WAL-Archiving + PITR

cmd/monitor/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"os"
 	"time"
 
 	"github.com/fly-apps/postgres-flex/internal/flypg"
@@ -16,6 +17,8 @@ var (
 
 	defaultDeadMemberRemovalThreshold   = time.Hour * 12
 	defaultInactiveSlotRemovalThreshold = time.Hour * 12
+
+	defaultBackupRetentionEvaluationThreshold = time.Hour * 1
 )
 
 func main() {
@@ -36,6 +39,18 @@ func main() {
 		}
 	}()
 
+	if os.Getenv("CLOUD_ARCHIVING_ENABLED") == "true" {
+		barman, err := flypg.NewBarman()
+		if err != nil {
+			panic(err)
+		}
+
+		log.Println("Monitoring backup retention")
+		barman.PrintRetentionPolicy()
+
+		go monitorBackupRetention(ctx, barman)
+	}
+
 	// Readonly monitor
 	log.Println("Monitoring cluster state")
 	go monitorClusterState(ctx, node)

cmd/monitor/monitor_backup_retention.go

@@ -0,0 +1,24 @@
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"github.com/fly-apps/postgres-flex/internal/flypg"
+)
+
+func monitorBackupRetention(ctx context.Context, barman *flypg.Barman) {
+	ticker := time.NewTicker(defaultBackupRetentionEvaluationThreshold)
+	defer ticker.Stop()
+	for range ticker.C {
+		result, err := barman.WALArchiveDelete(ctx)
+		if err != nil {
+			log.Printf("Backup retention failed with: %s", err)
+		}
+
+		if len(result) > 0 {
+			log.Printf("Backup retention response: %s", result)
+		}
+	}
+}

cmd/monitor/monitor_dead_members.go

@@ -37,8 +37,6 @@ func monitorDeadMembers(ctx context.Context, node *flypg.Node) error {
 	ticker := time.NewTicker(deadMemberMonitorFrequency)
 	defer ticker.Stop()
 
-	log.Printf("Pruning every %s...\n", removalThreshold)
-
 	for range ticker.C {
 		err := deadMemberMonitorTick(ctx, node, seenAt, removalThreshold)
 		if err != nil {

cmd/start/main.go

@@ -38,7 +38,6 @@ func main() {
 		}
 
 		svisor := supervisor.New("flybarman", 1*time.Minute)
-		svisor.AddProcess("cron", "/usr/sbin/cron -f", supervisor.WithRestart(0, 5*time.Second))
 		svisor.AddProcess("barman", fmt.Sprintf("tail -f %s", node.LogFile))
 		svisor.AddProcess("admin", "/usr/local/bin/start_admin_server",
 			supervisor.WithRestart(0, 5*time.Second),

internal/flybarman/node.go

@@ -139,8 +139,8 @@ wal_retention_policy = main
 		return fmt.Errorf("failed to write file %s: %s", n.RootPasswordConfigPath, err)
 	}
 
-	barmanCronFileContent := `* * * * * /usr/local/bin/barman_cron
-`
+	barmanCronFileContent := `* * * * * /usr/local/bin/barman_cron`
+
 	if err := os.WriteFile(n.BarmanCronFile, []byte(barmanCronFileContent), 0644); err != nil {
 		return fmt.Errorf("failed write %s: %s", n.BarmanCronFile, err)
 	}

internal/flypg/barman.go

@@ -0,0 +1,116 @@
+package flypg
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"strings"
+
+	"github.com/fly-apps/postgres-flex/internal/utils"
+)
+
+const (
+	barmanRecoveryDirectory = "/data/postgresql"
+)
+
+type Barman struct {
+	appName  string
+	provider string
+	endpoint string
+	bucket   string
+
+	// fullBackupFrequency string
+	minimumRedundancy string
+	retentionDays     string
+}
+
+func NewBarman() (*Barman, error) {
+	if err := validateBarman(); err != nil {
+		return nil, err
+	}
+
+	// TODO - Validate minimum and retention day values
+	minRedundancy := getenv("CLOUD_ARCHIVING_MINIMUM_REDUNDANCY", "3")
+	retentionDays := getenv("CLOUD_ARCHIVING_RETENTION_DAYS", "7")
+
+	return &Barman{
+		appName:  os.Getenv("FLY_APP_NAME"),
+		provider: "aws-s3",
+		endpoint: strings.TrimSpace(os.Getenv("AWS_ENDPOINT_URL")),
+		bucket:   strings.TrimSpace(os.Getenv("AWS_BUCKET_NAME")),
+
+		// fullBackupFrequency: getenv("CLOUD_ARCHIVING_FULL_BACKUP_FREQUENCY", "1"),
+		minimumRedundancy: minRedundancy,
+		retentionDays:     retentionDays,
+	}, nil
+}
+
+func (b *Barman) RetentionPolicy() string {
+	return fmt.Sprintf("'RECOVERY WINDOW OF %s days'", b.retentionDays)
+}
+
+// WALArchiveDelete deletes backups/WAL based on the specified retention policy.
+func (b *Barman) WALArchiveDelete(ctx context.Context) ([]byte, error) {
+	return utils.RunCommand(b.walArchiveDeleteCommand(), "postgres")
+}
+
+func (b *Barman) PrintRetentionPolicy() {
+	str := `
+	Retention Policy
+	-----------------
+	RECOVERY WINDOW OF %s days
+	MINIMUM BACKUP REDUNDANCY: %s
+	FULL BACKUP FREQUENCY: %s day(s)
+`
+	log.Printf(str, b.retentionDays, b.minimumRedundancy)
+}
+
+func (b *Barman) walArchiveDeleteCommand() string {
+	return fmt.Sprintf("barman-cloud-backup-delete --cloud-provider %s --endpoint-url %s --retention %s --minimum-redundancy %s s3://%s %s",
+		b.provider,
+		b.endpoint,
+		b.RetentionPolicy(),
+		b.minimumRedundancy,
+		b.bucket,
+		b.appName,
+	)
+}
+
+func (b *Barman) walArchiveCommand() string {
+	// TODO - Make compression configurable
+	return fmt.Sprintf("barman-cloud-wal-archive --cloud-provider %s --gzip --endpoint-url %s s3://%s %s %%p",
+		b.provider,
+		b.endpoint,
+		b.bucket,
+		b.appName,
+	)
+}
+
+func validateBarman() error {
+	if os.Getenv("AWS_ACCESS_KEY_ID") == "" {
+		return fmt.Errorf("AWS_ACCESS_KEY_ID secret must be set")
+	}
+
+	if os.Getenv("AWS_SECRET_ACCESS_KEY") == "" {
+		return fmt.Errorf("AWS_SECRET_ACCESS_KEY secret must be set")
+	}
+
+	if os.Getenv("AWS_BUCKET_NAME") == "" {
+		return fmt.Errorf("AWS_BUCKET_NAME envvar must be set")
+	}
+
+	if os.Getenv("AWS_ENDPOINT_URL") == "" {
+		return fmt.Errorf("AWS_ENDPOINT_URL envvar must be set")
+	}
+
+	return nil
+}
+
+func getenv(key, fallback string) string {
+	value := os.Getenv(key)
+	if len(value) == 0 {
+		return fallback
+	}
+	return value
+}