luciano/amazon-kinesis-client-go
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Removed stringifyNested, renameESReservedFields, and minimumTimestamp params from decoder. minimumTimestamp isn't needed anymore. stringifyNested and renameESReservedFields are very specific to the elasticsearch consumer and will be moved to the elasticsearch consumer.

Browse source
This commit is contained in:
Xavi Ramirez 2017-09-21 19:51:48 +00:00
parent 8d871591de
commit 2ecbc081cd
2 changed files with 37 additions and 196 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

60
decode/decode.go
View file

@ -5,7 +5,6 @@ import (
"fmt"
"regexp"
"strings"
"time"
"github.com/Clever/syslogparser/rfc3164"
)
@ -174,6 +173,17 @@ func getStringArray(json map[string]interface{}, key string) []string {
// LogRoutes a type alias to make it easier to add route specific filter functions
type LogRoutes []map[string]interface{}
// RuleNames returns a list of the names of the rules
func (l LogRoutes) RuleNames() []string {
names := []string{}
for _, route := range l {
name := getStringValue(route, "rule")
names = append(names, name)
}
return names
}
// MetricsRoutes filters the LogRoutes and returns a list of MetricsRoutes structs
func (l LogRoutes) MetricsRoutes() []MetricsRoute {
routes := []MetricsRoute{}
@ -354,7 +364,7 @@ func ExtractKVMeta(kvlog map[string]interface{}) KVMeta {
}
// ParseAndEnhance extracts fields from a log line, and does some post-processing to rename/add fields
func ParseAndEnhance(line string, env string, stringifyNested bool, renameESReservedFields bool, minimumTimestamp time.Time) (map[string]interface{}, error) {
func ParseAndEnhance(line string, env string) (map[string]interface{}, error) {
out := map[string]interface{}{}
syslogFields, err := FieldsFromSyslog(line)
@ -405,55 +415,9 @@ func ParseAndEnhance(line string, env string, stringifyNested bool, renameESRese
}
}
// ES dynamic mappings get finnicky once you start sending nested objects.
// E.g., if one app sends a field for the first time as an object, then any log
// sent by another app containing that field /not/ as an object will fail.
// One solution is to decode nested objects as strings.
if stringifyNested {
for k, v := range out {
_, ismap := v.(map[string]interface{})
_, isarr := v.([]interface{})
if ismap || isarr {
bs, _ := json.Marshal(v)
out[k] = string(bs)
}
}
}
// ES doesn't like fields that start with underscores.
if renameESReservedFields {
for oldKey, renamedKey := range esFieldRenames {
if val, ok := out[oldKey]; ok {
out[renamedKey] = val
delete(out, oldKey)
}
}
}
msgTime, ok := out["timestamp"].(time.Time)
if ok && !msgTime.After(minimumTimestamp) {
return map[string]interface{}{}, fmt.Errorf("message's timestamp < minimumTimestamp")
}
return out, nil
}
var esFieldRenames = map[string]string{
"_index": "kv__index",
"_uid": "kv__uid",
"_type": "kv__type",
"_id": "kv__id",
"_source": "kv__source",
"_size": "kv__size",
"_all": "kv__all",
"_field_names": "kv__field_names",
"_timestamp": "kv__timestamp",
"_ttl": "kv__ttl",
"_parent": "kv__parent",
"_routing": "kv__routing",
"_meta": "kv__meta",
}
const containerMeta = `([a-z0-9-]+)--([a-z0-9-]+)\/` + // env--app
`arn%3Aaws%3Aecs%3Aus-(west|east)-[1-2]%3A[0-9]{12}%3Atask%2F` + // ARN cruft
`([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})` // task-id

173
decode/decode_test.go
View file

@ -2,6 +2,7 @@ package decode
import (
"fmt"
"sort"
"testing"
"time"
@ -188,16 +189,9 @@ func TestSyslogDecoding(t *testing.T) {
}
}
type ParseAndEnhanceInput struct {
Line string
StringifyNested bool
RenameESReservedFields bool
MinimumTimestamp time.Time
}
type ParseAndEnhanceSpec struct {
Title string
Input ParseAndEnhanceInput
Line string
ExpectedOutput map[string]interface{}
ExpectedError error
}
@ -219,7 +213,7 @@ func TestParseAndEnhance(t *testing.T) {
specs := []ParseAndEnhanceSpec{
ParseAndEnhanceSpec{
Title: "Parses a Kayvee log line from an ECS app",
Input: ParseAndEnhanceInput{Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: 2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished"}`},
Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: 2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished"}`,
ExpectedOutput: map[string]interface{}{
"timestamp": logTime3,
"hostname": "ip-10-0-0-0",
@ -238,7 +232,7 @@ func TestParseAndEnhance(t *testing.T) {
},
ParseAndEnhanceSpec{
Title: "Parses a Kayvee log line from an ECS app, with override to container_app",
Input: ParseAndEnhanceInput{Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: 2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished","container_app":"force-app"}`},
Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: 2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished","container_app":"force-app"}`,
ExpectedOutput: map[string]interface{}{
"timestamp": logTime3,
"hostname": "ip-10-0-0-0",
@ -257,7 +251,7 @@ func TestParseAndEnhance(t *testing.T) {
},
ParseAndEnhanceSpec{
Title: "Parses a non-Kayvee log line",
Input: ParseAndEnhanceInput{Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: some log`},
Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: some log`,
ExpectedOutput: map[string]interface{}{
"timestamp": logTime3,
"hostname": "ip-10-0-0-0",
@ -272,13 +266,13 @@ func TestParseAndEnhance(t *testing.T) {
},
ParseAndEnhanceSpec{
Title: "Fails to parse non-RSyslog log line",
Input: ParseAndEnhanceInput{Line: `not rsyslog`},
Line: `not rsyslog`,
ExpectedOutput: map[string]interface{}{},
ExpectedError: &syslogparser.ParserError{},
},
ParseAndEnhanceSpec{
Title: "Parses JSON values",
Input: ParseAndEnhanceInput{Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: 2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "nested": {"a":"b"}}`},
Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: 2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "nested": {"a":"b"}}`,
ExpectedOutput: map[string]interface{}{
"timestamp": logTime3,
"hostname": "ip-10-0-0-0",
@ -296,141 +290,9 @@ func TestParseAndEnhance(t *testing.T) {
},
ExpectedError: nil,
},
ParseAndEnhanceSpec{
Title: "Has the option to stringify object values",
Input: ParseAndEnhanceInput{
Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: 2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "nested": {"a":"b"}}`,
StringifyNested: true,
},
ExpectedOutput: map[string]interface{}{
"timestamp": logTime3,
"hostname": "ip-10-0-0-0",
"programname": `env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef`,
"rawlog": `2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "nested": {"a":"b"}}`,
"title": "request_finished",
"decoder_msg_type": "Kayvee",
"prefix": "2017/04/05 21:57:46 some_file.go:10: ",
"postfix": "",
"env": "deploy-env",
"container_env": "env",
"container_app": "app",
"container_task": "abcd1234-1a3b-1a3b-1234-d76552f4b7ef",
"nested": `{"a":"b"}`,
},
ExpectedError: nil,
},
ParseAndEnhanceSpec{
Title: "Has the option to stringify array values",
Input: ParseAndEnhanceInput{
Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: 2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "nested": [{"a":"b"}]}`,
StringifyNested: true,
},
ExpectedOutput: map[string]interface{}{
"timestamp": logTime3,
"hostname": "ip-10-0-0-0",
"programname": `env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef`,
"rawlog": `2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "nested": [{"a":"b"}]}`,
"title": "request_finished",
"decoder_msg_type": "Kayvee",
"prefix": "2017/04/05 21:57:46 some_file.go:10: ",
"postfix": "",
"env": "deploy-env",
"container_env": "env",
"container_app": "app",
"container_task": "abcd1234-1a3b-1a3b-1234-d76552f4b7ef",
"nested": `[{"a":"b"}]`,
},
ExpectedError: nil,
},
ParseAndEnhanceSpec{
Title: "Has the option to rename reserved ES fields",
Input: ParseAndEnhanceInput{
Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: 2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "_source": "a"}`,
RenameESReservedFields: true,
},
ExpectedOutput: map[string]interface{}{
"timestamp": logTime3,
"hostname": "ip-10-0-0-0",
"programname": `env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef`,
"rawlog": `2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "_source": "a"}`,
"title": "request_finished",
"decoder_msg_type": "Kayvee",
"prefix": "2017/04/05 21:57:46 some_file.go:10: ",
"postfix": "",
"env": "deploy-env",
"container_env": "env",
"container_app": "app",
"container_task": "abcd1234-1a3b-1a3b-1234-d76552f4b7ef",
"kv__source": "a",
},
ExpectedError: nil,
},
ParseAndEnhanceSpec{
Title: "Errors if logTime < MinimumTimestamp",
Input: ParseAndEnhanceInput{
Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: 2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "_source": "a"}`,
RenameESReservedFields: true,
MinimumTimestamp: time.Now().Add(100 * time.Hour * 24 * 365), // good thru year 2117
},
ExpectedOutput: map[string]interface{}{},
ExpectedError: fmt.Errorf(""),
},
ParseAndEnhanceSpec{
Title: "Accepts logs if logTime > MinimumTimestamp",
Input: ParseAndEnhanceInput{
Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: 2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "_source": "a"}`,
RenameESReservedFields: true,
MinimumTimestamp: time.Now().Add(-100 * time.Hour * 24 * 365), // good thru year 2117
},
ExpectedOutput: map[string]interface{}{
"timestamp": logTime3,
"hostname": "ip-10-0-0-0",
"programname": `env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef`,
"rawlog": `2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "_source": "a"}`,
"title": "request_finished",
"decoder_msg_type": "Kayvee",
"prefix": "2017/04/05 21:57:46 some_file.go:10: ",
"postfix": "",
"env": "deploy-env",
"container_env": "env",
"container_app": "app",
"container_task": "abcd1234-1a3b-1a3b-1234-d76552f4b7ef",
"kv__source": "a",
},
ExpectedError: nil,
},
ParseAndEnhanceSpec{
Title: "Accepts logs if logTime > MinimumTimestamp",
Input: ParseAndEnhanceInput{
Line: `2017-04-05T21:57:46.794862+00:00 ip-10-0-0-0 env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef[3291]: 2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "_source": "a"}`,
RenameESReservedFields: true,
MinimumTimestamp: time.Now().Add(-100 * time.Hour * 24 * 365), // good thru year 2117
},
ExpectedOutput: map[string]interface{}{
"timestamp": logTime3,
"hostname": "ip-10-0-0-0",
"programname": `env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef`,
"rawlog": `2017/04/05 21:57:46 some_file.go:10: {"title":"request_finished", "_source": "a"}`,
"title": "request_finished",
"decoder_msg_type": "Kayvee",
"prefix": "2017/04/05 21:57:46 some_file.go:10: ",
"postfix": "",
"env": "deploy-env",
"container_env": "env",
"container_app": "app",
"container_task": "abcd1234-1a3b-1a3b-1234-d76552f4b7ef",
"kv__source": "a",
},
ExpectedError: nil,
},
ParseAndEnhanceSpec{
Title: "Log with timestamp time.RFC3339 format",
Input: ParseAndEnhanceInput{
Line: `2017-04-05T21:57:46+00:00 mongo-docker-pipeline-r10-4 diamond[24099] ` +
`Signal Received: 15`,
RenameESReservedFields: true,
MinimumTimestamp: time.Now().Add(-100 * time.Hour * 24 * 365), // year 2117
},
Line: `2017-04-05T21:57:46+00:00 mongo-docker-pipeline-r10-4 diamond[24099] Signal Received: 15`,
ExpectedOutput: map[string]interface{}{
"env": "deploy-env",
"hostname": "mongo-docker-pipeline-r10-4",
@ -443,7 +305,7 @@ func TestParseAndEnhance(t *testing.T) {
for _, spec := range specs {
t.Run(fmt.Sprintf(spec.Title), func(t *testing.T) {
assert := assert.New(t)
fields, err := ParseAndEnhance(spec.Input.Line, "deploy-env", spec.Input.StringifyNested, spec.Input.RenameESReservedFields, spec.Input.MinimumTimestamp)
fields, err := ParseAndEnhance(spec.Line, "deploy-env")
if spec.ExpectedError != nil {
assert.Error(err)
assert.IsType(spec.ExpectedError, err)
@ -523,6 +385,7 @@ func TestExtractKVMeta(t *testing.T) {
Team string
Language string
Version string
Names []string
ExpectedMetricsRoutes []MetricsRoute
ExpectedAnalyticsRoutes []AnalyticsRoute
ExpectedNotificationRoutes []NotificationRoute
@ -530,10 +393,12 @@ func TestExtractKVMeta(t *testing.T) {
}{
{
Description: "log line with no routes",
Names: []string{},
Log: map[string]interface{}{"hi": "hello!"},
},
{
Description: "empty _kvmeta",
Names: []string{},
Log: map[string]interface{}{
"hi": "hello!",
"_kvmeta": map[string]interface{}{},
@ -544,6 +409,7 @@ func TestExtractKVMeta(t *testing.T) {
Team: "green",
Version: "three",
Language: "tree",
Names: []string{},
Log: map[string]interface{}{
"hi": "hello!",
"_kvmeta": map[string]interface{}{
@ -558,6 +424,7 @@ func TestExtractKVMeta(t *testing.T) {
Team: "green",
Version: "three",
Language: "tree",
Names: []string{"cool", "cool2"},
ExpectedMetricsRoutes: []MetricsRoute{
{
Series: "1,1,2,3,5,8,13",
@ -601,6 +468,7 @@ func TestExtractKVMeta(t *testing.T) {
Team: "green",
Version: "christmas",
Language: "tree",
Names: []string{"what's-this?", "there's-app-invites-everywhere"},
ExpectedAnalyticsRoutes: []AnalyticsRoute{
{
Series: "what-is-this",
@ -637,6 +505,7 @@ func TestExtractKVMeta(t *testing.T) {
Team: "slack",
Version: "evergreen",
Language: "markdown-ish",
Names: []string{"did-you-know", "what's-the-catch"},
ExpectedNotificationRoutes: []NotificationRoute{
{
RuleName: "did-you-know",
@ -683,6 +552,7 @@ func TestExtractKVMeta(t *testing.T) {
Team: "a-team",
Version: "old",
Language: "jive",
Names: []string{"last-call", "watch-out-now"},
ExpectedAlertRoutes: []AlertRoute{
{
RuleName: "last-call",
@ -729,6 +599,9 @@ func TestExtractKVMeta(t *testing.T) {
Team: "diversity",
Version: "kv-routes",
Language: "understanding",
Names: []string{
"all-combos", "there's-app-invites-everywhere", "what's-the-catch", "last-call",
},
ExpectedMetricsRoutes: []MetricsRoute{
{
RuleName: "all-combos",
@ -807,6 +680,10 @@ func TestExtractKVMeta(t *testing.T) {
assert.Equal(test.Language, kvmeta.Language)
assert.Equal(test.Version, kvmeta.Version)
assert.EqualValues(
sort.StringSlice(test.Names), sort.StringSlice(kvmeta.Routes.RuleNames()),
)
assert.Equal(len(test.ExpectedMetricsRoutes), len(kvmeta.Routes.MetricsRoutes()))
for idx, route := range kvmeta.Routes.MetricsRoutes() {
expected := test.ExpectedMetricsRoutes[idx]
@ -853,7 +730,7 @@ func BenchmarkFieldsFromSyslog(b *testing.B) {
func BenchmarkParseAndEnhance(b *testing.B) {
for n := 0; n < b.N; n++ {
_, err := ParseAndEnhance(benchmarkLine, "env", false, false, time.Time{})
_, err := ParseAndEnhance(benchmarkLine, "env")
if err != nil {
b.FailNow()
}