Rotate PgBouncer logs using specified retention

Defaults to 1 day when no retention is set.

Issue: PGO-2169

internal/collector/config.go

@@ -5,12 +5,24 @@
 package collector
 
 import (
+	"context"
+	_ "embed"
+	"fmt"
+	"math"
+
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/yaml"
 
 	"github.com/crunchydata/postgres-operator/pkg/apis/postgres-operator.crunchydata.com/v1beta1"
 )
 
+// The contents of "logrotate.conf" as a string.
+// See: https://pkg.go.dev/embed
+//
+//go:embed "logrotate.conf"
+var logrotateConfigFormatString string
+
 // ComponentID represents a component identifier within an OpenTelemetry
 // Collector YAML configuration. Each value is a "type" followed by an optional
 // slash-then-name: `type[/name]`
@@ -102,3 +114,61 @@ func NewConfig(spec *v1beta1.InstrumentationSpec) *Config {
 
 	return config
 }
+
+// AddLogrotateConfig generates a logrotate configuration and adds it to the
+// provided configmap
+func AddLogrotateConfig(ctx context.Context, spec *v1beta1.InstrumentationSpec,
+	outInstanceConfigMap *corev1.ConfigMap, logFilePath, postrotateScript string,
+) error {
+	var err error
+	var retentionPeriod *v1beta1.Duration
+
+	if outInstanceConfigMap.Data == nil {
+		outInstanceConfigMap.Data = make(map[string]string)
+	}
+
+	// If retentionPeriod is set in the spec, use that value; otherwise, we want
+	// to use a reasonably short duration. Defaulting to 1 day.
+	if spec != nil && spec.Logs != nil && spec.Logs.RetentionPeriod != nil {
+		retentionPeriod = spec.Logs.RetentionPeriod
+	} else {
+		retentionPeriod, err = v1beta1.NewDuration("1d")
+		if err != nil {
+			return err
+		}
+	}
+
+	outInstanceConfigMap.Data["logrotate.conf"] = generateLogrotateConfig(logFilePath,
+		retentionPeriod, postrotateScript)
+
+	return err
+}
+
+// generateLogrotateConfig generates a configuration string for logrotate based
+// on the provided full log file path, retention period, and postrotate script
+func generateLogrotateConfig(logFilePath string, retentionPeriod *v1beta1.Duration,
+	postrotateScript string,
+) string {
+	number, interval := parseDurationForLogrotate(retentionPeriod)
+
+	return fmt.Sprintf(
+		logrotateConfigFormatString,
+		logFilePath,
+		number,
+		interval,
+		postrotateScript,
+	)
+}
+
+// parseDurationForLogrotate takes a retention period and returns the rotate
+// number and interval string that should be used in the logrotate config.
+// If the retentionPeriod is less than 24 hours, the function will return the
+// number of hours and "hourly"; otherwise, we will round up to the nearest day
+// and return the day count and "daily"
+func parseDurationForLogrotate(retentionPeriod *v1beta1.Duration) (int, string) {
+	hours := math.Ceil(retentionPeriod.AsDuration().Hours())
+	if hours < 24 {
+		return int(hours), "hourly"
+	}
+	return int(math.Ceil(hours / 24)), "daily"
+}

internal/collector/config_test.go

