Add gcp_errorreporting as a new log form

For now this ensures that on level ERROR or worse the appropriate minimal
fields are set, so the information gets scraped by Google Cloud Error
Reporting.

In turn this requires for StaticFields to allow for object values.

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/fsnotify/fsnotify v1.7.0
 	github.com/gofrs/uuid v4.4.0+incompatible
 	github.com/golang-jwt/jwt/v5 v5.2.1
+	github.com/google/go-cmp v0.6.0
 	github.com/gorilla/handlers v1.5.2
 	github.com/gorilla/mux v1.8.1
 	github.com/gorilla/websocket v1.5.3

internal/impl/pure/processor_log.go

@@ -33,7 +33,7 @@ The `+"`level`"+` field determines the log level of the printed events and can b
 
 == Structured fields
 
-It's also possible add custom fields to logs when the format is set to a structured form such as `+"`json` or `logfmt`"+` with the config field `+"<<fields_mapping, `fields_mapping`>>"+`:
+It's also possible add custom fields to logs when the format is set to a structured form such as `+"`gcp_errorreporting`, `json` or `logfmt`"+` with the config field `+"<<fields_mapping, `fields_mapping`>>"+`:
 
 `+"```yaml"+`
 pipeline:

internal/log/config.go

@@ -18,14 +18,14 @@ const (
 
 // Config holds configuration options for a logger object.
 type Config struct {
-	LogLevel      string            `yaml:"level"`
-	Format        string            `yaml:"format"`
-	AddTimeStamp  bool              `yaml:"add_timestamp"`
-	LevelName     string            `yaml:"level_name"`
-	MessageName   string            `yaml:"message_name"`
-	TimestampName string            `yaml:"timestamp_name"`
-	StaticFields  map[string]string `yaml:"static_fields"`
-	File          File              `yaml:"file"`
+	LogLevel      string         `yaml:"level"`
+	Format        string         `yaml:"format"`
+	AddTimeStamp  bool           `yaml:"add_timestamp"`
+	LevelName     string         `yaml:"level_name"`
+	MessageName   string         `yaml:"message_name"`
+	TimestampName string         `yaml:"timestamp_name"`
+	StaticFields  map[string]any `yaml:"static_fields"`
+	File          File           `yaml:"file"`
 }
 
 // File contains configuration for file based logging.
@@ -44,7 +44,7 @@ func NewConfig() Config {
 		LevelName:     "level",
 		TimestampName: "time",
 		MessageName:   "msg",
-		StaticFields: map[string]string{
+		StaticFields: map[string]any{
 			"@service": "benthos",
 		},
 	}
@@ -91,10 +91,17 @@ func FromParsed(pConf *docs.ParsedConfig) (conf Config, err error) {
 	if conf.TimestampName, err = pConf.FieldString(fieldTimestampName); err != nil {
 		return
 	}
-	if conf.StaticFields, err = pConf.FieldStringMap(fieldStaticFields); err != nil {
+
+	staticFields, err := pConf.FieldAnyMap(fieldStaticFields)
+	if err != nil {
 		return
 	}
 
+	conf.StaticFields = make(map[string]any, len(staticFields))
+	for k, v := range staticFields {
+		conf.StaticFields[k] = v.Raw()
+	}
+
 	if pConf.Contains(fieldFile) {
 		fConf := pConf.Namespace(fieldFile)
 		if conf.File.Path, err = fConf.FieldString(fieldFilePath); err != nil {

internal/log/config_test.go

@@ -0,0 +1,89 @@
+package log
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+)
+
+func TestFromParsedWithWithStaticFieldStrings(t *testing.T) {
+
+	spec := Spec()
+	pConf, err := spec.ParsedConfigFromAny(map[string]any{
+		"level":          "INFO",
+		"format":         "logfmt",
+		"add_timestamp":  false,
+		"level_name":     "level",
+		"timestamp_name": "time",
+		"message_name":   "msg",
+		"static_fields": map[string]any{
+			"@service": "benthos",
+		},
+		"file": map[string]any{
+			"path":                "",
+			"rotate":              false,
+			"rotate_max_age_days": 0,
+		},
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	config, err := FromParsed(pConf)
+	if err != nil {
+		t.Fatal("Failure in FromParsed:", err)
+	}
+	expected := Config{
+		LogLevel:      "INFO",
+		Format:        "logfmt",
+		LevelName:     "level",
+		MessageName:   "msg",
+		TimestampName: "time",
+		StaticFields: map[string]any{
+			"@service": "benthos",
+		},
+	}
+	diff := cmp.Diff(config, expected)
+	if diff != "" {
+		t.Fatalf("Unexpected parsed config (-got +want):\n%s\n", diff)
+	}
+}
+
+func TestFromParsedWithStaticFieldObjects(t *testing.T) {
+
+	spec := Spec()
+	pConf, err := spec.ParsedConfigFromAny(map[string]any{
+		"level":         "INFO",
+		"format":        "logfmt",
+		"add_timestamp": false,
+		"static_fields": map[string]any{
+			"serviceContext": map[string]any{
+				"service": "benthos",
+			},
+		},
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	config, err := FromParsed(pConf)
+	if err != nil {
+		t.Fatal("failure in FromParsed:", err)
+	}
+	expected := Config{
+		LogLevel:      "INFO",
+		Format:        "logfmt",
+		LevelName:     "level",
+		MessageName:   "msg",
+		TimestampName: "time",
+		StaticFields: map[string]any{
+			"serviceContext": map[string]any{
+				"service": string("benthos"),
+			},
+		},
+	}
+	diff := cmp.Diff(config, expected)
+	if diff != "" {
+		t.Fatalf("Unexpected parsed config (-got +want):\n%s\n", diff)
+	}
+}

internal/log/docs.go

@@ -10,12 +10,12 @@ func Spec() docs.FieldSpecs {
 		docs.FieldString(fieldLogLevel, "Set the minimum severity level for emitting logs.").HasOptions(
 			"OFF", "FATAL", "ERROR", "WARN", "INFO", "DEBUG", "TRACE", "ALL", "NONE",
 		).HasDefault("INFO").LinterFunc(nil),
-		docs.FieldString(fieldFormat, "Set the format of emitted logs.").HasOptions("json", "logfmt").HasDefault("logfmt"),
+		docs.FieldString(fieldFormat, "Set the format of emitted logs.").HasOptions("gcp_errorreporting", "json", "logfmt").HasDefault("logfmt"),
 		docs.FieldBool(fieldAddTimeStamp, "Whether to include timestamps in logs.").HasDefault(false),
 		docs.FieldString(fieldLevelName, "The name of the level field added to logs when the `format` is `json`.").HasDefault("level").Advanced(),
 		docs.FieldString(fieldTimestampName, "The name of the timestamp field added to logs when `add_timestamp` is set to `true` and the `format` is `json`.").HasDefault("time").Advanced(),
 		docs.FieldString(fieldMessageName, "The name of the message field added to logs when the `format` is `json`.").HasDefault("msg").Advanced(),
-		docs.FieldString(fieldStaticFields, "A map of key/value pairs to add to each structured log.").Map().HasDefault(map[string]any{
+		docs.FieldAnything(fieldStaticFields, "A map of key/value pairs to add to each structured log.").Map().HasDefault(map[string]any{
 			"@service": "benthos",
 		}),
 		docs.FieldObject(fieldFile, "Experimental: Specify fields for optionally writing logs to a file.").WithChildren(

internal/log/logrus.go

@@ -48,6 +48,36 @@ func New(stream io.Writer, fs ifs.FS, config Config) (Modular, error) {
 	logger.Out = stream
 
 	switch config.Format {
+	case "gcp_errorreporting":
+		logger.SetFormatter(&logrus.JSONFormatter{
+			DisableTimestamp: !config.AddTimeStamp,
+			FieldMap: logrus.FieldMap{
+				logrus.FieldKeyTime:  "eventTime",
+				logrus.FieldKeyMsg:   "message",
+				logrus.FieldKeyLevel: "level",
+			},
+		})
+		sca, ok := config.StaticFields["serviceContext"]
+		if !ok {
+			panic("missing static_field serviceContext. Google Cloud Error Reporting requires this field to be set.")
+		}
+		sc, ok := sca.(map[string]any)
+		if !ok {
+			panic("value of static_field serviceContext not an object. Google Cloud Error Reporting requires this field to be an object.")
+		}
+		srva, ok := sc["service"]
+		if !ok {
+			panic("missing static_field serviceContext.service. Google Cloud Error Reporting requires this field to be set.")
+		}
+		srv, ok := srva.(string)
+		if !ok {
+			panic("value of static_field serviceContext.service not a string. Google Cloud Error Reporting requires this field to be a string.")
+		}
+		if srv == "" {
+			panic("empty value of static_field serviceContext.service. Google Cloud Error Reporting requires this field to be set.")
+		}
+		// According to https://cloud.google.com/error-reporting/docs/formatting-error-messages#format-log-entry
+		// having a serviceContext.service and a message should be sufficient for Error Reporting to pick errors up automatically.
 	case "json":
 		logger.SetFormatter(&logrus.JSONFormatter{
 			DisableTimestamp: !config.AddTimeStamp,

internal/log/logrus_test.go

@@ -10,12 +10,44 @@ import (
 	"github.com/redpanda-data/benthos/v4/internal/filepath/ifs"
 )
 
+func TestGcpErrorReportingWithOtherNames(t *testing.T) {
+	loggerConfig := NewConfig()
+	loggerConfig.AddTimeStamp = false
+	loggerConfig.Format = "gcp_errorreporting"
+	loggerConfig.LogLevel = "ERROR"
+	loggerConfig.StaticFields = map[string]any{
+		"serviceContext": map[string]any{
+			"service": "benthos_service",
+			"version": "1.0",
+		},
+	}
+	loggerConfig.LevelName = "level"
+	loggerConfig.MessageName = "message"
+
+	var buf bytes.Buffer
+
+	logger, err := New(&buf, ifs.OS(), loggerConfig)
+	require.NoError(t, err)
+
+	logger = logger.WithFields(map[string]string{
+		"foo": "bar",
+	})
+	require.NoError(t, err)
+
+	logger.Error("Error message foo fields")
+
+	expected := `{"foo":"bar", "level":"error", "message":"Error message foo fields", "serviceContext":{"service":"benthos_service", "version":"1.0"}}
+`
+
+	require.JSONEq(t, expected, buf.String())
+}
+
 func TestLoggerWith(t *testing.T) {
 	loggerConfig := NewConfig()
 	loggerConfig.AddTimeStamp = false
 	loggerConfig.Format = "logfmt"
 	loggerConfig.LogLevel = "WARN"
-	loggerConfig.StaticFields = map[string]string{
+	loggerConfig.StaticFields = map[string]any{
 		"@service": "benthos_service",
 		"@system":  "foo",
 	}
@@ -48,7 +80,7 @@ func TestLoggerWithOddArgs(t *testing.T) {
 	loggerConfig.AddTimeStamp = false
 	loggerConfig.Format = "logfmt"
 	loggerConfig.LogLevel = "WARN"
-	loggerConfig.StaticFields = map[string]string{
+	loggerConfig.StaticFields = map[string]any{
 		"@service": "benthos_service",
 		"@system":  "foo",
 	}
@@ -76,7 +108,7 @@ func TestLoggerWithNonStringKeys(t *testing.T) {
 	loggerConfig.AddTimeStamp = false
 	loggerConfig.Format = "logfmt"
 	loggerConfig.LogLevel = "WARN"
-	loggerConfig.StaticFields = map[string]string{
+	loggerConfig.StaticFields = map[string]any{
 		"@service": "benthos_service",
 		"@system":  "foo",
 	}
@@ -106,7 +138,7 @@ func TestLoggerWithOtherNames(t *testing.T) {
 	loggerConfig.AddTimeStamp = false
 	loggerConfig.Format = "json"
 	loggerConfig.LogLevel = "WARN"
-	loggerConfig.StaticFields = map[string]string{
+	loggerConfig.StaticFields = map[string]any{
 		"@service": "benthos_service",
 		"@system":  "foo",
 	}

internal/log/tee_test.go

@@ -17,7 +17,7 @@ func TestLoggerTee(t *testing.T) {
 	loggerConfig.AddTimeStamp = false
 	loggerConfig.Format = "logfmt"
 	loggerConfig.LogLevel = "WARN"
-	loggerConfig.StaticFields = map[string]string{
+	loggerConfig.StaticFields = map[string]any{
 		"@service": "benthos_service",
 		"@system":  "foo",
 	}
@@ -29,7 +29,7 @@ func TestLoggerTee(t *testing.T) {
 	loggerConfig.AddTimeStamp = false
 	loggerConfig.Format = "logfmt"
 	loggerConfig.LogLevel = "DEBUG"
-	loggerConfig.StaticFields = map[string]string{
+	loggerConfig.StaticFields = map[string]any{
 		"@service": "benthos_service",
 		"@system":  "bar",
 	}