feat: add new binary data parser

Dockerfile

@@ -0,0 +1,33 @@
+# Dockerfile References: https://docs.docker.com/engine/reference/builder/
+
+# Start from the latest golang base image
+FROM golang:latest
+
+# Add Maintainer Info
+LABEL maintainer="Yuriy Olkhovyy <y.olkhovyy@gmail.com>"
+
+ENV GOPATH /go
+
+# Set the Current Working Directory inside the container
+WORKDIR /go/src/github.com/influxdata/telegraf
+
+
+# Copy go mod and sum files
+# COPY go.mod go.sum ./
+
+# Download all dependencies. Dependencies will be cached if the go.mod and go.sum files are not changed
+# RUN go mod download
+
+# Copy the source from the current directory to the Working Directory inside the container
+COPY . /go/src/github.com/influxdata/telegraf
+
+# Build the Go app
+RUN go get -u github.com/golang/dep/cmd/dep
+RUN go get ./...
+RUN make
+
+# Expose port 8080 to the outside world
+# EXPOSE 8080
+
+# Command to run the executable
+CMD ["./telegraf"]

internal/config/config.go

@@ -18,6 +18,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/influxdata/telegraf/plugins/parsers/bindata"
+
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/internal/models"
@@ -1450,6 +1452,66 @@ func getParserConfig(name string, tbl *ast.Table) (*parsers.Config, error) {
 		}
 	}
 
+	if node, ok := tbl.Fields["bindata_protocol"]; ok {
+		if kv, ok := node.(*ast.KeyValue); ok {
+			if str, ok := kv.Value.(*ast.String); ok {
+				c.BinDataProtocol = str.Value
+			}
+		}
+	}
+
+	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_fields"]; ok {
+		if bindataFields, ok := node.([]*ast.Table); ok {
+			for _, bindataField := range bindataFields {
+				var field bindata.Field
+				fmt.Printf("D! FIELD ------\n")
+				for _, prop := range bindataField.Fields {
+					if kv, ok := prop.(*ast.KeyValue); ok {
+						if str, ok := kv.Value.(*ast.String); ok {
+							fmt.Printf("D! %s %s\n", kv.Key, str.Value)
+							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.Atoi(integer.Value)
+							if err == nil {
+								fmt.Printf("D! %s %d\n", kv.Key, v)
+								switch kv.Key {
+								case "offset":
+									field.Offset = v
+								case "size":
+									field.Size = 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 {
@@ -1792,6 +1854,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_protocol")
+	delete(tbl.Fields, "bindata_endiannes")
+	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,21 @@
+
+
+
+[IBM ZID file](https://www.ibm.com/support/knowledgecenter/en/SSVRGU_8.5.3/com.ibm.designer.domino.main.doc/H_FIELD_OPTIONS_AND_TEXT_INFORMATION_DEFINITION_SYNTAX_2922_OVER.html)
+
+[IBM ZID Example](https://www.ibm.com/support/knowledgecenter/en/SSVRGU_10.0.0/basic/H_BINARY_INPUT_FILES_USING_FIXED_LENGTH_RECORDS_6416_OVER.html)
+
+
+
+```toml
+[fixed_record]
+  metric_name = "drone_status"
+  endiannes = "be"
+
+  fields = [
+    {name="version",type="uint16",offset=0},
+    {name="time",type="int32",offset=2},
+  ]
+
+```
+

plugins/parsers/bindata/config.toml

@@ -0,0 +1,3 @@
+[fixed_record]
+  metric_name = "drone_status"
+  endiannes = "be"

plugins/parsers/bindata/parser.go

@@ -0,0 +1,163 @@
+package bindata
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"reflect"
+	"strings"
+	"time"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/internal"
+	"github.com/influxdata/telegraf/metric"
+)
+
+const timeKey = "time"
+const timezone = "UTC"
+const defaultProtocol = "raw"
+
+// Field is ...
+type Field struct {
+	Name   string
+	Type   string
+	Offset int
+	Size   int
+}
+
+// BinData is ...
+type BinData struct {
+	MetricName string
+	Protocol   string
+	TimeFormat string
+	Endiannes  string
+	Fields     []Field
+	// TagKeys    []string
+	DefaultTags map[string]string
+}
+
+// Parse is ...
+func (binData *BinData) Parse(data []byte) ([]telegraf.Metric, error) {
+
+	if _, err := binData.getProtocol(); err != nil {
+		return nil, err
+	}
+
+	endiannes, err := binData.getEndiannes()
+	if err != nil {
+		return nil, err
+	}
+
+	fields := make(map[string]interface{})
+	reader := io.NewSectionReader(bytes.NewReader(data), 0, int64(len(data)))
+
+	for _, field := range binData.Fields {
+		fieldBuffer := make([]byte, field.Size)
+
+		if _, err := reader.ReadAt(fieldBuffer, int64(field.Offset)); err != nil {
+			return nil, err
+		}
+
+		fieldType := fieldTypes[strings.ToLower(field.Type)]
+		if fieldType == nil {
+			return nil, fmt.Errorf(`invalid field type "%s""`, field.Type)
+		}
+
+		if fieldType.Name() == "string" {
+			fields[field.Name] = string(fieldBuffer)
+		} else {
+			fieldValue := reflect.New(fieldType)
+			byteReader := bytes.NewReader(fieldBuffer)
+			binary.Read(byteReader, endiannes, fieldValue.Interface())
+			fields[field.Name] = fieldValue.Elem().Interface()
+		}
+	}
+
+	metricTime, err := binData.getTime(fields)
+	if err != nil {
+		return nil, err
+	}
+
+	metric, err := metric.New(binData.MetricName, nil, fields, metricTime)
+	if err != nil {
+		return nil, err
+	}
+
+	// metricTags := make(map[string]string)
+	// for _, tagKey := range binData.BinData.TagKeys {
+	// 	metricTags[tagKey] = ""
+	// }
+
+	return []telegraf.Metric{metric}, err
+}
+
+// ParseLine is ...
+func (binData *BinData) ParseLine(line string) (telegraf.Metric, error) {
+	fmt.Println("BinData.ParseLine() not supported")
+	return nil, nil
+}
+
+// SetDefaultTags is ...
+func (binData *BinData) SetDefaultTags(tags map[string]string) {
+	fmt.Println("BinData.SetDefaultTags() not supported")
+}
+
+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(),
+}
+
+func (binData BinData) getProtocol() (string, error) {
+	protocol := strings.ToLower(binData.Protocol)
+	if protocol == "" {
+	} else if protocol != defaultProtocol {
+		return defaultProtocol, fmt.Errorf("only protocol %s is supported", defaultProtocol)
+	}
+	return defaultProtocol, nil
+}
+
+func (binData BinData) getEndiannes() (binary.ByteOrder, error) {
+	var endiannes binary.ByteOrder
+	cfgEndiannes := strings.ToLower(binData.Endiannes)
+	if cfgEndiannes == "" || cfgEndiannes == "be" {
+		endiannes = binary.BigEndian
+	} else if cfgEndiannes == "le" {
+		endiannes = binary.LittleEndian
+	} else {
+		return nil, fmt.Errorf("invalid binmetric_endiannes %s", cfgEndiannes)
+	}
+	return endiannes, nil
+}
+
+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":
+			metricTime, err = internal.ParseTimestamp(binData.TimeFormat, int64(timeValue.(int32)), timezone)
+		case "unix_ms", "unix_us", "unix_ns":
+			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
+}