Promtail: adding pipeline stage inspector

clients/cmd/promtail/main.go

@@ -6,6 +6,9 @@ import (
 	"os"
 	"reflect"
 
+	// embed time zone data
+	_ "time/tzdata"
+
 	"k8s.io/klog"
 
 	"github.com/cortexproject/cortex/pkg/util/flagext"
@@ -15,9 +18,6 @@ import (
 	"github.com/prometheus/common/version"
 	"github.com/weaveworks/common/logging"
 
-	// embed time zone data
-	_ "time/tzdata"
-
 	"github.com/grafana/loki/clients/pkg/logentry/stages"
 	"github.com/grafana/loki/clients/pkg/promtail"
 	"github.com/grafana/loki/clients/pkg/promtail/config"
@@ -39,6 +39,7 @@ type Config struct {
 	dryRun          bool
 	configFile      string
 	configExpandEnv bool
+	inspect         bool
 }
 
 func (c *Config) RegisterFlags(f *flag.FlagSet) {
@@ -47,6 +48,7 @@ func (c *Config) RegisterFlags(f *flag.FlagSet) {
 	f.BoolVar(&c.logConfig, "log-config-reverse-order", false, "Dump the entire Loki config object at Info log "+
 		"level with the order reversed, reversing the order makes viewing the entries easier in Grafana.")
 	f.BoolVar(&c.dryRun, "dry-run", false, "Start Promtail but print entries instead of sending them to Loki.")
+	f.BoolVar(&c.inspect, "inspect", false, "Allows for detailed inspection of pipeline stages")
 	f.StringVar(&c.configFile, "config.file", "", "yaml file to load")
 	f.BoolVar(&c.configExpandEnv, "config.expand-env", false, "Expands ${var} in config according to the values of the environment variables.")
 	c.Config.RegisterFlags(f)
@@ -84,6 +86,10 @@ func main() {
 	// Use Stderr instead of files for the klog.
 	klog.SetOutput(os.Stderr)
 
+	if config.inspect {
+		stages.Inspect = true
+	}
+
 	// Set the global debug variable in the stages package which is used to conditionally log
 	// debug messages which otherwise cause huge allocations processing log lines for log messages never printed
 	if config.ServerConfig.Config.LogLevel.String() == "debug" {

clients/pkg/logentry/stages/inspector.go

@@ -0,0 +1,129 @@
+package stages
+
+import (
+	"fmt"
+	"io"
+	"strings"
+
+	"github.com/fatih/color"
+	"github.com/google/go-cmp/cmp"
+)
+
+type inspector struct {
+	writer    io.Writer
+	formatter *formatter
+}
+
+func newInspector(writer io.Writer, disableFormatting bool) *inspector {
+	f := &formatter{
+		red:    color.New(color.FgRed),
+		yellow: color.New(color.FgYellow),
+		green:  color.New(color.FgGreen),
+		bold:   color.New(color.Bold),
+	}
+
+	if disableFormatting {
+		f.disable()
+	}
+
+	return &inspector{
+		writer:    writer,
+		formatter: f,
+	}
+}
+
+type formatter struct {
+	red    *color.Color
+	yellow *color.Color
+	green  *color.Color
+	bold   *color.Color
+}
+
+func (f *formatter) disable() {
+	f.red.DisableColor()
+	f.yellow.DisableColor()
+	f.green.DisableColor()
+	f.bold.DisableColor()
+}
+
+func (i inspector) inspect(stageName string, before *Entry, after Entry) {
+	if before == nil {
+		fmt.Fprintln(i.writer, i.formatter.red.Sprintf("could not copy entry in '%s' stage; inspect aborted", stageName))
+		return
+	}
+
+	r := diffReporter{
+		formatter: i.formatter,
+	}
+
+	cmp.Equal(*before, after, cmp.Reporter(&r))
+
+	diff := r.String()
+	if strings.TrimSpace(diff) == "" {
+		diff = i.formatter.red.Sprintf("none")
+	}
+
+	fmt.Fprintf(i.writer, "[inspect: %s stage]: %s\n", i.formatter.bold.Sprintf("%s", stageName), diff)
+}
+
+// diffReporter is a simple custom reporter that only records differences
+// detected during comparison.
+type diffReporter struct {
+	path cmp.Path
+
+	formatter *formatter
+
+	diffs []string
+}
+
+func (r *diffReporter) PushStep(ps cmp.PathStep) {
+	r.path = append(r.path, ps)
+}
+
+func (r *diffReporter) Report(rs cmp.Result) {
+	if rs.Equal() {
+		return
+	}
+
+	vx, vy := r.path.Last().Values()
+
+	// TODO(dannyk): try using go-cmp to filter this condition out with Equal(), but for now this just makes it work
+	if fmt.Sprintf("%v", vx) == fmt.Sprintf("%v", vy) {
+		return
+	}
+
+	change := vx.IsValid()
+	addition := vy.IsValid()
+	removal := change && !addition
+	mod := addition && change
+
+	var titleColor *color.Color
+	switch {
+	case mod:
+		titleColor = r.formatter.yellow
+	case removal:
+		titleColor = r.formatter.red
+	default:
+		titleColor = r.formatter.green
+	}
+
+	r.diffs = append(r.diffs, titleColor.Sprintf("%#v:", r.path))
+
+	if removal {
+		r.diffs = append(r.diffs, r.formatter.red.Sprintf("\t-: %v", vx))
+	}
+	if mod {
+		r.diffs = append(r.diffs, r.formatter.yellow.Sprintf("\t-: %v", vx))
+	}
+	if addition {
+		r.diffs = append(r.diffs, r.formatter.green.Sprintf("\t+: %v", vy))
+	}
+}
+
+func (r *diffReporter) PopStep() {
+	r.path = r.path[:len(r.path)-1]
+}
+
+func (r *diffReporter) String() string {
+	return fmt.Sprintf("\n%s", strings.Join(r.diffs, "\n"))
+}

clients/pkg/logentry/stages/stage.go

@@ -1,12 +1,15 @@
 package stages
 
 import (
+	"os"
+	"runtime"
 	"time"
 
 	"github.com/go-kit/kit/log"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
+	"gopkg.in/yaml.v2"
 
 	"github.com/grafana/loki/clients/pkg/promtail/api"
 )
@@ -50,20 +53,51 @@ type Stage interface {
 	Run(chan Entry) chan Entry
 }
 
+func (entry *Entry) copy() *Entry {
+	out, err := yaml.Marshal(entry)
+	if err != nil {
+		return nil
+	}
+
+	var n *Entry
+	err = yaml.Unmarshal(out, &n)
+	if err != nil {
+		return nil
+	}
+
+	return n
+}
+
 // stageProcessor Allow to transform a Processor (old synchronous pipeline stage) into an async Stage
 type stageProcessor struct {
 	Processor
+
+	inspector *inspector
 }
 
 func (s stageProcessor) Run(in chan Entry) chan Entry {
 	return RunWith(in, func(e Entry) Entry {
+		var before *Entry
+
+		if Inspect {
+			before = e.copy()
+		}
+
 		s.Process(e.Labels, e.Extracted, &e.Timestamp, &e.Line)
+
+		if Inspect {
+			s.inspector.inspect(s.Processor.Name(), before, e)
+		}
+
 		return e
 	})
 }
 
 func toStage(p Processor) Stage {
-	return &stageProcessor{Processor: p}
+	return &stageProcessor{
+		Processor: p,
+		inspector: newInspector(os.Stderr, runtime.GOOS == "windows"),
+	}
 }
 
 // New creates a new stage for the given type and configuration.

clients/pkg/logentry/stages/util.go

@@ -14,6 +14,9 @@ var (
 	// debug level when debug level logging is not enabled. Log level allocations can become very expensive
 	// as we log numerous log entries per log line at debug level.
 	Debug = false
+
+	// Inspect is used to debug promtail pipelines by showing diffs between pipeline stages
+	Inspect = false
 )
 
 const (

docs/sources/clients/promtail/troubleshooting.md

@@ -19,6 +19,31 @@ To start Promtail in dry run mode use the flag `--dry-run` as shown in the examp
 cat my.log | promtail --stdin --dry-run --client.url http://127.0.0.1:3100/loki/api/v1/push
 ```
 
+## Inspecting pipeline stages
+
+Promtail can output all changes to log entries as each pipeline stage is executed.
+Each log entry contains four fields:
+- line
+- timestamp
+- labels
+- extracted fields
+
+Enable the inspection output using the `--inspect` command-line option.<br>
+The `--inspect` option can be used in combination with `--stdin` and `--dry-run`.
+
+```bash
+cat my.log | promtail --stdin --dry-run --inspect --client.url http://127.0.0.1:3100/loki/api/v1/push
+```
+
+![screenshot](../inspect.png)
+
+The output uses color to highlight changes. Additions are in green, modifications in yellow, and removals in red.
+
+If no changes are applied during a stage, that is usually an indication of a misconfiguration or undesired behavior.
+
+The `--inspect` flag should not be used in production, as the calculation of changes between pipeline stages impacts
+Promtail's performance.
+
 ## Pipe data to Promtail
 
 Promtail supports piping data for sending logs to Loki (via the flag `--stdin`). This is a very useful way to troubleshooting your configuration.

go.mod

@@ -29,6 +29,7 @@ require (
 	github.com/gofrs/uuid v4.0.0+incompatible
 	github.com/gogo/protobuf v1.3.2 // remember to update loki-build-image/Dockerfile too
 	github.com/golang/snappy v0.0.3
+	github.com/google/go-cmp v0.5.6
 	github.com/gorilla/mux v1.7.3
 	github.com/gorilla/websocket v1.4.2
 	github.com/grpc-ecosystem/go-grpc-middleware v1.2.2

go.sum

@@ -771,8 +771,9 @@ github.com/google/go-cmp v0.5.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.6 h1:BKbKCqvP6I+rmFHt06ZmyQtvB8xAkWdhFyr0ZUNZcxQ=
+github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-github v17.0.0+incompatible/go.mod h1:zLgOLi98H3fifZn+44m+umXrS52loVEgC2AApnigrVQ=
 github.com/google/go-github/v32 v32.1.0/go.mod h1:rIEpZD9CTDQwDK9GDrtMTycQNA4JU3qBsCizh3q2WCI=
 github.com/google/go-querystring v0.0.0-20170111101155-53e6ce116135/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=