From eaecf0ff90258b83cac29e8fac8cf1ea72c83b87 Mon Sep 17 00:00:00 2001
From: Edward Welch <edward.welch@grafana.com>
Date: Tue, 21 May 2019 17:28:46 -0400
Subject: [PATCH] Making the pipeline itself a Stage so that we can use it to
 better implement the Match stage (and it cleans up the Docker and CRI
 extensions some too) Still needs to fix the metrics test

---
 pkg/logentry/stages/extensions.go          | 142 +++++++--------------
 pkg/logentry/stages/extensions_test.go     |   4 +-
 pkg/logentry/stages/json_test.go           |   2 +-
 pkg/logentry/stages/match.go               |  84 ++++++++++--
 pkg/logentry/stages/match_test.go          |  42 ++++--
 pkg/logentry/stages/metrics_test.go        |   6 +-
 pkg/logentry/{ => stages}/pipeline.go      |  26 ++--
 pkg/logentry/{ => stages}/pipeline_test.go |  72 ++++++-----
 pkg/logentry/stages/regex_test.go          |   6 +-
 pkg/logentry/stages/stage.go               |  13 +-
 pkg/logentry/stages/timestamp.go           |   1 -
 pkg/promtail/scrape/scrape.go              |   4 +-
 pkg/promtail/targets/filetargetmanager.go  |   7 +-
 13 files changed, 229 insertions(+), 180 deletions(-)
 rename pkg/logentry/{ => stages}/pipeline.go (80%)
 rename pkg/logentry/{ => stages}/pipeline_test.go (71%)

diff --git a/pkg/logentry/stages/extensions.go b/pkg/logentry/stages/extensions.go
index c3560b0ddc..10566cfda4 100644
--- a/pkg/logentry/stages/extensions.go
+++ b/pkg/logentry/stages/extensions.go
@@ -1,114 +1,70 @@
 package stages
 
 import (
-	"time"
-
 	"github.com/go-kit/kit/log"
-	"github.com/prometheus/common/model"
 )
 
 const RFC3339Nano = "RFC3339Nano"
 
 // NewDocker creates a Docker json log format specific pipeline stage.
