Add pgBackRest repohost log collector

Issue: PGO-2058

internal/collector/pgbackrest.go

@@ -0,0 +1,106 @@
+// Copyright 2024 - 2025 Crunchy Data Solutions, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package collector
+
+import (
+	"context"
+	_ "embed"
+	"encoding/json"
+	"fmt"
+	"slices"
+
+	"github.com/crunchydata/postgres-operator/internal/feature"
+	"github.com/crunchydata/postgres-operator/internal/naming"
+	"github.com/crunchydata/postgres-operator/pkg/apis/postgres-operator.crunchydata.com/v1beta1"
+)
+
+// The contents of "pgbackrest_logs_transforms.yaml" as JSON.
+// See: https://pkg.go.dev/embed
+//
+//go:embed "generated/pgbackrest_logs_transforms.json"
+var pgBackRestLogsTransforms json.RawMessage
+
+func NewConfigForPgBackrestRepoHostPod(
+	ctx context.Context,
+	repos []v1beta1.PGBackRestRepo,
+) *Config {
+	config := NewConfig()
+
+	if feature.Enabled(ctx, feature.OpenTelemetryLogs) {
+
+		var directory string
+		for _, repo := range repos {
+			if repo.Volume != nil {
+				directory = fmt.Sprintf(naming.PGBackRestRepoLogPath, repo.Name)
+				break
+			}
+		}
+
+		// We should only enter this function if a PVC is assigned for a dedicated repohost
+		// but if we don't have one, exit early.
+		if directory == "" {
+			return config
+		}
+
+		// Keep track of what log records and files have been processed.
+		// Use a subdirectory of the logs directory to stay within the same failure domain.
+		config.Extensions["file_storage/pgbackrest_logs"] = map[string]any{
+			"directory":        directory + "/receiver",
+			"create_directory": true,
+			"fsync":            true,
+		}
+
+		// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/receiver/filelogreceiver#readme
+		config.Receivers["filelog/pgbackrest_log"] = map[string]any{
+			// Read the files and keep track of what has been processed.
+			"include": []string{
+				directory + "/*.log",
+			},
+			"storage": "file_storage/pgbackrest_logs",
+			// pgBackRest prints logs with a log prefix, which includes a timestamp
+			// as long as the timestamp is not turned off in the configuration.
+			// When pgBackRest starts a process, it also will print a newline
+			// (if the file has already been written to) and a process "banner"
+			// which looks like "-------------------PROCESS START-------------------\n".
+			// Therefore we break multiline on the timestamp or the 19 dashes that start the banner.
+			// - https://github.com/pgbackrest/pgbackrest/blob/main/src/common/log.c#L451
+			"multiline": map[string]string{
+				"line_start_pattern": `^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}|^-{19}`,
+			},
+		}
+
+		config.Processors["resource/pgbackrest"] = map[string]any{
+			"attributes": []map[string]any{
+				// Container and Namespace names need no escaping because they are DNS labels.
+				// Pod names need no escaping because they are DNS subdomains.
+				//
+				// https://kubernetes.io/docs/concepts/overview/working-with-objects/names
+				// https://github.com/open-telemetry/semantic-conventions/blob/v1.29.0/docs/resource/k8s.md
+				// https://github.com/open-telemetry/semantic-conventions/blob/v1.29.0/docs/general/logs.md
+				{"action": "insert", "key": "k8s.container.name", "value": naming.PGBackRestRepoContainerName},
+				{"action": "insert", "key": "k8s.namespace.name", "value": "${env:K8S_POD_NAMESPACE}"},
+				{"action": "insert", "key": "k8s.pod.name", "value": "${env:K8S_POD_NAME}"},
+			},
+		}
+
+		// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/processor/transformprocessor#readme
+		config.Processors["transform/pgbackrest_logs"] = map[string]any{
+			"log_statements": slices.Clone(pgBackRestLogsTransforms),
+		}
+
+		config.Pipelines["logs/pgbackrest"] = Pipeline{
+			Extensions: []ComponentID{"file_storage/pgbackrest_logs"},
+			Receivers:  []ComponentID{"filelog/pgbackrest_log"},
+			Processors: []ComponentID{
+				"resource/pgbackrest",
+				"transform/pgbackrest_logs",
+				SubSecondBatchProcessor,
+				CompactingProcessor,
+			},
+			Exporters: []ComponentID{DebugExporter},
+		}
+	}
+	return config
+}

internal/collector/pgbackrest_logs_transforms.yaml

