From c35bccf680b7755fa626a9dd7aa60702489e02fb Mon Sep 17 00:00:00 2001 From: Armando Ruocco Date: Tue, 29 Sep 2026 11:59:19 +0200 Subject: [PATCH 1/2] fix(sidecar)!: serve backups while the cluster bootstraps A cluster bootstrapping from a Klio backup received a dedicated `klio-restore` sidecar that only advertised the restore and WAL capabilities, later swapped for `klio-plugin` by a rollout once the primary was elected. CloudNativePG dispatches a backup as soon as the instance is ready, which lands in that window: the plugin client rejects it with "plugin does not support Backup service". A replica cluster created together with an immediate ScheduledBackup hit this every time. Inject the same `klio-plugin` sidecar in every phase and give the `cnpgi instance` server the restore capability too. The restore hook and the WAL service already pick their repository per request from the cluster definition, so nothing needs swapping once recovery completes, and the pod spec no longer changes after bootstrap. The `cnpgi restore` command is removed, and the `PluginConfiguration` CRD now accepts only `klio-plugin` in `spec.containers`: a leftover `klio-restore` entry must be removed before upgrading, as documented in the upgrade notes. Since one sidecar now reads the recovery source and replica source configurations from sibling files of the projected klio-config mount, the config watcher hashes the whole mount directory rather than only the file behind --config, and runs even when no --config is given, so a rotation of any of those Secrets restarts the sidecar. The mount path is defined once per module as KlioConfigDir. Also simplify the operator's selection of the PluginConfiguration the sidecar is customized from, and update the plugin configuration and OpenTelemetry docs for the single customizable container. The replica-cluster e2e now requests a backup right after creating the cluster, before its bootstrap completes. BREAKING CHANGE: the PluginConfiguration CRD no longer accepts a `klio-restore` entry in `spec.containers`. Closes #268 Co-authored-by: Gabriele Quaresima Co-authored-by: Francesco Canovai Assisted-by: Claude Assisted-by: OpenCode (Muse Spark 1.3 Free) Signed-off-by: Armando Ruocco Signed-off-by: Gabriele Quaresima Signed-off-by: Francesco Canovai --- core/cmd/cnpgi/common.go | 21 +- core/cmd/cnpgi/instance.go | 12 + core/cmd/cnpgi/restore.go | 99 ------- core/internal/cnpgi/configwatcher.go | 64 ++++- core/internal/cnpgi/configwatcher_test.go | 110 +++++++- core/internal/cnpgi/restore.go | 3 +- core/internal/cnpgi/start.go | 11 + core/internal/cnpgi/wal.go | 9 +- documentation/web/docs/user/api/_klio_api.md | 2 +- documentation/web/docs/user/opentelemetry.md | 14 +- .../web/docs/user/plugin_configuration.md | 67 ++--- documentation/web/docs/user/upgrade_notes.md | 11 + .../v1alpha1/plugin_configuration_types.go | 6 +- .../klio.cnpg.io_pluginconfigurations.yaml | 8 +- .../klio-operator.clusterserviceversion.yaml | 4 +- .../single/cluster/plugin_configuration.yaml | 7 - .../chart/crds/pluginconfiguration-crd.yaml | 8 +- operator/internal/cnpgi/lifecycle.go | 184 +++++++------ operator/internal/cnpgi/lifecycle_test.go | 253 +++++++++++++++++- operator/internal/klioconfig/config.go | 6 + .../e2e/backup_from_replica_cluster_test.go | 34 ++- 21 files changed, 621 insertions(+), 312 deletions(-) delete mode 100644 core/cmd/cnpgi/restore.go diff --git a/core/cmd/cnpgi/common.go b/core/cmd/cnpgi/common.go index f3ae78a99..4d3259610 100644 --- a/core/cmd/cnpgi/common.go +++ b/core/cmd/cnpgi/common.go @@ -25,6 +25,7 @@ import ( "fmt" "os" "os/signal" + "path/filepath" "syscall" "time" @@ -113,13 +114,21 @@ func runCNPGI( } } - // Add config file watcher so the sidecar restarts when the config changes + // Watch the whole config mount, not only the file behind --config: the + // restore and WAL services read the recovery source and replica source + // configurations from sibling files, and a rotated Secret must restart + // the sidecar for those too. Always watch the directory, even when no + // --config was passed (archive-less bootstrap or replica-source primary): + // those sidecars only consume sibling files. + watchDir := cnpgi.KlioConfigDir if configFile != "" { - if err := mgr.Add( - cnpgi.NewConfigFileWatcher(configFile, 10*time.Second), - ); err != nil { - return fmt.Errorf("while adding config watcher: %w", err) - } + watchDir = filepath.Dir(configFile) + } + + if err := mgr.Add( + cnpgi.NewConfigFileWatcher(watchDir, 10*time.Second), + ); err != nil { + return fmt.Errorf("while adding config watcher: %w", err) } // Start the manager and handle graceful shutdown diff --git a/core/cmd/cnpgi/instance.go b/core/cmd/cnpgi/instance.go index 57d4aefd1..4be079a97 100644 --- a/core/cmd/cnpgi/instance.go +++ b/core/cmd/cnpgi/instance.go @@ -55,8 +55,15 @@ var instanceCmd = &cobra.Command{ podName, _ := cmd.Flags().GetString("pod-name") clusterName, _ := cmd.Flags().GetString("cluster-name") clusterNamespace, _ := cmd.Flags().GetString("cluster-namespace") + pgData, _ := cmd.Flags().GetString("pgdata") + // One sidecar serves every phase of the instance: the restore hooks + // while the cluster bootstraps from a Klio backup, backup and WAL + // archiving afterwards. Each service picks its repository per request + // from the cluster definition, so nothing has to be swapped once the + // recovery completes, and a backup requested meanwhile is served. capabilities := func(server *cnpgi.CNPGI) { + server.AddRestoreCapability(pgData) server.AddBackupCapability(cnpgi.BackupCapabilityOptions{ Tier2: configuration.Tier2BackupEnabled, }) @@ -115,6 +122,11 @@ func init() { "", "The name of the current instance", ) + instanceCmd.Flags().String( + "pgdata", + "/var/lib/postgresql/data/pgdata", + "The PGDATA directory a restore is unpacked into", + ) CnpgiCmd.AddCommand(instanceCmd) } diff --git a/core/cmd/cnpgi/restore.go b/core/cmd/cnpgi/restore.go deleted file mode 100644 index db9a7287f..000000000 --- a/core/cmd/cnpgi/restore.go +++ /dev/null @@ -1,99 +0,0 @@ -/* -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 cnpgi - -import ( - "fmt" - - "github.com/cloudnative-pg/machinery/pkg/log" - "github.com/spf13/cobra" - "github.com/spf13/viper" - "k8s.io/apimachinery/pkg/types" - - "github.com/cloudnative-pg/klio/core/internal/cnpgi" - "github.com/cloudnative-pg/klio/core/pkg/config" -) - -// restoreJobCmd represents the job run command -// -//nolint:gochecknoglobals -var restoreJobCmd = &cobra.Command{ - Use: "restore [destination]", - Short: "Start the instance CNPG-I job restore server", - Hidden: true, - Args: cobra.ExactArgs(1), - RunE: func(cmd *cobra.Command, args []string) error { - configFile, _ := cmd.Root().PersistentFlags().GetString("config") - pluginPath, _ := cmd.Flags().GetString("plugin-path") - clusterName, _ := cmd.Flags().GetString("cluster-name") - clusterNamespace, _ := cmd.Flags().GetString("cluster-namespace") - debug, _ := cmd.PersistentFlags().GetBool("debug") - destination := args[0] - - var configuration config.Data - if err := viper.Unmarshal(&configuration); err != nil { - return fmt.Errorf("could not unmarshal configuration: %w", err) - } - - contextLogger := log.FromContext(cmd.Context()) - contextLogger.Info("Starting CNPG-I job restore server", - "pluginPath", pluginPath, - "destination", args[0], - ) - capabilities := func(server *cnpgi.CNPGI) { - server.AddRestoreCapability(destination) - server.AddWALCapability(cnpgi.WALCapabilityOptions{ - Debug: debug, - }) - } - - return runCNPGI( - cmd.Context(), - pluginPath, - configFile, - types.NamespacedName{ - Namespace: clusterNamespace, - Name: clusterName, - }, - capabilities, - nil, - ) - }, -} - -//nolint:gochecknoinits -func init() { - restoreJobCmd.Flags().String( - "cluster-name", - "", - "The name of the cluster object", - ) - restoreJobCmd.Flags().String( - "cluster-namespace", - "", - "The namespace of the cluster object", - ) - restoreJobCmd.Flags().String( - "plugin-path", - "/plugins", - "The directory where the Unix domain socket should be created", - ) - CnpgiCmd.AddCommand(restoreJobCmd) -} diff --git a/core/internal/cnpgi/configwatcher.go b/core/internal/cnpgi/configwatcher.go index f78950329..a728d39d4 100644 --- a/core/internal/cnpgi/configwatcher.go +++ b/core/internal/cnpgi/configwatcher.go @@ -26,6 +26,7 @@ import ( "errors" "fmt" "os" + "path/filepath" "time" "github.com/cloudnative-pg/machinery/pkg/log" @@ -35,25 +36,26 @@ import ( // ErrConfigFileChanged is returned when the config file changes on disk. var ErrConfigFileChanged = errors.New("config file changed, restarting") -// NewConfigFileWatcher creates a manager.RunnableFunc that polls a config file -// and returns an error when its content changes. This causes the manager to -// shut down, and the kubelet will restart the container. +// NewConfigFileWatcher creates a manager.RunnableFunc that polls a config file, +// or every file in a config directory, and returns an error when the content +// changes. This causes the manager to shut down, and the kubelet will restart +// the container. // // Polling is used instead of fsnotify because Kubernetes updates secret // volumes via symlink swaps, which fsnotify may not detect reliably. func NewConfigFileWatcher( - configFile string, + configPath string, interval time.Duration, ) manager.RunnableFunc { return func(ctx context.Context) error { logger := log.FromContext(ctx).WithName("config-watcher") - initialHash, err := hashFile(configFile) + initialHash, err := hashPath(configPath) if err != nil { - return fmt.Errorf("while reading initial config file: %w", err) + return fmt.Errorf("while reading initial config: %w", err) } - logger.Info("Config file watcher started", "file", configFile) + logger.Info("Config watcher started", "path", configPath) ticker := time.NewTicker(interval) defer ticker.Stop() @@ -63,15 +65,15 @@ func NewConfigFileWatcher( case <-ctx.Done(): return nil case <-ticker.C: - currentHash, err := hashFile(configFile) + currentHash, err := hashPath(configPath) if err != nil { - logger.Error(err, "Failed to read config file, will retry") + logger.Error(err, "Failed to read config, will retry") continue } if currentHash != initialHash { - logger.Info("Config file changed, shutting down for restart", - "file", configFile) + logger.Info("Config changed, shutting down for restart", + "path", configPath) return ErrConfigFileChanged } } @@ -79,6 +81,46 @@ func NewConfigFileWatcher( } } +// hashPath hashes a file, or every regular file directly inside a directory +// together with its name. Directory entries are resolved through symlinks so +// the projected Secret volume layout (`` linking into `..data/`) is +// followed, while its internal directories are skipped. +func hashPath(path string) (string, error) { + info, err := os.Stat(path) + if err != nil { + return "", fmt.Errorf("while reading path %q: %w", path, err) + } + if !info.IsDir() { + return hashFile(path) + } + + entries, err := os.ReadDir(path) + if err != nil { + return "", fmt.Errorf("while reading directory %q: %w", path, err) + } + + hash := sha256.New() + for _, entry := range entries { + entryPath := filepath.Join(path, entry.Name()) + entryInfo, err := os.Stat(entryPath) + if err != nil { + return "", fmt.Errorf("while reading path %q: %w", entryPath, err) + } + if entryInfo.IsDir() { + continue + } + + fileHash, err := hashFile(entryPath) + if err != nil { + return "", err + } + hash.Write([]byte(entry.Name())) + hash.Write([]byte(fileHash)) + } + + return hex.EncodeToString(hash.Sum(nil)), nil +} + // hashFile reads a file and returns its SHA256 hash as a hex string. func hashFile(path string) (string, error) { data, err := os.ReadFile(path) //nolint:gosec diff --git a/core/internal/cnpgi/configwatcher_test.go b/core/internal/cnpgi/configwatcher_test.go index 5e6fab13d..af56bec42 100644 --- a/core/internal/cnpgi/configwatcher_test.go +++ b/core/internal/cnpgi/configwatcher_test.go @@ -100,7 +100,7 @@ func TestConfigFileWatcherInitialReadFailure(t *testing.T) { err := watcher(ctx) require.Error(t, err) - assert.Contains(t, err.Error(), "while reading initial config file") + assert.Contains(t, err.Error(), "while reading initial config") } func TestConfigFileWatcherTransientReadError(t *testing.T) { @@ -181,3 +181,111 @@ func TestHashFile(t *testing.T) { assert.NotEmpty(t, hash) // SHA256 of empty string is a valid hash }) } + +// writeProjectedSecret lays out a directory the way the kubelet does for a +// projected Secret volume: a timestamped data directory, a `..data` symlink +// pointing at it, and one symlink per key at the top level. +func writeProjectedSecret(t *testing.T, dir string, files map[string]string) { + t.Helper() + + dataDir := filepath.Join(dir, "..2026_09_16_00_00_00.000000000") + require.NoError(t, os.MkdirAll(dataDir, 0o700)) + for name, content := range files { + require.NoError(t, os.WriteFile(filepath.Join(dataDir, name), []byte(content), 0o600)) + _ = os.Remove(filepath.Join(dir, name)) + require.NoError(t, os.Symlink(filepath.Join("..data", name), filepath.Join(dir, name))) + } + _ = os.Remove(filepath.Join(dir, "..data")) + require.NoError(t, os.Symlink(filepath.Base(dataDir), filepath.Join(dir, "..data"))) +} + +func TestHashPathDirectory(t *testing.T) { + t.Run("follows the projected secret layout", func(t *testing.T) { + dir := t.TempDir() + writeProjectedSecret(t, dir, map[string]string{ + "klio-archive": "archive", + "source": "recovery source", + }) + + hash, err := hashPath(dir) + require.NoError(t, err) + assert.NotEmpty(t, hash) + }) + + t.Run("changes when a sibling file changes", func(t *testing.T) { + dir := t.TempDir() + writeProjectedSecret(t, dir, map[string]string{ + "klio-archive": "archive", + "source": "recovery source", + }) + before, err := hashPath(dir) + require.NoError(t, err) + + writeProjectedSecret(t, dir, map[string]string{ + "klio-archive": "archive", + "source": "rotated recovery source", + }) + after, err := hashPath(dir) + require.NoError(t, err) + + assert.NotEqual(t, before, after) + }) + + t.Run("changes when a file is renamed", func(t *testing.T) { + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "a"), []byte("same"), 0o600)) + before, err := hashPath(dir) + require.NoError(t, err) + + require.NoError(t, os.Rename(filepath.Join(dir, "a"), filepath.Join(dir, "b"))) + after, err := hashPath(dir) + require.NoError(t, err) + + assert.NotEqual(t, before, after) + }) + + t.Run("is stable across polls", func(t *testing.T) { + dir := t.TempDir() + writeProjectedSecret(t, dir, map[string]string{"klio-archive": "archive"}) + + first, err := hashPath(dir) + require.NoError(t, err) + second, err := hashPath(dir) + require.NoError(t, err) + + assert.Equal(t, first, second) + }) + + t.Run("hashes a plain file like hashFile", func(t *testing.T) { + file := filepath.Join(t.TempDir(), testConfigFileName) + require.NoError(t, os.WriteFile(file, []byte("content"), 0o600)) + + fromPath, err := hashPath(file) + require.NoError(t, err) + fromFile, err := hashFile(file) + require.NoError(t, err) + + assert.Equal(t, fromFile, fromPath) + }) +} + +func TestConfigFileWatcherDetectsSiblingChange(t *testing.T) { + dir := t.TempDir() + writeProjectedSecret(t, dir, map[string]string{ + "klio-archive": "archive", + "source": "recovery source", + }) + + watcher := NewConfigFileWatcher(dir, 50*time.Millisecond) + + go func() { + time.Sleep(100 * time.Millisecond) + // Rewrite the file through its symlink, as a Secret rotation would. + _ = os.WriteFile(filepath.Join(dir, "source"), []byte("rotated recovery source"), 0o600) + }() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + assert.ErrorIs(t, watcher(ctx), ErrConfigFileChanged) +} diff --git a/core/internal/cnpgi/restore.go b/core/internal/cnpgi/restore.go index 599fb9490..b49bda814 100644 --- a/core/internal/cnpgi/restore.go +++ b/core/internal/cnpgi/restore.go @@ -25,6 +25,7 @@ import ( "fmt" "os" "os/exec" + "path/filepath" cnpgv1 "github.com/cloudnative-pg/api/pkg/api/v1" "github.com/cloudnative-pg/cloudnative-pg/pkg/postgres" @@ -61,7 +62,7 @@ func (impl restoreImpl) Restore( if err := json.Unmarshal(request.GetClusterDefinition(), &cluster); err != nil { return nil, fmt.Errorf("failed to unmarshal cluster definition: %w", err) } - configFile := "/var/lib/postgresql/klio/" + cluster.Spec.Bootstrap.Recovery.Source + configFile := filepath.Join(KlioConfigDir, cluster.Spec.Bootstrap.Recovery.Source) var backupID string var targetTime string diff --git a/core/internal/cnpgi/start.go b/core/internal/cnpgi/start.go index c00ae15ab..22c6f5e3b 100644 --- a/core/internal/cnpgi/start.go +++ b/core/internal/cnpgi/start.go @@ -33,6 +33,17 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" ) +// KlioConfigDir is the mount point of the projected klio-config volume, +// holding one file per resolved PluginConfiguration (e.g. klio-archive, +// the recovery source, the replica source). It must stay in sync with the +// mount path injected by the operator (KlioConfigDir in +// operator/internal/klioconfig). +const KlioConfigDir = "/var/lib/postgresql/klio" + +// ArchiveConfigKey is the config key of the archive PluginConfiguration. It +// must stay in sync with ArchiveConfigKey in operator/internal/klioconfig. +const ArchiveConfigKey = "klio-archive" + // CNPGI is the implementation of the PostgreSQL sidecar. type CNPGI struct { Client client.Client diff --git a/core/internal/cnpgi/wal.go b/core/internal/cnpgi/wal.go index 8365870f5..f1191d713 100644 --- a/core/internal/cnpgi/wal.go +++ b/core/internal/cnpgi/wal.go @@ -341,7 +341,10 @@ func getWalRepositoryConfigurationPath(cluster *cnpgv1.Cluster, instanceName str return repositoryName, err } -const backupRepositoryConfigPath = "/var/lib/postgresql/klio/klio-archive" +// backupRepositoryConfigPath is the config file of the archive repository. +// +//nolint:gochecknoglobals +var backupRepositoryConfigPath = filepath.Join(KlioConfigDir, ArchiveConfigKey) func getSourceRepositoryConfigPath(cluster *cnpgv1.Cluster) (string, error) { if cluster.Spec.ReplicaCluster == nil { @@ -349,7 +352,7 @@ func getSourceRepositoryConfigPath(cluster *cnpgv1.Cluster) (string, error) { } source := cluster.Spec.ReplicaCluster.Source - return filepath.Clean(filepath.Join("/var/lib/postgresql/klio", source)), nil + return filepath.Clean(filepath.Join(KlioConfigDir, source)), nil } func getBootstrapRepositoryConfigPath(cluster *cnpgv1.Cluster) (string, error) { @@ -363,5 +366,5 @@ func getBootstrapRepositoryConfigPath(cluster *cnpgv1.Cluster) (string, error) { return "", fmt.Errorf("cluster %s does not have a bootstrap recovery source configured", cluster.Name) } - return filepath.Clean(filepath.Join("/var/lib/postgresql/klio", cluster.Spec.Bootstrap.Recovery.Source)), nil + return filepath.Clean(filepath.Join(KlioConfigDir, cluster.Spec.Bootstrap.Recovery.Source)), nil } diff --git a/documentation/web/docs/user/api/_klio_api.md b/documentation/web/docs/user/api/_klio_api.md index 2d32d506d..56d4254e7 100644 --- a/documentation/web/docs/user/api/_klio_api.md +++ b/documentation/web/docs/user/api/_klio_api.md @@ -165,7 +165,7 @@ _Appears in:_ | `clusterName` _string_ | ClusterName is the name of the PostgreSQL cluster we are connecting to | True | | MinLength: 1
Required: \{\}
| | `pprof` _boolean_ | Pprof enables the pprof endpoint for performance profiling | | | Optional: \{\}
| | `mode` _[ServerMode](#servermode)_ | Mode selects the operation mode of the plugin. | True | standard | Enum: [standard read-only]
| -| `containers` _[Container](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.37/#container-v1-core) array_ | Containers allows defining a list of containers that will be merged with the Klio sidecar containers.
This enables users to customize the sidecars with additional environment variables, volume mounts,
resource limits, and other container settings without polluting the PostgreSQL container environment.
Merge behavior:
- Containers are matched by name (klio-plugin, klio-restore)
- User customizations serve as the base
- Klio required values (name, args, CONTAINER_NAME env var) always override user values
- User-defined environment variables and volume mounts are preserved
- Template defaults are applied only for fields not set by the user or Klio | | | MaxItems: 2
Optional: \{\}
| +| `containers` _[Container](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.37/#container-v1-core) array_ | Containers allows defining a list of containers that will be merged with the Klio sidecar containers.
This enables users to customize the sidecars with additional environment variables, volume mounts,
resource limits, and other container settings without polluting the PostgreSQL container environment.
Merge behavior:
- Containers are matched by name (klio-plugin)
- User customizations serve as the base
- Klio required values (name, args, CONTAINER_NAME env var) always override user values
- User-defined environment variables and volume mounts are preserved
- Template defaults are applied only for fields not set by the user or Klio | | | MaxItems: 1
Optional: \{\}
| #### PluginConfigurationStatus diff --git a/documentation/web/docs/user/opentelemetry.md b/documentation/web/docs/user/opentelemetry.md index 1d7ec731e..3c52aa8cd 100644 --- a/documentation/web/docs/user/opentelemetry.md +++ b/documentation/web/docs/user/opentelemetry.md @@ -623,10 +623,9 @@ spec: When deploying Klio as a CNPG Cluster plugin, configure OpenTelemetry by specifying the necessary environment variables in the `containers` section of -the `PluginConfiguration` spec. The available container names are: - -- `klio-plugin`: Main plugin sidecar for backup management -- `klio-restore`: Restore operations sidecar +the `PluginConfiguration` spec. The only container that can be customized +is `klio-plugin`: the plugin sidecar for backup, WAL and restore +operations. Create a `ConfigMap` for the shared OpenTelemetry configuration: @@ -670,13 +669,6 @@ spec: envFrom: - configMapRef: name: cluster-klio-otel-config - - name: klio-restore - env: - - name: OTEL_SERVICE_NAME - value: "klio-restore" - envFrom: - - configMapRef: - name: cluster-klio-otel-config ``` Mount the OpenTelemetry certificates using the Cluster's `projectedVolumeTemplate`. diff --git a/documentation/web/docs/user/plugin_configuration.md b/documentation/web/docs/user/plugin_configuration.md index a0243f425..fdb00b69c 100644 --- a/documentation/web/docs/user/plugin_configuration.md +++ b/documentation/web/docs/user/plugin_configuration.md @@ -10,9 +10,8 @@ CloudNativePG. It adds a `klio-plugin` container to each PostgreSQL instance pod, handling both backup creation/management and WAL streaming to the Klio server in real-time. -During recovery, Klio also injects a `klio-restore` container into the -CloudNativePG recovery Job to restore backups from the Klio server. See -[Available sidecar containers](#available-sidecar-containers) for details. +The same container serves the restore hooks while a cluster bootstraps from +a Klio backup, so a backup requested during recovery is served as well. ## Configuration @@ -529,12 +528,15 @@ unless necessary. ## Container customization The `PluginConfiguration` resource allows you to customize the Klio sidecar -containers by providing base container specifications that are used as the -foundation for the sidecars. This feature enables you to add custom environment +container by providing a base container specification that is used as the +foundation for the sidecar. This feature enables you to add custom environment variables, volume mounts, resource limits, and other container settings without modifying the PostgreSQL container environment. -### Basic example +### Example + +The following `PluginConfiguration` customizes the `klio-plugin` sidecar +with extra environment variables and resource limits: ```yaml apiVersion: klio.cnpg.io/v1alpha1 @@ -549,10 +551,17 @@ spec: containers: - name: klio-plugin env: - - name: CUSTOM_ENV_VAR - value: "my-value" - - name: DEBUG_LEVEL - value: "info" + - name: LOG_LEVEL + value: "debug" + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: "http://otel-collector:4317" + resources: + limits: + memory: "512Mi" + cpu: "1" + requests: + memory: "256Mi" + cpu: "500m" ``` ### How container merging works @@ -563,7 +572,7 @@ following merge behavior: 1. **Your container is the base**: When you define a container (e.g., `klio-plugin`), your specification serves as the starting point 1. **Klio enforces required values**: Klio sets its essential configuration: - - Container `name` (klio-plugin or klio-restore) + - Container `name` (`klio-plugin`) - Container `args` (the command arguments needed for operation) - `CONTAINER_NAME` environment variable 1. **Your customizations are preserved**: All other fields you define remain @@ -627,39 +636,3 @@ Klio sidecar, configured from the `source-plugin-config` `PluginConfiguration` (image, resources, environment variables, etc.) are applied to the sidecar following the same merging rules described above. - -### Available sidecar containers - -The following containers can be customized: - -- **`klio-plugin`**: Handles backup creation/management and WAL streaming to - the Klio server in PostgreSQL instance pods -- **`klio-restore`**: Restores backups during recovery jobs - -### Example: Resource limits and environment variables - -```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 - containers: - - name: klio-plugin - env: - - name: LOG_LEVEL - value: "debug" - - name: OTEL_EXPORTER_OTLP_ENDPOINT - value: "http://otel-collector:4317" - resources: - limits: - memory: "512Mi" - cpu: "1" - requests: - memory: "256Mi" - cpu: "500m" -``` diff --git a/documentation/web/docs/user/upgrade_notes.md b/documentation/web/docs/user/upgrade_notes.md index 90f3f6e1f..8c7340905 100644 --- a/documentation/web/docs/user/upgrade_notes.md +++ b/documentation/web/docs/user/upgrade_notes.md @@ -10,6 +10,17 @@ see the [Helm chart page](helm_chart.mdx#upgrades). ## 0.0.20 to 0.0.21 +### The `klio-restore` container is gone + +The `klio-plugin` sidecar now also serves the restore while a cluster +bootstraps from a Klio backup, so Klio no longer injects a separate +`klio-restore` container into the recovery Job. The `PluginConfiguration` +CRD only accepts `klio-plugin` in `spec.containers`: a `PluginConfiguration` +that still lists a `klio-restore` entry is rejected on its next update, and +its status cannot be written on Kubernetes versions without CRD validation +ratcheting. Remove the `klio-restore` entry and move any customization you +need onto `klio-plugin` before upgrading. + ### Migrating from the Multi-PVC Model Klio servers created until v0.0.20 used four separate diff --git a/operator/api/v1alpha1/plugin_configuration_types.go b/operator/api/v1alpha1/plugin_configuration_types.go index 6c3c8f612..e80d8e6f2 100644 --- a/operator/api/v1alpha1/plugin_configuration_types.go +++ b/operator/api/v1alpha1/plugin_configuration_types.go @@ -88,17 +88,17 @@ type PluginConfigurationSpec struct { // resource limits, and other container settings without polluting the PostgreSQL container environment. // // Merge behavior: - // - Containers are matched by name (klio-plugin, klio-restore) + // - Containers are matched by name (klio-plugin) // - User customizations serve as the base // - Klio required values (name, args, CONTAINER_NAME env var) always override user values // - User-defined environment variables and volume mounts are preserved // - Template defaults are applied only for fields not set by the user or Klio // // +optional - // +kubebuilder:validation:MaxItems=2 + // +kubebuilder:validation:MaxItems=1 // +listType=map // +listMapKey=name - // +kubebuilder:validation:XValidation:rule="self.all(c, c.name in ['klio-plugin', 'klio-restore'])",message="container name must be one of: klio-plugin, klio-restore" + // +kubebuilder:validation:XValidation:rule="self.all(c, c.name == 'klio-plugin')",message="container name must be klio-plugin" Containers []corev1.Container `json:"containers,omitempty"` } diff --git a/operator/config/crd/bases/klio.cnpg.io_pluginconfigurations.yaml b/operator/config/crd/bases/klio.cnpg.io_pluginconfigurations.yaml index 8f5664f66..ace9cf7cf 100644 --- a/operator/config/crd/bases/klio.cnpg.io_pluginconfigurations.yaml +++ b/operator/config/crd/bases/klio.cnpg.io_pluginconfigurations.yaml @@ -58,7 +58,7 @@ spec: resource limits, and other container settings without polluting the PostgreSQL container environment. Merge behavior: - - Containers are matched by name (klio-plugin, klio-restore) + - Containers are matched by name (klio-plugin) - User customizations serve as the base - Klio required values (name, args, CONTAINER_NAME env var) always override user values - User-defined environment variables and volume mounts are preserved @@ -1641,14 +1641,14 @@ spec: required: - name type: object - maxItems: 2 + maxItems: 1 type: array x-kubernetes-list-map-keys: - name x-kubernetes-list-type: map x-kubernetes-validations: - - message: 'container name must be one of: klio-plugin, klio-restore' - rule: self.all(c, c.name in ['klio-plugin', 'klio-restore']) + - message: container name must be klio-plugin + rule: self.all(c, c.name == 'klio-plugin') mode: default: standard description: Mode selects the operation mode of the plugin. diff --git a/operator/config/manifests/bases/klio-operator.clusterserviceversion.yaml b/operator/config/manifests/bases/klio-operator.clusterserviceversion.yaml index e975f1d29..eccc359e7 100644 --- a/operator/config/manifests/bases/klio-operator.clusterserviceversion.yaml +++ b/operator/config/manifests/bases/klio-operator.clusterserviceversion.yaml @@ -321,8 +321,8 @@ spec: x-descriptors: - urn:alm:descriptor:com.tectonic.ui:booleanSwitch - urn:alm:descriptor:com.tectonic.ui:advanced - - description: Advanced overrides for the Klio sidecar containers (klio-plugin, - klio-restore). Use at your own risk. + - description: Advanced overrides for the Klio sidecar container (klio-plugin). + Use at your own risk. displayName: Sidecar Container Overrides path: containers x-descriptors: diff --git a/operator/config/samples/opentelemetry/single/cluster/plugin_configuration.yaml b/operator/config/samples/opentelemetry/single/cluster/plugin_configuration.yaml index 8f94dfc0c..438556bdb 100644 --- a/operator/config/samples/opentelemetry/single/cluster/plugin_configuration.yaml +++ b/operator/config/samples/opentelemetry/single/cluster/plugin_configuration.yaml @@ -15,10 +15,3 @@ spec: envFrom: - configMapRef: name: cluster-klio-otel-config - - name: klio-restore - env: - - name: OTEL_SERVICE_NAME - value: "klio-restore" - envFrom: - - configMapRef: - name: cluster-klio-otel-config diff --git a/operator/dist/chart/crds/pluginconfiguration-crd.yaml b/operator/dist/chart/crds/pluginconfiguration-crd.yaml index 505840c38..847b1315e 100644 --- a/operator/dist/chart/crds/pluginconfiguration-crd.yaml +++ b/operator/dist/chart/crds/pluginconfiguration-crd.yaml @@ -57,7 +57,7 @@ spec: resource limits, and other container settings without polluting the PostgreSQL container environment. Merge behavior: - - Containers are matched by name (klio-plugin, klio-restore) + - Containers are matched by name (klio-plugin) - User customizations serve as the base - Klio required values (name, args, CONTAINER_NAME env var) always override user values - User-defined environment variables and volume mounts are preserved @@ -1640,14 +1640,14 @@ spec: required: - name type: object - maxItems: 2 + maxItems: 1 type: array x-kubernetes-list-map-keys: - name x-kubernetes-list-type: map x-kubernetes-validations: - - message: 'container name must be one of: klio-plugin, klio-restore' - rule: self.all(c, c.name in ['klio-plugin', 'klio-restore']) + - message: container name must be klio-plugin + rule: self.all(c, c.name == 'klio-plugin') mode: default: standard description: Mode selects the operation mode of the plugin. diff --git a/operator/internal/cnpgi/lifecycle.go b/operator/internal/cnpgi/lifecycle.go index 45bede273..9a8c97cf0 100644 --- a/operator/internal/cnpgi/lifecycle.go +++ b/operator/internal/cnpgi/lifecycle.go @@ -144,64 +144,80 @@ func (impl LifecycleImplementation) LifecycleHook( } } -// buildRestoreSidecar resolves the Klio recovery source plugin configuration and -// returns the restore-mode sidecar to inject into a recovery bootstrap. It -// returns ok=false when the cluster is not recovering through the Klio plugin. -func (impl LifecycleImplementation) buildRestoreSidecar( - ctx context.Context, - cluster *cnpgv1.Cluster, -) (corev1.Container, bool, error) { - contextLogger := log.FromContext(ctx).WithName("klio-restore-sidecar") - +// errRecoveryPluginConfigurationMissing is returned when the cluster recovers +// through the Klio plugin but the recovery source names no PluginConfiguration. +var errRecoveryPluginConfigurationMissing = errors.New( + "recovery plugin configuration missing '" + klioconfig.PluginConfigurationRefParam + "' parameter") + +// klioRecoverySource returns the config key of the external cluster the +// cluster bootstraps from, and whether that recovery goes through the Klio +// plugin. +func klioRecoverySource(cluster *cnpgv1.Cluster) (string, bool) { recoveryPluginConfig := cluster.GetRecoverySourcePlugin() if recoveryPluginConfig == nil || recoveryPluginConfig.Name != klioconfig.PluginName || !recoveryPluginConfig.IsEnabled() { - // not our plugin, skip - return corev1.Container{}, false, nil + return "", false } - if recoveryPluginConfig.Parameters[klioconfig.PluginConfigurationRefParam] == "" { - contextLogger.Warning("recovery plugin configuration missing 'ref' parameter") - return corev1.Container{}, false, errors.New("recovery plugin configuration missing 'ref' parameter") + recoveryExternalCluster, _ := cluster.ExternalCluster(cluster.Spec.Bootstrap.Recovery.Source) + + return recoveryExternalCluster.GetServerName(), true +} + +// selectPluginConfiguration picks the PluginConfiguration the sidecar of the +// given instance is customized from. It returns nil when the instance needs +// no sidecar at all. +// +// The same klio-plugin sidecar serves every phase: restore hooks and WAL +// restore during a bootstrap, backup and WAL archiving afterwards. It picks the +// repository per request from the cluster definition, so the pod spec does not +// change once recovery completes and a backup requested meanwhile is served. +func selectPluginConfiguration( + cluster *cnpgv1.Cluster, + podName string, + plugins klioconfig.ClusterPlugins, +) (*kliov1alpha1.PluginConfiguration, error) { + // While the cluster bootstraps from a Klio backup the restore reads the + // recovery source configuration, so it must be resolved whatever else the + // cluster does: failing here is clearer than a restore hook failing on a + // missing file later. + // During a bootstrap recovery no primary has been elected yet and there is no + // instance-manager cache. This gate relies on CurrentPrimary never reverting + // to empty once set. + + var recoveryPC *kliov1alpha1.PluginConfiguration + if cluster.Status.CurrentPrimary == "" { + if key, ok := klioRecoverySource(cluster); ok { + recoveryPC, ok = plugins[key] + if !ok { + return nil, errRecoveryPluginConfigurationMissing + } + } } - clusterPC := &kliov1alpha1.PluginConfiguration{} - if err := impl.Client.Get(ctx, - client.ObjectKey{ - Namespace: cluster.Namespace, - Name: recoveryPluginConfig.Parameters[klioconfig.PluginConfigurationRefParam], - }, - clusterPC); err != nil { - contextLogger.Error(err, "Failed to get client configuration") - return corev1.Container{}, false, fmt.Errorf("failed to get client configuration: %w", err) + if pc := plugins.ArchivePluginConfiguration(); pc != nil { + return pc, nil } - // Resolve the config key for the recovery source. - recoverySource := cluster.Spec.Bootstrap.Recovery.Source - recoveryExternalCluster, _ := cluster.ExternalCluster(recoverySource) - configKey := recoveryExternalCluster.GetServerName() + // No archive plugin. The sidecar is still needed while the cluster + // bootstraps from a Klio backup... + if recoveryPC != nil { + return recoveryPC, nil + } - // Build the restore sidecar with merge strategy: - // 1. Start from user customization if present (as the base) - // 2. Apply Klio required values (name, args, essential env vars) - // 3. Template defaults will be merged later in reconcilePodSpec - restoreSidecar := findUserContainer("klio-restore", clusterPC.Spec.Containers) - restoreSidecar.Args = []string{ - "cnpgi", - "restore", - "--config", "/var/lib/postgresql/klio/" + configKey, - pgdata, - } - restoreSidecar.Env = ensureEnvVar(restoreSidecar.Env, corev1.EnvVar{ - Name: "CONTAINER_NAME", - Value: "klio-restore", - }) + // ...and on the designated primary of a replica cluster, which restores + // WALs from the external source. + if !cluster.IsReplica() || cluster.Status.TargetPrimary != podName { + return nil, nil + } + ext, _ := cluster.ExternalCluster(cluster.Spec.ReplicaCluster.Source) - return restoreSidecar, true, nil + // A nil result means the cluster may be replicating using a different plugin + return plugins[ext.GetServerName()], nil } -// reconcileJob injects the restore sidecar into a recovery Job. Older +// reconcileJob injects the Klio sidecar into a recovery Job. Older // CloudNativePG releases run recovery bootstrap as a dedicated Job; newer // versions run it inside the instance pod instead (handled by reconcilePod). func (impl LifecycleImplementation) reconcileJob( @@ -217,11 +233,7 @@ func (impl LifecycleImplementation) reconcileJob( return nil, err } - restoreSidecar, ok, err := impl.buildRestoreSidecar(ctx, cluster) - if err != nil { - return nil, err - } - if !ok { + if _, ok := klioRecoverySource(cluster); !ok { // not our plugin, skip return nil, nil } @@ -246,6 +258,17 @@ func (impl LifecycleImplementation) reconcileJob( return nil, nil } + // A Job pod has no name at spec time: the sidecar never becomes the + // primary's WAL sender, so none is needed. + targetPC, err := selectPluginConfiguration(cluster, "", plugins) + if err != nil { + return nil, err + } + if targetPC == nil { + return nil, nil + } + archives := plugins.ArchivePluginConfiguration() != nil + mutatedJob := job.DeepCopy() cnpgGroup, cnpgVersion := cnpgGroupVersion(cluster) if err := reconcilePodSpec( @@ -253,7 +276,7 @@ func (impl LifecycleImplementation) reconcileJob( &mutatedJob.Spec.Template.Spec, jobRole, reconcilePodSpecConfiguration{ - sidecarsToEnrich: []corev1.Container{restoreSidecar}, + sidecarsToEnrich: []corev1.Container{buildInstanceSidecarTemplate("", cluster, targetPC, archives)}, cnpgGroup: cnpgGroup, cnpgVersion: cnpgVersion, plugins: plugins, @@ -296,43 +319,17 @@ func (impl LifecycleImplementation) reconcilePod( return nil, err } - // When CloudNativePG runs recovery bootstrap inside the instance pod (as an - // init container), the restore hooks must be served by a sidecar in this - // pod. Inject the restore sidecar while the cluster performs its initial - // recovery bootstrap. - if cluster.Status.CurrentPrimary == "" { - restoreSidecar, ok, rerr := impl.buildRestoreSidecar(ctx, cluster) - if rerr != nil { - return nil, rerr - } - if ok { - return impl.injectPodSidecars(ctx, cluster, pod, plugins, []corev1.Container{restoreSidecar}) - } + targetPC, err := selectPluginConfiguration(cluster, pod.Name, plugins) + if err != nil { + return nil, err } - - archiveConfigKey := klioconfig.ArchiveConfigKey - targetPC, ok := plugins[klioconfig.ArchiveConfigKey] - if !ok { - // No archive plugin. The only case where the instance sidecar is - // still needed is the designated primary of a replica cluster, - // which restores WALs from the external source. - if !cluster.IsReplica() || - cluster.Status.TargetPrimary != pod.Name { - return nil, nil - } - - replicaSource := cluster.Spec.ReplicaCluster.Source - ext, _ := cluster.ExternalCluster(replicaSource) - archiveConfigKey = "" - targetPC, ok = plugins[ext.GetServerName()] - if !ok { - // The cluster may be replicating using a different plugin - return nil, nil - } + if targetPC == nil { + return nil, nil } + archives := plugins.ArchivePluginConfiguration() != nil return impl.injectPodSidecars(ctx, cluster, pod, plugins, - []corev1.Container{buildInstanceSidecarTemplate(pod, cluster, targetPC, archiveConfigKey)}) + []corev1.Container{buildInstanceSidecarTemplate(pod.Name, cluster, targetPC, archives)}) } // injectPodSidecars enriches the pod spec with the given Klio sidecars and @@ -379,10 +376,10 @@ func (impl LifecycleImplementation) injectPodSidecars( } func buildInstanceSidecarTemplate( - pod *corev1.Pod, + podName string, cluster *cnpgv1.Cluster, clusterPC *kliov1alpha1.PluginConfiguration, - archiveConfigKey string, + archives bool, ) corev1.Container { // Merge strategy: // 1. Start from user customization if present (as the base) @@ -390,16 +387,17 @@ func buildInstanceSidecarTemplate( // 3. Template defaults will be merged later in reconcilePodSpec sidecar := corev1.Container{Name: KlioPluginContainerName} - args := []string{ - "cnpgi", - "instance", - "--pod-name", pod.Name, + args := []string{"cnpgi", "instance"} + if podName != "" { + args = append(args, "--pod-name", podName) + } + args = append(args, "--cluster-name", cluster.Name, "--cluster-namespace", cluster.Namespace, - } + ) - if archiveConfigKey != "" { - args = append(args, "--config", path.Join("/var/lib/postgresql/klio/", archiveConfigKey)) + if archives { + args = append(args, "--config", path.Join(klioconfig.KlioConfigDir, klioconfig.ArchiveConfigKey)) } if clusterPC != nil { @@ -470,7 +468,7 @@ func reconcilePodSpec( // NOSONAR }) volumeMounts = ensureVolumeMount(volumeMounts, corev1.VolumeMount{ Name: "klio-config", - MountPath: "/var/lib/postgresql/klio", + MountPath: klioconfig.KlioConfigDir, }) // Add the TLS volumes and mounts for the server and client secrets diff --git a/operator/internal/cnpgi/lifecycle_test.go b/operator/internal/cnpgi/lifecycle_test.go index b25a0092c..4065b2b75 100644 --- a/operator/internal/cnpgi/lifecycle_test.go +++ b/operator/internal/cnpgi/lifecycle_test.go @@ -30,6 +30,7 @@ import ( jsonpatch "github.com/evanphx/json-patch/v5" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -162,7 +163,7 @@ func TestFindUserContainer(t *testing.T) { containerName: KlioPluginContainerName, customContainers: []corev1.Container{ { - Name: "klio-restore", + Name: "other-container", Image: "other-image:latest", }, }, @@ -414,7 +415,7 @@ func TestBuildInstanceSidecarTemplate(t *testing.T) { }, } - result := buildInstanceSidecarTemplate(pod, cluster, clusterPC, klioconfig.ArchiveConfigKey) + result := buildInstanceSidecarTemplate(pod.Name, cluster, clusterPC, true) // Klio required values are set assert.Equal(t, KlioPluginContainerName, result.Name) @@ -455,7 +456,7 @@ func TestBuildInstanceSidecarTemplate(t *testing.T) { }, } - result := buildInstanceSidecarTemplate(pod, cluster, clusterPC, klioconfig.ArchiveConfigKey) + result := buildInstanceSidecarTemplate(pod.Name, cluster, clusterPC, true) assert.Equal(t, KlioPluginContainerName, result.Name) assert.Equal(t, []string{ @@ -477,7 +478,7 @@ func TestBuildInstanceSidecarTemplate(t *testing.T) { Spec: kliov1alpha1.PluginConfigurationSpec{}, } - result := buildInstanceSidecarTemplate(pod, cluster, clusterPC, klioconfig.ArchiveConfigKey) + result := buildInstanceSidecarTemplate(pod.Name, cluster, clusterPC, true) assert.Equal(t, KlioPluginContainerName, result.Name) assert.Equal(t, []string{ @@ -495,7 +496,7 @@ func TestBuildInstanceSidecarTemplate(t *testing.T) { }) t.Run("with nil clusterPC", func(t *testing.T) { - result := buildInstanceSidecarTemplate(pod, cluster, nil, klioconfig.ArchiveConfigKey) + result := buildInstanceSidecarTemplate(pod.Name, cluster, nil, true) assert.Equal(t, KlioPluginContainerName, result.Name) assert.Equal(t, []string{ @@ -509,7 +510,7 @@ func TestBuildInstanceSidecarTemplate(t *testing.T) { }) t.Run("without archiving", func(t *testing.T) { - result := buildInstanceSidecarTemplate(pod, cluster, nil, "") + result := buildInstanceSidecarTemplate(pod.Name, cluster, nil, false) assert.Equal(t, KlioPluginContainerName, result.Name) assert.Equal(t, []string{ @@ -538,7 +539,7 @@ func TestBuildInstanceSidecarTemplate(t *testing.T) { }, } - result := buildInstanceSidecarTemplate(pod, cluster, clusterPC, klioconfig.ArchiveConfigKey) + result := buildInstanceSidecarTemplate(pod.Name, cluster, clusterPC, true) // Should get the default template, not the custom one assert.Equal(t, KlioPluginContainerName, result.Name) @@ -647,10 +648,10 @@ func applyPatch[T client.Object](t *testing.T, resp *lifecycle.OperatorLifecycle return result } -// findInitContainer returns the init container with the given name, or nil. -func findInitContainer(pod *corev1.Pod, name string) *corev1.Container { +// findSidecar returns the klio-plugin init container, or nil. +func findSidecar(pod *corev1.Pod) *corev1.Container { for i := range pod.Spec.InitContainers { - if pod.Spec.InitContainers[i].Name == name { + if pod.Spec.InitContainers[i].Name == KlioPluginContainerName { return &pod.Spec.InitContainers[i] } } @@ -860,7 +861,7 @@ func TestReconcilePodPluginSelection(t *testing.T) { require.NotNil(t, resp, "primary pod of replica cluster should get a sidecar") patchedPod := applyPatch(t, resp, pod) - sidecar := findInitContainer(patchedPod, KlioPluginContainerName) + sidecar := findSidecar(patchedPod) require.NotNil(t, sidecar, "sidecar init container should be present") assert.Equal(t, "source-image:latest", sidecar.Image, "sidecar should use the image from the source PluginConfiguration") @@ -918,7 +919,7 @@ func TestReconcilePodPluginSelection(t *testing.T) { require.NotNil(t, resp) patchedPod := applyPatch(t, resp, pod) - sidecar := findInitContainer(patchedPod, KlioPluginContainerName) + sidecar := findSidecar(patchedPod) require.NotNil(t, sidecar, "sidecar init container should be present") assert.Equal(t, "archive-image:latest", sidecar.Image, "sidecar should use the image from the archive PluginConfiguration, not the source") @@ -1133,3 +1134,231 @@ func TestSidecarSecurityContext(t *testing.T) { assert.Nil(t, sc) }) } + +// TestBootstrapSidecar covers the sidecar injected while a cluster bootstraps +// from a Klio backup. It must be the same klio-plugin sidecar used afterwards, +// so a backup requested during recovery finds the backup capability (#268). +func TestBootstrapSidecar(t *testing.T) { + scheme := newTestScheme(t) + const ( + recoveryPCName = "recovery-pc" + sourceName = "source-cluster" + archivePCName = "archive-pc" + ) + makeCluster := func(withArchive bool) *cnpgv1.Cluster { + cluster := &cnpgv1.Cluster{ + TypeMeta: metav1.TypeMeta{ + APIVersion: cnpgv1.SchemeGroupVersion.String(), + Kind: "Cluster", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: testClusterName, + Namespace: testClusterNamespace, + }, + Spec: cnpgv1.ClusterSpec{ + Bootstrap: &cnpgv1.BootstrapConfiguration{ + Recovery: &cnpgv1.BootstrapRecovery{Source: sourceName}, + }, + ExternalClusters: []cnpgv1.ExternalCluster{ + { + Name: sourceName, + PluginConfiguration: &cnpgv1.PluginConfiguration{ + Name: klioconfig.PluginName, + Enabled: new(true), + Parameters: map[string]string{ + klioconfig.PluginConfigurationRefParam: recoveryPCName, + }, + }, + }, + }, + }, + } + if withArchive { + cluster.Spec.Plugins = []cnpgv1.PluginConfiguration{ + { + Name: klioconfig.PluginName, + Enabled: new(true), + Parameters: map[string]string{ + klioconfig.PluginConfigurationRefParam: archivePCName, + }, + }, + } + } + + return cluster + } + makePC := func(name, image string) *kliov1alpha1.PluginConfiguration { + return &kliov1alpha1.PluginConfiguration{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: testClusterNamespace}, + Spec: kliov1alpha1.PluginConfigurationSpec{ + ClusterName: testClusterName, + ServerAddress: "klio-server.example.com", + ClientSecretName: "client-secret", + ServerSecretName: "server-secret", + Containers: []corev1.Container{ + {Name: KlioPluginContainerName, Image: image}, + }, + }, + } + } + makePod := func() *corev1.Pod { + return &corev1.Pod{ + TypeMeta: metav1.TypeMeta{Kind: "Pod", APIVersion: "v1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: testPodName, + Namespace: testClusterNamespace, + }, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "postgres"}}, + }, + } + } + makeJob := func() *batchv1.Job { + return &batchv1.Job{ + TypeMeta: metav1.TypeMeta{Kind: "Job", APIVersion: "batch/v1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: testClusterName + "-1-full-recovery", + Namespace: testClusterNamespace, + }, + Spec: batchv1.JobSpec{ + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{"cnpg.io/jobRole": "full-recovery"}, + }, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "full-recovery"}}, + }, + }, + }, + } + } + assertInstanceSidecar := func(t *testing.T, sidecar *corev1.Container) { + t.Helper() + require.NotNil(t, sidecar, "klio-plugin sidecar must be injected") + require.Greater(t, len(sidecar.Args), 1) + assert.Equal(t, []string{"cnpgi", "instance"}, sidecar.Args[:2], + "bootstrap sidecar must run the full instance server") + assert.NotContains(t, sidecar.Args, "restore") + } + + t.Run("bootstrapping pod without archive gets the instance sidecar from the recovery PC", func(t *testing.T) { + cluster := makeCluster(false) + pod := makePod() + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(makePC(recoveryPCName, "recovery-image:latest")).Build() + impl := LifecycleImplementation{Client: fakeClient} + resp, err := impl.reconcilePod(context.Background(), cluster, buildLifecycleRequest(t, cluster, pod)) + require.NoError(t, err) + require.NotNil(t, resp) + patched := applyPatch(t, resp, pod) + sidecar := findSidecar(patched) + assertInstanceSidecar(t, sidecar) + assert.Equal(t, "recovery-image:latest", sidecar.Image) + assert.Contains(t, sidecar.Args, testPodName) + assert.NotContains(t, sidecar.Args, argConfig, "no archive: nothing to watch or archive") + }) + + t.Run("bootstrapping pod with archive gets the archive-configured instance sidecar", func(t *testing.T) { + cluster := makeCluster(true) + pod := makePod() + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects( + makePC(recoveryPCName, "recovery-image:latest"), + makePC(archivePCName, "archive-image:latest"), + ).Build() + impl := LifecycleImplementation{Client: fakeClient} + resp, err := impl.reconcilePod(context.Background(), cluster, buildLifecycleRequest(t, cluster, pod)) + require.NoError(t, err) + require.NotNil(t, resp) + patched := applyPatch(t, resp, pod) + sidecar := findSidecar(patched) + assertInstanceSidecar(t, sidecar) + assert.Equal(t, "archive-image:latest", sidecar.Image) + assert.Contains(t, sidecar.Args, expectedArchiveConfigPath) + }) + + t.Run("bootstrap spec equals steady-state spec so no rollout follows recovery", func(t *testing.T) { + cluster := makeCluster(true) + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects( + makePC(recoveryPCName, "recovery-image:latest"), + makePC(archivePCName, "archive-image:latest"), + ).Build() + impl := LifecycleImplementation{Client: fakeClient} + bootstrapPod := makePod() + resp, err := impl.reconcilePod(context.Background(), cluster, + buildLifecycleRequest(t, cluster, bootstrapPod)) + require.NoError(t, err) + bootstrapped := applyPatch(t, resp, bootstrapPod) + + cluster.Status.CurrentPrimary = testPodName + cluster.Status.TargetPrimary = testPodName + steadyPod := makePod() + resp, err = impl.reconcilePod(context.Background(), cluster, + buildLifecycleRequest(t, cluster, steadyPod)) + require.NoError(t, err) + steady := applyPatch(t, resp, steadyPod) + assert.Equal(t, steady.Spec, bootstrapped.Spec) + }) + + t.Run("recovery through klio with missing ref returns an error", func(t *testing.T) { + cluster := makeCluster(false) + cluster.Spec.ExternalClusters[0].PluginConfiguration.Parameters = nil + fakeClient := fake.NewClientBuilder().WithScheme(scheme).Build() + impl := LifecycleImplementation{Client: fakeClient} + _, err := impl.reconcilePod(context.Background(), cluster, buildLifecycleRequest(t, cluster, makePod())) + require.ErrorIs(t, err, errRecoveryPluginConfigurationMissing) + }) + + t.Run("recovery through klio with missing ref and an archive returns an error", func(t *testing.T) { + cluster := makeCluster(true) + cluster.Spec.ExternalClusters[0].PluginConfiguration.Parameters = nil + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(makePC(archivePCName, "archive-image:latest")).Build() + impl := LifecycleImplementation{Client: fakeClient} + _, err := impl.reconcilePod(context.Background(), cluster, buildLifecycleRequest(t, cluster, makePod())) + require.ErrorIs(t, err, errRecoveryPluginConfigurationMissing) + }) + + t.Run("recovery through another plugin without archive gets no sidecar", func(t *testing.T) { + cluster := makeCluster(false) + cluster.Spec.ExternalClusters[0].PluginConfiguration.Name = "other.plugin.io" + fakeClient := fake.NewClientBuilder().WithScheme(scheme).Build() + impl := LifecycleImplementation{Client: fakeClient} + resp, err := impl.reconcilePod(context.Background(), cluster, buildLifecycleRequest(t, cluster, makePod())) + require.NoError(t, err) + assert.Nil(t, resp) + }) + + t.Run("recovery job gets the instance sidecar without a pod name", func(t *testing.T) { + cluster := makeCluster(false) + job := makeJob() + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(makePC(recoveryPCName, "recovery-image:latest")).Build() + impl := LifecycleImplementation{Client: fakeClient} + resp, err := impl.reconcileJob(context.Background(), cluster, buildLifecycleRequest(t, cluster, job)) + require.NoError(t, err) + require.NotNil(t, resp) + patched := applyPatch(t, resp, job) + var sidecar *corev1.Container + for i := range patched.Spec.Template.Spec.InitContainers { + if patched.Spec.Template.Spec.InitContainers[i].Name == KlioPluginContainerName { + sidecar = &patched.Spec.Template.Spec.InitContainers[i] + } + } + assertInstanceSidecar(t, sidecar) + assert.Equal(t, "recovery-image:latest", sidecar.Image) + assert.NotContains(t, sidecar.Args, argPodName, "a job pod has no known name at spec time") + }) + + t.Run("recovery job through another plugin is left alone", func(t *testing.T) { + cluster := makeCluster(true) + cluster.Spec.ExternalClusters[0].PluginConfiguration.Name = "other.plugin.io" + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(makePC(archivePCName, "archive-image:latest")).Build() + impl := LifecycleImplementation{Client: fakeClient} + resp, err := impl.reconcileJob(context.Background(), cluster, buildLifecycleRequest(t, cluster, makeJob())) + require.NoError(t, err) + assert.Nil(t, resp) + }) +} diff --git a/operator/internal/klioconfig/config.go b/operator/internal/klioconfig/config.go index 8e6db3988..6731c5c6d 100644 --- a/operator/internal/klioconfig/config.go +++ b/operator/internal/klioconfig/config.go @@ -44,6 +44,12 @@ const ( // ArchiveConfigKey is the key used for the archive plugin configuration. ArchiveConfigKey = "klio-archive" + // KlioConfigDir is the mount point of the projected klio-config volume + // injected into instance pods, holding one file per resolved + // PluginConfiguration. It must stay in sync with KlioConfigDir in + // core/internal/cnpgi. + KlioConfigDir = "/var/lib/postgresql/klio" + // PluginConfigurationRefParam is the name of the parameter that contains // the reference to the Klio PluginConfiguration resource. PluginConfigurationRefParam = "pluginConfigurationRef" diff --git a/operator/test/e2e/backup_from_replica_cluster_test.go b/operator/test/e2e/backup_from_replica_cluster_test.go index a4f415be7..0f61c7bff 100644 --- a/operator/test/e2e/backup_from_replica_cluster_test.go +++ b/operator/test/e2e/backup_from_replica_cluster_test.go @@ -44,8 +44,8 @@ import ( "github.com/cloudnative-pg/klio/operator/test/utils/templates/secrets" ) -// ReplicaClusterBackupFeature verifies that an immediate backup taken from a -// freshly-created replica cluster completes. +// ReplicaClusterBackupFeature verifies that a backup requested as soon as a +// replica cluster is created, before its bootstrap completes, completes. type ReplicaClusterBackupFeature struct { scenario *commonBackupRestoreScenario @@ -220,8 +220,8 @@ func (f *ReplicaClusterBackupFeature) Setup() types.StepFunc { return f.scenario.Setup } -// Run backs up the source cluster, bootstraps the replica cluster, then takes an -// immediate backup of the replica cluster and asserts it completes. +// Run backs up the source cluster, creates the replica cluster together with a +// backup request for it, and asserts both the bootstrap and the backup complete. func (f *ReplicaClusterBackupFeature) Run() types.StepFunc { return func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { t.Helper() @@ -250,21 +250,41 @@ func (f *ReplicaClusterBackupFeature) Run() types.StepFunc { require.NoError(t, r.Create(ctx, f.replicaPluginConfiguration), "failed to create replica plugin configuration") require.NoError(t, r.Create(ctx, f.replicaCluster), "failed to create replica cluster") + + // Request the backup right away, as a ScheduledBackup with `immediate` + // created together with the cluster does. CloudNativePG dispatches it as + // soon as the instance is ready, which is while the cluster is still + // completing its bootstrap: the sidecar must already serve backups at + // that point (#268). + require.NoError(t, r.Create(ctx, f.replicaBackup), "failed to create replica backup") + + // The bootstrapping pod must be the one serving the backup: the same + // sidecar covers recovery and backups, so no rollout follows the + // bootstrap. Remember the first pod and check it survives. + firstPod := &corev1.Pod{} + require.NoError(t, wait.For(func(ctx context.Context) (bool, error) { + err := r.Get(ctx, f.replicaCluster.Name+"-1", f.replicaCluster.Namespace, firstPod) + return err == nil, nil + }, wait.WithTimeout(f.recoveryTimeout), wait.WithInterval(f.checkInterval)), + "replica cluster pod not created") + require.NoError(t, wait.For( machineryConditions.ClusterIsReady(r, f.replicaCluster), wait.WithTimeout(f.recoveryTimeout), wait.WithInterval(f.checkInterval), ), "replica cluster not ready") - // The immediate backup of the freshly-created replica cluster must - // complete. - require.NoError(t, r.Create(ctx, f.replicaBackup), "failed to create replica backup") require.NoError(t, wait.For( machineryConditions.BackupIsCompleted(r, f.replicaBackup), wait.WithTimeout(f.replicaBackupTimeout), wait.WithInterval(f.checkInterval), ), "replica cluster backup not completed") + currentPod := &corev1.Pod{} + require.NoError(t, r.Get(ctx, firstPod.Name, firstPod.Namespace, currentPod)) + require.Equal(t, firstPod.UID, currentPod.UID, + "the bootstrapping pod must not be recreated once recovery completes") + return ctx } } From 1aff8a0fdd2049b61f4cf77e2342c69b0d8d6444 Mon Sep 17 00:00:00 2001 From: Gabriele Quaresima Date: Wed, 23 Sep 2026 15:35:16 +0200 Subject: [PATCH 2/2] fix(replication): log the replication slot's real restart point after creation CREATE_REPLICATION_SLOT only fills in consistent_point for logical slots; for the physical slot created here it always comes back as 0/0, so logging it told us nothing about where send-wal will actually resume streaming from. Read the slot's real restart LSN and timeline back explicitly with READ_REPLICATION_SLOT right after creation instead, and log that. Also fixes ReadReplicationSlot's godoc, which wrongly described it as running IDENTIFY_SYSTEM. Signed-off-by: Gabriele Quaresima Signed-off-by: Francesco Canovai --- core/internal/client/sendwal/receiver.go | 13 +++++++++---- core/internal/client/sendwal/replslot.go | 6 +++--- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/core/internal/client/sendwal/receiver.go b/core/internal/client/sendwal/receiver.go index cd138611e..e68334e25 100644 --- a/core/internal/client/sendwal/receiver.go +++ b/core/internal/client/sendwal/receiver.go @@ -317,15 +317,20 @@ func (s *Process) ensureReplicationSlotExists( // command is built and parsed manually here. sql := fmt.Sprintf("CREATE_REPLICATION_SLOT %s PHYSICAL RESERVE_WAL", s.config.Source.Slot) - replicationSlotResult, err := pglogrepl.ParseCreateReplicationSlot(conn.Exec(ctx, sql)) + _, err = pglogrepl.ParseCreateReplicationSlot(conn.Exec(ctx, sql)) if err != nil { - return fmt.Errorf("while creating temporary replication slot: %w", err) + return fmt.Errorf("while creating replication slot: %w", err) } + slotResult, err = ReadReplicationSlot(ctx, conn, s.config.Source.Slot) + if err != nil { + return fmt.Errorf("while reading replication slot after creation: %w", err) + } contextLogger.Info( "Created replication slot", - "consistentPoint", replicationSlotResult.ConsistentPoint, - "name", replicationSlotResult.SlotName) + "slotName", s.config.Source.Slot, + "restartLSN", slotResult.RestartLSN, + "restartTLI", slotResult.RestartTLI) return nil } diff --git a/core/internal/client/sendwal/replslot.go b/core/internal/client/sendwal/replslot.go index debf2d32e..88559d9d3 100644 --- a/core/internal/client/sendwal/replslot.go +++ b/core/internal/client/sendwal/replslot.go @@ -46,14 +46,14 @@ func (e *ReadReplicationSlotParserError) Error() string { return e.reason } -// ParseReadReplicationSlotResult is the parsed result of the IDENTIFY_SYSTEM command. +// ParseReadReplicationSlotResult is the parsed result of the READ_REPLICATION_SLOT command. type ParseReadReplicationSlotResult struct { SlotType string RestartLSN pglogrepl.LSN RestartTLI int } -// ReadReplicationSlot executes the IDENTIFY_SYSTEM command. +// ReadReplicationSlot executes the READ_REPLICATION_SLOT command. func ReadReplicationSlot( ctx context.Context, conn *pgconn.PgConn, @@ -63,7 +63,7 @@ func ReadReplicationSlot( return ParseReadReplicationSlot(conn.Exec(ctx, sql)) } -// ParseReadReplicationSlot parses the result of the IDENTIFY_SYSTEM command. +// ParseReadReplicationSlot parses the result of the READ_REPLICATION_SLOT command. func ParseReadReplicationSlot(mrr *pgconn.MultiResultReader) (ParseReadReplicationSlotResult, error) { var rrs ParseReadReplicationSlotResult results, err := mrr.ReadAll()