Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: migrate value parser to new style #11407

Merged
merged 2 commits into from
Jun 29, 2022
Merged

feat: migrate value parser to new style #11407

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
412c82a
feat: migrate value parser to new style
powersj Jun 28, 2022
f7bf0ff
post-review changes
powersj Jun 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 26 additions & 13 deletions plugins/inputs/exec/exec_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,8 @@ import (

"github.com/stretchr/testify/require"

"github.com/influxdata/telegraf/plugins/parsers"
"github.com/influxdata/telegraf/plugins/parsers/json"
"github.com/influxdata/telegraf/plugins/parsers/value"
"github.com/influxdata/telegraf/testutil"
)

Expand Down Expand Up @@ -139,11 +139,15 @@ func TestCommandError(t *testing.T) {
}

func TestExecCommandWithGlob(t *testing.T) {
parser, err := parsers.NewValueParser("metric", "string", "", nil)
require.NoError(t, err)
parser := value.Parser{
MetricName: "metric",
DataType: "string",
}
require.NoError(t, parser.Init())

e := NewExec()
e.Commands = []string{"/bin/ech* metric_value"}
e.SetParser(parser)
e.parser = &parser
srebhan marked this conversation as resolved.
Show resolved Hide resolved

var acc testutil.Accumulator
require.NoError(t, acc.GatherError(e.Gather))
Expand All @@ -155,12 +159,15 @@ func TestExecCommandWithGlob(t *testing.T) {
}

func TestExecCommandWithoutGlob(t *testing.T) {
parser, err := parsers.NewValueParser("metric", "string", "", nil)
require.NoError(t, err)
parser := value.Parser{
MetricName: "metric",
DataType: "string",
}
require.NoError(t, parser.Init())

e := NewExec()
e.Commands = []string{"/bin/echo metric_value"}
e.SetParser(parser)
e.parser = &parser

var acc testutil.Accumulator
require.NoError(t, acc.GatherError(e.Gather))
Expand All @@ -172,11 +179,14 @@ func TestExecCommandWithoutGlob(t *testing.T) {
}

func TestExecCommandWithoutGlobAndPath(t *testing.T) {
parser, err := parsers.NewValueParser("metric", "string", "", nil)
require.NoError(t, err)
parser := value.Parser{
MetricName: "metric",
DataType: "string",
}
require.NoError(t, parser.Init())
e := NewExec()
e.Commands = []string{"echo metric_value"}
e.SetParser(parser)
e.parser = &parser

var acc testutil.Accumulator
require.NoError(t, acc.GatherError(e.Gather))
Expand All @@ -188,12 +198,15 @@ func TestExecCommandWithoutGlobAndPath(t *testing.T) {
}

func TestExecCommandWithEnv(t *testing.T) {
parser, err := parsers.NewValueParser("metric", "string", "", nil)
require.NoError(t, err)
parser := value.Parser{
MetricName: "metric",
DataType: "string",
}
require.NoError(t, parser.Init())
e := NewExec()
e.Commands = []string{"/bin/sh -c 'echo ${METRIC_NAME}'"}
e.Environment = []string{"METRIC_NAME=metric_value"}
e.SetParser(parser)
e.parser = &parser

var acc testutil.Accumulator
require.NoError(t, acc.GatherError(e.Gather))
Expand Down
8 changes: 7 additions & 1 deletion plugins/inputs/http/http_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (
"github.com/influxdata/telegraf/plugins/parsers"
"github.com/influxdata/telegraf/plugins/parsers/csv"
"github.com/influxdata/telegraf/plugins/parsers/json"
"github.com/influxdata/telegraf/plugins/parsers/value"
"github.com/influxdata/telegraf/testutil"
)

Expand Down Expand Up @@ -374,7 +375,12 @@ func TestOAuthClientCredentialsGrant(t *testing.T) {
})

tt.plugin.SetParserFunc(func() (telegraf.Parser, error) {
return parsers.NewValueParser("metric", "string", "", nil)
p := &value.Parser{
MetricName: "metric",
DataType: "string",
}
err := p.Init()
return p, err
})

err = tt.plugin.Init()
Expand Down
21 changes: 15 additions & 6 deletions plugins/inputs/kafka_consumer/kafka_consumer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,8 +271,11 @@ func (c *FakeConsumerGroupClaim) Messages() <-chan *sarama.ConsumerMessage {

func TestConsumerGroupHandler_Lifecycle(t *testing.T) {
acc := &testutil.Accumulator{}
parser := value.NewValueParser("cpu", "int", "", nil)
cg := NewConsumerGroupHandler(acc, 1, parser, testutil.Logger{})
parser := value.Parser{
MetricName: "cpu",
DataType: "int",
}
cg := NewConsumerGroupHandler(acc, 1, &parser, testutil.Logger{})
srebhan marked this conversation as resolved.
Show resolved Hide resolved

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
Expand Down Expand Up @@ -301,8 +304,11 @@ func TestConsumerGroupHandler_Lifecycle(t *testing.T) {

func TestConsumerGroupHandler_ConsumeClaim(t *testing.T) {
acc := &testutil.Accumulator{}
parser := value.NewValueParser("cpu", "int", "", nil)
cg := NewConsumerGroupHandler(acc, 1, parser, testutil.Logger{})
parser := value.Parser{
MetricName: "cpu",
DataType: "int",
}
cg := NewConsumerGroupHandler(acc, 1, &parser, testutil.Logger{})

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
Expand Down Expand Up @@ -415,8 +421,11 @@ func TestConsumerGroupHandler_Handle(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
acc := &testutil.Accumulator{}
parser := value.NewValueParser("cpu", "int", "", nil)
cg := NewConsumerGroupHandler(acc, 1, parser, testutil.Logger{})
parser := value.Parser{
MetricName: "cpu",
DataType: "int",
}
cg := NewConsumerGroupHandler(acc, 1, &parser, testutil.Logger{})
cg.MaxMessageLen = tt.maxMessageLen
cg.TopicTag = tt.topicTag

Expand Down
1 change: 1 addition & 0 deletions plugins/parsers/all/all.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
_ "github.com/influxdata/telegraf/plugins/parsers/csv"
_ "github.com/influxdata/telegraf/plugins/parsers/json"
_ "github.com/influxdata/telegraf/plugins/parsers/json_v2"
_ "github.com/influxdata/telegraf/plugins/parsers/value"
_ "github.com/influxdata/telegraf/plugins/parsers/wavefront"
_ "github.com/influxdata/telegraf/plugins/parsers/xpath"
)
13 changes: 0 additions & 13 deletions plugins/parsers/registry.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ import (
"github.com/influxdata/telegraf/plugins/parsers/prometheusremotewrite"
"github.com/influxdata/telegraf/plugins/parsers/temporary/json_v2"
"github.com/influxdata/telegraf/plugins/parsers/temporary/xpath"
"github.com/influxdata/telegraf/plugins/parsers/value"
)

// Creator is the function to create a new parser
Expand Down Expand Up @@ -203,9 +202,6 @@ func NewParser(config *Config) (Parser, error) {
var err error
var parser Parser
switch config.DataFormat {
case "value":
parser, err = NewValueParser(config.MetricName,
config.DataType, config.ValueFieldName, config.DefaultTags)
case "influx":
if config.InfluxParserType == "upstream" {
parser, err = NewInfluxUpstreamParser()
Expand Down Expand Up @@ -308,15 +304,6 @@ func NewGraphiteParser(
return graphite.NewGraphiteParser(separator, templates, defaultTags)
}

func NewValueParser(
metricName string,
dataType string,
fieldName string,
defaultTags map[string]string,
) (Parser, error) {
return value.NewValueParser(metricName, dataType, fieldName, defaultTags), nil
}

func NewDropwizardParser(
metricRegistryPath string,
timePath string,
Expand Down
52 changes: 35 additions & 17 deletions plugins/parsers/value/parser.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,16 +9,21 @@ import (

"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/plugins/parsers"
)

type ValueParser struct {
MetricName string
DataType string
DefaultTags map[string]string
FieldName string
type Parser struct {
DataType string `toml:"data_type"`
FieldName string `toml:"value_field_name"`
MetricName string `toml:"-"`
DefaultTags map[string]string `toml:"-"`
}

func (v *ValueParser) Parse(buf []byte) ([]telegraf.Metric, error) {
func (v *Parser) Parse(buf []byte) ([]telegraf.Metric, error) {
if v.FieldName == "" {
v.FieldName = "value"
}
srebhan marked this conversation as resolved.
Show resolved Hide resolved

vStr := string(bytes.TrimSpace(bytes.Trim(buf, "\x00")))

// unless it's a string, separate out any fields in the buffer,
Expand Down Expand Up @@ -54,7 +59,7 @@ func (v *ValueParser) Parse(buf []byte) ([]telegraf.Metric, error) {
return []telegraf.Metric{m}, nil
}

func (v *ValueParser) ParseLine(line string) (telegraf.Metric, error) {
func (v *Parser) ParseLine(line string) (telegraf.Metric, error) {
metrics, err := v.Parse([]byte(line))

if err != nil {
Expand All @@ -68,19 +73,32 @@ func (v *ValueParser) ParseLine(line string) (telegraf.Metric, error) {
return metrics[0], nil
}

func (v *ValueParser) SetDefaultTags(tags map[string]string) {
func (v *Parser) SetDefaultTags(tags map[string]string) {
v.DefaultTags = tags
}

func NewValueParser(metricName, dataType, fieldName string, defaultTags map[string]string) *ValueParser {
if fieldName == "" {
fieldName = "value"
}
// InitFromConfig is a compatibility function to construct the parser the old way
func (v *Parser) InitFromConfig(config *parsers.Config) error {
v.MetricName = config.MetricName
v.DefaultTags = config.DefaultTags
return v.Init()
}

return &ValueParser{
MetricName: metricName,
DataType: dataType,
DefaultTags: defaultTags,
FieldName: fieldName,
func (v *Parser) Init() error {
if v.FieldName == "" {
v.FieldName = "value"
}

return nil
}

func init() {
parsers.Add("value",
func(defaultMetricName string) telegraf.Parser {
return &Parser{
FieldName: "value",
MetricName: defaultMetricName,
}
},
)
}
Loading