dom/telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Refactor InfluxDB listener (#6974)

Browse Source 
Use streaming parser in InfluxDB listener
This commit is contained in:
reimda
2020-03-04 11:13:44 -07:00
committed by GitHub
parent ab8438dcc6
commit a0276385b1
21 changed files with 19898 additions and 18303 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
plugins/parsers/csv/parser.go
View File

@@ -13,6 +13,8 @@ import (
"github.com/influxdata/telegraf/metric"
)
type TimeFunc func() time.Time
type Parser struct {
MetricName string
HeaderRowCount int
@@ -31,7 +33,7 @@ type Parser struct {
TimeFunc func() time.Time
}
func (p *Parser) SetTimeFunc(fn metric.TimeFunc) {
func (p *Parser) SetTimeFunc(fn TimeFunc) {
p.TimeFunc = fn
}

6
plugins/parsers/dropwizard/parser.go
View File

@@ -17,6 +17,8 @@ import (
var fieldEscaper = strings.NewReplacer("\\", "\\\\", "\"", "\\\"")
var keyEscaper = strings.NewReplacer(" ", "\\ ", ",", "\\,", "=", "\\=")
type TimeFunc func() time.Time
// Parser parses json inputs containing dropwizard metrics,
// either top-level or embedded inside a json field.
// This parser is using gjson for retrieving paths within the json file.
@@ -48,7 +50,7 @@ type parser struct {
separator string
templateEngine *templating.Engine
timeFunc metric.TimeFunc
timeFunc TimeFunc
// seriesParser parses line protocol measurement + tags
seriesParser *influx.Parser
@@ -267,6 +269,6 @@ func (p *parser) readDWMetrics(metricType string, dwms interface{}, metrics []te
return metrics
}
func (p *parser) SetTimeFunc(f metric.TimeFunc) {
func (p *parser) SetTimeFunc(f TimeFunc) {
p.timeFunc = f
}

10
plugins/parsers/dropwizard/parser_test.go
View File

@@ -13,7 +13,7 @@ import (
"github.com/stretchr/testify/require"
)
var TimeFunc = func() time.Time {
var testTimeFunc = func() time.Time {
return time.Unix(0, 0)
}
@@ -528,7 +528,7 @@ func TestDropWizard(t *testing.T) {
map[string]interface{}{
"value": 42.0,
},
TimeFunc(),
testTimeFunc(),
),
),
},
@@ -547,7 +547,7 @@ func TestDropWizard(t *testing.T) {
map[string]interface{}{
"value": 42.0,
},
TimeFunc(),
testTimeFunc(),
),
),
},
@@ -573,7 +573,7 @@ func TestDropWizard(t *testing.T) {
map[string]interface{}{
"value": 42.0,
},
TimeFunc(),
testTimeFunc(),
),
),
},
@@ -584,7 +584,7 @@ func TestDropWizard(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
parser := NewParser()
parser.SetTimeFunc(TimeFunc)
parser.SetTimeFunc(testTimeFunc)
metrics, err := parser.Parse(tt.input)
tt.errFunc(t, err)

62
plugins/parsers/influx/handler.go
View File

@@ -10,43 +10,53 @@ import (
"github.com/influxdata/telegraf/metric"
)
// MetricHandler implements the Handler interface and produces telegraf.Metric.
type MetricHandler struct {
builder *metric.Builder
err error
precision time.Duration
err error
timePrecision time.Duration
timeFunc TimeFunc
metric telegraf.Metric
}
func NewMetricHandler() *MetricHandler {
return &MetricHandler{
builder: metric.NewBuilder(),
precision: time.Nanosecond,
timePrecision: time.Nanosecond,
timeFunc: time.Now,
}
}
func (h *MetricHandler) SetTimeFunc(f metric.TimeFunc) {
h.builder.TimeFunc = f
func (h *MetricHandler) SetTimePrecision(p time.Duration) {
h.timePrecision = p
// When the timestamp is omitted from the metric, the timestamp
// comes from the server clock, truncated to the nearest unit of
// measurement provided in precision.
//
// When a timestamp is provided in the metric, precsision is
// overloaded to hold the unit of measurement of the timestamp.
}
func (h *MetricHandler) SetTimePrecision(precision time.Duration) {
h.builder.TimePrecision = precision
h.precision = precision
func (h *MetricHandler) SetTimeFunc(f TimeFunc) {
h.timeFunc = f
}
func (h *MetricHandler) Metric() (telegraf.Metric, error) {
m, err := h.builder.Metric()
h.builder.Reset()
return m, err
if h.metric.Time().IsZero() {
h.metric.SetTime(h.timeFunc().Truncate(h.timePrecision))
}
return h.metric, nil
}
func (h *MetricHandler) SetMeasurement(name []byte) error {
h.builder.SetName(nameUnescape(name))
return nil
var err error
h.metric, err = metric.New(nameUnescape(name),
nil, nil, time.Time{})
return err
}
func (h *MetricHandler) AddTag(key []byte, value []byte) error {
tk := unescape(key)
tv := unescape(value)
h.builder.AddTag(tk, tv)
h.metric.AddTag(tk, tv)
return nil
}
@@ -59,7 +69,7 @@ func (h *MetricHandler) AddInt(key []byte, value []byte) error {
}
return err
}
h.builder.AddField(fk, fv)
h.metric.AddField(fk, fv)
return nil
}
@@ -72,7 +82,7 @@ func (h *MetricHandler) AddUint(key []byte, value []byte) error {
}
return err
}
h.builder.AddField(fk, fv)
h.metric.AddField(fk, fv)
return nil
}
@@ -85,14 +95,14 @@ func (h *MetricHandler) AddFloat(key []byte, value []byte) error {
}
return err
}
h.builder.AddField(fk, fv)
h.metric.AddField(fk, fv)
return nil
}
func (h *MetricHandler) AddString(key []byte, value []byte) error {
fk := unescape(key)
fv := stringFieldUnescape(value)
h.builder.AddField(fk, fv)
h.metric.AddField(fk, fv)
return nil
}
@@ -102,7 +112,7 @@ func (h *MetricHandler) AddBool(key []byte, value []byte) error {
if err != nil {
return errors.New("unparseable bool")
}
h.builder.AddField(fk, fv)
h.metric.AddField(fk, fv)
return nil
}
@@ -114,11 +124,9 @@ func (h *MetricHandler) SetTimestamp(tm []byte) error {
}
return err
}
ns := v * int64(h.precision)
h.builder.SetTime(time.Unix(0, ns))
//time precision is overloaded to mean time unit here
ns := v * int64(h.timePrecision)
h.metric.SetTime(time.Unix(0, ns))
return nil
}
func (h *MetricHandler) Reset() {
h.builder.Reset()
}

