Set correct permissions on the specified log directory

internal/collector/postgres.go

@@ -19,9 +19,60 @@ import (
 	"github.com/crunchydata/postgres-operator/pkg/apis/postgres-operator.crunchydata.com/v1beta1"
 )
 
+func PostgreSQLParameters(ctx context.Context,
+	inCluster *v1beta1.PostgresCluster,
+	outParameters *postgres.Parameters,
+) {
+	version := inCluster.Spec.PostgresVersion
+
+	if OpenTelemetryLogsEnabled(ctx, inCluster) {
+		var spec *v1beta1.InstrumentationLogsSpec
+		if inCluster != nil && inCluster.Spec.Instrumentation != nil {
+			spec = inCluster.Spec.Instrumentation.Logs
+		}
+
+		// Retain logs for a short time unless specified.
+		retention := metav1.Duration{Duration: 24 * time.Hour}
+		if spec != nil && spec.RetentionPeriod != nil {
+			retention = spec.RetentionPeriod.AsDuration()
+		}
+
+		// Rotate log files according to retention and name them for the OpenTelemetry Collector.
+		//
+		// The ".log" suffix is replaced by ".csv" for CSV log files, and
+		// the ".log" suffix is replaced by ".json" for JSON log files.
+		//
+		// https://www.postgresql.org/docs/current/runtime-config-logging.html
+		for k, v := range postgres.LogRotation(retention, "postgresql-", ".log") {
+			outParameters.Mandatory.Add(k, v)
+		}
+
+		// Enable logging to file. Postgres uses a "logging collector" to safely write concurrent messages.
+		// NOTE: That collector is designed to not lose messages. When it is overloaded, other Postgres processes block.
+		//
+		// https://www.postgresql.org/docs/current/runtime-config-logging.html
+		outParameters.Mandatory.Add("logging_collector", "on")
+
+		// PostgreSQL v8.3 adds support for CSV logging, and
+		// PostgreSQL v15 adds support for JSON logging.
+		// The latter is preferred because newlines are escaped as "\n", U+005C + U+006E.
+		if version >= 15 {
+			outParameters.Mandatory.Add("log_destination", "jsonlog")
+		} else {
+			outParameters.Mandatory.Add("log_destination", "csvlog")
+		}
+
+		// Log in a timezone the OpenTelemetry Collector understands.
+		outParameters.Mandatory.Add("log_timezone", "UTC")
+
+		// TODO(logs): Remove this call and do it in [postgres.NewParameters] regardless of the gate.
+		outParameters.Mandatory.Add("log_directory", fmt.Sprintf("%s/logs/postgres", postgres.DataStorage(inCluster)))
+	}
+}
+
 func NewConfigForPostgresPod(ctx context.Context,
 	inCluster *v1beta1.PostgresCluster,
-	outParameters *postgres.ParameterSet,
+	inParameters *postgres.ParameterSet,
 ) *Config {
 	config := NewConfig(inCluster.Spec.Instrumentation)
 
@@ -30,7 +81,7 @@ func NewConfigForPostgresPod(ctx context.Context,
 	EnablePatroniMetrics(ctx, inCluster, config)
 
 	// Logging
-	EnablePostgresLogging(ctx, inCluster, config, outParameters)
+	EnablePostgresLogging(ctx, inCluster, inParameters, config)
 	EnablePatroniLogging(ctx, inCluster, config)
 
 	return config
@@ -76,51 +127,18 @@ func postgresCSVNames(version int) string {
 func EnablePostgresLogging(
 	ctx context.Context,
 	inCluster *v1beta1.PostgresCluster,
+	inParameters *postgres.ParameterSet,
 	outConfig *Config,
-	outParameters *postgres.ParameterSet,
 ) {
 	var spec *v1beta1.InstrumentationLogsSpec
 	if inCluster != nil && inCluster.Spec.Instrumentation != nil {
 		spec = inCluster.Spec.Instrumentation.Logs
 	}
 
 	if OpenTelemetryLogsEnabled(ctx, inCluster) {
-		directory := postgres.LogDirectory()
+		directory := inParameters.Value("log_directory")
 		version := inCluster.Spec.PostgresVersion
 
-		// https://www.postgresql.org/docs/current/runtime-config-logging.html
-		outParameters.Add("logging_collector", "on")
-		outParameters.Add("log_directory", directory)
-
-		// PostgreSQL v8.3 adds support for CSV logging, and
-		// PostgreSQL v15 adds support for JSON logging. The latter is preferred
-		// because newlines are escaped as "\n", U+005C + U+006E.
-		if version < 15 {
-			outParameters.Add("log_destination", "csvlog")
-		} else {
-			outParameters.Add("log_destination", "jsonlog")
-		}
-
-		// If retentionPeriod is set in the spec, use that value; otherwise, we want
-		// to use a reasonably short duration. Defaulting to 1 day.
-		retentionPeriod := metav1.Duration{Duration: 24 * time.Hour}
-		if spec != nil && spec.RetentionPeriod != nil {
-			retentionPeriod = spec.RetentionPeriod.AsDuration()
-		}
-
-		// Rotate log files according to retention.
-		//
-		// The ".log" suffix is replaced by ".csv" for CSV log files, and
-		// the ".log" suffix is replaced by ".json" for JSON log files.
-		//
-		// https://www.postgresql.org/docs/current/runtime-config-logging.html
-		for k, v := range postgres.LogRotation(retentionPeriod, "postgresql-", ".log") {
-			outParameters.Add(k, v)
-		}
-
-		// Log in a timezone that the OpenTelemetry Collector will understand.
-		outParameters.Add("log_timezone", "UTC")
-
 		// 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.
 		// TODO(log-rotation): Create this directory during Collector startup.
@@ -145,8 +163,8 @@ func EnablePostgresLogging(
 			// The 2nd through 5th fields are optional, so match through to the 7th field.
 			// This should do a decent job of not matching the middle of some SQL statement.
 			//
-			// The number of fields has changed over the years, but the first few
-			// are always formatted the same way.
+			// The number of fields has changed over the years, but the first few are always formatted the same way.
+			// [PostgreSQLParameters] ensures the timezone is UTC.
 			//
 			// NOTE: This regexp is invoked in multi-line mode. https://go.dev/s/re2syntax
 			"multiline": map[string]string{

internal/collector/postgres_test.go

@@ -31,9 +31,11 @@ func TestEnablePostgresLogging(t *testing.T) {
 		}`)
 
 		config := NewConfig(nil)
-		params := postgres.NewParameterSet()
+		params := postgres.NewParameters()
 
-		EnablePostgresLogging(ctx, cluster, config, params)
+		// NOTE: This call is necessary only because the default "log_directory" is not absolute.
+		PostgreSQLParameters(ctx, cluster, &params)
+		EnablePostgresLogging(ctx, cluster, params.Mandatory, config)
 
 		result, err := config.ToYAML()
 		assert.NilError(t, err)
@@ -293,9 +295,11 @@ service:
 		cluster.Spec.Instrumentation = testInstrumentationSpec()
 
 		config := NewConfig(cluster.Spec.Instrumentation)
-		params := postgres.NewParameterSet()
+		params := postgres.NewParameters()
 
-		EnablePostgresLogging(ctx, cluster, config, params)
+		// NOTE: This call is necessary only because the default "log_directory" is not absolute.
+		PostgreSQLParameters(ctx, cluster, &params)
+		EnablePostgresLogging(ctx, cluster, params.Mandatory, config)
 
 		result, err := config.ToYAML()
 		assert.NilError(t, err)

internal/controller/postgrescluster/controller.go

@@ -339,7 +339,7 @@ func (r *Reconciler) Reconcile(
 			ctx, cluster, clusterConfigMap, clusterReplicationSecret, rootCA,
 			clusterPodService, instanceServiceAccount, instances, patroniLeaderService,
 			primaryCertificate, clusterVolumes, exporterQueriesConfig, exporterWebConfig,
-			backupsSpecFound, otelConfig,
+			backupsSpecFound, otelConfig, pgParameters,
 		)
 	}
 

internal/controller/postgrescluster/instance.go

@@ -593,6 +593,7 @@ func (r *Reconciler) reconcileInstanceSets(
 	exporterQueriesConfig, exporterWebConfig *corev1.ConfigMap,
 	backupsSpecFound bool,
 	otelConfig *collector.Config,
+	pgParameters *postgres.ParameterSet,
 ) error {
 
 	// Go through the observed instances and check if a primary has been determined.
@@ -630,7 +631,7 @@ func (r *Reconciler) reconcileInstanceSets(
 			patroniLeaderService, primaryCertificate,
 			findAvailableInstanceNames(*set, instances, clusterVolumes),
 			numInstancePods, clusterVolumes, exporterQueriesConfig, exporterWebConfig,
-			backupsSpecFound, otelConfig,
+			backupsSpecFound, otelConfig, pgParameters,
 		)
 
 		if err == nil {
@@ -1066,6 +1067,7 @@ func (r *Reconciler) scaleUpInstances(
 	exporterQueriesConfig, exporterWebConfig *corev1.ConfigMap,
 	backupsSpecFound bool,
 	otelConfig *collector.Config,
+	pgParameters *postgres.ParameterSet,
 ) ([]*appsv1.StatefulSet, error) {
 	log := logging.FromContext(ctx)
 
@@ -1112,7 +1114,7 @@ func (r *Reconciler) scaleUpInstances(
 			rootCA, clusterPodService, instanceServiceAccount,
 			patroniLeaderService, primaryCertificate, instances[i],
 			numInstancePods, clusterVolumes, exporterQueriesConfig, exporterWebConfig,
-			backupsSpecFound, otelConfig,
+			backupsSpecFound, otelConfig, pgParameters,
 		)
 	}
 	if err == nil {
@@ -1144,6 +1146,7 @@ func (r *Reconciler) reconcileInstance(
 	exporterQueriesConfig, exporterWebConfig *corev1.ConfigMap,
 	backupsSpecFound bool,
 	otelConfig *collector.Config,
+	pgParameters *postgres.ParameterSet,
 ) error {
 	log := logging.FromContext(ctx).WithValues("instance", instance.Name)
 	ctx = logging.NewContext(ctx, log)
@@ -1187,7 +1190,7 @@ func (r *Reconciler) reconcileInstance(
 		postgres.InstancePod(
 			ctx, cluster, spec,
 			primaryCertificate, replicationCertSecretProjection(clusterReplicationSecret),
-			postgresDataVolume, postgresWALVolume, tablespaceVolumes,
+			postgresDataVolume, postgresWALVolume, tablespaceVolumes, pgParameters,
 			&instance.Spec.Template)
 
 		if backupsSpecFound {

internal/controller/postgrescluster/postgres.go

@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	"github.com/crunchydata/postgres-operator/internal/collector"
 	"github.com/crunchydata/postgres-operator/internal/feature"
 	"github.com/crunchydata/postgres-operator/internal/initialize"
 	"github.com/crunchydata/postgres-operator/internal/logging"
@@ -130,6 +131,7 @@ func (*Reconciler) generatePostgresParameters(
 	ctx context.Context, cluster *v1beta1.PostgresCluster, backupsSpecFound bool,
 ) *postgres.ParameterSet {
 	builtin := postgres.NewParameters()
+	collector.PostgreSQLParameters(ctx, cluster, &builtin)
 	pgaudit.PostgreSQLParameters(&builtin)
 	pgbackrest.PostgreSQLParameters(cluster, &builtin, backupsSpecFound)
 	pgmonitor.PostgreSQLParameters(ctx, cluster, &builtin)

internal/postgres/config.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"fmt"
 	"math"
+	"path"
 	"strings"
 
 	corev1 "k8s.io/api/core/v1"
@@ -89,13 +90,13 @@ func ConfigDirectory(cluster *v1beta1.PostgresCluster) string {
 // DataDirectory returns the absolute path to the "data_directory" of cluster.
 // - https://www.postgresql.org/docs/current/runtime-config-file-locations.html
 func DataDirectory(cluster *v1beta1.PostgresCluster) string {
-	return fmt.Sprintf("%s/pg%d", dataMountPath, cluster.Spec.PostgresVersion)
+	return fmt.Sprintf("%s/pg%d", DataStorage(cluster), cluster.Spec.PostgresVersion)
 }
 
-// LogDirectory returns the absolute path to the "log_directory" of cluster.
-// - https://www.postgresql.org/docs/current/runtime-config-logging.html
-func LogDirectory() string {
-	return fmt.Sprintf("%s/logs/postgres", dataMountPath)
+// DataStorage returns the absolute path to the disk where cluster stores its data.
+// Use [DataDirectory] for the exact directory that Postgres uses.
+func DataStorage(cluster *v1beta1.PostgresCluster) string {
+	return dataMountPath
 }
 
 // LogRotation returns parameters that rotate log files while keeping a minimum amount.
@@ -310,11 +311,33 @@ done
 // PostgreSQL.
 func startupCommand(
 	ctx context.Context,
-	cluster *v1beta1.PostgresCluster, instance *v1beta1.PostgresInstanceSetSpec,
+	cluster *v1beta1.PostgresCluster,
+	instance *v1beta1.PostgresInstanceSetSpec,
+	parameters *ParameterSet,
 ) []string {
 	version := fmt.Sprint(cluster.Spec.PostgresVersion)
+	dataDir := DataDirectory(cluster)
 	walDir := WALDirectory(cluster, instance)
 
+	// TODO: describe this?
+	mkdirs := make([]string, 0, 6)
+	mkdir := func(b, p string) {
+		if path.IsAbs(p) {
+			p = path.Clean(p)
+		} else {
+			p = path.Join(dataDir, p)
+		}
+
+		// Create directories unless they are in the empty Postgres data directory.
+		mkdirs = append(mkdirs,
+			`[[ `+shell.QuoteWord(p)+` != "${postgres_data_directory}"* || -f "${postgres_data_directory}/PG_VERSION" ]] &&`,
+			`{ (`+shell.MakeDirectories(b, p)+`) || halt "$(permissions `+shell.QuoteWord(p)+` ||:)"; }`,
+		)
+	}
+	mkdir(dataMountPath, naming.PGBackRestPGDataLogPath)
+	mkdir(dataMountPath, naming.PatroniPGDataLogPath)
+	mkdir(dataDir, parameters.Value("log_directory"))
+
 	// If the user requests tablespaces, we want to make sure the directories exist with the
 	// correct owner and permissions.
 	tablespaceCmd := ""
@@ -442,13 +465,7 @@ chmod +x /tmp/pg_rewind_tde.sh
 		`else (halt Permissions!); fi ||`,
 		`halt "$(permissions "${postgres_data_directory}" ||:)"`,
 
-		// Create log directories.
-		`(` + shell.MakeDirectories(dataMountPath, naming.PGBackRestPGDataLogPath) + `) ||`,
-		`halt "$(permissions ` + naming.PGBackRestPGDataLogPath + ` ||:)"`,
-		`(` + shell.MakeDirectories(dataMountPath, naming.PatroniPGDataLogPath) + `) ||`,
-		`halt "$(permissions ` + naming.PatroniPGDataLogPath + ` ||:)"`,
-		`(` + shell.MakeDirectories(dataMountPath, LogDirectory()) + `) ||`,
-		`halt "$(permissions ` + LogDirectory() + ` ||:)"`,
+		strings.Join(mkdirs, "\n"),
 
 		// Copy replication client certificate files
 		// from the /pgconf/tls/replication directory to the /tmp/replication directory in order

internal/postgres/config_test.go

@@ -40,6 +40,13 @@ func TestDataDirectory(t *testing.T) {
 	assert.Equal(t, DataDirectory(cluster), "/pgdata/pg12")
 }
 
+func TestDataStorage(t *testing.T) {
+	cluster := new(v1beta1.PostgresCluster)
+	cluster.Spec.PostgresVersion = rand.IntN(20)
+
+	assert.Equal(t, DataStorage(cluster), "/pgdata")
+}
+
 func TestLogRotation(t *testing.T) {
 	t.Parallel()
 
@@ -543,8 +550,10 @@ func TestStartupCommand(t *testing.T) {
 	cluster.Spec.PostgresVersion = 13
 	instance := new(v1beta1.PostgresInstanceSetSpec)
 
+	parameters := NewParameters().Default
+
 	ctx := context.Background()
-	command := startupCommand(ctx, cluster, instance)
+	command := startupCommand(ctx, cluster, instance, parameters)
 
 	// Expect a bash command with an inline script.
 	assert.DeepEqual(t, command[:3], []string{"bash", "-ceu", "--"})
@@ -579,7 +588,7 @@ func TestStartupCommand(t *testing.T) {
 				},
 			},
 		}
-		command := startupCommand(ctx, cluster, instance)
+		command := startupCommand(ctx, cluster, instance, parameters)
 		assert.Assert(t, len(command) > 3)
 		assert.Assert(t, strings.Contains(command[3], `cat << "EOF" > /tmp/pg_rewind_tde.sh
 #!/bin/sh

internal/postgres/parameters.go

@@ -48,6 +48,17 @@ func NewParameters() Parameters {
 	// - https://www.postgresql.org/docs/current/auth-password.html
 	parameters.Default.Add("password_encryption", "scram-sha-256")
 
+	// Explicitly use Postgres' default log directory. This value is relative to the "data_directory" parameter.
+	//
+	// Controllers look at this parameter to grant group-write S_IWGRP on the directory.
+	// Postgres does not grant this permission on the directories it creates.
+	//
+	// TODO(logs): A better default would be *outside* the data directory.
+	// This parameter needs to be configurable and documented before the default can change.
+	//
+	// PostgreSQL must be reloaded when changing this parameter.
+	parameters.Default.Add("log_directory", "log")
+
 	// Pod "securityContext.fsGroup" ensures processes and filesystems agree on a GID;
 	// use the same permissions for group and owner.
 	// This allows every process in the pod to read Postgres log files.

internal/postgres/parameters_test.go

@@ -28,6 +28,8 @@ func TestNewParameters(t *testing.T) {
 	assert.DeepEqual(t, parameters.Default.AsMap(), map[string]string{
 		"jit": "off",
 
+		"log_directory": "log",
+
 		"password_encryption": "scram-sha-256",
 	})
 }

internal/postgres/reconcile.go

@@ -68,6 +68,7 @@ func InstancePod(ctx context.Context,
 	inClusterCertificates, inClientCertificates *corev1.SecretProjection,
 	inDataVolume, inWALVolume *corev1.PersistentVolumeClaim,
 	inTablespaceVolumes []*corev1.PersistentVolumeClaim,
+	inParameters *ParameterSet,
 	outInstancePod *corev1.PodTemplateSpec,
 ) {
 	certVolumeMount := corev1.VolumeMount{
@@ -201,7 +202,7 @@ func InstancePod(ctx context.Context,
 	startup := corev1.Container{
 		Name: naming.ContainerPostgresStartup,
 
-		Command: startupCommand(ctx, inCluster, inInstanceSpec),
+		Command: startupCommand(ctx, inCluster, inInstanceSpec, inParameters),
 		Env:     Environment(inCluster),
 
 		Image:           container.Image,