From 9d4c6cdaf6438f220ead4b1f1224859fb127e41f Mon Sep 17 00:00:00 2001 From: ermakov-oleg Date: Fri, 25 Sep 2026 13:59:25 +0200 Subject: [PATCH 1/3] fix(e2e): run MinIO from the Chainguard images Claude-Session: https://claude.ai/code/session_01MK4vZ6U6TZvXVLPViN1crd Signed-off-by: ermakov-oleg --- test/e2e/internal/objectstore/minio.go | 37 +++++++++++++++++--------- 1 file changed, 25 insertions(+), 12 deletions(-) diff --git a/test/e2e/internal/objectstore/minio.go b/test/e2e/internal/objectstore/minio.go index 4fbe332..68d5b16 100644 --- a/test/e2e/internal/objectstore/minio.go +++ b/test/e2e/internal/objectstore/minio.go @@ -34,6 +34,15 @@ import ( pgbackrestApi "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/api" ) +// MinIO no longer publishes its images (minio/minio is gone from Docker Hub, quay.io/minio/minio +// requires authentication), so the e2e tests use the Chainguard builds. +const ( + minioImage = "cgr.dev/chainguard/minio:latest" + minioClientImage = "cgr.dev/chainguard/minio-client:latest-dev" + minioUID = int64(65532) + minioCertsPath = "/certs" +) + // NewMinioObjectStoreResources creates the resources required to create a Minio object store. func NewMinioObjectStoreResources(namespace, name string) *Resources { return &Resources{ @@ -69,10 +78,15 @@ func newMinioDeployment(namespace, name string) *appsv1.Deployment { }, }, Spec: corev1.PodSpec{ + // The certificates are generated as the user minio runs as, so that it can read the key. + SecurityContext: &corev1.PodSecurityContext{ + RunAsUser: ptr.To(minioUID), + RunAsGroup: ptr.To(minioUID), + FSGroup: ptr.To(minioUID), + }, // Pgbackrest only allows HTTPS connections to S3 endpoints. // That means minio must be configured in HTTPS mode. It's enabled - // automatically if the ${HOME}/.minio/certs directory contains - // certificates. + // automatically if the --certs-dir directory contains certificates. InitContainers: []corev1.Container{ { Name: "generate-certs", @@ -83,9 +97,9 @@ func newMinioDeployment(namespace, name string) *appsv1.Deployment { "-newkey", "rsa:4096", "-keyout", - "/root/.minio/certs/private.key", + minioCertsPath + "/private.key", "-out", - "/root/.minio/certs/public.crt", + minioCertsPath + "/public.crt", "-sha256", "-days", "3650", @@ -95,17 +109,16 @@ func newMinioDeployment(namespace, name string) *appsv1.Deployment { VolumeMounts: []corev1.VolumeMount{ { Name: "certs", - MountPath: "/root/.minio/certs", + MountPath: minioCertsPath, }, }, }, }, Containers: []corev1.Container{ { - Name: name, - // TODO: renovate the image - Image: "minio/minio:latest", - Args: []string{"server", "/data"}, + Name: name, + Image: minioImage, + Args: []string{"server", "/data", "--certs-dir", minioCertsPath}, Ports: []corev1.ContainerPort{ { ContainerPort: 9000, @@ -143,7 +156,7 @@ func newMinioDeployment(namespace, name string) *appsv1.Deployment { }, { Name: "certs", - MountPath: "/root/.minio/certs", + MountPath: minioCertsPath, }, }, }, @@ -194,8 +207,8 @@ func newMinioProvisioningJob(namespace, name string) *batchv1.Job { Containers: []corev1.Container{ { Name: name + "-provisioner", - Image: "minio/minio:latest", - Command: []string{"bash"}, + Image: minioClientImage, + Command: []string{"/bin/sh"}, Args: []string{ "-c", fmt.Sprintf("mc alias set local https://%s:9000 $MINIO_ROOT_USER $MINIO_ROOT_PASSWORD --insecure;\n", name) + From 569c2ad86a5350c8dc25d44f9ddd6665d7dd4e2b Mon Sep 17 00:00:00 2001 From: ermakov-oleg Date: Mon, 21 Sep 2026 15:56:37 +0200 Subject: [PATCH 2/3] feat: delete Backup objects not present in the pgBackRest catalog CloudNativePG does not delete Backup objects; the component taking the backups is expected to. The plugin runs pgbackrest expire in the repository but left the Kubernetes objects behind, so they piled up without bound and the operator kept all of them in its informer cache. Add a CatalogMaintenanceRunnable to the instance sidecar, ported from plugin-barman-cloud (internal/cnpgi/instance/retention.go) and reduced to the one operation pgBackRest does not do itself. On the current primary it periodically reads the catalog with pgbackrest info and deletes the completed Backup objects of the cluster whose backup ID is not in the catalog anymore. The catalog is acted upon only when it is for the expected stanza, reports every configured repository as readable (pgbackrest info flags an unreadable repository inside the JSON and still exits 0) and lists at least one backup: an empty catalog is also what a recreated stanza looks like. The Backup RPC now records the cluster UID, the stanza and the repository locations in status.pluginMetadata, and only objects whose recorded location equals the current one are considered; objects created by earlier plugin versions carry no location and are never deleted. Deletes carry the object UID as a precondition, Backups that completed after the catalog was read are skipped, and the list is narrowed to the cnpg.io/cluster label, paginated and jittered. New Archive field instanceSidecarConfiguration.catalogMaintenanceIntervalSeconds: default 1800, 0 disables the maintenance, otherwise at least 60. No RBAC change: the instance service account already holds list/get/delete on backups from CloudNativePG. Co-authored-by: chobostar Claude-Session: https://claude.ai/code/session_01NRY9m8yhU4NVCCBHnURaL9 Signed-off-by: ermakov-oleg --- .wordlist.txt | 1 + README.md | 29 + api/v1/archive_types.go | 8 + api/v1/zz_generated.deepcopy.go | 5 + .../pgbackrest.cnpg.opera.com_archives.yaml | 13 + internal/cmd/instance/main.go | 2 + internal/cnpgi/instance/backup.go | 8 +- .../cnpgi/instance/catalog_maintenance.go | 359 +++++++++ .../instance/catalog_maintenance_test.go | 733 ++++++++++++++++++ internal/cnpgi/instance/manager.go | 22 +- internal/cnpgi/instance/suite_test.go | 30 + internal/cnpgi/instance/types.go | 107 +++ internal/cnpgi/instance/types_test.go | 83 ++ internal/pgbackrest/catalog/catalog.go | 58 +- internal/pgbackrest/catalog/catalog_test.go | 102 +++ manifest.yaml | 13 + 16 files changed, 1561 insertions(+), 12 deletions(-) create mode 100644 internal/cnpgi/instance/catalog_maintenance.go create mode 100644 internal/cnpgi/instance/catalog_maintenance_test.go create mode 100644 internal/cnpgi/instance/suite_test.go create mode 100644 internal/cnpgi/instance/types.go create mode 100644 internal/cnpgi/instance/types_test.go diff --git a/.wordlist.txt b/.wordlist.txt index 31d4f20..b36b4ad 100644 --- a/.wordlist.txt +++ b/.wordlist.txt @@ -6,6 +6,7 @@ MinIO PITR TODO TLS +UID WAL WALs api diff --git a/README.md b/README.md index 130a4f8..781e551 100644 --- a/README.md +++ b/README.md @@ -192,6 +192,7 @@ spec: limits: memory: "2Gi" cpu: "2" + catalogMaintenanceIntervalSeconds: 1800 ``` > [!IMPORTANT] @@ -199,6 +200,34 @@ spec: > it's possible to disable key verification and use self-signed keys, using HTTP > endpoint is not possible. +### Backup objects cleanup + +pgBackRest expires backups from the repository on its own, but it does not know about +the CNPG `Backup` objects that point to them. The sidecar running on the primary deletes +the `completed` `Backup` objects of the cluster whose backup ID is not in the pgBackRest +catalog anymore, every `instanceSidecarConfiguration.catalogMaintenanceIntervalSeconds` +(default `1800`; `0` disables the maintenance, otherwise at least `60`); an unset or +unreadable `Archive` and a failed cycle fall back to 30 minutes. + +The catalog is acted upon only when it is the catalog of the expected stanza, reports +every configured repository as readable (`ok` or `no backup`; with several repositories +the stanza status `mixed` is the normal answer) and lists at least one backup. An empty +catalog is never acted upon, because a freshly recreated stanza looks exactly like one. + +A `Backup` is matched by location: the plugin records the cluster UID, the stanza and +the repository locations (`endpointURL/bucket` plus `destinationPath`, in configuration +order) in `Backup.status.pluginMetadata`, and deletes only the objects whose recorded +values all equal the current ones. Pointing an `Archive` at another bucket, adding, removing or +reordering its repositories, or renaming the stanza makes the objects recorded earlier +impossible to match: they are never deleted and have to be removed by hand, as are the objects +created by plugin versions before this feature, which carry no location at all. Each +cycle logs how many objects it kept for these reasons. + +Only `Backup` objects labeled `cnpg.io/cluster=` are considered. Both +`ScheduledBackup` and `kubectl cnpg backup` set the label; a hand-applied `Backup` +manifest without it is never cleaned up. Failed backups are not in the catalog and are +left untouched. + ### Configuring WAL Archiving Once the `Archive` is defined, you can configure a PostgreSQL cluster to archive WALs by diff --git a/api/v1/archive_types.go b/api/v1/archive_types.go index 2faf24e..33fde07 100644 --- a/api/v1/archive_types.go +++ b/api/v1/archive_types.go @@ -35,6 +35,14 @@ type InstanceSidecarConfiguration struct { // SecurityContext for the sidecar container // +optional SecurityContext *corev1.SecurityContext `json:"securityContext,omitempty"` + // The catalogMaintenanceIntervalSeconds defines how often the sidecar + // running on the primary instance compares the Backup objects of the + // cluster with the pgBackRest catalog and deletes the ones that are + // not present in the catalog anymore. 0 disables the maintenance. + // +kubebuilder:default:=1800 + // +kubebuilder:validation:XValidation:rule="self == 0 || self >= 60",message="catalogMaintenanceIntervalSeconds must be 0 (disabled) or at least 60" + // +optional + CatalogMaintenanceIntervalSeconds *int32 `json:"catalogMaintenanceIntervalSeconds,omitempty"` } // ArchiveSpec defines the desired state of Archive. diff --git a/api/v1/zz_generated.deepcopy.go b/api/v1/zz_generated.deepcopy.go index 9b8c09d..a94cd0c 100644 --- a/api/v1/zz_generated.deepcopy.go +++ b/api/v1/zz_generated.deepcopy.go @@ -132,6 +132,11 @@ func (in *InstanceSidecarConfiguration) DeepCopyInto(out *InstanceSidecarConfigu *out = new(corev1.SecurityContext) (*in).DeepCopyInto(*out) } + if in.CatalogMaintenanceIntervalSeconds != nil { + in, out := &in.CatalogMaintenanceIntervalSeconds, &out.CatalogMaintenanceIntervalSeconds + *out = new(int32) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InstanceSidecarConfiguration. diff --git a/config/crd/bases/pgbackrest.cnpg.opera.com_archives.yaml b/config/crd/bases/pgbackrest.cnpg.opera.com_archives.yaml index f9b60dd..4a605c0 100644 --- a/config/crd/bases/pgbackrest.cnpg.opera.com_archives.yaml +++ b/config/crd/bases/pgbackrest.cnpg.opera.com_archives.yaml @@ -412,6 +412,19 @@ spec: description: InstanceSidecarConfiguration defines the configuration for the sidecar that runs in the instance pods. properties: + catalogMaintenanceIntervalSeconds: + default: 1800 + description: |- + The catalogMaintenanceIntervalSeconds defines how often the sidecar + running on the primary instance compares the Backup objects of the + cluster with the pgBackRest catalog and deletes the ones that are + not present in the catalog anymore. 0 disables the maintenance. + format: int32 + type: integer + x-kubernetes-validations: + - message: catalogMaintenanceIntervalSeconds must be 0 (disabled) + or at least 60 + rule: self == 0 || self >= 60 env: description: The environment to be explicitly passed to the sidecar items: diff --git a/internal/cmd/instance/main.go b/internal/cmd/instance/main.go index cbeac95..f6c5dc9 100644 --- a/internal/cmd/instance/main.go +++ b/internal/cmd/instance/main.go @@ -18,6 +18,7 @@ func NewCmd() *cobra.Command { RunE: func(cmd *cobra.Command, _ []string) error { requiredSettings := []string{ "namespace", + "cluster-name", "pod-name", "spool-directory", } @@ -33,6 +34,7 @@ func NewCmd() *cobra.Command { } _ = viper.BindEnv("namespace", "NAMESPACE") + _ = viper.BindEnv("cluster-name", "CLUSTER_NAME") _ = viper.BindEnv("pod-name", "POD_NAME") _ = viper.BindEnv("pgdata", "PGDATA") _ = viper.BindEnv("spool-directory", "SPOOL_DIRECTORY") diff --git a/internal/cnpgi/instance/backup.go b/internal/cnpgi/instance/backup.go index 979d0b3..7c47a9d 100644 --- a/internal/cnpgi/instance/backup.go +++ b/internal/cnpgi/instance/backup.go @@ -31,7 +31,6 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" pgbackrestv1 "github.com/operasoftware/cnpg-plugin-pgbackrest/api/v1" - "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/cnpgi/metadata" "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/cnpgi/operator/config" pgbackrestBackup "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/backup" "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/catalog" @@ -160,10 +159,7 @@ func (b BackupServiceImplementation) Backup( EndLsn: executedBackupInfo.Backups[0].LSN.Stop, InstanceId: b.InstanceName, Online: true, - Metadata: map[string]string{ - "version": metadata.Data.Version, - "name": metadata.Data.Name, - "displayName": metadata.Data.DisplayName, - }, + Metadata: newBackupResultMetadata( + configuration.Cluster.UID, configuration.Stanza, &archive.Spec.Configuration).toMap(), }, nil } diff --git a/internal/cnpgi/instance/catalog_maintenance.go b/internal/cnpgi/instance/catalog_maintenance.go new file mode 100644 index 0000000..490fc0e --- /dev/null +++ b/internal/cnpgi/instance/catalog_maintenance.go @@ -0,0 +1,359 @@ +/* +Copyright The CloudNativePG Contributors +Copyright 2025, Opera Norway AS + +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. +*/ + +package instance + +import ( + "context" + "errors" + "fmt" + "math/rand/v2" + "slices" + "strings" + "time" + + cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1" + cnpgutils "github.com/cloudnative-pg/cloudnative-pg/pkg/utils" + "github.com/cloudnative-pg/machinery/pkg/log" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + + pgbackrestv1 "github.com/operasoftware/cnpg-plugin-pgbackrest/api/v1" + "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/cnpgi/metadata" + "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/cnpgi/operator/config" + "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/catalog" + pgbackrestCommand "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/command" + pgbackrestCredentials "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/credentials" + "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/utils" +) + +const ( + // defaultCatalogMaintenanceInterval mirrors the CRD default; it also paces the loop + // when the Archive cannot be read, when a cycle fails and while maintenance is disabled. + defaultCatalogMaintenanceInterval = 30 * time.Minute + + // backupListPageSize keeps a maintenance cycle within the small memory + // limit of the sidecar: a namespace can hold tens of thousands of Backups. + backupListPageSize = 500 +) + +// errEmptyCatalog is a sentinel because the caller logs it below the other rejections: +// an empty catalog is the normal state of a cluster whose first backup is still to come. +var errEmptyCatalog = errors.New("the catalog is empty") + +// catalogUsable tells whether the catalog can decide that a backup absent from it is +// gone. Only a complete answer for the expected stanza with at least one backup can: +// an empty catalog is also what a freshly recreated stanza looks like. +func catalogUsable(c *catalog.Catalog, stanza string, repositories int) error { + if c.Stanza != stanza { + return fmt.Errorf("the catalog describes the stanza %q instead of %q", c.Stanza, stanza) + } + if !c.DescribesRepositories(repositories) { + return fmt.Errorf( + "the catalog does not describe all %d configured repositories (stanza status %d %q; repositories: %s)", + repositories, c.Status.Code, c.Status.Message, describeRepoStatuses(c)) + } + if len(c.Backups) == 0 { + return errEmptyCatalog + } + return nil +} + +// describeRepoStatuses renders the per-repository status codes as "key=code", so that +// the rejection names the repository that could not be read, not only that one could not. +func describeRepoStatuses(c *catalog.Catalog) string { + statuses := make([]string, len(c.Repos)) + for idx := range c.Repos { + statuses[idx] = fmt.Sprintf("%d=%d", c.Repos[idx].Key, c.Repos[idx].Status.Code) + } + return strings.Join(statuses, " ") +} + +// deleteBackupsNotInCatalog deletes the completed Backup objects of the +// given cluster whose backup ID is not in the pgBackRest catalog anymore. +// +// The catalog is the source of truth: pgBackRest expires backups on its +// own after every backup, so a Backup object whose ID has disappeared +// from `pgbackrest info` points to data that no longer exists. +func deleteBackupsNotInCatalog( + ctx context.Context, + cli client.Client, + cluster *cnpgv1.Cluster, + current backupResultMetadata, + backupIDs []string, + catalogReadAt time.Time, +) error { + contextLogger := log.FromContext(ctx) + contextLogger.Debug("Checking the catalog to delete backups not present anymore") + + var errs []error + keptElsewhere := 0 + continueToken := "" + restarted := false + for { + var backups cnpgv1.BackupList + if err := cli.List(ctx, &backups, + client.InNamespace(cluster.GetNamespace()), + // CloudNativePG labels every Backup it creates with its cluster; hand-written + // Backups without the label are left alone rather than listed fleet-wide. + client.MatchingLabels{cnpgutils.ClusterLabelName: cluster.GetName()}, + client.Limit(backupListPageSize), + client.Continue(continueToken), + ); err != nil { + if apierrors.IsResourceExpired(err) && !restarted { + // The apiserver compacted the snapshot behind the token; start over once. + // The restarted pass sees every object again, so what the aborted one + // counted and failed to delete is re-attempted rather than reported twice. + restarted = true + continueToken = "" + errs = nil + keptElsewhere = 0 + continue + } + return fmt.Errorf("while getting backups: %w", err) + } + + for idx := range backups.Items { + backup := &backups.Items[idx] + if !concernsCluster(backup, cluster.GetName()) { + continue + } + + if !useSameBackupLocation(&backup.Status, current) { + keptElsewhere++ + continue + } + + if slices.Contains(backupIDs, backup.Status.BackupID) { + continue + } + + // A backup that completed after the catalog was read cannot be in it yet. + if backup.Status.StoppedAt != nil && backup.Status.StoppedAt.After(catalogReadAt) { + continue + } + + contextLogger.Info("Deleting backup not in the catalog", + "backup", backup.Name, "backupID", backup.Status.BackupID) + // The listed object may have been replaced by a same-named one for a live + // backup; the precondition makes that a Conflict instead of a wrong delete. + preconditions := client.Preconditions{UID: &backup.UID} + if err := cli.Delete(ctx, backup, preconditions); err != nil && + !apierrors.IsNotFound(err) && !apierrors.IsConflict(err) { + errs = append(errs, fmt.Errorf( + "while deleting backup %s/%s: %w", backup.Namespace, backup.Name, err)) + } + } + + continueToken = backups.Continue + if continueToken == "" { + break + } + } + + if keptElsewhere > 0 { + contextLogger.Info("Kept Backup objects recorded against another location or without location metadata", + "count", keptElsewhere) + } + + if len(errs) > 0 { + return fmt.Errorf("got errors while deleting Backups not in the catalog: %w", errors.Join(errs...)) + } + + return nil +} + +// concernsCluster tells whether the catalog can say anything about the object at all: +// only a completed backup of this cluster carries an ID to look up in it. +func concernsCluster(backup *cnpgv1.Backup, clusterName string) bool { + return backup.Spec.Cluster.Name == clusterName && + backup.Status.Phase == cnpgv1.BackupPhaseCompleted && + backup.Status.BackupID != "" +} + +// useSameBackupLocation tells whether the backup was taken by this plugin for this +// cluster incarnation, stanza and repositories. Any entry missing from the recorded +// metadata (older plugin versions) means "unknown location" and keeps the object. +func useSameBackupLocation(backup *cnpgv1.BackupStatus, current backupResultMetadata) bool { + if backup.Method != cnpgv1.BackupMethodPlugin { + return false + } + + recorded := newBackupResultMetadataFromMap(backup.PluginMetadata) + return recorded.pluginName != "" && recorded.pluginName == current.pluginName && + recorded.clusterUID != "" && recorded.clusterUID == current.clusterUID && + recorded.stanza != "" && recorded.stanza == current.stanza && + recorded.repositories != "" && recorded.repositories == current.repositories +} + +// CatalogMaintenanceRunnable periodically reconciles the Backup objects +// of the cluster with the pgBackRest catalog. pgBackRest expires backups +// on its own after every backup; this runnable removes the Kubernetes +// objects left behind. +type CatalogMaintenanceRunnable struct { + Client client.Client + ClusterKey types.NamespacedName + CurrentPodName string +} + +// Start runs the maintenance loop until the context is cancelled. +func (c *CatalogMaintenanceRunnable) Start(ctx context.Context) error { + contextLogger := log.FromContext(ctx) + contextLogger.Info("Starting catalog maintenance runnable") + + // Spread the fleet over the interval: without this, a rollout would fire + // the first Backup list of every primary at the very same second. + select { + case <-time.After(initialDelay()): + case <-ctx.Done(): + return nil + } + + for { + period, err := c.cycle(ctx) + if err != nil { + contextLogger.Error(err, "Catalog maintenance failed") + } + + select { + case <-time.After(nextInterval(period)): + case <-ctx.Done(): + return nil + } + } +} + +// initialDelay is how long the runnable waits before its first cycle, spreading +// the fleet uniformly over one full default interval. +func initialDelay() time.Duration { + return rand.N(defaultCatalogMaintenanceInterval) // #nosec G404 -- spreading load, not a secret +} + +// nextInterval guards against a zero or negative period, which would +// otherwise spin `pgbackrest info` against the object store in a tight loop. +func nextInterval(period time.Duration) time.Duration { + if period <= 0 { + return defaultCatalogMaintenanceInterval + } + return period +} + +// cycle runs one maintenance pass and returns the interval to wait before +// the next one. A zero interval means "use the default". It runs maintenance +// on the current primary only, mirroring where backups are taken. +func (c *CatalogMaintenanceRunnable) cycle(ctx context.Context) (time.Duration, error) { + contextLogger := log.FromContext(ctx) + + var cluster cnpgv1.Cluster + if err := c.Client.Get(ctx, c.ClusterKey, &cluster); err != nil { + return 0, err + } + + enabledPlugins := cnpgv1.GetPluginConfigurationEnabledPluginNames(cluster.Spec.Plugins) + if !slices.Contains(enabledPlugins, metadata.PluginName) { + contextLogger.Debug("Skipping catalog maintenance: plugin is not enabled for backups") + // The Archive is not read here, so there is no configured interval to honor. + return defaultCatalogMaintenanceInterval, nil + } + + if cluster.Status.CurrentPrimary != c.CurrentPodName { + contextLogger.Debug("Skipping catalog maintenance, not the current primary", + "currentPrimary", cluster.Status.CurrentPrimary, "podName", c.CurrentPodName) + // The Archive is not read here, so there is no configured interval to honor. + return defaultCatalogMaintenanceInterval, nil + } + + configuration := config.NewFromCluster(&cluster) + if configuration == nil || len(configuration.PgbackrestObjectName) == 0 { + return 0, fmt.Errorf("invalid configuration, missing pgbackrestObjectName parameter") + } + + var archive pgbackrestv1.Archive + if err := c.Client.Get(ctx, configuration.GetArchiveObjectKey(), &archive); err != nil { + return 0, err + } + + interval, enabled := maintenanceInterval(&archive) + if !enabled { + contextLogger.Debug("Catalog maintenance is disabled for this Archive", "archive", archive.Name) + return interval, nil + } + + if err := c.maintenance(ctx, &cluster, &archive, configuration); err != nil { + return 0, err + } + + return interval, nil +} + +// maintenanceInterval returns the configured interval and whether maintenance is +// enabled at all (0 disables it; nil means the CRD default). +func maintenanceInterval(archive *pgbackrestv1.Archive) (time.Duration, bool) { + seconds := archive.Spec.InstanceSidecarConfiguration.CatalogMaintenanceIntervalSeconds + if seconds == nil { + return defaultCatalogMaintenanceInterval, true + } + if *seconds == 0 { + return defaultCatalogMaintenanceInterval, false + } + return time.Duration(*seconds) * time.Second, true +} + +// maintenance reads the catalog and deletes the stale Backup objects. +// cycle runs it on the current primary only. +func (c *CatalogMaintenanceRunnable) maintenance( + ctx context.Context, + cluster *cnpgv1.Cluster, + archive *pgbackrestv1.Archive, + configuration *config.PluginConfiguration, +) error { + contextLogger := log.FromContext(ctx) + + env, err := pgbackrestCredentials.EnvSetBackupCloudCredentials( + ctx, + c.Client, + archive.Namespace, + &archive.Spec.Configuration, + utils.SanitizedEnviron(), + ) + if err != nil { + return fmt.Errorf("while setting backup cloud credentials: %w", err) + } + + catalogReadAt := time.Now() + backupList, err := pgbackrestCommand.GetBackupList(ctx, &archive.Spec.Configuration, configuration.Stanza, env) + if err != nil { + return fmt.Errorf("while reading the backup list: %w", err) + } + + if err := catalogUsable(backupList, configuration.Stanza, len(archive.Spec.Configuration.Repositories)); err != nil { + if errors.Is(err, errEmptyCatalog) { + contextLogger.Info( + "Skipping catalog maintenance: the catalog is empty — a cluster without a completed backup yet, "+ + "or a recreated stanza", + "stanza", configuration.Stanza) + return nil + } + contextLogger.Warning("Skipping catalog maintenance", "reason", err.Error(), "stanza", configuration.Stanza) + return nil + } + + current := newBackupResultMetadata(cluster.UID, configuration.Stanza, &archive.Spec.Configuration) + return deleteBackupsNotInCatalog( + ctx, c.Client, cluster, current, backupList.GetBackupIDs(), catalogReadAt) +} diff --git a/internal/cnpgi/instance/catalog_maintenance_test.go b/internal/cnpgi/instance/catalog_maintenance_test.go new file mode 100644 index 0000000..4dccce3 --- /dev/null +++ b/internal/cnpgi/instance/catalog_maintenance_test.go @@ -0,0 +1,733 @@ +/* +Copyright The CloudNativePG Contributors +Copyright 2025, Opera Norway AS + +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. +*/ + +package instance + +import ( + "context" + "errors" + "time" + + cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1" + cnpgutils "github.com/cloudnative-pg/cloudnative-pg/pkg/utils" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + pgbackrestv1 "github.com/operasoftware/cnpg-plugin-pgbackrest/api/v1" + "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/cnpgi/metadata" + pgbackrestApi "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/api" + "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/catalog" +) + +const ( + testNamespace = "default" + testClusterUID = types.UID("cluster-uid-1") + testStanza = "cluster-example" +) + +var ( + testConfiguration = pgbackrestApi.PgbackrestConfiguration{ + Repositories: []pgbackrestApi.PgbackrestRepository{ + {EndpointURL: "https://s3.example.com", Bucket: "bucket", DestinationPath: "/cluster-example"}, + }, + } + testLocation = newBackupResultMetadata(testClusterUID, testStanza, &testConfiguration) +) + +func newTestCluster() *cnpgv1.Cluster { + return &cnpgv1.Cluster{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: testNamespace, + Name: "cluster-example", + UID: testClusterUID, + }, + } +} + +// newTestBackup builds a completed plugin backup of the given cluster with +// this plugin's metadata. Tests override single fields to build the +// negative cases. The cluster label mirrors what CloudNativePG sets on +// every Backup it creates. +func newTestBackup(name, clusterName, backupID string) *cnpgv1.Backup { + return &cnpgv1.Backup{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: testNamespace, + Name: name, + Labels: map[string]string{cnpgutils.ClusterLabelName: clusterName}, + }, + Spec: cnpgv1.BackupSpec{ + Cluster: cnpgv1.LocalObjectReference{Name: clusterName}, + }, + Status: cnpgv1.BackupStatus{ + Phase: cnpgv1.BackupPhaseCompleted, + Method: cnpgv1.BackupMethodPlugin, + BackupID: backupID, + PluginMetadata: testLocation.toMap(), + }, + } +} + +func remainingBackupNames(ctx context.Context, cli client.Client) []string { + var list cnpgv1.BackupList + ExpectWithOffset(1, cli.List(ctx, &list, client.InNamespace(testNamespace))).To(Succeed()) + names := make([]string, 0, len(list.Items)) + for _, b := range list.Items { + names = append(names, b.Name) + } + return names +} + +var _ = Describe("catalogUsable", func() { + usable := func() *catalog.Catalog { + return &catalog.Catalog{ + Stanza: testStanza, + Status: catalog.PgbackrestStanzaStatus{Code: catalog.StanzaStatusCodeOk}, + Repos: []catalog.PgbackrestRepo{{Key: 1, Status: catalog.PgbackrestStanzaStatus{Code: catalog.StanzaStatusCodeOk}}}, + Backups: []catalog.PgbackrestBackup{{ID: "20250101-000000F"}}, + } + } + + It("accepts a complete catalog of the expected stanza with at least one backup", func() { + Expect(catalogUsable(usable(), testStanza, 1)).To(Succeed()) + }) + + It("rejects a catalog of another stanza", func() { + Expect(catalogUsable(usable(), "other-stanza", 1)).To(MatchError(ContainSubstring("stanza"))) + }) + + It("rejects a catalog that does not describe every configured repository", func() { + c := usable() + c.Repos = append(c.Repos, catalog.PgbackrestRepo{Key: 2, Status: catalog.PgbackrestStanzaStatus{Code: 99}}) + c.Status.Code = catalog.StanzaStatusCodeMixed + Expect(catalogUsable(c, testStanza, 2)).To(MatchError(ContainSubstring("repositor"))) + }) + + It("reports the status of every repository when the catalog is incomplete", func() { + c := usable() + c.Repos = append(c.Repos, catalog.PgbackrestRepo{Key: 2, Status: catalog.PgbackrestStanzaStatus{Code: 99}}) + c.Status.Code = catalog.StanzaStatusCodeMixed + Expect(catalogUsable(c, testStanza, 2)).To(MatchError(ContainSubstring("1=0 2=99"))) + }) + + It("rejects a catalog whose stanza has not been created yet", func() { + c := usable() + c.Status.Code = catalog.StanzaStatusCodeMissing + c.Repos[0].Status.Code = catalog.StanzaStatusCodeMissing + c.Backups = nil + Expect(catalogUsable(c, testStanza, 1)).To(MatchError(ContainSubstring("repositor"))) + }) + + It("rejects an empty catalog", func() { + c := usable() + c.Backups = nil + c.Status.Code = catalog.StanzaStatusCodeNoBackup + c.Repos[0].Status.Code = catalog.StanzaStatusCodeNoBackup + Expect(catalogUsable(c, testStanza, 1)).To(MatchError(errEmptyCatalog)) + }) +}) + +var _ = Describe("useSameBackupLocation", func() { + cluster := newTestCluster() + + It("accepts a plugin backup of this cluster taken by this plugin", func() { + status := newTestBackup("b", cluster.Name, "id").Status + Expect(useSameBackupLocation(&status, testLocation)).To(BeTrue()) + }) + + It("rejects non-plugin backup methods", func() { + status := newTestBackup("b", cluster.Name, "id").Status + status.Method = cnpgv1.BackupMethodBarmanObjectStore + Expect(useSameBackupLocation(&status, testLocation)).To(BeFalse()) + }) + + It("rejects a different cluster UID", func() { + status := newTestBackup("b", cluster.Name, "id").Status + status.PluginMetadata = newBackupResultMetadata(types.UID("other-uid"), testStanza, &testConfiguration).toMap() + Expect(useSameBackupLocation(&status, testLocation)).To(BeFalse()) + }) + + It("rejects a different plugin", func() { + status := newTestBackup("b", cluster.Name, "id").Status + status.PluginMetadata["pluginName"] = "barman-cloud.cloudnative-pg.io" + Expect(useSameBackupLocation(&status, testLocation)).To(BeFalse()) + }) + + It("rejects backups without plugin metadata", func() { + status := newTestBackup("b", cluster.Name, "id").Status + status.PluginMetadata = nil + Expect(useSameBackupLocation(&status, testLocation)).To(BeFalse()) + }) + + It("rejects backups from older plugin versions without clusterUID", func() { + status := newTestBackup("b", cluster.Name, "id").Status + status.PluginMetadata = map[string]string{ + "version": "0.5.2", + "name": metadata.Data.Name, + "displayName": metadata.Data.DisplayName, + } + Expect(useSameBackupLocation(&status, testLocation)).To(BeFalse()) + }) + + It("rejects a backup taken against another stanza", func() { + status := newTestBackup("b", cluster.Name, "id").Status + status.PluginMetadata = newBackupResultMetadata(testClusterUID, "another-stanza", &testConfiguration).toMap() + Expect(useSameBackupLocation(&status, testLocation)).To(BeFalse()) + }) + + It("rejects backups from older plugin versions without stanza", func() { + status := newTestBackup("b", cluster.Name, "id").Status + delete(status.PluginMetadata, "stanza") + Expect(useSameBackupLocation(&status, testLocation)).To(BeFalse()) + }) + + It("rejects a backup written to other repositories", func() { + status := newTestBackup("b", cluster.Name, "id").Status + status.PluginMetadata["repositories"] = "https://other/bucket/cluster-example" + Expect(useSameBackupLocation(&status, testLocation)).To(BeFalse()) + }) + + It("rejects backups from older plugin versions without repositories", func() { + status := newTestBackup("b", cluster.Name, "id").Status + delete(status.PluginMetadata, "repositories") + Expect(useSameBackupLocation(&status, testLocation)).To(BeFalse()) + }) +}) + +var _ = Describe("deleteBackupsNotInCatalog", func() { + var cluster *cnpgv1.Cluster + + BeforeEach(func() { + cluster = newTestCluster() + }) + + It("deletes completed backups whose ID is not in the catalog and keeps the others", func(ctx SpecContext) { + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects( + newTestBackup("in-catalog", cluster.Name, "20250101-000000F"), + newTestBackup("stale", cluster.Name, "20240101-000000F"), + ).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, []string{"20250101-000000F"}, time.Now())).To(Succeed()) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("in-catalog")) + }) + + It("keeps backups of other clusters", func(ctx SpecContext) { + // Labelled for our cluster (so it is listed) but spec'd for another one, to + // exercise the Spec.Cluster.Name guard rather than the label filter. + other := newTestBackup("other", cluster.Name, "stale-id") + other.Spec.Cluster.Name = "another-cluster" + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(other).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("other")) + }) + + It("keeps backups that are not completed", func(ctx SpecContext) { + failed := newTestBackup("failed", cluster.Name, "") + failed.Status.Phase = cnpgv1.BackupPhaseFailed + started := newTestBackup("started", cluster.Name, "") + started.Status.Phase = cnpgv1.BackupPhaseStarted + pending := newTestBackup("pending", cluster.Name, "") + pending.Status.Phase = cnpgv1.BackupPhasePending + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(failed, started, pending).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("failed", "started", "pending")) + }) + + It("keeps backups with a different clusterUID", func(ctx SpecContext) { + recreated := newTestBackup("old-incarnation", cluster.Name, "stale-id") + recreated.Status.PluginMetadata = newBackupResultMetadata(types.UID("previous-uid"), testStanza, &testConfiguration).toMap() + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(recreated).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("old-incarnation")) + }) + + It("keeps a backup that completed after the catalog was read", func(ctx SpecContext) { + catalogReadAt := time.Now() + fresh := newTestBackup("fresh", cluster.Name, "fresh-id") + fresh.Status.StoppedAt = &metav1.Time{Time: catalogReadAt.Add(time.Minute)} + older := newTestBackup("older", cluster.Name, "older-id") + older.Status.StoppedAt = &metav1.Time{Time: catalogReadAt.Add(-time.Minute)} + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(fresh, older).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, catalogReadAt)).To(Succeed()) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("fresh")) + }) + + It("keeps backups taken against another stanza", func(ctx SpecContext) { + moved := newTestBackup("other-stanza", cluster.Name, "stale-id") + moved.Status.PluginMetadata = newBackupResultMetadata(testClusterUID, "another-stanza", &testConfiguration).toMap() + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(moved).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("other-stanza")) + }) + + It("keeps backups written to other repositories", func(ctx SpecContext) { + moved := newTestBackup("other-repositories", cluster.Name, "stale-id") + otherConfiguration := pgbackrestApi.PgbackrestConfiguration{ + Repositories: []pgbackrestApi.PgbackrestRepository{ + {EndpointURL: "https://other", Bucket: "bucket", DestinationPath: "/cluster-example"}, + }, + } + moved.Status.PluginMetadata = newBackupResultMetadata(testClusterUID, testStanza, &otherConfiguration).toMap() + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(moved).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("other-repositories")) + }) + + It("keeps backups without plugin metadata", func(ctx SpecContext) { + legacy := newTestBackup("legacy", cluster.Name, "stale-id") + legacy.Status.PluginMetadata = nil + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(legacy).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("legacy")) + }) + + It("keeps every backup whose recorded location does not match the current one", func(ctx SpecContext) { + // The two shapes the cycle reports together: a backup written elsewhere and + // one from a plugin version that recorded no location at all. + moved := newTestBackup("other-repositories", cluster.Name, "stale-1") + otherConfiguration := pgbackrestApi.PgbackrestConfiguration{ + Repositories: []pgbackrestApi.PgbackrestRepository{ + {EndpointURL: "https://other", Bucket: "bucket", DestinationPath: "/cluster-example"}, + }, + } + moved.Status.PluginMetadata = newBackupResultMetadata(testClusterUID, testStanza, &otherConfiguration).toMap() + legacy := newTestBackup("legacy", cluster.Name, "stale-2") + legacy.Status.PluginMetadata = nil + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(moved, legacy).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("other-repositories", "legacy")) + }) + + It("keeps backups taken with another method", func(ctx SpecContext) { + barman := newTestBackup("barman", cluster.Name, "stale-id") + barman.Status.Method = cnpgv1.BackupMethodBarmanObjectStore + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(barman).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("barman")) + }) + + It("ignores NotFound on delete", func(ctx SpecContext) { + cli := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(newTestBackup("stale", cluster.Name, "stale-id")). + WithInterceptorFuncs(interceptor.Funcs{ + Delete: func(_ context.Context, _ client.WithWatch, obj client.Object, _ ...client.DeleteOption) error { + return apierrors.NewNotFound( + schema.GroupResource{Group: "postgresql.cnpg.io", Resource: "backups"}, obj.GetName()) + }, + }).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + }) + + It("ignores Conflict on delete", func(ctx SpecContext) { + cli := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(newTestBackup("recreated", cluster.Name, "stale-id")). + WithInterceptorFuncs(interceptor.Funcs{ + Delete: func(_ context.Context, _ client.WithWatch, obj client.Object, _ ...client.DeleteOption) error { + return apierrors.NewConflict( + schema.GroupResource{Group: "postgresql.cnpg.io", Resource: "backups"}, obj.GetName(), + errors.New("the UID in the precondition does not match")) + }, + }).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + }) + + It("deletes with the UID of the listed object as a precondition", func(ctx SpecContext) { + var seen *types.UID + stale := newTestBackup("stale", cluster.Name, "stale-id") + stale.UID = types.UID("uid-stale") + cli := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(stale). + WithInterceptorFuncs(interceptor.Funcs{ + Delete: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.DeleteOption) error { + options := &client.DeleteOptions{} + options.ApplyOptions(opts) + if options.Preconditions != nil { + seen = options.Preconditions.UID + } + return c.Delete(ctx, obj, opts...) + }, + }).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(seen).NotTo(BeNil()) + Expect(*seen).NotTo(BeEmpty()) + }) + + It("keeps completed backups that carry no backup ID", func(ctx SpecContext) { + cli := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(newTestBackup("no-id", cluster.Name, "")).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("no-id")) + }) + + It("keeps deleting after a failure and returns an aggregated error", func(ctx SpecContext) { + cli := fake.NewClientBuilder().WithScheme(scheme). + WithObjects( + newTestBackup("fails", cluster.Name, "stale-1"), + newTestBackup("succeeds", cluster.Name, "stale-2"), + ). + WithInterceptorFuncs(interceptor.Funcs{ + Delete: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.DeleteOption) error { + if obj.GetName() == "fails" { + return errors.New("boom") + } + return c.Delete(ctx, obj, opts...) + }, + }).Build() + + err := deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now()) + + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("default/fails")) + Expect(err.Error()).To(ContainSubstring("boom")) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("fails")) + }) + + It("lists only the Backups labelled with the cluster name", func(ctx SpecContext) { + var selector string + cli := fake.NewClientBuilder().WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error { + options := &client.ListOptions{} + options.ApplyOptions(opts) + if options.LabelSelector != nil { + selector = options.LabelSelector.String() + } + return c.List(ctx, list, opts...) + }, + }).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(selector).To(Equal("cnpg.io/cluster=cluster-example")) + }) + + It("walks every page of the list", func(ctx SpecContext) { + // The fake client does not paginate; emulate two pages with the interceptor. + calls := 0 + cli := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(newTestBackup("stale-1", cluster.Name, "s1"), newTestBackup("stale-2", cluster.Name, "s2")). + WithInterceptorFuncs(interceptor.Funcs{ + List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error { + calls++ + if err := c.List(ctx, list, opts...); err != nil { + return err + } + backups := list.(*cnpgv1.BackupList) + if calls == 1 { + // First page: one object and a token. The second call returns whatever + // is still there (the first object is gone by then) and no token. + backups.Items = backups.Items[:1] + backups.Continue = "page-2" + } else { + backups.Continue = "" + } + return nil + }, + }).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(calls).To(Equal(2)) + Expect(remainingBackupNames(ctx, cli)).To(BeEmpty()) + }) + + It("restarts the pass once when the continue token expired", func(ctx SpecContext) { + calls := 0 + cli := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(newTestBackup("stale", cluster.Name, "s1")). + WithInterceptorFuncs(interceptor.Funcs{ + List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error { + calls++ + options := &client.ListOptions{} + options.ApplyOptions(opts) + switch { + case calls == 1: + if err := c.List(ctx, list, opts...); err != nil { + return err + } + list.(*cnpgv1.BackupList).Items = nil + list.(*cnpgv1.BackupList).Continue = "expired" + return nil + case options.Continue == "expired": + return apierrors.NewResourceExpired("the continue token has expired") + default: + return c.List(ctx, list, opts...) + } + }, + }).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(calls).To(Equal(3)) + Expect(remainingBackupNames(ctx, cli)).To(BeEmpty()) + }) + + It("forgets the delete errors of the pass aborted by an expired continue token", func(ctx SpecContext) { + listCalls := 0 + deleteCalls := 0 + cli := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(newTestBackup("stale", cluster.Name, "s1")). + WithInterceptorFuncs(interceptor.Funcs{ + List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error { + listCalls++ + options := &client.ListOptions{} + options.ApplyOptions(opts) + switch { + case listCalls == 1: + if err := c.List(ctx, list, opts...); err != nil { + return err + } + list.(*cnpgv1.BackupList).Continue = "expired" + return nil + case options.Continue == "expired": + return apierrors.NewResourceExpired("the continue token has expired") + default: + return c.List(ctx, list, opts...) + } + }, + Delete: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.DeleteOption) error { + deleteCalls++ + if deleteCalls == 1 { + return errors.New("boom") + } + return c.Delete(ctx, obj, opts...) + }, + }).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(deleteCalls).To(Equal(2)) + Expect(remainingBackupNames(ctx, cli)).To(BeEmpty()) + }) + + It("keeps backups without the cluster label", func(ctx SpecContext) { + unlabelled := newTestBackup("stale-unlabelled", cluster.Name, "stale-id") + unlabelled.Labels = nil + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(unlabelled).Build() + + Expect(deleteBackupsNotInCatalog(ctx, cli, cluster, testLocation, nil, time.Now())).To(Succeed()) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("stale-unlabelled")) + }) +}) + +func newTestArchive(intervalSeconds *int32) *pgbackrestv1.Archive { + return &pgbackrestv1.Archive{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: testNamespace, + Name: "archive", + }, + Spec: pgbackrestv1.ArchiveSpec{ + Configuration: testConfiguration, + InstanceSidecarConfiguration: pgbackrestv1.InstanceSidecarConfiguration{ + CatalogMaintenanceIntervalSeconds: intervalSeconds, + }, + }, + } +} + +// newTestClusterWithPlugin enables this plugin on the cluster and points it +// at the given archive. primary is Status.CurrentPrimary. +func newTestClusterWithPlugin(archiveName, primary string) *cnpgv1.Cluster { + cluster := newTestCluster() + params := map[string]string{} + if archiveName != "" { + params["pgbackrestObjectName"] = archiveName + } + cluster.Spec.Plugins = []cnpgv1.PluginConfiguration{{ + Name: metadata.PluginName, + Parameters: params, + }} + cluster.Status.CurrentPrimary = primary + return cluster +} + +var _ = Describe("nextInterval", func() { + It("keeps a positive period", func() { + Expect(nextInterval(30 * time.Minute)).To(Equal(30 * time.Minute)) + }) + + It("falls back to the default for zero", func() { + Expect(nextInterval(0)).To(Equal(defaultCatalogMaintenanceInterval)) + }) + + It("falls back to the default for negative values", func() { + Expect(nextInterval(-5 * time.Second)).To(Equal(defaultCatalogMaintenanceInterval)) + }) +}) + +var _ = Describe("maintenanceInterval", func() { + It("defaults to the CRD default and stays enabled when unset", func() { + interval, enabled := maintenanceInterval(newTestArchive(nil)) + Expect(enabled).To(BeTrue()) + Expect(interval).To(Equal(defaultCatalogMaintenanceInterval)) + }) + + It("honours a configured interval", func() { + interval, enabled := maintenanceInterval(newTestArchive(ptr.To[int32](900))) + Expect(enabled).To(BeTrue()) + Expect(interval).To(Equal(15 * time.Minute)) + }) + + It("disables maintenance and returns the default when set to zero", func() { + interval, enabled := maintenanceInterval(newTestArchive(ptr.To[int32](0))) + Expect(enabled).To(BeFalse()) + Expect(interval).To(Equal(defaultCatalogMaintenanceInterval)) + }) +}) + +var _ = Describe("CatalogMaintenanceRunnable.cycle", func() { + const podName = "cluster-example-1" + + newRunnable := func(cli client.Client) *CatalogMaintenanceRunnable { + return &CatalogMaintenanceRunnable{ + Client: cli, + ClusterKey: types.NamespacedName{Namespace: testNamespace, Name: "cluster-example"}, + CurrentPodName: podName, + } + } + + It("returns an error when the cluster does not exist", func(ctx SpecContext) { + cli := fake.NewClientBuilder().WithScheme(scheme).Build() + + period, err := newRunnable(cli).cycle(ctx) + + Expect(err).To(HaveOccurred()) + Expect(period).To(BeZero()) + }) + + It("skips the cycle when the plugin is not enabled on the cluster", func(ctx SpecContext) { + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(newTestCluster()).Build() + + period, err := newRunnable(cli).cycle(ctx) + + Expect(err).NotTo(HaveOccurred()) + Expect(period).To(Equal(defaultCatalogMaintenanceInterval)) + }) + + It("returns an error when the archive name is missing", func(ctx SpecContext) { + cli := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(newTestClusterWithPlugin("", podName)).Build() + + _, err := newRunnable(cli).cycle(ctx) + + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("pgbackrestObjectName")) + }) + + It("returns an error when the archive does not exist", func(ctx SpecContext) { + cli := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(newTestClusterWithPlugin("archive", podName)).Build() + + _, err := newRunnable(cli).cycle(ctx) + + Expect(apierrors.IsNotFound(err)).To(BeTrue()) + }) + + It("skips maintenance on a replica", func(ctx SpecContext) { + // On a replica no pgbackrest command runs, so the fake client is enough + // to drive the whole cycle. The stale backup must survive. + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects( + newTestClusterWithPlugin("archive", "cluster-example-2"), + newTestArchive(ptr.To[int32](900)), + newTestBackup("stale", "cluster-example", "stale-id"), + ).Build() + + period, err := newRunnable(cli).cycle(ctx) + + Expect(err).NotTo(HaveOccurred()) + // The Archive is not read on a replica, so the interval it configures never comes into play. + Expect(period).To(Equal(defaultCatalogMaintenanceInterval)) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("stale")) + }) + + It("does not read the Archive on a replica", func(ctx SpecContext) { + archiveReads := 0 + cli := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(newTestClusterWithPlugin("archive", "cluster-example-2"), newTestArchive(ptr.To[int32](900))). + WithInterceptorFuncs(interceptor.Funcs{ + Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, isArchive := obj.(*pgbackrestv1.Archive); isArchive { + archiveReads++ + } + return c.Get(ctx, key, obj, opts...) + }, + }).Build() + + period, err := newRunnable(cli).cycle(ctx) + Expect(err).NotTo(HaveOccurred()) + Expect(archiveReads).To(BeZero()) + Expect(period).To(Equal(defaultCatalogMaintenanceInterval)) + }) + + It("does nothing when maintenance is disabled", func(ctx SpecContext) { + // Primary pod, disabled interval, a stale backup: no pgbackrest call is made (the + // test environment has none), the backup survives, the Archive is polled again later. + cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects( + newTestClusterWithPlugin("archive", podName), newTestArchive(ptr.To[int32](0)), + newTestBackup("stale", "cluster-example", "stale-id")).Build() + period, err := newRunnable(cli).cycle(ctx) + Expect(err).NotTo(HaveOccurred()) + Expect(period).To(Equal(defaultCatalogMaintenanceInterval)) + Expect(remainingBackupNames(ctx, cli)).To(ConsistOf("stale")) + }) +}) + +var _ = Describe("initialDelay", func() { + It("stays inside the default maintenance interval", func() { + for range 100 { + delay := initialDelay() + Expect(delay).To(BeNumerically(">=", 0)) + Expect(delay).To(BeNumerically("<", defaultCatalogMaintenanceInterval)) + } + }) +}) + +var _ = Describe("CatalogMaintenanceRunnable.Start", func() { + It("returns when the context is cancelled", func() { + // Start must return nil promptly once the context is cancelled, whether + // it is sleeping or between cycles. + cli := fake.NewClientBuilder().WithScheme(scheme).Build() + runnable := &CatalogMaintenanceRunnable{ + Client: cli, + ClusterKey: types.NamespacedName{Namespace: testNamespace, Name: "cluster-example"}, + CurrentPodName: "cluster-example-1", + } + ctx, cancel := context.WithCancel(context.Background()) + + done := make(chan error, 1) + go func() { done <- runnable.Start(ctx) }() + cancel() + + Eventually(done, "2s").Should(Receive(BeNil())) + }) +}) diff --git a/internal/cnpgi/instance/manager.go b/internal/cnpgi/instance/manager.go index 8365f24..f3aa985 100644 --- a/internal/cnpgi/instance/manager.go +++ b/internal/cnpgi/instance/manager.go @@ -25,6 +25,7 @@ import ( "github.com/spf13/viper" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientgoscheme "k8s.io/client-go/kubernetes/scheme" ctrl "sigs.k8s.io/controller-runtime" @@ -48,15 +49,20 @@ func Start(ctx context.Context) error { setupLog := log.FromContext(ctx) setupLog.Info("Starting pgbackrest instance plugin") podName := viper.GetString("pod-name") + clusterName := viper.GetString("cluster-name") + namespace := viper.GetString("namespace") controllerOptions := ctrl.Options{ Scheme: scheme, Client: client.Options{ Cache: &client.CacheOptions{ + // Backups are listed namespace-wide by the catalog maintenance; + // an informer would keep every Backup of the namespace in memory. DisableFor: []client.Object{ &corev1.Secret{}, &pgbackrestv1.Archive{}, &cnpgv1.Cluster{}, + &cnpgv1.Backup{}, }, }, }, @@ -68,8 +74,10 @@ func Start(ctx context.Context) error { return err } + customCacheClient := extendedclient.NewExtendedClient(mgr.GetClient()) + if err := mgr.Add(&CNPGI{ - Client: extendedclient.NewExtendedClient(mgr.GetClient()), + Client: customCacheClient, InstanceName: podName, // TODO: improve PGDataPath: viper.GetString("pgdata"), @@ -81,6 +89,18 @@ func Start(ctx context.Context) error { return err } + if err := mgr.Add(&CatalogMaintenanceRunnable{ + Client: customCacheClient, + ClusterKey: types.NamespacedName{ + Namespace: namespace, + Name: clusterName, + }, + CurrentPodName: podName, + }); err != nil { + setupLog.Error(err, "unable to create catalog maintenance runnable") + return err + } + if err := mgr.Start(ctx); err != nil { return err } diff --git a/internal/cnpgi/instance/suite_test.go b/internal/cnpgi/instance/suite_test.go new file mode 100644 index 0000000..d67bb96 --- /dev/null +++ b/internal/cnpgi/instance/suite_test.go @@ -0,0 +1,30 @@ +/* +Copyright The CloudNativePG Contributors +Copyright 2025, Opera Norway AS + +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. +*/ + +package instance + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestInstance(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Instance Suite") +} diff --git a/internal/cnpgi/instance/types.go b/internal/cnpgi/instance/types.go new file mode 100644 index 0000000..77c982e --- /dev/null +++ b/internal/cnpgi/instance/types.go @@ -0,0 +1,107 @@ +/* +Copyright The CloudNativePG Contributors +Copyright 2025, Opera Norway AS + +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. +*/ + +package instance + +import ( + "fmt" + "strings" + + "k8s.io/apimachinery/pkg/types" + + "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/cnpgi/metadata" + pgbackrestApi "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/api" +) + +const ( + backupResultMetadataVersionKey = "version" + backupResultMetadataNameKey = "name" + backupResultMetadataDisplayNameKey = "displayName" + backupResultMetadataClusterUIDKey = "clusterUID" + backupResultMetadataPluginNameKey = "pluginName" + backupResultMetadataStanzaKey = "stanza" + backupResultMetadataRepositoriesKey = "repositories" +) + +// backupResultMetadata is the content of Backup.status.pluginMetadata written +// by this plugin: the identity keys let the catalog maintenance tell apart the +// backups taken by this plugin for this cluster, stanza and repositories from +// anything else. +type backupResultMetadata struct { + version string + name string + displayName string + clusterUID string + pluginName string + stanza string + repositories string +} + +func (b backupResultMetadata) toMap() map[string]string { + return map[string]string{ + backupResultMetadataVersionKey: b.version, + backupResultMetadataNameKey: b.name, + backupResultMetadataDisplayNameKey: b.displayName, + backupResultMetadataClusterUIDKey: b.clusterUID, + backupResultMetadataPluginNameKey: b.pluginName, + backupResultMetadataStanzaKey: b.stanza, + backupResultMetadataRepositoriesKey: b.repositories, + } +} + +func newBackupResultMetadata( + clusterUID types.UID, + stanza string, + configuration *pgbackrestApi.PgbackrestConfiguration, +) backupResultMetadata { + return backupResultMetadata{ + clusterUID: string(clusterUID), + stanza: stanza, + repositories: describeRepositories(configuration), + version: metadata.Data.Version, + name: metadata.Data.Name, + displayName: metadata.Data.DisplayName, + pluginName: metadata.PluginName, + } +} + +// describeRepositories renders where the stanza is written, in pgBackRest's repository +// order, so that an Archive later pointed elsewhere is told apart from this one. +func describeRepositories(configuration *pgbackrestApi.PgbackrestConfiguration) string { + locations := make([]string, len(configuration.Repositories)) + for idx := range configuration.Repositories { + repository := &configuration.Repositories[idx] + locations[idx] = fmt.Sprintf("%s/%s%s", repository.EndpointURL, repository.Bucket, repository.DestinationPath) + } + return strings.Join(locations, ",") +} + +func newBackupResultMetadataFromMap(m map[string]string) backupResultMetadata { + if m == nil { + return backupResultMetadata{} + } + + return backupResultMetadata{ + version: m[backupResultMetadataVersionKey], + name: m[backupResultMetadataNameKey], + displayName: m[backupResultMetadataDisplayNameKey], + clusterUID: m[backupResultMetadataClusterUIDKey], + pluginName: m[backupResultMetadataPluginNameKey], + stanza: m[backupResultMetadataStanzaKey], + repositories: m[backupResultMetadataRepositoriesKey], + } +} diff --git a/internal/cnpgi/instance/types_test.go b/internal/cnpgi/instance/types_test.go new file mode 100644 index 0000000..d4bd881 --- /dev/null +++ b/internal/cnpgi/instance/types_test.go @@ -0,0 +1,83 @@ +/* +Copyright The CloudNativePG Contributors +Copyright 2025, Opera Norway AS + +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. +*/ + +package instance + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "k8s.io/apimachinery/pkg/types" + + "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/cnpgi/metadata" + pgbackrestApi "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/api" +) + +var _ = Describe("backupResultMetadata", func() { + oneRepository := &pgbackrestApi.PgbackrestConfiguration{ + Repositories: []pgbackrestApi.PgbackrestRepository{ + {EndpointURL: "https://s3.example.com", Bucket: "bucket", DestinationPath: "/path"}, + }, + } + + It("serializes the cluster UID, the stanza, the plugin identity, the static plugin data and the repositories", func() { + m := newBackupResultMetadata(types.UID("uid-1234"), "cluster-example", oneRepository).toMap() + + Expect(m).To(Equal(map[string]string{ + "version": metadata.Data.Version, + "name": metadata.Data.Name, + "displayName": metadata.Data.DisplayName, + "clusterUID": "uid-1234", + "pluginName": metadata.PluginName, + "stanza": "cluster-example", + "repositories": "https://s3.example.com/bucket/path", + })) + }) + + It("renders two repositories in configuration order, joined by a comma", func() { + configuration := &pgbackrestApi.PgbackrestConfiguration{ + Repositories: []pgbackrestApi.PgbackrestRepository{ + {EndpointURL: "https://a", Bucket: "b1", DestinationPath: "/p1"}, + {EndpointURL: "https://c", Bucket: "b2", DestinationPath: "/p2"}, + }, + } + + Expect(describeRepositories(configuration)).To(Equal("https://a/b1/p1,https://c/b2/p2")) + }) + + It("round-trips through a map", func() { + original := newBackupResultMetadata(types.UID("uid-1234"), "cluster-example", oneRepository) + + Expect(newBackupResultMetadataFromMap(original.toMap())).To(Equal(original)) + }) + + It("returns empty metadata for a nil map", func() { + Expect(newBackupResultMetadataFromMap(nil)).To(Equal(backupResultMetadata{})) + }) + + It("ignores unknown keys and tolerates missing ones", func() { + m := newBackupResultMetadataFromMap(map[string]string{ + "version": "0.1.0", + "foo": "bar", + }) + + Expect(m.version).To(Equal("0.1.0")) + Expect(m.clusterUID).To(BeEmpty()) + Expect(m.pluginName).To(BeEmpty()) + Expect(m.stanza).To(BeEmpty()) + Expect(m.repositories).To(BeEmpty()) + }) +}) diff --git a/internal/pgbackrest/catalog/catalog.go b/internal/pgbackrest/catalog/catalog.go index 6c8171e..d504a8e 100644 --- a/internal/pgbackrest/catalog/catalog.go +++ b/internal/pgbackrest/catalog/catalog.go @@ -336,13 +336,22 @@ type Catalog struct { Databases []PgbackrestBackupDatabase `json:"db"` Encryption string `json:"cipher"` Status PgbackrestStanzaStatus `json:"status"` + Repos []PgbackrestRepo `json:"repo"` } -// StanzaStatusCodeMissing is the "pgbackrest info" stanza status code returned when -// the stanza has not been created in the repository yet (its archive.info is absent). -// While the stanza is in this state WAL archiving cannot succeed until -// "pgbackrest stanza-create" has been run. -const StanzaStatusCodeMissing = 1 +const ( + // StanzaStatusCodeOk means everything that was asked about could be read and is consistent. + StanzaStatusCodeOk = 0 + // StanzaStatusCodeMissing means the stanza has not been created in the repository yet + // (its archive.info is absent); WAL archiving cannot succeed until stanza-create runs. + StanzaStatusCodeMissing = 1 + // StanzaStatusCodeNoBackup means the repository could be read in full and holds no backup yet. + StanzaStatusCodeNoBackup = 2 + // StanzaStatusCodeMixed means the repositories of the stanza do not all report the same + // status. With one repository holding backups and another still empty this is the + // expected answer, so it says nothing about any repository having failed. + StanzaStatusCodeMixed = 4 +) // PgbackrestStanzaStatus is the "status" object that "pgbackrest info" reports for a // stanza. Only the fields we act on are parsed; the rest of the object is ignored. @@ -357,6 +366,45 @@ func (catalog *Catalog) StanzaMissing() bool { return catalog.Status.Code == StanzaStatusCodeMissing } +// PgbackrestRepo is one of the repositories of a stanza, as "pgbackrest info" reports +// it. Only the fields we act on are parsed. +type PgbackrestRepo struct { + // The one-based index of the repository in the pgbackrest configuration + Key int `json:"key"` + // "pgbackrest info" does not fail when one repository cannot be read: it reports the + // failure here and still returns what the other repositories hold. + Status PgbackrestStanzaStatus `json:"status"` +} + +// DescribesRepositories reports whether the catalog is a complete and consistent +// description of the given number of repositories — what a caller needs before +// concluding that a backup absent from it does not exist anymore. +func (catalog *Catalog) DescribesRepositories(repositories int) bool { + // pgBackRest writes a backup to a single repository, so a stanza spanning several + // reports "mixed" as a matter of course; what the repositories themselves report decides. + if catalog.Status.Code != StanzaStatusCodeMixed && !describesContent(catalog.Status.Code) { + return false + } + + if len(catalog.Repos) != repositories { + return false + } + + for _, repo := range catalog.Repos { + if !describesContent(repo.Status.Code) { + return false + } + } + + return true +} + +// describesContent tells whether a status code belongs to a repository whose content +// could be read in full; an empty repository has been read as completely as a used one. +func describesContent(statusCode int) bool { + return statusCode == StanzaStatusCodeOk || statusCode == StanzaStatusCodeNoBackup +} + // NewSingleBackupCatalogFromPgbackrestInfo parses the output of pgbackrest info // targeting a single backup via "--set". // While structure is the same as for the full backups list there is only a single diff --git a/internal/pgbackrest/catalog/catalog_test.go b/internal/pgbackrest/catalog/catalog_test.go index 0cb8919..e487e93 100644 --- a/internal/pgbackrest/catalog/catalog_test.go +++ b/internal/pgbackrest/catalog/catalog_test.go @@ -167,6 +167,108 @@ var _ = Describe("pgbackrest info parsing", func() { Expect(result.Status.Message).To(Equal("missing stanza path")) }) + It("must parse the repositories of a complete answer", func() { + result, err := NewCatalogFromPgbackrestInfo(pgbackrestInfoOutput) + Expect(err).ToNot(HaveOccurred()) + Expect(result.Repos).To(HaveLen(1)) + Expect(result.Repos[0].Key).To(Equal(1)) + Expect(result.Repos[0].Status.Code).To(Equal(StanzaStatusCodeOk)) + Expect(result.DescribesRepositories(1)).To(BeTrue()) + + // the stanza is configured with a repository this answer says nothing about + Expect(result.DescribesRepositories(2)).To(BeFalse()) + }) + + It("must accept a stanza whose second repository holds no backup yet", func() { + // pgBackRest writes a backup to a single repository, so a stanza spanning two + // of them reports "mixed" as a matter of course + emptySecondRepoOutput := `[ + { + "archive": [], + "backup": [ + { + "annotation": { "cnpg-backup-name": "backup-20250401132030" }, + "archive": { "start": "00000001000000000000000D", "stop": "00000001000000000000000D" }, + "database": { "id": 1, "repo-key": 1 }, + "error": false, + "label": "20250401-132030F", + "lsn": { "start": "0/D000028", "stop": "0/D000158" }, + "prior": null, + "reference": null, + "timestamp": { "start": 1743513630, "stop": 1743513632 }, + "type": "full" + } + ], + "cipher": "none", + "db": [], + "name": "cluster-example-pgbackrest", + "repo": [ + { "cipher": "none", "key": 1, "status": { "code": 0, "message": "ok" } }, + { "cipher": "none", "key": 2, "status": { "code": 2, "message": "no valid backups" } } + ], + "status": { "code": 4, "message": "different across repos" } + } +]` + result, err := NewCatalogFromPgbackrestInfo(emptySecondRepoOutput) + Expect(err).ToNot(HaveOccurred()) + Expect(result.DescribesRepositories(2)).To(BeTrue()) + }) + + It("must refuse a stanza whose second repository could not be read", func() { + partialOutput := `[ + { + "archive": [], + "backup": [ + { + "annotation": { "cnpg-backup-name": "backup-20250401132030" }, + "archive": { "start": "00000001000000000000000D", "stop": "00000001000000000000000D" }, + "database": { "id": 1, "repo-key": 1 }, + "error": false, + "label": "20250401-132030F", + "lsn": { "start": "0/D000028", "stop": "0/D000158" }, + "prior": null, + "reference": null, + "timestamp": { "start": 1743513630, "stop": 1743513632 }, + "type": "full" + } + ], + "cipher": "none", + "db": [], + "name": "cluster-example-pgbackrest", + "repo": [ + { "cipher": "none", "key": 1, "status": { "code": 0, "message": "ok" } }, + { "cipher": "none", "key": 2, "status": { "code": 99, "message": "other" } } + ], + "status": { "code": 4, "message": "different across repos" } + } +]` + result, err := NewCatalogFromPgbackrestInfo(partialOutput) + Expect(err).ToNot(HaveOccurred()) + + // the backups of the repository that could not be read are simply absent + Expect(result.Backups).To(HaveLen(1)) + Expect(result.DescribesRepositories(2)).To(BeFalse()) + }) + + DescribeTable("weighs every criterion of a complete description on its own", + func(stanzaCode int, repoCodes []int, configured int, expected bool) { + c := &Catalog{Status: PgbackrestStanzaStatus{Code: stanzaCode}} + for idx, code := range repoCodes { + c.Repos = append(c.Repos, PgbackrestRepo{Key: idx + 1, Status: PgbackrestStanzaStatus{Code: code}}) + } + Expect(c.DescribesRepositories(configured)).To(Equal(expected)) + }, + Entry("everything ok", 0, []int{0}, 1, true), + Entry("an empty repository alongside a used one", 4, []int{0, 2}, 2, true), + Entry("every repository empty", 2, []int{2, 2}, 2, true), + Entry("one repository unreadable", 0, []int{0, 99}, 2, false), + Entry("one repository missing the stanza", 0, []int{0, 1}, 2, false), + Entry("a database mismatch across repositories", 5, []int{0, 0}, 2, false), + Entry("a stanza missing everywhere", 1, []int{1}, 1, false), + Entry("fewer repositories than configured", 0, []int{0}, 2, false), + Entry("more repositories than configured", 0, []int{0, 0}, 1, false), + ) + // It("can find the closest backup info when there is one", func() { // recoveryTarget := &v1.RecoveryTarget{TargetTime: time.Now().Format("2006-01-02 15:04:04")} // closestBackupInfo, err := catalog.FindBackupInfo(recoveryTarget) diff --git a/manifest.yaml b/manifest.yaml index 7ad05bb..f1c50cd 100644 --- a/manifest.yaml +++ b/manifest.yaml @@ -411,6 +411,19 @@ spec: description: InstanceSidecarConfiguration defines the configuration for the sidecar that runs in the instance pods. properties: + catalogMaintenanceIntervalSeconds: + default: 1800 + description: |- + The catalogMaintenanceIntervalSeconds defines how often the sidecar + running on the primary instance compares the Backup objects of the + cluster with the pgBackRest catalog and deletes the ones that are + not present in the catalog anymore. 0 disables the maintenance. + format: int32 + type: integer + x-kubernetes-validations: + - message: catalogMaintenanceIntervalSeconds must be 0 (disabled) + or at least 60 + rule: self == 0 || self >= 60 env: description: The environment to be explicitly passed to the sidecar items: From a474cdf1232e4b08a39060d768d01d47c53bbb48 Mon Sep 17 00:00:00 2001 From: ermakov-oleg Date: Mon, 21 Sep 2026 18:12:54 +0200 Subject: [PATCH 3/3] fix(e2e): pin CloudNativePG to v1.30.0 The e2e tests installed CloudNativePG from its main branch. Since cloudnative-pg#11319 (2026-08-27) new instances bootstrap through an init container instead of a Job, and the recovery of a cluster needs the restore hooks served from the instance sidecar, which this plugin does not do yet. Every recovery-based test therefore hung on the bootstrap init container. Run the suite against the latest release the plugin supports; the API in go.mod is pinned to the same version. Claude-Session: https://claude.ai/code/session_01NRY9m8yhU4NVCCBHnURaL9 Signed-off-by: ermakov-oleg --- test/e2e/e2e_suite_test.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/test/e2e/e2e_suite_test.go b/test/e2e/e2e_suite_test.go index 5ca11f9..6977669 100644 --- a/test/e2e/e2e_suite_test.go +++ b/test/e2e/e2e_suite_test.go @@ -50,7 +50,14 @@ var _ = SynchronizedBeforeSuite(func(ctx SpecContext) []byte { Fail(fmt.Sprintf("failed to create Kubernetes client: %v", err)) } - if err = e2etestenv.Setup(ctx, cl); err != nil { + // CloudNativePG main moved the bootstrap of new instances from a Job to an init container + // (cloudnative-pg#11319); the plugin serves the restore hooks from the job sidecar only, so the + // e2e tests run against the latest release the plugin supports. + if err = e2etestenv.Setup(ctx, cl, + e2etestenv.WithCNPGKustomizationRef("v1.30.0"), + e2etestenv.WithCNPGImageName("ghcr.io/cloudnative-pg/cloudnative-pg"), + e2etestenv.WithCNPGImageTag("1.30.0"), + ); err != nil { Fail(fmt.Sprintf("failed to setup environment: %v", err)) }