refactor: abstract structs event.Raw and event.Slo into interfaces and add UID to each event

CHANGELOG.md

@@ -5,8 +5,16 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
-### Fixed
+### Added
+- [#70](https://github.com/seznam/slo-exporter/pull/70) Every event has a unique id set, so it can be filtered in logs.
+  The `kafkaIngester`, `tailer` and `envoyAccessLogServer` modules
+  has new config option `eventIdMetadataKey` to use value of this metadata key as the unique identifier.
+  (Mostly should be used with a trace id for example).
 
+### Changed
+- _Internal:_ Event structs `Raw` and `Slo` has been abstracted into interfaces.
+
+### Fixed
 - [#71](https://github.com/seznam/slo-exporter/pull/71) Fix corner cases in StringMap.Merge(), StringMap.Without()
 
 ## [v6.9.0] 2021-07-14

docs/modules/envoy_access_log_server.md

@@ -131,4 +131,6 @@ Please note that some of the keys may not be present	|
 address: ":18090"
 # gracefulShutdownTimeout for the GRPC server. Please note also the existence of 'maximumGracefulShutdownDuration' global config option which is effectively an upper boundary of here-specified timeout value.
 gracefulShutdownTimeout: "5s"
+# eventIdMetadataKey its value will be used as a unique id for the generated event if present (hint: use a trace ID if possible).
+eventIdMetadataKey: <string>
 ```

docs/modules/kafka_ingester.md

@@ -28,6 +28,8 @@ retentionTime: <duration>
 # fallbackStartOffset determines from whence the consumer group should begin consuming when it finds a partition without a committed offset.
 # Default: FirstOffset
 fallbackStartOffset: <LastOffset|FirstOffset>
+# eventIdMetadataKey its value will be used as a unique id for the generated event if present (hint: use a trace ID if possible).
+eventIdMetadataKey: <string>
 ```
 
 

docs/modules/tailer.md

@@ -35,5 +35,7 @@ positionPersistenceInterval: "2s"
 loglineParseRegexp: '^(?P<ip>[A-Fa-f0-9.:]{4,50}) \S+ \S+ \[(?P<time>.*?)\] "(?P<request>.*?)" (?P<statusCode>\d+) \d+ "(?P<referer>.*?)" uag="(?P<userAgent>[^"]+)" "[^"]+" ua="[^"]+" rt="(?P<requestDuration>\d+(\.\d+)??)".+ignore-slo="(?P<ignoreSloHeader>[^"]*)" slo-domain="(?P<sloDomain>[^"]*)" slo-app="(?P<sloApp>[^"]*)" slo-class="(?P<sloClass>[^"]*)" slo-endpoint="(?P<sloEndpoint>[^"]*)" slo-result="(?P<sloResult>[^"]*)"'    # emptyGroupRE defines RE used to decide whether some of the RE match groups specified in loglineParseRegexp is empty and this its assigned variable should be kept unitialized
 # Value, that will be treated as empty value.
 emptyGroupRE: '^-$'
+# eventIdMetadataKey its value will be used as a unique id for the generated event if present (hint: use a trace ID if possible).
+eventIdMetadataKey: <string>
 ```
 

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/go-kit/kit v0.10.0
 	github.com/go-test/deep v1.0.6
 	github.com/golang/protobuf v1.4.2
+	github.com/google/uuid v1.3.0
 	github.com/gorilla/mux v1.7.4
 	github.com/grafana/loki v1.5.0
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.1-0.20191002090509-6af20e3a5340

go.sum

@@ -349,6 +349,8 @@ github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm4
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go v2.0.2+incompatible/go.mod h1:SFVmujtThgffbyetf+mdk2eWhX2bMyUtNHzFKcPA9HY=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=

pkg/dynamic_classifier/dynamic_classifier.go

@@ -60,8 +60,8 @@ type DynamicClassifier struct {
 	unclassifiedEventMetadataKeys []string
 	eventsMetric                  *prometheus.CounterVec
 	observer                      pipeline.EventProcessingDurationObserver
-	inputChannel                  chan *event.Raw
-	outputChannel                 chan *event.Raw
+	inputChannel                  chan event.Raw
+	outputChannel                 chan event.Raw
 	logger                        logrus.FieldLogger
 	done                          bool
 }
@@ -102,11 +102,11 @@ func (dc *DynamicClassifier) Stop() {
 	return
 }
 
-func (dc *DynamicClassifier) SetInputChannel(channel chan *event.Raw) {
+func (dc *DynamicClassifier) SetInputChannel(channel chan event.Raw) {
 	dc.inputChannel = channel
 }
 
-func (dc *DynamicClassifier) OutputChannel() chan *event.Raw {
+func (dc *DynamicClassifier) OutputChannel() chan event.Raw {
 	return dc.outputChannel
 }
 
@@ -129,8 +129,8 @@ func New(conf classifierConfig, logger logrus.FieldLogger) (*DynamicClassifier,
 		exactMatches:                  newMemoryExactMatcher(logger),
 		regexpMatches:                 newRegexpMatcher(logger),
 		unclassifiedEventMetadataKeys: conf.UnclassifiedEventMetadataKeys,
-		inputChannel:                  make(chan *event.Raw),
-		outputChannel:                 make(chan *event.Raw),
+		inputChannel:                  make(chan event.Raw),
+		outputChannel:                 make(chan event.Raw),
 		done:                          false,
 		logger:                        logger,
 	}
@@ -235,7 +235,7 @@ func (dc *DynamicClassifier) loadMatchesFromCSV(matcher matcher, path string) er
 		sloApp := line[1]
 		sloClass := line[2]
 		sloEndpoint := line[3]
-		classification := &event.SloClassification{
+		classification := event.SloClassification{
 			Domain: sloDomain,
 			App:    sloApp,
 			Class:  sloClass,
@@ -251,14 +251,14 @@ func (dc *DynamicClassifier) loadMatchesFromCSV(matcher matcher, path string) er
 }
 
 // Classify classifies endpoint by updating its Classification field
-func (dc *DynamicClassifier) Classify(newEvent *event.Raw) (bool, error) {
+func (dc *DynamicClassifier) Classify(newEvent event.Raw) (bool, error) {
 	var (
 		classificationErrors error
-		classification       *event.SloClassification
+		classification       event.SloClassification
 		classifiedBy         matcherType
 	)
 	if newEvent.IsClassified() {
-		if err := dc.exactMatches.set(newEvent.EventKey(), newEvent.SloClassification); err != nil {
+		if err := dc.exactMatches.set(newEvent.EventKey(), newEvent.SloClassification()); err != nil {
 			return true, fmt.Errorf("failed to set the exact matcher: %w", err)
 		}
 		return true, nil
@@ -272,20 +272,20 @@ func (dc *DynamicClassifier) Classify(newEvent *event.Raw) (bool, error) {
 			dc.logger.Errorf("error while classifying event: %+v", err)
 			classificationErrors = multierror.Append(classificationErrors, err)
 		}
-		if classification != nil {
+		if classification.IsClassified() {
 			classifiedBy = classifier.getType()
 			break
 		}
 	}
 
-	if classification == nil {
-		dc.reportEvent(unclassifiedEventLabel, string(classifiedBy), newEvent.Metadata)
+	if !classification.IsClassified() {
+		dc.reportEvent(unclassifiedEventLabel, string(classifiedBy), newEvent.Metadata())
 		return false, classificationErrors
 	}
 
 	dc.logger.Debugf("event '%s' matched by %s matcher", newEvent.EventKey(), classifiedBy)
-	newEvent.UpdateSLOClassification(classification)
-	dc.reportEvent(classifiedEventLabel, string(classifiedBy), newEvent.Metadata)
+	newEvent.SetSLOClassification(classification)
+	dc.reportEvent(classifiedEventLabel, string(classifiedBy), newEvent.Metadata())
 
 	// Those matched by regex we want to write to the exact matcher so it is cached
 	if classifiedBy == regexpMatcherType {
@@ -311,7 +311,7 @@ func (dc *DynamicClassifier) DumpCSV(w io.Writer, matcherType string) error {
 	return matcher.dumpCSV(w)
 }
 
-func (dc *DynamicClassifier) classifyByMatch(matcher matcher, event *event.Raw) (*event.SloClassification, error) {
+func (dc *DynamicClassifier) classifyByMatch(matcher matcher, event event.Raw) (event.SloClassification, error) {
 	return matcher.get(event.EventKey())
 }
 

pkg/dynamic_classifier/dynamic_classifier_test.go

@@ -4,9 +4,10 @@ package dynamic_classifier
 //revive:enable:var-naming
 
 import (
+	"github.com/seznam/slo-exporter/pkg/event"
+	"github.com/seznam/slo-exporter/pkg/stringmap"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
-	"github.com/seznam/slo-exporter/pkg/event"
 	"path/filepath"
 	"reflect"
 	"regexp"
@@ -68,17 +69,15 @@ func TestClassificationByExactMatches(t *testing.T) {
 
 	data := []struct {
 		endpoint               string
-		expectedClassification *event.SloClassification
+		expectedClassification event.SloClassification
 		expectedOk             bool
 	}{
 		{"GET:/testing-endpoint", newSloClassification("test-domain", "test-app", "test-class"), true},
-		{"non-classified-endpoint", nil, false},
+		{"non-classified-endpoint", event.SloClassification{}, false},
 	}
 
 	for _, ec := range data {
-		newEvent := &event.Raw{
-			SloClassification: ec.expectedClassification,
-		}
+		newEvent := event.NewRaw("", 1, stringmap.StringMap{}, &ec.expectedClassification)
 		newEvent.SetEventKey(ec.endpoint)
 
 		ok, err := classifier.Classify(newEvent)
@@ -87,8 +86,8 @@ func TestClassificationByExactMatches(t *testing.T) {
 		}
 
 		assert.Equal(t, ec.expectedOk, ok)
-		if !reflect.DeepEqual(ec.expectedClassification, newEvent.SloClassification) {
-			t.Errorf("Classification does not match %+v != %+v", ec.expectedClassification, newEvent.SloClassification)
+		if !reflect.DeepEqual(ec.expectedClassification, newEvent.SloClassification()) {
+			t.Errorf("Classification does not match %+v != %+v", ec.expectedClassification, newEvent.SloClassification())
 		}
 	}
 }
@@ -101,18 +100,16 @@ func TestClassificationByRegexpMatches(t *testing.T) {
 
 	data := []struct {
 		endpoint               string
-		expectedClassification *event.SloClassification
+		expectedClassification event.SloClassification
 		expectedOk             bool
 	}{
 		{"/api/test/asdf", newSloClassification("test-domain", "test-app", "test-class"), true},
 		{"/api/asdf", newSloClassification("test-domain", "test-app", "test-class-all"), true},
-		{"non-classified-endpoint", nil, false},
+		{"non-classified-endpoint", event.SloClassification{}, false},
 	}
 
 	for _, ec := range data {
-		newEvent := &event.Raw{
-			SloClassification: ec.expectedClassification,
-		}
+		newEvent := event.NewRaw("", 1, stringmap.StringMap{}, &ec.expectedClassification)
 		newEvent.SetEventKey(ec.endpoint)
 
 		ok, err := classifier.Classify(newEvent)
@@ -121,21 +118,19 @@ func TestClassificationByRegexpMatches(t *testing.T) {
 		}
 
 		assert.Equal(t, ec.expectedOk, ok)
-		if !reflect.DeepEqual(ec.expectedClassification, newEvent.SloClassification) {
-			t.Errorf("Classification does not match %+v != %+v", ec.expectedClassification, newEvent.SloClassification)
+		if !reflect.DeepEqual(ec.expectedClassification, newEvent.SloClassification()) {
+			t.Errorf("Classification does not match %+v != %+v", ec.expectedClassification, newEvent.SloClassification())
 		}
 	}
 }
 
 func Test_DynamicClassifier_Classify_UpdatesEmptyCache(t *testing.T) {
 	eventKey := "GET:/testing-endpoint"
-	classifiedEvent := &event.Raw{
-		SloClassification: &event.SloClassification{
-			Domain: "domain",
-			App:    "app",
-			Class:  "class",
-		},
-	}
+	classifiedEvent := event.NewRaw("", 1, stringmap.StringMap{}, &event.SloClassification{
+		Domain: "domain",
+		App:    "app",
+		Class:  "class",
+	})
 	classifiedEvent.SetEventKey(eventKey)
 
 	// test that classified event updates an empty exact matches cache
@@ -148,21 +143,19 @@ func Test_DynamicClassifier_Classify_UpdatesEmptyCache(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error while getting the tested event key from exact Matches classifier: %v", err)
 	}
-	if !reflect.DeepEqual(classifiedEvent.SloClassification, classification) {
-		t.Errorf("event classification '%+v' did not propagate to classifier exact matches cache: %+v", classifiedEvent.SloClassification, classification)
+	if !reflect.DeepEqual(classifiedEvent.SloClassification(), classification) {
+		t.Errorf("event classification '%+v' did not propagate to classifier exact matches cache: %+v", classifiedEvent.SloClassification(), classification)
 	}
 }
 
 // test that classified event updates dynamic classifier cache as initialized from golden file
 func Test_DynamicClassifier_Classify_OverridesCacheFromConfig(t *testing.T) {
 	eventKey := "GET:/testing-endpoint"
-	classifiedEvent := &event.Raw{
-		SloClassification: &event.SloClassification{
-			Domain: "domain",
-			App:    "app",
-			Class:  "class",
-		},
-	}
+	classifiedEvent := event.NewRaw("", 1, stringmap.StringMap{}, &event.SloClassification{
+		Domain: "domain",
+		App:    "app",
+		Class:  "class",
+	})
 	classifiedEvent.SetEventKey(eventKey)
 
 	classifier := newClassifier(t, classifierConfig{RegexpMatchesCsvFiles: goldenFile(t)})
@@ -179,8 +172,8 @@ func Test_DynamicClassifier_Classify_OverridesCacheFromConfig(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error while getting the tested event key from exact Matches classifier: %v", err)
 	}
-	if !reflect.DeepEqual(classifiedEvent.SloClassification, classification) {
-		t.Errorf("classifier cache '%+v' for event_key '%s' was not updated with classification from the classified event '%+v'.", classifiedEvent.SloClassification, eventKey, classification)
+	if !reflect.DeepEqual(classifiedEvent.SloClassification(), classification) {
+		t.Errorf("classifier cache '%+v' for event_key '%s' was not updated with classification from the classified event '%+v'.", classifiedEvent.SloClassification(), eventKey, classification)
 	}
 }
 
@@ -191,13 +184,11 @@ func Test_DynamicClassifier_Classify_OverridesCacheFromPreviousClassifiedEvent(t
 
 	classifier := newClassifier(t, classifierConfig{})
 	for _, eventClass := range eventClasses {
-		classifiedEvent := &event.Raw{
-			SloClassification: &event.SloClassification{
-				Domain: "domain",
-				App:    "app",
-				Class:  eventClass,
-			},
-		}
+		classifiedEvent := event.NewRaw("", 1, stringmap.StringMap{}, &event.SloClassification{
+			Domain: "domain",
+			App:    "app",
+			Class:  eventClass,
+		})
 		classifiedEvent.SetEventKey(eventKey)
 
 		ok, err := classifier.Classify(classifiedEvent)
@@ -208,8 +199,8 @@ func Test_DynamicClassifier_Classify_OverridesCacheFromPreviousClassifiedEvent(t
 		if err != nil {
 			t.Fatalf("error while getting the tested event key from exact Matches classifier: %v", err)
 		}
-		if !reflect.DeepEqual(classifiedEvent.SloClassification, classification) {
-			t.Errorf("classifier cache '%+v' for event_key '%s' was not updated with classification from the classified event '%+v'.", classifiedEvent.SloClassification, eventKey, classification)
+		if !reflect.DeepEqual(classifiedEvent.SloClassification(), classification) {
+			t.Errorf("classifier cache '%+v' for event_key '%s' was not updated with classification from the classified event '%+v'.", classifiedEvent.SloClassification(), eventKey, classification)
 		}
 	}
 }

pkg/dynamic_classifier/matcher.go

@@ -12,7 +12,7 @@ type matcherType string
 
 type matcher interface {
 	getType() matcherType
-	set(key string, classification *event.SloClassification) error
-	get(key string) (*event.SloClassification, error)
+	set(key string, classification event.SloClassification) error
+	get(key string) (event.SloClassification, error)
 	dumpCSV(w io.Writer) error
 }

pkg/dynamic_classifier/matcher_test.go

@@ -6,8 +6,8 @@ package dynamic_classifier
 import (
 	"bytes"
 	"fmt"
-	"github.com/sirupsen/logrus"
 	"github.com/seznam/slo-exporter/pkg/event"
+	"github.com/sirupsen/logrus"
 	"io/ioutil"
 	"path/filepath"
 	"reflect"
@@ -17,8 +17,8 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func newSloClassification(domain string, app string, class string) *event.SloClassification {
-	return &event.SloClassification{
+func newSloClassification(domain string, app string, class string) event.SloClassification {
+	return event.SloClassification{
 		Domain: domain,
 		App:    app,
 		Class:  class,
@@ -30,15 +30,15 @@ func TestMatcher(t *testing.T) {
 	cases := []struct {
 		matcher     matcher
 		key         string
-		value       *event.SloClassification
+		value       event.SloClassification
 		wantedKey   string
-		wantedValue *event.SloClassification
+		wantedValue event.SloClassification
 		setErr      string
 		getErr      string
 	}{
 		{newMemoryExactMatcher(logger), "test", newSloClassification("test-domain", "test-app", "test-class"), "test", newSloClassification("test-domain", "test-app", "test-class"), "", ""},
 		{newMemoryExactMatcher(logger), "", newSloClassification("test-domain", "test-app", "test-class"), "", newSloClassification("test-domain", "test-app", "test-class"), "", ""},
-		{newMemoryExactMatcher(logger), "test", newSloClassification("test-domain", "test-app", "test-class"), "aaa", nil, "", ""},
+		{newMemoryExactMatcher(logger), "test", newSloClassification("test-domain", "test-app", "test-class"), "aaa", event.SloClassification{}, "", ""},
 		{newRegexpMatcher(logger), ".*", newSloClassification("test-domain", "test-app", "test-class"), "aaa", newSloClassification("test-domain", "test-app", "test-class"), "", ""},
 		{newRegexpMatcher(logger), ".*****", newSloClassification("test-domain", "test-app", "test-class"), "aaa", newSloClassification("test-domain", "test-app", "test-class"), "failed to create new regexp endpoint classification: error parsing regexp: invalid nested repetition operator: `**`", ""},
 	}