Repository: honeycombio/honeycomb-kubernetes-agent
Pull request #53 (2019-10-17 14:08:07)

processors/route_event: A simple value based event router
This lets one split events in kubernetes into different datasets based on a
simple route matching logic.

This is mostly useful for services that serve multiple different backends, like
load balancers, ingresses and databases. It was specifically made to handle
logs out of `nginx-ingress` to be able to split them based on the service it
waas routing data to.

Spindel: All review comments are welcome, I'm still trying to figure out Go here, so if something isn't idiomatic or could be better, please let me know. 

Followup on review comments will happen after the weekend.
tredman: @Spindel looks like a few simple conflicts. If you're busy I can sort them out later today.

Patch:
diff --git a/docs/configuration-reference.md b/docs/configuration-reference.md
index bada909..cc06624 100644
--- a/docs/configuration-reference.md
+++ b/docs/configuration-reference.md
@@ -154,7 +154,6 @@ key | value | description
 field | string | The name of the event field to match against the blacklist
 values | list  | The set of field values that cause this processor to drop an event
 
-
 ### keep_event
 
 The `keep_event` processor will remove all events NOT matching one of the
@@ -171,6 +170,46 @@ key | value | description
 field | string | The name of the field to match against the whitelist
 values | list  | The name of the event field to match against the whitelist
 
+
+### route_event
+
+The `route_event` processor will route events to a different dataset depending
+on the value of a field.  This is mostly useful for load balancing web servers
+or other resources that serve multiple different backends, but where you still
+wish to track a common identifier between systems.
+
+
+**Options:**
+key | value | description
+:--|:--|:--
+field | string | The name of the event field to base routing on
+routes | list | A list of routing configurations
+
+**Routes:**
+
+key | value | description
+:--|:--|:--
+value   | string | Route if the event field is exactly this value
+dataset | string | The dataset to route to
+
+
+Example configuration
+```
+processors:
+- route_events:
+    field: host
+    routes:
+      - value: api.example.com
+        dataset: api
+      - value: www.example.com
+        dataset: web
+```
+
+The above configuration would route events where `host` exists and is equal to
+`api.example.com` to the `api` dataset. It would also route requests to
+`www.example.com` to the `www` dataset. All others would be routed to the
+default dataset for this watcher.
+
 ### rename_field
 
 The `rename_field` processor will rename the specified field in all events, if it exists, before sending them to Honeycomb. You can use this to standardize field names across data sources, for example. Note that if the `new` field already exists, it will be overwritten with the value in the `original` field.