34919
plugins/parsers/influx/machine.go
View File

File diff suppressed because it is too large Load Diff

160
plugins/parsers/influx/machine.go.rl
View File

@@ -2,6 +2,7 @@ package influx
import (
"errors"
"io"
)
var (
@@ -70,8 +71,8 @@ action goto_align {
fgoto align;
}
action found_metric {
foundMetric = true
action begin_metric {
m.beginMetric = true
}
action name {
@@ -84,11 +85,11 @@ action name {
}
action tagkey {
key = m.text()
m.key = m.text()
}
action tagvalue {
err = m.handler.AddTag(key, m.text())
err = m.handler.AddTag(m.key, m.text())
if err != nil {
fhold;
fnext discard_line;
@@ -97,11 +98,11 @@ action tagvalue {
}
action fieldkey {
key = m.text()
m.key = m.text()
}
action integer {
err = m.handler.AddInt(key, m.text())
err = m.handler.AddInt(m.key, m.text())
if err != nil {
fhold;
fnext discard_line;
@@ -110,7 +111,7 @@ action integer {
}
action unsigned {
err = m.handler.AddUint(key, m.text())
err = m.handler.AddUint(m.key, m.text())
if err != nil {
fhold;
fnext discard_line;
@@ -119,7 +120,7 @@ action unsigned {
}
action float {
err = m.handler.AddFloat(key, m.text())
err = m.handler.AddFloat(m.key, m.text())
if err != nil {
fhold;
fnext discard_line;
@@ -128,7 +129,7 @@ action float {
}
action bool {
err = m.handler.AddBool(key, m.text())
err = m.handler.AddBool(m.key, m.text())
if err != nil {
fhold;
fnext discard_line;
@@ -137,7 +138,7 @@ action bool {
}
action string {
err = m.handler.AddString(key, m.text())
err = m.handler.AddString(m.key, m.text())
if err != nil {
fhold;
fnext discard_line;
@@ -161,15 +162,20 @@ action incr_newline {
}
action eol {
m.finishMetric = true
fnext align;
fbreak;
}
action finish_metric {
m.finishMetric = true
}
ws =
[\t\v\f ];
newline =
'\r'? '\n' %to(incr_newline);
'\r'? '\n' >incr_newline;
non_zero_digit =
[1-9];
@@ -273,7 +279,7 @@ line_without_term =
main :=
(line_with_term*
(line_with_term | line_without_term?)
) >found_metric
) >begin_metric %eof(finish_metric)
;
# The discard_line machine discards the current line. Useful for recovering
@@ -299,7 +305,7 @@ align :=
# Series is a machine for matching measurement+tagset
series :=
(measurement >err(name_error) tagset eol_break?)
>found_metric
>begin_metric
;
}%%
@@ -317,14 +323,17 @@ type Handler interface {
}
type machine struct {
data []byte
cs int
p, pe, eof int
pb int
lineno int
sol int
handler Handler
initState int
data []byte
cs int
p, pe, eof int
pb int
lineno int
sol int
handler Handler
initState int
key []byte
beginMetric bool
finishMetric bool
}
func NewMachine(handler Handler) *machine {
@@ -368,6 +377,9 @@ func (m *machine) SetData(data []byte) {
m.sol = 0
m.pe = len(data)
m.eof = len(data)
m.key = nil
m.beginMetric = false
m.finishMetric = false
%% write init;
m.cs = m.initState
@@ -382,10 +394,15 @@ func (m *machine) Next() error {
return EOF
}
var err error
var key []byte
foundMetric := false
m.key = nil
m.beginMetric = false
m.finishMetric = false
return m.exec()
}
func (m *machine) exec() error {
var err error
%% write exec;
if err != nil {
@@ -405,7 +422,7 @@ func (m *machine) Next() error {
//
// Otherwise we have successfully parsed a metric line, so if we are at
// the EOF we will report it the next call.
if !foundMetric && m.p == m.pe && m.pe == m.eof {
if !m.beginMetric && m.p == m.pe && m.pe == m.eof {
return EOF
}
@@ -437,3 +454,96 @@ func (m *machine) Column() int {
func (m *machine) text() []byte {
return m.data[m.pb:m.p]
}
type streamMachine struct {
machine *machine
reader io.Reader
}
func NewStreamMachine(r io.Reader, handler Handler) *streamMachine {
m := &streamMachine{
machine: NewMachine(handler),
reader: r,
}
m.machine.SetData(make([]byte, 1024))
m.machine.pe = 0
m.machine.eof = -1
return m
}
func (m *streamMachine) Next() error {
// Check if we are already at EOF, this should only happen if called again
// after already returning EOF.
if m.machine.p == m.machine.pe && m.machine.pe == m.machine.eof {
return EOF
}
copy(m.machine.data, m.machine.data[m.machine.p:])
m.machine.pe = m.machine.pe - m.machine.p
m.machine.sol = m.machine.sol - m.machine.p
m.machine.pb = 0
m.machine.p = 0
m.machine.eof = -1
m.machine.key = nil
m.machine.beginMetric = false
m.machine.finishMetric = false
for {
// Expand the buffer if it is full
if m.machine.pe == len(m.machine.data) {
expanded := make([]byte, 2 * len(m.machine.data))
copy(expanded, m.machine.data)
m.machine.data = expanded
}
n, err := m.reader.Read(m.machine.data[m.machine.pe:])
if n == 0 && err == io.EOF {
m.machine.eof = m.machine.pe
} else if err != nil && err != io.EOF {
return err
}
m.machine.pe += n
err = m.machine.exec()
if err != nil {
return err
}
// If we have successfully parsed a full metric line break out
if m.machine.finishMetric {
break
}
}
return nil
}
// Position returns the current byte offset into the data.
func (m *streamMachine) Position() int {
return m.machine.Position()
}
// LineOffset returns the byte offset of the current line.
func (m *streamMachine) LineOffset() int {
return m.machine.LineOffset()
}
// LineNumber returns the current line number. Lines are counted based on the
// regular expression `\r?\n`.
func (m *streamMachine) LineNumber() int {
return m.machine.LineNumber()
}
// Column returns the current column.
func (m *streamMachine) Column() int {
return m.machine.Column()
}
// LineText returns the text of the current line that has been parsed so far.
func (m *streamMachine) LineText() string {
return string(m.machine.data[0:m.machine.p])
}

508
plugins/parsers/influx/machine_test.go
View File

@@ -1,8 +1,10 @@
package influx_test
import (
"bytes"
"errors"
"fmt"
"io"
"testing"
"github.com/influxdata/telegraf/plugins/parsers/influx"
@@ -14,41 +16,59 @@ type TestingHandler struct {
}
func (h *TestingHandler) SetMeasurement(name []byte) error {
n := make([]byte, len(name))
copy(n, name)
mname := Result{
Name: Measurement,
Value: name,
Value: n,
}
h.results = append(h.results, mname)
return nil
}
func (h *TestingHandler) AddTag(key []byte, value []byte) error {
k := make([]byte, len(key))
copy(k, key)
v := make([]byte, len(value))
copy(v, value)
tagkey := Result{
Name: TagKey,
Value: key,
Value: k,
}
tagvalue := Result{
Name: TagValue,
Value: value,
Value: v,
}
h.results = append(h.results, tagkey, tagvalue)
return nil
}
func (h *TestingHandler) AddInt(key []byte, value []byte) error {
k := make([]byte, len(key))
copy(k, key)
v := make([]byte, len(value))
copy(v, value)
fieldkey := Result{
Name: FieldKey,
Value: key,
Value: k,
}
fieldvalue := Result{
Name: FieldInt,
Value: value,
Value: v,
}
h.results = append(h.results, fieldkey, fieldvalue)
return nil
}
func (h *TestingHandler) AddUint(key []byte, value []byte) error {
k := make([]byte, len(key))
copy(k, key)
v := make([]byte, len(value))
copy(v, value)
fieldkey := Result{
Name: FieldKey,
Value: key,
@@ -62,48 +82,66 @@ func (h *TestingHandler) AddUint(key []byte, value []byte) error {
}
func (h *TestingHandler) AddFloat(key []byte, value []byte) error {
k := make([]byte, len(key))
copy(k, key)
v := make([]byte, len(value))
copy(v, value)
fieldkey := Result{
Name: FieldKey,
Value: key,
Value: k,
}
fieldvalue := Result{
Name: FieldFloat,
Value: value,
Value: v,
}
h.results = append(h.results, fieldkey, fieldvalue)
return nil
}
func (h *TestingHandler) AddString(key []byte, value []byte) error {
k := make([]byte, len(key))
copy(k, key)
v := make([]byte, len(value))
copy(v, value)
fieldkey := Result{
Name: FieldKey,
Value: key,
Value: k,
}
fieldvalue := Result{
Name: FieldString,
Value: value,
Value: v,
}
h.results = append(h.results, fieldkey, fieldvalue)
return nil
}
func (h *TestingHandler) AddBool(key []byte, value []byte) error {
k := make([]byte, len(key))
copy(k, key)
v := make([]byte, len(value))
copy(v, value)
fieldkey := Result{
Name: FieldKey,
Value: key,
Value: k,
}
fieldvalue := Result{
Name: FieldBool,
Value: value,
Value: v,
}
h.results = append(h.results, fieldkey, fieldvalue)
return nil
}
func (h *TestingHandler) SetTimestamp(tm []byte) error {
t := make([]byte, len(tm))
copy(t, tm)
timestamp := Result{
Name: Timestamp,
Value: tm,
Value: t,
}
h.results = append(h.results, timestamp)
return nil
@@ -1676,63 +1714,64 @@ func TestMachine(t *testing.T) {
}
}
var positionTests = []struct {
name string
input []byte
lineno int
column int
}{
{
name: "empty string",
input: []byte(""),
lineno: 1,
column: 1,
},
{
name: "minimal",
input: []byte("cpu value=42"),
lineno: 1,
column: 13,
},
{
name: "one newline",
input: []byte("cpu value=42\ncpu value=42"),
lineno: 2,
column: 13,
},
{
name: "several newlines",
input: []byte("cpu value=42\n\n\n"),
lineno: 4,
column: 1,
},
{
name: "error on second line",
input: []byte("cpu value=42\ncpu value=invalid"),
lineno: 2,
column: 11,
},
{
name: "error after comment line",
input: []byte("cpu value=42\n# comment\ncpu value=invalid"),
lineno: 3,
column: 11,
},
{
name: "dos line endings",
input: []byte("cpu value=42\r\ncpu value=invalid"),
lineno: 2,
column: 11,
},
{
name: "mac line endings not supported",
input: []byte("cpu value=42\rcpu value=invalid"),
lineno: 1,
column: 14,
},
}
func TestMachinePosition(t *testing.T) {
var tests = []struct {
name string
input []byte
lineno int
column int
}{
{
name: "empty string",
input: []byte(""),
lineno: 1,
column: 1,
},
{
name: "minimal",
input: []byte("cpu value=42"),
lineno: 1,
column: 13,
},
{
name: "one newline",
input: []byte("cpu value=42\ncpu value=42"),
lineno: 2,
column: 13,
},
{
name: "several newlines",
input: []byte("cpu value=42\n\n\n"),
lineno: 4,
column: 1,
},
{
name: "error on second line",
input: []byte("cpu value=42\ncpu value=invalid"),
lineno: 2,
column: 11,
},
{
name: "error after comment line",
input: []byte("cpu value=42\n# comment\ncpu value=invalid"),
lineno: 3,
column: 11,
},
{
name: "dos line endings",
input: []byte("cpu value=42\r\ncpu value=invalid"),
lineno: 2,
column: 11,
},
{
name: "mac line endings not supported",
input: []byte("cpu value=42\rcpu value=invalid"),
lineno: 1,
column: 14,
},
}
for _, tt := range tests {
for _, tt := range positionTests {
t.Run(tt.name, func(t *testing.T) {
handler := &TestingHandler{}
fsm := influx.NewMachine(handler)
@@ -1932,135 +1971,136 @@ func (h *MockHandler) SetTimestamp(tm []byte) error {
return h.SetTimestampF(tm)
}
var errorRecoveryTests = []struct {
name string
input []byte
handler *MockHandler
results []Result
}{
{
name: "integer",
input: []byte("cpu value=43i\ncpu value=42i"),
handler: &MockHandler{
SetMeasurementF: func(name []byte) error {
return nil
},
AddIntF: func(name, value []byte) error {
if string(value) != "42i" {
return errors.New("handler error")
}
return nil
},
},
results: []Result{
{
Name: Measurement,
Value: []byte("cpu"),
},
{
Name: Error,
err: errors.New("handler error"),
},
{
Name: Measurement,
Value: []byte("cpu"),
},
{
Name: FieldKey,
Value: []byte("value"),
},
{
Name: FieldInt,
Value: []byte("42i"),
},
{
Name: Success,
},
},
},
{
name: "integer with timestamp",
input: []byte("cpu value=43i 1516241192000000000\ncpu value=42i"),
handler: &MockHandler{
SetMeasurementF: func(name []byte) error {
return nil
},
AddIntF: func(name, value []byte) error {
if string(value) != "42i" {
return errors.New("handler error")
}
return nil
},
},
results: []Result{
{
Name: Measurement,
Value: []byte("cpu"),
},
{
Name: Error,
err: errors.New("handler error"),
},
{
Name: Measurement,
Value: []byte("cpu"),
},
{
Name: FieldKey,
Value: []byte("value"),
},
{
Name: FieldInt,
Value: []byte("42i"),
},
{
Name: Success,
},
},
},
{
name: "unsigned",
input: []byte("cpu value=43u\ncpu value=42u"),
handler: &MockHandler{
SetMeasurementF: func(name []byte) error {
return nil
},
AddUintF: func(name, value []byte) error {
if string(value) != "42u" {
return errors.New("handler error")
}
return nil
},
},
results: []Result{
{
Name: Measurement,
Value: []byte("cpu"),
},
{
Name: Error,
err: errors.New("handler error"),
},
{
Name: Measurement,
Value: []byte("cpu"),
},
{
Name: FieldKey,
Value: []byte("value"),
},
{
Name: FieldUint,
Value: []byte("42u"),
},
{
Name: Success,
},
},
},
}
func TestHandlerErrorRecovery(t *testing.T) {
var tests = []struct {
name string
input []byte
handler *MockHandler
results []Result
}{
{
name: "integer",
input: []byte("cpu value=43i\ncpu value=42i"),
handler: &MockHandler{
SetMeasurementF: func(name []byte) error {
return nil
},
AddIntF: func(name, value []byte) error {
if string(value) != "42i" {
return errors.New("handler error")
}
return nil
},
},
results: []Result{
{
Name: Measurement,
Value: []byte("cpu"),
},
{
Name: Error,
err: errors.New("handler error"),
},
{
Name: Measurement,
Value: []byte("cpu"),
},
{
Name: FieldKey,
Value: []byte("value"),
},
{
Name: FieldInt,
Value: []byte("42i"),
},
{
Name: Success,
},
},
},
{
name: "integer with timestamp",
input: []byte("cpu value=43i 1516241192000000000\ncpu value=42i"),
handler: &MockHandler{
SetMeasurementF: func(name []byte) error {
return nil
},
AddIntF: func(name, value []byte) error {
if string(value) != "42i" {
return errors.New("handler error")
}
return nil
},
},
results: []Result{
{
Name: Measurement,
Value: []byte("cpu"),
},
{
Name: Error,
err: errors.New("handler error"),
},
{
Name: Measurement,
Value: []byte("cpu"),
},
{
Name: FieldKey,
Value: []byte("value"),
},
{
Name: FieldInt,
Value: []byte("42i"),
},
{
Name: Success,
},
},
},
{
name: "unsigned",
input: []byte("cpu value=43u\ncpu value=42u"),
handler: &MockHandler{
SetMeasurementF: func(name []byte) error {
return nil
},
AddUintF: func(name, value []byte) error {
if string(value) != "42u" {
return errors.New("handler error")
}
return nil
},
},
results: []Result{
{
Name: Measurement,
Value: []byte("cpu"),
},
{
Name: Error,
err: errors.New("handler error"),
},
{
Name: Measurement,
Value: []byte("cpu"),
},
{
Name: FieldKey,
Value: []byte("value"),
},
{
Name: FieldUint,
Value: []byte("42u"),
},
{
Name: Success,
},
},
},
}
for _, tt := range tests {
for _, tt := range errorRecoveryTests {
t.Run(tt.name, func(t *testing.T) {
fsm := influx.NewMachine(tt.handler)
fsm.SetData(tt.input)
@@ -2078,3 +2118,79 @@ func TestHandlerErrorRecovery(t *testing.T) {
})
}
}
func TestStreamMachine(t *testing.T) {
type testcase struct {
name string
input io.Reader
results []Result
err error
}
var tc []testcase
for _, tt := range tests {
tc = append(tc, testcase{
name: tt.name,
input: bytes.NewBuffer([]byte(tt.input)),
results: tt.results,
err: tt.err,
})
}
for _, tt := range tc {
t.Run(tt.name, func(t *testing.T) {
handler := &TestingHandler{}
fsm := influx.NewStreamMachine(tt.input, handler)
// Parse only up to 20 metrics; to avoid any bugs where the parser
// isn't terminated.
for i := 0; i < 20; i++ {
err := fsm.Next()
if err != nil && err == influx.EOF {
break
}
handler.Result(err)
}
results := handler.Results()
require.Equal(t, tt.results, results)
})
}
}
func TestStreamMachinePosition(t *testing.T) {
type testcase struct {
name string
input io.Reader
lineno int
column int
}
var tc []testcase
for _, tt := range positionTests {
tc = append(tc, testcase{
name: tt.name,
input: bytes.NewBuffer([]byte(tt.input)),
lineno: tt.lineno,
column: tt.column,
})
}
for _, tt := range tc {
t.Run(tt.name, func(t *testing.T) {
handler := &TestingHandler{}
fsm := influx.NewStreamMachine(tt.input, handler)
// Parse until an error or eof
for i := 0; i < 20; i++ {
err := fsm.Next()
if err != nil {
break
}
}
require.Equal(t, tt.lineno, fsm.LineNumber(), "lineno")
require.Equal(t, tt.column, fsm.Column(), "column")
})
}
}

104
plugins/parsers/influx/parser.go
View File

@@ -3,8 +3,10 @@ package influx
import (
"errors"
"fmt"
"io"
"strings"
"sync"
"time"
"github.com/influxdata/telegraf"
)
@@ -17,6 +19,9 @@ var (
ErrNoMetric = errors.New("no metric in line")
)
type TimeFunc func() time.Time
// ParseError indicates a error in the parsing of the text.
type ParseError struct {
Offset int
LineOffset int
@@ -38,6 +43,8 @@ func (e *ParseError) Error() string {
return fmt.Sprintf("metric parse error: %s at %d:%d: %q", e.msg, e.LineNumber, e.Column, buffer)
}
// Parser is an InfluxDB Line Protocol parser that implements the
// parsers.Parser interface.
type Parser struct {
DefaultTags map[string]string
@@ -62,6 +69,10 @@ func NewSeriesParser(handler *MetricHandler) *Parser {
}
}
func (h *Parser) SetTimeFunc(f TimeFunc) {
h.handler.SetTimeFunc(f)
}
func (p *Parser) Parse(input []byte) ([]telegraf.Metric, error) {
p.Lock()
defer p.Unlock()
@@ -75,7 +86,6 @@ func (p *Parser) Parse(input []byte) ([]telegraf.Metric, error) {
}
if err != nil {
p.handler.Reset()
return nil, &ParseError{
Offset: p.machine.Position(),
LineOffset: p.machine.LineOffset(),
@@ -88,7 +98,6 @@ func (p *Parser) Parse(input []byte) ([]telegraf.Metric, error) {
metric, err := p.handler.Metric()
if err != nil {
p.handler.Reset()
return nil, err
}
@@ -126,10 +135,93 @@ func (p *Parser) applyDefaultTags(metrics []telegraf.Metric) {
}
for _, m := range metrics {
for k, v := range p.DefaultTags {
if !m.HasTag(k) {
m.AddTag(k, v)
}
p.applyDefaultTagsSingle(m)
}
}
func (p *Parser) applyDefaultTagsSingle(metric telegraf.Metric) {
for k, v := range p.DefaultTags {
if !metric.HasTag(k) {
metric.AddTag(k, v)
}
}
}
// StreamParser is an InfluxDB Line Protocol parser. It is not safe for
// concurrent use in multiple goroutines.
type StreamParser struct {
machine *streamMachine
handler *MetricHandler
}
func NewStreamParser(r io.Reader) *StreamParser {
handler := NewMetricHandler()
return &StreamParser{
machine: NewStreamMachine(r, handler),
handler: handler,
}
}
// SetTimeFunc changes the function used to determine the time of metrics
// without a timestamp. The default TimeFunc is time.Now. Useful mostly for
// testing, or perhaps if you want all metrics to have the same timestamp.
func (h *StreamParser) SetTimeFunc(f TimeFunc) {
h.handler.SetTimeFunc(f)
}
func (h *StreamParser) SetTimePrecision(u time.Duration) {
h.handler.SetTimePrecision(u)
}
// Next parses the next item from the stream. You can repeat calls to this
// function until it returns EOF.
func (p *StreamParser) Next() (telegraf.Metric, error) {
err := p.machine.Next()
if err == EOF {
return nil, EOF
}
if err != nil {
return nil, &ParseError{
Offset: p.machine.Position(),
LineOffset: p.machine.LineOffset(),
LineNumber: p.machine.LineNumber(),
Column: p.machine.Column(),
msg: err.Error(),
buf: p.machine.LineText(),
}
}
metric, err := p.handler.Metric()
if err != nil {
return nil, err
}
return metric, nil
}
// Position returns the current byte offset into the data.
func (p *StreamParser) Position() int {
return p.machine.Position()
}
// LineOffset returns the byte offset of the current line.
func (p *StreamParser) LineOffset() int {
return p.machine.LineOffset()
}
// LineNumber returns the current line number. Lines are counted based on the
// regular expression `\r?\n`.
func (p *StreamParser) LineNumber() int {
return p.machine.LineNumber()
}
// Column returns the current column.
func (p *StreamParser) Column() int {
return p.machine.Column()
}
// LineText returns the text of the current line that has been parsed so far.
func (p *StreamParser) LineText() string {
return p.machine.LineText()
}

162
plugins/parsers/influx/parser_test.go
View File

@@ -1,6 +1,7 @@
package influx
import (
"bytes"
"strconv"
"strings"
"testing"
@@ -8,6 +9,7 @@ import (
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/require"
)
@@ -23,12 +25,11 @@ var DefaultTime = func() time.Time {
}
var ptests = []struct {
name string
input []byte
timeFunc func() time.Time
precision time.Duration
metrics []telegraf.Metric
err error
name string
input []byte
timeFunc func() time.Time
metrics []telegraf.Metric
err error
}{
{
name: "minimal",
@@ -495,7 +496,7 @@ var ptests = []struct {
err: nil,
},
{
name: "no timestamp full precision",
name: "no timestamp",
input: []byte("cpu value=42"),
timeFunc: func() time.Time {
return time.Unix(42, 123456789)
@@ -514,27 +515,6 @@ var ptests = []struct {
},
err: nil,
},
{
name: "no timestamp partial precision",
input: []byte("cpu value=42"),
timeFunc: func() time.Time {
return time.Unix(42, 123456789)
},
precision: 1 * time.Millisecond,
metrics: []telegraf.Metric{
Metric(
metric.New(
"cpu",
map[string]string{},
map[string]interface{}{
"value": 42.0,
},
time.Unix(42, 123000000),
),
),
},
err: nil,
},
{
name: "multiple lines",
input: []byte("cpu value=42\ncpu value=42"),
@@ -651,14 +631,11 @@ func TestParser(t *testing.T) {
for _, tt := range ptests {
t.Run(tt.name, func(t *testing.T) {
handler := NewMetricHandler()
handler.SetTimeFunc(DefaultTime)
if tt.timeFunc != nil {
handler.SetTimeFunc(tt.timeFunc)
}
if tt.precision > 0 {
handler.SetTimePrecision(tt.precision)
}
parser := NewParser(handler)
parser.SetTimeFunc(DefaultTime)
if tt.timeFunc != nil {
parser.SetTimeFunc(tt.timeFunc)
}
metrics, err := parser.Parse(tt.input)
require.Equal(t, tt.err, err)
@@ -688,14 +665,41 @@ func BenchmarkParser(b *testing.B) {
}
}
func TestStreamParser(t *testing.T) {
for _, tt := range ptests {
t.Run(tt.name, func(t *testing.T) {
r := bytes.NewBuffer(tt.input)
parser := NewStreamParser(r)
parser.SetTimeFunc(DefaultTime)
if tt.timeFunc != nil {
parser.SetTimeFunc(tt.timeFunc)
}
var i int
for {
m, err := parser.Next()
if err != nil {
if err == EOF {
break
}
require.Equal(t, tt.err, err)
break
}
testutil.RequireMetricEqual(t, tt.metrics[i], m)
i++
}
})
}
}
func TestSeriesParser(t *testing.T) {
var tests = []struct {
name string
input []byte
timeFunc func() time.Time
precision time.Duration
metrics []telegraf.Metric
err error
name string
input []byte
timeFunc func() time.Time
metrics []telegraf.Metric
err error
}{
{
name: "empty",
@@ -749,14 +753,10 @@ func TestSeriesParser(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
handler := NewMetricHandler()
handler.SetTimeFunc(DefaultTime)
if tt.timeFunc != nil {
handler.SetTimeFunc(tt.timeFunc)
}
if tt.precision > 0 {
handler.SetTimePrecision(tt.precision)
}
parser := NewSeriesParser(handler)
if tt.timeFunc != nil {
parser.SetTimeFunc(tt.timeFunc)
}
metrics, err := parser.Parse(tt.input)
require.Equal(t, tt.err, err)
@@ -791,6 +791,11 @@ func TestParserErrorString(t *testing.T) {
input: []byte("cpu " + strings.Repeat("ab", maxErrorBufferSize) + "=invalid\ncpu value=42"),
errString: "metric parse error: expected field at 1:2054: \"cpu " + strings.Repeat("ab", maxErrorBufferSize)[:maxErrorBufferSize-4] + "...\"",
},
{
name: "multiple line error",
input: []byte("cpu value=42\ncpu value=invalid\ncpu value=42\ncpu value=invalid"),
errString: `metric parse error: expected field at 2:11: "cpu value=invalid"`,
},
}
for _, tt := range ptests {
@@ -803,3 +808,64 @@ func TestParserErrorString(t *testing.T) {
})
}
}
func TestStreamParserErrorString(t *testing.T) {
var ptests = []struct {
name string
input []byte
errs []string
}{
{
name: "multiple line error",
input: []byte("cpu value=42\ncpu value=invalid\ncpu value=42"),
errs: []string{
`metric parse error: expected field at 2:11: "cpu value="`,
},
},
{
name: "handler error",
input: []byte("cpu value=9223372036854775808i\ncpu value=42"),
errs: []string{
`metric parse error: value out of range at 1:31: "cpu value=9223372036854775808i"`,
},
},
{
name: "buffer too long",
input: []byte("cpu " + strings.Repeat("ab", maxErrorBufferSize) + "=invalid\ncpu value=42"),
errs: []string{
"metric parse error: expected field at 1:2054: \"cpu " + strings.Repeat("ab", maxErrorBufferSize)[:maxErrorBufferSize-4] + "...\"",
},
},
{
name: "multiple errors",
input: []byte("foo value=1asdf2.0\nfoo value=2.0\nfoo value=3asdf2.0\nfoo value=4.0"),
errs: []string{
`metric parse error: expected field at 1:12: "foo value=1"`,
`metric parse error: expected field at 3:12: "foo value=3"`,
},
},
}
for _, tt := range ptests {
t.Run(tt.name, func(t *testing.T) {
parser := NewStreamParser(bytes.NewBuffer(tt.input))
var errs []error
for i := 0; i < 20; i++ {
_, err := parser.Next()
if err == EOF {
break
}
if err != nil {
errs = append(errs, err)
}
}
require.Equal(t, len(tt.errs), len(errs))
for i, err := range errs {
require.Equal(t, tt.errs[i], err.Error())
}
})
}
}