diff --git a/core/cmd/backup/run.go b/core/cmd/backup/run.go
index d12916e3..4c11ef43 100644
--- a/core/cmd/backup/run.go
+++ b/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,15 @@ func runBackup(cmd *cobra.Command, _ []string) error {
}
defer kopiaClient.Close(cmd.Context())
+ // Set the per-cluster tier1 compression policy before uploading so that
+ // the base backup snapshot is compressed. This overrides the repository
+ // 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 +149,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 +194,86 @@ func runBackup(cmd *cobra.Command, _ []string) error {
return nil
}
+// setTier1CompressionPolicy applies the per-cluster tier1 compression policy,
+// if configured, to the cluster's source in the tier1 repository.
+func setTier1CompressionPolicy(
+ ctx context.Context,
+ client *kopia.MultiConnection,
+ configuration *config.Data,
+) error {
+ policy := toKopiaCompressionPolicy(configuration.Tier1CompressionPolicy)
+ if policy.IsZero() {
+ return nil
+ }
+
+ 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 returns an empty string
+// when no policy is configured or serialization fails.
+func marshalTier2CompressionPolicy(ctx context.Context, configuration *config.Data) string {
+ policy := toKopiaCompressionPolicy(configuration.Tier2CompressionPolicy)
+ if policy.IsZero() {
+ return ""
+ }
+
+ 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.
diff --git a/core/cmd/server/server.go b/core/cmd/server/server.go
index dca7fa20..f10ce430 100644
--- a/core/cmd/server/server.go
+++ b/core/cmd/server/server.go
@@ -28,11 +28,56 @@ 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 a no-op
+// when the policy carries no settings. 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.CompressionServerConfig,
+) error {
+ if compression.IsZero() {
+ return nil
+ }
+
+ 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 +237,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")
@@ -229,6 +270,12 @@ func runServer(ctx context.Context, opts serverOpts) error {
tier2RWConfigFileName = tier2Configs.rwConfigFileName
tier2ROConfigFileName = tier2Configs.roConfigFileName
+ 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,
diff --git a/core/internal/client/klioclient/interfaces.go b/core/internal/client/klioclient/interfaces.go
index 6c4642f4..26514056 100644
--- a/core/internal/client/klioclient/interfaces.go
+++ b/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)
diff --git a/core/internal/client/klioclient/kopia/multiconnect.go b/core/internal/client/klioclient/kopia/multiconnect.go
index c1a6ad33..86d73eca 100644
--- a/core/internal/client/klioclient/kopia/multiconnect.go
+++ b/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,
diff --git a/core/internal/client/klioclient/kopia/retention.go b/core/internal/client/klioclient/kopia/retention.go
index f7653280..e4428f55 100644
--- a/core/internal/client/klioclient/kopia/retention.go
+++ b/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)
diff --git a/core/internal/consumer/backup.go b/core/internal/consumer/backup.go
index b338d654..d4e54b5e 100644
--- a/core/internal/consumer/backup.go
+++ b/core/internal/consumer/backup.go
@@ -236,6 +236,21 @@ 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. A direct write is
+ // unavoidable here (the consumer has no tier2 server connection) and is
+ // safe: it only writes a policy manifest, never rewrites a live backup.
+ if p := task.Tier2CompressionPolicy; p != nil && !p.IsZero() && 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,
diff --git a/core/internal/grpc/klio_wal.pb.go b/core/internal/grpc/klio_wal.pb.go
index 207d602c..9cc51d6d 100644
--- a/core/internal/grpc/klio_wal.pb.go
+++ b/core/internal/grpc/klio_wal.pb.go
@@ -728,8 +728,10 @@ type CloseBackupRequest struct {
SendToTier2 bool `protobuf:"varint,8,opt,name=send_to_tier2,json=sendToTier2,proto3" json:"send_to_tier2,omitempty"`
// When present, set the tier2 retention policy to the specified JSON-serialized policy.
Tier2RetentionPolicy string `protobuf:"bytes,9,opt,name=tier2_retention_policy,json=tier2RetentionPolicy,proto3" json:"tier2_retention_policy,omitempty"`
- unknownFields protoimpl.UnknownFields
- sizeCache protoimpl.SizeCache
+ // When present, set the tier2 compression policy to the specified JSON-serialized policy.
+ Tier2CompressionPolicy string `protobuf:"bytes,10,opt,name=tier2_compression_policy,json=tier2CompressionPolicy,proto3" json:"tier2_compression_policy,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
}
func (x *CloseBackupRequest) Reset() {
@@ -818,6 +820,13 @@ func (x *CloseBackupRequest) GetTier2RetentionPolicy() string {
return ""
}
+func (x *CloseBackupRequest) GetTier2CompressionPolicy() string {
+ if x != nil {
+ return x.Tier2CompressionPolicy
+ }
+ return ""
+}
+
// This is sent by the WAL server in response to a CloseBackupRequest
// message.
type CloseBackupResult struct {
@@ -921,7 +930,7 @@ const file_proto_klio_wal_proto_rawDesc = "" +
"\fStartWALFile\x12!\n" +
"\fklio_version\x18\x01 \x01(\x04R\vklioVersion\x12\x1f\n" +
"\vfile_length\x18\x02 \x01(\x04R\n" +
- "fileLength\"\xa7\x02\n" +
+ "fileLength\"\xe1\x02\n" +
"\x12CloseBackupRequest\x12!\n" +
"\fcluster_name\x18\x01 \x01(\tR\vclusterName\x12\x1f\n" +
"\vbackup_name\x18\x03 \x01(\tR\n" +
@@ -931,7 +940,9 @@ const file_proto_klio_wal_proto_rawDesc = "" +
"\aend_wal\x18\x06 \x01(\tR\x06endWal\x12!\n" +
"\fsegment_size\x18\a \x01(\x04R\vsegmentSize\x12\"\n" +
"\rsend_to_tier2\x18\b \x01(\bR\vsendToTier2\x124\n" +
- "\x16tier2_retention_policy\x18\t \x01(\tR\x14tier2RetentionPolicy\"f\n" +
+ "\x16tier2_retention_policy\x18\t \x01(\tR\x14tier2RetentionPolicy\x128\n" +
+ "\x18tier2_compression_policy\x18\n" +
+ " \x01(\tR\x16tier2CompressionPolicy\"f\n" +
"\x11CloseBackupResult\x12%\n" +
"\x0etier2_schedule\x18\x01 \x01(\bR\rtier2Schedule\x12*\n" +
"\x11missing_wal_files\x18\x02 \x03(\tR\x0fmissingWalFiles2\xd8\x03\n" +
diff --git a/core/internal/kopia/data.go b/core/internal/kopia/data.go
index 3915c69e..eaa2031b 100644
--- a/core/internal/kopia/data.go
+++ b/core/internal/kopia/data.go
@@ -162,6 +162,26 @@ type RetentionPolicy struct {
KeepAnnual *int `json:"keepAnnual,omitempty"`
}
+// CompressionPolicy describes the compression policy for a source.
+type CompressionPolicy struct {
+ // Algorithm is the name of the Kopia compression algorithm to use.
+ // The special value "none" disables compression.
+ Algorithm string `json:"compressionAlgorithm,omitempty"`
+
+ // MinSize is the minimum file size, in bytes, to attempt compression for.
+ // Files smaller than this are stored uncompressed. Zero means no minimum.
+ MinSize int64 `json:"compressionMinSize,omitempty"`
+
+ // MaxSize is the maximum file size, in bytes, to attempt compression for.
+ // Files larger than this are stored uncompressed. Zero means no maximum.
+ MaxSize int64 `json:"compressionMaxSize,omitempty"`
+}
+
+// IsZero reports whether the policy carries no compression settings.
+func (p CompressionPolicy) IsZero() bool {
+ return p.Algorithm == "" && p.MinSize == 0 && p.MaxSize == 0
+}
+
// Target is used to point a Kopia transaction to the set of snapshots
// having the specified Hostname and Username.
type Target struct {
diff --git a/core/internal/kopia/policy.go b/core/internal/kopia/policy.go
index 43998dc6..f2c8ebd7 100644
--- a/core/internal/kopia/policy.go
+++ b/core/internal/kopia/policy.go
@@ -107,6 +107,80 @@ func (s *Client) SetKopiaPolicy(
return nil
}
+// SetKopiaCompressionPolicy sets the compression policy for a source. This
+// overrides the repository-wide global policy for that source.
+func (s *Client) SetKopiaCompressionPolicy(
+ ctx context.Context,
+ t Target,
+ policy CompressionPolicy,
+) error {
+ return s.setKopiaCompressionPolicy(ctx, t.String(), policy)
+}
+
+// SetKopiaGlobalCompressionPolicy sets the repository-wide (global)
+// compression policy, which applies to every source that does not define its
+// own compression policy.
+func (s *Client) SetKopiaGlobalCompressionPolicy(
+ ctx context.Context,
+ policy CompressionPolicy,
+) error {
+ return s.setKopiaCompressionPolicy(ctx, "--global", policy)
+}
+
+// setKopiaCompressionPolicy runs `kopia policy set` with the compression flags
+// against the passed policy target (a "user@host" source or the "--global"
+// selector).
+func (s *Client) setKopiaCompressionPolicy(
+ ctx context.Context,
+ policyTarget string,
+ policy CompressionPolicy,
+) error {
+ contextLogger := log.FromContext(ctx)
+
+ args := buildCompressionPolicyArgs(s.ConfigFile, policyTarget, policy)
+
+ contextLogger.Info("Setting Kopia compression policy", "args", args)
+
+ setPolicyCmd := exec.CommandContext(ctx, s.KopiaBinary, args...) //nolint:gosec
+ setPolicyCmd.Env = s.kopiaEnvironmentVariables()
+
+ if err := RunWithLogCapture(ctx, setPolicyCmd, nil); err != nil {
+ return fmt.Errorf("error while setting Kopia compression policy: %w", err)
+ }
+
+ return nil
+}
+
+// buildCompressionPolicyArgs builds the argument list for the
+// `kopia policy set` compression command. policyTarget is either a
+// "user@host" source or the "--global" selector. Only the flags for the
+// fields that are set are emitted, so an unset field leaves the corresponding
+// Kopia policy value untouched.
+func buildCompressionPolicyArgs(configFile, policyTarget string, policy CompressionPolicy) []string {
+ args := []string{
+ "policy",
+ "set",
+ "--config-file=" + configFile,
+ // Kopia's own on-disk log files are suppressed on every invocation:
+ // Klio captures the subprocess output via RunWithLogCapture and
+ // re-emits it as structured logs, so the files would only accumulate
+ // redundantly on the cache volume.
+ "--disable-file-logging",
+ }
+
+ if policy.Algorithm != "" {
+ args = append(args, "--compression="+policy.Algorithm)
+ }
+ if policy.MinSize > 0 {
+ args = append(args, "--compression-min-size="+strconv.FormatInt(policy.MinSize, 10))
+ }
+ if policy.MaxSize > 0 {
+ args = append(args, "--compression-max-size="+strconv.FormatInt(policy.MaxSize, 10))
+ }
+
+ return append(args, policyTarget)
+}
+
// ApplyKopiaPolicy applies the retention policy by expiring old snapshots.
func (s *Client) ApplyKopiaPolicy(ctx context.Context, t Target) error {
contextLogger := log.FromContext(ctx)
diff --git a/core/internal/kopia/policy_test.go b/core/internal/kopia/policy_test.go
new file mode 100644
index 00000000..8ab623b2
--- /dev/null
+++ b/core/internal/kopia/policy_test.go
@@ -0,0 +1,90 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package kopia
+
+import (
+ "slices"
+ "strings"
+ "testing"
+)
+
+// assertArgContains fails the test when args does not contain want.
+func assertArgContains(t *testing.T, args []string, want string) {
+ t.Helper()
+ if !slices.Contains(args, want) {
+ t.Errorf("expected args to contain %q, got %v", want, args)
+ }
+}
+
+// assertNoArgWithPrefix fails the test when any arg starts with prefix.
+func assertNoArgWithPrefix(t *testing.T, args []string, prefix string) {
+ t.Helper()
+ for _, a := range args {
+ if strings.HasPrefix(a, prefix) {
+ t.Errorf("expected no arg with prefix %q, got %v", prefix, args)
+ }
+ }
+}
+
+func TestBuildCompressionPolicyArgs(t *testing.T) {
+ t.Run("per-source target", func(t *testing.T) {
+ target := Target{Username: "user", Hostname: "cluster"}
+ args := buildCompressionPolicyArgs("/etc/kopia/config", target.String(),
+ CompressionPolicy{Algorithm: "zstd"})
+
+ assertArgContains(t, args, "--compression=zstd")
+ assertArgContains(t, args, "--config-file=/etc/kopia/config")
+ assertArgContains(t, args, "user@cluster")
+ assertNoArgWithPrefix(t, args, "--global")
+ })
+
+ t.Run("global target", func(t *testing.T) {
+ args := buildCompressionPolicyArgs("/etc/kopia/config", "--global",
+ CompressionPolicy{Algorithm: "s2-default"})
+
+ assertArgContains(t, args, "--compression=s2-default")
+ assertArgContains(t, args, "--global")
+ })
+
+ t.Run("min and max size flags", func(t *testing.T) {
+ args := buildCompressionPolicyArgs("/etc/kopia/config", "--global",
+ CompressionPolicy{Algorithm: "zstd", MinSize: 4096, MaxSize: 1048576})
+
+ assertArgContains(t, args, "--compression-min-size=4096")
+ assertArgContains(t, args, "--compression-max-size=1048576")
+ })
+
+ t.Run("unset fields emit no flag", func(t *testing.T) {
+ args := buildCompressionPolicyArgs("/etc/kopia/config", "--global",
+ CompressionPolicy{MinSize: 4096})
+
+ assertNoArgWithPrefix(t, args, "--compression=")
+ assertNoArgWithPrefix(t, args, "--compression-max-size=")
+ assertArgContains(t, args, "--compression-min-size=4096")
+ })
+
+ t.Run("last argument is the target", func(t *testing.T) {
+ args := buildCompressionPolicyArgs("/etc/kopia/config", "--global",
+ CompressionPolicy{Algorithm: "none"})
+ if got := args[len(args)-1]; got != "--global" {
+ t.Errorf("expected the target to be the last argument, got %q in %v", got, args)
+ }
+ })
+}
diff --git a/core/internal/queue/backup.go b/core/internal/queue/backup.go
index 49688bda..955030b9 100644
--- a/core/internal/queue/backup.go
+++ b/core/internal/queue/backup.go
@@ -41,6 +41,9 @@ type BackupTask struct {
// The retention policy to apply to tier2.
Tier2RetentionPolicy *kopia.RetentionPolicy `json:"tier2RetentionPolicy,omitzero"`
+
+ // The compression policy to apply to tier2.
+ Tier2CompressionPolicy *kopia.CompressionPolicy `json:"tier2CompressionPolicy,omitzero"`
}
// Cluster returns the name of the cluster associated with this task.
diff --git a/core/internal/server/walserver/backup.go b/core/internal/server/walserver/backup.go
index 76645ae1..fefe4d85 100644
--- a/core/internal/server/walserver/backup.go
+++ b/core/internal/server/walserver/backup.go
@@ -91,10 +91,21 @@ func (w *Implementation) scheduleBackupRelay(ctx context.Context, request *grpc.
}
}
+ var tier2Compression *kopia.CompressionPolicy
+ if request.GetTier2CompressionPolicy() != "" {
+ var policy kopia.CompressionPolicy
+ if err := json.Unmarshal([]byte(request.GetTier2CompressionPolicy()), &policy); err != nil {
+ contextLogger.Error(err, "Unable to unmarshal tier2 compression policy, skipping")
+ } else {
+ tier2Compression = &policy
+ }
+ }
+
if err := w.queue.NotifyBackupReceived(ctx, &queue.BackupTask{
- ClusterName: request.GetClusterName(),
- SendToTier2: request.GetSendToTier2(),
- Tier2RetentionPolicy: tier2Policy,
+ ClusterName: request.GetClusterName(),
+ SendToTier2: request.GetSendToTier2(),
+ Tier2RetentionPolicy: tier2Policy,
+ Tier2CompressionPolicy: tier2Compression,
}); err != nil {
return fmt.Errorf("while sending task to queue: %w", err)
}
diff --git a/core/pkg/config/client.go b/core/pkg/config/client.go
index a6726d59..cbca13c0 100644
--- a/core/pkg/config/client.go
+++ b/core/pkg/config/client.go
@@ -38,6 +38,12 @@ type Data struct {
// Tier2RetentionPolicy is the retention policy to be applied to tier2.
Tier2RetentionPolicy *RetentionPolicy `json:"tier2_retention,omitempty" mapstructure:"tier2_retention"`
+ // Tier1CompressionPolicy is the compression policy to be applied to tier1.
+ Tier1CompressionPolicy *CompressionPolicy `json:"tier1_compression,omitempty" mapstructure:"tier1_compression"`
+
+ // Tier2CompressionPolicy is the compression policy to be applied to tier2.
+ Tier2CompressionPolicy *CompressionPolicy `json:"tier2_compression,omitempty" mapstructure:"tier2_compression"`
+
// Tier1Enabled records whether the client archives base backups and WAL
// to tier1. False on read-only clients, which do not write to tier1 but
// may still restore from it when Client.Wal.Address is set. Not consulted
diff --git a/core/pkg/config/client_validate.go b/core/pkg/config/client_validate.go
index 1690d9e1..202c7b54 100644
--- a/core/pkg/config/client_validate.go
+++ b/core/pkg/config/client_validate.go
@@ -38,6 +38,14 @@ func (d *Data) Validate() error {
errs = errors.Join(errs, err)
}
+ if err := d.Tier1CompressionPolicy.Validate(); err != nil {
+ errs = errors.Join(errs, err)
+ }
+
+ if err := d.Tier2CompressionPolicy.Validate(); err != nil {
+ errs = errors.Join(errs, err)
+ }
+
return errs
}
diff --git a/core/pkg/config/compression.go b/core/pkg/config/compression.go
new file mode 100644
index 00000000..62119c55
--- /dev/null
+++ b/core/pkg/config/compression.go
@@ -0,0 +1,90 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package config
+
+import (
+ "errors"
+ "fmt"
+)
+
+// CompressionPolicy configures the Kopia compression policy applied to base
+// backup data.
+type CompressionPolicy struct {
+ // Algorithm is the name of the Kopia compression algorithm to use.
+ // The special value "none" disables compression.
+ Algorithm string `json:"algorithm,omitempty" mapstructure:"algorithm"`
+
+ // MinSize is the minimum file size, in bytes, to attempt compression for.
+ // Files smaller than this are stored uncompressed. Zero means no minimum.
+ MinSize int64 `json:"min_size,omitempty" mapstructure:"min_size"`
+
+ // MaxSize is the maximum file size, in bytes, to attempt compression for.
+ // Files larger than this are stored uncompressed. Zero means no maximum.
+ MaxSize int64 `json:"max_size,omitempty" mapstructure:"max_size"`
+}
+
+// ErrInvalidCompressionAlgorithm is returned when an unsupported compression
+// algorithm is configured.
+var ErrInvalidCompressionAlgorithm = errors.New("invalid compression algorithm")
+
+// IsValidCompressionAlgorithm returns true when the passed algorithm is a
+// supported Kopia compression algorithm. The list matches the compressors
+// registered by Kopia, plus the special value "none" that explicitly disables
+// compression. It must be kept in sync with the CompressionAlgorithm enum in
+// the operator CRD types.
+func IsValidCompressionAlgorithm(algorithm string) bool {
+ switch algorithm {
+ case "none",
+ "deflate-best-compression",
+ "deflate-best-speed",
+ "deflate-default",
+ "gzip",
+ "gzip-best-compression",
+ "gzip-best-speed",
+ "pgzip",
+ "pgzip-best-compression",
+ "pgzip-best-speed",
+ "s2-better",
+ "s2-default",
+ "s2-parallel-4",
+ "s2-parallel-8",
+ "zstd",
+ "zstd-better-compression",
+ "zstd-fastest":
+ return true
+ default:
+ return false
+ }
+}
+
+// Validate checks that the configured compression algorithm is supported.
+// A nil policy or an empty algorithm is considered valid and means the Kopia
+// default (inherited) policy is left untouched.
+func (c *CompressionPolicy) Validate() error {
+ if c == nil || c.Algorithm == "" {
+ return nil
+ }
+
+ if !IsValidCompressionAlgorithm(c.Algorithm) {
+ return fmt.Errorf("%w: %q", ErrInvalidCompressionAlgorithm, c.Algorithm)
+ }
+
+ return nil
+}
diff --git a/core/pkg/config/compression_test.go b/core/pkg/config/compression_test.go
new file mode 100644
index 00000000..82bc969c
--- /dev/null
+++ b/core/pkg/config/compression_test.go
@@ -0,0 +1,118 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package config
+
+import (
+ "errors"
+ "testing"
+)
+
+func TestIsValidCompressionAlgorithm(t *testing.T) {
+ tests := []struct {
+ algorithm string
+ want bool
+ }{
+ {"none", true},
+ {"zstd", true},
+ {"zstd-fastest", true},
+ {"zstd-better-compression", true},
+ {"s2-default", true},
+ {"gzip", true},
+ {"pgzip-best-compression", true},
+ {"deflate-default", true},
+ {"", false},
+ {"zstd-best-compression", false},
+ {"lz4", false},
+ {"bogus", false},
+ }
+ for _, tt := range tests {
+ t.Run(tt.algorithm, func(t *testing.T) {
+ if got := IsValidCompressionAlgorithm(tt.algorithm); got != tt.want {
+ t.Errorf("IsValidCompressionAlgorithm(%q) = %v, want %v", tt.algorithm, got, tt.want)
+ }
+ })
+ }
+}
+
+func TestCompressionPolicyValidate(t *testing.T) {
+ tests := []struct {
+ name string
+ policy *CompressionPolicy
+ wantErr bool
+ }{
+ {"nil policy", nil, false},
+ {"empty algorithm", &CompressionPolicy{}, false},
+ {"valid algorithm", &CompressionPolicy{Algorithm: "zstd"}, false},
+ {"none algorithm", &CompressionPolicy{Algorithm: "none"}, false},
+ {"invalid algorithm", &CompressionPolicy{Algorithm: "bogus"}, true},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ err := tt.policy.Validate()
+ if (err != nil) != tt.wantErr {
+ t.Errorf("Validate() error = %v, wantErr %v", err, tt.wantErr)
+ }
+ if tt.wantErr && !errors.Is(err, ErrInvalidCompressionAlgorithm) {
+ t.Errorf("Validate() error = %v, want it to wrap ErrInvalidCompressionAlgorithm", err)
+ }
+ })
+ }
+}
+
+func TestDataValidateCompression(t *testing.T) {
+ base := func() Data {
+ return Data{
+ Client: ClientConfig{
+ ClusterName: "cluster",
+ Base: BaseRepositoryClientConfig{
+ URL: "https://example",
+ ServerCertPath: "s",
+ ClientCertPath: "c",
+ ClientKeyPath: "k",
+ },
+ },
+ }
+ }
+
+ t.Run("valid tier1 and tier2 compression", func(t *testing.T) {
+ d := base()
+ d.Tier1CompressionPolicy = &CompressionPolicy{Algorithm: "zstd"}
+ d.Tier2CompressionPolicy = &CompressionPolicy{Algorithm: "s2-default"}
+ if err := d.Validate(); err != nil {
+ t.Errorf("Validate() unexpected error = %v", err)
+ }
+ })
+
+ t.Run("invalid tier1 compression", func(t *testing.T) {
+ d := base()
+ d.Tier1CompressionPolicy = &CompressionPolicy{Algorithm: "bogus"}
+ if err := d.Validate(); !errors.Is(err, ErrInvalidCompressionAlgorithm) {
+ t.Errorf("Validate() error = %v, want ErrInvalidCompressionAlgorithm", err)
+ }
+ })
+
+ t.Run("invalid tier2 compression", func(t *testing.T) {
+ d := base()
+ d.Tier2CompressionPolicy = &CompressionPolicy{Algorithm: "bogus"}
+ if err := d.Validate(); !errors.Is(err, ErrInvalidCompressionAlgorithm) {
+ t.Errorf("Validate() error = %v, want ErrInvalidCompressionAlgorithm", err)
+ }
+ })
+}
diff --git a/core/pkg/config/server.go b/core/pkg/config/server.go
index 09b77188..2b688344 100644
--- a/core/pkg/config/server.go
+++ b/core/pkg/config/server.go
@@ -71,6 +71,11 @@ type Tier1Config struct {
// Wal is the configuration of the Wal server
Wal WalServerConfig `mapstructure:"wal"`
+
+ // Compression is the repository-wide (global) compression policy applied
+ // to base backups stored on tier1. When empty, the Kopia default (no
+ // compression) is left untouched.
+ Compression CompressionServerConfig `mapstructure:"compression"`
}
// Tier2Config is the configuration of tier 2.
@@ -98,10 +103,36 @@ type Tier2Config struct {
// CacheDirectory is the directory of the Kopia cache
CacheDirectory string `mapstructure:"cache"`
+ // Compression is the repository-wide (global) compression policy applied
+ // to base backups stored on tier2. When empty, the Kopia default (no
+ // compression) is left untouched.
+ Compression CompressionServerConfig `mapstructure:"compression"`
+
// S3 contains the configuration parameters for an S3-based tier 2
S3 S3Configuration `json:"s3" mapstructure:"s3"`
}
+// CompressionServerConfig is the repository-wide (global) compression policy
+// applied to a tier when the Kopia server starts.
+type CompressionServerConfig struct {
+ // Algorithm is the name of the Kopia compression algorithm to use.
+ // The special value "none" disables compression.
+ Algorithm string `mapstructure:"algorithm"`
+
+ // MinSize is the minimum file size, in bytes, to attempt compression for.
+ // Zero means no minimum.
+ MinSize int64 `mapstructure:"min_size"`
+
+ // MaxSize is the maximum file size, in bytes, to attempt compression for.
+ // Zero means no maximum.
+ MaxSize int64 `mapstructure:"max_size"`
+}
+
+// IsZero reports whether the compression policy carries no settings.
+func (c CompressionServerConfig) IsZero() bool {
+ return c.Algorithm == "" && c.MinSize == 0 && c.MaxSize == 0
+}
+
// BaseServerConfig is the configuration that will be used for
// the kopia server.
type BaseServerConfig struct {
diff --git a/core/pkg/config/server_validate.go b/core/pkg/config/server_validate.go
index e66db7fe..14de6002 100644
--- a/core/pkg/config/server_validate.go
+++ b/core/pkg/config/server_validate.go
@@ -21,6 +21,7 @@ package config
import (
"errors"
+ "fmt"
"path/filepath"
)
@@ -76,6 +77,10 @@ func (c *Tier1Config) Validate() error {
if err := c.Wal.Validate(); err != nil {
errs = errors.Join(errs, err)
}
+ if c.Compression.Algorithm != "" && !IsValidCompressionAlgorithm(c.Compression.Algorithm) {
+ errs = errors.Join(errs, fmt.Errorf("invalid tier1 config: %w: %q",
+ ErrInvalidCompressionAlgorithm, c.Compression.Algorithm))
+ }
return errs
}
@@ -105,6 +110,11 @@ func (c *Tier2Config) Validate() error {
errs = errors.Join(errs, err)
}
+ if c.Compression.Algorithm != "" && !IsValidCompressionAlgorithm(c.Compression.Algorithm) {
+ errs = errors.Join(errs, fmt.Errorf("invalid tier2 config: %w: %q",
+ ErrInvalidCompressionAlgorithm, c.Compression.Algorithm))
+ }
+
return errs
}
diff --git a/core/pkg/config/server_validate_test.go b/core/pkg/config/server_validate_test.go
index 5850e427..c4bda5e3 100644
--- a/core/pkg/config/server_validate_test.go
+++ b/core/pkg/config/server_validate_test.go
@@ -105,6 +105,56 @@ func TestTierConfigsValidate(t *testing.T) {
})
}
+func TestTierConfigCompressionValidate(t *testing.T) {
+ validTier1 := func() Tier1Config {
+ return Tier1Config{
+ EncryptionKeyFile: "/path/to/key",
+ Base: BaseServerConfig{
+ CacheDirectory: "cache",
+ RepositoryDirectory: "repo",
+ ListenAddress: "address",
+ },
+ Wal: WalServerConfig{
+ ListenAddress: "address",
+ WALPath: "walPath",
+ },
+ }
+ }
+
+ t.Run("Tier1 valid compression", func(t *testing.T) {
+ cfg := validTier1()
+ cfg.Compression = CompressionServerConfig{Algorithm: "zstd"}
+ if err := cfg.Validate(); err != nil {
+ t.Errorf("Validate() unexpected error = %v", err)
+ }
+ })
+
+ t.Run("Tier1 empty compression", func(t *testing.T) {
+ cfg := validTier1()
+ if err := cfg.Validate(); err != nil {
+ t.Errorf("Validate() unexpected error = %v", err)
+ }
+ })
+
+ t.Run("Tier1 invalid compression", func(t *testing.T) {
+ cfg := validTier1()
+ cfg.Compression = CompressionServerConfig{Algorithm: "bogus"}
+ if err := cfg.Validate(); err == nil {
+ t.Errorf("Validate() expected an error for invalid compression")
+ }
+ })
+
+ t.Run("Tier2 invalid compression", func(t *testing.T) {
+ cfg := Tier2Config{
+ S3: S3Configuration{Enabled: false},
+ Compression: CompressionServerConfig{Algorithm: "bogus"},
+ }
+ if err := cfg.Validate(); err == nil {
+ t.Errorf("Validate() expected an error for invalid compression")
+ }
+ })
+}
+
func TestS3ConfigurationValidate(t *testing.T) {
tests := []struct {
name string
diff --git a/core/proto/klio_wal.proto b/core/proto/klio_wal.proto
index afbc3c15..2d936e0c 100644
--- a/core/proto/klio_wal.proto
+++ b/core/proto/klio_wal.proto
@@ -161,6 +161,9 @@ message CloseBackupRequest {
// When present, set the tier2 retention policy to the specified JSON-serialized policy.
string tier2_retention_policy = 9;
+
+ // When present, set the tier2 compression policy to the specified JSON-serialized policy.
+ string tier2_compression_policy = 10;
}
// This is sent by the WAL server in response to a CloseBackupRequest
diff --git a/documentation/.wordlist.txt b/documentation/.wordlist.txt
index 838d92a0..7307decd 100644
--- a/documentation/.wordlist.txt
+++ b/documentation/.wordlist.txt
@@ -28,6 +28,8 @@ CloudNativePG's
ClusterMetadata
ClusterName
CodeReady
+CompressionAlgorithm
+CompressionPolicy
DNS
Debian
DeleteBackup
@@ -75,6 +77,7 @@ KeepHourly
KeepLatest
KeepMonthly
KeepWeekly
+KiB
Klio
Klio's
KlioAPI
@@ -89,8 +92,10 @@ Liveness
LocalObjectReference
MaxConcurrentDownloads
MaxItems
+MaxSize
MiB
MinLength
+MinSize
ModeReadOnly
ModeStandard
NATS
@@ -193,6 +198,7 @@ WALPrefetch
WALPrefetchConfiguration
WALs
YAML
+Zstandard
api
args
aspirational
@@ -226,7 +232,10 @@ datacenters
decrypt
decrypted
deduplicated
+deduplicated
+deduplicates
deduplication
+deflate
deployable
dev
distroless
@@ -280,6 +289,7 @@ observability
opentelemetry
otel
otlp
+pgzip
plaintext
pluginConfigurationRef
podSecurityContext
@@ -299,6 +309,7 @@ pullPolicy
pullSecrets
quickstart
readinessProbe
+recompressed
renewBefore
restorable
rollout
@@ -324,3 +335,4 @@ verifications
wal
wals
yaml
+zstd
diff --git a/documentation/web/docs/developer/_protocol.md b/documentation/web/docs/developer/_protocol.md
index 4a1665e2..72af7d03 100644
--- a/documentation/web/docs/developer/_protocol.md
+++ b/documentation/web/docs/developer/_protocol.md
@@ -305,6 +305,7 @@ been completed.
| segment_size | [uint64](#uint64) | | The size of a WAL segment. Needed to generate the sequence of WAL files between the start and the end. |
| send_to_tier2 | [bool](#bool) | | Require this backup to be sent to tier2. |
| tier2_retention_policy | [string](#string) | | When present, set the tier2 retention policy to the specified JSON-serialized policy. |
+| tier2_compression_policy | [string](#string) | | When present, set the tier2 compression policy to the specified JSON-serialized policy. |
diff --git a/documentation/web/docs/developer/kopia_internals.md b/documentation/web/docs/developer/kopia_internals.md
index 462bdf20..c6134b73 100644
--- a/documentation/web/docs/developer/kopia_internals.md
+++ b/documentation/web/docs/developer/kopia_internals.md
@@ -157,6 +157,9 @@ Metadata is stored in **manifests** - JSON documents identified by labels rather
1. **Snapshot Manifests**: Record what was backed up, when, and the root object ID
2. **Policy Manifests**: Define retention rules, compression settings, etc.
+ Policy changes are read at snapshot-creation time, so they affect only
+ future snapshots; changing the compression algorithm never rewrites data
+ already stored.
3. **Other Manifests**: Maintenance schedules, ACLs, etc.
Each manifest has:
diff --git a/documentation/web/docs/developer/running-e2e-tests.md b/documentation/web/docs/developer/running-e2e-tests.md
index 5c173203..a9cbbaff 100644
--- a/documentation/web/docs/developer/running-e2e-tests.md
+++ b/documentation/web/docs/developer/running-e2e-tests.md
@@ -123,6 +123,11 @@ The E2E tests are located in `operator/test/e2e/` and include:
(`RecoverClusterFromTier2Pitr`)
- **`tier2_retention_test.go`** - Backup and WAL retention policy
enforcement in tier2 storage (`Tier2Retention`)
+- **`compression_test.go`** - Kopia compression policies: verifies the
+ repository-wide policy set on the Server applies globally and that the
+ per-cluster policy set on the PluginConfiguration overrides it, by
+ inspecting `kopia policy show` for the `--global` and `user@host`
+ targets on tier2 (`Compression`)
- **`wal_retention_test.go`** - WAL retention queue-awareness: verifies
server-side tier1 retention prunes old WALs only after they reach tier2,
driven by backup completion rather than a client command
diff --git a/documentation/web/docs/user/api/_klio_api.md b/documentation/web/docs/user/api/_klio_api.md
index 751b54d1..a5318a01 100644
--- a/documentation/web/docs/user/api/_klio_api.md
+++ b/documentation/web/docs/user/api/_klio_api.md
@@ -29,6 +29,43 @@ _Appears in:_
| `pvcTemplate` _[PersistentVolumeClaimSpec](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.37/#persistentvolumeclaimspec-v1-core)_ | | True | | |
+#### CompressionAlgorithm
+
+_Underlying type:_ _string_
+
+CompressionAlgorithm is the name of a Kopia compression algorithm.
+The special value "none" disables compression.
+
+_Validation:_
+- Enum: [none deflate-best-compression deflate-best-speed deflate-default gzip gzip-best-compression gzip-best-speed pgzip pgzip-best-compression pgzip-best-speed s2-better s2-default s2-parallel-4 s2-parallel-8 zstd zstd-better-compression zstd-fastest]
+
+_Appears in:_
+- [CompressionPolicy](#compressionpolicy)
+
+
+
+#### CompressionPolicy
+
+
+
+CompressionPolicy configures the Kopia compression policy applied to base
+backup data.
+
+
+
+_Appears in:_
+- [Tier1Configuration](#tier1configuration)
+- [Tier1PluginConfiguration](#tier1pluginconfiguration)
+- [Tier2Configuration](#tier2configuration)
+- [Tier2PluginConfiguration](#tier2pluginconfiguration)
+
+| Field | Description | Required | Default | Validation |
+| --- | --- | --- | --- | --- |
+| `algorithm` _[CompressionAlgorithm](#compressionalgorithm)_ | Algorithm is the name of the Kopia compression algorithm to use. | True | | Enum: [none deflate-best-compression deflate-best-speed deflate-default gzip gzip-best-compression gzip-best-speed pgzip pgzip-best-compression pgzip-best-speed s2-better s2-default s2-parallel-4 s2-parallel-8 zstd zstd-better-compression zstd-fastest]
Required: \{\}
|
+| `minSize` _integer_ | MinSize is the minimum file size, in bytes, to attempt compression for.
Files smaller than this are stored uncompressed. Zero means no minimum. | | | Minimum: 0
Optional: \{\}
|
+| `maxSize` _integer_ | MaxSize is the maximum file size, in bytes, to attempt compression for.
Files larger than this are stored uncompressed. Zero means no maximum. | | | Minimum: 0
Optional: \{\}
|
+
+
#### Data
@@ -366,6 +403,7 @@ _Appears in:_
| `data` _[Data](#data)_ | Data is the configuration of the PVC that should be used
for the base backups. | True | | |
| `encryptionKeyFile` _[FileSource](#filesource)_ | EncryptionKeyFile specifies the Age-encrypted encryption key file. | True | | ExactlyOneOf: [fileReference]
|
| `identityFile` _[FileSource](#filesource)_ | IdentityFile specifies the Age identity (private key) file used to
decrypt the encryption key. | True | | ExactlyOneOf: [fileReference]
|
+| `compression` _[CompressionPolicy](#compressionpolicy)_ | Compression defines the repository-wide (global) compression policy
applied to base backups stored on tier1. Individual clusters can
override it through their PluginConfiguration. | | | Optional: \{\}
|
#### Tier1PluginConfiguration
@@ -382,6 +420,7 @@ _Appears in:_
| Field | Description | Required | Default | Validation |
| --- | --- | --- | --- | --- |
| `retention` _[RetentionPolicy](#retentionpolicy)_ | RetentionPolicy defines how many backups we should keep | | | Optional: \{\}
|
+| `compression` _[CompressionPolicy](#compressionpolicy)_ | Compression defines the compression policy applied to this cluster's
base backups on tier1. It overrides the tier1 repository-wide policy
configured on the Server. | | | Optional: \{\}
|
#### Tier2Configuration
@@ -401,6 +440,7 @@ _Appears in:_
| `s3` _[S3Configuration](#s3configuration)_ | S3 contains the configuration parameters for an S3-based tier 2. | True | | |
| `encryptionKeyFile` _[FileSource](#filesource)_ | EncryptionKeyFile specifies the Age-encrypted encryption key file. | True | | ExactlyOneOf: [fileReference]
|
| `identityFile` _[FileSource](#filesource)_ | IdentityFile specifies the Age identity (private key) file used to
decrypt the encryption key. | True | | ExactlyOneOf: [fileReference]
|
+| `compression` _[CompressionPolicy](#compressionpolicy)_ | Compression defines the repository-wide (global) compression policy
applied to base backups stored on tier2. Individual clusters can
override it through their PluginConfiguration. | | | Optional: \{\}
|
#### Tier2PluginConfiguration
@@ -419,6 +459,7 @@ _Appears in:_
| `enableBackup` _boolean_ | EnableBackup controls whether WAL and base backups should be stored in tier2 | | | Optional: \{\}
|
| `enableRecovery` _boolean_ | EnableRecovery controls whether tier2 should be included in the recovery source list | | | Optional: \{\}
|
| `retention` _[RetentionPolicy](#retentionpolicy)_ | RetentionPolicy defines how many backups we should keep | | | Optional: \{\}
|
+| `compression` _[CompressionPolicy](#compressionpolicy)_ | Compression defines the compression policy applied to this cluster's
base backups on tier2. It overrides the tier2 repository-wide policy
configured on the Server. | | | Optional: \{\}
|
#### WALPrefetchConfiguration
diff --git a/documentation/web/docs/user/klio_server.md b/documentation/web/docs/user/klio_server.md
index 20717323..d741978e 100644
--- a/documentation/web/docs/user/klio_server.md
+++ b/documentation/web/docs/user/klio_server.md
@@ -296,6 +296,42 @@ merged with the default Klio `server` container. If you do not need to add
containers or modify the default one, you must still include an empty list.
:::
+### Compression
+
+By default, Kopia stores base backup data uncompressed. You can set a
+repository-wide compression policy per tier on the `Server`. This becomes the
+default for every cluster that backs up to this server; individual clusters can
+override it through their
+[PluginConfiguration](./plugin_configuration.md#compression-policies).
+
+```yaml
+apiVersion: klio.cnpg.io/v1alpha1
+kind: Server
+metadata:
+ name: klio-server
+spec:
+ # ...
+ tier1:
+ # ...
+ compression:
+ algorithm: zstd
+ tier2:
+ # ...
+ compression:
+ algorithm: zstd-better-compression
+```
+
+The `algorithm` field accepts any compression algorithm supported by Kopia
+(for example `zstd`, `s2-default`, `gzip`, or `none` to disable compression).
+The optional `minSize` and `maxSize` fields (in bytes, `0` meaning no limit)
+restrict compression to files within a size range, matching the per-cluster
+[PluginConfiguration](./plugin_configuration.md#compression-policies) fields.
+The global policy is applied to the repository when the server starts, so it
+only affects backups taken afterwards; existing backups are not recompressed.
+No manual step is needed beyond configuring the field. Because content is
+deduplicated by hash, only new or changed data is compressed with the new
+algorithm, so storage savings appear gradually as data churns.
+
### Node Affinity and Tolerations
To dedicate specific nodes for Klio workloads (e.g., for performance isolation
diff --git a/documentation/web/docs/user/plugin_configuration.md b/documentation/web/docs/user/plugin_configuration.md
index 81b59813..347170d4 100644
--- a/documentation/web/docs/user/plugin_configuration.md
+++ b/documentation/web/docs/user/plugin_configuration.md
@@ -267,6 +267,69 @@ keepAnnual: 1
Set a rule to `0` to disable that retention level.
+### Compression policies
+
+By default, Kopia stores base backup data uncompressed. You can enable
+compression for a cluster's base backups by configuring a compression policy
+per tier. The policy overrides the repository-wide default configured on the
+[Klio server](./klio_server.md#compression):
+
+```yaml
+apiVersion: klio.cnpg.io/v1alpha1
+kind: PluginConfiguration
+metadata:
+ name: klio-plugin-config
+spec:
+ serverAddress: klio-server.default
+ clientSecretName: cluster-example-klio-user
+ serverSecretName: klio-server-tls
+ clusterName: cluster-example
+ tier1:
+ compression:
+ algorithm: zstd
+ tier2:
+ enableBackup: true
+ compression:
+ algorithm: zstd-better-compression
+```
+
+The `algorithm` field selects one of the compression algorithms supported by
+Kopia. Common choices are:
+
+- `none`: disable compression (the default behavior).
+- `zstd`, `zstd-fastest`, `zstd-better-compression`: the Zstandard family,
+ offering a good balance of ratio and speed.
+- `s2-default`, `s2-better`: the S2 family, optimized for throughput.
+- `gzip`, `gzip-best-compression`, `gzip-best-speed`.
+- `pgzip`, `pgzip-best-compression`, `pgzip-best-speed`.
+- `deflate-default`, `deflate-best-compression`, `deflate-best-speed`.
+
+You can optionally bound which files are compressed by size, in bytes:
+
+- `minSize`: files smaller than this are stored uncompressed. Useful to skip
+ the per-file overhead of compressing very small files.
+- `maxSize`: files larger than this are stored uncompressed.
+
+Both default to `0`, which means no limit. For example, to compress only files
+of at least 4 KiB:
+
+```yaml
+ tier1:
+ compression:
+ algorithm: zstd
+ minSize: 4096
+```
+
+Compression only affects backups taken after the policy is applied; existing
+backups are not recompressed. WAL files are always compressed independently
+and are not affected by this setting.
+
+No manual step is needed for a policy change to take effect: the next backup
+picks it up automatically. Because Kopia deduplicates content by hash, a new
+algorithm applies only to data that is new or changed in later backups; blocks
+already stored uncompressed are reused as they are, so repository savings
+appear gradually as data churns rather than all at once.
+
### Operation Mode
The `mode` field controls whether the plugin can perform both backup and
diff --git a/operator/api/v1alpha1/plugin_configuration_types.go b/operator/api/v1alpha1/plugin_configuration_types.go
index 83d07b7d..e0b68948 100644
--- a/operator/api/v1alpha1/plugin_configuration_types.go
+++ b/operator/api/v1alpha1/plugin_configuration_types.go
@@ -106,6 +106,12 @@ type Tier1PluginConfiguration struct {
// RetentionPolicy defines how many backups we should keep
// +optional
RetentionPolicy *RetentionPolicy `json:"retention,omitempty" mapstructure:"retention"`
+
+ // Compression defines the compression policy applied to this cluster's
+ // base backups on tier1. It overrides the tier1 repository-wide policy
+ // configured on the Server.
+ // +optional
+ Compression *CompressionPolicy `json:"compression,omitempty" mapstructure:"compression"`
}
// Tier2PluginConfiguration configures tier2 backup and recovery settings.
@@ -122,6 +128,37 @@ type Tier2PluginConfiguration struct {
// RetentionPolicy defines how many backups we should keep
// +optional
RetentionPolicy *RetentionPolicy `json:"retention,omitempty" mapstructure:"retention"`
+
+ // Compression defines the compression policy applied to this cluster's
+ // base backups on tier2. It overrides the tier2 repository-wide policy
+ // configured on the Server.
+ // +optional
+ Compression *CompressionPolicy `json:"compression,omitempty" mapstructure:"compression"`
+}
+
+// CompressionAlgorithm is the name of a Kopia compression algorithm.
+// The special value "none" disables compression.
+// +kubebuilder:validation:Enum=none;deflate-best-compression;deflate-best-speed;deflate-default;gzip;gzip-best-compression;gzip-best-speed;pgzip;pgzip-best-compression;pgzip-best-speed;s2-better;s2-default;s2-parallel-4;s2-parallel-8;zstd;zstd-better-compression;zstd-fastest
+type CompressionAlgorithm string
+
+// CompressionPolicy configures the Kopia compression policy applied to base
+// backup data.
+type CompressionPolicy struct {
+ // Algorithm is the name of the Kopia compression algorithm to use.
+ // +kubebuilder:validation:Required
+ Algorithm CompressionAlgorithm `json:"algorithm" mapstructure:"algorithm"`
+
+ // MinSize is the minimum file size, in bytes, to attempt compression for.
+ // Files smaller than this are stored uncompressed. Zero means no minimum.
+ // +optional
+ // +kubebuilder:validation:Minimum=0
+ MinSize int64 `json:"minSize,omitempty" mapstructure:"minSize"`
+
+ // MaxSize is the maximum file size, in bytes, to attempt compression for.
+ // Files larger than this are stored uncompressed. Zero means no maximum.
+ // +optional
+ // +kubebuilder:validation:Minimum=0
+ MaxSize int64 `json:"maxSize,omitempty" mapstructure:"maxSize"`
}
// WALPrefetchConfiguration configures WAL prefetching during recovery.
diff --git a/operator/api/v1alpha1/server_types.go b/operator/api/v1alpha1/server_types.go
index 4f98cb5f..91087ad0 100644
--- a/operator/api/v1alpha1/server_types.go
+++ b/operator/api/v1alpha1/server_types.go
@@ -191,6 +191,12 @@ type Tier1Configuration struct {
// IdentityFile specifies the Age identity (private key) file used to
// decrypt the encryption key.
IdentityFile FileSource `json:"identityFile"`
+
+ // Compression defines the repository-wide (global) compression policy
+ // applied to base backups stored on tier1. Individual clusters can
+ // override it through their PluginConfiguration.
+ // +optional
+ Compression *CompressionPolicy `json:"compression,omitempty"`
}
// Tier2Configuration is the tier 2 configuration.
@@ -208,6 +214,12 @@ type Tier2Configuration struct {
// IdentityFile specifies the Age identity (private key) file used to
// decrypt the encryption key.
IdentityFile FileSource `json:"identityFile"`
+
+ // Compression defines the repository-wide (global) compression policy
+ // applied to base backups stored on tier2. Individual clusters can
+ // override it through their PluginConfiguration.
+ // +optional
+ Compression *CompressionPolicy `json:"compression,omitempty"`
}
// S3Configuration is the configuration to a S3 defined tier 2.
diff --git a/operator/api/v1alpha1/zz_generated.deepcopy.go b/operator/api/v1alpha1/zz_generated.deepcopy.go
index d0ac006c..ea0d7278 100644
--- a/operator/api/v1alpha1/zz_generated.deepcopy.go
+++ b/operator/api/v1alpha1/zz_generated.deepcopy.go
@@ -46,6 +46,21 @@ func (in *Cache) DeepCopy() *Cache {
return out
}
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CompressionPolicy) DeepCopyInto(out *CompressionPolicy) {
+ *out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CompressionPolicy.
+func (in *CompressionPolicy) DeepCopy() *CompressionPolicy {
+ if in == nil {
+ return nil
+ }
+ out := new(CompressionPolicy)
+ in.DeepCopyInto(out)
+ return out
+}
+
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *Data) DeepCopyInto(out *Data) {
*out = *in
@@ -511,6 +526,11 @@ func (in *Tier1Configuration) DeepCopyInto(out *Tier1Configuration) {
in.Data.DeepCopyInto(&out.Data)
in.EncryptionKeyFile.DeepCopyInto(&out.EncryptionKeyFile)
in.IdentityFile.DeepCopyInto(&out.IdentityFile)
+ if in.Compression != nil {
+ in, out := &in.Compression, &out.Compression
+ *out = new(CompressionPolicy)
+ **out = **in
+ }
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Tier1Configuration.
@@ -531,6 +551,11 @@ func (in *Tier1PluginConfiguration) DeepCopyInto(out *Tier1PluginConfiguration)
*out = new(RetentionPolicy)
(*in).DeepCopyInto(*out)
}
+ if in.Compression != nil {
+ in, out := &in.Compression, &out.Compression
+ *out = new(CompressionPolicy)
+ **out = **in
+ }
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Tier1PluginConfiguration.
@@ -554,6 +579,11 @@ func (in *Tier2Configuration) DeepCopyInto(out *Tier2Configuration) {
}
in.EncryptionKeyFile.DeepCopyInto(&out.EncryptionKeyFile)
in.IdentityFile.DeepCopyInto(&out.IdentityFile)
+ if in.Compression != nil {
+ in, out := &in.Compression, &out.Compression
+ *out = new(CompressionPolicy)
+ **out = **in
+ }
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Tier2Configuration.
@@ -574,6 +604,11 @@ func (in *Tier2PluginConfiguration) DeepCopyInto(out *Tier2PluginConfiguration)
*out = new(RetentionPolicy)
(*in).DeepCopyInto(*out)
}
+ if in.Compression != nil {
+ in, out := &in.Compression, &out.Compression
+ *out = new(CompressionPolicy)
+ **out = **in
+ }
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Tier2PluginConfiguration.
diff --git a/operator/config/crd/bases/klio.cnpg.io_pluginconfigurations.yaml b/operator/config/crd/bases/klio.cnpg.io_pluginconfigurations.yaml
index 830fbbc1..f08ebaf1 100644
--- a/operator/config/crd/bases/klio.cnpg.io_pluginconfigurations.yaml
+++ b/operator/config/crd/bases/klio.cnpg.io_pluginconfigurations.yaml
@@ -1615,6 +1615,51 @@ spec:
tier1:
description: Tier1 is the Tier 1 configuration
properties:
+ compression:
+ description: |-
+ Compression defines the compression policy applied to this cluster's
+ base backups on tier1. It overrides the tier1 repository-wide policy
+ configured on the Server.
+ properties:
+ algorithm:
+ description: Algorithm is the name of the Kopia compression
+ algorithm to use.
+ enum:
+ - none
+ - deflate-best-compression
+ - deflate-best-speed
+ - deflate-default
+ - gzip
+ - gzip-best-compression
+ - gzip-best-speed
+ - pgzip
+ - pgzip-best-compression
+ - pgzip-best-speed
+ - s2-better
+ - s2-default
+ - s2-parallel-4
+ - s2-parallel-8
+ - zstd
+ - zstd-better-compression
+ - zstd-fastest
+ type: string
+ maxSize:
+ description: |-
+ MaxSize is the maximum file size, in bytes, to attempt compression for.
+ Files larger than this are stored uncompressed. Zero means no maximum.
+ format: int64
+ minimum: 0
+ type: integer
+ minSize:
+ description: |-
+ MinSize is the minimum file size, in bytes, to attempt compression for.
+ Files smaller than this are stored uncompressed. Zero means no minimum.
+ format: int64
+ minimum: 0
+ type: integer
+ required:
+ - algorithm
+ type: object
retention:
description: RetentionPolicy defines how many backups we should
keep
@@ -1654,6 +1699,51 @@ spec:
tier2:
description: Tier2 is the Tier 2 configuration
properties:
+ compression:
+ description: |-
+ Compression defines the compression policy applied to this cluster's
+ base backups on tier2. It overrides the tier2 repository-wide policy
+ configured on the Server.
+ properties:
+ algorithm:
+ description: Algorithm is the name of the Kopia compression
+ algorithm to use.
+ enum:
+ - none
+ - deflate-best-compression
+ - deflate-best-speed
+ - deflate-default
+ - gzip
+ - gzip-best-compression
+ - gzip-best-speed
+ - pgzip
+ - pgzip-best-compression
+ - pgzip-best-speed
+ - s2-better
+ - s2-default
+ - s2-parallel-4
+ - s2-parallel-8
+ - zstd
+ - zstd-better-compression
+ - zstd-fastest
+ type: string
+ maxSize:
+ description: |-
+ MaxSize is the maximum file size, in bytes, to attempt compression for.
+ Files larger than this are stored uncompressed. Zero means no maximum.
+ format: int64
+ minimum: 0
+ type: integer
+ minSize:
+ description: |-
+ MinSize is the minimum file size, in bytes, to attempt compression for.
+ Files smaller than this are stored uncompressed. Zero means no minimum.
+ format: int64
+ minimum: 0
+ type: integer
+ required:
+ - algorithm
+ type: object
enableBackup:
description: EnableBackup controls whether WAL and base backups
should be stored in tier2
diff --git a/operator/config/crd/bases/klio.cnpg.io_servers.yaml b/operator/config/crd/bases/klio.cnpg.io_servers.yaml
index 2242fec1..440cdff5 100644
--- a/operator/config/crd/bases/klio.cnpg.io_servers.yaml
+++ b/operator/config/crd/bases/klio.cnpg.io_servers.yaml
@@ -8976,6 +8976,51 @@ spec:
required:
- pvcTemplate
type: object
+ compression:
+ description: |-
+ Compression defines the repository-wide (global) compression policy
+ applied to base backups stored on tier1. Individual clusters can
+ override it through their PluginConfiguration.
+ properties:
+ algorithm:
+ description: Algorithm is the name of the Kopia compression
+ algorithm to use.
+ enum:
+ - none
+ - deflate-best-compression
+ - deflate-best-speed
+ - deflate-default
+ - gzip
+ - gzip-best-compression
+ - gzip-best-speed
+ - pgzip
+ - pgzip-best-compression
+ - pgzip-best-speed
+ - s2-better
+ - s2-default
+ - s2-parallel-4
+ - s2-parallel-8
+ - zstd
+ - zstd-better-compression
+ - zstd-fastest
+ type: string
+ maxSize:
+ description: |-
+ MaxSize is the maximum file size, in bytes, to attempt compression for.
+ Files larger than this are stored uncompressed. Zero means no maximum.
+ format: int64
+ minimum: 0
+ type: integer
+ minSize:
+ description: |-
+ MinSize is the minimum file size, in bytes, to attempt compression for.
+ Files smaller than this are stored uncompressed. Zero means no minimum.
+ format: int64
+ minimum: 0
+ type: integer
+ required:
+ - algorithm
+ type: object
data:
description: |-
Data is the configuration of the PVC that should be used
@@ -13276,6 +13321,51 @@ spec:
required:
- pvcTemplate
type: object
+ compression:
+ description: |-
+ Compression defines the repository-wide (global) compression policy
+ applied to base backups stored on tier2. Individual clusters can
+ override it through their PluginConfiguration.
+ properties:
+ algorithm:
+ description: Algorithm is the name of the Kopia compression
+ algorithm to use.
+ enum:
+ - none
+ - deflate-best-compression
+ - deflate-best-speed
+ - deflate-default
+ - gzip
+ - gzip-best-compression
+ - gzip-best-speed
+ - pgzip
+ - pgzip-best-compression
+ - pgzip-best-speed
+ - s2-better
+ - s2-default
+ - s2-parallel-4
+ - s2-parallel-8
+ - zstd
+ - zstd-better-compression
+ - zstd-fastest
+ type: string
+ maxSize:
+ description: |-
+ MaxSize is the maximum file size, in bytes, to attempt compression for.
+ Files larger than this are stored uncompressed. Zero means no maximum.
+ format: int64
+ minimum: 0
+ type: integer
+ minSize:
+ description: |-
+ MinSize is the minimum file size, in bytes, to attempt compression for.
+ Files smaller than this are stored uncompressed. Zero means no minimum.
+ format: int64
+ minimum: 0
+ type: integer
+ required:
+ - algorithm
+ type: object
encryptionKeyFile:
description: EncryptionKeyFile specifies the Age-encrypted encryption
key file.
diff --git a/operator/dist/chart/crds/pluginconfiguration-crd.yaml b/operator/dist/chart/crds/pluginconfiguration-crd.yaml
index 3b2e1408..e94dba22 100644
--- a/operator/dist/chart/crds/pluginconfiguration-crd.yaml
+++ b/operator/dist/chart/crds/pluginconfiguration-crd.yaml
@@ -1614,6 +1614,51 @@ spec:
tier1:
description: Tier1 is the Tier 1 configuration
properties:
+ compression:
+ description: |-
+ Compression defines the compression policy applied to this cluster's
+ base backups on tier1. It overrides the tier1 repository-wide policy
+ configured on the Server.
+ properties:
+ algorithm:
+ description: Algorithm is the name of the Kopia compression
+ algorithm to use.
+ enum:
+ - none
+ - deflate-best-compression
+ - deflate-best-speed
+ - deflate-default
+ - gzip
+ - gzip-best-compression
+ - gzip-best-speed
+ - pgzip
+ - pgzip-best-compression
+ - pgzip-best-speed
+ - s2-better
+ - s2-default
+ - s2-parallel-4
+ - s2-parallel-8
+ - zstd
+ - zstd-better-compression
+ - zstd-fastest
+ type: string
+ maxSize:
+ description: |-
+ MaxSize is the maximum file size, in bytes, to attempt compression for.
+ Files larger than this are stored uncompressed. Zero means no maximum.
+ format: int64
+ minimum: 0
+ type: integer
+ minSize:
+ description: |-
+ MinSize is the minimum file size, in bytes, to attempt compression for.
+ Files smaller than this are stored uncompressed. Zero means no minimum.
+ format: int64
+ minimum: 0
+ type: integer
+ required:
+ - algorithm
+ type: object
retention:
description: RetentionPolicy defines how many backups we should
keep
@@ -1653,6 +1698,51 @@ spec:
tier2:
description: Tier2 is the Tier 2 configuration
properties:
+ compression:
+ description: |-
+ Compression defines the compression policy applied to this cluster's
+ base backups on tier2. It overrides the tier2 repository-wide policy
+ configured on the Server.
+ properties:
+ algorithm:
+ description: Algorithm is the name of the Kopia compression
+ algorithm to use.
+ enum:
+ - none
+ - deflate-best-compression
+ - deflate-best-speed
+ - deflate-default
+ - gzip
+ - gzip-best-compression
+ - gzip-best-speed
+ - pgzip
+ - pgzip-best-compression
+ - pgzip-best-speed
+ - s2-better
+ - s2-default
+ - s2-parallel-4
+ - s2-parallel-8
+ - zstd
+ - zstd-better-compression
+ - zstd-fastest
+ type: string
+ maxSize:
+ description: |-
+ MaxSize is the maximum file size, in bytes, to attempt compression for.
+ Files larger than this are stored uncompressed. Zero means no maximum.
+ format: int64
+ minimum: 0
+ type: integer
+ minSize:
+ description: |-
+ MinSize is the minimum file size, in bytes, to attempt compression for.
+ Files smaller than this are stored uncompressed. Zero means no minimum.
+ format: int64
+ minimum: 0
+ type: integer
+ required:
+ - algorithm
+ type: object
enableBackup:
description: EnableBackup controls whether WAL and base backups
should be stored in tier2
diff --git a/operator/dist/chart/crds/server-crd.yaml b/operator/dist/chart/crds/server-crd.yaml
index 627c8976..c430327b 100644
--- a/operator/dist/chart/crds/server-crd.yaml
+++ b/operator/dist/chart/crds/server-crd.yaml
@@ -8975,6 +8975,51 @@ spec:
required:
- pvcTemplate
type: object
+ compression:
+ description: |-
+ Compression defines the repository-wide (global) compression policy
+ applied to base backups stored on tier1. Individual clusters can
+ override it through their PluginConfiguration.
+ properties:
+ algorithm:
+ description: Algorithm is the name of the Kopia compression
+ algorithm to use.
+ enum:
+ - none
+ - deflate-best-compression
+ - deflate-best-speed
+ - deflate-default
+ - gzip
+ - gzip-best-compression
+ - gzip-best-speed
+ - pgzip
+ - pgzip-best-compression
+ - pgzip-best-speed
+ - s2-better
+ - s2-default
+ - s2-parallel-4
+ - s2-parallel-8
+ - zstd
+ - zstd-better-compression
+ - zstd-fastest
+ type: string
+ maxSize:
+ description: |-
+ MaxSize is the maximum file size, in bytes, to attempt compression for.
+ Files larger than this are stored uncompressed. Zero means no maximum.
+ format: int64
+ minimum: 0
+ type: integer
+ minSize:
+ description: |-
+ MinSize is the minimum file size, in bytes, to attempt compression for.
+ Files smaller than this are stored uncompressed. Zero means no minimum.
+ format: int64
+ minimum: 0
+ type: integer
+ required:
+ - algorithm
+ type: object
data:
description: |-
Data is the configuration of the PVC that should be used
@@ -13275,6 +13320,51 @@ spec:
required:
- pvcTemplate
type: object
+ compression:
+ description: |-
+ Compression defines the repository-wide (global) compression policy
+ applied to base backups stored on tier2. Individual clusters can
+ override it through their PluginConfiguration.
+ properties:
+ algorithm:
+ description: Algorithm is the name of the Kopia compression
+ algorithm to use.
+ enum:
+ - none
+ - deflate-best-compression
+ - deflate-best-speed
+ - deflate-default
+ - gzip
+ - gzip-best-compression
+ - gzip-best-speed
+ - pgzip
+ - pgzip-best-compression
+ - pgzip-best-speed
+ - s2-better
+ - s2-default
+ - s2-parallel-4
+ - s2-parallel-8
+ - zstd
+ - zstd-better-compression
+ - zstd-fastest
+ type: string
+ maxSize:
+ description: |-
+ MaxSize is the maximum file size, in bytes, to attempt compression for.
+ Files larger than this are stored uncompressed. Zero means no maximum.
+ format: int64
+ minimum: 0
+ type: integer
+ minSize:
+ description: |-
+ MinSize is the minimum file size, in bytes, to attempt compression for.
+ Files smaller than this are stored uncompressed. Zero means no minimum.
+ format: int64
+ minimum: 0
+ type: integer
+ required:
+ - algorithm
+ type: object
encryptionKeyFile:
description: EncryptionKeyFile specifies the Age-encrypted encryption
key file.
diff --git a/operator/internal/controller/server_envbuilder.go b/operator/internal/controller/server_envbuilder.go
index c9f22758..19d3eeca 100644
--- a/operator/internal/controller/server_envbuilder.go
+++ b/operator/internal/controller/server_envbuilder.go
@@ -21,6 +21,7 @@ package controller
import (
"path"
+ "strconv"
machineryapi "github.com/cloudnative-pg/machinery/pkg/api"
corev1 "k8s.io/api/core/v1"
@@ -152,6 +153,8 @@ func (e *envBuilder) getCoreEnvVars() []corev1.EnvVar {
Value: "/queue",
})
+ tier1Envs = appendCompressionEnvs(tier1Envs, "TIER1", e.tier1.Compression)
+
result = append(result, tier1Envs...)
}
@@ -232,9 +235,40 @@ func (e *envBuilder) getTier2EnvVars() []corev1.EnvVar {
result = appendEnvIfNotEmpty(result, "TIER2_S3_PREFIX", e.tier2.S3.Prefix)
result = appendEnvIfNotEmpty(result, "TIER2_S3_REGION", e.tier2.S3.Region)
+ result = appendCompressionEnvs(result, "TIER2", e.tier2.Compression)
+
return result
}
+// appendCompressionEnvs emits the compression policy environment variables for
+// the given tier prefix (e.g. "TIER1"). Only the fields that are set produce a
+// variable, so an unset field leaves the corresponding Kopia policy untouched.
+func appendCompressionEnvs(
+ envs []corev1.EnvVar,
+ prefix string,
+ compression *kliov1alpha1.CompressionPolicy,
+) []corev1.EnvVar {
+ if compression == nil {
+ return envs
+ }
+
+ envs = appendEnvIfNotEmpty(envs, prefix+"_COMPRESSION_ALGORITHM", string(compression.Algorithm))
+ if compression.MinSize > 0 {
+ envs = append(envs, corev1.EnvVar{
+ Name: prefix + "_COMPRESSION_MIN_SIZE",
+ Value: strconv.FormatInt(compression.MinSize, 10),
+ })
+ }
+ if compression.MaxSize > 0 {
+ envs = append(envs, corev1.EnvVar{
+ Name: prefix + "_COMPRESSION_MAX_SIZE",
+ Value: strconv.FormatInt(compression.MaxSize, 10),
+ })
+ }
+
+ return envs
+}
+
func appendEnvIfNotEmpty(envs []corev1.EnvVar, name, value string) []corev1.EnvVar {
if value != "" {
return append(envs, corev1.EnvVar{Name: name, Value: value})
diff --git a/operator/internal/controller/server_envbuilder_test.go b/operator/internal/controller/server_envbuilder_test.go
index c41ca69f..95da8dc7 100644
--- a/operator/internal/controller/server_envbuilder_test.go
+++ b/operator/internal/controller/server_envbuilder_test.go
@@ -101,6 +101,106 @@ func TestGetCoreEnvVarsIncludesTier1EnvVars(t *testing.T) {
assert.Equal(t, "/files/tier1-identity/identity.txt", identityFile.Value)
}
+func TestGetCoreEnvVarsIncludesTier1Compression(t *testing.T) {
+ t.Run("compression set with sizes", func(t *testing.T) {
+ builder := &envBuilder{
+ tier1: &kliov1alpha1.Tier1Configuration{
+ EncryptionKeyFile: newTestFileSource("enc-secret", "encryption-key.age"),
+ IdentityFile: newTestFileSource("id-secret", "identity.txt"),
+ Compression: &kliov1alpha1.CompressionPolicy{
+ Algorithm: "zstd",
+ MinSize: 4096,
+ MaxSize: 1048576,
+ },
+ },
+ }
+
+ envVars := builder.getCoreEnvVars()
+
+ algorithm := findEnvVar(envVars, "TIER1_COMPRESSION_ALGORITHM")
+ require.NotNil(t, algorithm)
+ assert.Equal(t, "zstd", algorithm.Value)
+
+ minSize := findEnvVar(envVars, "TIER1_COMPRESSION_MIN_SIZE")
+ require.NotNil(t, minSize)
+ assert.Equal(t, "4096", minSize.Value)
+
+ maxSize := findEnvVar(envVars, "TIER1_COMPRESSION_MAX_SIZE")
+ require.NotNil(t, maxSize)
+ assert.Equal(t, "1048576", maxSize.Value)
+ })
+
+ t.Run("algorithm only omits size vars", func(t *testing.T) {
+ builder := &envBuilder{
+ tier1: &kliov1alpha1.Tier1Configuration{
+ EncryptionKeyFile: newTestFileSource("enc-secret", "encryption-key.age"),
+ IdentityFile: newTestFileSource("id-secret", "identity.txt"),
+ Compression: &kliov1alpha1.CompressionPolicy{Algorithm: "zstd"},
+ },
+ }
+
+ envVars := builder.getCoreEnvVars()
+ require.NotNil(t, findEnvVar(envVars, "TIER1_COMPRESSION_ALGORITHM"))
+ assert.Nil(t, findEnvVar(envVars, "TIER1_COMPRESSION_MIN_SIZE"))
+ assert.Nil(t, findEnvVar(envVars, "TIER1_COMPRESSION_MAX_SIZE"))
+ })
+
+ t.Run("compression unset", func(t *testing.T) {
+ builder := &envBuilder{
+ tier1: &kliov1alpha1.Tier1Configuration{
+ EncryptionKeyFile: newTestFileSource("enc-secret", "encryption-key.age"),
+ IdentityFile: newTestFileSource("id-secret", "identity.txt"),
+ },
+ }
+
+ assert.Nil(t, findEnvVar(builder.getCoreEnvVars(), "TIER1_COMPRESSION_ALGORITHM"))
+ })
+}
+
+func TestGetTier2EnvVarsIncludesCompression(t *testing.T) {
+ t.Run("compression set with sizes", func(t *testing.T) {
+ builder := &envBuilder{
+ tier2: &kliov1alpha1.Tier2Configuration{
+ S3: &kliov1alpha1.S3Configuration{
+ BucketName: "test-bucket",
+ },
+ EncryptionKeyFile: newTestFileSource("enc-secret", "encryption-key.age"),
+ IdentityFile: newTestFileSource("id-secret", "identity.txt"),
+ Compression: &kliov1alpha1.CompressionPolicy{
+ Algorithm: "s2-default",
+ MinSize: 8192,
+ },
+ },
+ }
+
+ envVars := builder.getTier2EnvVars()
+
+ algorithm := findEnvVar(envVars, "TIER2_COMPRESSION_ALGORITHM")
+ require.NotNil(t, algorithm)
+ assert.Equal(t, "s2-default", algorithm.Value)
+
+ minSize := findEnvVar(envVars, "TIER2_COMPRESSION_MIN_SIZE")
+ require.NotNil(t, minSize)
+ assert.Equal(t, "8192", minSize.Value)
+
+ assert.Nil(t, findEnvVar(envVars, "TIER2_COMPRESSION_MAX_SIZE"))
+ })
+
+ t.Run("compression unset", func(t *testing.T) {
+ builder := &envBuilder{
+ tier2: &kliov1alpha1.Tier2Configuration{
+ S3: &kliov1alpha1.S3Configuration{
+ BucketName: "test-bucket",
+ },
+ EncryptionKeyFile: newTestFileSource("enc-secret", "encryption-key.age"),
+ IdentityFile: newTestFileSource("id-secret", "identity.txt"),
+ },
+ }
+
+ assert.Nil(t, findEnvVar(builder.getTier2EnvVars(), "TIER2_COMPRESSION_ALGORITHM"))
+ })
+}
+
func TestGetCoreEnvVarsOnlyTLSWhenNoTier1(t *testing.T) {
builder := &envBuilder{
tier1: nil,
diff --git a/operator/internal/klioconfig/config.go b/operator/internal/klioconfig/config.go
index 89b43d0d..0c4143e4 100644
--- a/operator/internal/klioconfig/config.go
+++ b/operator/internal/klioconfig/config.go
@@ -96,6 +96,8 @@ func GenerateConfig(
klioTier1RetentionPolicy := convertTier1RetentionPolicy(spec.Tier1)
klioTier2RetentionPolicy := convertTier2RetentionPolicy(spec.Tier2)
+ klioTier1CompressionPolicy := convertTier1CompressionPolicy(spec.Tier1)
+ klioTier2CompressionPolicy := convertTier2CompressionPolicy(spec.Tier2)
walPrefetch := spec.GetWALPrefetch()
klioConfig := &config.Data{
@@ -121,8 +123,10 @@ func GenerateConfig(
ClientKeyPath: path.Join(clientCertPath, tlsKeyFile),
},
},
- Tier1RetentionPolicy: klioTier1RetentionPolicy,
- Tier2RetentionPolicy: klioTier2RetentionPolicy,
+ Tier1RetentionPolicy: klioTier1RetentionPolicy,
+ Tier2RetentionPolicy: klioTier2RetentionPolicy,
+ Tier1CompressionPolicy: klioTier1CompressionPolicy,
+ Tier2CompressionPolicy: klioTier2CompressionPolicy,
WALPrefetch: config.WALPrefetchConfig{
Count: walPrefetch.Count,
MaxConcurrentDownloads: walPrefetch.MaxConcurrentDownloads,
@@ -183,6 +187,34 @@ func convertTier2RetentionPolicy(tier2 *kliov1alpha1.Tier2PluginConfiguration) *
return convertRetentionPolicy(tier2.RetentionPolicy)
}
+func convertCompressionPolicy(p *kliov1alpha1.CompressionPolicy) *config.CompressionPolicy {
+ if p == nil {
+ return nil
+ }
+
+ return &config.CompressionPolicy{
+ Algorithm: string(p.Algorithm),
+ MinSize: p.MinSize,
+ MaxSize: p.MaxSize,
+ }
+}
+
+func convertTier1CompressionPolicy(tier1 *kliov1alpha1.Tier1PluginConfiguration) *config.CompressionPolicy {
+ if tier1 == nil {
+ return nil
+ }
+
+ return convertCompressionPolicy(tier1.Compression)
+}
+
+func convertTier2CompressionPolicy(tier2 *kliov1alpha1.Tier2PluginConfiguration) *config.CompressionPolicy {
+ if tier2 == nil {
+ return nil
+ }
+
+ return convertCompressionPolicy(tier2.Compression)
+}
+
// configuration holds the CNPG and Klio plugin configurations for a single config key.
// It is used internally by getConfigurations and is not exported.
type configuration struct {
diff --git a/operator/internal/klioconfig/config_test.go b/operator/internal/klioconfig/config_test.go
index 7ffc5d47..0a4d70ae 100644
--- a/operator/internal/klioconfig/config_test.go
+++ b/operator/internal/klioconfig/config_test.go
@@ -289,6 +289,31 @@ func TestGenerateConfig(t *testing.T) {
assert.Equal(t, new(10), cfg.Tier2RetentionPolicy.KeepLatest)
},
},
+ {
+ name: "tier1 and tier2 compression policies are set",
+ spec: kliov1alpha1.PluginConfigurationSpec{
+ ServerAddress: testServerAddress,
+ Mode: kliov1alpha1.ModeStandard,
+ ClusterName: testClusterName,
+ Tier1: &kliov1alpha1.Tier1PluginConfiguration{
+ Compression: &kliov1alpha1.CompressionPolicy{Algorithm: "zstd", MinSize: 4096},
+ },
+ Tier2: &kliov1alpha1.Tier2PluginConfiguration{
+ EnableBackup: true,
+ Compression: &kliov1alpha1.CompressionPolicy{Algorithm: "s2-default", MaxSize: 1048576},
+ },
+ },
+ configKey: ArchiveConfigKey,
+ assertions: func(t *testing.T, cfg *config.Data) {
+ t.Helper()
+ assert.NotNil(t, cfg.Tier1CompressionPolicy)
+ assert.Equal(t, "zstd", cfg.Tier1CompressionPolicy.Algorithm)
+ assert.Equal(t, int64(4096), cfg.Tier1CompressionPolicy.MinSize)
+ assert.NotNil(t, cfg.Tier2CompressionPolicy)
+ assert.Equal(t, "s2-default", cfg.Tier2CompressionPolicy.Algorithm)
+ assert.Equal(t, int64(1048576), cfg.Tier2CompressionPolicy.MaxSize)
+ },
+ },
{
name: "source config has default values",
spec: kliov1alpha1.PluginConfigurationSpec{
@@ -387,6 +412,64 @@ func TestConvertTier1RetentionPolicy(t *testing.T) {
})
}
+func TestConvertCompressionPolicy(t *testing.T) {
+ t.Run("nil returns nil", func(t *testing.T) {
+ assert.Nil(t, convertCompressionPolicy(nil))
+ })
+
+ t.Run("copies the algorithm and sizes", func(t *testing.T) {
+ result := convertCompressionPolicy(&kliov1alpha1.CompressionPolicy{
+ Algorithm: "zstd",
+ MinSize: 4096,
+ MaxSize: 1048576,
+ })
+
+ assert.NotNil(t, result)
+ assert.Equal(t, "zstd", result.Algorithm)
+ assert.Equal(t, int64(4096), result.MinSize)
+ assert.Equal(t, int64(1048576), result.MaxSize)
+ })
+}
+
+func TestConvertTier1CompressionPolicy(t *testing.T) {
+ t.Run("nil tier1 returns nil", func(t *testing.T) {
+ assert.Nil(t, convertTier1CompressionPolicy(nil))
+ })
+
+ t.Run("tier1 with nil compression returns nil", func(t *testing.T) {
+ assert.Nil(t, convertTier1CompressionPolicy(&kliov1alpha1.Tier1PluginConfiguration{}))
+ })
+
+ t.Run("tier1 with compression", func(t *testing.T) {
+ result := convertTier1CompressionPolicy(&kliov1alpha1.Tier1PluginConfiguration{
+ Compression: &kliov1alpha1.CompressionPolicy{Algorithm: "gzip"},
+ })
+
+ assert.NotNil(t, result)
+ assert.Equal(t, "gzip", result.Algorithm)
+ })
+}
+
+func TestConvertTier2CompressionPolicy(t *testing.T) {
+ t.Run("nil tier2 returns nil", func(t *testing.T) {
+ assert.Nil(t, convertTier2CompressionPolicy(nil))
+ })
+
+ t.Run("tier2 with nil compression returns nil", func(t *testing.T) {
+ assert.Nil(t, convertTier2CompressionPolicy(&kliov1alpha1.Tier2PluginConfiguration{EnableBackup: true}))
+ })
+
+ t.Run("tier2 with compression", func(t *testing.T) {
+ result := convertTier2CompressionPolicy(&kliov1alpha1.Tier2PluginConfiguration{
+ EnableBackup: true,
+ Compression: &kliov1alpha1.CompressionPolicy{Algorithm: "s2-default"},
+ })
+
+ assert.NotNil(t, result)
+ assert.Equal(t, "s2-default", result.Algorithm)
+ })
+}
+
func TestConvertTier2RetentionPolicy(t *testing.T) {
t.Run("nil tier2 returns nil", func(t *testing.T) {
result := convertTier2RetentionPolicy(nil)
diff --git a/operator/pkg/config/client.go b/operator/pkg/config/client.go
index a6726d59..cbca13c0 100644
--- a/operator/pkg/config/client.go
+++ b/operator/pkg/config/client.go
@@ -38,6 +38,12 @@ type Data struct {
// Tier2RetentionPolicy is the retention policy to be applied to tier2.
Tier2RetentionPolicy *RetentionPolicy `json:"tier2_retention,omitempty" mapstructure:"tier2_retention"`
+ // Tier1CompressionPolicy is the compression policy to be applied to tier1.
+ Tier1CompressionPolicy *CompressionPolicy `json:"tier1_compression,omitempty" mapstructure:"tier1_compression"`
+
+ // Tier2CompressionPolicy is the compression policy to be applied to tier2.
+ Tier2CompressionPolicy *CompressionPolicy `json:"tier2_compression,omitempty" mapstructure:"tier2_compression"`
+
// Tier1Enabled records whether the client archives base backups and WAL
// to tier1. False on read-only clients, which do not write to tier1 but
// may still restore from it when Client.Wal.Address is set. Not consulted
diff --git a/operator/pkg/config/compression.go b/operator/pkg/config/compression.go
new file mode 100644
index 00000000..62119c55
--- /dev/null
+++ b/operator/pkg/config/compression.go
@@ -0,0 +1,90 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package config
+
+import (
+ "errors"
+ "fmt"
+)
+
+// CompressionPolicy configures the Kopia compression policy applied to base
+// backup data.
+type CompressionPolicy struct {
+ // Algorithm is the name of the Kopia compression algorithm to use.
+ // The special value "none" disables compression.
+ Algorithm string `json:"algorithm,omitempty" mapstructure:"algorithm"`
+
+ // MinSize is the minimum file size, in bytes, to attempt compression for.
+ // Files smaller than this are stored uncompressed. Zero means no minimum.
+ MinSize int64 `json:"min_size,omitempty" mapstructure:"min_size"`
+
+ // MaxSize is the maximum file size, in bytes, to attempt compression for.
+ // Files larger than this are stored uncompressed. Zero means no maximum.
+ MaxSize int64 `json:"max_size,omitempty" mapstructure:"max_size"`
+}
+
+// ErrInvalidCompressionAlgorithm is returned when an unsupported compression
+// algorithm is configured.
+var ErrInvalidCompressionAlgorithm = errors.New("invalid compression algorithm")
+
+// IsValidCompressionAlgorithm returns true when the passed algorithm is a
+// supported Kopia compression algorithm. The list matches the compressors
+// registered by Kopia, plus the special value "none" that explicitly disables
+// compression. It must be kept in sync with the CompressionAlgorithm enum in
+// the operator CRD types.
+func IsValidCompressionAlgorithm(algorithm string) bool {
+ switch algorithm {
+ case "none",
+ "deflate-best-compression",
+ "deflate-best-speed",
+ "deflate-default",
+ "gzip",
+ "gzip-best-compression",
+ "gzip-best-speed",
+ "pgzip",
+ "pgzip-best-compression",
+ "pgzip-best-speed",
+ "s2-better",
+ "s2-default",
+ "s2-parallel-4",
+ "s2-parallel-8",
+ "zstd",
+ "zstd-better-compression",
+ "zstd-fastest":
+ return true
+ default:
+ return false
+ }
+}
+
+// Validate checks that the configured compression algorithm is supported.
+// A nil policy or an empty algorithm is considered valid and means the Kopia
+// default (inherited) policy is left untouched.
+func (c *CompressionPolicy) Validate() error {
+ if c == nil || c.Algorithm == "" {
+ return nil
+ }
+
+ if !IsValidCompressionAlgorithm(c.Algorithm) {
+ return fmt.Errorf("%w: %q", ErrInvalidCompressionAlgorithm, c.Algorithm)
+ }
+
+ return nil
+}
diff --git a/operator/pkg/config/server.go b/operator/pkg/config/server.go
index 09b77188..2b688344 100644
--- a/operator/pkg/config/server.go
+++ b/operator/pkg/config/server.go
@@ -71,6 +71,11 @@ type Tier1Config struct {
// Wal is the configuration of the Wal server
Wal WalServerConfig `mapstructure:"wal"`
+
+ // Compression is the repository-wide (global) compression policy applied
+ // to base backups stored on tier1. When empty, the Kopia default (no
+ // compression) is left untouched.
+ Compression CompressionServerConfig `mapstructure:"compression"`
}
// Tier2Config is the configuration of tier 2.
@@ -98,10 +103,36 @@ type Tier2Config struct {
// CacheDirectory is the directory of the Kopia cache
CacheDirectory string `mapstructure:"cache"`
+ // Compression is the repository-wide (global) compression policy applied
+ // to base backups stored on tier2. When empty, the Kopia default (no
+ // compression) is left untouched.
+ Compression CompressionServerConfig `mapstructure:"compression"`
+
// S3 contains the configuration parameters for an S3-based tier 2
S3 S3Configuration `json:"s3" mapstructure:"s3"`
}
+// CompressionServerConfig is the repository-wide (global) compression policy
+// applied to a tier when the Kopia server starts.
+type CompressionServerConfig struct {
+ // Algorithm is the name of the Kopia compression algorithm to use.
+ // The special value "none" disables compression.
+ Algorithm string `mapstructure:"algorithm"`
+
+ // MinSize is the minimum file size, in bytes, to attempt compression for.
+ // Zero means no minimum.
+ MinSize int64 `mapstructure:"min_size"`
+
+ // MaxSize is the maximum file size, in bytes, to attempt compression for.
+ // Zero means no maximum.
+ MaxSize int64 `mapstructure:"max_size"`
+}
+
+// IsZero reports whether the compression policy carries no settings.
+func (c CompressionServerConfig) IsZero() bool {
+ return c.Algorithm == "" && c.MinSize == 0 && c.MaxSize == 0
+}
+
// BaseServerConfig is the configuration that will be used for
// the kopia server.
type BaseServerConfig struct {
diff --git a/operator/test/e2e/compression_test.go b/operator/test/e2e/compression_test.go
new file mode 100644
index 00000000..c5fdb92c
--- /dev/null
+++ b/operator/test/e2e/compression_test.go
@@ -0,0 +1,530 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package e2e
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "strings"
+ "testing"
+ "time"
+
+ cnpgv1 "github.com/cloudnative-pg/api/pkg/api/v1"
+ "github.com/stretchr/testify/require"
+ corev1 "k8s.io/api/core/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "sigs.k8s.io/e2e-framework/klient/k8s/resources"
+ "sigs.k8s.io/e2e-framework/klient/wait"
+ "sigs.k8s.io/e2e-framework/pkg/envconf"
+ "sigs.k8s.io/e2e-framework/pkg/types"
+
+ kliov1alpha1 "github.com/cloudnative-pg/klio/operator/api/v1alpha1"
+ "github.com/cloudnative-pg/klio/operator/test/klio/infra"
+ "github.com/cloudnative-pg/klio/operator/test/klio/testconfig"
+ machineryConditions "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/conditions"
+ "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/namespaces"
+ "github.com/cloudnative-pg/klio/operator/test/utils/templates/certificates"
+ "github.com/cloudnative-pg/klio/operator/test/utils/templates/cnpg"
+ "github.com/cloudnative-pg/klio/operator/test/utils/templates/klio"
+ "github.com/cloudnative-pg/klio/operator/test/utils/templates/rustfs"
+ "github.com/cloudnative-pg/klio/operator/test/utils/templates/secrets"
+)
+
+const (
+ // globalCompressionAlgorithm is the repository-wide compression policy set
+ // on the Server. Clusters without an override inherit it.
+ globalCompressionAlgorithm = "s2-default"
+
+ // clusterCompressionAlgorithm is the per-cluster compression policy set on
+ // the PluginConfiguration. It must differ from globalCompressionAlgorithm
+ // so we can prove the override takes precedence over the global policy.
+ clusterCompressionAlgorithm = "zstd"
+
+ // globalCompressionMinSize and clusterCompressionMinSize exercise the
+ // optional minSize bound on the Server (global) and the PluginConfiguration
+ // (per-cluster) respectively. They differ so the override is provable.
+ globalCompressionMinSize = 2048
+ clusterCompressionMinSize = 4096
+
+ // tier1KopiaConfigPattern and tier2RWKopiaConfigPattern glob the Kopia
+ // config files the server creates at startup, via their companion password
+ // files. tier1 is the local filesystem repository; tier2 is the read-write
+ // S3 repository.
+ tier1KopiaConfigPattern = "/tmp/kopiaconfig_tier1_*.kopia-password"
+ tier2RWKopiaConfigPattern = "/tmp/kopiaconfig_tier2_rw_*.kopia-password"
+
+ compressionServerContainerName = "server"
+ compressionKlioPodSuffix = "-klio-0"
+)
+
+// compressionScenario contains all resources needed for compression testing.
+// It reuses the tier2 infrastructure so that both the tier2 global policy
+// (Server) and the tier2 per-cluster policy (PluginConfiguration) can be
+// verified against the same Kopia repository.
+type compressionScenario struct {
+ namespace *corev1.Namespace
+
+ // tier2Infra bundles the shared RustFS + Klio Server (tier2) bring-up.
+ tier2Infra infra.Tier2
+
+ // klioServer is a shortcut to tier2Infra.KlioServer used by the verifier.
+ klioServer *kliov1alpha1.Server
+
+ // Source cluster
+ cnpgCluster *cnpgv1.Cluster
+ klioPluginConfiguration *kliov1alpha1.PluginConfiguration
+ backup *cnpgv1.Backup
+
+ name string
+}
+
+// Setup creates all resources for compression testing.
+func (s *compressionScenario) Setup(
+ ctx context.Context,
+ t *testing.T,
+ cfg *envconf.Config,
+) context.Context {
+ t.Helper()
+
+ t.Logf("Creating resources for compression feature: %s", s.name)
+ r, err := resources.New(cfg.Client().RESTConfig())
+ require.NoError(t, err, "failed to create resources client")
+
+ createNamespace(ctx, t, r, s.namespace)
+
+ // Bring up the shared RustFS + tier2 Klio Server infrastructure.
+ s.tier2Infra.ParallelSetup(ctx, t, r)
+
+ t.Logf("Deploying source CNPG cluster and plugin configuration...")
+ require.NoError(t, r.Create(ctx, s.klioPluginConfiguration),
+ "failed to create Klio plugin configuration")
+ require.NoError(t, r.Create(ctx, s.cnpgCluster), "failed to create CNPG source cluster")
+
+ require.NoError(t, wait.For(
+ machineryConditions.ClusterIsReady(r, s.cnpgCluster),
+ wait.WithTimeout(4*time.Minute),
+ wait.WithInterval(10*time.Second),
+ ), "source cluster not ready")
+
+ t.Logf("All resources ready for compression feature: %s", s.name)
+
+ return ctx
+}
+
+// Teardown deletes all resources.
+func (s *compressionScenario) Teardown(
+ ctx context.Context,
+ t *testing.T,
+ cfg *envconf.Config,
+) context.Context {
+ t.Helper()
+
+ t.Logf("Tearing down resources for compression feature: %s", s.name)
+ r, err := resources.New(cfg.Client().RESTConfig())
+ require.NoError(t, err, "failed to create resources client")
+ namespaces.DumpNamespaceOnFailure(ctx, t, r, testCfg.LogDir, s.namespace.Name, testconfig.DumpedKinds())
+ require.NoError(t, r.Delete(ctx, s.namespace), "failed to delete namespace")
+ t.Logf("Resources torn down for compression feature: %s", s.name)
+
+ return ctx
+}
+
+// kopiaConfigFile discovers an ephemeral Kopia config file the server created
+// at startup, by globbing its companion password file with the passed pattern.
+func (s *compressionScenario) kopiaConfigFile(
+ ctx context.Context,
+ r *resources.Resources,
+ pattern string,
+) (string, error) {
+ podName := s.klioServer.Name + compressionKlioPodSuffix
+
+ var stdout, stderr bytes.Buffer
+ findCmd := []string{"sh", "-c", "ls " + pattern + " 2>/dev/null"}
+ if err := r.ExecInPod(
+ ctx, s.namespace.Name, podName, compressionServerContainerName, findCmd, &stdout, &stderr,
+ ); err != nil {
+ return "", fmt.Errorf("failed to locate kopia config %q: %w; stderr: %s", pattern, err, stderr.String())
+ }
+
+ passwordFile := strings.TrimSpace(stdout.String())
+ if passwordFile == "" {
+ return "", fmt.Errorf("no kopia config file found matching %q", pattern)
+ }
+
+ return strings.TrimSuffix(passwordFile, ".kopia-password"), nil
+}
+
+// compressionOfPolicy runs `kopia policy show --json` against the
+// passed config and returns the effective compression settings.
+func (s *compressionScenario) compressionOfPolicy(
+ ctx context.Context,
+ r *resources.Resources,
+ configFile string,
+ target string,
+) (effectiveCompression, error) {
+ podName := s.klioServer.Name + compressionKlioPodSuffix
+
+ var stdout, stderr bytes.Buffer
+ showCmd := []string{
+ "kopia", "policy", "show", target,
+ "--disable-file-logging",
+ "--config-file=" + configFile,
+ "--json",
+ }
+ if err := r.ExecInPod(
+ ctx, s.namespace.Name, podName, compressionServerContainerName, showCmd, &stdout, &stderr,
+ ); err != nil {
+ return effectiveCompression{},
+ fmt.Errorf("failed to show kopia policy for %q: %w; stderr: %s", target, err, stderr.String())
+ }
+
+ var policy struct {
+ Compression effectiveCompression `json:"compression"`
+ }
+ if err := json.Unmarshal(stdout.Bytes(), &policy); err != nil {
+ return effectiveCompression{}, fmt.Errorf("failed to parse kopia policy %q: %w", stdout.String(), err)
+ }
+
+ return policy.Compression, nil
+}
+
+// effectiveCompression is the subset of a Kopia policy's compression settings
+// that the test asserts on.
+type effectiveCompression struct {
+ CompressorName string `json:"compressorName"`
+ MinSize int64 `json:"minSize"`
+}
+
+// clusterPolicyTarget finds the "user@host" policy target for the cluster in
+// the tier2 repository. The per-cluster policy only exists once the backup has
+// been relayed to tier2, so this returns an empty string until then.
+func (s *compressionScenario) clusterPolicyTarget(
+ ctx context.Context,
+ r *resources.Resources,
+ configFile string,
+) (string, error) {
+ podName := s.klioServer.Name + compressionKlioPodSuffix
+
+ var stdout, stderr bytes.Buffer
+ listCmd := []string{
+ "kopia", "policy", "list",
+ "--disable-file-logging",
+ "--config-file=" + configFile,
+ "--json",
+ }
+ if err := r.ExecInPod(
+ ctx, s.namespace.Name, podName, compressionServerContainerName, listCmd, &stdout, &stderr,
+ ); err != nil {
+ return "", fmt.Errorf("failed to list kopia policies: %w; stderr: %s", err, stderr.String())
+ }
+
+ var policies []struct {
+ Target struct {
+ Host string `json:"host"`
+ User string `json:"userName"`
+ } `json:"target"`
+ }
+ if err := json.Unmarshal(stdout.Bytes(), &policies); err != nil {
+ return "", fmt.Errorf("failed to parse kopia policy list %q: %w", stdout.String(), err)
+ }
+
+ for _, p := range policies {
+ if p.Target.Host == s.cnpgCluster.Name && p.Target.User != "" {
+ return p.Target.User + "@" + p.Target.Host, nil
+ }
+ }
+
+ return "", nil
+}
+
+// CompressionFeature verifies the global and per-cluster compression policies.
+type CompressionFeature struct {
+ name string
+ scenario *compressionScenario
+}
+
+// Name returns the name of the feature.
+func (f *CompressionFeature) Name() string {
+ return f.name
+}
+
+// Setup initializes the test resources.
+func (f *CompressionFeature) Setup() types.StepFunc {
+ return f.scenario.Setup
+}
+
+// Run verifies that the repository-wide compression policy configured on the
+// Server is applied globally, and that the per-cluster policy configured on the
+// PluginConfiguration overrides it for the cluster's own source. Both tier1 and
+// tier2 repositories are checked.
+func (f *CompressionFeature) Run() types.StepFunc {
+ return func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
+ t.Helper()
+ t.Log("Running compression policy test")
+
+ // The per-cluster policies must differ from the global one, otherwise
+ // the override assertions below would pass vacuously.
+ require.NotEqual(t, globalCompressionAlgorithm, clusterCompressionAlgorithm,
+ "test misconfigured: global and per-cluster algorithms must differ")
+
+ r, err := resources.New(cfg.Client().RESTConfig())
+ require.NoError(t, err, "failed to create resources client")
+
+ // A backup is required so the base data reaches both tiers, which is
+ // when the per-cluster compression policies are applied (tier1 by
+ // `klio backup run`, tier2 by the consumer during the relay).
+ t.Log("Creating backup with tier2 enabled...")
+ require.NoError(t, r.Create(ctx, f.scenario.backup), "failed to create backup")
+ err = wait.For(
+ machineryConditions.BackupIsCompleted(r, f.scenario.backup),
+ wait.WithTimeout(3*time.Minute),
+ wait.WithInterval(10*time.Second),
+ )
+ require.NoError(t, err, "backup not completed")
+
+ f.scenario.verifyTierCompression(ctx, t, r, "tier1", tier1KopiaConfigPattern)
+ f.scenario.verifyTierCompression(ctx, t, r, "tier2", tier2RWKopiaConfigPattern)
+
+ return ctx
+ }
+}
+
+// verifyTierCompression asserts that, for the repository selected by
+// configPattern, the global compression policy matches the Server setting and
+// the cluster's own source policy matches the per-cluster override.
+func (s *compressionScenario) verifyTierCompression(
+ ctx context.Context,
+ t *testing.T,
+ r *resources.Resources,
+ tier string,
+ configPattern string,
+) {
+ t.Helper()
+
+ configFile, err := s.kopiaConfigFile(ctx, r, configPattern)
+ require.NoError(t, err, "[%s] failed to discover kopia config file", tier)
+
+ // The global policy is applied when the server starts, so it is already
+ // present. Verify its algorithm and minSize match what the Server requested.
+ t.Logf("[%s] verifying the global compression policy...", tier)
+ globalCompression, err := s.compressionOfPolicy(ctx, r, configFile, "--global")
+ require.NoError(t, err, "[%s] failed to read the global compression policy", tier)
+ require.Equal(t, globalCompressionAlgorithm, globalCompression.CompressorName,
+ "[%s] unexpected global compression algorithm", tier)
+ require.Equal(t, int64(globalCompressionMinSize), globalCompression.MinSize,
+ "[%s] unexpected global compression minSize", tier)
+
+ // The per-cluster policy is applied during the backup, so poll until it
+ // appears with the expected algorithm and minSize.
+ t.Logf("[%s] waiting for the per-cluster compression policy...", tier)
+ var clusterCompression effectiveCompression
+ err = wait.For(
+ func(ctx context.Context) (bool, error) {
+ target, err := s.clusterPolicyTarget(ctx, r, configFile)
+ if err != nil || target == "" {
+ return false, err
+ }
+ clusterCompression, err = s.compressionOfPolicy(ctx, r, configFile, target)
+ if err != nil {
+ return false, err
+ }
+
+ return clusterCompression.CompressorName == clusterCompressionAlgorithm &&
+ clusterCompression.MinSize == clusterCompressionMinSize, nil
+ },
+ wait.WithTimeout(3*time.Minute),
+ wait.WithInterval(10*time.Second),
+ )
+ require.NoError(t, err,
+ "[%s] per-cluster compression policy did not become %q/minSize=%d (last seen %+v)",
+ tier, clusterCompressionAlgorithm, clusterCompressionMinSize, clusterCompression)
+
+ t.Logf("[%s] compression policies verified: global=%+v, cluster=%+v",
+ tier, globalCompression, clusterCompression)
+}
+
+// Teardown cleans up resources after the test.
+func (f *CompressionFeature) Teardown() types.StepFunc {
+ return f.scenario.Teardown
+}
+
+// newCompressionScenario creates a new compression test scenario.
+func newCompressionScenario(name string, namespace string) *compressionScenario {
+ const (
+ cnpgClusterName = "pg-compression"
+
+ klioServerName = "klio"
+
+ selfSignedIssuerName = "selfsigned-issuer"
+ caCertificateName = klioServerName + "-ca"
+ caIssuerName = caCertificateName + "-issuer"
+ serverCertificateName = klioServerName + "-server"
+ cnpgClientCertName = cnpgClusterName + "-client"
+
+ rustfsName = "rustfs"
+ rustfsSecretName = rustfsName + "-secret"
+ rustfsConfigMapName = rustfsName + "-config"
+ rustfsCreateBucketJobName = rustfsName
+
+ encryptionSecretName = "encryption"
+ encryptionPassword = "testencryptionpassword123"
+
+ pluginConfigurationName = "klio-plugin-configuration"
+
+ backupName = "test-backup"
+
+ s3Prefix = "tier2"
+ )
+
+ namespaceObj := &corev1.Namespace{
+ ObjectMeta: metav1.ObjectMeta{Name: namespace},
+ }
+
+ issuer := certificates.GetSelfSignedIssuerObject(selfSignedIssuerName, namespace)
+
+ rustfsSecret := rustfs.GetRustFSSecret(rustfsSecretName, namespace)
+ rustfsConfigMap := rustfs.GetRustFSConfigMap(rustfsConfigMapName, namespace)
+ rustfsCertificate := rustfs.GetRustFSCertificate(rustfsName, namespace, issuer)
+ rustfsService := rustfs.GetRustFSService(rustfsName, namespace)
+ rustfsDeployment := rustfs.GetRustFSDeployment(rustfsName, namespace)
+ rustfsCreateBucketJob := rustfs.GetRustFSCreateBucketJob(
+ rustfsCreateBucketJobName, namespace, rustfs.RustFSBucketName)
+
+ caCertificate := certificates.GetCACertificateObject(caCertificateName, namespace, issuer)
+ caIssuer := certificates.GetCAIssuerObject(caIssuerName, namespace, caCertificate.Spec.SecretName)
+ serverCertificate := certificates.GetCertificateObject(serverCertificateName, namespace, []string{klioServerName},
+ issuer)
+ userCertificate := certificates.GetUserCertificateObject(
+ cnpgClientCertName, namespace, cnpgClientCertName+"@"+cnpgClusterName, caIssuer)
+
+ ageSecrets := secrets.GetKlioAgeEncryptionSecrets(encryptionSecretName, namespace, encryptionPassword)
+
+ klioServer := klio.GetServerWithTier2Object(
+ klioServerName,
+ namespace,
+ klio.ServerWithTier2TemplateOptions{
+ ServerTemplateOptions: klio.ServerTemplateOptions{
+ Image: testCfg.ServerImage,
+ StorageClass: testCfg.StorageClass,
+ TLSSecretName: serverCertificate.Spec.SecretName,
+ ClientCASecretName: caCertificate.Spec.SecretName,
+ Encryption: klio.EncryptionOptions{
+ EncryptionKeySecretName: ageSecrets.EncryptionKeySecret.Name,
+ EncryptionKeyFileName: "encryption-key.age",
+ IdentitySecretName: ageSecrets.IdentitySecret.Name,
+ IdentityFileName: "identity.txt",
+ },
+ },
+ Tier2Encryption: klio.EncryptionOptions{
+ EncryptionKeySecretName: ageSecrets.EncryptionKeySecret.Name,
+ EncryptionKeyFileName: "encryption-key.age",
+ IdentitySecretName: ageSecrets.IdentitySecret.Name,
+ IdentityFileName: "identity.txt",
+ },
+ S3: klio.Tier2S3Options{
+ S3BucketName: rustfs.RustFSBucketName,
+ S3Prefix: s3Prefix,
+ S3Endpoint: rustfs.GetRustFSEndpoint(rustfsName, namespace),
+ S3Region: rustfs.RustFSRegion,
+ S3AccessKeySecretName: rustfsSecret.Name,
+ S3SecretKeySecretName: rustfsSecret.Name,
+ S3CABundleSecretName: rustfsCertificate.Spec.SecretName,
+ },
+ },
+ )
+ // Repository-wide (global) compression policy on the Server: every cluster
+ // that does not override it inherits this policy.
+ klioServer.Spec.Tier1.Compression = &kliov1alpha1.CompressionPolicy{
+ Algorithm: globalCompressionAlgorithm,
+ MinSize: globalCompressionMinSize,
+ }
+ klioServer.Spec.Tier2.Compression = &kliov1alpha1.CompressionPolicy{
+ Algorithm: globalCompressionAlgorithm,
+ MinSize: globalCompressionMinSize,
+ }
+
+ cnpgCluster := cnpg.GetCnpgClusterObject(
+ cnpgClusterName, namespace, 1, pluginConfigurationName,
+ cnpg.ClusterTemplateOptions{StorageClass: testCfg.StorageClass})
+
+ klioPluginConfiguration := klio.GetPluginConfigurationObject(
+ pluginConfigurationName,
+ namespace,
+ klio.PluginConfigurationTemplateOptions{
+ ServerCertificate: serverCertificate,
+ ClientCertificate: userCertificate,
+ ClusterName: cnpgClusterName,
+ EnableTier2Backup: true,
+ EnableTier2Recovery: false,
+ Mode: kliov1alpha1.ModeStandard,
+ },
+ )
+ // Per-cluster compression policy overriding the Server global policy for
+ // this cluster's own source, on both tiers.
+ klioPluginConfiguration.Spec.Tier1 = &kliov1alpha1.Tier1PluginConfiguration{
+ Compression: &kliov1alpha1.CompressionPolicy{
+ Algorithm: clusterCompressionAlgorithm,
+ MinSize: clusterCompressionMinSize,
+ },
+ }
+ klioPluginConfiguration.Spec.Tier2.Compression = &kliov1alpha1.CompressionPolicy{
+ Algorithm: clusterCompressionAlgorithm,
+ MinSize: clusterCompressionMinSize,
+ }
+
+ backup := cnpg.GetCnpgBackupObject(backupName, namespace, cnpgv1.BackupTargetPrimary, cnpgCluster)
+
+ return &compressionScenario{
+ namespace: namespaceObj,
+ tier2Infra: infra.Tier2{
+ Issuer: issuer,
+ RustfsSecret: rustfsSecret,
+ RustfsConfigMap: rustfsConfigMap,
+ RustfsCertificate: rustfsCertificate,
+ RustfsService: rustfsService,
+ RustfsDeployment: rustfsDeployment,
+ RustfsCreateBucketJob: rustfsCreateBucketJob,
+ ServerCertificate: serverCertificate,
+ CaCertificate: caCertificate,
+ CaIssuer: caIssuer,
+ UserCertificate: userCertificate,
+ EncryptionSecret: ageSecrets.EncryptionKeySecret,
+ IdentitySecret: ageSecrets.IdentitySecret,
+ KlioServer: klioServer,
+ },
+ klioServer: klioServer,
+ cnpgCluster: cnpgCluster,
+ klioPluginConfiguration: klioPluginConfiguration,
+ backup: backup,
+ name: name,
+ }
+}
+
+// Compression returns a Feature that verifies the global and per-cluster
+// Kopia compression policies are applied and that the per-cluster policy
+// overrides the global one.
+func Compression(namespace string) *CompressionFeature {
+ return &CompressionFeature{
+ name: "Compression",
+ scenario: newCompressionScenario("Compression", namespace),
+ }
+}
diff --git a/operator/test/e2e/main_test.go b/operator/test/e2e/main_test.go
index b0b41e72..b8492cac 100644
--- a/operator/test/e2e/main_test.go
+++ b/operator/test/e2e/main_test.go
@@ -60,6 +60,7 @@ func TestMain(m *testing.M) {
runner.RegisterFeature(RecoverClusterFromTier2Pitr(envconf.RandomName("recovery-from-tier2-pitr", 32)))
runner.RegisterFeature(PluginConfigurationUpdate(envconf.RandomName("plugin-config-update", 32)))
runner.RegisterFeature(Tier2Retention(envconf.RandomName("tier2-retention", 32)))
+ runner.RegisterFeature(Compression(envconf.RandomName("compression", 32)))
runner.RegisterFeature(WALRetentionQueueAwareness(envconf.RandomName("wal-retention-queue", 32)))
runner.RegisterFeature(ServerTierReconfiguration(envconf.RandomName("server-reconfig", 32)))
runner.RegisterFeature(PVCResize(envconf.RandomName("pvc-resize", 32)))