Allows to pass inlined pipeline stages to the docker driver.

cmd/docker-driver/README.md

@@ -110,7 +110,54 @@ Loki can received a set of labels along with log line. These labels are used to
 
 By default the Docker driver will add the `filename` where the log is written, the `host` where the log has been generated as well as the `container_name`. Additionally `swarm_stack` and `swarm_service` are added for Docker Swarm deployments.
 
-You can add more labels by using `loki-external-labels`,`loki-pipeline-stage-file`,`labels`,`env` and `env-regex` options as described below.
+You can add more labels by using `loki-external-labels`,`loki-pipeline-stages`,`loki-pipeline-stage-file`,`labels`,`env` and `env-regex` options as described below.
+
+## Pipeline stages
+
+While you can provide `loki-pipeline-stage-file` it can be hard to mount the configuration file to the driver root filesystem.
+This is why another option `loki-pipeline-stages` is available allowing your to pass a list of stages inlined.
+
+The example [docker-compose](./docker-compose.yaml) below configures 2 stages, one to extract level values and one to set it as a label:
+
+```yaml
+version: "3"
+services:
+  nginx:
+    image: grafana/grafana
+    logging:
+      driver: loki
+      options:
+        loki-url: http://host.docker.internal:3100/loki/api/v1/push
+        loki-pipeline-stages: |
+          - regex:
+              expression: '(level|lvl|severity)=(?P<level>\w+)'
+          - labels:
+              level:
+    ports:
+      - "3000:3000"
+```
+
+> Note the `loki-pipeline-stages: |` allowing to keep the indentation correct.
+
+When using docker run you can also pass the value via a string parameter like such:
+
+```bash
+read -d '' stages << EOF
+- regex:
+     expression: '(level|lvl|severity)=(?P<level>\\\w+)'
+- labels:
+    level:
+EOF
+
+docker run --log-driver=loki \
+    --log-opt loki-url="http://host.docker.internal:3100/loki/api/v1/push" \
+    --log-opt loki-pipeline-stages="$stages" \
+    -p 3000:3000 grafana/grafana
+```
+
+This is a bit more difficult as you need to properly escape bash special characters. (note `\\\w+` for `\w+`)
+
+Providing both `loki-pipeline-stage-file` and `loki-pipeline-stages` will cause an error.
 
 ## log-opt options
 
@@ -127,6 +174,7 @@ To specify additional logging driver options, you can use the --log-opt NAME=VAL
 | `loki-max-backoff`              |    No     |           `10s`            | The maximum amount of time to wait before retrying a batch. Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".                                                                                                                                                   |
 | `loki-retries`                  |    No     |            `10`            | The maximum amount of retries for a log batch.                                                                                                                                                                                                                                |
 | `loki-pipeline-stage-file`      |    No     |                            | The location of a pipeline stage configuration file ([example](./pipeline-example.yaml)). Pipeline stages allows to parse log lines to extract more labels. [see documentation](../../docs/logentry/processing-log-lines.md)                                                  |
