From 5f9adbee4825bc6c089e7656f600bef59f886636 Mon Sep 17 00:00:00 2001 From: Gabriele Quaresima Date: Wed, 26 Aug 2026 17:15:43 +0200 Subject: [PATCH] feat(compression): allow configuring the Kopia compression policy Kopia repositories were created without a compression policy, so base backup data was stored uncompressed. Allow users to configure a compression policy repository-wide on the Server (applied to the global Kopia policy at server start) and per-cluster on the PluginConfiguration (applied to the cluster's own source, overriding the global policy). Both tier1 and tier2 are covered: tier1 is set through the Kopia server during the backup, while tier2 travels over gRPC and is applied by the backup consumer before the relay. The policy exposes the compression algorithm plus the optional minSize and maxSize bounds (in bytes) that restrict which files are compressed. Kopia's only-compress and never-compress extension lists are left out on purpose: they select files by extension, which does not fit a PostgreSQL data directory (relation files are numeric and carry no meaningful extension), and WAL files are compressed on a separate path regardless. Closes #126 Assisted-by: Claude Signed-off-by: Gabriele Quaresima --- core/cmd/backup/run.go | 127 ++++- core/cmd/server/server.go | 59 +- core/internal/client/klioclient/interfaces.go | 3 + .../client/klioclient/kopia/multiconnect.go | 13 + .../client/klioclient/kopia/retention.go | 5 + core/internal/consumer/backup.go | 15 + core/internal/grpc/klio_wal.pb.go | 19 +- core/internal/kopia/data.go | 20 + core/internal/kopia/policy.go | 74 +++ core/internal/kopia/policy_test.go | 90 +++ core/internal/queue/backup.go | 3 + core/internal/server/walserver/backup.go | 17 +- core/pkg/config/client.go | 6 + core/pkg/config/client_validate.go | 8 + core/pkg/config/compression.go | 90 +++ core/pkg/config/compression_test.go | 118 ++++ core/pkg/config/server.go | 31 + core/pkg/config/server_validate.go | 10 + core/pkg/config/server_validate_test.go | 50 ++ core/proto/klio_wal.proto | 3 + documentation/.wordlist.txt | 12 + documentation/web/docs/developer/_protocol.md | 1 + .../web/docs/developer/kopia_internals.md | 3 + .../web/docs/developer/running-e2e-tests.md | 5 + documentation/web/docs/user/api/_klio_api.md | 41 ++ documentation/web/docs/user/klio_server.md | 36 ++ .../web/docs/user/plugin_configuration.md | 63 +++ .../v1alpha1/plugin_configuration_types.go | 37 ++ operator/api/v1alpha1/server_types.go | 12 + .../api/v1alpha1/zz_generated.deepcopy.go | 35 ++ .../klio.cnpg.io_pluginconfigurations.yaml | 90 +++ .../crd/bases/klio.cnpg.io_servers.yaml | 90 +++ .../chart/crds/pluginconfiguration-crd.yaml | 90 +++ operator/dist/chart/crds/server-crd.yaml | 90 +++ .../internal/controller/server_envbuilder.go | 34 ++ .../controller/server_envbuilder_test.go | 100 ++++ operator/internal/klioconfig/config.go | 36 +- operator/internal/klioconfig/config_test.go | 83 +++ operator/pkg/config/client.go | 6 + operator/pkg/config/compression.go | 90 +++ operator/pkg/config/server.go | 31 + operator/test/e2e/compression_test.go | 530 ++++++++++++++++++ operator/test/e2e/main_test.go | 1 + 43 files changed, 2236 insertions(+), 41 deletions(-) create mode 100644 core/internal/kopia/policy_test.go create mode 100644 core/pkg/config/compression.go create mode 100644 core/pkg/config/compression_test.go create mode 100644 operator/pkg/config/compression.go create mode 100644 operator/test/e2e/compression_test.go 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)))