diff --git a/processors/processors.go b/processors/processors.go
index 4f53d16..9b2b3e9 100644
--- a/processors/processors.go
+++ b/processors/processors.go
@@ -45,6 +45,8 @@ func NewProcessorFromConfig(config map[string]map[string]interface{}) (Processor
 func NewProcessor(name string, options map[string]interface{}) (Processor, error) {
 	var p Processor
 	switch name {
+	case "route_event":
+		p = &EventRouter{}
 	case "request_shape":
 		p = &RequestShaper{}
 	case "drop_field":
diff --git a/processors/route_event.go b/processors/route_event.go
new file mode 100644
index 0000000..87b6ff2
--- /dev/null
+++ b/processors/route_event.go
@@ -0,0 +1,76 @@
+package processors
+
+import (
+	"errors"
+	"fmt"
+	"github.com/honeycombio/honeycomb-kubernetes-agent/event"
+	"github.com/mitchellh/mapstructure"
+	"github.com/sirupsen/logrus"
+)
+
+type EventRouter struct {
+	config *eventRouterConfig
+	routes map[string]string
+}
+
+type eventRouterConfig struct {
+	Field  string
+	Routes []eventRoute
+}
+
+type eventRoute struct {
+	Dataset string
+	Value   string
+}
+
+var (
+	ErrDuplicateEventRoute    = errors.New("route_event requires all values to be unique")
+	ErrMissingFieldEventRoute = errors.New("route_event requires field to be set")
+)
+
+func (f *EventRouter) Init(options map[string]interface{}) error {
+	config := &eventRouterConfig{}
+	err := mapstructure.Decode(options, config)
+	if err != nil {
+		return err
+	}
+
+	if config.Field == "" {
+		return ErrMissingFieldEventRoute
+	}
+	f.routes = make(map[string]string, len(config.Routes))
+
+	for _, route := range config.Routes {
+		// Same route pointing to multiple datasets?
+		if _, ok := f.routes[route.Value]; ok {
+			return ErrDuplicateEventRoute
+		}
+		f.routes[route.Value] = route.Dataset
+	}
+	f.config = config
+	return nil
+}
+
+func (f *EventRouter) Process(ev *event.Event) bool {
+	if ev.Data == nil {
+		return true
+	}
+	val, ok := ev.Data[f.config.Field]
+	if !ok {
+		return true
+	}
+
+	valString, ok := val.(string)
+	if !ok {
+		logrus.WithFields(logrus.Fields{
+			"key":   f.config.Field,
+			"value": val,
+			"type":  fmt.Sprintf("%T", val)}).
+			Debug("Not routing field of non-string type")
+		return true
+	}
+	if dataset, ok := f.routes[valString]; ok {
+		ev.Dataset = dataset
+	}
+	return true
+}


Test patch:
diff --git a/handlers/handlers_test.go b/handlers/handlers_test.go
index f3f9d24..0458e4e 100644
--- a/handlers/handlers_test.go
+++ b/handlers/handlers_test.go
@@ -440,6 +440,52 @@ func TestRenameField(t *testing.T) {
 	assert.Equal(t, mt.events[0], expected)
 }
 
+func TestEventRouter(t *testing.T) {
+	mt := &MockTransmitter{}
+
+	cfg, err := watcherConfigFromYAML(`
+dataset: kubernetestest
+parser: json
+processors:
+- route_event:
+    field: host
+    routes:
+      - value: api
+        dataset: api
+      - value: www
+        dataset: not-api`)
+	assert.NoError(t, err)
+
+	hf, err := NewLineHandlerFactoryFromConfig(cfg, &unwrappers.RawLogUnwrapper{}, mt)
+	assert.NoError(t, err)
+	handler := hf.New("/tmp/testpath")
+	handler.Handle(`{"host": "default", "another": "field"}`)
+	handler.Handle(`{"host": "api", "another": "field"}`)
+	handler.Handle(`{"host": "www", "another": "field"}`)
+	assert.Equal(t, len(mt.events), 3)
+	expected0 := &event.Event{
+		Data:       map[string]interface{}{"host": "default", "another": "field"},
+		Dataset:    "kubernetestest",
+		Path:       "/tmp/testpath",
+		RawMessage: `{"host": "default", "another": "field"}`,
+	}
+	expected1 := &event.Event{
+		Data:       map[string]interface{}{"host": "api", "another": "field"},
+		Dataset:    "api",
+		Path:       "/tmp/testpath",
+		RawMessage: `{"host": "api", "another": "field"}`,
+	}
+	expected2 := &event.Event{
+		Data:       map[string]interface{}{"host": "www", "another": "field"},
+		Dataset:    "not-api",
+		Path:       "/tmp/testpath",
+		RawMessage: `{"host": "www", "another": "field"}`,
+	}
+	assert.Equal(t, mt.events[0], expected0)
+	assert.Equal(t, mt.events[1], expected1)
+	assert.Equal(t, mt.events[2], expected2)
+}
+
 func TestEventDropper(t *testing.T) {
 	mt := &MockTransmitter{}
 
@@ -456,7 +502,8 @@ processors:
 	hf, err := NewLineHandlerFactoryFromConfig(cfg, &unwrappers.RawLogUnwrapper{}, mt)
 	assert.NoError(t, err)
 	handler := hf.New("/tmp/testpath")
-	handler.Handle(`{"service": "dropthis", "another": "field"}`)
+
+  handler.Handle(`{"service": "dropthis", "another": "field"}`)
 	handler.Handle(`{"service": "keepme", "another": "field"}`)
 	handler.Handle(`{"no_service": "keepme", "another": "field"}`)
 	assert.Equal(t, len(mt.events), 2)