-func NewDocker(logger log.Logger) (Stage, error) {
-
-	// JSON Stage to extract output, stream, and timestamp
-	jsonCfg := &JSONConfig{
-		Expressions: map[string]string{
-			"output":    "log",
-			"stream":    "stream",
-			"timestamp": "time",
-		},
-	}
-	jsonStage, err := New(logger, StageTypeJSON, jsonCfg, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	// Set the stream as a label
-	lblCfg := &LabelsConfig{
-		"stream": nil,
-	}
-	labelStage, err := New(logger, StageTypeLabel, lblCfg, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	// Parse the time into the current timestamp
+func NewDocker(logger log.Logger, jobName string) (Stage, error) {
 	t := "timestamp"
 	f := RFC3339Nano
-	tsCfg := &TimestampConfig{
-		&t,
-		&f,
-	}
-	tsStage, err := New(logger, StageTypeTimestamp, tsCfg, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	// Set the output to the log message
 	o := "output"
-	outputCfg := &OutputConfig{
-		&o,
-	}
-	outputStage, err := New(logger, StageTypeOutput, outputCfg, nil)
-	if err != nil {
-		return nil, err
-	}
+	stages := PipelineStages{
+		PipelineStage{
+			StageTypeJSON: JSONConfig{
+				Expressions: map[string]string{
+					"output":    "log",
+					"stream":    "stream",
+					"timestamp": "time",
+				},
+			}},
+		PipelineStage{
+			StageTypeLabel: LabelsConfig{
+				"stream": nil,
+			}},
+		PipelineStage{
+			StageTypeTimestamp: TimestampConfig{
+				&t,
+				&f,
+			}},
+		PipelineStage{
+			StageTypeOutput: OutputConfig{
+				&o,
+			},
+		}}
 
-	return StageFunc(func(labels model.LabelSet, extracted map[string]interface{}, t *time.Time, entry *string) {
-		jsonStage.Process(labels, extracted, t, entry)
-		labelStage.Process(labels, extracted, t, entry)
-		tsStage.Process(labels, extracted, t, entry)
-		outputStage.Process(labels, extracted, t, entry)
-	}), nil
+	return NewPipeline(logger, stages, jobName+"_docker")
 }
 
 // NewCRI creates a CRI format specific pipeline stage
-func NewCRI(logger log.Logger) (Stage, error) {
-	regexCfg := &RegexConfig{
-		"^(?s)(?P<time>\\S+?) (?P<stream>stdout|stderr) (?P<flags>\\S+?) (?P<content>.*)$",
-	}
-	regexStage, err := New(logger, StageTypeRegex, regexCfg, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	// Set the stream as a label
-	lblCfg := &LabelsConfig{
-		"stream": nil,
-	}
-	labelStage, err := New(logger, StageTypeLabel, lblCfg, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	// Parse the time into the current timestamp
+func NewCRI(logger log.Logger, jobName string) (Stage, error) {
 	t := "time"
 	f := RFC3339Nano
-	tsCfg := &TimestampConfig{
-		&t,
-		&f,
-	}
-	tsStage, err := New(logger, StageTypeTimestamp, tsCfg, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	// Set the output to the log message
 	o := "content"
-	outputCfg := &OutputConfig{
-		&o,
-	}
-	outputStage, err := New(logger, StageTypeOutput, outputCfg, nil)
-	if err != nil {
-		return nil, err
+	stages := PipelineStages{
+		PipelineStage{
+			StageTypeRegex: RegexConfig{
+				"^(?s)(?P<time>\\S+?) (?P<stream>stdout|stderr) (?P<flags>\\S+?) (?P<content>.*)$",
+			},
+		},
+		PipelineStage{
+			StageTypeLabel: LabelsConfig{
+				"stream": nil,
+			},
+		},
+		PipelineStage{
+			StageTypeTimestamp: TimestampConfig{
+				&t,
+				&f,
+			},
+		},
+		PipelineStage{
+			StageTypeOutput: OutputConfig{
+				&o,
+			},
+		},
 	}
-
-	return StageFunc(func(labels model.LabelSet, extracted map[string]interface{}, t *time.Time, entry *string) {
-		regexStage.Process(labels, extracted, t, entry)
-		labelStage.Process(labels, extracted, t, entry)
-		tsStage.Process(labels, extracted, t, entry)
-		outputStage.Process(labels, extracted, t, entry)
-	}), nil
+	return NewPipeline(logger, stages, jobName+"_cri")
 }
diff --git a/pkg/logentry/stages/extensions_test.go b/pkg/logentry/stages/extensions_test.go
index f75ac7f21e..a5530480a5 100644
--- a/pkg/logentry/stages/extensions_test.go
+++ b/pkg/logentry/stages/extensions_test.go
@@ -64,7 +64,7 @@ func TestNewDocker(t *testing.T) {
 		tt := tt
 		t.Run(tName, func(t *testing.T) {
 			t.Parallel()
-			p, err := NewDocker(util.Logger)
+			p, err := NewDocker(util.Logger, "test")
 			if err != nil {
 				t.Fatalf("failed to create Docker parser: %s", err)
 			}
@@ -140,7 +140,7 @@ func TestNewCri(t *testing.T) {
 		tt := tt
 		t.Run(tName, func(t *testing.T) {
 			t.Parallel()
-			p, err := NewCRI(util.Logger)
+			p, err := NewCRI(util.Logger, "test")
 			if err != nil {
 				t.Fatalf("failed to create CRI parser: %s", err)
 			}
diff --git a/pkg/logentry/stages/json_test.go b/pkg/logentry/stages/json_test.go
index 470f1bf501..ab8173fdbd 100644
--- a/pkg/logentry/stages/json_test.go
+++ b/pkg/logentry/stages/json_test.go
@@ -174,7 +174,7 @@ func TestJSONParser_Parse(t *testing.T) {
 		tt := tt
 		t.Run(tName, func(t *testing.T) {
 			t.Parallel()
-			p, err := New(util.Logger, StageTypeJSON, tt.config, nil)
+			p, err := New(util.Logger, "test", StageTypeJSON, tt.config, nil)
 			if err != nil {
 				t.Fatalf("failed to create json parser: %s", err)
 			}
diff --git a/pkg/logentry/stages/match.go b/pkg/logentry/stages/match.go
index 7100db9dce..58427ad162 100644
--- a/pkg/logentry/stages/match.go
+++ b/pkg/logentry/stages/match.go
@@ -3,27 +3,85 @@ package stages
 import (
 	"time"
 
+	"github.com/go-kit/kit/log"
+	"github.com/mitchellh/mapstructure"
+	"github.com/pkg/errors"
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/pkg/labels"
+
 	"github.com/grafana/loki/pkg/logql"
+)
 
-	"github.com/prometheus/common/model"
+const (
+	ErrEmptyMatchStageConfig = "match stage config cannot be empty"
+	ErrPipelineNameRequired  = "match stage must specify a pipeline name which is used in the exported metrics"
+	ErrSelectorRequired      = "selector statement required for match stage"
+	ErrMatchRequiresStages   = "match stage requires at least one additional stage to be defined in '- stages'"
+	ErrSelectorSyntax        = "invalid selector syntax for match stage"
 )
 
-// withMatcher runs a stage if matcher matches an entry labelset.
-func withMatcher(s Stage, matcher string) (Stage, error) {
-	if matcher == "" {
-		return s, nil
+type MatcherConfig struct {
+	PipelineName *string        `mapstructure:"pipeline_name"`
+	Selector     *string        `mapstructure:"selector"`
+	Stages       PipelineStages `mapstructure:"stages"`
+}
+
+func validateMatcherConfig(cfg *MatcherConfig) ([]*labels.Matcher, error) {
+	if cfg == nil {
+		return nil, errors.New(ErrEmptyMatchStageConfig)
+	}
+	if cfg.PipelineName == nil || *cfg.PipelineName == "" {
+		return nil, errors.New(ErrPipelineNameRequired)
+	}
+	if cfg.Selector == nil || *cfg.Selector == "" {
+		return nil, errors.New(ErrSelectorRequired)
+	}
+	if cfg.Stages == nil || len(cfg.Stages) == 0 {
+		return nil, errors.New(ErrMatchRequiresStages)
+	}
+	matchers, err := logql.ParseMatchers(*cfg.Selector)
+	if err != nil {
+		return nil, errors.Wrap(err, ErrSelectorSyntax)
+	}
+	return matchers, nil
+}
+
+func newMatcherStage(logger log.Logger, config interface{}) (Stage, error) {
+	cfg := &MatcherConfig{}
+	err := mapstructure.Decode(config, cfg)
+	if err != nil {
+		return nil, err
 	}
-	matchers, err := logql.ParseMatchers(matcher)
+	matchers, err := validateMatcherConfig(cfg)
 	if err != nil {
 		return nil, err
 	}
 
-	return StageFunc(func(labels model.LabelSet, extracted map[string]interface{}, t *time.Time, entry *string) {
-		for _, filter := range matchers {
-			if !filter.Matches(string(labels[model.LabelName(filter.Name)])) {
-				return
-			}
+	pl, err := NewPipeline(logger, cfg.Stages, *cfg.PipelineName)
+	if err != nil {
+		return nil, errors.Wrapf(err, "match stage %s failed to create pipeline", *cfg.PipelineName)
+	}
+
+	return &matcherStage{
+		cfgs:     cfg,
+		logger:   logger,
+		matchers: matchers,
+		pipeline: pl,
+	}, nil
+}
+
+type matcherStage struct {
+	cfgs     *MatcherConfig
+	matchers []*labels.Matcher
+	logger   log.Logger
+	pipeline Stage
+}
+
+func (m *matcherStage) Process(labels model.LabelSet, extracted map[string]interface{}, t *time.Time, entry *string) {
+	for _, filter := range m.matchers {
+		if !filter.Matches(string(labels[model.LabelName(filter.Name)])) {
+			return
 		}
-		s.Process(labels, extracted, t, entry)
-	}), nil
+	}
+	m.pipeline.Process(labels, extracted, t, entry)
 }
diff --git a/pkg/logentry/stages/match_test.go b/pkg/logentry/stages/match_test.go
index a0a631f3f1..1605b4a55f 100644
--- a/pkg/logentry/stages/match_test.go
+++ b/pkg/logentry/stages/match_test.go
@@ -5,16 +5,12 @@ import (
 	"testing"
 	"time"
 
-	"github.com/prometheus/common/model"
+	"github.com/cortexproject/cortex/pkg/util"
 )
 
-type hasRunStage bool
+var pipelineName = "pl_name"
 
-func (h *hasRunStage) Process(labels model.LabelSet, exctracted map[string]interface{}, t *time.Time, entry *string) {
-	*h = true
-}
-
-func Test_withMatcher(t *testing.T) {
+func TestMatcher(t *testing.T) {
 	t.Parallel()
 	tests := []struct {
 		matcher string
@@ -28,7 +24,7 @@ func Test_withMatcher(t *testing.T) {
 		{"foo", map[string]string{"foo": "bar"}, false, true},
 		{"{}", map[string]string{"foo": "bar"}, false, true},
 		{"{", map[string]string{"foo": "bar"}, false, true},
-		{"", map[string]string{"foo": "bar"}, true, false},
+		{"", map[string]string{"foo": "bar"}, true, true},
 		{"{foo=\"bar\"}", map[string]string{"foo": "bar"}, true, false},
 		{"{foo=\"\"}", map[string]string{"foo": "bar"}, false, false},
 		{"{foo=\"\"}", map[string]string{}, true, false},
@@ -38,20 +34,40 @@ func Test_withMatcher(t *testing.T) {
 		{"{foo=\"bar\",bar=~\"te.*\"}", map[string]string{"foo": "bar", "bar": "test"}, true, false},
 		{"{foo=\"bar\",bar!~\"te.*\"}", map[string]string{"foo": "bar", "bar": "test"}, false, false},
 	}
+
 	for _, tt := range tests {
 		t.Run(fmt.Sprintf("%s/%s", tt.matcher, tt.labels), func(t *testing.T) {
-			hasRun := hasRunStage(false)
-			s, err := withMatcher(&hasRun, tt.matcher)
+			// Build a match config which has a simple label stage that when matched will add the test_label to
+			// the labels in the pipeline.
+			matchConfig := MatcherConfig{
+				&pipelineName,
+				&tt.matcher,
+				PipelineStages{
+					PipelineStage{
+						StageTypeLabel: LabelsConfig{
+							"test_label": nil,
+						},
+					},
+				},
+			}
+			s, err := newMatcherStage(util.Logger, matchConfig)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("withMatcher() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
 			if s != nil {
 				ts, entry := time.Now(), ""
-				s.Process(toLabelSet(tt.labels), map[string]interface{}{}, &ts, &entry)
+				extracted := map[string]interface{}{
+					"test_label": "unimportant value",
+				}
+				labels := toLabelSet(tt.labels)
+				s.Process(labels, extracted, &ts, &entry)
 
-				if bool(hasRun) != tt.shouldRun {
-					t.Error("stage ran but should have not")
+				// test_label should only be in the label set if the stage ran
+				if _, ok := labels["test_label"]; ok {
+					if !tt.shouldRun {
+						t.Error("stage ran but should have not")
+					}
 				}
 			}
 		})
diff --git a/pkg/logentry/stages/metrics_test.go b/pkg/logentry/stages/metrics_test.go
index 4af01123cc..455140074a 100644
--- a/pkg/logentry/stages/metrics_test.go
+++ b/pkg/logentry/stages/metrics_test.go
@@ -80,15 +80,15 @@ func Test_withMetric(t *testing.T) {
 	}
 
 	registry := prometheus.NewRegistry()
-	jsonStage, err := New(util.Logger, StageTypeJSON, jsonConfig, registry)
+	jsonStage, err := New(util.Logger, "test", StageTypeJSON, jsonConfig, registry)
 	if err != nil {
 		t.Fatalf("failed to create stage with metrics: %v", err)
 	}
-	regexStage, err := New(util.Logger, StageTypeRegex, regexConfig, registry)
+	regexStage, err := New(util.Logger, "test", StageTypeRegex, regexConfig, registry)
 	if err != nil {
 		t.Fatalf("failed to create stage with metrics: %v", err)
 	}
-	metricStage, err := New(util.Logger, StageTypeMetric, metricsConfig, registry)
+	metricStage, err := New(util.Logger, "test", StageTypeMetric, metricsConfig, registry)
 	if err != nil {
 		t.Fatalf("failed to create stage with metrics: %v", err)
 	}
diff --git a/pkg/logentry/pipeline.go b/pkg/logentry/stages/pipeline.go
similarity index 80%
rename from pkg/logentry/pipeline.go
rename to pkg/logentry/stages/pipeline.go
index ec12d86e3f..478c14d8a7 100644
--- a/pkg/logentry/pipeline.go
+++ b/pkg/logentry/stages/pipeline.go
@@ -1,4 +1,4 @@
-package logentry
+package stages
 
 import (
 	"time"
@@ -10,7 +10,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/common/model"
 
-	"github.com/grafana/loki/pkg/logentry/stages"
 	"github.com/grafana/loki/pkg/promtail/api"
 )
 
@@ -24,20 +23,23 @@ var (
 )
 
 // PipelineStages contains configuration for each stage within a pipeline
-type PipelineStages []interface{}
+type PipelineStages = []interface{}
+
+// PipelineStage contains configuration for a single pipeline stage
+type PipelineStage = map[interface{}]interface{}
 
 // Pipeline pass down a log entry to each stage for mutation and/or label extraction.
 type Pipeline struct {
 	logger  log.Logger
-	stages  []stages.Stage
+	stages  []Stage
 	jobName string
 }
 
 // NewPipeline creates a new log entry pipeline from a configuration
 func NewPipeline(logger log.Logger, stgs PipelineStages, jobName string) (*Pipeline, error) {
-	st := []stages.Stage{}
+	st := []Stage{}
 	for _, s := range stgs {
-		stage, ok := s.(map[interface{}]interface{})
+		stage, ok := s.(PipelineStage)
 		if !ok {
 			return nil, errors.Errorf("invalid YAML config, "+
 				"make sure each stage of your pipeline is a YAML object (must end with a `:`), check stage `- %s`", s)
@@ -50,7 +52,7 @@ func NewPipeline(logger log.Logger, stgs PipelineStages, jobName string) (*Pipel
 			if !ok {
 				return nil, errors.New("pipeline stage key must be a string")
 			}
-			newStage, err := stages.New(logger, name, config, prometheus.DefaultRegisterer)
+			newStage, err := New(logger, jobName, name, config, prometheus.DefaultRegisterer)
 			if err != nil {
 				return nil, errors.Wrapf(err, "invalid %s stage config", name)
 			}
@@ -64,10 +66,9 @@ func NewPipeline(logger log.Logger, stgs PipelineStages, jobName string) (*Pipel
 	}, nil
 }
 
-// Process mutates an entry and its metadata by using multiple configure stage.
-func (p *Pipeline) Process(labels model.LabelSet, ts *time.Time, entry *string) {
+// Process implements Stage allowing a pipeline stage to also be an entire pipeline
+func (p *Pipeline) Process(labels model.LabelSet, extracted map[string]interface{}, ts *time.Time, entry *string) {
 	start := time.Now()
-	extracted := map[string]interface{}{}
 	for i, stage := range p.stages {
 		level.Debug(p.logger).Log("msg", "processing pipeline", "stage", i, "labels", labels, "time", ts, "entry", entry)
 		stage.Process(labels, extracted, ts, entry)
@@ -80,13 +81,14 @@ func (p *Pipeline) Process(labels model.LabelSet, ts *time.Time, entry *string)
 // Wrap implements EntryMiddleware
 func (p *Pipeline) Wrap(next api.EntryHandler) api.EntryHandler {
 	return api.EntryHandlerFunc(func(labels model.LabelSet, timestamp time.Time, line string) error {
-		p.Process(labels, &timestamp, &line)
+		extracted := map[string]interface{}{}
+		p.Process(labels, extracted, &timestamp, &line)
 		return next.Handle(labels, timestamp, line)
 	})
 }
 
 // AddStage adds a stage to the pipeline
-func (p *Pipeline) AddStage(stage stages.Stage) {
+func (p *Pipeline) AddStage(stage Stage) {
 	p.stages = append(p.stages, stage)
 }
 
diff --git a/pkg/logentry/pipeline_test.go b/pkg/logentry/stages/pipeline_test.go
similarity index 71%
rename from pkg/logentry/pipeline_test.go
rename to pkg/logentry/stages/pipeline_test.go
index a64bcfa803..85e0e1dbc5 100644
--- a/pkg/logentry/pipeline_test.go
+++ b/pkg/logentry/stages/pipeline_test.go
@@ -1,4 +1,4 @@
-package logentry
+package stages
 
 import (
 	"testing"
@@ -9,34 +9,32 @@ import (
 	"github.com/go-kit/kit/log/level"
 	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"gopkg.in/yaml.v2"
 )
 
-var rawTestLine = `{"log":"11.11.11.11 - frank [25/Jan/2000:14:00:01 -0500] \"GET /1986.js HTTP/1.1\" 200 932 \"-\" \"Mozilla/5.0 (Windows; U; Windows NT 5.1; de; rv:1.9.1.7) Gecko/20091221 Firefox/3.5.7 GTB6\"","stream":"stderr","time":"2019-04-30T02:12:41.8443515Z"}`
-var processedTestLine = `11.11.11.11 - frank [25/Jan/2000:14:00:01 -0500] "GET /1986.js HTTP/1.1" 200 932 "-" "Mozilla/5.0 (Windows; U; Windows NT 5.1; de; rv:1.9.1.7) Gecko/20091221 Firefox/3.5.7 GTB6"`
+var (
+	ct                = time.Now()
+	rawTestLine       = `{"log":"11.11.11.11 - frank [25/Jan/2000:14:00:01 -0500] \"GET /1986.js HTTP/1.1\" 200 932 \"-\" \"Mozilla/5.0 (Windows; U; Windows NT 5.1; de; rv:1.9.1.7) Gecko/20091221 Firefox/3.5.7 GTB6\"","stream":"stderr","time":"2019-04-30T02:12:41.8443515Z"}`
+	processedTestLine = `11.11.11.11 - frank [25/Jan/2000:14:00:01 -0500] "GET /1986.js HTTP/1.1" 200 932 "-" "Mozilla/5.0 (Windows; U; Windows NT 5.1; de; rv:1.9.1.7) Gecko/20091221 Firefox/3.5.7 GTB6"`
+)
 
 var testYaml = `
 pipeline_stages:
 - match:
-    selector:
+    pipeline_name: "test_pipeline"
+    selector: "{match=\"true\"}"
     stages:
-    - match:
-       matchers:
-       stages:
-       - json:
     - docker:
     - regex:
-- docker:
-- regex:
-    expression: "^(?P<ip>\\S+) (?P<identd>\\S+) (?P<user>\\S+) \\[(?P<timestamp>[\\w:/]+\\s[+\\-]\\d{4})\\] \"(?P<action>\\S+)\\s?(?P<path>\\S+)?\\s?(?P<protocol>\\S+)?\" (?P<status>\\d{3}|-) (?P<size>\\d+|-)\\s?\"?(?P<referer>[^\"]*)\"?\\s?\"?(?P<useragent>[^\"]*)?\"?$"
-- timestamp:
-    source: timestamp
-    format: "02/Jan/2006:15:04:05 -0700"
-- labels:
-    action:
-    status_code: "status"
-- metrics:
+        expression: "^(?P<ip>\\S+) (?P<identd>\\S+) (?P<user>\\S+) \\[(?P<timestamp>[\\w:/]+\\s[+\\-]\\d{4})\\] \"(?P<action>\\S+)\\s?(?P<path>\\S+)?\\s?(?P<protocol>\\S+)?\" (?P<status>\\d{3}|-) (?P<size>\\d+|-)\\s?\"?(?P<referer>[^\"]*)\"?\\s?\"?(?P<useragent>[^\"]*)?\"?$"
+    - timestamp:
+        source: timestamp
+        format: "02/Jan/2006:15:04:05 -0700"
+    - labels:
+        action:
+        status_code: "status"
 `
 
 func loadConfig(yml string) PipelineStages {
@@ -54,9 +52,7 @@ func TestNewPipeline(t *testing.T) {
 	if err != nil {
 		panic(err)
 	}
-	if len(p.stages) != 2 {
-		t.Fatal("missing stages")
-	}
+	require.Equal(t, 1, len(p.stages))
 }
 
 func TestPipeline_MultiStage(t *testing.T) {
@@ -89,23 +85,35 @@ func TestPipeline_MultiStage(t *testing.T) {
 			time.Now(),
 			time.Date(2000, 01, 25, 14, 00, 01, 0, est),
 			map[model.LabelName]model.LabelValue{
-				"test": "test",
+				"match": "true",
 			},
 			map[model.LabelName]model.LabelValue{
-				"test":        "test",
+				"match":       "true",
 				"stream":      "stderr",
 				"action":      "GET",
 				"status_code": "200",
 			},
 		},
+		"no match": {
+			rawTestLine,
+			rawTestLine,
+			ct,
+			ct,
+			map[model.LabelName]model.LabelValue{
+				"nomatch": "true",
+			},
+			map[model.LabelName]model.LabelValue{
+				"nomatch": "true",
+			},
+		},
 	}
 
 	for tName, tt := range tests {
 		tt := tt
 		t.Run(tName, func(t *testing.T) {
 			t.Parallel()
-
-			p.Process(tt.labels, &tt.t, &tt.entry)
+			extracted := map[string]interface{}{}
+			p.Process(tt.labels, extracted, &tt.t, &tt.entry)
 
 			assert.Equal(t, tt.expectedLabels, tt.labels, "did not get expected labels")
 			assert.Equal(t, tt.expectedEntry, tt.entry, "did not receive expected log entry")
@@ -144,7 +152,8 @@ func TestPipeline_Output(t *testing.T) {
 	lbls := model.LabelSet{}
 	ts := time.Now()
 	entry := testOutputLogLine
-	pl.Process(lbls, &ts, &entry)
+	extracted := map[string]interface{}{}
+	pl.Process(lbls, extracted, &ts, &entry)
 	assert.Equal(t, "this is a log line", entry)
 }
 
@@ -181,7 +190,8 @@ func TestPipeline_Labels(t *testing.T) {
 	}
 	ts := time.Now()
 	entry := testLabelsLogLine
-	pl.Process(lbls, &ts, &entry)
+	extracted := map[string]interface{}{}
+	pl.Process(lbls, extracted, &ts, &entry)
 	assert.Equal(t, expectedLbls, lbls)
 }
 
@@ -212,7 +222,8 @@ func TestPipeline_Timestamp(t *testing.T) {
 	lbls := model.LabelSet{}
 	ts := time.Now()
 	entry := testTimestampLogLine
-	pl.Process(lbls, &ts, &entry)
+	extracted := map[string]interface{}{}
+	pl.Process(lbls, extracted, &ts, &entry)
 	assert.Equal(t, time.Date(2012, 11, 01, 22, 8, 41, 0, time.FixedZone("", 0)), ts)
 }
 
@@ -224,7 +235,7 @@ var (
 	debugLogger = level.NewFilter(l, level.AllowDebug())
 )
 
-func Benchmark(b *testing.B) {
+func BenchmarkPipeline(b *testing.B) {
 	benchmarks := []struct {
 		name   string
 		stgs   PipelineStages
@@ -254,7 +265,8 @@ func Benchmark(b *testing.B) {
 			ts := time.Now()
 			for i := 0; i < b.N; i++ {
 				entry := bm.entry
-				pl.Process(lb, &ts, &entry)
+				extracted := map[string]interface{}{}
+				pl.Process(lb, extracted, &ts, &entry)
 			}
 		})
 	}
diff --git a/pkg/logentry/stages/regex_test.go b/pkg/logentry/stages/regex_test.go
index 02eef99584..a33d2f9fff 100644
--- a/pkg/logentry/stages/regex_test.go
+++ b/pkg/logentry/stages/regex_test.go
@@ -129,7 +129,7 @@ func TestRegexParser_Parse(t *testing.T) {
 		tt := tt
 		t.Run(tName, func(t *testing.T) {
 			t.Parallel()
-			p, err := New(util.Logger, StageTypeRegex, tt.config, nil)
+			p, err := New(util.Logger, "test", StageTypeRegex, tt.config, nil)
 			if err != nil {
 				t.Fatalf("failed to create regex parser: %s", err)
 			}
@@ -143,7 +143,7 @@ func TestRegexParser_Parse(t *testing.T) {
 	}
 }
 
-func Benchmark(b *testing.B) {
+func BenchmarkRegexStage(b *testing.B) {
 	benchmarks := []struct {
 		name   string
 		config map[string]interface{}
@@ -171,7 +171,7 @@ func Benchmark(b *testing.B) {
 	}
 	for _, bm := range benchmarks {
 		b.Run(bm.name, func(b *testing.B) {
-			stage, err := New(util.Logger, StageTypeRegex, bm.config, nil)
+			stage, err := New(util.Logger, "test", StageTypeRegex, bm.config, nil)
 			if err != nil {
 				panic(err)
 			}
diff --git a/pkg/logentry/stages/stage.go b/pkg/logentry/stages/stage.go
index 821a28f82a..2d96d44746 100644
--- a/pkg/logentry/stages/stage.go
+++ b/pkg/logentry/stages/stage.go
@@ -18,6 +18,7 @@ const (
 	StageTypeOutput    = "output"
 	StageTypeDocker    = "docker"
 	StageTypeCRI       = "cri"
+	StageTypeMatch     = "match"
 )
 
 // Stage takes an existing set of labels, timestamp and log entry and returns either a possibly mutated
@@ -35,17 +36,18 @@ func (s StageFunc) Process(labels model.LabelSet, extracted map[string]interface
 }
 
 // New creates a new stage for the given type and configuration.
-func New(logger log.Logger, stageType string, cfg interface{}, registerer prometheus.Registerer) (Stage, error) {
+func New(logger log.Logger, jobName string, stageType string,
+	cfg interface{}, registerer prometheus.Registerer) (Stage, error) {
 	var s Stage
 	var err error
 	switch stageType {
 	case StageTypeDocker:
-		s, err = NewDocker(logger)
+		s, err = NewDocker(logger, jobName)
 		if err != nil {
 			return nil, err
 		}
 	case StageTypeCRI:
-		s, err = NewCRI(logger)
+		s, err = NewCRI(logger, jobName)
 		if err != nil {
 			return nil, err
 		}
@@ -79,6 +81,11 @@ func New(logger log.Logger, stageType string, cfg interface{}, registerer promet
 		if err != nil {
 			return nil, err
 		}
+	case StageTypeMatch:
+		s, err = newMatcherStage(logger, cfg)
+		if err != nil {
+			return nil, err
+		}
 	default:
 		return nil, errors.Errorf("Unknown stage type: %s", stageType)
 	}
diff --git a/pkg/logentry/stages/timestamp.go b/pkg/logentry/stages/timestamp.go
index 6357e6b8f7..128bf8426e 100644
--- a/pkg/logentry/stages/timestamp.go
+++ b/pkg/logentry/stages/timestamp.go
@@ -15,7 +15,6 @@ const (
 	ErrEmptyTimestampStageConfig = "timestamp stage config cannot be empty"
 	ErrTimestampSourceRequired   = "timestamp source value is required if timestamp is specified"
 	ErrTimestampFormatRequired   = "timestamp format is required"
-	ErrInvalidTimestampFormat    = "failed to parse timestamp format: %s"
 )
 
 // TimestampConfig configures timestamp extraction
diff --git a/pkg/promtail/scrape/scrape.go b/pkg/promtail/scrape/scrape.go
index d2e8402297..8a0f0c1592 100644
--- a/pkg/promtail/scrape/scrape.go
+++ b/pkg/promtail/scrape/scrape.go
@@ -6,7 +6,7 @@ import (
 	sd_config "github.com/prometheus/prometheus/discovery/config"
 	"github.com/prometheus/prometheus/pkg/relabel"
 
-	"github.com/grafana/loki/pkg/logentry"
+	"github.com/grafana/loki/pkg/logentry/stages"
 	"github.com/grafana/loki/pkg/promtail/api"
 )
 
@@ -14,7 +14,7 @@ import (
 type Config struct {
 	JobName                string                           `yaml:"job_name,omitempty"`
 	EntryParser            api.EntryParser                  `yaml:"entry_parser"`
-	PipelineStages         logentry.PipelineStages          `yaml:"pipeline_stages,omitempty"`
+	PipelineStages         stages.PipelineStages            `yaml:"pipeline_stages,omitempty"`
 	RelabelConfigs         []*relabel.Config                `yaml:"relabel_configs,omitempty"`
 	ServiceDiscoveryConfig sd_config.ServiceDiscoveryConfig `yaml:",inline"`
 }
diff --git a/pkg/promtail/targets/filetargetmanager.go b/pkg/promtail/targets/filetargetmanager.go
index 640b4eca50..de04ee6f35 100644
--- a/pkg/promtail/targets/filetargetmanager.go
+++ b/pkg/promtail/targets/filetargetmanager.go
@@ -21,7 +21,6 @@ import (
 	"github.com/prometheus/prometheus/relabel"
 
 	"github.com/grafana/loki/pkg/helpers"
-	"github.com/grafana/loki/pkg/logentry"
 	"github.com/grafana/loki/pkg/logentry/stages"
 	"github.com/grafana/loki/pkg/promtail/api"
 	"github.com/grafana/loki/pkg/promtail/positions"
@@ -77,7 +76,7 @@ func NewFileTargetManager(
 
 	config := map[string]sd_config.ServiceDiscoveryConfig{}
 	for _, cfg := range scrapeConfigs {
-		pipeline, err := logentry.NewPipeline(log.With(logger, "component", "pipeline"), cfg.PipelineStages, cfg.JobName)
+		pipeline, err := stages.NewPipeline(log.With(logger, "component", "pipeline"), cfg.PipelineStages, cfg.JobName)
 		if err != nil {
 			return nil, err
 		}
@@ -87,14 +86,14 @@ func NewFileTargetManager(
 			switch cfg.EntryParser {
 			case api.CRI:
 				level.Warn(logger).Log("msg", "WARNING!!! entry_parser config is deprecated, please change to pipeline_stages")
-				cri, err := stages.NewCRI(logger)
+				cri, err := stages.NewCRI(logger, cfg.JobName)
 				if err != nil {
 					return nil, err
 				}
 				pipeline.AddStage(cri)
 			case api.Docker:
 				level.Warn(logger).Log("msg", "WARNING!!! entry_parser config is deprecated, please change to pipeline_stages")
-				docker, err := stages.NewDocker(logger)
+				docker, err := stages.NewDocker(logger, cfg.JobName)
 				if err != nil {
 					return nil, err
 				}