Add LoggingConfiguration struct for providing log path to pgbackrest processes.

Add ability to send pgbackrest logs to additional volume for repo host processes.

Author: dsessler7

config/crd/bases/postgres-operator.crunchydata.com_postgresclusters.yaml

@@ -154,7 +154,7 @@ spec:
                     properties:
                       configuration:
                         description: |-
-                          Projected volumes containing custom pgBackRest configuration.  These files are mounted
+                          Projected volumes containing custom pgBackRest configuration. These files are mounted
                           under "/etc/pgbackrest/conf.d" alongside any pgBackRest configuration generated by the
                           PostgreSQL Operator:
                           https://pgbackrest.org/configuration.html
@@ -1597,6 +1597,13 @@ spec:
                                 x-kubernetes-list-type: map
                             type: object
                         type: object
+                      log:
+                        description: Logging configuration for pgbackrest processes
+                          running in postgres instance pods.
+                        properties:
+                          path:
+                            type: string
+                        type: object
                       manual:
                         description: Defines details for manual pgBackRest backup
                           Jobs
@@ -2565,6 +2572,13 @@ spec:
                                     x-kubernetes-list-type: atomic
                                 type: object
                             type: object
+                          log:
+                            description: Logging configuration for pgbackrest processes
+                              running in the repo host pod.
+                            properties:
+                              path:
+                                type: string
+                            type: object
                           priorityClassName:
                             description: |-
                               Priority class name for the pgBackRest repo host pod. Changing this value
@@ -4588,6 +4602,10 @@ spec:
                     required:
                     - repos
                     type: object
+                    x-kubernetes-validations:
+                    - message: log path is restricted to an existing additional volume
+                      rule: '!has(self.repoHost) || !has(self.repoHost.log) || !has(self.repoHost.log.path)
+                        || self.repoHost.volumes.additional.exists(x, self.repoHost.log.path.startsWith("/volumes/"+x.name))'
                   snapshots:
                     description: VolumeSnapshot configuration
                     properties:
@@ -19087,7 +19105,7 @@ spec:
                     properties:
                       configuration:
                         description: |-
-                          Projected volumes containing custom pgBackRest configuration.  These files are mounted
+                          Projected volumes containing custom pgBackRest configuration. These files are mounted
                           under "/etc/pgbackrest/conf.d" alongside any pgBackRest configuration generated by the
                           PostgreSQL Operator:
                           https://pgbackrest.org/configuration.html
@@ -20530,6 +20548,13 @@ spec:
                                 x-kubernetes-list-type: map
                             type: object
                         type: object
+                      log:
+                        description: Logging configuration for pgbackrest processes
+                          running in postgres instance pods.
+                        properties:
+                          path:
+                            type: string
+                        type: object
                       manual:
                         description: Defines details for manual pgBackRest backup
                           Jobs
@@ -21498,6 +21523,13 @@ spec:
                                     x-kubernetes-list-type: atomic
                                 type: object
                             type: object
+                          log:
+                            description: Logging configuration for pgbackrest processes
+                              running in the repo host pod.
+                            properties:
+                              path:
+                                type: string
+                            type: object
                           priorityClassName:
                             description: |-
                               Priority class name for the pgBackRest repo host pod. Changing this value

internal/collector/helpers_test.go

@@ -5,9 +5,22 @@
 package collector
 
 import (
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/crunchydata/postgres-operator/internal/initialize"
 	"github.com/crunchydata/postgres-operator/pkg/apis/postgres-operator.crunchydata.com/v1beta1"
 )
 