+| `loki-pipeline-stages`          |    No     |                            | The list of pipeline stages provided as a string [see](#pipeline-stages) and  [see documentation](../../docs/logentry/processing-log-lines.md)                                                                                                                                |
 | `loki-tenant-id`                |    No     |                            | Set the tenant id (http header`X-Scope-OrgID`) when sending logs to Loki. It can be overrides by a pipeline stage.                                                                                                                                                            |
 | `loki-tls-ca-file`              |    No     |                            | Set the path to a custom certificate authority.                                                                                                                                                                                                                               |
 | `loki-tls-cert-file`            |    No     |                            | Set the path to a client certificate file.                                                                                                                                                                                                                                    |

cmd/docker-driver/config.go

@@ -14,6 +14,7 @@ import (
 	"github.com/docker/docker/daemon/logger"
 	"github.com/docker/docker/daemon/logger/templates"
 	"github.com/prometheus/common/model"
+	yaml "gopkg.in/yaml.v2"
 
 	"github.com/grafana/loki/pkg/helpers"
 	"github.com/grafana/loki/pkg/logentry/stages"
@@ -24,24 +25,25 @@ import (
 const (
 	driverName = "loki"
 
-	cfgExternalLabelsKey = "loki-external-labels"
-	cfgURLKey            = "loki-url"
-	cfgTLSCAFileKey      = "loki-tls-ca-file"
-	cfgTLSCertFileKey    = "loki-tls-cert-file"
-	cfgTLSKeyFileKey     = "loki-tls-key-file"
-	cfgTLSServerNameKey  = "loki-tls-server-name"
-	cfgTLSInsecure       = "loki-tls-insecure-skip-verify"
-	cfgProxyURLKey       = "loki-proxy-url"
-	cfgTimeoutKey        = "loki-timeout"
-	cfgBatchWaitKey      = "loki-batch-wait"
-	cfgBatchSizeKey      = "loki-batch-size"
-	cfgMinBackoffKey     = "loki-min-backoff"
-	cfgMaxBackoffKey     = "loki-max-backoff"
-	cfgMaxRetriesKey     = "loki-retries"
-	cfgPipelineStagesKey = "loki-pipeline-stage-file"
-	cfgTenantIDKey       = "loki-tenant-id"
-	cfgNofile            = "no-file"
-	cfgKeepFile          = "keep-file"
+	cfgExternalLabelsKey     = "loki-external-labels"
+	cfgURLKey                = "loki-url"
+	cfgTLSCAFileKey          = "loki-tls-ca-file"
+	cfgTLSCertFileKey        = "loki-tls-cert-file"
+	cfgTLSKeyFileKey         = "loki-tls-key-file"
+	cfgTLSServerNameKey      = "loki-tls-server-name"
+	cfgTLSInsecure           = "loki-tls-insecure-skip-verify"
+	cfgProxyURLKey           = "loki-proxy-url"
+	cfgTimeoutKey            = "loki-timeout"
+	cfgBatchWaitKey          = "loki-batch-wait"
+	cfgBatchSizeKey          = "loki-batch-size"
+	cfgMinBackoffKey         = "loki-min-backoff"
+	cfgMaxBackoffKey         = "loki-max-backoff"
+	cfgMaxRetriesKey         = "loki-retries"
+	cfgPipelineStagesFileKey = "loki-pipeline-stage-file"
+	cfgPipelineStagesKey     = "loki-pipeline-stages"
+	cfgTenantIDKey           = "loki-tenant-id"
+	cfgNofile                = "no-file"
+	cfgKeepFile              = "keep-file"
 
 	swarmServiceLabelKey = "com.docker.swarm.service.name"
 	swarmStackLabelKey   = "com.docker.stack.namespace"
@@ -102,6 +104,7 @@ func validateDriverOpt(loggerInfo logger.Info) error {
 		case cfgMaxBackoffKey:
 		case cfgMaxRetriesKey:
 		case cfgPipelineStagesKey:
+		case cfgPipelineStagesFileKey:
 		case cfgTenantIDKey:
 		case cfgNofile:
 		case cfgKeepFile:
@@ -276,21 +279,37 @@ func parseConfig(logCtx logger.Info) (*config, error) {
 	labels[targets.FilenameLabel] = model.LabelValue(logCtx.LogPath)
 
 	// parse pipeline stages
-	var pipeline PipelineConfig
-	pipelineFile, ok := logCtx.Config[cfgPipelineStagesKey]
-	if ok {
-		if err := helpers.LoadConfig(pipelineFile, &pipeline); err != nil {
-			return nil, fmt.Errorf("%s: error loading config file %s: %s", driverName, pipelineFile, err)
-		}
+	pipeline, err := parsePipeline(logCtx)
+	if err != nil {
+		return nil, err
 	}
-
 	return &config{
 		labels:       labels,
 		clientConfig: clientConfig,
 		pipeline:     pipeline,
 	}, nil
 }
 
+func parsePipeline(logCtx logger.Info) (PipelineConfig, error) {
+	var pipeline PipelineConfig
+	pipelineFile, okFile := logCtx.Config[cfgPipelineStagesFileKey]
+	pipelineString, okString := logCtx.Config[cfgPipelineStagesKey]
+	if okFile && okString {
+		return pipeline, fmt.Errorf("only one of %s or %s can be configured", cfgPipelineStagesFileKey, cfgPipelineStagesFileKey)
+	}
+	if okFile {
+		if err := helpers.LoadConfig(pipelineFile, &pipeline); err != nil {
+			return pipeline, fmt.Errorf("error loading config file %s: %s", pipelineFile, err)
+		}
+	}
+	if okString {
+		if err := yaml.UnmarshalStrict([]byte(pipelineString), &pipeline.PipelineStages); err != nil {
+			return pipeline, err
+		}
+	}
+	return pipeline, nil
+}
+
 func expandLabelValue(info logger.Info, defaultTemplate string) (string, error) {
 	tmpl, err := templates.NewParse("label_value", defaultTemplate)
 	if err != nil {

cmd/docker-driver/config_test.go

@@ -0,0 +1,82 @@
+package main
+
+import (
+	"fmt"
+	"io/ioutil"
+	"os"
+	"reflect"
+	"testing"
+
+	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/docker/docker/daemon/logger"
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/grafana/loki/pkg/logentry/stages"
+)
+
+var pipelineString = `
+- regex:
+     expression: '(level|lvl|severity)=(?P<level>\w+)'
+- labels:
+    level:
+`
+var pipeline = PipelineConfig{
+	PipelineStages: []interface{}{
+		map[interface{}]interface{}{
+			"regex": map[interface{}]interface{}{
+				"expression": "(level|lvl|severity)=(?P<level>\\w+)",
+			},
+		},
+		map[interface{}]interface{}{
+			"labels": map[interface{}]interface{}{
+				"level": nil,
+			},
+		},
+	},
+}
+
+func Test_parsePipeline(t *testing.T) {
+	f, err := ioutil.TempFile("/tmp", "Test_parsePipeline")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.Remove(f.Name())
+
+	_, err = f.Write([]byte(fmt.Sprintf("pipeline_stages:\n%s", pipelineString)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	tests := []struct {
+		name    string
+		logCtx  logger.Info
+		want    PipelineConfig
+		wantErr bool
+	}{
+		{"no config", logger.Info{Config: map[string]string{}}, PipelineConfig{}, false},
+		{"double config", logger.Info{Config: map[string]string{cfgPipelineStagesFileKey: "", cfgPipelineStagesKey: ""}}, PipelineConfig{}, true},
+		{"string config", logger.Info{Config: map[string]string{cfgPipelineStagesKey: pipelineString}}, pipeline, false},
+		{"string wrong", logger.Info{Config: map[string]string{cfgPipelineStagesKey: "pipelineString"}}, PipelineConfig{}, true},
+		{"file config", logger.Info{Config: map[string]string{cfgPipelineStagesFileKey: f.Name()}}, pipeline, false},
+		{"file wrong", logger.Info{Config: map[string]string{cfgPipelineStagesFileKey: "foo"}}, PipelineConfig{}, true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := parsePipeline(tt.logCtx)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("parsePipeline() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("parsePipeline() = %v, want %v", got, tt.want)
+			}
+
+			// all configs are supposed to be valid
+			name := "foo"
+			_, err = stages.NewPipeline(util.Logger, got.PipelineStages, &name, prometheus.DefaultRegisterer)
+			if err != nil {
+				t.Error(err)
+			}
+
+		})
+	}
+}

cmd/docker-driver/docker-compose.yaml

@@ -0,0 +1,15 @@
+version: "3"
+services:
+  nginx:
+    image: grafana/grafana
+    logging:
+      driver: loki
+      options:
+        loki-url: http://host.docker.internal:3100/loki/api/v1/push
+        loki-pipeline-stages: |
+          - regex:
+              expression: '(level|lvl|severity)=(?P<level>\w+)'
+          - labels:
+              level:
+    ports:
+      - "3000:3000"