@@ -8,6 +8,8 @@ import (
 	"testing"
 
 	"gotest.tools/v3/assert"
+
+	"github.com/crunchydata/postgres-operator/pkg/apis/postgres-operator.crunchydata.com/v1beta1"
 )
 
 func TestConfigToYAML(t *testing.T) {
@@ -61,3 +63,138 @@ service:
 `)
 	})
 }
+
+func TestGenerateLogrotateConfig(t *testing.T) {
+	for _, tt := range []struct {
+		logFilePath      string
+		retentionPeriod  string
+		postrotateScript string
+		result           string
+	}{
+		{
+			logFilePath:      "/this/is/a/file.path",
+			retentionPeriod:  "12h",
+			postrotateScript: "echo 'Hello, World'",
+			result: `/this/is/a/file.path {
+      rotate 12
+      missingok
+      sharedscripts
+      notifempty
+      nocompress
+      hourly
+      postrotate
+            echo 'Hello, World'
+      endscript
+}
+`,
+		},
+		{
+			logFilePath:      "/tmp/test.log",
+			retentionPeriod:  "5 days",
+			postrotateScript: "",
+			result: `/tmp/test.log {
+      rotate 5
+      missingok
+      sharedscripts
+      notifempty
+      nocompress
+      daily
+      postrotate
+
+      endscript
+}
+`,
+		},
+		{
+			logFilePath:      "/tmp/test.log",
+			retentionPeriod:  "5wk",
+			postrotateScript: "pkill -HUP --exact pgbouncer",
+			result: `/tmp/test.log {
+      rotate 35
+      missingok
+      sharedscripts
+      notifempty
+      nocompress
+      daily
+      postrotate
+            pkill -HUP --exact pgbouncer
+      endscript
+}
+`,
+		},
+	} {
+		t.Run(tt.retentionPeriod, func(t *testing.T) {
+			duration, err := v1beta1.NewDuration(tt.retentionPeriod)
+			assert.NilError(t, err)
+			result := generateLogrotateConfig(tt.logFilePath, duration, tt.postrotateScript)
+			assert.Equal(t, tt.result, result)
+		})
+	}
+}
+
+func TestParseDurationForLogrotate(t *testing.T) {
+	for _, tt := range []struct {
+		retentionPeriod string
+		number          int
+		interval        string
+	}{
+		{
+			retentionPeriod: "1 h 20 min",
+			number:          2,
+			interval:        "hourly",
+		},
+		{
+			retentionPeriod: "12h",
+			number:          12,
+			interval:        "hourly",
+		},
+		{
+			retentionPeriod: "24hr",
+			number:          1,
+			interval:        "daily",
+		},
+		{
+			retentionPeriod: "35hour",
+			number:          2,
+			interval:        "daily",
+		},
+		{
+			retentionPeriod: "36 hours",
+			number:          2,
+			interval:        "daily",
+		},
+		{
+			retentionPeriod: "3d",
+			number:          3,
+			interval:        "daily",
+		},
+		{
+			retentionPeriod: "365day",
+			number:          365,
+			interval:        "daily",
+		},
+		{
+			retentionPeriod: "1w",
+			number:          7,
+			interval:        "daily",
+		},
+		{
+			retentionPeriod: "4wk",
+			number:          28,
+			interval:        "daily",
+		},
+		{
+			retentionPeriod: "52week",
+			number:          364,
+			interval:        "daily",
+		},
+	} {
+		t.Run(tt.retentionPeriod, func(t *testing.T) {
+			duration, err := v1beta1.NewDuration(tt.retentionPeriod)
+			assert.NilError(t, err)
+			number, interval := parseDurationForLogrotate(duration)
+			assert.Equal(t, tt.number, number)
+			assert.Equal(t, tt.interval, interval)
+		})
+	}
+}

internal/collector/instance.go

@@ -41,11 +41,13 @@ func AddToPod(
 	outPod *corev1.PodSpec,
 	volumeMounts []corev1.VolumeMount,
 	sqlQueryPassword string,
+	includeLogrotate bool,
 ) {
 	if !(feature.Enabled(ctx, feature.OpenTelemetryLogs) || feature.Enabled(ctx, feature.OpenTelemetryMetrics)) {
 		return
 	}
 
+	// Create volume and volume mount for otel collector config
 	configVolumeMount := corev1.VolumeMount{
 		Name:      "collector-config",
 		MountPath: "/etc/otel-collector",
@@ -71,11 +73,15 @@ func AddToPod(
 		configVolume.Projected.Sources = append(configVolume.Projected.Sources, spec.Config.Files...)
 	}
 
+	// Add configVolume to the pod's volumes
+	outPod.Volumes = append(outPod.Volumes, configVolume)
+
+	// Create collector container
 	container := corev1.Container{
 		Name:            naming.ContainerCollector,
 		Image:           config.CollectorContainerImage(spec),
 		ImagePullPolicy: pullPolicy,
-		Command:         []string{"/otelcol-contrib", "--config", "/etc/otel-collector/config.yaml"},
+		Command:         startCommand(includeLogrotate),
 		Env: []corev1.EnvVar{
 			{
 				Name: "K8S_POD_NAMESPACE",
@@ -99,6 +105,32 @@ func AddToPod(
 		VolumeMounts:    append(volumeMounts, configVolumeMount),
 	}
 
+	// If this is a pod that uses logrotate for log rotation, add config volume
+	// and mount for logrotate config
+	if includeLogrotate {
+		logrotateConfigVolumeMount := corev1.VolumeMount{
+			Name:      "logrotate-config",
+			MountPath: "/etc/logrotate.d",
+			ReadOnly:  true,
+		}
+		logrotateConfigVolume := corev1.Volume{Name: logrotateConfigVolumeMount.Name}
+		logrotateConfigVolume.Projected = &corev1.ProjectedVolumeSource{
+			Sources: []corev1.VolumeProjection{{
+				ConfigMap: &corev1.ConfigMapProjection{
+					LocalObjectReference: corev1.LocalObjectReference{
+						Name: inInstanceConfigMap.Name,
+					},
+					Items: []corev1.KeyToPath{{
+						Key:  "logrotate.conf",
+						Path: "logrotate.conf",
+					}},
+				},
+			}},
+		}
+		container.VolumeMounts = append(container.VolumeMounts, logrotateConfigVolumeMount)
+		outPod.Volumes = append(outPod.Volumes, logrotateConfigVolume)
+	}
+
 	if feature.Enabled(ctx, feature.OpenTelemetryMetrics) {
 		container.Ports = []corev1.ContainerPort{{
 			ContainerPort: int32(8889),
@@ -108,5 +140,26 @@ func AddToPod(
 	}
 
 	outPod.Containers = append(outPod.Containers, container)
-	outPod.Volumes = append(outPod.Volumes, configVolume)
+}
+
+// startCommand generates the command script used by the collector container
+func startCommand(includeLogrotate bool) []string {
+	var startScript = `
+/otelcol-contrib --config /etc/otel-collector/config.yaml
+`
+
+	if includeLogrotate {
+		startScript = `
+/otelcol-contrib --config /etc/otel-collector/config.yaml &
+
+exec {fd}<> <(:||:)
+while read -r -t 5 -u "${fd}" ||:; do
+	logrotate -s /tmp/logrotate.status /etc/logrotate.d/logrotate.conf
+done
+`
+	}
+
+	wrapper := `monitor() {` + startScript + `}; export -f monitor; exec -a "$0" bash -ceu monitor`
+
+	return []string{"bash", "-ceu", "--", wrapper, "collector"}
 }

internal/collector/logrotate.conf

@@ -0,0 +1,11 @@
+%s {
+      rotate %d
+      missingok
+      sharedscripts
+      notifempty
+      nocompress
+      %s
+      postrotate
+            %s
+      endscript
+}

internal/collector/pgbouncer.go

@@ -22,6 +22,10 @@ import (
 //go:embed "generated/pgbouncer_metrics_queries.json"
 var pgBouncerMetricsQueries json.RawMessage
 
+// PGBouncerPostRotateScript is the script that is run after pgBouncer's log
+// files have been rotated. The pgbouncer process is sent a sighup signal.
+const PGBouncerPostRotateScript = "pkill -HUP --exact pgbouncer"
+
 // NewConfigForPgBouncerPod creates a config for the OTel collector container
 // that runs as a sidecar in the pgBouncer Pod
 func NewConfigForPgBouncerPod(
@@ -62,7 +66,11 @@ func EnablePgBouncerLogging(ctx context.Context,
 		// 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"},
+			// We want to watch the ".log.1" file as well as it is possible that
+			// a log entry or two will end up there after the original ".log"
+			// file is renamed to ".log.1" during rotation. OTel will not create
+			// duplicate log entries.
+			"include": []string{directory + "/*.log", directory + "/*.log.1"},
 			"storage": "file_storage/pgbouncer_logs",
 		}
 

internal/collector/pgbouncer_test.go

@@ -78,6 +78,7 @@ receivers:
   filelog/pgbouncer_log:
     include:
     - /tmp/*.log
+    - /tmp/*.log.1
     storage: file_storage/pgbouncer_logs
 service:
   extensions:
@@ -166,6 +167,7 @@ receivers:
   filelog/pgbouncer_log:
     include:
     - /tmp/*.log
+    - /tmp/*.log.1
     storage: file_storage/pgbouncer_logs
 service:
   extensions:

internal/controller/postgrescluster/instance.go

@@ -1202,8 +1202,10 @@ func (r *Reconciler) reconcileInstance(
 
 	if err == nil &&
 		(feature.Enabled(ctx, feature.OpenTelemetryLogs) || feature.Enabled(ctx, feature.OpenTelemetryMetrics)) {
+		// TODO: Setting the includeLogrotate argument to false for now. This
+		// should be changed when we implement log rotation for postgres
 		collector.AddToPod(ctx, cluster.Spec.Instrumentation, cluster.Spec.ImagePullPolicy, instanceConfigMap, &instance.Spec.Template.Spec,
-			[]corev1.VolumeMount{postgres.DataVolumeMount()}, "")
+			[]corev1.VolumeMount{postgres.DataVolumeMount()}, "", false)
 	}
 
 	// Add pgMonitor resources to the instance Pod spec
@@ -1407,6 +1409,7 @@ func (r *Reconciler) reconcileInstanceConfigMap(
 			naming.LabelInstance:    instance.Name,
 		})
 
+	// If OTel logging or metrics is enabled, add collector config
 	if err == nil && (feature.Enabled(ctx, feature.OpenTelemetryLogs) || feature.Enabled(ctx, feature.OpenTelemetryMetrics)) {
 		err = collector.AddToConfigMap(ctx, otelConfig, instanceConfigMap)
 	}