+var (
+	// TODO (testing): With the new RELATED_IMAGES defaulting behavior, tests could be refactored
+	// to reference those environment variables instead of hard coded image values
+	CrunchyPostgresHAImage = "registry.developers.crunchydata.com/crunchydata/crunchy-postgres:ubi8-13.6-1"
+	CrunchyPGBackRestImage = "registry.developers.crunchydata.com/crunchydata/crunchy-pgbackrest:ubi8-2.38-0"
+	CrunchyPGBouncerImage  = "registry.developers.crunchydata.com/crunchydata/crunchy-pgbouncer:ubi8-1.16-2"
+)
+
 func testInstrumentationSpec() *v1beta1.InstrumentationSpec {
 	spec := v1beta1.InstrumentationSpec{
 		Config: &v1beta1.InstrumentationConfigSpec{
@@ -30,3 +43,58 @@ func testInstrumentationSpec() *v1beta1.InstrumentationSpec {
 
 	return spec.DeepCopy()
 }
+
+// Copied from postgrescluster package
+func testVolumeClaimSpec() v1beta1.VolumeClaimSpec {
+	// Defines a volume claim spec that can be used to create instances
+	return v1beta1.VolumeClaimSpec{
+		AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+		Resources: corev1.VolumeResourceRequirements{
+			Requests: map[corev1.ResourceName]resource.Quantity{
+				corev1.ResourceStorage: resource.MustParse("1Gi"),
+			},
+		},
+	}
+}
+
+// Copied from postgrescluster package (and then edited)
+func testCluster() *v1beta1.PostgresCluster {
+	// Defines a base cluster spec that can be used by tests to generate a
+	// cluster with an expected number of instances
+	cluster := v1beta1.PostgresCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "hippo",
+		},
+		Spec: v1beta1.PostgresClusterSpec{
+			PostgresVersion: 13,
+			Image:           CrunchyPostgresHAImage,
+			ImagePullSecrets: []corev1.LocalObjectReference{{
+				Name: "myImagePullSecret"},
+			},
+			InstanceSets: []v1beta1.PostgresInstanceSetSpec{{
+				Name:                "instance1",
+				Replicas:            initialize.Int32(1),
+				DataVolumeClaimSpec: testVolumeClaimSpec(),
+			}},
+			Backups: v1beta1.Backups{
+				PGBackRest: v1beta1.PGBackRestArchive{
+					Image: CrunchyPGBackRestImage,
+					Repos: []v1beta1.PGBackRestRepo{{
+						Name: "repo1",
+						Volume: &v1beta1.RepoPVC{
+							VolumeClaimSpec: testVolumeClaimSpec(),
+						},
+					}},
+					RepoHost: &v1beta1.PGBackRestRepoHost{},
+				},
+			},
+			Proxy: &v1beta1.PostgresProxySpec{
+				PGBouncer: &v1beta1.PGBouncerPodSpec{
+					Image: CrunchyPGBouncerImage,
+				},
+			},
+			Instrumentation: testInstrumentationSpec(),
+		},
+	}
+	return cluster.DeepCopy()
+}

internal/collector/pgbackrest.go

@@ -23,17 +23,25 @@ var pgBackRestLogsTransforms json.RawMessage
 
 func NewConfigForPgBackrestRepoHostPod(
 	ctx context.Context,
-	spec *v1beta1.InstrumentationSpec,
+	cluster *v1beta1.PostgresCluster,
 	repos []v1beta1.PGBackRestRepo,
 ) *Config {
-	config := NewConfig(spec)
+	config := NewConfig(cluster.Spec.Instrumentation)
 
-	if OpenTelemetryLogsEnabled(ctx, spec) {
+	if OpenTelemetryLogsEnabled(ctx, cluster) {
 
 		var directory string
 		for _, repo := range repos {
 			if repo.Volume != nil {
-				directory = fmt.Sprintf(naming.PGBackRestRepoLogPath, repo.Name)
+				// If the user has set a log path in the spec, use it.
+				// Otherwise, default to /pgbackrest/repo#/log
+				if cluster.Spec.Backups.PGBackRest.RepoHost != nil &&
+					cluster.Spec.Backups.PGBackRest.RepoHost.Log != nil &&
+					cluster.Spec.Backups.PGBackRest.RepoHost.Log.Path != "" {
+					directory = cluster.Spec.Backups.PGBackRest.RepoHost.Log.Path
+				} else {
+					directory = fmt.Sprintf(naming.PGBackRestRepoLogPath, repo.Name)
+				}
 				break
 			}
 		}
@@ -99,8 +107,9 @@ func NewConfigForPgBackrestRepoHostPod(
 		// If there are exporters to be added to the logs pipelines defined in
 		// the spec, add them to the pipeline. Otherwise, add the DebugExporter.
 		exporters := []ComponentID{DebugExporter}
-		if spec != nil && spec.Logs != nil && spec.Logs.Exporters != nil {
-			exporters = slices.Clone(spec.Logs.Exporters)
+		if cluster.Spec.Instrumentation != nil && cluster.Spec.Instrumentation.Logs != nil &&
+			cluster.Spec.Instrumentation.Logs.Exporters != nil {
+			exporters = slices.Clone(cluster.Spec.Instrumentation.Logs.Exporters)
 		}
 
 		config.Pipelines["logs/pgbackrest"] = Pipeline{

internal/collector/pgbackrest_test.go

@@ -30,8 +30,9 @@ func TestNewConfigForPgBackrestRepoHostPod(t *testing.T) {
 		}
 		var instrumentation *v1beta1.InstrumentationSpec
 		require.UnmarshalInto(t, &instrumentation, `{}`)
-
-		config := NewConfigForPgBackrestRepoHostPod(ctx, instrumentation, repos)
+		cluster := testCluster()
+		cluster.Spec.Instrumentation = instrumentation
+		config := NewConfigForPgBackrestRepoHostPod(ctx, cluster, repos)
 
 		result, err := config.ToYAML()
 		assert.NilError(t, err)
@@ -136,8 +137,8 @@ service:
 				Volume: new(v1beta1.RepoPVC),
 			},
 		}
-
-		config := NewConfigForPgBackrestRepoHostPod(ctx, testInstrumentationSpec(), repos)
+		cluster := testCluster()
+		config := NewConfigForPgBackrestRepoHostPod(ctx, cluster, repos)
 
 		result, err := config.ToYAML()
 		assert.NilError(t, err)
@@ -231,6 +232,119 @@ service:
       - groupbyattrs/compact
       receivers:
       - filelog/pgbackrest_log
+`)
+	})
+
+	t.Run("LogPathDefined", 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),
+			},
+		}
+		cluster := testCluster()
+		cluster.Spec.Backups.PGBackRest.RepoHost.Log = &v1beta1.LoggingConfiguration{
+			Path: "/test/path",
+		}
+		config := NewConfigForPgBackrestRepoHostPod(ctx, cluster, 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
+  googlecloud:
+    log:
+      default_log_name: opentelemetry.io/collector-exported-log
+    project: google-project-name
+extensions:
+  file_storage/pgbackrest_logs:
+    create_directory: false
+    directory: /test/path/receiver
+    fsync: true
+processors:
+  batch/1s:
+    timeout: 1s
+  batch/200ms:
+    timeout: 200ms
+  batch/logs:
+    send_batch_size: 8192
+    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}
+    - action: insert
+      key: process.executable.name
+      value: pgbackrest
+  resourcedetection:
+    detectors: []
+    override: false
+    timeout: 30s
+  transform/pgbackrest_logs:
+    log_statements:
+    - statements:
+      - set(instrumentation_scope.name, "pgbackrest")
+      - set(instrumentation_scope.schema_url, "https://opentelemetry.io/schemas/1.29.0")
+      - 'merge_maps(log.cache, ExtractPatterns(log.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(log.body) > 0'
+      - set(log.severity_text, log.cache["error_severity"]) where IsString(log.cache["error_severity"])
+      - set(log.severity_number, SEVERITY_NUMBER_TRACE) where log.severity_text ==
+        "TRACE"
+      - set(log.severity_number, SEVERITY_NUMBER_DEBUG) where log.severity_text ==
+        "DEBUG"
+      - set(log.severity_number, SEVERITY_NUMBER_DEBUG2) where log.severity_text ==
+        "DETAIL"
+      - set(log.severity_number, SEVERITY_NUMBER_INFO) where log.severity_text ==
+        "INFO"
+      - set(log.severity_number, SEVERITY_NUMBER_WARN) where log.severity_text ==
+        "WARN"
+      - set(log.severity_number, SEVERITY_NUMBER_ERROR) where log.severity_text ==
+        "ERROR"
+      - set(log.time, Time(log.cache["timestamp"], "%Y-%m-%d %H:%M:%S.%L")) where
+        IsString(log.cache["timestamp"])
+      - set(log.attributes["process.pid"], log.cache["process_id"])
+      - set(log.attributes["log.record.original"], log.body)
+      - set(log.body, log.cache["message"])
+receivers:
+  filelog/pgbackrest_log:
+    include:
+    - /test/path/*.log
+    - /test/path/*.log.1
+    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:
+      - googlecloud
+      processors:
+      - resource/pgbackrest
+      - transform/pgbackrest_logs
+      - resourcedetection
+      - batch/logs
+      - groupbyattrs/compact
+      receivers:
+      - filelog/pgbackrest_log
 `)
 	})
 }