Refactor sending repo host pgbackrest logs to additional volume.

Author: dsessler7

internal/collector/helpers_test.go

@@ -5,22 +5,9 @@
 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{
@@ -43,58 +30,3 @@ 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

@@ -8,7 +8,6 @@ import (
 	"context"
 	_ "embed"
 	"encoding/json"
-	"fmt"
 	"slices"
 
 	"github.com/crunchydata/postgres-operator/internal/naming"
@@ -23,28 +22,13 @@ var pgBackRestLogsTransforms json.RawMessage
 
 func NewConfigForPgBackrestRepoHostPod(
 	ctx context.Context,
-	cluster *v1beta1.PostgresCluster,
+	spec *v1beta1.InstrumentationSpec,
 	repos []v1beta1.PGBackRestRepo,
+	directory string,
 ) *Config {
-	config := NewConfig(cluster.Spec.Instrumentation)
+	config := NewConfig(spec)
 
-	if OpenTelemetryLogsEnabled(ctx, cluster) {
-
-		var directory string
-		for _, repo := range repos {
-			if repo.Volume != nil {
-				// 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
-			}
-		}
+	if OpenTelemetryLogsEnabled(ctx, spec) {
 
 		// We should only enter this function if a PVC is assigned for a dedicated repohost
 		// but if we don't have one, exit early.
@@ -107,9 +91,8 @@ 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 cluster.Spec.Instrumentation != nil && cluster.Spec.Instrumentation.Logs != nil &&
-			cluster.Spec.Instrumentation.Logs.Exporters != nil {
-			exporters = slices.Clone(cluster.Spec.Instrumentation.Logs.Exporters)
+		if spec != nil && spec.Logs != nil && spec.Logs.Exporters != nil {
+			exporters = slices.Clone(spec.Logs.Exporters)
 		}
 
 		config.Pipelines["logs/pgbackrest"] = Pipeline{

internal/collector/pgbackrest_test.go

@@ -30,9 +30,7 @@ func TestNewConfigForPgBackrestRepoHostPod(t *testing.T) {
 		}
 		var instrumentation *v1beta1.InstrumentationSpec
 		require.UnmarshalInto(t, &instrumentation, `{}`)
-		cluster := testCluster()
-		cluster.Spec.Instrumentation = instrumentation
-		config := NewConfigForPgBackrestRepoHostPod(ctx, cluster, repos)
+		config := NewConfigForPgBackrestRepoHostPod(ctx, instrumentation, repos, "/test/directory")
 
 		result, err := config.ToYAML()
 		assert.NilError(t, err)
@@ -44,7 +42,7 @@ exporters:
 extensions:
   file_storage/pgbackrest_logs:
     create_directory: false
-    directory: /pgbackrest/repo1/log/receiver
+    directory: /test/directory/receiver
     fsync: true
 processors:
   batch/1s:
@@ -102,8 +100,8 @@ processors:
 receivers:
   filelog/pgbackrest_log:
     include:
-    - /pgbackrest/repo1/log/*.log
-    - /pgbackrest/repo1/log/*.log.1
+    - /test/directory/*.log
+    - /test/directory/*.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
@@ -137,8 +135,7 @@ service:
 				Volume: new(v1beta1.RepoPVC),
 			},
 		}
-		cluster := testCluster()
-		config := NewConfigForPgBackrestRepoHostPod(ctx, cluster, repos)
+		config := NewConfigForPgBackrestRepoHostPod(ctx, testInstrumentationSpec(), repos, "/another/directory")
 
 		result, err := config.ToYAML()
 		assert.NilError(t, err)
@@ -154,7 +151,7 @@ exporters:
 extensions:
   file_storage/pgbackrest_logs:
     create_directory: false
-    directory: /pgbackrest/repo1/log/receiver
+    directory: /another/directory/receiver
     fsync: true
 processors:
   batch/1s:
@@ -212,121 +209,8 @@ processors:
 receivers:
   filelog/pgbackrest_log:
     include:
-    - /pgbackrest/repo1/log/*.log
-    - /pgbackrest/repo1/log/*.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
-`)
-	})
-
-	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
+    - /another/directory/*.log
+    - /another/directory/*.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

internal/pgbackrest/config.go

@@ -133,31 +133,33 @@ func CreatePGBackRestConfigMapIntent(ctx context.Context, postgresCluster *v1bet
 			).String()
 
 		if collector.OpenTelemetryLogsOrMetricsEnabled(ctx, postgresCluster) {
+			// Get pgbackrest log path for repo host pod
+			var pgBackRestLogPath string
+			for _, repo := range postgresCluster.Spec.Backups.PGBackRest.Repos {
+				if repo.Volume != nil {
+					// If the user has set a log path in the spec, use it.
+					// Otherwise, default to /pgbackrest/repo#/log
+					if postgresCluster.Spec.Backups.PGBackRest.RepoHost != nil &&
+						postgresCluster.Spec.Backups.PGBackRest.RepoHost.Log != nil &&
+						postgresCluster.Spec.Backups.PGBackRest.RepoHost.Log.Path != "" {
+						pgBackRestLogPath = postgresCluster.Spec.Backups.PGBackRest.RepoHost.Log.Path
+					} else {
+						pgBackRestLogPath = fmt.Sprintf(naming.PGBackRestRepoLogPath, repo.Name)
+					}
+					break
+				}
+			}
 
 			err = collector.AddToConfigMap(ctx, collector.NewConfigForPgBackrestRepoHostPod(
 				ctx,
-				postgresCluster,
+				postgresCluster.Spec.Instrumentation,
 				postgresCluster.Spec.Backups.PGBackRest.Repos,
+				pgBackRestLogPath,
 			), cm)
 
 			// If OTel logging is enabled, add logrotate config for the RepoHost
 			if err == nil &&
 				collector.OpenTelemetryLogsEnabled(ctx, postgresCluster) {
-				var pgBackRestLogPath string
-				for _, repo := range postgresCluster.Spec.Backups.PGBackRest.Repos {
-					if repo.Volume != nil {
-						// If the user has set a log path in the spec, use it.
-						// Otherwise, default to /pgbackrest/repo#/log
-						if postgresCluster.Spec.Backups.PGBackRest.RepoHost != nil &&
-							postgresCluster.Spec.Backups.PGBackRest.RepoHost.Log != nil &&
-							postgresCluster.Spec.Backups.PGBackRest.RepoHost.Log.Path != "" {
-							pgBackRestLogPath = postgresCluster.Spec.Backups.PGBackRest.RepoHost.Log.Path
-						} else {
-							pgBackRestLogPath = fmt.Sprintf(naming.PGBackRestRepoLogPath, repo.Name)
-						}
-						break
-					}
-				}
 
 				collector.AddLogrotateConfigs(ctx, postgresCluster.Spec.Instrumentation, cm, []collector.LogrotateConfig{{
 					LogFiles: []string{pgBackRestLogPath + "/*.log"},