Repository: honeycombio/honeycomb-kubernetes-agent
Pull request #43 (2019-05-06 17:20:11)

[processors] add additional fields processor
Addresses https://github.com/honeycombio/honeycomb-kubernetes-agent/issues/42


Patch:
diff --git a/config/config.go b/config/config.go
index f0e25ab..a9a55a9 100644
--- a/config/config.go
+++ b/config/config.go
@@ -7,12 +7,13 @@ import (
 )
 
 type Config struct {
-	APIHost        string `yaml:"apiHost"`
-	WriteKey       string `yaml:"writekey"`
-	Watchers       []*WatcherConfig
-	Verbosity      string
-	LegacyLogPaths bool `yaml:"legacyLogPaths"`
-	SplitLogging   bool `yaml:"splitLogging"`
+	APIHost          string `yaml:"apiHost"`
+	WriteKey         string `yaml:"writekey"`
+	Watchers         []*WatcherConfig
+	Verbosity        string
+	LegacyLogPaths   bool                   `yaml:"legacyLogPaths"`
+	SplitLogging     bool                   `yaml:"splitLogging"`
+	AdditionalFields map[string]interface{} `yaml:"additionalFields"`
 }
 
 type WatcherConfig struct {
diff --git a/docs/configuration-reference.md b/docs/configuration-reference.md
index 9b6631e..6465aae 100644
--- a/docs/configuration-reference.md
+++ b/docs/configuration-reference.md
@@ -90,6 +90,19 @@ More parsers will be added in the future. If you'd like to see support for addit
 ## Processors
 Processors transform events after they're parsed. Currently, the following processors are supported:
 
+### additional_fields
+
+The `additional_fields` processor accepts a static map of field names and values and appends those to every event it processes. These values will overwrite existing fields of the same name, if they exist:
+
+**Example:**
+
+```yaml
+processors:
+  - additional_fields:
+      environment: production
+      owner: me@example.com
+```
+
 ### sample
 The `sample` processor will only send a subset of events to Honeycomb. Honeycomb natively supports sampled event streams, allowing you to send a representative subset of events while still getting high-fidelity query results.
 
@@ -190,6 +203,21 @@ into
 }
 ```
 
+## Global Configuration Options
+
+### additionalFields
+
+You can append arbitrary key/value pairs to all events processed by the agent by defining the `additionalFields` section of the config. These values will overwrite existing fields of the same name, if they exist:
+
+```yaml
+writekey: "YOUR_HONEYCOMB_WRITEKEY_HERE"
+watchers:
+  - # ...
+additionalFields:
+  cluster_name: my cluster
+  owner: me@example.com
+```
+
 ## Sample configurations
 
 Here are some example configurations for the Honeycomb agent.
diff --git a/examples/quickstart.yaml b/examples/quickstart.yaml
index 42d16ab..786bd66 100644
--- a/examples/quickstart.yaml
+++ b/examples/quickstart.yaml
@@ -46,10 +46,16 @@ data:
       - dataset: kubernetes-logs
         labelSelector: ""
         parser: glog
+        # you can append arbitrary field information to watchers
+        # additionalFields:
+        #   owner_email: me@example.com
     verbosity: debug
     # support older container log paths, see:
     # https://github.com/kubernetes/community/blob/master/contributors/design-proposals/node/kubelet-cri-logging.md
     legacyLogPaths: true
+    # you can append arbitrary fields to all events sent from this agent
+    #additionalFields:
+    #  cluster_name: "dev_cluster"
 
 # Daemonset
 ---
diff --git a/main.go b/main.go
index 5234b2d..2be9fd0 100644
--- a/main.go
+++ b/main.go
@@ -154,6 +154,7 @@ func main() {
 				stateRecorder,
 				kubeClient,
 				config.LegacyLogPaths,
+				config.AdditionalFields,
 			)
 			pt.Start()
 			defer pt.Stop()
diff --git a/podtailer/podtailer.go b/podtailer/podtailer.go
index 5d33be5..2511f01 100644
--- a/podtailer/podtailer.go
+++ b/podtailer/podtailer.go
@@ -30,14 +30,15 @@ const maxPodWatcherRetries = 3
 // PodSetTailer is responsible for watching for all pods that match the
 // criteria defined by config, and managing tailers for each pod.
 type PodSetTailer struct {
-	config         *config.WatcherConfig
-	nodeSelector   string
-	transmitter    transmission.Transmitter
-	stateRecorder  tailer.StateRecorder
-	kubeClient     corev1.PodsGetter
-	stop           chan bool
-	wg             sync.WaitGroup
-	legacyLogPaths bool
+	config                 *config.WatcherConfig
+	nodeSelector           string
+	transmitter            transmission.Transmitter
+	stateRecorder          tailer.StateRecorder
+	kubeClient             corev1.PodsGetter
+	stop                   chan bool
+	wg                     sync.WaitGroup
+	legacyLogPaths         bool
+	additionalFieldsGlobal map[string]interface{}
 }
 
 func NewPodSetTailer(
@@ -47,15 +48,17 @@ func NewPodSetTailer(
 	stateRecorder tailer.StateRecorder,
 	kubeClient corev1.PodsGetter,
 	legacyLogPaths bool,
+	additionalFieldsGlobal map[string]interface{},
 ) *PodSetTailer {
 	return &PodSetTailer{
-		config:         config,
-		nodeSelector:   nodeSelector,
-		transmitter:    transmitter,
-		stateRecorder:  stateRecorder,
-		kubeClient:     kubeClient,
-		stop:           make(chan bool),
-		legacyLogPaths: legacyLogPaths,
+		config:                 config,
+		nodeSelector:           nodeSelector,
+		transmitter:            transmitter,
+		stateRecorder:          stateRecorder,
+		kubeClient:             kubeClient,
+		stop:                   make(chan bool),
+		legacyLogPaths:         legacyLogPaths,
+		additionalFieldsGlobal: additionalFieldsGlobal,
 	}
 }
 
@@ -273,14 +276,21 @@ func (pt *PodSetTailer) watcherForPod(pod *v1.Pod, containerName string, podWatc
 	// one is specified
 	filterFunc := determineFilterFunc(pod, containerName, pt.legacyLogPaths)
 
-	k8sMetadataProcessor := &processors.KubernetesMetadataProcessor{
-		PodGetter: podWatcher,
-		UID:       pod.UID}
+	additionalProcessors := []processors.Processor{
+		&processors.KubernetesMetadataProcessor{
+			PodGetter: podWatcher,
+			UID:       pod.UID,
+		},
+	}
+	// the additionalFields section at the root of the config applies to all pod watchers
+	if pt.additionalFieldsGlobal != nil {
+		additionalProcessors = append(additionalProcessors, &processors.AdditionalFieldsProcessor{AdditionalFields: pt.additionalFieldsGlobal})
+	}
 	handlerFactory, err := handlers.NewLineHandlerFactoryFromConfig(
 		pt.config,
 		&unwrappers.DockerJSONLogUnwrapper{},
 		pt.transmitter,
-		k8sMetadataProcessor)
+		additionalProcessors...)
 	if err != nil {
 		// This shouldn't happen, since we check for configuration errors
 		// before actually setting up the watcher
diff --git a/processors/additional_fields.go b/processors/additional_fields.go
new file mode 100644
index 0000000..b17ac13
--- /dev/null
+++ b/processors/additional_fields.go
@@ -0,0 +1,27 @@
+package processors
+
+import (
+	"github.com/honeycombio/honeycomb-kubernetes-agent/event"
+)
+
+type AdditionalFieldsProcessor struct {
+	AdditionalFields map[string]interface{}
+}
+
+func (a *AdditionalFieldsProcessor) Init(options map[string]interface{}) error {
+	// options expects a map of string->values
+	if options != nil {
+		a.AdditionalFields = options
+	} else {
+		a.AdditionalFields = make(map[string]interface{})
+	}
+	return nil
+}
+
+func (a *AdditionalFieldsProcessor) Process(ev *event.Event) bool {
+	for k, v := range a.AdditionalFields {
+		ev.Data[k] = v
+	}
+
+	return true
+}
diff --git a/processors/processors.go b/processors/processors.go
index 4479d24..aa2a6f6 100644
--- a/processors/processors.go
+++ b/processors/processors.go
@@ -55,6 +55,8 @@ func NewProcessor(name string, options map[string]interface{}) (Processor, error
 		p = &TimeFieldExtractor{}
 	case "rename_field":
 		p = &FieldRenamer{}
+	case "additional_fields":
+		p = &AdditionalFieldsProcessor{}
 	default:
 		return nil, fmt.Errorf("Unknown processor type %s", name)
 	}


Test patch:
diff --git a/config/config_test.go b/config/config_test.go
index 6ed3636..b7dea2e 100644
--- a/config/config_test.go
+++ b/config/config_test.go
@@ -26,3 +26,17 @@ func TestParsing(t *testing.T) {
 		}
 	}
 }
+
+func TestAdditionalFieldsParsing(t *testing.T) {
+	path, _ := filepath.Abs(filepath.Join("testdata", "basic.yaml"))
+	c, err := ReadFromFile(path)
+	assert.NoError(t, err)
+
+	assert.Equal(t, map[string]interface{}{"foo": 1, "bar": "2"}, c.AdditionalFields)
+
+	path, _ = filepath.Abs(filepath.Join("testdata", "parser_with_options.yaml"))
+	c, err = ReadFromFile(path)
+	assert.NoError(t, err)
+
+	assert.Equal(t, map[string]interface{}{"foo": "1", "bar": 2}, c.Watchers[0].Processors[0]["additional_fields"])
+}
diff --git a/config/testdata/basic.yaml b/config/testdata/basic.yaml
index fe15f48..fa3b1fc 100644
--- a/config/testdata/basic.yaml
+++ b/config/testdata/basic.yaml
@@ -5,3 +5,6 @@ watchers:
 - labelSelector: app=nginx
   dataset: testdataset
   parser: json
+additionalFields:
+  foo: 1
+  bar: "2"
\ No newline at end of file
diff --git a/config/testdata/parser_with_options.yaml b/config/testdata/parser_with_options.yaml
index e5c66a9..862c125 100644
--- a/config/testdata/parser_with_options.yaml
+++ b/config/testdata/parser_with_options.yaml
@@ -4,3 +4,7 @@ watchers:
       name: nginx
       options:
         log_format: default
+    processors:
+      - additional_fields:
+          foo: "1"
+          bar: 2
diff --git a/processors/additional_fields_test.go b/processors/additional_fields_test.go
new file mode 100644
index 0000000..781e192
--- /dev/null
+++ b/processors/additional_fields_test.go
@@ -0,0 +1,31 @@
+package processors
+
+import (
+	"testing"
+
+	"github.com/honeycombio/honeycomb-kubernetes-agent/event"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAdditionalFieldsProcess(t *testing.T) {
+	fields := map[string]interface{}{"foo": 1, "bar": "two"}
+
+	p := &AdditionalFieldsProcessor{}
+	p.Init(fields)
+
+	ev := &event.Event{}
+	ev.Data = make(map[string]interface{})
+	p.Process(ev)
+	assert.Equal(t, ev.Data["foo"], 1)
+	assert.Equal(t, ev.Data["bar"], "two")
+}
+
+func TestAdditionalFieldsProcessNilMap(t *testing.T) {
+	p := &AdditionalFieldsProcessor{}
+	p.Init(nil)
+
+	ev := &event.Event{}
+	ev.Data = make(map[string]interface{})
+	// shouldn't crash
+	p.Process(ev)
+}