feat: separate recovery and cluster object store

Signed-off-by: Leonardo Cecchi <[email protected]>

Author: leonardoce

docs/examples/cluster-restore-archive.yaml

@@ -0,0 +1,27 @@
+apiVersion: postgresql.cnpg.io/v1
+kind: Cluster
+metadata:
+  name: cluster-restore
+spec:
+  instances: 3
+  imagePullPolicy: IfNotPresent
+
+  bootstrap:
+    recovery:
+      source: source
+
+  plugins:
+  - name: barman-cloud.cloudnative-pg.io
+    parameters:
+      barmanObjectName: minio-store-bis
+
+  externalClusters:
+  - name: source
+    plugin:
+      name: barman-cloud.cloudnative-pg.io
+      parameters:
+        barmanObjectName: minio-store
+        serverName: cluster-example
+
+  storage:
+    size: 1Gi

docs/examples/minio-store-bis.yaml

@@ -0,0 +1,23 @@
+apiVersion: barmancloud.cnpg.io/v1
+kind: ObjectStore
+metadata:
+  name: minio-store-bis
+spec:
+  configuration:
+    destinationPath: s3://backups/
+    endpointURL: http://minio-bis:9000
+    s3Credentials:
+      accessKeyId:
+        name: minio-bis
+        key: ACCESS_KEY_ID
+      secretAccessKey:
+        name: minio-bis
+        key: ACCESS_SECRET_KEY
+    wal:
+      compression: gzip
+    data:
+      additionalCommandArgs:
+        - "--min-chunk-size=5MB"
+        - "--read-timeout=60"
+        - "-vv"
+    

docs/minio-bis/minio-deployment.yaml

@@ -0,0 +1,42 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: minio-bis
+  labels:
+    app: minio-bis
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: minio-bis
+  template:
+    metadata:
+      labels:
+        app: minio-bis
+    spec:
+      containers:
+      - name: minio
+        image: minio/minio
+        ports:
+        - containerPort: 9000
+        volumeMounts:
+        - mountPath: /data
+          name: data
+        args:
+        - server
+        - /data
+        env:
+        - name: MINIO_ACCESS_KEY
+          valueFrom:
+            secretKeyRef: 
+              name: minio-bis
+              key: ACCESS_KEY_ID
+        - name: MINIO_SECRET_KEY
+          valueFrom:
+            secretKeyRef: 
+              name: minio-bis
+              key: ACCESS_SECRET_KEY
+      volumes:
+      - name: data
+        persistentVolumeClaim:
+          claimName: minio-bis

docs/minio-bis/minio-pvc.yaml

@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: minio-bis
+spec:
+  accessModes:
+    - ReadWriteOnce
+  volumeMode: Filesystem
+  resources:
+    requests:
+      storage: 1Gi

docs/minio-bis/minio-secret.yaml

@@ -0,0 +1,7 @@
+apiVersion: v1
+data:
+  ACCESS_KEY_ID: dVo5YWVndW9OZ29vY2g4bG9odHU4aXRlaTJhaHY0ZGE=
+  ACCESS_SECRET_KEY: ZWV6b2hkOFNpbm9oeG9od2VpbmdvbjhhaXI1T2h5b2g=
+kind: Secret
+metadata:
+  name: minio-bis

docs/minio-bis/minio-service.yaml

@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: minio-bis
+spec:
+  selector:
+    app: minio-bis
+  ports:
+    - protocol: TCP
+      port: 9000
+      targetPort: 9000

internal/cmd/instance/main.go

@@ -36,12 +36,16 @@ func NewCmd() *cobra.Command {
 	}
 
 	_ = viper.BindEnv("namespace", "NAMESPACE")
-	_ = viper.BindEnv("barman-object-name", "BARMAN_OBJECT_NAME")
 	_ = viper.BindEnv("cluster-name", "CLUSTER_NAME")
 	_ = viper.BindEnv("pod-name", "POD_NAME")
 	_ = viper.BindEnv("pgdata", "PGDATA")
 	_ = viper.BindEnv("spool-directory", "SPOOL_DIRECTORY")
+
+	_ = viper.BindEnv("barman-object-name", "BARMAN_OBJECT_NAME")
 	_ = viper.BindEnv("server-name", "SERVER_NAME")
 