@@ -0,0 +1,43 @@
+# This list of transform statements configures an OTel Transform Processor to
+# parse pgbackrest logs.
+#
+# https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/processor/transformprocessor#readme
+
+- context: log
+  statements:
+    - set(instrumentation_scope.name, "pgbackrest")
+    - set(instrumentation_scope.schema_url, "https://opentelemetry.io/schemas/1.29.0")
+
+    # Regex the pgbackrest log to capture the following groups:
+    # 1) the timestamp (form YYYY-MM-DD HH:MM:SS.sss)
+    # 2) the process id (form `P` + 2 or 3 digits)
+    # 3) the log level (form INFO, WARN, etc.)
+    # 4) the message (anything else, including newline -- we can do this because we have a multiline block on the receiver)
+    - >-
+      merge_maps(cache,
+      ExtractPatterns(body, "^(?<timestamp>\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}\\.\\d{3}) (?<process_id>P\\d{2,3})\\s*(?<error_severity>\\S*): (?<message>(?s).*)$"),
+      "insert")
+      where Len(body) > 0
+
+    # The log severity is the "error_severity" field.
+    # https://opentelemetry.io/docs/specs/otel/logs/data-model/#field-severitytext
+    # https://pgbackrest.org/configuration.html#section-log/option-log-level-file
+    - set(severity_text, cache["error_severity"]) where IsString(cache["error_severity"])
+    - set(severity_number, SEVERITY_NUMBER_TRACE) where severity_text == "TRACE"
+    - set(severity_number, SEVERITY_NUMBER_DEBUG) where severity_text == "DEBUG"
+    - set(severity_number, SEVERITY_NUMBER_DEBUG2) where severity_text == "DETAIL"
+    - set(severity_number, SEVERITY_NUMBER_INFO) where severity_text == "INFO"
+    - set(severity_number, SEVERITY_NUMBER_WARN) where severity_text == "WARN"
+    - set(severity_number, SEVERITY_NUMBER_ERROR) where severity_text == "ERROR"
+
+    # https://opentelemetry.io/docs/specs/otel/logs/data-model/#field-timestamp
+    - set(time, Time(cache["timestamp"], "%Y-%m-%d %H:%M:%S.%L")) where IsString(cache["timestamp"])
+
+    # https://github.com/open-telemetry/semantic-conventions/blob/v1.29.0/docs/attributes-registry/process.md
+    - set(attributes["process.pid"], cache["process_id"])
+
+    # Keep the unparsed log record in a standard attribute,
+    # and replace the log record body with the message field.
+    # https://github.com/open-telemetry/semantic-conventions/blob/v1.29.0/docs/general/logs.md
+    - set(attributes["log.record.original"], body)
+    - set(body, cache["message"])

internal/collector/pgbackrest_test.go

@@ -0,0 +1,105 @@
+// Copyright 2024 - 2025 Crunchy Data Solutions, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package collector
+
+import (
+	"context"
+	"testing"
+
+	"gotest.tools/v3/assert"
+
+	"github.com/crunchydata/postgres-operator/internal/feature"
+	"github.com/crunchydata/postgres-operator/pkg/apis/postgres-operator.crunchydata.com/v1beta1"
+)
+
+func TestNewConfigForPgBackrestRepoHostPod(t *testing.T) {
+	t.Run("Enabled", func(t *testing.T) {
+		gate := feature.NewGate()
+		assert.NilError(t, gate.SetFromMap(map[string]bool{
+			feature.OpenTelemetryLogs: true,
+		}))
+		ctx := feature.NewContext(context.Background(), gate)
+		repos := []v1beta1.PGBackRestRepo{
+			{
+				Name:   "repo1",
+				Volume: new(v1beta1.RepoPVC),
+			},
+		}
+
+		config := NewConfigForPgBackrestRepoHostPod(ctx, repos)
+
+		result, err := config.ToYAML()
+		assert.NilError(t, err)
+		assert.DeepEqual(t, result, `# Generated by postgres-operator. DO NOT EDIT.
+# Your changes will not be saved.
+exporters:
+  debug:
+    verbosity: detailed
+extensions:
+  file_storage/pgbackrest_logs:
+    create_directory: true
+    directory: /pgbackrest/repo1/log/receiver
+    fsync: true
+processors:
+  batch/1s:
+    timeout: 1s
+  batch/200ms:
+    timeout: 200ms
+  groupbyattrs/compact: {}
+  resource/pgbackrest:
+    attributes:
+    - action: insert
+      key: k8s.container.name
+      value: pgbackrest
+    - action: insert
+      key: k8s.namespace.name
+      value: ${env:K8S_POD_NAMESPACE}
+    - action: insert
+      key: k8s.pod.name
+      value: ${env:K8S_POD_NAME}
+  transform/pgbackrest_logs:
+    log_statements:
+    - context: log
+      statements:
+      - set(instrumentation_scope.name, "pgbackrest")
+      - set(instrumentation_scope.schema_url, "https://opentelemetry.io/schemas/1.29.0")
+      - 'merge_maps(cache, ExtractPatterns(body, "^(?<timestamp>\\d{4}-\\d{2}-\\d{2}
+        \\d{2}:\\d{2}:\\d{2}\\.\\d{3}) (?<process_id>P\\d{2,3})\\s*(?<error_severity>\\S*):
+        (?<message>(?s).*)$"), "insert") where Len(body) > 0'
+      - set(severity_text, cache["error_severity"]) where IsString(cache["error_severity"])
+      - set(severity_number, SEVERITY_NUMBER_TRACE) where severity_text == "TRACE"
+      - set(severity_number, SEVERITY_NUMBER_DEBUG) where severity_text == "DEBUG"
+      - set(severity_number, SEVERITY_NUMBER_DEBUG2) where severity_text == "DETAIL"
+      - set(severity_number, SEVERITY_NUMBER_INFO) where severity_text == "INFO"
+      - set(severity_number, SEVERITY_NUMBER_WARN) where severity_text == "WARN"
+      - set(severity_number, SEVERITY_NUMBER_ERROR) where severity_text == "ERROR"
+      - set(time, Time(cache["timestamp"], "%Y-%m-%d %H:%M:%S.%L")) where IsString(cache["timestamp"])
+      - set(attributes["process.pid"], cache["process_id"])
+      - set(attributes["log.record.original"], body)
+      - set(body, cache["message"])
+receivers:
+  filelog/pgbackrest_log:
+    include:
+    - /pgbackrest/repo1/log/*.log
+    multiline:
+      line_start_pattern: ^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}|^-{19}
+    storage: file_storage/pgbackrest_logs
+service:
+  extensions:
+  - file_storage/pgbackrest_logs
+  pipelines:
+    logs/pgbackrest:
+      exporters:
+      - debug
+      processors:
+      - resource/pgbackrest
+      - transform/pgbackrest_logs
+      - batch/200ms
+      - groupbyattrs/compact
+      receivers:
+      - filelog/pgbackrest_log
+`)
+	})
+}

