feat: add new binary data parser

config/config.go

@@ -25,6 +25,7 @@ import (
 	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/outputs"
 	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/parsers/bindata"
 	"github.com/influxdata/telegraf/plugins/processors"
 	"github.com/influxdata/telegraf/plugins/serializers"
 	"github.com/influxdata/toml"
@@ -1492,6 +1493,61 @@ func getParserConfig(name string, tbl *ast.Table) (*parsers.Config, error) {
 		}
 	}
 
+	if node, ok := tbl.Fields["bindata_endiannes"]; ok {
+		if kv, ok := node.(*ast.KeyValue); ok {
+			if str, ok := kv.Value.(*ast.String); ok {
+				c.BinDataEndiannes = str.Value
+			}
+		}
+	}
+
+	if node, ok := tbl.Fields["bindata_time_format"]; ok {
+		if kv, ok := node.(*ast.KeyValue); ok {
+			if str, ok := kv.Value.(*ast.String); ok {
+				c.BinDataTimeFormat = str.Value
+			}
+		}
+	}
+
+	if node, ok := tbl.Fields["bindata_string_encoding"]; ok {
+		if kv, ok := node.(*ast.KeyValue); ok {
+			if str, ok := kv.Value.(*ast.String); ok {
+				c.BinDataStringEncoding = str.Value
+			}
+		}
+	}
+
+	if node, ok := tbl.Fields["bindata_fields"]; ok {
+		if bindataFields, ok := node.([]*ast.Table); ok {
+			for _, bindataField := range bindataFields {
+				var field bindata.Field
+				for _, prop := range bindataField.Fields {
+					if kv, ok := prop.(*ast.KeyValue); ok {
+						if str, ok := kv.Value.(*ast.String); ok {
+							switch kv.Key {
+							case "name":
+								field.Name = str.Value
+							case "type":
+								field.Type = str.Value
+							default:
+							}
+						} else if integer, ok := kv.Value.(*ast.Integer); ok {
+							v, err := strconv.ParseUint(integer.Value, 10, 32)
+							if err == nil {
+								switch kv.Key {
+								case "size":
+									field.Size = uint(v)
+								default:
+								}
+							}
+						}
+					}
+				}
+				c.BinDataFields = append(c.BinDataFields, field)
+			}
+		}
+	}
+
 	if node, ok := tbl.Fields["json_string_fields"]; ok {
 		if kv, ok := node.(*ast.KeyValue); ok {
 			if ary, ok := kv.Value.(*ast.Array); ok {
@@ -1854,6 +1910,10 @@ func getParserConfig(name string, tbl *ast.Table) (*parsers.Config, error) {
 	delete(tbl.Fields, "separator")
 	delete(tbl.Fields, "templates")
 	delete(tbl.Fields, "tag_keys")
+	delete(tbl.Fields, "bindata_endiannes")
+	delete(tbl.Fields, "bindata_string_encoding")
+	delete(tbl.Fields, "bindata_time_format")
+	delete(tbl.Fields, "bindata_fields")
 	delete(tbl.Fields, "json_name_key")
 	delete(tbl.Fields, "json_query")
 	delete(tbl.Fields, "json_string_fields")

plugins/parsers/bindata/README.md

@@ -0,0 +1,55 @@
+# BinData
+
+The "BinData" parser translates binary records consisting of multiple fields into Telegraf metrics. It supports:
+
+- Little- and Big-Endian encoding
+- bool, int8/uint8, int16/uint16, int32/uint32, int64/uint64, float32/float64 field types
+- UTF-8 and ASCII-encoded strings
+- unix, unix_ms, unix_us and unix_ns timestamp
+
+### Configuration
+
+```toml
+[[inputs.mqtt_consumer]]
+  name_override = "drone_status"
+
+  ## Data format to consume.
+  ## Each data format has its own unique set of configuration options, read
+  ## more about them here:
+  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
+  data_format = "bindata"
+
+  ## Numeric fields endiannes, "be" or "le", default "be"
+  bindata_endiannes = "be"
+
+  ## Timestamp format - "unix", "unix_ms", "unix_us", "unix_ns", default "unix"
+  bindata_time_format = "unix"
+
+  ## String encoding, default "UTF-8"
+  bindata_string_encoding = "UTF-8"
+
+  ## Binary data descriptor
+  ## Fields are described by:
+  ## - name - field name, use Golang naming conventions
+  ## - type - field type, use Golang type names
+  ## - size - size in bytes, obligatory for fields with type "string" and "padding", ignored in numeric and bool fields
+  ## Field with case incensitive name "time" has special meaning - it is used as metric time and must be of type
+  ## - int32 for bindata_time_format = "unix", or of type
+  ## - int64 for bindata_time_format = "unix_ms", "unix_us", "unix_ns".
+  ## Use padding when auto-generated metric time is preferred.
+  bindata_fields = [
+    {name="Version",type="uint16"},
+    {name="Time",type="int32"},
+    {name="Latitude",type="float64"},
+    {name="Longitude",type="float64"},
+    {name="Altitude",type="float32"},
+    {name="Heading",type="float32"},
+    {name="Elevation",type="float32"},
+    {name="Bank",type="float32"},
+    {name="GroundSpeed",type="float32"},
+    {name="AirSpeed",type="float32"},
+    {name="None",type="padding", size=16},
+    {name="Status",type="string",size=7},
+    {name="StatusOK",type="bool"},
+  ]
+```

plugins/parsers/bindata/parser.go

@@ -0,0 +1,216 @@
+package bindata
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"reflect"
+	"strings"
+	"time"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/internal"
+	"github.com/influxdata/telegraf/metric"
+)
+
+const timeKey = "time"
+const timezone = "UTC"
+const defaultStringEncoding = "UTF-8"
+const defaultTimeFormat = "unix"
+
+// Field is a binary data field descriptor
+type Field struct {
+	Name string
+	Type string
+	Size uint
+}
+
+// BinData is a binary data parser
+type BinData struct {
+	metricName     string
+	timeFormat     string
+	endiannes      string
+	byteOrder      binary.ByteOrder
+	stringEncoding string
+	fields         []Field
+	DefaultTags    map[string]string
+}
+
+// Supported field types
+var fieldTypes = map[string]reflect.Type{
+	"bool":    reflect.TypeOf((*bool)(nil)).Elem(),
+	"uint8":   reflect.TypeOf((*uint8)(nil)).Elem(),
+	"int8":    reflect.TypeOf((*int8)(nil)).Elem(),
+	"uint16":  reflect.TypeOf((*uint16)(nil)).Elem(),
+	"int16":   reflect.TypeOf((*int16)(nil)).Elem(),
+	"uint32":  reflect.TypeOf((*uint32)(nil)).Elem(),
+	"int32":   reflect.TypeOf((*int32)(nil)).Elem(),
+	"uint64":  reflect.TypeOf((*uint64)(nil)).Elem(),
+	"int64":   reflect.TypeOf((*int64)(nil)).Elem(),
+	"float32": reflect.TypeOf((*float32)(nil)).Elem(),
+	"float64": reflect.TypeOf((*float64)(nil)).Elem(),
+	"string":  reflect.TypeOf((*string)(nil)).Elem(),
+	"padding": reflect.TypeOf((*[]byte)(nil)).Elem(),
+}
+
+// NewBinDataParser is BinData factory
+func NewBinDataParser(
+	metricName string,
+	timeFormat string,
+	endiannes string,
+	stringEncoding string,
+	fields []Field,
+	defaultTags map[string]string,
+) (*BinData, error) {
+
+	// Time format
+	switch timeFormat {
+	case "":
+		timeFormat = defaultTimeFormat
+	case "unix", "unix_ms", "unix_us", "unix_ns":
+	default:
+		return nil, fmt.Errorf("invalid time format %s", timeFormat)
+	}
+
+	// Endiannes
+	var byteOrder binary.ByteOrder
+	endiannes = strings.ToLower(endiannes)
+	switch endiannes {
+	case "", "be":
+		byteOrder = binary.BigEndian
+	case "le":
+		byteOrder = binary.LittleEndian
+	default:
+		return nil, fmt.Errorf("invalid bindata_endiannes %s", endiannes)
+	}
+
+	// String encoding
+	if stringEncoding == "" {
+		stringEncoding = defaultStringEncoding
+	}
+	stringEncoding = strings.ToUpper(stringEncoding)
+	if stringEncoding != defaultStringEncoding {
+		return nil, fmt.Errorf(`invalid string encoding %s`, stringEncoding)
+	}
+
+	// Field types, names and sizes
+	knownFields := make(map[string]bool)
+	for i := 0; i < len(fields); i++ {
+		fieldType, ok := fieldTypes[strings.ToLower(fields[i].Type)]
+		if !ok {
+			return nil, fmt.Errorf(`invalid field type %s`, fields[i].Type)
+		}
+
+		if fields[i].Type != "padding" {
+			// Check for duplicate field names
+			fieldName := fields[i].Name
+			if _, ok := knownFields[fieldName]; ok {
+				return nil, fmt.Errorf(`duplicate field name %s`, fieldName)
+			}
+			knownFields[fieldName] = true
+
+			// Time field type check
+			if fieldName == "time" {
+				if timeFormat == "unix" &&
+					fieldType != reflect.TypeOf((*int32)(nil)).Elem() {
+					return nil, fmt.Errorf(`invalid time type, must be int32`)
+				}
+				if (timeFormat == "unix_ms" || timeFormat == "unix_us" || timeFormat == "unix_ns") &&
+					fieldType != reflect.TypeOf((*int64)(nil)).Elem() {
+					return nil, fmt.Errorf(`invalid time type, must be int64`)
+				}
+			}
+
+			// Overwrite non-string and non-padding field size
+			if fieldType.Name() != "string" {
+				fields[i].Size = uint(fieldType.Size())
+			}
+		}
+	}
+
+	return &BinData{
+		metricName:     metricName,
+		timeFormat:     timeFormat,
+		endiannes:      endiannes,
+		byteOrder:      byteOrder,
+		stringEncoding: stringEncoding,
+		fields:         fields,
+		DefaultTags:    defaultTags,
+	}, nil
+}
+
+// SetDefaultTags implements Parser.SetDefaultTags()
+func (binData *BinData) SetDefaultTags(tags map[string]string) {
+	binData.DefaultTags = tags
+}
+
+// Parse implements Parser.Parse()
+func (binData *BinData) Parse(data []byte) ([]telegraf.Metric, error) {
+
+	fields := make(map[string]interface{})
+	var offset uint = 0
+
+	for _, field := range binData.fields {
+		if offset > uint(len(data)) || offset+field.Size > uint(len(data)) {
+			return nil, fmt.Errorf("invalid offset/size in field %s", field.Name)
+		}
+		if field.Type != "padding" {
+			fieldBuffer := data[offset : offset+field.Size]
+			switch field.Type {
+			case "string":
+				fields[field.Name] = string(fieldBuffer)
+			default:
+				fieldValue := reflect.New(fieldTypes[field.Type])
+				byteReader := bytes.NewReader(fieldBuffer)
+				binary.Read(byteReader, binData.byteOrder, fieldValue.Interface())
+				fields[field.Name] = fieldValue.Elem().Interface()
+			}
+		}
+		offset += field.Size
+	}
+
+	metricTime, err := binData.getTime(fields)
+	if err != nil {
+		return nil, err
+	}
+
+	metric, err := metric.New(binData.metricName, binData.DefaultTags, fields, metricTime)
+	if err != nil {
+		return nil, err
+	}
+
+	return []telegraf.Metric{metric}, err
+}
+
+// ParseLine implements Parser.ParseLine()
+func (binData *BinData) ParseLine(line string) (telegraf.Metric, error) {
+	return nil, fmt.Errorf("BinData.ParseLine() not supported")
+}
+
+func (binData *BinData) getTime(fields map[string]interface{}) (time.Time, error) {
+	nilTime := new(time.Time)
+	metricTime := time.Now()
+	timeValue := fields[timeKey]
+	if timeValue != nil {
+		var err error
+		switch binData.timeFormat {
+		case "unix":
+			if _, ok := timeValue.(int32); !ok {
+				return *nilTime, fmt.Errorf("invalid time type, must be int32")
+			}
+			metricTime, err = internal.ParseTimestamp(binData.timeFormat, int64(timeValue.(int32)), timezone)
+		case "unix_ms", "unix_us", "unix_ns":
+			if _, ok := timeValue.(int64); !ok {
+				return *nilTime, fmt.Errorf("invalid time type, must be int64")
+			}
+			metricTime, err = internal.ParseTimestamp(binData.timeFormat, int64(timeValue.(int64)), timezone)
+		default:
+			return *nilTime, fmt.Errorf("invalid time format %s", binData.timeFormat)
+		}
+		if err != nil {
+			return *nilTime, err
+		}
+		delete(fields, timeKey)
+	}
+	return metricTime, nil
+}