.circleci/config.yml

@@ -6,7 +6,6 @@ jobs:
     - image: circleci/golang:1.13-stretch
     - image: circleci/mongo:3.2.20-jessie-ram
     environment:
-      GOPRIVATE: github.com/Clever/*
       CIRCLE_ARTIFACTS: /tmp/circleci-artifacts
       CIRCLE_TEST_REPORTS: /tmp/circleci-test-results
     steps:
@@ -18,13 +17,7 @@ jobs:
     - run:
         command: mkdir -p $CIRCLE_ARTIFACTS $CIRCLE_TEST_REPORTS
         name: Set up CircleCI artifacts directories
-    - run:
-        command: git config --global "url.ssh://git@github.com/Clever".insteadOf "https://github.com/Clever"
-    - run:
-        name: Add github.com to known hosts
-        command: mkdir -p ~/.ssh && touch ~/.ssh/known_hosts && echo 'github.com ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQCj7ndNxQowgcQnjshcLrqPEiiphnt+VTTvDP6mHBL9j1aNUkY4Ue1gvwnGLVlOhGeYrnZaMgRK6+PKCUXaDbC7qtbW8gIkhL7aGCsOr/C56SJMy/BCZfxd1nWzAOxSDPgVsmerOBYfNqltV9/hWCqBywINIR+5dIg6JTJ72pcEpEjcYgXkE2YEFXV1JHnsKgbLWNlhScqb2UmyRkQyytRLtL+38TGxkxCflmO+5Z8CSSNY7GidjMIZ7Q4zMjA2n1nGrlTDkzwDCsw+wqFPGQA179cnfGWOWRVruj16z6XyvxvjJwbz0wQZ75XK5tKSb7FNyeIEs4TT4jk+S4dhPeAUC5y+bDYirYgM4GC7uEnztnZyaVWQ7B381AK4Qdrwt51ZqExKbQpTUNn+EjqoTwvqNj4kqx5QUCI0ThS/YkOxJCXmPUWZbhjpCg56i+2aB6CmK2JGhn57K5mj0MNdBXA4/WnwH6XoPWJzK5Nyu2zB3nAZp+S5hpQs+p1vN1/wsjk=' >> ~/.ssh/known_hosts
     - run: make install_deps
     - run: make build
     - run: make bench
     - run: make test
-    - run: if [ "${CIRCLE_BRANCH}" == "master" ]; then $HOME/ci-scripts/circleci/github-release $GH_RELEASE_TOKEN; fi;

.github/CODEOWNERS

@@ -1 +0,0 @@
-* @Clever/eng-infra

.github/workflows/notify-ci-status.yml

@@ -1,20 +0,0 @@
-name: Notify CI status
-
-on:
-  check_suite:
-    types: [completed]
-  status:
-
-jobs:
-  call-workflow:
-    if: >-
-      (github.event.branches[0].name == github.event.repository.default_branch &&
-        (github.event.state == 'error' || github.event.state == 'failure')) ||
-      (github.event.check_suite.head_branch == github.event.repository.default_branch &&
-        github.event.check_suite.conclusion != 'success')
-    uses: Clever/ci-scripts/.github/workflows/reusable-notify-ci-status.yml@master
-    secrets:
-      CIRCLE_CI_INTEGRATIONS_URL: ${{ secrets.CIRCLE_CI_INTEGRATIONS_URL }}
-      CIRCLE_CI_INTEGRATIONS_USERNAME: ${{ secrets.CIRCLE_CI_INTEGRATIONS_USERNAME }}
-      CIRCLE_CI_INTEGRATIONS_PASSWORD: ${{ secrets.CIRCLE_CI_INTEGRATIONS_PASSWORD }}
-      SLACK_BOT_TOKEN: ${{ secrets.DAPPLE_BOT_TOKEN }}

Gopkg.lock

@@ -0,0 +1,84 @@
+# This file is autogenerated, do not edit; changes may be undone by the next 'dep ensure'.
+
+
+[[projects]]
+  branch = "master"
+  name = "github.com/Clever/syslogparser"
+  packages = [
+    ".",
+    "rfc3164"
+  ]
+  revision = "fb28ad3e4340c046323b7beba685a72fd12ecbe8"
+
+[[projects]]
+  name = "github.com/davecgh/go-spew"
+  packages = ["spew"]
+  revision = "6d212800a42e8ab5c146b8ace3490ee17e5225f9"
+
+[[projects]]
+  branch = "master"
+  name = "github.com/jeromer/syslogparser"
+  packages = ["."]
+  revision = "0e4ae46ea3f08de351074b643d649d5d00661a3c"
+
+[[projects]]
+  name = "github.com/pmezard/go-difflib"
+  packages = ["difflib"]
+  revision = "d8ed2627bdf02c080bf22230dbb337003b7aba2d"
+
+[[projects]]
+  name = "github.com/stretchr/testify"
+  packages = [
+    "assert",
+    "require"
+  ]
+  revision = "69483b4bd14f5845b5a1e55bca19e954e827f1d0"
+  version = "v1.1.4"
+
+[[projects]]
+  name = "github.com/xeipuuv/gojsonpointer"
+  packages = ["."]
+  revision = "e0fe6f68307607d540ed8eac07a342c33fa1b54a"
+
+[[projects]]
+  branch = "master"
+  name = "github.com/xeipuuv/gojsonreference"
+  packages = ["."]
+  revision = "e02fc20de94c78484cd5ffb007f8af96be030a45"
+
+[[projects]]
+  name = "github.com/xeipuuv/gojsonschema"
+  packages = ["."]
+  revision = "e18f0065e8c148fcf567ac43a3f8f5b66ac0720b"
+
+[[projects]]
+  name = "golang.org/x/net"
+  packages = ["context"]
+  revision = "a6577fac2d73be281a500b310739095313165611"
+
+[[projects]]
+  name = "golang.org/x/time"
+  packages = ["rate"]
+  revision = "f51c12702a4d776e4c1fa9b0fabab841babae631"
+
+[[projects]]
+  name = "gopkg.in/Clever/kayvee-go.v6"
+  packages = [
+    ".",
+    "logger",
+    "router"
+  ]
+  revision = "096364e316a52652d3493be702d8105d8d01db84"
+  version = "v6.6.0"
+
+[[projects]]
+  name = "gopkg.in/yaml.v2"
+  packages = ["."]
+  revision = "a5b47d31c556af34a302ce5d659e6fea44d90de0"
+
+[solve-meta]
+  analyzer-name = "dep"
+  analyzer-version = 1
+  inputs-digest = "4699293f3632dd38561ff60477aa7cc1ecaadc5808b974d017099e2189679286"
+  solver-name = "gps-cdcl"
+  solver-version = 1

Gopkg.toml

@@ -0,0 +1,36 @@
+
+# Gopkg.toml example
+#
+# Refer to https://github.com/golang/dep/blob/master/docs/Gopkg.toml.md
+# for detailed Gopkg.toml documentation.
+#
+# required = ["github.com/user/thing/cmd/thing"]
+# ignored = ["github.com/user/project/pkgX", "bitbucket.org/user/project/pkgA/pkgY"]
+#
+# [[constraint]]
+#   name = "github.com/user/project"
+#   version = "1.0.0"
+#
+# [[constraint]]
+#   name = "github.com/user/project2"
+#   branch = "dev"
+#   source = "github.com/myfork/project2"
+#
+# [[override]]
+#  name = "github.com/x/y"
+#  version = "2.4.0"
+
+
+[[constraint]]
+  branch = "master"
+  name = "github.com/Clever/syslogparser"
+
+[[constraint]]
+  name = "github.com/stretchr/testify"
+
+[[constraint]]
+  name = "golang.org/x/time"
+
+[[constraint]]
+  name = "gopkg.in/Clever/kayvee-go.v6"
+  version = "6.0.0"

Makefile

@@ -61,5 +61,5 @@ $(PKGS): golang-test-all-deps
 
 
 
-install_deps:
-	go mod vendor
+install_deps: golang-dep-vendor-deps
+	$(call golang-dep-vendor)

VERSION

@@ -1 +0,0 @@
-1.0.0

batchconsumer/stats/stats.go

@@ -30,17 +30,16 @@ type datum struct {
 var queue = make(chan datum, 1000)
 
 func init() {
-	countData := map[string]int{}
-	gaugeData := map[string]int{}
+	data := map[string]int{}
 	tick := time.Tick(time.Minute)
 	go func() {
 		for {
 			select {
 			case d := <-queue:
 				if d.category == "counter" {
-					countData[d.name] = countData[d.name] + d.value
+					data[d.name] = data[d.name] + d.value
 				} else if d.category == "gauge" {
-					gaugeData[d.name] = d.value
+					data[d.name] = d.value
 				} else {
 					log.ErrorD("unknown-stat-category", logger.M{"category": d.category})
 				}
@@ -49,14 +48,10 @@ func init() {
 				for _, k := range DefaultCounters {
 					tmp[k] = 0
 				}
-				for k, v := range countData {
-					tmp[k] = v
-				}
-				for k, v := range gaugeData {
+				for k, v := range data {
 					tmp[k] = v
 				}
 				log.InfoD("stats", tmp)
-				countData = map[string]int{}
 			}
 		}
 	}()

batchconsumer/writer.go

@@ -57,6 +57,19 @@ func (b *batchedWriter) Initialize(shardID string, checkpointer kcl.Checkpointer
 	return nil
 }
 
+func (b *batchedWriter) splitMessageIfNecessary(record []byte) ([][]byte, error) {
+	// We handle two types of records:
+	// - records emitted from CWLogs Subscription
+	// - records emitted from KPL
+	if !splitter.IsGzipped(record) {
+		// Process a single message, from KPL
+		return [][]byte{record}, nil
+	}
+
+	// Process a batch of messages from a CWLogs Subscription
+	return splitter.GetMessagesFromGzippedInput(record)
+}
+
 func (b *batchedWriter) ProcessRecords(records []kcl.Record) error {
 	var pair kcl.SequencePair
 	prevPair := b.lastProcessedSeq
@@ -80,7 +93,7 @@ func (b *batchedWriter) ProcessRecords(records []kcl.Record) error {
 			return err
 		}
 
-		messages, err := splitter.SplitMessageIfNecessary(data)
+		messages, err := b.splitMessageIfNecessary(data)
 		if err != nil {
 			return err
 		}

batchconsumer/writer_test.go

@@ -352,7 +352,7 @@ func TestProcessRecordsMutliBatchWithIgnores(t *testing.T) {
 	assert.Equal("tag3", string(mocksender.batches["tag3"][0][2]))
 }
 
-func TestStaggeredCheckpointing(t *testing.T) {
+func TestStaggeredCheckpionting(t *testing.T) {
 	assert := assert.New(t)
 
 	mockFailedLogsFile := logger.New("testing")

decode/decode.go

@@ -2,11 +2,9 @@ package decode
 
 import (
 	"encoding/json"
-	"errors"
 	"fmt"
 	"regexp"
 	"strings"
-	"time"
 
 	"github.com/Clever/syslogparser/rfc3164"
 )
@@ -17,18 +15,9 @@ var reservedFields = []string{
 	"prefix",
 	"postfix",
 	"decoder_msg_type",
-
-	// used by all logs
 	"timestamp",
 	"hostname",
 	"rawlog",
-
-	// Used by Kayvee
-	"prefix",
-	"postfix",
-
-	// Set to the deepest step of parsing that succeeded
-	"decoder_msg_type",
 }
 
 func stringInSlice(s string, slice []string) bool {
@@ -396,21 +385,17 @@ 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) (map[string]interface{}, error) {
-	syslog, syslogErr := FieldsFromSyslog(line)
-	if syslogErr == nil {
-		return decodeSyslog(syslog, env)
+	out := map[string]interface{}{}
+
+	syslogFields, err := FieldsFromSyslog(line)
+	if err != nil {
+		return map[string]interface{}{}, err
 	}
-
-	fluentLog, fluentErr := FieldsFromFluentbitLog(line)
-	if fluentErr == nil {
-		return decodeFluent(fluentLog, env)
+	for k, v := range syslogFields {
+		out[k] = v
 	}
-
-	return nil, fmt.Errorf("unable to parse log line with errors `%v` and `%v`", syslogErr, fluentErr)
-}
-
-func decodeSyslog(syslog map[string]interface{}, env string) (map[string]interface{}, error) {
-	rawlog := syslog["rawlog"].(string)
+	rawlog := syslogFields["rawlog"].(string)
+	programname := syslogFields["programname"].(string)
 
 	// Try pulling Kayvee fields out of message
 	kvFields, err := FieldsFromKayvee(rawlog)
@@ -420,39 +405,56 @@ func decodeSyslog(syslog map[string]interface{}, env string) (map[string]interfa
 		}
 	} else {
 		for k, v := range kvFields {
-			syslog[k] = v
-		}
-	}
-
-	// Try pulling RDS slowquery logs fields out of message
-	if syslog["hostname"] == "aws-rds" {
-		slowQueryFields := FieldsFromRDSSlowquery(rawlog)
-		for k, v := range slowQueryFields {
-			syslog[k] = v
+			out[k] = v
 		}
 	}
 
 	// Inject additional fields that are useful in log-searching and other business logic
-	syslog["env"] = env
+	out["env"] = env
 
-	// this can error, which indicates inability to extract container meta. That's fine, we can ignore that.
-	addContainterMetaToSyslog(syslog)
+	// Sometimes its useful to force `container_{env,app,task}`. A specific use-case is writing Docker events.
+	// A separate container monitors for start/stop events, but we set the container values in such a way that
+	// the logs for these events will appear in context for the app that the user is looking at instead of the
+	// docker-events app.
+	forceEnv := ""
+	forceApp := ""
+	forceTask := ""
+	if cEnv, ok := out["container_env"]; ok {
+		forceEnv = cEnv.(string)
+	}
+	if cApp, ok := out["container_app"]; ok {
+		forceApp = cApp.(string)
+	}
+	if cTask, ok := out["container_task"]; ok {
+		forceTask = cTask.(string)
+	}
+	meta, err := getContainerMeta(programname, forceEnv, forceApp, forceTask)
+	if err == nil {
+		for k, v := range meta {
+			out[k] = v
+		}
+	}
 
-	return syslog, nil
+	// Try pulling RDS slowquery logs fields out of message
+	if out["hostname"] == "aws-rds" {
+		slowQueryFields := FieldsFromRDSSlowquery(rawlog)
+		for k, v := range slowQueryFields {
+			out[k] = v
+		}
+	}
+
+	return out, nil
 }
 
 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}|[a-z0-9]{32}|jr_[a-z0-9-]+)` // task-id (ECS, both EC2 and Fargate), Glue Job ID
+	`([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}|[a-z0-9]{32})` // task-id (ECS, both EC2 and Fargate)
 
 var containerMetaRegex = regexp.MustCompile(containerMeta)
 
-var errBadProgramname = errors.New("invalid or missing programname")
-
-func addContainterMetaToSyslog(syslog map[string]interface{}) (map[string]interface{}, error) {
-	programname, ok := syslog["programname"].(string)
-	if !ok || programname == "" {
-		return syslog, errBadProgramname
+func getContainerMeta(programname, forceEnv, forceApp, forceTask string) (map[string]string, error) {
+	if programname == "" {
+		return map[string]string{}, fmt.Errorf("no programname")
 	}
 
 	env := ""
@@ -463,116 +465,25 @@ func addContainterMetaToSyslog(syslog map[string]interface{}) (map[string]interf
 		env = matches[0][1]
 		app = matches[0][2]
 		task = matches[0][4]
-	} else {
-		return syslog, errBadProgramname
 	}
 
-	// Sometimes its useful to force `container_{env,app,task}`. A specific use-case is writing Docker events.
-	// A separate container monitors for start/stop events, but we set the container values in such a way that
-	// the logs for these events will appear in context for the app that the user is looking at instead of the
-	// docker-events app.
-	if existingEnv, ok := syslog["container_env"].(string); !ok || existingEnv == "" {
-		syslog["container_env"] = env
+	if forceEnv != "" {
+		env = forceEnv
 	}
-	if existingApp, ok := syslog["container_app"].(string); !ok || existingApp == "" {
-		syslog["container_app"] = app
+	if forceApp != "" {
+		app = forceApp
 	}
-	if existingTask, ok := syslog["container_task"].(string); !ok || existingTask == "" {
-		syslog["container_task"] = task
+	if forceTask != "" {
+		task = forceTask
 	}
 
-	return syslog, nil
-}
-
-const (
-	// These are the fields in the incoming fluentbit JSON that we expect the timestamp and log
-	// They are referenced below by the FluentLog type; those are the ones that matter.
-	fluentbitTimestampField = "fluent_ts"
-	fluentbitLogField       = "log"
-
-	// This is what we get from the fluentbig config: `time_key_format: "%Y-%m-%dT%H:%M:%S.%L%z"`
-	fluentTimeFormat = "2006-01-02T15:04:05.999-0700"
-)
-
-// FluentLog represents is the set of fields extracted from an incoming fluentbit log
-// The struct tags must line up with the JSON schema in the fluentbit configuration, see the comment for FieldsFromFluentBitlog
-type FluentLog struct {
-	Log            *string `json:"log"`
-	Timestamp      *string `json:"fluent_ts"`
-	TaskArn        string  `json:"ecs_task_arn"`
-	TaskDefinition string  `json:"ecs_task_definition"`
-}
-
-// FieldsFromFluentbitLog parses JSON object with fields indicating that it's coming from FluentBit
-// Its return value shares a common interface with the Syslog output - with the same four key fields
-// Unlike FieldsFromSyslog, it accepts its argument as bytes, since it will be used as bytes immediately and bytes is what comes off the firehose
-// In theory, the format we recieve is highly customizable, so we'll be making the following assumptions:
-// - All logs are coming from aws-fargate with the ecs-metadata fields (ecs_cluster, ecs_task_arn, ecs_task_definition) enabled
-// - The timestamp is in the field given by the FluentBitTimestampField constant in this package
-// - The timestamp is of the format of the constant fluentTimestampFormat
-// - The log is in the field given by the FluentBitlogField constant in this package
-// - The ecs_task_definition is of the from {environment}--{application}--.*
-func FieldsFromFluentbitLog(line string) (*FluentLog, error) {
-	fluentFields := FluentLog{}
-	if err := json.Unmarshal([]byte(line), &fluentFields); err != nil {
-		return nil, BadLogFormatError{Format: "fluentbit", DecodingError: err}
-	}
-
-	return &fluentFields, nil
-}
-
-func decodeFluent(fluentLog *FluentLog, env string) (map[string]interface{}, error) {
-	out := map[string]interface{}{}
-	if fluentLog.Timestamp == nil || *fluentLog.Timestamp == "" {
-		return nil, BadLogFormatError{Format: "fluentbit", DecodingError: fmt.Errorf("no timestamp found in input field %s", fluentbitTimestampField)}
-	}
-
-	if timeValue, err := time.Parse(fluentTimeFormat, *fluentLog.Timestamp); err == nil {
-		out["timestamp"] = timeValue
-	} else {
-		return nil, BadLogFormatError{Format: "fluentbit", DecodingError: fmt.Errorf("timestamp has bad format: %s", *fluentLog.Timestamp)}
-	}
-
-	if fluentLog.Log == nil {
-		return nil, BadLogFormatError{Format: "fluentbit", DecodingError: fmt.Errorf("no log found in input field %s", fluentbitLogField)}
-	}
-	log := *fluentLog.Log
-	out["rawlog"] = log
-
-	out["decoder_msg_type"] = "fluentbit"
-	out["hostname"] = "aws-fargate"
-
-	// best effort to add container_env|app|task
-	if parts := strings.SplitN(fluentLog.TaskDefinition, "--", 3); len(parts) == 3 {
-		out["container_env"] = parts[0]
-		out["container_app"] = parts[1]
-	}
-	if idx := strings.LastIndex(fluentLog.TaskArn, "/"); idx != -1 {
-		out["container_task"] = fluentLog.TaskArn[idx+1:]
-	}
-
-	kvFields, err := FieldsFromKayvee(log)
-	if err == nil {
-		for k, v := range kvFields {
-			out[k] = v
-		}
-	}
-
-	// Inject additional fields that are useful in log-searching and other business logic; mimicking syslog behavior
-	out["env"] = env
-
-	return out, nil
-}
-
-type BadLogFormatError struct {
-	Format        string
-	DecodingError error
-}
-
-func (b BadLogFormatError) Error() string {
-	return fmt.Sprintf("trying to decode log as format %s: %v", b.Format, b.DecodingError)
-}
-
-func (b BadLogFormatError) Unwrap() error {
-	return b.DecodingError
+	if env == "" || app == "" || task == "" {
+		return map[string]string{}, fmt.Errorf("unable to get one or more of env/app/task")
+	}
+
+	return map[string]string{
+		"container_env":  env,
+		"container_app":  app,
+		"container_task": task,
+	}, nil
 }

decode/decode_test.go

@@ -1,7 +1,6 @@
 package decode
 
 import (
-	"errors"
 	"fmt"
 	"sort"
 	"testing"
@@ -222,10 +221,10 @@ func TestSyslogDecoding(t *testing.T) {
 			fields, err := FieldsFromSyslog(spec.Input)
 			if spec.ExpectedError != nil {
 				assert.Error(err)
-				assert.True(errors.As(err, &spec.ExpectedError))
-				return
+				assert.IsType(spec.ExpectedError, err)
+			} else {
+				assert.NoError(err)
 			}
-			assert.NoError(err)
 			assert.Equal(spec.ExpectedOutput, fields)
 		})
 	}
@@ -252,11 +251,6 @@ func TestParseAndEnhance(t *testing.T) {
 	}
 	logTime3 = logTime3.UTC()
 
-	logTime4, err := time.Parse(fluentTimeFormat, "2020-08-13T21:10:57.000+0000")
-	if err != nil {
-		t.Fatal(err)
-	}
-
 	specs := []ParseAndEnhanceSpec{
 		ParseAndEnhanceSpec{
 			Title: "Parses a Kayvee log line from an ECS app",
@@ -313,10 +307,10 @@ func TestParseAndEnhance(t *testing.T) {
 			ExpectedError: nil,
 		},
 		ParseAndEnhanceSpec{
-			Title:          "Fails to parse non-RSyslog, non-Fluent log line",
+			Title:          "Fails to parse non-RSyslog log line",
 			Line:           `not rsyslog`,
 			ExpectedOutput: map[string]interface{}{},
-			ExpectedError:  fmt.Errorf(""),
+			ExpectedError:  &syslogparser.ParserError{},
 		},
 		ParseAndEnhanceSpec{
 			Title: "Parses JSON values",
@@ -385,161 +379,6 @@ select sleep(2);`,
 				"user_id":          "868",
 			},
 		},
-		{
-			Title: "Valid fluent log",
-			Line: `{
-  "container_id": "b17752cdd47a480712eca6c9774d782f68b50a876446faec5e26c9e8846bd29e",
-  "container_name": "/ecs-env--app--d--2-2-env--app-e295a2dcf6f9ac849101",
-  "ecs_cluster": "arn:aws:ecs:us-east-1:589690932525:cluster/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1",
-  "ecs_task_arn": "arn:aws:ecs:us-east-1:589690932525:task/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1/7c1ccb5cb5c44582808b9e516b479eb6",
-  "ecs_task_definition": "env--app--d--2:2",
-  "fluent_ts": "2020-08-13T21:10:57.000+0000",
-  "log": "2020/08/13 21:10:57 {\"title\": \"request-finished\"}",
-  "source": "stderr"
-}`,
-			ExpectedOutput: map[string]interface{}{
-				"timestamp":        logTime4,
-				"rawlog":           "2020/08/13 21:10:57 {\"title\": \"request-finished\"}",
-				"hostname":         "aws-fargate",
-				"decoder_msg_type": "Kayvee",
-				"title":            "request-finished",
-				"container_app":    "app",
-				"container_env":    "env",
-				"container_task":   "7c1ccb5cb5c44582808b9e516b479eb6",
-				"prefix":           "2020/08/13 21:10:57 ",
-				"postfix":          "",
-				"env":              "deploy-env",
-			},
-			ExpectedError: nil,
-		},
-		{
-			Title: "fluent log missing ecs_task_definition still parses and has container_task",
-			Line: `{
-  "container_id": "b17752cdd47a480712eca6c9774d782f68b50a876446faec5e26c9e8846bd29e",
-  "container_name": "/ecs-env--app--d--2-2-env--app-e295a2dcf6f9ac849101",
-  "ecs_cluster": "arn:aws:ecs:us-east-1:589690932525:cluster/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1",
-  "ecs_task_arn": "arn:aws:ecs:us-east-1:589690932525:task/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1/7c1ccb5cb5c44582808b9e516b479eb6",
-  "fluent_ts": "2020-08-13T21:10:57.000+0000",
-  "log": "2020/08/13 21:10:57 {\"title\": \"request-finished\"}",
-  "source": "stderr"
-}`,
-			ExpectedOutput: map[string]interface{}{
-				"timestamp":        logTime4,
-				"rawlog":           "2020/08/13 21:10:57 {\"title\": \"request-finished\"}",
-				"hostname":         "aws-fargate",
-				"decoder_msg_type": "Kayvee",
-				"title":            "request-finished",
-				"container_task":   "7c1ccb5cb5c44582808b9e516b479eb6",
-				"prefix":           "2020/08/13 21:10:57 ",
-				"postfix":          "",
-				"env":              "deploy-env",
-			},
-		},
-		{
-			Title: "fluent log with bad ecs_task_definition format still parses and has container_task",
-			Line: `{
-  "container_id": "b17752cdd47a480712eca6c9774d782f68b50a876446faec5e26c9e8846bd29e",
-  "container_name": "/ecs-env--app--d--2-2-env--app-e295a2dcf6f9ac849101",
-  "ecs_cluster": "arn:aws:ecs:us-east-1:589690932525:cluster/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1",
-  "ecs_task_arn": "arn:aws:ecs:us-east-1:589690932525:task/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1/7c1ccb5cb5c44582808b9e516b479eb6",
-  "ecs_task_definition": "env--app",
-  "fluent_ts": "2020-08-13T21:10:57.000+0000",
-  "log": "2020/08/13 21:10:57 {\"title\": \"request-finished\"}",
-  "source": "stderr"
-}`,
-			ExpectedOutput: map[string]interface{}{
-				"timestamp":        logTime4,
-				"rawlog":           "2020/08/13 21:10:57 {\"title\": \"request-finished\"}",
-				"hostname":         "aws-fargate",
-				"decoder_msg_type": "Kayvee",
-				"title":            "request-finished",
-				"container_task":   "7c1ccb5cb5c44582808b9e516b479eb6",
-				"prefix":           "2020/08/13 21:10:57 ",
-				"postfix":          "",
-				"env":              "deploy-env",
-			},
-		},
-		{
-			Title: "fluent log missing ecs_task_arn still parses and has containter_env and app",
-			Line: `{
-  "container_id": "b17752cdd47a480712eca6c9774d782f68b50a876446faec5e26c9e8846bd29e",
-  "container_name": "/ecs-env--app--d--2-2-env--app-e295a2dcf6f9ac849101",
-  "ecs_cluster": "arn:aws:ecs:us-east-1:589690932525:cluster/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1",
-  "ecs_task_definition": "env--app--d--2:2",
-  "fluent_ts": "2020-08-13T21:10:57.000+0000",
-  "log": "2020/08/13 21:10:57 {\"title\": \"request-finished\"}",
-  "source": "stderr"
-}`,
-			ExpectedOutput: map[string]interface{}{
-				"timestamp":        logTime4, // time.Date(2020, 8, 13, 21, 10, 57, 0, time.UTC),
-				"rawlog":           "2020/08/13 21:10:57 {\"title\": \"request-finished\"}",
-				"hostname":         "aws-fargate",
-				"decoder_msg_type": "Kayvee",
-				"title":            "request-finished",
-				"container_env":    "env",
-				"container_app":    "app",
-				"prefix":           "2020/08/13 21:10:57 ",
-				"postfix":          "",
-				"env":              "deploy-env",
-			},
-		},
-		{
-			Title: "fluent log missing timestamp is an error",
-			Line: `{
-  "container_id": "b17752cdd47a480712eca6c9774d782f68b50a876446faec5e26c9e8846bd29e",
-  "container_name": "/ecs-env--app--d--2-2-env--app-e295a2dcf6f9ac849101",
-  "ecs_cluster": "arn:aws:ecs:us-east-1:589690932525:cluster/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1",
-  "ecs_task_arn": "arn:aws:ecs:us-east-1:589690932525:task/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1/7c1ccb5cb5c44582808b9e516b479eb6",
-  "ecs_task_definition": "env--app--d--2:2",
-  "ts": "2020-08-13T21:10:57.000+0000",
-  "log": "2020/08/13 21:10:57 {\"deploy_env\":\"env\",\"level\":\"info\",\"pod-account\":\"589690932525\",\"pod-id\":\"2870425d-e4b1-4869-b2b7-a6d9fade9be1\",\"pod-region\":\"us-east-1\",\"pod-shortname\":\"us-east-1-dev-canary-2870425d\",\"source\":\"app-service\",\"title\":\"NumFDs\",\"type\":\"gauge\",\"value\":33,\"via\":\"process-metrics\"}",
-  "source": "stderr"
-}`,
-			ExpectedOutput: nil,
-			ExpectedError:  BadLogFormatError{},
-		},
-		{
-			Title: "fluent log with bad timestamp format is an error",
-			Line: `{
-  "container_id": "b17752cdd47a480712eca6c9774d782f68b50a876446faec5e26c9e8846bd29e",
-  "container_name": "/ecs-env--app--d--2-2-env--app-e295a2dcf6f9ac849101",
-  "ecs_cluster": "arn:aws:ecs:us-east-1:589690932525:cluster/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1",
-  "ecs_task_arn": "arn:aws:ecs:us-east-1:589690932525:task/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1/7c1ccb5cb5c44582808b9e516b479eb6",
-  "ecs_task_definition": "env--app--d--2:2",
-  "fluent_ts": "2020-08-13T21:10:57.000Z",
-  "log": "2020/08/13 21:10:57 {\"deploy_env\":\"env\",\"level\":\"info\",\"pod-account\":\"589690932525\",\"pod-id\":\"2870425d-e4b1-4869-b2b7-a6d9fade9be1\",\"pod-region\":\"us-east-1\",\"pod-shortname\":\"us-east-1-dev-canary-2870425d\",\"source\":\"app-service\",\"title\":\"NumFDs\",\"type\":\"gauge\",\"value\":33,\"via\":\"process-metrics\"}",
-  "source": "stderr"
-}`,
-			ExpectedOutput: nil,
-			ExpectedError:  BadLogFormatError{},
-		},
-		{
-			Title: "Valid fluent log with overrides for container_env, app, and task",
-			Line: `{
-  "container_id": "b17752cdd47a480712eca6c9774d782f68b50a876446faec5e26c9e8846bd29e",
-  "container_name": "/ecs-env--app--d--2-2-env--app-e295a2dcf6f9ac849101",
-  "ecs_cluster": "arn:aws:ecs:us-east-1:589690932525:cluster/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1",
-  "ecs_task_arn": "arn:aws:ecs:us-east-1:589690932525:task/pod-2870425d-e4b1-4869-b2b7-a6d9fade9be1/7c1ccb5cb5c44582808b9e516b479eb6",
-  "ecs_task_definition": "env--app--d--2:2",
-  "fluent_ts": "2020-08-13T21:10:57.000+0000",
-  "log": "2020/08/13 21:10:57 {\"title\": \"request-finished\", \"container_env\": \"env2\", \"container_task\": \"task\", \"container_app\": \"app2\"}",
-  "source": "stderr"
-}`,
-			ExpectedOutput: map[string]interface{}{
-				"timestamp":        logTime4,
-				"rawlog":           "2020/08/13 21:10:57 {\"title\": \"request-finished\", \"container_env\": \"env2\", \"container_task\": \"task\", \"container_app\": \"app2\"}",
-				"hostname":         "aws-fargate",
-				"decoder_msg_type": "Kayvee",
-				"title":            "request-finished",
-				"container_app":    "app2",
-				"container_env":    "env2",
-				"container_task":   "task",
-				"prefix":           "2020/08/13 21:10:57 ",
-				"postfix":          "",
-				"env":              "deploy-env",
-			},
-			ExpectedError: nil,
-		},
 	}
 	for _, spec := range specs {
 		t.Run(fmt.Sprintf(spec.Title), func(t *testing.T) {
@@ -547,114 +386,72 @@ select sleep(2);`,
 			fields, err := ParseAndEnhance(spec.Line, "deploy-env")
 			if spec.ExpectedError != nil {
 				assert.Error(err)
-				assert.True(errors.As(err, &spec.ExpectedError))
-				return
+				assert.IsType(spec.ExpectedError, err)
+			} else {
+				assert.NoError(err)
 			}
-			assert.NoError(err)
 			assert.Equal(spec.ExpectedOutput, fields)
 		})
 	}
 }
 
 func TestGetContainerMeta(t *testing.T) {
+	assert := assert.New(t)
 
-	type containerMetaSpec struct {
-		description       string
-		input             map[string]interface{}
-		wantErr           bool
-		wantContainerEnv  string
-		wantContainerApp  string
-		wantContainerTask string
-	}
+	t.Log("Must have a programname to get container meta")
+	programname := ""
+	_, err := getContainerMeta(programname, "", "", "")
+	assert.Error(err)
 
-	tests := []containerMetaSpec{
-		{
-			description: "Must have a programname to get container meta",
-			input: map[string]interface{}{
-				"programname": "",
-			},
-			wantErr: true,
-		},
-		{
-			description: "Can parse a programname",
-			input: map[string]interface{}{
-				"programname": `env1--app2/arn%3Aaws%3Aecs%3Aus-west-1%3A589690932525%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef`,
-			},
-			wantContainerEnv:  "env1",
-			wantContainerApp:  "app2",
-			wantContainerTask: "abcd1234-1a3b-1a3b-1234-d76552f4b7ef",
-		},
-		{
-			description: "Can override just 'env'",
-			input: map[string]interface{}{
-				"programname":   `env1--app2/arn%3Aaws%3Aecs%3Aus-west-1%3A589690932525%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef`,
-				"container_env": "force-env",
-			},
-			wantContainerEnv:  "force-env",
-			wantContainerApp:  "app2",
-			wantContainerTask: "abcd1234-1a3b-1a3b-1234-d76552f4b7ef",
-		},
-		{
-			description: "Can override just 'app'",
-			input: map[string]interface{}{
-				"programname":   `env1--app2/arn%3Aaws%3Aecs%3Aus-west-1%3A589690932525%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef`,
-				"container_app": "force-app",
-			},
-			wantContainerEnv:  "env1",
-			wantContainerApp:  "force-app",
-			wantContainerTask: "abcd1234-1a3b-1a3b-1234-d76552f4b7ef",
-		},
-		{
-			description: "Can override just 'task'",
-			input: map[string]interface{}{
-				"programname":    `env1--app2/arn%3Aaws%3Aecs%3Aus-west-1%3A589690932525%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef`,
-				"container_task": "force-task",
-			},
-			wantContainerEnv:  "env1",
-			wantContainerApp:  "app2",
-			wantContainerTask: "force-task",
-		},
-		{
-			description: "Can override all fields",
-			input: map[string]interface{}{
-				"programname":    `env1--app2/arn%3Aaws%3Aecs%3Aus-west-1%3A589690932525%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef`,
-				"container_task": "force-task",
-				"container_app":  "force-app",
-				"container_env":  "force-env",
-			},
-			wantContainerEnv:  "force-env",
-			wantContainerApp:  "force-app",
-			wantContainerTask: "force-task",
-		},
-	}
+	t.Log("Can parse a programname")
+	programname = `env1--app2/arn%3Aaws%3Aecs%3Aus-west-1%3A589690932525%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef`
+	meta, err := getContainerMeta(programname, "", "", "")
+	assert.NoError(err)
+	assert.Equal(map[string]string{
+		"container_env":  "env1",
+		"container_app":  "app2",
+		"container_task": "abcd1234-1a3b-1a3b-1234-d76552f4b7ef",
+	}, meta)
 
-	for _, tcase := range tests {
-		assert := assert.New(t)
+	t.Log("Can override just 'env'")
+	overrideEnv := "force-env"
+	meta, err = getContainerMeta(programname, overrideEnv, "", "")
+	assert.NoError(err)
+	assert.Equal(map[string]string{
+		"container_env":  overrideEnv,
+		"container_app":  "app2",
+		"container_task": "abcd1234-1a3b-1a3b-1234-d76552f4b7ef",
+	}, meta)
 
-		syslog := map[string]interface{}{}
-		for k, v := range tcase.input {
-			syslog[k] = v
-		}
+	t.Log("Can override just 'app'")
+	overrideApp := "force-app"
+	meta, err = getContainerMeta(programname, "", overrideApp, "")
+	assert.NoError(err)
+	assert.Equal(map[string]string{
+		"container_env":  "env1",
+		"container_app":  overrideApp,
+		"container_task": "abcd1234-1a3b-1a3b-1234-d76552f4b7ef",
+	}, meta)
 
-		newSyslog, err := addContainterMetaToSyslog(syslog)
-		if tcase.wantErr {
-			assert.Error(err)
-			continue
-		}
-		for k, v := range newSyslog {
-			switch k {
-			case "container_env":
-				assert.Equal(v, tcase.wantContainerEnv)
-			case "container_app":
-				assert.Equal(v, tcase.wantContainerApp)
-			case "container_task":
-				assert.Equal(v, tcase.wantContainerTask)
-			default:
-				assert.Equal(v, tcase.input[k])
-			}
-		}
-	}
+	t.Log("Can override just 'task'")
+	overrideTask := "force-task"
+	meta, err = getContainerMeta(programname, "", "", overrideTask)
+	assert.NoError(err)
+	assert.Equal(map[string]string{
+		"container_env":  "env1",
+		"container_app":  "app2",
+		"container_task": overrideTask,
+	}, meta)
 
+	t.Log("Can override all fields")
+	programname = `env--app/arn%3Aaws%3Aecs%3Aus-west-1%3A999988887777%3Atask%2Fabcd1234-1a3b-1a3b-1234-d76552f4b7ef`
+	meta, err = getContainerMeta(programname, overrideEnv, overrideApp, overrideTask)
+	assert.NoError(err)
+	assert.Equal(map[string]string{
+		"container_env":  overrideEnv,
+		"container_app":  overrideApp,
+		"container_task": overrideTask,
+	}, meta)
 }
 
 func TestExtractKVMeta(t *testing.T) {

go.mod

@@ -1,22 +0,0 @@
-module github.com/amazon-kinesis-client-go
-
-go 1.13
-
-require (
-	github.com/Clever/amazon-kinesis-client-go v1.0.0
-	github.com/Clever/syslogparser v0.0.0-20170816194131-fb28ad3e4340
-	github.com/a8m/kinesis-producer v0.2.0
-	github.com/aws/aws-sdk-go v1.35.28 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/golang/protobuf v1.5.2
-	github.com/jeromer/syslogparser v0.0.0-20190429161531-5fbaaf06d9e7 // indirect
-	github.com/jpillora/backoff v0.0.0-20170918002102-8eab2debe79d // indirect
-	github.com/stretchr/testify v1.7.0
-	github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect
-	github.com/xeipuuv/gojsonschema v1.2.1-0.20200424115421-065759f9c3d7 // indirect
-	golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e
-	gopkg.in/Clever/kayvee-go.v6 v6.24.1
-	gopkg.in/yaml.v2 v2.3.1-0.20200602174213-b893565b90ca // indirect
-	gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776 // indirect
-	launchpad.net/gocheck v0.0.0-20140225173054-000000000087 // indirect
-)

go.sum

@@ -1,63 +0,0 @@
-github.com/Clever/amazon-kinesis-client-go v1.0.0 h1:rG+hkHCSe+xMGk3asESg18tiEj8X6To8vj0TRa1pKvQ=
-github.com/Clever/amazon-kinesis-client-go v1.0.0/go.mod h1:jLfi8QusUjdsYazmEH0DlSO4gE1WRzVE7cs4c/fOFdI=
-github.com/Clever/syslogparser v0.0.0-20170816194131-fb28ad3e4340 h1:wr8lTqPJZQpZkZgllQtrD96SecXkAu5MxzY3yJgsuCg=
-github.com/Clever/syslogparser v0.0.0-20170816194131-fb28ad3e4340/go.mod h1:e7Yy7RTiIMU9pZ+dcSviX3cpod8e0CEeTUPTgBbKlRE=
-github.com/a8m/kinesis-producer v0.2.0 h1:Bd5Oi4dczbTLPIZwVbm02464LIFgBqmViFj//b098xc=
-github.com/a8m/kinesis-producer v0.2.0/go.mod h1:CxoFe0Y49udKMnQPkC5S5VmZZy6a+Bef9otuoH96Pv0=
-github.com/aws/aws-sdk-go v1.35.28 h1:S2LuRnfC8X05zgZLC8gy/Sb82TGv2Cpytzbzz7tkeHc=
-github.com/aws/aws-sdk-go v1.35.28/go.mod h1:tlPOdRjfxPBpNIwqDj61rmsnA85v9jc0Ps9+muhnW+k=
-github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
-github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
-github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw=
-github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
-github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
-github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/jeromer/syslogparser v0.0.0-20190429161531-5fbaaf06d9e7 h1:FRDp/rRU2sThfssdHIUATCKJOms8R6A9rrhdC1sLqFI=
-github.com/jeromer/syslogparser v0.0.0-20190429161531-5fbaaf06d9e7/go.mod h1:mQyv/QAgjs9+PTi/iXveno+U86nKGsltjqf3ilYx4Bg=
-github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=
-github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
-github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8=
-github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
-github.com/jpillora/backoff v0.0.0-20170918002102-8eab2debe79d h1:ix3WmphUvN0GDd0DO9MH0v6/5xTv+Xm1bPN+1UJn58k=
-github.com/jpillora/backoff v0.0.0-20170918002102-8eab2debe79d/go.mod h1:2iMrUgbbvHEiQClaW2NsSzMyGHqN+rDFqY705q49KG0=
-github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
-github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
-github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
-github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
-github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb h1:zGWFAtiMcyryUHoUjUJX0/lt1H2+i2Ka2n+D3DImSNo=
-github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
-github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 h1:EzJWgHovont7NscjpAxXsDA8S8BMYve8Y5+7cuRE7R0=
-github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
-github.com/xeipuuv/gojsonschema v1.2.1-0.20200424115421-065759f9c3d7 h1:tdnG+ZILOafvA29+pYKP3gauEbigHll3PHsf1GQa2ms=
-github.com/xeipuuv/gojsonschema v1.2.1-0.20200424115421-065759f9c3d7/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
-golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/net v0.0.0-20200202094626-16171245cfb2 h1:CCH4IOTTfewWjGOlSp+zGcjutRKlBEZQ6wTn8ozI/nI=
-golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
-golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
-golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e h1:EHBhcS0mlXEAVwNyO2dLfjToGsyY4j24pTs2ScHnX7s=
-golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
-google.golang.org/protobuf v1.26.0 h1:bxAC2xTBsZGibn2RTntX0oH50xLsqy1OxA9tTL3p/lk=
-google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-gopkg.in/Clever/kayvee-go.v6 v6.24.1 h1:lHxjlTFj57mfrq06PiCKIAJ0QuHHEqsyvSYvzb+gSFg=
-gopkg.in/Clever/kayvee-go.v6 v6.24.1/go.mod h1:G0m6nBZj7Kdz+w2hiIaawmhXl5zp7E/K0ashol3Kb2A=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v2 v2.3.1-0.20200602174213-b893565b90ca h1:oivFrl3Vo+KfpUmTDJvz91I+BWzDPOQ+0CNR5jwTHcg=
-gopkg.in/yaml.v2 v2.3.1-0.20200602174213-b893565b90ca/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776 h1:tQIYjPdBoyREyB9XMu+nnTclpTYkz2zFM+lzLJFO4gQ=
-gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-launchpad.net/gocheck v0.0.0-20140225173054-000000000087 h1:Izowp2XBH6Ya6rv+hqbceQyw/gSGoXfH/UPoTGduL54=
-launchpad.net/gocheck v0.0.0-20140225173054-000000000087/go.mod h1:hj7XX3B/0A+80Vse0e+BUHsHMTEhd0O4cpUHr/e/BUM=

golang.mk

@@ -1,18 +1,15 @@
 # This is the default Clever Golang Makefile.
 # It is stored in the dev-handbook repo, github.com/Clever/dev-handbook
 # Please do not alter this file directly.
-GOLANG_MK_VERSION := 1.0.0
+GOLANG_MK_VERSION := 0.4.0
 
 SHELL := /bin/bash
 SYSTEM := $(shell uname -a | cut -d" " -f1 | tr '[:upper:]' '[:lower:]')
-.PHONY: golang-test-deps golang-ensure-curl-installed
+.PHONY: golang-test-deps bin/dep golang-ensure-curl-installed
 
 # set timezone to UTC for golang to match circle and deploys
 export TZ=UTC
 
-# go build flags for use across all commands which accept them
-GO_BUILD_FLAGS := "-mod=vendor"
-
 # if the gopath includes several directories, use only the first
 GOPATH=$(shell echo $$GOPATH | cut -d: -f1)
 
@@ -35,21 +32,57 @@ _ := $(if  \
 endef
 
 # FGT is a utility that exits with 1 whenever any stderr/stdout output is recieved.
-# We pin its version since its a simple tool that does its job as-is;
-# so we're defended against it breaking or changing in the future.
 FGT := $(GOPATH)/bin/fgt
 $(FGT):
-	go get github.com/GeertJohan/fgt@262f7b11eec07dc7b147c44641236f3212fee89d
+	go get github.com/GeertJohan/fgt
 
 golang-ensure-curl-installed:
 	@command -v curl >/dev/null 2>&1 || { echo >&2 "curl not installed. Please install curl."; exit 1; }
 
+DEP_VERSION = v0.4.1
+DEP_INSTALLED := $(shell [[ -e "bin/dep" ]] && bin/dep version | grep version | grep -v go | cut -d: -f2 | tr -d '[:space:]')
+# Dep is a tool used to manage Golang dependencies. It is the offical vendoring experiment, but
+# not yet the official tool for Golang.
+ifeq ($(DEP_VERSION),$(DEP_INSTALLED))
+bin/dep: # nothing to do, dep is already up-to-date
+else
+CACHED_DEP = /tmp/dep-$(DEP_VERSION)
+bin/dep: golang-ensure-curl-installed
+	@echo "Updating dep..."
+	@mkdir -p bin
+	@if [ ! -f $(CACHED_DEP) ]; then curl -o $(CACHED_DEP) -sL https://github.com/golang/dep/releases/download/$(DEP_VERSION)/dep-$(SYSTEM)-amd64; fi;
+	@cp $(CACHED_DEP) bin/dep
+	@chmod +x bin/dep || true
+endif
+
+# figure out "github.com/<org>/<repo>"
+# `go list` will fail if there are no .go files in the directory
+# if this is the case, fall back to assuming github.com/Clever
+REF = $(shell go list || echo github.com/Clever/$(notdir $(shell pwd)))
+golang-verify-no-self-references:
+	@if grep -q -i "$(REF)" Gopkg.lock; then echo "Error: Gopkg.lock includes a self-reference ($(REF)), which is not allowed. See: https://github.com/golang/dep/issues/1690" && exit 1; fi;
+	@if grep -q -i "$(REF)" Gopkg.toml; then echo "Error: Gopkg.toml includes a self-reference ($(REF)), which is not allowed. See: https://github.com/golang/dep/issues/1690" && exit 1; fi;
+
+golang-dep-vendor-deps: bin/dep golang-verify-no-self-references
+
+# golang-godep-vendor is a target for saving dependencies with the dep tool
+# to the vendor/ directory. All nested vendor/ directories are deleted via
+# the prune command.
+# In CI, -vendor-only is used to avoid updating the lock file.
+ifndef CI
+define golang-dep-vendor
+bin/dep ensure -v
+endef
+else
+define golang-dep-vendor
+bin/dep ensure -v -vendor-only
+endef
+endif
+
 # Golint is a tool for linting Golang code for common errors.
-# We pin its version because an update could add a new lint check which would make
-# previously passing tests start failing without changing our code.
 GOLINT := $(GOPATH)/bin/golint
 $(GOLINT):
-	go get golang.org/x/lint/golint@738671d3881b9731cc63024d5d88cf28db875626
+	go get golang.org/x/lint/golint
 
 # golang-fmt-deps requires the FGT tool for checking output
 golang-fmt-deps: $(FGT)
@@ -58,7 +91,7 @@ golang-fmt-deps: $(FGT)
 # arg1: pkg path
 define golang-fmt
 @echo "FORMATTING $(1)..."
-@PKG_PATH=$$(go list -f '{{.Dir}}' $(1)); $(FGT) gofmt -l=true $${PKG_PATH}/*.go
+@$(FGT) gofmt -l=true $(GOPATH)/src/$(1)/*.go
 endef
 
 # golang-lint-deps requires the golint tool for golang linting.
@@ -68,7 +101,7 @@ golang-lint-deps: $(GOLINT)
 # arg1: pkg path
 define golang-lint
 @echo "LINTING $(1)..."
-@PKG_PATH=$$(go list -f '{{.Dir}}' $(1)); find $${PKG_PATH}/*.go -type f | grep -v gen_ | xargs $(GOLINT)
+@find $(GOPATH)/src/$(1)/*.go -type f | grep -v gen_ | xargs $(GOLINT)
 endef
 
 # golang-lint-deps-strict requires the golint tool for golang linting.
@@ -79,7 +112,7 @@ golang-lint-deps-strict: $(GOLINT) $(FGT)
 # arg1: pkg path
 define golang-lint-strict
 @echo "LINTING $(1)..."
-@PKG_PATH=$$(go list -f '{{.Dir}}' $(1)); find $${PKG_PATH}/*.go -type f | grep -v gen_ | xargs $(FGT) $(GOLINT)
+@find $(GOPATH)/src/$(1)/*.go -type f | grep -v gen_ | xargs $(FGT) $(GOLINT)
 endef
 
 # golang-test-deps is here for consistency
@@ -89,7 +122,7 @@ golang-test-deps:
 # arg1: pkg path
 define golang-test
 @echo "TESTING $(1)..."
-@go test $(GO_BUILD_FLAGS) -v $(1)
+@go test -v $(1)
 endef
 
 # golang-test-strict-deps is here for consistency
@@ -99,7 +132,7 @@ golang-test-strict-deps:
 # arg1: pkg path
 define golang-test-strict
 @echo "TESTING $(1)..."
-@go test -v $(GO_BUILD_FLAGS) -race $(1)
+@go test -v -race $(1)
 endef
 
 # golang-vet-deps is here for consistency
@@ -109,7 +142,7 @@ golang-vet-deps:
 # arg1: pkg path
 define golang-vet
 @echo "VETTING $(1)..."
-@go vet $(GO_BUILD_FLAGS) $(1)
+@go vet $(GOPATH)/src/$(1)/*.go
 endef
 
 # golang-test-all-deps installs all dependencies needed for different test cases.
@@ -143,14 +176,14 @@ endef
 define golang-build
 @echo "BUILDING..."
 @if [ -z "$$CI" ]; then \
-	go build $(GO_BUILD_FLAGS) -o bin/$(2) $(1); \
+	go build -o bin/$(2) $(1); \
 else \
 	echo "-> Building CGO binary"; \
-	CGO_ENABLED=0 go build $(GO_BUILD_FLAGS) -installsuffix cgo -o bin/$(2) $(1); \
+	CGO_ENABLED=0 go build -installsuffix cgo -o bin/$(2) $(1); \
 fi;
 endef
 
 # golang-update-makefile downloads latest version of golang.mk
 golang-update-makefile:
-	@wget https://raw.githubusercontent.com/Clever/dev-handbook/master/make/golang-v1.mk -O /tmp/golang.mk 2>/dev/null
+	@wget https://raw.githubusercontent.com/Clever/dev-handbook/master/make/golang.mk -O /tmp/golang.mk 2>/dev/null
 	@if ! grep -q $(GOLANG_MK_VERSION) /tmp/golang.mk; then cp /tmp/golang.mk golang.mk && echo "golang.mk updated"; else echo "golang.mk is up-to-date"; fi

splitter/splitter.go

@@ -1,150 +1,16 @@
-// Package splitter provides functions for decoding various kinds of records that might come off of a kinesis stream.
-// It is equipped to with the functions to unbundle KPL aggregates and CloudWatch log bundles,
-// as well as apply appropriate decompression.
-// KCL applications would be most interested in `SplitMessageIfNecessary` which can handle zlibbed records as well as
-// CloudWatch bundles. KCL automatically unbundles KPL aggregates before passing the records to the consumer.
-// Non-KCL applications (such as Lambdas consuming KPL-produced aggregates) should either use
-// - KPLDeaggregate if the consumer purely wants to unbundle KPL aggregates, but will handle the raw records themselves.
-// - Deaggregate if the consumer wants to apply the same decompress and split logic as SplitMessageIfNecessary
-//   in addition to the KPL splitting.
 package splitter
 
 import (
 	"bytes"
 	"compress/gzip"
-	"compress/zlib"
-	"crypto/md5"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"regexp"
 	"strconv"
 	"time"
-
-	kpl "github.com/a8m/kinesis-producer"
-	"github.com/golang/protobuf/proto"
 )
 
-// The Amazon Kinesis Producer Library (KPL) aggregates multiple logical user records into a single
-// Amazon Kinesis record for efficient puts.
-// https://github.com/awslabs/amazon-kinesis-producer/blob/master/aggregation-format.md
-var kplMagicNumber = []byte{0xF3, 0x89, 0x9A, 0xC2}
-
-// IsKPLAggregate checks a record for a KPL aggregate prefix.
-// It is not necessary to call this before calling KPLDeaggregate.
-func IsKPLAggregate(data []byte) bool {
-	return bytes.HasPrefix(data, kplMagicNumber)
-}
-
-// KPLDeaggregate takes a Kinesis record and converts it to one or more user records by applying KPL deaggregation.
-// If the record begins with the 4-byte magic prefix that KPL uses, the single Kinesis record is split into its component user records.
-// Otherwise, the return value is a singleton containing the original record.
-func KPLDeaggregate(kinesisRecord []byte) ([][]byte, error) {
-	if !IsKPLAggregate(kinesisRecord) {
-		return [][]byte{kinesisRecord}, nil
-	}
-	src := kinesisRecord[len(kplMagicNumber) : len(kinesisRecord)-md5.Size]
-	checksum := kinesisRecord[len(kinesisRecord)-md5.Size:]
-	recordSum := md5.Sum(src)
-	for i, b := range checksum {
-		if b != recordSum[i] {
-			// either the data is corrupted or this is not a KPL aggregate
-			// either way, return the data as-is
-			return [][]byte{kinesisRecord}, nil
-		}
-	}
-	dest := new(kpl.AggregatedRecord)
-	err := proto.Unmarshal(src, dest)
-	if err != nil {
-		return nil, fmt.Errorf("unmarshalling proto: %v", err)
-	}
-	var records [][]byte
-	for _, userRecord := range dest.GetRecords() {
-		records = append(records, userRecord.Data)
-	}
-	return records, nil
-}
-
-// Deaggregate is a combination of KPLDeaggregate and SplitMessageIfNecessary
-// First it tries to KPL-deaggregate. If unsuccessful, it calls SplitIfNecessary on the original record.
-// If successful, it iterates over the individual user records and attempts to unzlib them.
-// If a record inside an aggregate is in zlib format, the output will contain the unzlibbed version.
-// If it is not zlibbed, the output will contain the record verbatim
-// A similar result can be optained by calling KPLDeaggregate, then iterating over the results and callin SplitMessageIfNecessary.
-// This function makes the assumption that after KPL-deaggregating, the results are not CloudWatch aggregates, so it doesn't need to check them for a gzip header.
-// Also it lets us iterate over the user records one less time, since KPLDeaggregate loops over the records and we would need to loop again to unzlib.
-//
-// See the SplitMessageIfNecessary documentation for the format of output for CloudWatch log bundles.
-func Deaggregate(kinesisRecord []byte) ([][]byte, error) {
-	if !IsKPLAggregate(kinesisRecord) {
-		return SplitMessageIfNecessary(kinesisRecord)
-	}
-	src := kinesisRecord[len(kplMagicNumber) : len(kinesisRecord)-md5.Size]
-	checksum := kinesisRecord[len(kinesisRecord)-md5.Size:]
-	recordSum := md5.Sum(src)
-	for i, b := range checksum {
-		if b != recordSum[i] {
-			// either the data is corrupted or this is not a KPL aggregate
-			// either way, return the data as-is
-			return [][]byte{kinesisRecord}, nil
-		}
-	}
-	dest := new(kpl.AggregatedRecord)
-	err := proto.Unmarshal(src, dest)
-	if err != nil {
-		return nil, fmt.Errorf("unmarshalling proto: %v", err)
-	}
-	var records [][]byte
-	for _, userRecord := range dest.GetRecords() {
-		record, err := unzlib(userRecord.Data)
-		if err != nil {
-			return nil, fmt.Errorf("unzlibbing record: %w", err)
-		}
-		records = append(records, record)
-	}
-	return records, nil
-}
-
-// SplitMessageIfNecessary recieves a user-record and returns a slice of one or more records.
-// if the record is coming off of a kinesis stream and might be KPL aggregated, it needs to be deaggregated before calling this.
-// This function handles three types of records:
-// - records emitted from CWLogs Subscription (which are gzip compressed)
-// - zlib compressed records (e.g. as compressed and emitted by Kinesis plugin for Fluent Bit
-// - any other record (left unchanged)
-//
-// CloudWatch logs come as structured JSON. In the process of splitting, they are converted
-// into an rsyslog format that allows fairly uniform parsing of the result across the
-// AWS services that might emit logs to CloudWatch.
-// Note that these timezone used in these syslog records is guessed based on the local env.
-// If you need consistent timezones, set TZ=UTC in your environment.
-func SplitMessageIfNecessary(userRecord []byte) ([][]byte, error) {
-	// First try the record as a CWLogs record
-	if IsGzipped(userRecord) {
-		return GetMessagesFromGzippedInput(userRecord)
-	}
-
-	unzlibRecord, err := unzlib(userRecord)
-	if err != nil {
-		return nil, err
-	}
-
-	// Process a single message, from KPL
-	return [][]byte{unzlibRecord}, nil
-}
-
-func unzlib(input []byte) ([]byte, error) {
-	zlibReader, err := zlib.NewReader(bytes.NewReader(input))
-	if err == nil {
-		unzlibRecord, err := ioutil.ReadAll(zlibReader)
-		if err != nil {
-			return nil, fmt.Errorf("reading zlib-compressed record: %v", err)
-		}
-		return unzlibRecord, nil
-	}
-	return input, nil
-
-}
-
 // LogEvent is a single log line within a LogEventBatch
 type LogEvent struct {
 	ID        string              `json:"id"`
@@ -250,11 +116,6 @@ var awsFargateLogStreamRegex = regexp.MustCompile(`^fargate/([a-z0-9-]+)--([a-z0
 // RDS slowquery log groups are in the form of /aws/rds/cluster/<database name>/slowquery
 var awsRDSLogGroupRegex = regexp.MustCompile(`^/aws/rds/cluster/([a-z0-9-]+)/slowquery$`)
 
-// glue log groups are of the form /aws-glue/jobs/<env>/<app>/<pod-id>
-// glue log streams are of the form <job id>-<"driver" | "1" | "progress-bar">
-var awsGlueLogGroupRegex = regexp.MustCompile(`^/aws-glue/jobs/([a-z0-9-]+)/([a-z0-9-]+)/([a-z0-9-]+)$`)
-var awsGlueLogStreamRegex = regexp.MustCompile(`^(jr_[a-z0-9-]+)-.*$`)
-
 // arn and task cruft to satisfy parsing later on: https://github.com/Clever/amazon-kinesis-client-go/blob/94aacdf8339bd2cc8400d3bcb323dc1bce2c8422/decode/decode.go#L421-L425
 const arnCruft = `/arn%3Aaws%3Aecs%3Aus-east-1%3A999988887777%3Atask%2F`
 const taskCruft = `12345678-1234-1234-1234-555566667777`
@@ -367,32 +228,6 @@ func splitAWSRDS(b LogEventBatch) ([]RSysLogMessage, bool) {
 	return out, true
 }
 
-func splitAWSGlue(b LogEventBatch) ([]RSysLogMessage, bool) {
-	matches := awsGlueLogGroupRegex.FindAllStringSubmatch(b.LogGroup, 1)
-	if len(matches) != 1 {
-		return nil, false
-	}
-	env := matches[0][1]
-	app := matches[0][2]
-
-	streamMatches := awsGlueLogStreamRegex.FindAllStringSubmatch(b.LogStream, 1)
-	if len(streamMatches) != 1 {
-		return nil, false
-	}
-	jobID := streamMatches[0][1]
-
-	out := []RSysLogMessage{}
-	for _, event := range b.LogEvents {
-		out = append(out, RSysLogMessage{
-			Timestamp:   event.Timestamp.Time(),
-			ProgramName: env + "--" + app + arnCruft + jobID,
-			Hostname:    "aws-glue",
-			Message:     event.Message,
-		})
-	}
-	return out, true
-}
-
 func splitDefault(b LogEventBatch) []RSysLogMessage {
 	out := []RSysLogMessage{}
 	for _, event := range b.LogEvents {
@@ -426,8 +261,6 @@ func Split(b LogEventBatch) [][]byte {
 		return stringify(rsyslogMsgs)
 	} else if rsyslogMsgs, ok := splitAWSRDS(b); ok {
 		return stringify(rsyslogMsgs)
-	} else if rsyslogMsgs, ok := splitAWSGlue(b); ok {
-		return stringify(rsyslogMsgs)
 	}
 	return stringify(splitDefault(b))
 }

splitter/splitter_test.go

@@ -3,29 +3,15 @@ package splitter
 import (
 	"bytes"
 	"compress/gzip"
-	"compress/zlib"
-	"crypto/md5"
 	b64 "encoding/base64"
 	"encoding/json"
-	"os"
 	"testing"
-	"time"
 
 	"github.com/Clever/amazon-kinesis-client-go/decode"
-	kpl "github.com/a8m/kinesis-producer"
-	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-func TestMain(m *testing.M) {
-	// In the conversion of CloudWatch LogEvent struct to an RSyslog struct to a string,
-	// the timezone used in the final string depends on the locally set timezone.
-	// in order for tests to pass, we set TZ to UTC
-	os.Setenv("TZ", "UTC")
-	os.Exit(m.Run())
-}
-
 func TestUnpacking(t *testing.T) {
 	input := "H4sIAAAAAAAAADWOTQuCQBRF/8ow6wj6ENRdhLXIClJoERKTvsZHOiPzxiLE/96YtTzcy72n4zUQCQnpuwEe8vXxkJ6O8XUfJclqG/EJ1y8FZkgq3RYvYfMy1pJcUGm5NbptXDZSYg2IekRqb5QbbCxqtcHKgiEeXrJvL3qCsgN2HIuxbtFpWFG7sdky8L1ZECwXc9+b/PUGgXPMfnrspxeydQn5A5VkJYjKlkzfWeGWUInhme1QASEx+qpNeZ/1H1PFPn3yAAAA"
 
@@ -267,290 +253,3 @@ func TestSplitRDS(t *testing.T) {
 		assert.Equal(t, "Slow query: select * from table.", enhanced["rawlog"])
 	}
 }
-
-func TestSplitGlue(t *testing.T) {
-	input := LogEventBatch{
-		MessageType:         "DATA_MESSAGE",
-		Owner:               "123456789012",
-		LogGroup:            "/aws-glue/jobs/clever-dev/analytics-district-participation/aae75f00",
-		LogStream:           "jr_8927660fecacbe026ccab656cb80befea8102ac2023df531b92889b112aada28-1",
-		SubscriptionFilters: []string{"ForwardLogsToKinesis"},
-		LogEvents: []LogEvent{
-			{
-				ID:        "99999992379011144044923130086453437181614530551221780480",
-				Timestamp: NewUnixTimestampMillis(1498519943285),
-				Message:   "foo bar.",
-			},
-		},
-	}
-	lines := Split(input)
-	expected := [][]byte{
-		[]byte(`2017-06-26T23:32:23.285001+00:00 aws-glue clever-dev--analytics-district-participation/arn%3Aaws%3Aecs%3Aus-east-1%3A999988887777%3Atask%2Fjr_8927660fecacbe026ccab656cb80befea8102ac2023df531b92889b112aada28: foo bar.`),
-	}
-	assert.Equal(t, expected, lines)
-	for _, line := range expected {
-		enhanced, err := decode.ParseAndEnhance(string(line), "")
-		require.Nil(t, err)
-		assert.Equal(t, "aws-glue", enhanced["hostname"])
-		assert.Equal(t, "clever-dev", enhanced["container_env"])
-		assert.Equal(t, "analytics-district-participation", enhanced["container_app"])
-		assert.Equal(t, "jr_8927660fecacbe026ccab656cb80befea8102ac2023df531b92889b112aada28", enhanced["container_task"])
-	}
-}
-
-func TestSplitIfNecesary(t *testing.T) {
-
-	// We provide three different inputs to batchedWriter.splitMessageIfNecessary
-	// plain text
-	// zlib compressed text
-	// gzip compressed CloudWatch logs batch
-	// we verify that the split function matches the input against the correct splitter
-	// and decodes it.
-
-	assert := assert.New(t)
-
-	plainTextInput := []byte("hello, world!")
-
-	records, err := SplitMessageIfNecessary(plainTextInput)
-	assert.NoError(err)
-	assert.Equal(
-		records,
-		[][]byte{[]byte("hello, world!")},
-	)
-
-	var z bytes.Buffer
-	zbuf := zlib.NewWriter(&z)
-	zbuf.Write([]byte("hello, world!"))
-	zbuf.Close()
-	zlibSingleInput := z.Bytes()
-
-	records, err = SplitMessageIfNecessary(zlibSingleInput)
-	assert.NoError(err)
-	assert.Equal(
-		records,
-		[][]byte{[]byte("hello, world!")},
-	)
-
-	// the details of this part aren't super important since the actual functionality is
-	// tested in other tests; for this test we just want to make sure that split function
-	// correctly realizes it's gzip and call the appropriate CW-log-splitting logic
-	var g bytes.Buffer
-	gbuf := gzip.NewWriter(&g)
-	cwLogBatch := LogEventBatch{
-		MessageType:         "test",
-		Owner:               "test",
-		LogGroup:            "test",
-		LogStream:           "test",
-		SubscriptionFilters: []string{""},
-		LogEvents: []LogEvent{{
-			ID:        "test",
-			Timestamp: UnixTimestampMillis(time.Date(2020, time.September, 9, 9, 10, 10, 0, time.UTC)),
-			Message:   "test",
-		}},
-	}
-	cwLogBatchJSON, _ := json.Marshal(cwLogBatch)
-	gbuf.Write(cwLogBatchJSON)
-	gbuf.Close()
-	gzipBatchInput := g.Bytes()
-
-	expectedRecord := []byte("2020-09-09T09:10:10.000001+00:00 test test--test/arn%3Aaws%3Aecs%3Aus-east-1%3A999988887777%3Atask%2F12345678-1234-1234-1234-555566667777: test")
-	records, err = SplitMessageIfNecessary(gzipBatchInput)
-	assert.NoError(err)
-	assert.Equal(
-		records,
-		[][]byte{expectedRecord},
-	)
-}
-
-func createKPLAggregate(input [][]byte, compress bool) []byte {
-	var partitionKeyIndex uint64 = 0
-
-	records := []*kpl.Record{}
-	for _, log := range input {
-		if compress {
-			var z bytes.Buffer
-			zbuf := zlib.NewWriter(&z)
-			zbuf.Write(log)
-			zbuf.Close()
-			log = z.Bytes()
-		}
-		records = append(records, &kpl.Record{
-			PartitionKeyIndex: &partitionKeyIndex,
-			Data:              log,
-		})
-	}
-
-	logProto, err := proto.Marshal(&kpl.AggregatedRecord{
-		PartitionKeyTable: []string{"ecs_task_arn"},
-		Records:           records,
-	})
-	if err != nil {
-		panic(err)
-	}
-	log := append(kplMagicNumber, logProto...)
-	logHash := md5.Sum(logProto)
-	return append(log, logHash[0:16]...)
-}
-
-func TestKPLDeaggregate(t *testing.T) {
-	type test struct {
-		description string
-		input       []byte
-		output      [][]byte
-		shouldError bool
-	}
-
-	tests := []test{
-		{
-			description: "non-aggregated record",
-			input:       []byte("hello"),
-			output:      [][]byte{[]byte("hello")},
-			shouldError: false,
-		},
-		{
-			description: "one kpl-aggregated record",
-			input: createKPLAggregate(
-				[][]byte{[]byte("hello")},
-				false,
-			),
-			output:      [][]byte{[]byte("hello")},
-			shouldError: false,
-		},
-		{
-			description: "three kpl-aggregated record",
-			input: createKPLAggregate([][]byte{
-				[]byte("hello, "),
-				[]byte("world"),
-				[]byte("!"),
-			},
-				false,
-			),
-			output: [][]byte{
-				[]byte("hello, "),
-				[]byte("world"),
-				[]byte("!"),
-			},
-			shouldError: false,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.description, func(t *testing.T) {
-			out, err := KPLDeaggregate(tt.input)
-			if tt.shouldError {
-				assert.Error(t, err)
-				return
-			}
-			assert.NoError(t, err)
-			assert.Equal(t, out, tt.output)
-		})
-
-	}
-}
-
-func TestDeaggregate(t *testing.T) {
-	type test struct {
-		description string
-		input       []byte
-		output      [][]byte
-		shouldError bool
-	}
-
-	tests := []test{
-		{
-			description: "non-aggregated record",
-			input:       []byte("hello"),
-			output:      [][]byte{[]byte("hello")},
-			shouldError: false,
-		},
-		{
-			description: "one kpl-aggregated record",
-			input: createKPLAggregate(
-				[][]byte{[]byte("hello")},
-				false,
-			),
-			output:      [][]byte{[]byte("hello")},
-			shouldError: false,
-		},
-		{
-			description: "three kpl-aggregated record",
-			input: createKPLAggregate([][]byte{
-				[]byte("hello, "),
-				[]byte("world"),
-				[]byte("!"),
-			},
-				false,
-			),
-			output: [][]byte{
-				[]byte("hello, "),
-				[]byte("world"),
-				[]byte("!"),
-			},
-			shouldError: false,
-		},
-		{
-			description: "one kpl-aggregated zlib record",
-			input: createKPLAggregate(
-				[][]byte{[]byte("hello")},
-				true,
-			),
-			output:      [][]byte{[]byte("hello")},
-			shouldError: false,
-		},
-		{
-			description: "three kpl-aggregated zlib record",
-			input: createKPLAggregate([][]byte{
-				[]byte("hello, "),
-				[]byte("world"),
-				[]byte("!"),
-			},
-				true,
-			),
-			output: [][]byte{
-				[]byte("hello, "),
-				[]byte("world"),
-				[]byte("!"),
-			},
-			shouldError: false,
-		},
-	}
-
-	var g bytes.Buffer
-	gbuf := gzip.NewWriter(&g)
-	cwLogBatch := LogEventBatch{
-		MessageType:         "test",
-		Owner:               "test",
-		LogGroup:            "test",
-		LogStream:           "test",
-		SubscriptionFilters: []string{""},
-		LogEvents: []LogEvent{{
-			ID:        "test",
-			Timestamp: UnixTimestampMillis(time.Date(2020, time.September, 9, 9, 10, 10, 0, time.UTC)),
-			Message:   "test",
-		}},
-	}
-	cwLogBatchJSON, _ := json.Marshal(cwLogBatch)
-	gbuf.Write(cwLogBatchJSON)
-	gbuf.Close()
-	gzipBatchInput := g.Bytes()
-
-	tests = append(tests, test{
-		description: "cloudwatch log batch",
-		input:       gzipBatchInput,
-		output:      [][]byte{[]byte("2020-09-09T09:10:10.000001+00:00 test test--test/arn%3Aaws%3Aecs%3Aus-east-1%3A999988887777%3Atask%2F12345678-1234-1234-1234-555566667777: test")},
-		shouldError: false,
-	})
-	for _, tt := range tests {
-		t.Run(tt.description, func(t *testing.T) {
-			out, err := Deaggregate(tt.input)
-
-			if tt.shouldError {
-				assert.Error(t, err)
-				return
-			}
-			assert.NoError(t, err)
-			assert.Equal(t, out, tt.output)
-		})
-
-	}
-}