+	_ = viper.BindEnv("recovery-barman-object-name", "RECOVERY_BARMAN_OBJECT_NAME")
+	_ = viper.BindEnv("recovery-server-name", "RECOVERY_SERVER_NAME")
+
 	return cmd
 }

internal/cmd/restore/main.go

@@ -23,8 +23,11 @@ func NewCmd() *cobra.Command {
 				"cluster-name",
 				"pod-name",
 				"spool-directory",
-				"barman-object-name",
-				"server-name",
+
+				// IMPORTANT: barman-object-name and server-name is not required here
+				// as we are restoring an existing cluster.
+				"recovery-barman-object-name",
+				"recovery-server-name",
 			}
 
 			for _, k := range requiredSettings {
@@ -42,8 +45,12 @@ func NewCmd() *cobra.Command {
 	_ = viper.BindEnv("pod-name", "POD_NAME")
 	_ = viper.BindEnv("pgdata", "PGDATA")
 	_ = viper.BindEnv("spool-directory", "SPOOL_DIRECTORY")
+
 	_ = viper.BindEnv("barman-object-name", "BARMAN_OBJECT_NAME")
 	_ = viper.BindEnv("server-name", "SERVER_NAME")
 
+	_ = viper.BindEnv("recovery-barman-object-name", "RECOVERY_BARMAN_OBJECT_NAME")
+	_ = viper.BindEnv("recovery-server-name", "RECOVERY_SERVER_NAME")
+
 	return cmd
 }

internal/cnpgi/common/wal.go

@@ -24,15 +24,19 @@ import (
 
 // WALServiceImplementation is the implementation of the WAL Service
 type WALServiceImplementation struct {
-	ServerName       string
-	BarmanObjectKey  client.ObjectKey
+	wal.UnimplementedWALServer
 	ClusterObjectKey client.ObjectKey
 	Client           client.Client
 	InstanceName     string
 	SpoolDirectory   string
 	PGDataPath       string
 	PGWALPath        string
-	wal.UnimplementedWALServer
+
+	BarmanObjectKey client.ObjectKey
+	ServerName      string
+
+	RecoveryBarmanObjectKey client.ObjectKey
+	RecoveryServerName      string
 }
 
 // GetCapabilities implements the WALService interface
@@ -123,22 +127,54 @@ func (w WALServiceImplementation) Restore(
 	ctx context.Context,
 	request *wal.WALRestoreRequest,
 ) (*wal.WALRestoreResult, error) {
-	contextLogger := log.FromContext(ctx)
-	startTime := time.Now()
+	// TODO: build full paths
+	walName := request.GetSourceWalName()
+	destinationPath := request.GetDestinationFileName()
 
 	var cluster cnpgv1.Cluster
 	if err := w.Client.Get(ctx, w.ClusterObjectKey, &cluster); err != nil {
 		return nil, err
 	}
 
 	var objectStore barmancloudv1.ObjectStore
-	if err := w.Client.Get(ctx, w.BarmanObjectKey, &objectStore); err != nil {
-		return nil, err
+	var serverName string
+
+	switch {
+	case cluster.IsReplica() && cluster.Status.CurrentPrimary == w.InstanceName:
+		// Designated primary on replica cluster, using recovery object store
+		serverName = w.RecoveryServerName
+		if err := w.Client.Get(ctx, w.RecoveryBarmanObjectKey, &objectStore); err != nil {
+			return nil, err
+		}
+
+	case cluster.Status.CurrentPrimary == "":
+		// Recovery from object store, using recovery object store
+		serverName = w.RecoveryServerName
+		if err := w.Client.Get(ctx, w.RecoveryBarmanObjectKey, &objectStore); err != nil {
+			return nil, err
+		}
+
+	default:
+		// Using cluster object store
+		serverName = w.ServerName
+		if err := w.Client.Get(ctx, w.BarmanObjectKey, &objectStore); err != nil {
+			return nil, err
+		}
 	}
 
-	// TODO: build full paths
-	walName := request.GetSourceWalName()
-	destinationPath := request.GetDestinationFileName()
+	return &wal.WALRestoreResult{}, w.RestoreEx(ctx, &cluster, &objectStore, serverName, walName, destinationPath)
+}
+
+func (w WALServiceImplementation) RestoreEx(
+	ctx context.Context,
+	cluster *cnpgv1.Cluster,
+	objectStore *barmancloudv1.ObjectStore,
+	serverName string,
+	walName string,
+	destinationPath string,
+) error {
+	contextLogger := log.FromContext(ctx)
+	startTime := time.Now()
 
 	barmanConfiguration := &objectStore.Spec.Configuration
 
@@ -151,37 +187,37 @@ func (w WALServiceImplementation) Restore(
 		os.Environ(),
 	)
 	if err != nil {
-		return nil, fmt.Errorf("while getting recover credentials: %w", err)
+		return fmt.Errorf("while getting recover credentials: %w", err)
 	}
 	env = MergeEnv(env, credentialsEnv)
 
-	options, err := barmanCommand.CloudWalRestoreOptions(ctx, barmanConfiguration, w.ServerName)
+	options, err := barmanCommand.CloudWalRestoreOptions(ctx, barmanConfiguration, serverName)
 	if err != nil {
-		return nil, fmt.Errorf("while getting barman-cloud-wal-restore options: %w", err)
+		return fmt.Errorf("while getting barman-cloud-wal-restore options: %w", err)
 	}
 
 	// Create the restorer
 	var walRestorer *barmanRestorer.WALRestorer
 	if walRestorer, err = barmanRestorer.New(ctx, env, w.SpoolDirectory); err != nil {
-		return nil, fmt.Errorf("while creating the restorer: %w", err)
+		return fmt.Errorf("while creating the restorer: %w", err)
 	}
 
 	// Step 1: check if this WAL file is not already in the spool
 	var wasInSpool bool
 	if wasInSpool, err = walRestorer.RestoreFromSpool(walName, destinationPath); err != nil {
-		return nil, fmt.Errorf("while restoring a file from the spool directory: %w", err)
+		return fmt.Errorf("while restoring a file from the spool directory: %w", err)
 	}
 	if wasInSpool {
 		contextLogger.Info("Restored WAL file from spool (parallel)",
 			"walName", walName,
 		)
-		return nil, nil
+		return nil
 	}
 
 	// We skip this step if streaming connection is not available
-	if isStreamingAvailable(&cluster, w.InstanceName) {
+	if isStreamingAvailable(cluster, w.InstanceName) {
 		if err := checkEndOfWALStreamFlag(walRestorer); err != nil {
-			return nil, err
+			return err
 		}
 	}
 
@@ -194,7 +230,7 @@ func (w WALServiceImplementation) Restore(
 	if IsWALFile(walName) {
 		// If this is a regular WAL file, we try to prefetch
 		if walFilesList, err = gatherWALFilesToRestore(walName, maxParallel); err != nil {
-			return nil, fmt.Errorf("while generating the list of WAL files to restore: %w", err)
+			return fmt.Errorf("while generating the list of WAL files to restore: %w", err)
 		}
 	} else {
 		// This is not a regular WAL file, we fetch it directly
@@ -209,18 +245,18 @@ func (w WALServiceImplementation) Restore(
 	// is the one that PostgreSQL has requested to restore.
 	// The failure has already been logged in walRestorer.RestoreList method
 	if walStatus[0].Err != nil {
-		return nil, walStatus[0].Err
+		return walStatus[0].Err
 	}
 
 	// We skip this step if streaming connection is not available
 	endOfWALStream := isEndOfWALStream(walStatus)
-	if isStreamingAvailable(&cluster, w.InstanceName) && endOfWALStream {
+	if isStreamingAvailable(cluster, w.InstanceName) && endOfWALStream {
 		contextLogger.Info(
 			"Set end-of-wal-stream flag as one of the WAL files to be prefetched was not found")
 
 		err = walRestorer.SetEndOfWALStream()
 		if err != nil {
-			return nil, err
+			return err
 		}
 	}
 
@@ -241,7 +277,7 @@ func (w WALServiceImplementation) Restore(
 		"downloadTotalTime", time.Since(downloadStartTime),
 		"totalTime", time.Since(startTime))
 
-	return &wal.WALRestoreResult{}, nil
+	return nil
 }
 
 // Status implements the WALService interface