flow: Add 'write_to' argument to logging block (#4620)

Signed-off-by: Paschalis Tsilias <[email protected]>

CHANGELOG.md

@@ -18,6 +18,9 @@ Main (unreleased)
 
 - Clustering: Enable peer discovery with the go-discover package. (@tpaschalis)
 
+- Flow: Allow the `logging` configuration block to tee the Agent's logs to one
+  or more loki.* components. (@tpaschalis)
+
 - New Grafana Agent Flow components:
 
   - `prometheus.exporter.gcp` - scrape GCP metrics (@tburgessdev)

docs/sources/flow/reference/config-blocks/logging.md

@@ -26,6 +26,7 @@ Name | Type | Description | Default | Required
 ---- | ---- | ----------- | ------- | --------
 `level` | `string` | Level at which log lines should be written | `"info"` | no
 `format` | `string` | Format to use for writing log lines | `"logfmt"` | no
+`write_to` | `list(LogsReceiver)` | List of receivers to send log entries to | | no
 
 ### Log level
 
@@ -45,6 +46,16 @@ The following strings are recognized as valid log line formats:
 
 [logfmt]: https://brandur.org/logfmt
 
+### Log receivers
+
+The `write_to` argument allows the Agent to tee its log entries to one or more
+`loki.*` component log receivers in addition to the default [location][].
+This, for example can be the export of a `loki.write` component to ship log
+entries directly to Loki, or a `loki.relabel` component to add a certain label
+first.
+
+[location]: #log-location
+
 ## Log location
 
 Grafana Agent writes all logs to `stderr`.

pkg/flow/logging/logger.go

@@ -4,9 +4,13 @@ import (
 	"fmt"
 	"io"
 	"sync"
+	"time"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
+	"github.com/grafana/agent/component/common/loki"
+	"github.com/grafana/loki/pkg/logproto"
+	"github.com/prometheus/common/model"
 )
 
 // Logger implements the github.com/go-kit/log.Logger interface. It supports
@@ -50,12 +54,18 @@ func (l *Logger) Update(o Options) error {
 
 func buildLogger(w io.Writer, o Options) (log.Logger, error) {
 	var l log.Logger
+	var wr io.Writer
+	wr = w
+
+	if len(o.WriteTo) > 0 {
+		wr = io.MultiWriter(w, &lokiWriter{o.WriteTo})
+	}
 
 	switch o.Format {
 	case FormatLogfmt:
-		l = log.NewLogfmtLogger(log.NewSyncWriter(w))
+		l = log.NewLogfmtLogger(log.NewSyncWriter(wr))
 	case FormatJSON:
-		l = log.NewJSONLogger(log.NewSyncWriter(w))
+		l = log.NewJSONLogger(log.NewSyncWriter(wr))
 	default:
 		return nil, fmt.Errorf("unrecognized log format %q", o.Format)
 	}
@@ -65,3 +75,24 @@ func buildLogger(w io.Writer, o Options) (log.Logger, error) {
 	l = log.With(l, "ts", log.DefaultTimestampUTC)
 	return l, nil
 }
+
+type lokiWriter struct {
+	f []loki.LogsReceiver
+}
+
+func (fw *lokiWriter) Write(p []byte) (int, error) {
+	for _, receiver := range fw.f {
+		select {
+		case receiver.Chan() <- loki.Entry{
+			Labels: model.LabelSet{"component": "agent"},
+			Entry: logproto.Entry{
+				Timestamp: time.Now(),
+				Line:      string(p),
+			},
+		}:
+		default:
+			return 0, fmt.Errorf("lokiWriter failed to forward entry, channel was blocked")
+		}
+	}
+	return len(p), nil
+}

pkg/flow/logging/options.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/go-kit/log/level"
+	"github.com/grafana/agent/component/common/loki"
 	"github.com/grafana/agent/pkg/river"
 )
 
@@ -13,7 +14,7 @@ type Options struct {
 	Level  Level  `river:"level,attr,optional"`
 	Format Format `river:"format,attr,optional"`
 
-	// TODO: log sink parameter (e.g., to use the Windows Event logger)
+	WriteTo []loki.LogsReceiver `river:"write_to,attr,optional"`
 }
 
 // DefaultOptions holds defaults for creating a Logger.