Add new line protocol parser and serializer, influxdb output (#3924)

internal/config/config.go

@@ -9,6 +9,7 @@ import (
 	"math"
 	"os"
 	"path/filepath"
+
 	"regexp"
 	"runtime"
 	"sort"
@@ -1366,6 +1367,30 @@ func buildSerializer(name string, tbl *ast.Table) (serializers.Serializer, error
 		}
 	}
 
+	if node, ok := tbl.Fields["influx_max_line_bytes"]; ok {
+		if kv, ok := node.(*ast.KeyValue); ok {
+			if integer, ok := kv.Value.(*ast.Integer); ok {
+				v, err := integer.Int()
+				if err != nil {
+					return nil, err
+				}
+				c.InfluxMaxLineBytes = int(v)
+			}
+		}
+	}
+
+	if node, ok := tbl.Fields["influx_sort_fields"]; ok {
+		if kv, ok := node.(*ast.KeyValue); ok {
+			if b, ok := kv.Value.(*ast.Boolean); ok {
+				var err error
+				c.InfluxSortFields, err = b.Boolean()
+				if err != nil {
+					return nil, err
+				}
+			}
+		}
+	}
+
 	if node, ok := tbl.Fields["json_timestamp_units"]; ok {
 		if kv, ok := node.(*ast.KeyValue); ok {
 			if str, ok := kv.Value.(*ast.String); ok {
@@ -1382,6 +1407,8 @@ func buildSerializer(name string, tbl *ast.Table) (serializers.Serializer, error
 		}
 	}
 
+	delete(tbl.Fields, "influx_max_line_bytes")
+	delete(tbl.Fields, "influx_sort_fields")
 	delete(tbl.Fields, "data_format")
 	delete(tbl.Fields, "prefix")
 	delete(tbl.Fields, "template")

internal/models/makemetric.go

@@ -2,8 +2,6 @@ package models
 
 import (
 	"log"
-	"math"
-	"strings"
 	"time"
 
 	"github.com/influxdata/telegraf"
@@ -78,84 +76,6 @@ func makemetric(
 		}
 	}
 
-	for k, v := range tags {
-		if strings.HasSuffix(k, `\`) {
-			log.Printf("D! Measurement [%s] tag [%s] "+
-				"ends with a backslash, skipping", measurement, k)
-			delete(tags, k)
-			continue
-		} else if strings.HasSuffix(v, `\`) {
-			log.Printf("D! Measurement [%s] tag [%s] has a value "+
-				"ending with a backslash, skipping", measurement, k)
-			delete(tags, k)
-			continue
-		}
-	}
-
-	for k, v := range fields {
-		if strings.HasSuffix(k, `\`) {
-			log.Printf("D! Measurement [%s] field [%s] "+
-				"ends with a backslash, skipping", measurement, k)
-			delete(fields, k)
-			continue
-		}
-		// Validate uint64 and float64 fields
-		// convert all int & uint types to int64
-		switch val := v.(type) {
-		case nil:
-			// delete nil fields
-			delete(fields, k)
-		case uint:
-			fields[k] = int64(val)
-			continue
-		case uint8:
-			fields[k] = int64(val)
-			continue
-		case uint16:
-			fields[k] = int64(val)
-			continue
-		case uint32:
-			fields[k] = int64(val)
-			continue
-		case int:
-			fields[k] = int64(val)
-			continue
-		case int8:
-			fields[k] = int64(val)
-			continue
-		case int16:
-			fields[k] = int64(val)
-			continue
-		case int32:
-			fields[k] = int64(val)
-			continue
-		case uint64:
-			// InfluxDB does not support writing uint64
-			if val < uint64(9223372036854775808) {
-				fields[k] = int64(val)
-			} else {
-				fields[k] = int64(9223372036854775807)
-			}
-			continue
-		case float32:
-			fields[k] = float64(val)
-			continue
-		case float64:
-			// NaNs are invalid values in influxdb, skip measurement
-			if math.IsNaN(val) || math.IsInf(val, 0) {
-				log.Printf("D! Measurement [%s] field [%s] has a NaN or Inf "+
-					"field, skipping",
-					measurement, k)
-				delete(fields, k)
-				continue
-			}
-		case string:
-			fields[k] = v
-		default:
-			fields[k] = v
-		}
-	}
-
 	m, err := metric.New(measurement, tags, fields, t, mType)
 	if err != nil {
 		log.Printf("Error adding point [%s]: %s\n", measurement, err.Error())

internal/models/running_aggregator_test.go

@@ -1,7 +1,6 @@
 package models
 
 import (
-	"fmt"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -167,69 +166,6 @@ func TestAddDropOriginal(t *testing.T) {
 	assert.False(t, ra.Add(m2))
 }
 
-// make an untyped, counter, & gauge metric
-func TestMakeMetricA(t *testing.T) {
-	now := time.Now()
-	ra := NewRunningAggregator(&TestAggregator{}, &AggregatorConfig{
-		Name: "TestRunningAggregator",
-	})
-	assert.Equal(t, "aggregators.TestRunningAggregator", ra.Name())
-
-	m := ra.MakeMetric(
-		"RITest",
-		map[string]interface{}{"value": int(101)},
-		map[string]string{},
-		telegraf.Untyped,
-		now,
-	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("RITest value=101i %d\n", now.UnixNano()),
-		m.String(),
-	)
-	assert.Equal(
-		t,
-		m.Type(),
-		telegraf.Untyped,
-	)
-
-	m = ra.MakeMetric(
-		"RITest",
-		map[string]interface{}{"value": int(101)},
-		map[string]string{},
-		telegraf.Counter,
-		now,
-	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("RITest value=101i %d\n", now.UnixNano()),
-		m.String(),
-	)
-	assert.Equal(
-		t,
-		m.Type(),
-		telegraf.Counter,
-	)
-
-	m = ra.MakeMetric(
-		"RITest",
-		map[string]interface{}{"value": int(101)},
-		map[string]string{},
-		telegraf.Gauge,
-		now,
-	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("RITest value=101i %d\n", now.UnixNano()),
-		m.String(),
-	)
-	assert.Equal(
-		t,
-		m.Type(),
-		telegraf.Gauge,
-	)
-}
-
 type TestAggregator struct {
 	sum int64
 }

internal/models/running_input.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/serializers/influx"
 	"github.com/influxdata/telegraf/selfstat"
 )
 
@@ -75,7 +76,11 @@ func (r *RunningInput) MakeMetric(
 	)
 
 	if r.trace && m != nil {
-		fmt.Print("> " + m.String())
+		s := influx.NewSerializer()
+		octets, err := s.Serialize(m)
+		if err == nil {
+			fmt.Print("> " + string(octets))
+		}
 	}
 
 	r.MetricsGathered.Incr(1)

internal/models/running_input_test.go

@@ -1,12 +1,11 @@
 package models
 
 import (
-	"fmt"
-	"math"
 	"testing"
 	"time"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/metric"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -45,77 +44,17 @@ func TestMakeMetricNilFields(t *testing.T) {
 		telegraf.Untyped,
 		now,
 	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("RITest value=101i %d\n", now.UnixNano()),
-		m.String(),
-	)
-}
 
-// make an untyped, counter, & gauge metric
-func TestMakeMetric(t *testing.T) {
-	now := time.Now()
-	ri := NewRunningInput(&testInput{}, &InputConfig{
-		Name: "TestRunningInput",
-	})
-
-	ri.SetTrace(true)
-	assert.Equal(t, true, ri.Trace())
-	assert.Equal(t, "inputs.TestRunningInput", ri.Name())
-
-	m := ri.MakeMetric(
-		"RITest",
-		map[string]interface{}{"value": int(101)},
+	expected, err := metric.New("RITest",
 		map[string]string{},
-		telegraf.Untyped,
+		map[string]interface{}{
+			"value": int(101),
+		},
 		now,
 	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("RITest value=101i %d\n", now.UnixNano()),
-		m.String(),
-	)
-	assert.Equal(
-		t,
-		m.Type(),
-		telegraf.Untyped,
-	)
+	require.NoError(t, err)
 
-	m = ri.MakeMetric(
-		"RITest",
-		map[string]interface{}{"value": int(101)},
-		map[string]string{},
-		telegraf.Counter,
-		now,
-	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("RITest value=101i %d\n", now.UnixNano()),
-		m.String(),
-	)
-	assert.Equal(
-		t,
-		m.Type(),
-		telegraf.Counter,
-	)
-
-	m = ri.MakeMetric(
-		"RITest",
-		map[string]interface{}{"value": int(101)},
-		map[string]string{},
-		telegraf.Gauge,
-		now,
-	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("RITest value=101i %d\n", now.UnixNano()),
-		m.String(),
-	)
-	assert.Equal(
-		t,
-		m.Type(),
-		telegraf.Gauge,
-	)
+	require.Equal(t, expected, m)
 }
 
 func TestMakeMetricWithPluginTags(t *testing.T) {
@@ -137,11 +76,18 @@ func TestMakeMetricWithPluginTags(t *testing.T) {
 		telegraf.Untyped,
 		now,
 	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("RITest,foo=bar value=101i %d\n", now.UnixNano()),
-		m.String(),
+
+	expected, err := metric.New("RITest",
+		map[string]string{
+			"foo": "bar",
+		},
+		map[string]interface{}{
+			"value": 101,
+		},
+		now,
 	)
+	require.NoError(t, err)
+	require.Equal(t, expected, m)
 }
 
 func TestMakeMetricFilteredOut(t *testing.T) {
@@ -187,87 +133,17 @@ func TestMakeMetricWithDaemonTags(t *testing.T) {
 		telegraf.Untyped,
 		now,
 	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("RITest,foo=bar value=101i %d\n", now.UnixNano()),
-		m.String(),
-	)
-}
-
-// make an untyped, counter, & gauge metric
-func TestMakeMetricInfFields(t *testing.T) {
-	inf := math.Inf(1)
-	ninf := math.Inf(-1)
-	now := time.Now()
-	ri := NewRunningInput(&testInput{}, &InputConfig{
-		Name: "TestRunningInput",
-	})
-
-	ri.SetTrace(true)
-	assert.Equal(t, true, ri.Trace())
-
-	m := ri.MakeMetric(
-		"RITest",
-		map[string]interface{}{
-			"value": int(101),
-			"inf":   inf,
-			"ninf":  ninf,
+	expected, err := metric.New("RITest",
+		map[string]string{
+			"foo": "bar",
+		},
+		map[string]interface{}{
+			"value": 101,
 		},
-		map[string]string{},
-		telegraf.Untyped,
 		now,
 	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("RITest value=101i %d\n", now.UnixNano()),
-		m.String(),
-	)
-}
-
-func TestMakeMetricAllFieldTypes(t *testing.T) {
-	now := time.Now()
-	ri := NewRunningInput(&testInput{}, &InputConfig{
-		Name: "TestRunningInput",
-	})
-
-	ri.SetTrace(true)
-	assert.Equal(t, true, ri.Trace())
-
-	m := ri.MakeMetric(
-		"RITest",
-		map[string]interface{}{
-			"a": int(10),
-			"b": int8(10),
-			"c": int16(10),
-			"d": int32(10),
-			"e": uint(10),
-			"f": uint8(10),
-			"g": uint16(10),
-			"h": uint32(10),
-			"i": uint64(10),
-			"j": float32(10),
-			"k": uint64(9223372036854775810),
-			"l": "foobar",
-			"m": true,
-		},
-		map[string]string{},
-		telegraf.Untyped,
-		now,
-	)
-	assert.Contains(t, m.String(), "a=10i")
-	assert.Contains(t, m.String(), "b=10i")
-	assert.Contains(t, m.String(), "c=10i")
-	assert.Contains(t, m.String(), "d=10i")
-	assert.Contains(t, m.String(), "e=10i")
-	assert.Contains(t, m.String(), "f=10i")
-	assert.Contains(t, m.String(), "g=10i")
-	assert.Contains(t, m.String(), "h=10i")
-	assert.Contains(t, m.String(), "i=10i")
-	assert.Contains(t, m.String(), "j=10")
-	assert.NotContains(t, m.String(), "j=10i")
-	assert.Contains(t, m.String(), "k=9223372036854775807i")
-	assert.Contains(t, m.String(), "l=\"foobar\"")
-	assert.Contains(t, m.String(), "m=true")
+	require.NoError(t, err)
+	require.Equal(t, expected, m)
 }
 
 func TestMakeMetricNameOverride(t *testing.T) {
@@ -284,11 +160,15 @@ func TestMakeMetricNameOverride(t *testing.T) {
 		telegraf.Untyped,
 		now,
 	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("foobar value=101i %d\n", now.UnixNano()),
-		m.String(),
+	expected, err := metric.New("foobar",
+		nil,
+		map[string]interface{}{
+			"value": 101,
+		},
+		now,
 	)
+	require.NoError(t, err)
+	require.Equal(t, expected, m)
 }
 
 func TestMakeMetricNamePrefix(t *testing.T) {
@@ -305,11 +185,15 @@ func TestMakeMetricNamePrefix(t *testing.T) {
 		telegraf.Untyped,
 		now,
 	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("foobar_RITest value=101i %d\n", now.UnixNano()),
-		m.String(),
+	expected, err := metric.New("foobar_RITest",
+		nil,
+		map[string]interface{}{
+			"value": 101,
+		},
+		now,
 	)
+	require.NoError(t, err)
+	require.Equal(t, expected, m)
 }
 
 func TestMakeMetricNameSuffix(t *testing.T) {
@@ -326,134 +210,15 @@ func TestMakeMetricNameSuffix(t *testing.T) {
 		telegraf.Untyped,
 		now,
 	)
-	assert.Equal(
-		t,
-		fmt.Sprintf("RITest_foobar value=101i %d\n", now.UnixNano()),
-		m.String(),
+	expected, err := metric.New("RITest_foobar",
+		nil,
+		map[string]interface{}{
+			"value": 101,
+		},
+		now,
 	)
-}
-
-func TestMakeMetric_TrailingSlash(t *testing.T) {
-	now := time.Now()
-
-	tests := []struct {
-		name                string
-		measurement         string
-		fields              map[string]interface{}
-		tags                map[string]string
-		expectedNil         bool
-		expectedMeasurement string
-		expectedFields      map[string]interface{}
-		expectedTags        map[string]string
-	}{
-		{
-			name:        "Measurement cannot have trailing slash",
-			measurement: `cpu\`,
-			fields: map[string]interface{}{
-				"value": int64(42),
-			},
-			tags:        map[string]string{},
-			expectedNil: true,
-		},
-		{
-			name:        "Field key with trailing slash dropped",
-			measurement: `cpu`,
-			fields: map[string]interface{}{
-				"value": int64(42),
-				`bad\`:  `xyzzy`,
-			},
-			tags:                map[string]string{},
-			expectedMeasurement: `cpu`,
-			expectedFields: map[string]interface{}{
-				"value": int64(42),
-			},
-			expectedTags: map[string]string{},
-		},
-		{
-			name:        "Field value with trailing slash okay",
-			measurement: `cpu`,
-			fields: map[string]interface{}{
-				"value": int64(42),
-				"ok":    `xyzzy\`,
-			},
-			tags:                map[string]string{},
-			expectedMeasurement: `cpu`,
-			expectedFields: map[string]interface{}{
-				"value": int64(42),
-				"ok":    `xyzzy\`,
-			},
-			expectedTags: map[string]string{},
-		},
-		{
-			name:        "Must have one field after dropped",
-			measurement: `cpu`,
-			fields: map[string]interface{}{
-				"bad": math.NaN(),
-			},
-			tags:        map[string]string{},
-			expectedNil: true,
-		},
-		{
-			name:        "Tag key with trailing slash dropped",
-			measurement: `cpu`,
-			fields: map[string]interface{}{
-				"value": int64(42),
-			},
-			tags: map[string]string{
-				`host\`: "localhost",
-				"a":     "x",
-			},
-			expectedMeasurement: `cpu`,
-			expectedFields: map[string]interface{}{
-				"value": int64(42),
-			},
-			expectedTags: map[string]string{
-				"a": "x",
-			},
-		},
-		{
-			name:        "Tag value with trailing slash dropped",
-			measurement: `cpu`,
-			fields: map[string]interface{}{
-				"value": int64(42),
-			},
-			tags: map[string]string{
-				`host`: `localhost\`,
-				"a":    "x",
-			},
-			expectedMeasurement: `cpu`,
-			expectedFields: map[string]interface{}{
-				"value": int64(42),
-			},
-			expectedTags: map[string]string{
-				"a": "x",
-			},
-		},
-	}
-
-	ri := NewRunningInput(&testInput{}, &InputConfig{
-		Name: "TestRunningInput",
-	})
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			m := ri.MakeMetric(
-				tc.measurement,
-				tc.fields,
-				tc.tags,
-				telegraf.Untyped,
-				now)
-
-			if tc.expectedNil {
-				require.Nil(t, m)
-			} else {
-				require.NotNil(t, m)
-				require.Equal(t, tc.expectedMeasurement, m.Name())
-				require.Equal(t, tc.expectedFields, m.Fields())
-				require.Equal(t, tc.expectedTags, m.Tags())
-			}
-		})
-	}
+	require.NoError(t, err)
+	require.Equal(t, expected, m)
 }
 
 type testInput struct{}