Parse PgBouncer logs using the OTel Collector

Issue: PGO-2056

internal/collector/pgbouncer.go

@@ -12,6 +12,7 @@ import (
 	"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"
 )
 
@@ -33,11 +34,120 @@ func NewConfigForPgBouncerPod(
 
 	config := NewConfig()
 
+	EnablePgBouncerLogging(ctx, cluster, config)
 	EnablePgBouncerMetrics(ctx, config, sqlQueryUsername)
 
 	return config
 }
 
+// EnablePgBouncerLogging adds necessary configuration to the collector config to collect
+// logs from pgBouncer when the OpenTelemetryLogging feature flag is enabled.
+func EnablePgBouncerLogging(ctx context.Context,
+	inCluster *v1beta1.PostgresCluster,
+	outConfig *Config) {
+	if feature.Enabled(ctx, feature.OpenTelemetryLogs) {
+		directory := naming.PGBouncerLogPath
+
+		// 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.
+		//
+		// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/extension/storage/filestorage#readme
+		outConfig.Extensions["file_storage/pgbouncer_logs"] = map[string]any{
+			"directory":        directory + "/receiver",
+			"create_directory": true,
+			"fsync":            true,
+		}
+
+		// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/receiver/filelogreceiver#readme
+		outConfig.Receivers["filelog/pgbouncer_log"] = map[string]any{
+			// Read the log files and keep track of what has been processed.
+			"include": []string{directory + "/*.log"},
+			"storage": "file_storage/pgbouncer_logs",
+		}
+
+		// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/processor/resourceprocessor#readme
+		outConfig.Processors["resource/pgbouncer"] = 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.ContainerPGBouncer},
+				{"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/pgbouncer_logs"] = map[string]any{
+			"log_statements": []map[string]any{{
+				"context": "log",
+				"statements": []string{
+					// Set instrumentation scope
+					`set(instrumentation_scope.name, "pgbouncer")`,
+
+					// Extract timestamp, pid, log level, and message and store in cache.
+					`merge_maps(cache, ExtractPatterns(body, ` +
+						`"^(?<timestamp>\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}\\.\\d{3} [A-Z]{3}) ` +
+						`\\[(?<pid>\\d+)\\] (?<log_level>[A-Z]+) (?<msg>.*$)"), "insert")`,
+
+					// https://opentelemetry.io/docs/specs/otel/logs/data-model/#field-severitytext
+					`set(severity_text, cache["log_level"])`,
+
+					// Map pgBouncer (libusual) "logging levels" to OpenTelemetry severity levels.
+					//
+					// https://github.com/libusual/libusual/blob/master/usual/logging.c
+					// https://opentelemetry.io/docs/specs/otel/logs/data-model/#field-severitynumber
+					// https://opentelemetry.io/docs/specs/otel/logs/data-model-appendix/#appendix-b-severitynumber-example-mappings
+					// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/pkg/ottl/contexts/ottllog#enums
+					`set(severity_number, SEVERITY_NUMBER_DEBUG)  where severity_text == "NOISE" or severity_text == "DEBUG"`,
+					`set(severity_number, SEVERITY_NUMBER_INFO)   where severity_text == "LOG"`,
+					`set(severity_number, SEVERITY_NUMBER_WARN)   where severity_text == "WARNING"`,
+					`set(severity_number, SEVERITY_NUMBER_ERROR)  where severity_text == "ERROR"`,
+					`set(severity_number, SEVERITY_NUMBER_FATAL)  where severity_text == "FATAL"`,
+
+					// Parse the timestamp.
+					// The format is neither RFC 3339 nor ISO 8601:
+					//
+					// The date and time are separated by a single space U+0020,
+					// followed by a dot U+002E, milliseconds, another space U+0020,
+					// then a timezone abbreviation.
+					//
+					// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/pkg/stanza/docs/types/timestamp.md
+					`set(time, Time(cache["timestamp"], "%F %T.%L %Z"))`,
+
+					// 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 pid as attribute
+					`set(attributes["process.pid"], cache["pid"])`,
+
+					// Set the log message to body.
+					`set(body, cache["msg"])`,
+				},
+			}},
+		}
+
+		outConfig.Pipelines["logs/pgbouncer"] = Pipeline{
+			Extensions: []ComponentID{"file_storage/pgbouncer_logs"},
+			Receivers:  []ComponentID{"filelog/pgbouncer_log"},
+			Processors: []ComponentID{
+				"resource/pgbouncer",
+				"transform/pgbouncer_logs",
+				SubSecondBatchProcessor,
+				CompactingProcessor,
+			},
+			Exporters: []ComponentID{DebugExporter},
+		}
+	}
+}
+
+// EnablePgBouncerMetrics adds necessary configuration to the collector config to scrape
+// metrics from pgBouncer when the OpenTelemetryMetrics feature flag is enabled.
 func EnablePgBouncerMetrics(ctx context.Context, config *Config, sqlQueryUsername string) {
 	if feature.Enabled(ctx, feature.OpenTelemetryMetrics) {
 		// Add Prometheus exporter

internal/collector/pgbouncer_test.go

@@ -0,0 +1,98 @@
+// 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 TestEnablePgBouncerLogging(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)
+
+		config := NewConfig()
+
+		EnablePgBouncerLogging(ctx, new(v1beta1.PostgresCluster), config)
+
+		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/pgbouncer_logs:
+    create_directory: true
+    directory: /tmp/receiver
+    fsync: true
+processors:
+  batch/1s:
+    timeout: 1s
+  batch/200ms:
+    timeout: 200ms
+  groupbyattrs/compact: {}
+  resource/pgbouncer:
+    attributes:
+    - action: insert
+      key: k8s.container.name
+      value: pgbouncer
+    - action: insert
+      key: k8s.namespace.name
+      value: ${env:K8S_POD_NAMESPACE}
+    - action: insert
+      key: k8s.pod.name
+      value: ${env:K8S_POD_NAME}
+  transform/pgbouncer_logs:
+    log_statements:
+    - context: log
+      statements:
+      - set(instrumentation_scope.name, "pgbouncer")
+      - merge_maps(cache, ExtractPatterns(body, "^(?<timestamp>\\d{4}-\\d{2}-\\d{2}
+        \\d{2}:\\d{2}:\\d{2}\\.\\d{3} [A-Z]{3}) \\[(?<pid>\\d+)\\] (?<log_level>[A-Z]+)
+        (?<msg>.*$)"), "insert")
+      - set(severity_text, cache["log_level"])
+      - set(severity_number, SEVERITY_NUMBER_DEBUG)  where severity_text == "NOISE"
+        or severity_text == "DEBUG"
+      - set(severity_number, SEVERITY_NUMBER_INFO)   where severity_text == "LOG"
+      - set(severity_number, SEVERITY_NUMBER_WARN)   where severity_text == "WARNING"
+      - set(severity_number, SEVERITY_NUMBER_ERROR)  where severity_text == "ERROR"
+      - set(severity_number, SEVERITY_NUMBER_FATAL)  where severity_text == "FATAL"
+      - set(time, Time(cache["timestamp"], "%F %T.%L %Z"))
+      - set(attributes["log.record.original"], body)
+      - set(attributes["process.pid"], cache["pid"])
+      - set(body, cache["msg"])
+receivers:
+  filelog/pgbouncer_log:
+    include:
+    - /tmp/*.log
+    storage: file_storage/pgbouncer_logs
+service:
+  extensions:
+  - file_storage/pgbouncer_logs
+  pipelines:
+    logs/pgbouncer:
+      exporters:
+      - debug
+      processors:
+      - resource/pgbouncer
+      - transform/pgbouncer_logs
+      - batch/200ms
+      - groupbyattrs/compact
+      receivers:
+      - filelog/pgbouncer_log
+`)
+	})
+}

internal/controller/postgrescluster/pgbouncer.go

@@ -465,6 +465,9 @@ func (r *Reconciler) generatePGBouncerDeployment(
 		pgbouncer.Pod(ctx, cluster, configmap, primaryCertificate, secret, &deploy.Spec.Template.Spec)
 	}
 
+	// Add tmp directory and volume for log files
+	addTMPEmptyDir(&deploy.Spec.Template)
+
 	return deploy, true, err
 }
 

internal/naming/names.go

@@ -155,6 +155,9 @@ const (
 	// dedicated repo host, if configured.
 	PGBackRestRepoLogPath = "/pgbackrest/%s/log"
 
+	// PGBouncerLogPath is the pgBouncer default log path configuration
+	PGBouncerLogPath = "/tmp"
+
 	// suffix used with postgrescluster name for associated configmap.
 	// for instance, if the cluster is named 'mycluster', the
 	// configmap will be named 'mycluster-pgbackrest-config'

internal/pgbouncer/config.go

@@ -126,7 +126,12 @@ func clusterINI(ctx context.Context, cluster *v1beta1.PostgresCluster) string {
 		"unix_socket_dir": "",
 	}
 
-	// When OTel metrics are enabled, allow pgbouncer's postgres user
+	// If OpenTelemetryLogs feature is enabled, enable logging to file
+	if feature.Enabled(ctx, feature.OpenTelemetryLogs) {
+		global["logfile"] = naming.PGBouncerLogPath + "/pgbouncer.log"
+	}
+
+	// When OTel metrics are enabled, allow pgBouncer's postgres user
 	// to run read-only console queries on pgBouncer's virtual db
 	if feature.Enabled(ctx, feature.OpenTelemetryMetrics) {
 		global["stats_users"] = PostgresqlUser