Compare commits

..

4 Commits

Author SHA1 Message Date
Greg Linton
f88f665cd9 Add missing bracket 2018-06-28 11:49:07 -06:00
Ayrdrie
a2df042d92 Merge branch 'master' into logparser-divide-by-zero 2018-06-22 11:49:51 -06:00
Ayrdrie Palmer
c7a72b9a9d Update test for divide by zero error 2018-06-21 22:41:04 -06:00
Ayrdrie Palmer
09f884b4f0 Fix for divide by zero error by defaulting to current year if no year is given in timestamp 2018-06-21 22:15:33 -06:00
17 changed files with 53 additions and 133 deletions

View File

@@ -104,10 +104,9 @@ but can be overridden using the `name_override` config option.
#### JSON Configuration: #### JSON Configuration:
The JSON data format supports specifying "tag keys" and "field keys". If specified, keys The JSON data format supports specifying "tag keys". If specified, keys
will be searched for in the root-level and any nested lists of the JSON blob. If the key(s) exist, will be searched for in the root-level of the JSON blob. If the key(s) exist,
they will be applied as tags or fields to the Telegraf metrics. If "field_keys" is not specified, they will be applied as tags to the Telegraf metrics.
all int and float values will be set as fields by default.
For example, if you had this configuration: For example, if you had this configuration:
@@ -174,7 +173,6 @@ For example, if the following configuration:
"my_tag_1", "my_tag_1",
"my_tag_2" "my_tag_2"
] ]
field_keys = ["b_c"]
``` ```
with this JSON output from a command: with this JSON output from a command:
@@ -200,11 +198,11 @@ with this JSON output from a command:
] ]
``` ```
Your Telegraf metrics would get tagged with "my_tag_1" and "my_tag_2" and fielded with "b_c" Your Telegraf metrics would get tagged with "my_tag_1" and "my_tag_2"
``` ```
exec_mycollector,my_tag_1=foo,my_tag_2=baz b_c=6 exec_mycollector,my_tag_1=foo,my_tag_2=baz a=5,b_c=6
exec_mycollector,my_tag_1=bar,my_tag_2=baz b_c=8 exec_mycollector,my_tag_1=bar,my_tag_2=baz a=7,b_c=8
``` ```
# Value: # Value:

View File

