Postgres Restore (#306)

api/v1/postgres_types.go

@@ -11,6 +11,7 @@ import (
 	"reflect"
 	"strconv"
 	"strings"
+	"time"
 
 	"regexp"
 
@@ -155,6 +156,9 @@ type PostgresSpec struct {
 	// BackupSecretRef reference to the secret where the backup credentials are stored
 	BackupSecretRef string `json:"backupSecretRef,omitempty"`
 
+	// PostgresRestore
+	PostgresRestore *PostgresRestore `json:"restore,omitempty"`
+
 	// PostgresConnection Connection info of a streaming host, independant of the current role (leader or standby)
 	PostgresConnection *PostgresConnection `json:"connection,omitempty"`
 
@@ -187,6 +191,14 @@ type Size struct {
 	StorageSize string `json:"storageSize,omitempty"`
 }
 
+// Restore defines what to restore from where
+type PostgresRestore struct {
+	// SourcePostgresID internal ID of the Postgres instance to whose backup to restore
+	SourcePostgresID string `json:"postgresID,omitempty"`
+	// Timestamp The point in time to recover. Must be set, or the clone with switch from WALs from the S3 to a basebackup via direct sql connection (which won't work when the source db is managed by another posgres-operator)
+	Timestamp string `json:"timestamp,omitempty"`
+}
+
 // PostgresStatus defines the observed state of Postgres
 type PostgresStatus struct {
 	// INSERT ADDITIONAL STATUS FIELD - define observed state of cluster
@@ -463,7 +475,7 @@ func (p *Postgres) ToPeripheralResourceLookupKey() types.NamespacedName {
 	}
 }
 
-func (p *Postgres) ToUnstructuredZalandoPostgresql(z *zalando.Postgresql, c *corev1.ConfigMap, sc string, pgParamBlockList map[string]bool) (*unstructured.Unstructured, error) {
+func (p *Postgres) ToUnstructuredZalandoPostgresql(z *zalando.Postgresql, c *corev1.ConfigMap, sc string, pgParamBlockList map[string]bool, rbs *BackupConfig, srcDB *Postgres) (*unstructured.Unstructured, error) {
 	if z == nil {
 		z = &zalando.Postgresql{}
 	}
@@ -526,8 +538,8 @@ func (p *Postgres) ToUnstructuredZalandoPostgresql(z *zalando.Postgresql, c *cor
 			"pgcrypto":   "public",
 		},
 		PreparedSchemas: map[string]zalando.PreparedSchema{
-			"data":    zalando.PreparedSchema{},
-			"history": zalando.PreparedSchema{},
+			"data":    {},
+			"history": {},
 		},
 	}
 
@@ -541,6 +553,23 @@ func (p *Postgres) ToUnstructuredZalandoPostgresql(z *zalando.Postgresql, c *cor
 		z.Spec.AllowedSourceRanges = p.Spec.AccessList.SourceRanges
 	}
 
+	if p.Spec.PostgresRestore != nil && rbs != nil && srcDB != nil {
+		// make sure there is always a value set. The operator will fall back to CLONE_WITH_BASEBACKUP, which assumes the source db's credentials are existing within the same namespace, which is not the case with the postgreslet.
+		if p.Spec.PostgresRestore.Timestamp == "" {
+			// e.g. 2021-12-07T15:28:00+01:00
+			p.Spec.PostgresRestore.Timestamp = time.Now().Format(time.RFC3339)
+		}
+
+		z.Spec.Clone = &zalando.CloneDescription{
+			ClusterName:       srcDB.ToPeripheralResourceName(),
+			EndTimestamp:      p.Spec.PostgresRestore.Timestamp,
+			S3Endpoint:        rbs.S3Endpoint,
+			S3AccessKeyId:     rbs.S3AccessKey,
+			S3SecretAccessKey: rbs.S3SecretKey,
+			S3ForcePathStyle:  pointer.Bool(true),
+		}
+	}
+
 	// Enable replication (using unstructured json)
 	if p.IsReplicationPrimary() {
 		// delete field

api/v1/postgres_types_test.go

@@ -9,9 +9,12 @@ package v1
 import (
 	"regexp"
 	"testing"
+	"time"
 
 	"github.com/google/uuid"
+	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 )
 
@@ -216,3 +219,148 @@ func TestPostgres_ToPeripheralResourceName(t *testing.T) {
 		})
 	}
 }
+
+func TestPostgresRestoreTimestamp_ToUnstructuredZalandoPostgresql(t *testing.T) {
+	tests := []struct {
+		name             string
+		spec             PostgresSpec
+		c                *corev1.ConfigMap
+		sc               string
+		pgParamBlockList map[string]bool
+		rbs              *BackupConfig
+		srcDB            *Postgres
+		want             string
+		wantErr          bool
+	}{
+		{
+			name: "empty timestamp initialized with current time",
+			spec: PostgresSpec{
+				Size: &Size{
+					CPU:          "1",
+					Memory:       "4Gi",
+					SharedBuffer: "64Mi",
+				},
+				PostgresRestore: &PostgresRestore{
+					Timestamp: "",
+				},
+			},
+			c:                nil,
+			sc:               "fake-storage-class",
+			pgParamBlockList: map[string]bool{},
+			rbs:              &BackupConfig{},
+			srcDB: &Postgres{
+				ObjectMeta: v1.ObjectMeta{
+					Name: uuid.NewString(),
+				},
+				Spec: PostgresSpec{
+					Tenant:      "tenant",
+					Description: "description",
+				},
+			},
+			want:    time.Now().Format(time.RFC3339), // I know this is not perfect, let's just hope we always finish within the same second...
+			wantErr: false,
+		},
+		{
+			name: "undefined timestamp initialized with current time",
+			spec: PostgresSpec{
+				Size: &Size{
+					CPU:          "1",
+					Memory:       "4Gi",
+					SharedBuffer: "64Mi",
+				},
+				PostgresRestore: &PostgresRestore{},
+			},
+			c:                nil,
+			sc:               "fake-storage-class",
+			pgParamBlockList: map[string]bool{},
+			rbs:              &BackupConfig{},
+			srcDB: &Postgres{
+				ObjectMeta: v1.ObjectMeta{
+					Name: uuid.NewString(),
+				},
+				Spec: PostgresSpec{
+					Tenant:      "tenant",
+					Description: "description",
+				},
+			},
+			want:    time.Now().Format(time.RFC3339), // I know this is not perfect, let's just hope we always finish within the same second...
+			wantErr: false,
+		},
+		{
+			name: "given timestamp is passed along",
+			spec: PostgresSpec{
+				Size: &Size{
+					CPU:          "1",
+					Memory:       "4Gi",
+					SharedBuffer: "64Mi",
+				},
+				PostgresRestore: &PostgresRestore{
+					Timestamp: "invalid but whatever",
+				},
+			},
+			c:                nil,
+			sc:               "fake-storage-class",
+			pgParamBlockList: map[string]bool{},
+			rbs:              &BackupConfig{},
+			srcDB: &Postgres{
+				ObjectMeta: v1.ObjectMeta{
+					Name: uuid.NewString(),
+				},
+				Spec: PostgresSpec{
+					Tenant:      "tenant",
+					Description: "description",
+				},
+			},
+			want:    "invalid but whatever",
+			wantErr: false,
+		},
+		{
+			name: "fail on purpose",
+			spec: PostgresSpec{
+				Size: &Size{
+					CPU:          "1",
+					Memory:       "4Gi",
+					SharedBuffer: "64Mi",
+				},
+				PostgresRestore: &PostgresRestore{
+					Timestamp: "apples",
+				},
+			},
+			c:                nil,
+			sc:               "fake-storage-class",
+			pgParamBlockList: map[string]bool{},
+			rbs:              &BackupConfig{},
+			srcDB: &Postgres{
+				ObjectMeta: v1.ObjectMeta{
+					Name: uuid.NewString(),
+				},
+				Spec: PostgresSpec{
+					Tenant:      "tenant",
+					Description: "description",
+				},
+			},
+			want:    "oranges",
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt // pin!
+		t.Run(tt.name, func(t *testing.T) {
+			p := &Postgres{
+				Spec: tt.spec,
+			}
+			got, _ := p.ToUnstructuredZalandoPostgresql(nil, tt.c, tt.sc, tt.pgParamBlockList, tt.rbs, tt.srcDB)
+
+			jsonZ, err := runtime.DefaultUnstructuredConverter.ToUnstructured(got)
+			if err != nil {
+				t.Errorf("failed to convert to unstructured zalando postgresql: %v", err)
+			}
+			jsonSpec, _ := jsonZ["spec"].(map[string]interface{})
+			jsonClone, _ := jsonSpec["clone"].(map[string]interface{})
+
+			if !tt.wantErr && tt.want != jsonClone["timestamp"] {
+				t.Errorf("Spec.Clone.Timestamp was %v, but expected %v", jsonClone["timestamp"], tt.want)
+			}
+		})
+	}
+}

config/crd/bases/database.fits.cloud_postgres.yaml

@@ -126,6 +126,20 @@ spec:
               projectID:
                 description: ProjectID metal project ID
                 type: string
+              restore:
+                description: PostgresRestore
+                properties:
+                  postgresID:
+                    description: SourcePostgresID internal ID of the Postgres instance
+                      to whose backup to restore
+                    type: string
+                  timestamp:
+                    description: Timestamp The point in time to recover. Must be set,
+                      or the clone with switch from WALs from the S3 to a basebackup
+                      via direct sql connection (which won't work when the source
+                      db is managed by another posgres-operator)
+                    type: string
+                type: object
               size:
                 description: Size of the database
                 properties:

controllers/postgres_controller.go

@@ -239,6 +239,32 @@ func (r *PostgresReconciler) createOrUpdateZalandoPostgresql(ctx context.Context
 		c = nil
 	}
 
+	var restoreBackupConfig *pg.BackupConfig
+	var restoreSouceInstance *pg.Postgres
+	if instance.Spec.PostgresRestore != nil {
+		if instance.Spec.PostgresRestore.SourcePostgresID == "" {
+			return fmt.Errorf("restore requested, but no source configured")
+		}
+		srcNs := types.NamespacedName{
+			Namespace: instance.Namespace,
+			Name:      instance.Spec.PostgresRestore.SourcePostgresID,
+		}
+		src := &pg.Postgres{}
+		if err := r.CtrlClient.Get(ctx, srcNs, src); err != nil {
+			r.recorder.Eventf(instance, "Warning", "Error", "failed to get resource: %v", err)
+			return err
+		}
+		log.Info("source for restore fetched", "postgres", instance)
+
+		bc, err := r.getBackupConfig(ctx, instance.Namespace, src.Spec.BackupSecretRef)
+		if err != nil {
+			return err
+		}
+
+		restoreBackupConfig = bc
+		restoreSouceInstance = src
+	}
+
 	// Get zalando postgresql and create one if none.
 	rawZ, err := r.getZalandoPostgresql(ctx, instance)
 	if err != nil {
@@ -247,7 +273,7 @@ func (r *PostgresReconciler) createOrUpdateZalandoPostgresql(ctx context.Context
 			return fmt.Errorf("failed to fetch zalando postgresql: %w", err)
 		}
 
-		u, err := instance.ToUnstructuredZalandoPostgresql(nil, c, r.StorageClass, r.PgParamBlockList)
+		u, err := instance.ToUnstructuredZalandoPostgresql(nil, c, r.StorageClass, r.PgParamBlockList, restoreBackupConfig, restoreSouceInstance)
 		if err != nil {
 			return fmt.Errorf("failed to convert to unstructured zalando postgresql: %w", err)
 		}
@@ -263,7 +289,7 @@ func (r *PostgresReconciler) createOrUpdateZalandoPostgresql(ctx context.Context
 	// Update zalando postgresql
 	mergeFrom := client.MergeFrom(rawZ.DeepCopy())
 
-	u, err := instance.ToUnstructuredZalandoPostgresql(rawZ, c, r.StorageClass, r.PgParamBlockList)
+	u, err := instance.ToUnstructuredZalandoPostgresql(rawZ, c, r.StorageClass, r.PgParamBlockList, restoreBackupConfig, restoreSouceInstance)
 	if err != nil {
 		return fmt.Errorf("failed to convert to unstructured zalando postgresql: %w", err)
 	}
@@ -316,24 +342,9 @@ func (r *PostgresReconciler) updatePodEnvironmentConfigMap(ctx context.Context,
 		return nil
 	}
 
-	// fetch secret
-	backupSecret := &corev1.Secret{}
-	backupNamespace := types.NamespacedName{
-		Name:      p.Spec.BackupSecretRef,
-		Namespace: p.Namespace,
-	}
-	if err := r.CtrlClient.Get(ctx, backupNamespace, backupSecret); err != nil {
-		return fmt.Errorf("error while getting the backup secret from control plane cluster: %w", err)
-	}
-
-	backupConfigJSON, ok := backupSecret.Data[pg.BackupConfigKey]
-	if !ok {
-		return fmt.Errorf("no backupConfig stored in the secret")
-	}
-	var backupConfig pg.BackupConfig
-	err := json.Unmarshal(backupConfigJSON, &backupConfig)
+	backupConfig, err := r.getBackupConfig(ctx, p.Namespace, p.Spec.BackupSecretRef)
 	if err != nil {
-		return fmt.Errorf("unable to unmarshal backupconfig:%w", err)
+		return err
 	}
 
 	s3url, err := url.Parse(backupConfig.S3Endpoint)
@@ -714,3 +725,26 @@ func (r *PostgresReconciler) updatePatroniConfig(ctx context.Context, instance *
 
 	return nil
 }
+
+func (r *PostgresReconciler) getBackupConfig(ctx context.Context, ns, name string) (*pg.BackupConfig, error) {
+	// fetch secret
+	backupSecret := &corev1.Secret{}
+	backupNamespace := types.NamespacedName{
+		Name:      name,
+		Namespace: ns,
+	}
+	if err := r.CtrlClient.Get(ctx, backupNamespace, backupSecret); err != nil {
+		return nil, fmt.Errorf("error while getting the backup secret from control plane cluster: %w", err)
+	}
+
+	backupConfigJSON, ok := backupSecret.Data[pg.BackupConfigKey]
+	if !ok {
+		return nil, fmt.Errorf("no backupConfig stored in the secret")
+	}
+	var backupConfig pg.BackupConfig
+	err := json.Unmarshal(backupConfigJSON, &backupConfig)
+	if err != nil {
+		return nil, fmt.Errorf("unable to unmarshal backupconfig:%w", err)
+	}
+	return &backupConfig, nil
+}