Added basic multiline support based on stage-chains idea

pkg/logentry/stages/extensions.go

@@ -8,7 +8,7 @@ import (
 const RFC3339Nano = "RFC3339Nano"
 
 // NewDocker creates a Docker json log format specific pipeline stage.
-func NewDocker(logger log.Logger, registerer prometheus.Registerer) (Stage, error) {
+func NewDocker(logger log.Logger, registerer prometheus.Registerer) (*Pipeline, error) {
 	stages := PipelineStages{
 		PipelineStage{
 			StageTypeJSON: JSONConfig{
@@ -36,7 +36,7 @@ func NewDocker(logger log.Logger, registerer prometheus.Registerer) (Stage, erro
 }
 
 // NewCRI creates a CRI format specific pipeline stage
-func NewCRI(logger log.Logger, registerer prometheus.Registerer) (Stage, error) {
+func NewCRI(logger log.Logger, registerer prometheus.Registerer) (*Pipeline, error) {
 	stages := PipelineStages{
 		PipelineStage{
 			StageTypeRegex: RegexConfig{

pkg/logentry/stages/extensions_test.go

@@ -71,11 +71,12 @@ func TestNewDocker(t *testing.T) {
 			}
 			lbs := toLabelSet(tt.labels)
 			extr := map[string]interface{}{}
-			p.Process(lbs, extr, &tt.t, &tt.entry)
+			result := &resultChain{}
+			p.Process(lbs, extr, tt.t, tt.entry, result)
 
-			assertLabels(t, tt.expectedLabels, lbs)
-			assert.Equal(t, tt.expectedEntry, tt.entry, "did not receive expected log entry")
-			if tt.t.Unix() != tt.expectedT.Unix() {
+			assertLabels(t, tt.expectedLabels, result.labels)
+			assert.Equal(t, tt.expectedEntry, result.entry, "did not receive expected log entry")
+			if result.time.Unix() != tt.expectedT.Unix() {
 				t.Fatalf("mismatch ts want: %s got:%s", tt.expectedT, tt.t)
 			}
 		})
@@ -147,11 +148,12 @@ func TestNewCri(t *testing.T) {
 			}
 			lbs := toLabelSet(tt.labels)
 			extr := map[string]interface{}{}
-			p.Process(lbs, extr, &tt.t, &tt.entry)
+			result := &resultChain{}
+			p.Process(lbs, extr, tt.t, tt.entry, result)
 
-			assertLabels(t, tt.expectedLabels, lbs)
-			assert.Equal(t, tt.expectedEntry, tt.entry, "did not receive expected log entry")
-			if tt.t.Unix() != tt.expectedT.Unix() {
+			assertLabels(t, tt.expectedLabels, result.labels)
+			assert.Equal(t, tt.expectedEntry, result.entry, "did not receive expected log entry")
+			if result.time.Unix() != tt.expectedT.Unix() {
 				t.Fatalf("mismatch ts want: %s got:%s", tt.expectedT, tt.t)
 			}
 		})

pkg/logentry/stages/json.go

@@ -92,7 +92,9 @@ func parseJSONConfig(config interface{}) (*JSONConfig, error) {
 }
 
 // Process implements Stage
-func (j *jsonStage) Process(labels model.LabelSet, extracted map[string]interface{}, t *time.Time, entry *string) {
+func (j *jsonStage) Process(labels model.LabelSet, extracted map[string]interface{}, time time.Time, entry string, chain StageChain) {
+	defer chain.NextStage(labels, extracted, time, entry) // we can use defer, as we don't update time or entry
+
 	// If a source key is provided, the json stage should process it
 	// from the exctracted map, otherwise should fallback to the entry
 	input := entry
@@ -113,19 +115,12 @@ func (j *jsonStage) Process(labels model.LabelSet, extracted map[string]interfac
 			return
 		}
 
-		input = &value
-	}
-
-	if input == nil {
-		if Debug {
-			level.Debug(j.logger).Log("msg", "cannot parse a nil entry")
-		}
-		return
+		input = value
 	}
 
 	var data map[string]interface{}
 
-	if err := json.Unmarshal([]byte(*input), &data); err != nil {
+	if err := json.Unmarshal([]byte(input), &data); err != nil {
 		if Debug {
 			level.Debug(j.logger).Log("msg", "failed to unmarshal log line", "err", err)
 		}

pkg/logentry/stages/json_test.go

@@ -91,8 +91,9 @@ func TestPipeline_JSON(t *testing.T) {
 			ts := time.Now()
 			entry := testData.entry
 			extracted := map[string]interface{}{}
-			pl.Process(lbls, extracted, &ts, &entry)
-			assert.Equal(t, testData.expectedExtract, extracted)
+			result := &resultChain{}
+			pl.Process(lbls, extracted, ts, entry, result)
+			assert.Equal(t, testData.expectedExtract, result.extracted)
 		})
 	}
 }
@@ -351,10 +352,10 @@ func TestJSONParser_Parse(t *testing.T) {
 			}
 			lbs := model.LabelSet{}
 			extr := tt.extracted
-			ts := time.Now()
-			p.Process(lbs, extr, &ts, &tt.entry)
+			result := &resultChain{}
+			p.Process(lbs, extr, time.Now(), tt.entry, result)
 
-			assert.Equal(t, tt.expectedExtract, extr)
+			assert.Equal(t, tt.expectedExtract, result.extracted)
 		})
 	}
 }

pkg/logentry/stages/labels.go

@@ -62,7 +62,7 @@ type labelStage struct {
 }
 
 // Process implements Stage
-func (l *labelStage) Process(labels model.LabelSet, extracted map[string]interface{}, t *time.Time, entry *string) {
+func (l *labelStage) Process(labels model.LabelSet, extracted map[string]interface{}, time time.Time, entry string, chain StageChain) {
 	for lName, lSrc := range l.cfgs {
 		if lValue, ok := extracted[*lSrc]; ok {
 			s, err := getString(lValue)
@@ -82,6 +82,8 @@ func (l *labelStage) Process(labels model.LabelSet, extracted map[string]interfa
 			labels[model.LabelName(lName)] = labelValue
 		}
 	}
+
+	chain.NextStage(labels, extracted, time, entry)
 }
 
 // Name implements Stage

pkg/logentry/stages/labels_test.go

@@ -42,11 +42,10 @@ func TestLabelsPipeline_Labels(t *testing.T) {
 		"level": "WARN",
 		"app":   "loki",
 	}
-	ts := time.Now()
-	entry := testLabelsLogLine
 	extracted := map[string]interface{}{}
-	pl.Process(lbls, extracted, &ts, &entry)
-	assert.Equal(t, expectedLbls, lbls)
+	result := &resultChain{}
+	pl.Process(lbls, extracted, time.Now(), testLabelsLogLine, result)
+	assert.Equal(t, expectedLbls, result.labels)
 }
 
 var (
@@ -157,8 +156,11 @@ func TestLabelStage_Process(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			st.Process(test.inputLabels, test.extractedData, nil, nil)
-			assert.Equal(t, test.expectedLabels, test.inputLabels)
+
+			result := &resultChain{}
+
+			st.Process(test.inputLabels, test.extractedData, time.Now(), "", result)
+			assert.Equal(t, test.expectedLabels, result.labels)
 		})
 	}
 }

pkg/logentry/stages/match.go

@@ -68,7 +68,7 @@ func validateMatcherConfig(cfg *MatcherConfig) (logql.LogSelectorExpr, error) {
 }
 
 // newMatcherStage creates a new matcherStage from config
-func newMatcherStage(logger log.Logger, jobName *string, config interface{}, registerer prometheus.Registerer) (Stage, error) {
+func newMatcherStage(logger log.Logger, jobName *string, config interface{}, registerer prometheus.Registerer) (*matcherStage, error) {
 	cfg := &MatcherConfig{}
 	err := mapstructure.Decode(config, cfg)
 	if err != nil {
@@ -111,24 +111,27 @@ func newMatcherStage(logger log.Logger, jobName *string, config interface{}, reg
 type matcherStage struct {
 	matchers []*labels.Matcher
 	filter   logql.Filter
-	pipeline Stage
+	pipeline *Pipeline
 	action   string
 }
 
 // Process implements Stage
-func (m *matcherStage) Process(labels model.LabelSet, extracted map[string]interface{}, t *time.Time, entry *string) {
+func (m *matcherStage) Process(labels model.LabelSet, extracted map[string]interface{}, time time.Time, entry string, chain StageChain) {
 	for _, filter := range m.matchers {
 		if !filter.Matches(string(labels[model.LabelName(filter.Name)])) {
+			chain.NextStage(labels, extracted, time, entry)
 			return
 		}
 	}
-	if m.filter == nil || m.filter([]byte(*entry)) {
+	if m.filter == nil || m.filter([]byte(entry)) {
 		switch m.action {
 		case MatchActionDrop:
 			// Adds the drop label to not be sent by the api.EntryHandler
 			labels[dropLabel] = ""
+			chain.NextStage(labels, extracted, time, entry)
 		case MatchActionKeep:
-			m.pipeline.Process(labels, extracted, t, entry)
+			// run all stages in the pipeline, and then continue with our original chain afterwards
+			m.pipeline.Process(labels, extracted, time, entry, matcherChain{chain: chain})
 		}
 	}
 }
@@ -137,3 +140,11 @@ func (m *matcherStage) Process(labels model.LabelSet, extracted map[string]inter
 func (m *matcherStage) Name() string {
 	return StageTypeMatch
 }
+
+type matcherChain struct {
+	chain StageChain
+}
+
+func (mc matcherChain) NextStage(labels model.LabelSet, extracted map[string]interface{}, time time.Time, entry string) {
+	mc.chain.NextStage(labels, extracted, time, entry)
+}

pkg/logentry/stages/match_test.go

@@ -65,18 +65,18 @@ func TestMatchPipeline(t *testing.T) {
 		t.Fatal(err)
 	}
 	lbls := model.LabelSet{}
-	ts := time.Now()
+
 	// Process the first log line which should extract the output from the `message` field
-	entry := testMatchLogLineApp1
 	extracted := map[string]interface{}{}
-	pl.Process(lbls, extracted, &ts, &entry)
-	assert.Equal(t, "app1 log line", entry)
+	result := &resultChain{}
+	pl.Process(lbls, extracted, time.Now(), testMatchLogLineApp1, result)
+	assert.Equal(t, "app1 log line", result.entry)
 
 	// Process the second log line which should extract the output from the `msg` field
-	entry = testMatchLogLineApp2
 	extracted = map[string]interface{}{}
-	pl.Process(lbls, extracted, &ts, &entry)
-	assert.Equal(t, "app2 log line", entry)
+	result = &resultChain{}
+	pl.Process(lbls, extracted, time.Now(), testMatchLogLineApp2, result)
+	assert.Equal(t, "app2 log line", result.entry)
 
 	got, err := registry.Gather()
 	if err != nil {
@@ -170,21 +170,21 @@ func TestMatcher(t *testing.T) {
 				return
 			}
 			if s != nil {
-				ts, entry := time.Now(), "foo"
 				extracted := map[string]interface{}{
 					"test_label": "unimportant value",
 				}
 				labels := toLabelSet(tt.labels)
-				s.Process(labels, extracted, &ts, &entry)
+				result := &resultChain{}
+				s.Process(labels, extracted, time.Now(), "foo", result)
 
 				// test_label should only be in the label set if the stage ran
-				if _, ok := labels["test_label"]; ok {
+				if _, ok := result.labels["test_label"]; ok {
 					if !tt.shouldRun {
 						t.Error("stage ran but should have not")
 					}
 				}
 				if tt.shouldDrop {
-					if _, ok := labels[dropLabel]; !ok {
+					if _, ok := result.labels[dropLabel]; !ok {
 						t.Error("stage should have been dropped")
 					}
 				}

pkg/logentry/stages/metrics.go

@@ -115,7 +115,7 @@ type metricStage struct {
 }
 
 // Process implements Stage
-func (m *metricStage) Process(labels model.LabelSet, extracted map[string]interface{}, t *time.Time, entry *string) {
+func (m *metricStage) Process(labels model.LabelSet, extracted map[string]interface{}, time time.Time, entry string, chain StageChain) {
 	for name, collector := range m.metrics {
 		if v, ok := extracted[*m.cfg[name].Source]; ok {
 			switch vec := collector.(type) {
@@ -128,6 +128,8 @@ func (m *metricStage) Process(labels model.LabelSet, extracted map[string]interf
 			}
 		}
 	}
+
+	chain.NextStage(labels, extracted, time, entry)
 }
 
 // Name implements Stage

pkg/logentry/stages/metrics_test.go

@@ -84,12 +84,9 @@ func TestMetricsPipeline(t *testing.T) {
 		t.Fatal(err)
 	}
 	lbls := model.LabelSet{}
-	ts := time.Now()
 	extracted := map[string]interface{}{}
-	entry := testMetricLogLine1
-	pl.Process(lbls, extracted, &ts, &entry)
-	entry = testMetricLogLine2
-	pl.Process(lbls, extracted, &ts, &entry)
+	pl.Process(lbls, extracted, time.Now(), testMetricLogLine1, &resultChain{})
+	pl.Process(lbls, extracted, time.Now(), testMetricLogLine2, &resultChain{})
 
 	if err := testutil.GatherAndCompare(registry,
 		strings.NewReader(expectedMetrics)); err != nil {
@@ -244,14 +241,14 @@ func TestMetricStage_Process(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed to create stage with metrics: %v", err)
 	}
-	var ts = time.Now()
-	var entry = logFixture
+
 	extr := map[string]interface{}{}
-	jsonStage.Process(labelFoo, extr, &ts, &entry)
-	regexStage.Process(labelFoo, extr, &ts, &regexLogFixture)
-	metricStage.Process(labelFoo, extr, &ts, &entry)
+
+	jsonStage.Process(labelFoo, extr, time.Now(), logFixture, &resultChain{})
+	regexStage.Process(labelFoo, extr, time.Now(), regexLogFixture, &resultChain{})
+	metricStage.Process(labelFoo, extr, time.Now(), logFixture, &resultChain{})
 	// Process the same extracted values again with different labels so we can verify proper metric/label assignments
-	metricStage.Process(labelFu, extr, &ts, &entry)
+	metricStage.Process(labelFu, extr, time.Now(), logFixture, &resultChain{})
 
 	names := metricNames(metricsConfig)
 	if err := testutil.GatherAndCompare(registry,