feat(compression): allow configuring the Kopia compression policy

AGENTS.md

@@ -138,6 +138,7 @@ Klio-only assertions) must live outside `machinery` — e.g. under
 - These files must be kept in sync:
   - `operator/pkg/config/server.go` ↔ `core/pkg/config/server.go`
   - `operator/pkg/config/client.go` ↔ `core/pkg/config/client.go`
+  - `operator/pkg/config/compression.go` ↔ `core/pkg/config/compression.go`
 
 - When you change a metric in `core/internal/opentelemetry/catalog.go`
   (rename, add, remove, or change a metric's unit, type, or attributes),
@@ -198,6 +199,11 @@ confirm after that warning.
   those steps: tier1/tier2 retention apply, tier2 relay/migrate, tier2 policy
   set, and tier1 unpin. This is a deliberate, contained exception — not a pattern
   to copy, and one that should be removed in the future.
+- A second, narrower exception: `applyGlobalCompressionPolicy` in
+  `core/cmd/server/server.go` sets the repository-wide (global) compression
+  policy with a raw `kopia.Client{ConfigFile: ...}`, before the tier's Kopia
+  server starts. This is safe only because no server is running yet to hold a
+  stale cache. Do not reuse this pattern once the server is up.
 - A direct write that **rewrites the manifest of a live backup** MUST be followed
   by `refreshTier1KopiaServer` / `refreshTier2KopiaServer` so the servers
   reconcile their caches; skipping the refresh is a bug. The tier1 unpin is the

core/cmd/backup/run.go

@@ -20,6 +20,7 @@ SPDX-License-Identifier: Apache-2.0
 package backup
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -95,6 +96,13 @@ func runBackup(cmd *cobra.Command, _ []string) error {
 	}
 	defer kopiaClient.Close(cmd.Context())
 
+	// Overrides the repository's global policy for this cluster's source.
+	if err := setTier1CompressionPolicy(cmd.Context(), kopiaClient, &configuration); err != nil {
+		return cli.NewCodedError(
+			fmt.Errorf("while setting the tier1 compression policy: %w", err),
+			backupfailure.RepositoryError.ExitCode)
+	}
+
 	conn, err := pgx.Connect(cmd.Context(), configuration.Source.StandardDSN)
 	if err != nil {
 		return cli.NewCodedError(
@@ -139,34 +147,19 @@ func runBackup(cmd *cobra.Command, _ []string) error {
 	}
 
 	for {
-		var tier2RetentionPolicy string
-		if configuration.Tier2RetentionPolicy != nil {
-			policy := kopiaWrapper.RetentionPolicy{
-				KeepLatest:  configuration.Tier2RetentionPolicy.KeepLatest,
-				KeepHourly:  configuration.Tier2RetentionPolicy.KeepHourly,
-				KeepDaily:   configuration.Tier2RetentionPolicy.KeepDaily,
-				KeepWeekly:  configuration.Tier2RetentionPolicy.KeepWeekly,
-				KeepMonthly: configuration.Tier2RetentionPolicy.KeepMonthly,
-				KeepAnnual:  configuration.Tier2RetentionPolicy.KeepAnnual,
-			}
-
-			content, err := json.Marshal(policy)
-			if err != nil {
-				contextLogger.Error(err, "Error while serializing the tier2 retention policy, skipping")
-			} else {
-				tier2RetentionPolicy = string(content)
-			}
-		}
+		//nolint:gosec // postgres timeline is uint32 in practice, fits int32
+		timeline := int32(metadata.Timeline)
 
 		result, err := grpcClient.CloseBackup(cmd.Context(), &grpc.CloseBackupRequest{
-			ClusterName:          kopiaClient.GetHostname(),
-			BackupName:           metadata.Name,
-			Timeline:             int32(metadata.Timeline), //nolint:gosec // postgres timeline is uint32 in practice, fits int32
-			StartWal:             metadata.StartWAL,
-			EndWal:               metadata.EndWAL,
-			SegmentSize:          metadata.SegmentSize,
-			SendToTier2:          tier2,
-			Tier2RetentionPolicy: tier2RetentionPolicy,
+			ClusterName:            kopiaClient.GetHostname(),
+			BackupName:             metadata.Name,
+			Timeline:               timeline,
+			StartWal:               metadata.StartWAL,
+			EndWal:                 metadata.EndWAL,
+			SegmentSize:            metadata.SegmentSize,
+			SendToTier2:            tier2,
+			Tier2RetentionPolicy:   marshalTier2RetentionPolicy(cmd.Context(), &configuration),
+			Tier2CompressionPolicy: marshalTier2CompressionPolicy(cmd.Context(), &configuration),
 		})
 		if err != nil {
 			return cli.NewCodedError(
@@ -199,6 +192,86 @@ func runBackup(cmd *cobra.Command, _ []string) error {
 	return nil
 }
 
+// setTier1CompressionPolicy applies the per-cluster tier1 compression policy
+// to the cluster's source in the tier1 repository. It is always applied, even
+// when unconfigured, so that removing the compression section resets the
+// source back to inheriting the repository's global policy instead of
+// leaving a stale, previously-set override in place.
+func setTier1CompressionPolicy(
+	ctx context.Context,
+	client *kopia.MultiConnection,
+	configuration *config.Data,
+) error {
+	policy := toKopiaCompressionPolicy(configuration.Tier1CompressionPolicy)
+
+	target := kopiaWrapper.Target{
+		Username: client.GetUsername(),
+		Hostname: client.GetHostname(),
+	}
+
+	return client.SetCompressionPolicy(ctx, target, policy)
+}
+
+// toKopiaCompressionPolicy converts a config compression policy into the Kopia
+// wrapper representation. A nil input yields the zero policy.
+func toKopiaCompressionPolicy(p *config.CompressionPolicy) kopiaWrapper.CompressionPolicy {
+	if p == nil {
+		return kopiaWrapper.CompressionPolicy{}
+	}
+
+	return kopiaWrapper.CompressionPolicy{
+		Algorithm: p.Algorithm,
+		MinSize:   p.MinSize,
+		MaxSize:   p.MaxSize,
+	}
+}
+
+// marshalTier2RetentionPolicy serializes the tier2 retention policy to the
+// JSON representation expected by the WAL server. It returns an empty string
+// when no policy is configured or serialization fails.
+func marshalTier2RetentionPolicy(ctx context.Context, configuration *config.Data) string {
+	if configuration.Tier2RetentionPolicy == nil {
+		return ""
+	}
+
+	policy := kopiaWrapper.RetentionPolicy{
+		KeepLatest:  configuration.Tier2RetentionPolicy.KeepLatest,
+		KeepHourly:  configuration.Tier2RetentionPolicy.KeepHourly,
+		KeepDaily:   configuration.Tier2RetentionPolicy.KeepDaily,
+		KeepWeekly:  configuration.Tier2RetentionPolicy.KeepWeekly,
+		KeepMonthly: configuration.Tier2RetentionPolicy.KeepMonthly,
+		KeepAnnual:  configuration.Tier2RetentionPolicy.KeepAnnual,
+	}
+
+	content, err := json.Marshal(policy)
+	if err != nil {
+		log.FromContext(ctx).Error(err, "Error while serializing the tier2 retention policy, skipping")
+
+		return ""
+	}
+
+	return string(content)
+}
+
+// marshalTier2CompressionPolicy serializes the tier2 compression policy to the
+// JSON representation expected by the WAL server. It is always serialized,
+// even when unconfigured, so that removing the compression section resets the
+// cluster's tier2 source back to inheriting the repository's global policy
+// instead of leaving a stale, previously-set override in place. It returns an
+// empty string only when serialization fails.
+func marshalTier2CompressionPolicy(ctx context.Context, configuration *config.Data) string {
+	policy := toKopiaCompressionPolicy(configuration.Tier2CompressionPolicy)
+
+	content, err := json.Marshal(policy)
+	if err != nil {
+		log.FromContext(ctx).Error(err, "Error while serializing the tier2 compression policy, skipping")
+
+		return ""
+	}
+
+	return string(content)
+}
+
 //nolint:gochecknoinits
 func init() {
 	// Here you will define your flags and configuration settings.

core/cmd/server/server.go

@@ -28,11 +28,55 @@ import (
 	"github.com/cloudnative-pg/machinery/pkg/log"
 	"github.com/thejerf/suture/v4"
 
+	"github.com/cloudnative-pg/klio/core/internal/kopia"
 	"github.com/cloudnative-pg/klio/core/internal/server"
 	"github.com/cloudnative-pg/klio/core/internal/server/kopiaconfig"
 	"github.com/cloudnative-pg/klio/core/pkg/config"
 )
 
+// applyGlobalCompressionPolicy sets the repository-wide (global) Kopia
+// compression policy using the passed persistent config file. It is always
+// applied, even when unconfigured, so that removing the compression section
+// resets the global policy back to Kopia's built-in default instead of
+// leaving a stale, previously-set policy in place. This runs before the
+// Kopia servers start, so the direct write to the repository predates any
+// server cache.
+func applyGlobalCompressionPolicy(
+	ctx context.Context,
+	configFile string,
+	compression config.CompressionPolicy,
+) error {
+	kopiaBinary, err := kopia.LookupBinary()
+	if err != nil {
+		return err
+	}
+
+	client := &kopia.Client{
+		KopiaBinary: kopiaBinary,
+		ConfigFile:  configFile,
+	}
+
+	return client.SetKopiaGlobalCompressionPolicy(ctx, kopia.CompressionPolicy{
+		Algorithm: compression.Algorithm,
+		MinSize:   compression.MinSize,
+		MaxSize:   compression.MaxSize,
+	})
+}
+
+// setupTier1KopiaConfig connects the tier1 config file to the repository and
+// applies the tier1 repository-wide compression policy.
+func setupTier1KopiaConfig(ctx context.Context, configFile string, cfg *config.Tier1Config) error {
+	if err := kopiaconfig.CreateTier1KopiaConfigFile(ctx, configFile, cfg); err != nil {
+		return fmt.Errorf("error creating tier1 kopia config file: %w", err)
+	}
+
+	if err := applyGlobalCompressionPolicy(ctx, configFile, cfg.Compression); err != nil {
+		return fmt.Errorf("error setting tier1 global compression policy: %w", err)
+	}
+
+	return nil
+}
+
 type serverOpts struct {
 	tier1 bool
 	tier2 bool
@@ -192,12 +236,8 @@ func runServer(ctx context.Context, opts serverOpts) error {
 			}
 		}()
 
-		if err := kopiaconfig.CreateTier1KopiaConfigFile(
-			ctx,
-			tier1ConfigFileName,
-			&opts.cfg.Tier1,
-		); err != nil {
-			return fmt.Errorf("error creating tier1 kopia config file: %w", err)
+		if err := setupTier1KopiaConfig(ctx, tier1ConfigFileName, &opts.cfg.Tier1); err != nil {
+			return err
 		}
 
 		tier1 := suture.NewSimple("tier1")
@@ -215,6 +255,7 @@ func runServer(ctx context.Context, opts serverOpts) error {
 	}
 
 	// Configure tier2
+	//nolint:nestif
 	if opts.tier2 {
 		if err := opts.cfg.RequireTier2(); err != nil {
 			return fmt.Errorf("tier 2 opts.cfg validation error: %w", err)
@@ -229,6 +270,18 @@ func runServer(ctx context.Context, opts serverOpts) error {
 		tier2RWConfigFileName = tier2Configs.rwConfigFileName
 		tier2ROConfigFileName = tier2Configs.roConfigFileName
 
+		// A read-only server (tier1 disabled) never takes backups, so it must
+		// never write to the shared tier2 global policy: doing so would reset
+		// it on every restart, clobbering whatever the tier1-enabled server
+		// that actually owns backups has configured.
+		if opts.tier1 {
+			if err := applyGlobalCompressionPolicy(
+				ctx, tier2RWConfigFileName, opts.cfg.Tier2.Compression,
+			); err != nil {
+				return fmt.Errorf("error setting tier2 global compression policy: %w", err)
+			}
+		}
+
 		tier2 := suture.NewSimple("tier2")
 		tier2.Add(&server.Tier2KopiaServer{
 			Config:      opts.cfg,

core/internal/client/klioclient/interfaces.go

@@ -86,6 +86,9 @@ type Client interface {
 	// SetRetentionPolicy sets the retention policy for backups of this cluster.
 	SetRetentionPolicy(ctx context.Context, t kopia.Target, p kopia.RetentionPolicy) error
 
+	// SetCompressionPolicy sets the compression policy for backups of this cluster.
+	SetCompressionPolicy(ctx context.Context, t kopia.Target, policy kopia.CompressionPolicy) error
+
 	// GetRetentionPolicy gets the currently applied retention policy for this cluster.
 	GetRetentionPolicy(ctx context.Context, t kopia.Target) (*kopia.RetentionPolicy, error)
 

core/internal/client/klioclient/kopia/multiconnect.go

@@ -124,6 +124,19 @@ func (s *MultiConnection) SetRetentionPolicy(
 	return s.Tier1.SetRetentionPolicy(ctx, t, p)
 }
 
+// SetCompressionPolicy implements the Client interface.
+func (s *MultiConnection) SetCompressionPolicy(
+	ctx context.Context,
+	t kopia.Target,
+	policy kopia.CompressionPolicy,
+) error {
+	if s.Tier1 == nil {
+		return ErrUnsupportedWriteOperation
+	}
+
+	return s.Tier1.SetCompressionPolicy(ctx, t, policy)
+}
+
 // GetRetentionPolicy implements the Client interface.
 func (s *MultiConnection) GetRetentionPolicy(
 	ctx context.Context,

core/internal/client/klioclient/kopia/retention.go

@@ -30,6 +30,11 @@ func (s *Connection) SetRetentionPolicy(ctx context.Context, t kopia.Target, p k
 	return s.kopia.SetKopiaPolicy(ctx, t, &p)
 }
 
+// SetCompressionPolicy sets the compression policy for backups of this cluster.
+func (s *Connection) SetCompressionPolicy(ctx context.Context, t kopia.Target, policy kopia.CompressionPolicy) error {
+	return s.kopia.SetKopiaCompressionPolicy(ctx, t, policy)
+}
+
 // GetRetentionPolicy gets the currently applied retention policy for this cluster.
 func (s *Connection) GetRetentionPolicy(ctx context.Context, t kopia.Target) (*kopia.RetentionPolicy, error) {
 	policy, err := s.kopia.GetCurrentKopiaPolicy(ctx, t)

core/internal/consumer/backup.go

@@ -236,6 +236,24 @@ func (d *Backup) relayAndMaintain(ctx context.Context, task *queue.BackupTask, e
 func (d *Backup) relayTier2(ctx context.Context, task *queue.BackupTask, entries []kopia.Manifest) error {
 	sources := manifestListToDescriptors(entries)
 
+	// Set the per-cluster tier2 compression policy before migrating so that
+	// the data relayed to tier2 is compressed. This overrides the tier2
+	// repository global policy for this cluster's source. Applied even when
+	// the policy is zero, so removing the compression section resets the
+	// source back to inheriting the global policy instead of leaving a stale
+	// override in place. A direct write is unavoidable here (the consumer has
+	// no tier2 server connection) and is safe: it only writes a policy
+	// manifest.
+	if p := task.Tier2CompressionPolicy; p != nil && len(entries) > 0 {
+		target := kopia.Target{
+			Username: entries[0].Source.UserName,
+			Hostname: task.ClusterName,
+		}
+		if err := d.tier2Kopia.SetKopiaCompressionPolicy(ctx, target, *p); err != nil {
+			return fmt.Errorf("while setting the tier2 compression policy: %w", err)
+		}
+	}
+
 	if err := d.tier2Kopia.MigrateSnapshots(ctx, kopia.SnapshotMigrateOpts{
 		SourceConfig: d.opts.Tier1KopiaConfig,
 		Sources:      sources,