internal/collector/postgres.go

@@ -202,5 +202,59 @@ func EnablePostgresLogging(
 			},
 			Exporters: []ComponentID{DebugExporter},
 		}
+
+		// pgBackRest pipeline
+		outConfig.Extensions["file_storage/pgbackrest_logs"] = map[string]any{
+			"directory":        naming.PGBackRestPGDataLogPath + "/receiver",
+			"create_directory": true,
+			"fsync":            true,
+		}
+
+		outConfig.Receivers["filelog/pgbackrest_log"] = map[string]any{
+			"include": []string{naming.PGBackRestPGDataLogPath + "/*.log"},
+			"storage": "file_storage/pgbackrest_logs",
+
+			// pgBackRest prints logs with a log prefix, which includes a timestamp
+			// as long as the timestamp is not turned off in the configuration.
+			// When pgBackRest starts a process, it also will print a newline
+			// (if the file has already been written to) and a process "banner"
+			// which looks like "-------------------PROCESS START-------------------\n".
+			// Therefore we break multiline on the timestamp or the 19 dashes that start the banner.
+			// - https://github.com/pgbackrest/pgbackrest/blob/main/src/common/log.c#L451
+			"multiline": map[string]string{
+				"line_start_pattern": `^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}|^-{19}`,
+			},
+		}
+
+		// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/processor/resourceprocessor#readme
+		outConfig.Processors["resource/pgbackrest"] = map[string]any{
+			"attributes": []map[string]any{
+				// Container and Namespace names need no escaping because they are DNS labels.
+				// Pod names need no escaping because they are DNS subdomains.
+				//
+				// https://kubernetes.io/docs/concepts/overview/working-with-objects/names
+				// https://github.com/open-telemetry/semantic-conventions/blob/v1.29.0/docs/resource/k8s.md
+				{"action": "insert", "key": "k8s.container.name", "value": naming.ContainerDatabase},
+				{"action": "insert", "key": "k8s.namespace.name", "value": "${env:K8S_POD_NAMESPACE}"},
+				{"action": "insert", "key": "k8s.pod.name", "value": "${env:K8S_POD_NAME}"},
+			},
+		}
+
+		// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/processor/transformprocessor#readme
+		outConfig.Processors["transform/pgbackrest_logs"] = map[string]any{
+			"log_statements": slices.Clone(pgBackRestLogsTransforms),
+		}
+
+		outConfig.Pipelines["logs/pgbackrest"] = Pipeline{
+			Extensions: []ComponentID{"file_storage/pgbackrest_logs"},
+			Receivers:  []ComponentID{"filelog/pgbackrest_log"},
+			Processors: []ComponentID{
+				"resource/pgbackrest",
+				"transform/pgbackrest_logs",
+				SubSecondBatchProcessor,
+				CompactingProcessor,
+			},
+			Exporters: []ComponentID{DebugExporter},
+		}
 	}
 }