@@ -1261,18 +1261,6 @@ func buildParser(name string, tbl *ast.Table) (parsers.Parser, error) {
} }
} }
if node, ok := tbl.Fields["field_keys"]; ok {
if kv, ok := node.(*ast.KeyValue); ok {
if ary, ok := kv.Value.(*ast.Array); ok {
for _, elem := range ary.Value {
if str, ok := elem.(*ast.String); ok {
c.FieldKeys = append(c.FieldKeys, str.Value)
}
}
}
}
}
if node, ok := tbl.Fields["data_type"]; ok { if node, ok := tbl.Fields["data_type"]; ok {
if kv, ok := node.(*ast.KeyValue); ok { if kv, ok := node.(*ast.KeyValue); ok {
if str, ok := kv.Value.(*ast.String); ok { if str, ok := kv.Value.(*ast.String); ok {
@@ -1356,7 +1344,6 @@ func buildParser(name string, tbl *ast.Table) (parsers.Parser, error) {
delete(tbl.Fields, "separator") delete(tbl.Fields, "separator")
delete(tbl.Fields, "templates") delete(tbl.Fields, "templates")
delete(tbl.Fields, "tag_keys") delete(tbl.Fields, "tag_keys")
delete(tbl.Fields, "field_keys")
delete(tbl.Fields, "data_type") delete(tbl.Fields, "data_type")
delete(tbl.Fields, "collectd_auth_file") delete(tbl.Fields, "collectd_auth_file")
delete(tbl.Fields, "collectd_security_level") delete(tbl.Fields, "collectd_security_level")

View File

@@ -143,10 +143,7 @@ func TestConfig_LoadDirectory(t *testing.T) {
"Testdata did not produce correct memcached metadata.") "Testdata did not produce correct memcached metadata.")
ex := inputs.Inputs["exec"]().(*exec.Exec) ex := inputs.Inputs["exec"]().(*exec.Exec)
p, err := parsers.NewParser(&parsers.Config{ p, err := parsers.NewJSONParser("exec", nil, nil)
MetricName: "exec",
DataFormat: "json",
})
assert.NoError(t, err) assert.NoError(t, err)
ex.SetParser(p) ex.SetParser(p)
ex.Command = "/usr/bin/myothercollector --foo=bar" ex.Command = "/usr/bin/myothercollector --foo=bar"

View File

@@ -93,10 +93,7 @@ func (r runnerMock) Run(e *Exec, command string, acc telegraf.Accumulator) ([]by
} }
func TestExec(t *testing.T) { func TestExec(t *testing.T) {
parser, _ := parsers.NewParser(&parsers.Config{ parser, _ := parsers.NewJSONParser("exec", []string{}, nil)
DataFormat: "json",
MetricName: "exec",
})
e := &Exec{ e := &Exec{
runner: newRunnerMock([]byte(validJson), nil), runner: newRunnerMock([]byte(validJson), nil),
Commands: []string{"testcommand arg1"}, Commands: []string{"testcommand arg1"},
@@ -122,10 +119,7 @@ func TestExec(t *testing.T) {
} }
func TestExecMalformed(t *testing.T) { func TestExecMalformed(t *testing.T) {
parser, _ := parsers.NewParser(&parsers.Config{ parser, _ := parsers.NewJSONParser("exec", []string{}, nil)
DataFormat: "json",
MetricName: "exec",
})
e := &Exec{ e := &Exec{
runner: newRunnerMock([]byte(malformedJson), nil), runner: newRunnerMock([]byte(malformedJson), nil),
Commands: []string{"badcommand arg1"}, Commands: []string{"badcommand arg1"},
@@ -138,10 +132,7 @@ func TestExecMalformed(t *testing.T) {
} }
func TestCommandError(t *testing.T) { func TestCommandError(t *testing.T) {
parser, _ := parsers.NewParser(&parsers.Config{ parser, _ := parsers.NewJSONParser("exec", []string{}, nil)
DataFormat: "json",
MetricName: "exec",
})
e := &Exec{ e := &Exec{
runner: newRunnerMock(nil, fmt.Errorf("exit status code 1")), runner: newRunnerMock(nil, fmt.Errorf("exit status code 1")),
Commands: []string{"badcommand"}, Commands: []string{"badcommand"},

View File

@@ -26,11 +26,7 @@ func TestHTTPwithJSONFormat(t *testing.T) {
URLs: []string{url}, URLs: []string{url},
} }
metricName := "metricName" metricName := "metricName"
p, _ := parsers.NewJSONParser(metricName, nil, nil)
p, _ := parsers.NewParser(&parsers.Config{
DataFormat: "json",
MetricName: "metricName",
})
plugin.SetParser(p) plugin.SetParser(p)
var acc testutil.Accumulator var acc testutil.Accumulator
@@ -67,11 +63,8 @@ func TestHTTPHeaders(t *testing.T) {
URLs: []string{url}, URLs: []string{url},
Headers: map[string]string{header: headerValue}, Headers: map[string]string{header: headerValue},
} }
metricName := "metricName"
p, _ := parsers.NewParser(&parsers.Config{ p, _ := parsers.NewJSONParser(metricName, nil, nil)
DataFormat: "json",
MetricName: "metricName",
})
plugin.SetParser(p) plugin.SetParser(p)
var acc testutil.Accumulator var acc testutil.Accumulator
@@ -90,10 +83,7 @@ func TestInvalidStatusCode(t *testing.T) {
} }
metricName := "metricName" metricName := "metricName"
p, _ := parsers.NewParser(&parsers.Config{ p, _ := parsers.NewJSONParser(metricName, nil, nil)
DataFormat: "json",
MetricName: metricName,
})
plugin.SetParser(p) plugin.SetParser(p)
var acc testutil.Accumulator var acc testutil.Accumulator
@@ -115,10 +105,8 @@ func TestMethod(t *testing.T) {
Method: "POST", Method: "POST",
} }
p, _ := parsers.NewParser(&parsers.Config{ metricName := "metricName"
DataFormat: "json", p, _ := parsers.NewJSONParser(metricName, nil, nil)
MetricName: "metricName",
})
plugin.SetParser(p) plugin.SetParser(p)
var acc testutil.Accumulator var acc testutil.Accumulator

View File

@@ -181,12 +181,7 @@ func (h *HttpJson) gatherServer(
"server": serverURL, "server": serverURL,
} }
parser, err := parsers.NewParser(&parsers.Config{ parser, err := parsers.NewJSONParser(msrmnt_name, h.TagKeys, tags)
DataFormat: "json",
MetricName: msrmnt_name,
TagKeys: h.TagKeys,
DefaultTags: tags,
})
if err != nil { if err != nil {
return err return err
} }

View File

@@ -125,10 +125,7 @@ func TestRunParserAndGatherJSON(t *testing.T) {
k.acc = &acc k.acc = &acc
defer close(k.done) defer close(k.done)
k.parser, _ = parsers.NewParser(&parsers.Config{ k.parser, _ = parsers.NewJSONParser("kafka_json_test", []string{}, nil)
DataFormat: "json",
MetricName: "kafka_json_test",
})
go k.receiver() go k.receiver()
in <- saramaMsg(testMsgJSON) in <- saramaMsg(testMsgJSON)
acc.Wait(1) acc.Wait(1)

View File

@@ -125,10 +125,7 @@ func TestRunParserAndGatherJSON(t *testing.T) {
k.acc = &acc k.acc = &acc
defer close(k.done) defer close(k.done)
k.parser, _ = parsers.NewParser(&parsers.Config{ k.parser, _ = parsers.NewJSONParser("kafka_json_test", []string{}, nil)
DataFormat: "json",
MetricName: "kafka_json_test",
})
go k.receiver() go k.receiver()
in <- saramaMsg(testMsgJSON) in <- saramaMsg(testMsgJSON)
acc.Wait(1) acc.Wait(1)

View File

@@ -340,6 +340,9 @@ func (p *Parser) ParseLine(line string) (telegraf.Metric, error) {
ts, err := time.ParseInLocation(t, v, p.loc) ts, err := time.ParseInLocation(t, v, p.loc)
if err == nil { if err == nil {
if ts.Year() == 0 {
ts = ts.AddDate(timestamp.Year(), 0, 0)
}
timestamp = ts timestamp = ts
} else { } else {
log.Printf("E! Error parsing %s to time layout [%s]: %s", v, t, err) log.Printf("E! Error parsing %s to time layout [%s]: %s", v, t, err)
@@ -469,7 +472,6 @@ func (t *tsModder) tsMod(ts time.Time) time.Time {
t.rollover = 0 t.rollover = 0
return ts return ts
} }
if ts.Equal(t.last) { if ts.Equal(t.last) {
t.dupe = ts t.dupe = ts
} }

View File

@@ -1000,3 +1000,20 @@ func TestReplaceTimestampComma(t *testing.T) {
//Convert Nanosecond to milisecond for compare //Convert Nanosecond to milisecond for compare
require.Equal(t, 555, m.Time().Nanosecond()/1000000) require.Equal(t, 555, m.Time().Nanosecond()/1000000)
} }
func TestEmptyYearInTimestamp(t *testing.T) {
p := &Parser{
Patterns: []string{`%{APPLE_SYSLOG_TIME_SHORT:timestamp:ts-"Jan 2 15:04:05"} %{HOSTNAME} %{APP_NAME:app_name}\[%{NUMBER:pid:int}\]%{GREEDYDATA:message}`},
CustomPatterns: `
APPLE_SYSLOG_TIME_SHORT %{MONTH} +%{MONTHDAY} %{TIME}
APP_NAME [a-zA-Z0-9\.]+
`,
}
require.NoError(t, p.Compile())
p.ParseLine("Nov 6 13:57:03 generic iTunes[6504]: info> Scale factor of main display = 2.0")
m, err := p.ParseLine("Nov 6 13:57:03 generic iTunes[6504]: objc[6504]: Object descriptor was null.")
require.NoError(t, err)
require.NotNil(t, m)
require.Equal(t, 2018, m.Time().Year())
}

View File

@@ -172,10 +172,7 @@ func TestRunParserAndGatherJSON(t *testing.T) {
n.acc = &acc n.acc = &acc
defer close(n.done) defer close(n.done)
n.parser, _ = parsers.NewParser(&parsers.Config{ n.parser, _ = parsers.NewJSONParser("nats_json_test", []string{}, nil)
DataFormat: "json",
MetricName: "nats_json_test",
})
go n.receiver() go n.receiver()
in <- mqttMsg(testMsgJSON) in <- mqttMsg(testMsgJSON)

View File

@@ -108,10 +108,7 @@ func TestRunParserAndGatherJSON(t *testing.T) {
n.acc = &acc n.acc = &acc
defer close(n.done) defer close(n.done)
n.parser, _ = parsers.NewParser(&parsers.Config{ n.parser, _ = parsers.NewJSONParser("nats_json_test", []string{}, nil)
DataFormat: "json",
MetricName: "nats_json_test",
})
n.wg.Add(1) n.wg.Add(1)
go n.receiver() go n.receiver()
in <- natsMsg(testMsgJSON) in <- natsMsg(testMsgJSON)

View File

@@ -300,10 +300,7 @@ func TestRunParserJSONMsg(t *testing.T) {
listener.acc = &acc listener.acc = &acc
defer close(listener.done) defer close(listener.done)
listener.parser, _ = parsers.NewParser(&parsers.Config{ listener.parser, _ = parsers.NewJSONParser("udp_json_test", []string{}, nil)
DataFormat: "json",
MetricName: "udp_json_test",
})
listener.wg.Add(1) listener.wg.Add(1)
go listener.tcpParser() go listener.tcpParser()

View File

@@ -193,10 +193,7 @@ func TestRunParserJSONMsg(t *testing.T) {
listener.acc = &acc listener.acc = &acc
defer close(listener.done) defer close(listener.done)
listener.parser, _ = parsers.NewParser(&parsers.Config{ listener.parser, _ = parsers.NewJSONParser("udp_json_test", []string{}, nil)
DataFormat: "json",
MetricName: "udp_json_test",
})
listener.wg.Add(1) listener.wg.Add(1)
go listener.udpParser() go listener.udpParser()

View File

@@ -20,7 +20,6 @@ var (
type JSONParser struct { type JSONParser struct {
MetricName string MetricName string
TagKeys []string TagKeys []string
FieldKeys []string
DefaultTags map[string]string DefaultTags map[string]string
} }
@@ -87,17 +86,6 @@ func (p *JSONParser) switchFieldToTag(tags map[string]string, fields map[string]
} }
} }
//if field_keys is specified, only those values should be reported as fields
if len(p.FieldKeys) > 0 {
nFields := make(map[string]interface{})
for _, name := range p.FieldKeys {
if fields[name] != nil {
nFields[name] = fields[name]
}
}
return tags, nFields
}
//remove any additional string/bool values from fields //remove any additional string/bool values from fields
for k := range fields { for k := range fields {
switch fields[k].(type) { switch fields[k].(type) {

View File

@@ -1,7 +1,6 @@
package json package json
import ( import (
"log"
"testing" "testing"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
@@ -449,28 +448,22 @@ func TestJSONParseNestedArray(t *testing.T) {
"total_devices": 5, "total_devices": 5,
"total_threads": 10, "total_threads": 10,
"shares": { "shares": {
"total": 5, "total": 5,
"accepted": 5, "accepted": 5,
"rejected": 0, "rejected": 0,
"avg_find_time": 4, "avg_find_time": 4,
"tester": "work", "tester": "work",
"tester2": "don't want this", "tester2": "don't want this",
"tester3": { "tester3": 7.93
"hello":"sup",
"fun":"money",
"break":9
}
} }
}` }`
parser := JSONParser{ parser := JSONParser{
MetricName: "json_test", MetricName: "json_test",
TagKeys: []string{"total_devices", "total_threads", "shares_tester3_fun"}, TagKeys: []string{"total_devices", "total_threads", "shares_tester", "shares_tester3"},
FieldKeys: []string{"shares_tester", "shares_tester3_break"},
} }
metrics, err := parser.Parse([]byte(testString)) metrics, err := parser.Parse([]byte(testString))
log.Printf("m[0] name: %v, tags: %v, fields: %v", metrics[0].Name(), metrics[0].Tags(), metrics[0].Fields())
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, len(parser.TagKeys), len(metrics[0].Tags())) require.Equal(t, len(parser.TagKeys), len(metrics[0].Tags()))
} }

View File

@@ -56,8 +56,6 @@ type Config struct {
// TagKeys only apply to JSON data // TagKeys only apply to JSON data
TagKeys []string TagKeys []string
// FieldKeys only apply to JSON
FieldKeys []string
// MetricName applies to JSON & value. This will be the name of the measurement. // MetricName applies to JSON & value. This will be the name of the measurement.
MetricName string MetricName string
@@ -97,8 +95,8 @@ func NewParser(config *Config) (Parser, error) {
var parser Parser var parser Parser
switch config.DataFormat { switch config.DataFormat {
case "json": case "json":
parser, err = newJSONParser(config.MetricName, parser, err = NewJSONParser(config.MetricName,
config.TagKeys, config.FieldKeys, config.DefaultTags) config.TagKeys, config.DefaultTags)
case "value": case "value":
parser, err = NewValueParser(config.MetricName, parser, err = NewValueParser(config.MetricName,
config.DataType, config.DefaultTags) config.DataType, config.DefaultTags)
@@ -128,22 +126,6 @@ func NewParser(config *Config) (Parser, error) {
return parser, err return parser, err
} }
func newJSONParser(
metricName string,
tagKeys []string,
fieldKeys []string,
defaultTags map[string]string,
) (Parser, error) {
parser := &json.JSONParser{
MetricName: metricName,
TagKeys: tagKeys,
FieldKeys: fieldKeys,
DefaultTags: defaultTags,
}
return parser, nil
}
//Deprecated: Use NewParser to get a JSONParser object
func NewJSONParser( func NewJSONParser(
metricName string, metricName string,
tagKeys []string, tagKeys []string,