CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(parsers.value): Add support for automatic fallback for numeric types (#13506)

This commit is contained in:
Sven Rebhan 2023-06-29 12:13:34 +02:00 committed by GitHub
parent c35cabda9e
commit e2fb181a80
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 302 additions and 251 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
plugins/outputs/mqtt/mqtt_test.go
View File

@ -16,7 +16,6 @@ import (
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/plugins/common/mqtt"
"github.com/influxdata/telegraf/plugins/parsers/influx"
"github.com/influxdata/telegraf/plugins/parsers/value"
influxSerializer "github.com/influxdata/telegraf/plugins/serializers/influx"
"github.com/influxdata/telegraf/testutil"
)
@ -598,13 +597,6 @@ func TestIntegrationMQTTLayoutHomieV4(t *testing.T) {
require.NoError(t, container.Start(), "failed to start container")
defer container.Terminate()
// Setup the parser / serializer pair
parser := &value.Parser{
MetricName: "test",
DataType: "auto",
}
require.NoError(t, parser.Init())
// Setup the plugin
url := fmt.Sprintf("tcp://%s:%s", container.Address, container.Ports[servicePort])
topic := "homie"

34
plugins/parsers/value/README.md
View File

@ -25,14 +25,34 @@ as the parsed metric.
data_type = "integer" # required
```
### Metric name
It is recommended to set `name_override` to a measurement name that makes sense
for your metric, otherwise it will just be set to the name of the plugin.
### Datatype
You **must** tell Telegraf what type of metric to collect by using the
`data_type` configuration option. Available options are:
1. integer
2. float or long
3. string
4. boolean
- `integer`: converts the received data to an integer value. This setting will
produce an error on non-integer data.
- `float`: converts the received data to a floating-point value. This setting
will treat integers as floating-point values and produces an error
on data that cannot be converted (e.g. strings).
- `string`: outputs the data as a string.
- `boolean`: converts the received data to a boolean value. This setting will
produce an error on any data except for `true` and `false` strings.
- `auto_integer`: converts the received data to an integer value if possible and
will return the data as string otherwise. This is helpful for
mixed-type data.
- `auto_float`: converts the received data to a floating-point value if possible
and will return the data as string otherwise. This is helpful
for mixed-type data. Integer data will be treated as
floating-point values.
**Note:** It is also recommended that you set `name_override` to a measurement
name that makes sense for your metric, otherwise it will just be set to the
name of the plugin.
**NOTE**: The `auto` conversions might convert data to their prioritized type
by accident, for example if a string data-source provides `"55"` it will be
converted to integer/float. This might break outputs that require the same
datatype within a field or column. It is thus recommended to use *strict* typing
whenever possible.

51
plugins/parsers/value/parser.go
View File

@ -19,6 +19,29 @@ type Parser struct {
DefaultTags map[string]string `toml:"-"`
}
func (v *Parser) Init() error {
switch v.DataType {
case "", "int", "integer":
v.DataType = "int"
case "float", "long":
v.DataType = "float"
case "str", "string":
v.DataType = "string"
case "bool", "boolean":
v.DataType = "bool"
case "auto_integer", "auto_float":
// Do nothing both are valid
default:
return fmt.Errorf("unknown datatype %q", v.DataType)
}
if v.FieldName == "" {
v.FieldName = "value"
}
return nil
}
func (v *Parser) Parse(buf []byte) ([]telegraf.Metric, error) {
vStr := string(bytes.TrimSpace(bytes.Trim(buf, "\x00")))
@ -35,14 +58,26 @@ func (v *Parser) Parse(buf []byte) ([]telegraf.Metric, error) {
var value interface{}
var err error
switch v.DataType {
case "", "int", "integer":
case "int":
value, err = strconv.Atoi(vStr)
case "float", "long":
case "float":
value, err = strconv.ParseFloat(vStr, 64)
case "str", "string":
case "string":
value = vStr
case "bool", "boolean":
case "bool":
value, err = strconv.ParseBool(vStr)
case "auto_integer":
value, err = strconv.Atoi(vStr)
if err != nil {
value = vStr
err = nil
}
case "auto_float":
value, err = strconv.ParseFloat(vStr, 64)
if err != nil {
value = vStr
err = nil
}
}
if err != nil {
return nil, err
@ -73,14 +108,6 @@ func (v *Parser) SetDefaultTags(tags map[string]string) {
v.DefaultTags = tags
}
func (v *Parser) Init() error {
if v.FieldName == "" {
v.FieldName = "value"
}
return nil
}
func init() {
parsers.Add("value",
func(defaultMetricName string) telegraf.Parser {

460
plugins/parsers/value/parser_test.go
View File

@ -2,273 +2,277 @@ package value
import (
"testing"
"time"
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/require"
)
func TestParseValidValues(t *testing.T) {
parser := Parser{
MetricName: "value_test",
DataType: "integer",
tests := []struct {
name string
dtype string
input []byte
expected interface{}
}{
{
name: "integer",
dtype: "integer",
input: []byte("55"),
expected: int64(55),
},
{
name: "float",
dtype: "float",
input: []byte("64"),
expected: float64(64),
},
{
name: "string",
dtype: "string",
input: []byte("foobar"),
expected: "foobar",
},
{
name: "boolean",
dtype: "boolean",
input: []byte("true"),
expected: true,
},
{
name: "multiple integers",
dtype: "integer",
input: []byte(`55 45 223 12 999`),
expected: int64(999),
},
{
name: "auto integer",
dtype: "auto_integer",
input: []byte("55"),
expected: int64(55),
},
{
name: "auto integer with string",
dtype: "auto_integer",
input: []byte("foobar"),
expected: "foobar",
},
{
name: "auto integer with float",
dtype: "auto_integer",
input: []byte("55.0"),
expected: "55.0",
},
{
name: "auto float",
dtype: "auto_float",
input: []byte("64.2"),
expected: float64(64.2),
},
{
name: "auto float with string",
dtype: "auto_float",
input: []byte("foobar"),
expected: "foobar",
},
{
name: "auto float with integer",
dtype: "auto_float",
input: []byte("64"),
expected: float64(64),
},
}
require.NoError(t, parser.Init())
metrics, err := parser.Parse([]byte("55"))
require.NoError(t, err)
require.Len(t, metrics, 1)
require.Equal(t, "value_test", metrics[0].Name())
require.Equal(t, map[string]interface{}{
"value": int64(55),
}, metrics[0].Fields())
require.Equal(t, map[string]string{}, metrics[0].Tags())
parser = Parser{
MetricName: "value_test",
DataType: "float",
}
require.NoError(t, parser.Init())
metrics, err = parser.Parse([]byte("64"))
require.NoError(t, err)
require.Len(t, metrics, 1)
require.Equal(t, "value_test", metrics[0].Name())
require.Equal(t, map[string]interface{}{
"value": float64(64),
}, metrics[0].Fields())
require.Equal(t, map[string]string{}, metrics[0].Tags())
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
expected := metric.New(
"value_test",
map[string]string{},
map[string]interface{}{"value": tt.expected},
time.Unix(0, 0),
)
parser = Parser{
MetricName: "value_test",
DataType: "string",
plugin := Parser{
MetricName: "value_test",
DataType: tt.dtype,
}
require.NoError(t, plugin.Init())
actual, err := plugin.Parse(tt.input)
require.NoError(t, err)
require.Len(t, actual, 1)
testutil.RequireMetricEqual(t, expected, actual[0], testutil.IgnoreTime())
})
}
require.NoError(t, parser.Init())
metrics, err = parser.Parse([]byte("foobar"))
require.NoError(t, err)
require.Len(t, metrics, 1)
require.Equal(t, "value_test", metrics[0].Name())
require.Equal(t, map[string]interface{}{
"value": "foobar",
}, metrics[0].Fields())
require.Equal(t, map[string]string{}, metrics[0].Tags())
parser = Parser{
MetricName: "value_test",
DataType: "boolean",
}
require.NoError(t, parser.Init())
metrics, err = parser.Parse([]byte("true"))
require.NoError(t, err)
require.Len(t, metrics, 1)
require.Equal(t, "value_test", metrics[0].Name())
require.Equal(t, map[string]interface{}{
"value": true,
}, metrics[0].Fields())
require.Equal(t, map[string]string{}, metrics[0].Tags())
}
func TestParseMultipleValues(t *testing.T) {
parser := Parser{
MetricName: "value_test",
DataType: "integer",
func TestParseLineValidValues(t *testing.T) {
tests := []struct {
name string
dtype string
input string
expected interface{}
}{
{
name: "integer",
dtype: "integer",
input: "55",
expected: int64(55),
},
{
name: "float",
dtype: "float",
input: "64",
expected: float64(64),
},
{
name: "string",
dtype: "string",
input: "foobar",
expected: "foobar",
},
{
name: "boolean",
dtype: "boolean",
input: "true",
expected: true,
},
{
name: "multiple integers",
dtype: "integer",
input: `55 45 223 12 999`,
expected: int64(999),
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
expected := metric.New(
"value_test",
map[string]string{},
map[string]interface{}{"value": tt.expected},
time.Unix(0, 0),
)
plugin := Parser{
MetricName: "value_test",
DataType: tt.dtype,
}
require.NoError(t, plugin.Init())
actual, err := plugin.ParseLine(tt.input)
require.NoError(t, err)
testutil.RequireMetricEqual(t, expected, actual, testutil.IgnoreTime())
})
}
require.NoError(t, parser.Init())
metrics, err := parser.Parse([]byte(`55
45
223
12
999
`))
require.NoError(t, err)
require.Len(t, metrics, 1)
require.Equal(t, "value_test", metrics[0].Name())
require.Equal(t, map[string]interface{}{
"value": int64(999),
}, metrics[0].Fields())
require.Equal(t, map[string]string{}, metrics[0].Tags())
}
func TestParseCustomFieldName(t *testing.T) {
parser := Parser{
MetricName: "value_test",
DataType: "integer",
FieldName: "penguin",
}
require.NoError(t, parser.Init())
parser.FieldName = "penguin"
metrics, err := parser.Parse([]byte(`55`))
require.NoError(t, err)
require.Equal(t, map[string]interface{}{
"penguin": int64(55),
}, metrics[0].Fields())
}
func TestParseLineValidValues(t *testing.T) {
parser := Parser{
MetricName: "value_test",
DataType: "integer",
}
require.NoError(t, parser.Init())
metric, err := parser.ParseLine("55")
require.NoError(t, err)
require.Equal(t, "value_test", metric.Name())
require.Equal(t, map[string]interface{}{
"value": int64(55),
}, metric.Fields())
require.Equal(t, map[string]string{}, metric.Tags())
parser = Parser{
MetricName: "value_test",
DataType: "float",
}
require.NoError(t, parser.Init())
metric, err = parser.ParseLine("64")
require.NoError(t, err)
require.Equal(t, "value_test", metric.Name())
require.Equal(t, map[string]interface{}{
"value": float64(64),
}, metric.Fields())
require.Equal(t, map[string]string{}, metric.Tags())
parser = Parser{
MetricName: "value_test",
DataType: "string",
}
require.NoError(t, parser.Init())
metric, err = parser.ParseLine("foobar")
require.NoError(t, err)
require.Equal(t, "value_test", metric.Name())
require.Equal(t, map[string]interface{}{
"value": "foobar",
}, metric.Fields())
require.Equal(t, map[string]string{}, metric.Tags())
parser = Parser{
MetricName: "value_test",
DataType: "boolean",
}
require.NoError(t, parser.Init())
metric, err = parser.ParseLine("true")
require.NoError(t, err)
require.Equal(t, "value_test", metric.Name())
require.Equal(t, map[string]interface{}{
"value": true,
}, metric.Fields())
require.Equal(t, map[string]string{}, metric.Tags())
require.Equal(t, map[string]interface{}{"penguin": int64(55)}, metrics[0].Fields())
}
func TestParseInvalidValues(t *testing.T) {
parser := Parser{
MetricName: "value_test",
DataType: "integer",
tests := []struct {
name string
dtype string
input []byte
}{
{
name: "integer",
dtype: "integer",
input: []byte("55.0"),
},
{
name: "float",
dtype: "float",
input: []byte("foobar"),
},
{
name: "boolean",
dtype: "boolean",
input: []byte("213"),
},
}
require.NoError(t, parser.Init())
metrics, err := parser.Parse([]byte("55.0"))
require.Error(t, err)
require.Len(t, metrics, 0)
parser = Parser{
MetricName: "value_test",
DataType: "float",
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
plugin := Parser{
MetricName: "value_test",
DataType: tt.dtype,
}
require.NoError(t, plugin.Init())
actual, err := plugin.Parse(tt.input)
require.ErrorContains(t, err, "invalid syntax")
require.Empty(t, actual)
})
}
require.NoError(t, parser.Init())
metrics, err = parser.Parse([]byte("foobar"))
require.Error(t, err)
require.Len(t, metrics, 0)
parser = Parser{
MetricName: "value_test",
DataType: "boolean",
}
require.NoError(t, parser.Init())
metrics, err = parser.Parse([]byte("213"))
require.Error(t, err)
require.Len(t, metrics, 0)
}
func TestParseLineInvalidValues(t *testing.T) {
parser := Parser{
MetricName: "value_test",
DataType: "integer",
tests := []struct {
name string
dtype string
input string
}{
{
name: "integer",
dtype: "integer",
input: "55.0",
},
{
name: "float",
dtype: "float",
input: "foobar",
},
{
name: "boolean",
dtype: "boolean",
input: "213",
},
}
require.NoError(t, parser.Init())
_, err := parser.ParseLine("55.0")
require.Error(t, err)
parser = Parser{
MetricName: "value_test",
DataType: "float",
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
plugin := Parser{
MetricName: "value_test",
DataType: tt.dtype,
}
require.NoError(t, plugin.Init())
actual, err := plugin.ParseLine(tt.input)
require.ErrorContains(t, err, "invalid syntax")
require.Empty(t, actual)
})
}
require.NoError(t, parser.Init())
_, err = parser.ParseLine("foobar")
require.Error(t, err)
parser = Parser{
MetricName: "value_test",
DataType: "boolean",
}
require.NoError(t, parser.Init())
_, err = parser.ParseLine("213")
require.Error(t, err)
}
func TestParseValidValuesDefaultTags(t *testing.T) {
parser := Parser{
expected := metric.New(
"value_test",
map[string]string{"test": "tag"},
map[string]interface{}{"value": int64(55)},
time.Unix(0, 0),
)
plugin := Parser{
MetricName: "value_test",
DataType: "integer",
}
require.NoError(t, parser.Init())
parser.SetDefaultTags(map[string]string{"test": "tag"})
metrics, err := parser.Parse([]byte("55"))
require.NoError(t, err)
require.Len(t, metrics, 1)
require.Equal(t, "value_test", metrics[0].Name())
require.Equal(t, map[string]interface{}{
"value": int64(55),
}, metrics[0].Fields())
require.Equal(t, map[string]string{"test": "tag"}, metrics[0].Tags())
require.NoError(t, plugin.Init())
plugin.SetDefaultTags(map[string]string{"test": "tag"})
parser = Parser{
MetricName: "value_test",
DataType: "float",
}
require.NoError(t, parser.Init())
parser.SetDefaultTags(map[string]string{"test": "tag"})
metrics, err = parser.Parse([]byte("64"))
actual, err := plugin.Parse([]byte("55"))
require.NoError(t, err)
require.Len(t, metrics, 1)
require.Equal(t, "value_test", metrics[0].Name())
require.Equal(t, map[string]interface{}{
"value": float64(64),
}, metrics[0].Fields())
require.Equal(t, map[string]string{"test": "tag"}, metrics[0].Tags())
require.Len(t, actual, 1)
parser = Parser{
MetricName: "value_test",
DataType: "string",
}
require.NoError(t, parser.Init())
parser.SetDefaultTags(map[string]string{"test": "tag"})
metrics, err = parser.Parse([]byte("foobar"))
require.NoError(t, err)
require.Len(t, metrics, 1)
require.Equal(t, "value_test", metrics[0].Name())
require.Equal(t, map[string]interface{}{
"value": "foobar",
}, metrics[0].Fields())
require.Equal(t, map[string]string{"test": "tag"}, metrics[0].Tags())
parser = Parser{
MetricName: "value_test",
DataType: "boolean",
}
require.NoError(t, parser.Init())
parser.SetDefaultTags(map[string]string{"test": "tag"})
metrics, err = parser.Parse([]byte("true"))
require.NoError(t, err)
require.Len(t, metrics, 1)
require.Equal(t, "value_test", metrics[0].Name())
require.Equal(t, map[string]interface{}{
"value": true,
}, metrics[0].Fields())
require.Equal(t, map[string]string{"test": "tag"}, metrics[0].Tags())
testutil.RequireMetricEqual(t, expected, actual[0], testutil.IgnoreTime())
}
func TestParseValuesWithNullCharacter(t *testing.T) {
@ -286,3 +290,11 @@ func TestParseValuesWithNullCharacter(t *testing.T) {
}, metrics[0].Fields())
require.Equal(t, map[string]string{}, metrics[0].Tags())
}
func TestInvalidDatatype(t *testing.T) {
parser := Parser{
MetricName: "value_test",
DataType: "foo",
}
require.ErrorContains(t, parser.Init(), "unknown datatype")
}