feat: parallel WAL upload (#45)

While parallel upload was documented as supported, it actually wasn't.

Added some e2e tests for it although this kind of concurrency is tricky to validate.

Signed-off-by: Szymon Soloch <[email protected]>

Author: Agalin

internal/cnpgi/common/wal.go

@@ -136,14 +136,41 @@ func (w WALServiceImplementation) Archive(
 	if err != nil {
 		return nil, err
 	}
-	walList := arch.GatherWALFilesToArchive(ctx, request.GetSourceFileName(), 1)
+
+	maxParallel := 1
+	if archive.Spec.Configuration.Wal != nil && archive.Spec.Configuration.Wal.MaxParallel > 1 {
+		maxParallel = archive.Spec.Configuration.Wal.MaxParallel
+	}
+
+	walList := arch.GatherWALFilesToArchive(ctx, request.GetSourceFileName(), maxParallel)
+
+	// Log the batch of WAL files prepared for archiving
+	contextLogger.Info("WAL archive batch prepared",
+		"requestedWalFile", request.GetSourceFileName(),
+		"maxParallel", maxParallel,
+		"walFiles", walList)
+
 	result := arch.ArchiveList(ctx, walList, options)
+	successfulArchives := 0
+	var lastErr error
 	for _, archiverResult := range result {
-		if archiverResult.Err != nil {
-			return nil, archiverResult.Err
+		if archiverResult.Err == nil {
+			successfulArchives++
+		} else {
+			lastErr = archiverResult.Err
 		}
 	}
 
+	contextLogger.Info("WAL archive batch completed",
+		"requestedWalFile", request.GetSourceFileName(),
+		"maxParallel", maxParallel,
+		"successfulArchives", successfulArchives,
+		"failedArchives", len(walList)-successfulArchives)
+
+	if lastErr != nil {
+		return nil, lastErr
+	}
+
 	return &wal.WALArchiveResult{}, nil
 }
 

internal/pgbackrest/archiver/command_test.go

@@ -19,6 +19,7 @@ package archiver
 
 import (
 	"os"
+	"path/filepath"
 	"strings"
 
 	pgbackrestApi "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/api"
@@ -75,3 +76,161 @@ var _ = Describe("pgbackrestWalArchiveOptions", func() {
 				))
 	})
 })
+
+var _ = Describe("GatherWALFilesToArchive", func() {
+	var tempPgData string
+	var archiveStatusDir string
+	var archiver *WALArchiver
+
+	BeforeEach(func(ctx SpecContext) {
+		var err error
+
+		tempPgData, err = os.MkdirTemp("", "pgdata-test-*")
+		Expect(err).ToNot(HaveOccurred())
+
+		// Archiver uses the env variable to determine directory root.
+		os.Setenv("PGDATA", tempPgData)
+
+		archiveStatusDir = filepath.Join(tempPgData, "pg_wal", "archive_status")
+		err = os.MkdirAll(archiveStatusDir, 0755)
+		Expect(err).ToNot(HaveOccurred())
+
+		tempEmptyWalArchivePath := filepath.Join(tempPgData, "empty-wal-archive")
+		_, err = os.Create(tempEmptyWalArchivePath)
+		Expect(err).ToNot(HaveOccurred())
+
+		archiver, err = New(ctx, nil, filepath.Join(tempPgData, "spool"), tempPgData, tempEmptyWalArchivePath)
+		Expect(err).ToNot(HaveOccurred())
+	})
+
+	AfterEach(func() {
+		// Clean up temp directory
+		if tempPgData != "" {
+			os.RemoveAll(tempPgData)
+		}
+		os.Unsetenv("PGDATA")
+	})
+
+	// Helper function to create .ready files
+	createReadyFile := func(walName string) {
+		path := filepath.Join(archiveStatusDir, walName+".ready")
+		err := os.WriteFile(path, []byte{}, 0644)
+		Expect(err).ToNot(HaveOccurred())
+	}
+
+	Context("when parallel=1", func() {
+		It("should gather only the requested file", func(ctx SpecContext) {
+			// Create .ready files for multiple WAL files
+			createReadyFile("000000010000000000000001")
+			createReadyFile("000000010000000000000002")
+			createReadyFile("000000010000000000000003")
+
+			walList := archiver.GatherWALFilesToArchive(ctx, "pg_wal/000000010000000000000001", 1)
+
+			Expect(walList).To(ConsistOf("pg_wal/000000010000000000000001"))
+		})
+
+		It("should handle when no other .ready files exist", func(ctx SpecContext) {
+			// Only create the requested file
+			createReadyFile("000000010000000000000001")
+
+			walList := archiver.GatherWALFilesToArchive(ctx, "pg_wal/000000010000000000000001", 1)
+
+			Expect(walList).To(ConsistOf("pg_wal/000000010000000000000001"))
+		})
+	})
+
+	Context("when parallel>1", func() {
+		It("should gather multiple files when parallel=4", func(ctx SpecContext) {
+			// Create .ready files for multiple WAL files
+			createReadyFile("000000010000000000000001")
+			createReadyFile("000000010000000000000002")
+			createReadyFile("000000010000000000000003")
+			createReadyFile("000000010000000000000004")
+
+			walList := archiver.GatherWALFilesToArchive(ctx, "pg_wal/000000010000000000000001", 4)
+
+			Expect(walList).To(ConsistOf(
+				"pg_wal/000000010000000000000001",
+				"pg_wal/000000010000000000000002",
+				"pg_wal/000000010000000000000003",
+				"pg_wal/000000010000000000000004",
+			))
+		})
+
+		It("should not exceed parallel limit even when more files are ready", func(ctx SpecContext) {
+			// Create many .ready files
+			for i := 1; i <= 10; i++ {
+				createReadyFile("00000001000000000000000" + string(rune('0'+i)))
+			}
+
+			walList := archiver.GatherWALFilesToArchive(ctx, "pg_wal/000000010000000000000001", 3)
+
+			Expect(walList).To(HaveLen(3))
+		})
+
+		It("should handle when fewer files exist than parallel limit", func(ctx SpecContext) {
+			// Create only 2 .ready files but request parallel=5
+			createReadyFile("000000010000000000000001")
+			createReadyFile("000000010000000000000002")
+
+			walList := archiver.GatherWALFilesToArchive(ctx, "pg_wal/000000010000000000000001", 5)
+
+			// Should only get the files that exist
+			Expect(walList).To(ConsistOf(
+				"pg_wal/000000010000000000000001",
+				"pg_wal/000000010000000000000002",
+			))
+		})
+	})
+
+	Context("edge cases", func() {
+		It("should handle empty archive_status directory", func(ctx SpecContext) {
+			// Don't create any .ready files
+
+			walList := archiver.GatherWALFilesToArchive(ctx, "pg_wal/000000010000000000000001", 3)
+
+			// Should still return the requested file
+			Expect(walList).To(ConsistOf("pg_wal/000000010000000000000001"))
+		})
+
+	})
+
+	Context("other files in directory", func() {
+		It("should ignore non-.ready files in archive_status", func(ctx SpecContext) {
+			// Create .ready files
+			createReadyFile("000000010000000000000001")
+			createReadyFile("000000010000000000000002")
+
+			// Create .done files (should be ignored)
+			donePath := filepath.Join(archiveStatusDir, "000000010000000000000003.done")
+			err := os.WriteFile(donePath, []byte{}, 0644)
+			Expect(err).ToNot(HaveOccurred())
+
+			// Create a random file (should be ignored)
+			randomPath := filepath.Join(archiveStatusDir, "random.txt")
+			err = os.WriteFile(randomPath, []byte{}, 0644)
+			Expect(err).ToNot(HaveOccurred())
+
+			walList := archiver.GatherWALFilesToArchive(ctx, "pg_wal/000000010000000000000001", 5)
+
+			// Should only get .ready files, not .done or other files
+			Expect(walList).To(ConsistOf(
+				"pg_wal/000000010000000000000001",
+				"pg_wal/000000010000000000000002",
+			))
+		})
+		It("should handle timeline history files", func(ctx SpecContext) {
+			// Create timeline history file
+			createReadyFile("00000002.history")
+			createReadyFile("000000010000000000000001")
+
+			walList := archiver.GatherWALFilesToArchive(ctx, "pg_wal/00000002.history", 2)
+
+			Expect(walList).To(ConsistOf(
+				"pg_wal/00000002.history",
+				"pg_wal/000000010000000000000001",
+			))
+		})
+	})
+})

test/e2e/e2e_suite_test.go

@@ -34,6 +34,7 @@ import (
 	"github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/kustomize"
 
 	_ "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/tests/backup"
+	_ "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/tests/parallelarchive"
 	_ "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/tests/replicacluster"
 
 	. "github.com/onsi/ginkgo/v2"

test/e2e/internal/logs/doc.go

@@ -0,0 +1,18 @@
+/*
+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 logs provides utilities for retrieving and parsing pod logs in e2e tests
+package logs

test/e2e/internal/logs/logs.go

@@ -0,0 +1,112 @@
+/*
+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 logs
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+// GetPodContainerLogs retrieves logs from a specific container in a pod and parses them as JSON.
+// All logs are expected to be valid structured JSON. Returns an error if any line is not valid JSON.
+func GetPodContainerLogs(
+	ctx context.Context,
+	clientSet *kubernetes.Clientset,
+	namespace, podName, containerName string,
+	sinceSeconds *int64,
+) ([]map[string]any, error) {
+	podLogOpts := corev1.PodLogOptions{
+		Container: containerName,
+	}
+	if sinceSeconds != nil {
+		podLogOpts.SinceSeconds = sinceSeconds
+	}
+
+	req := clientSet.CoreV1().Pods(namespace).GetLogs(podName, &podLogOpts)
+	stream, err := req.Stream(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("error opening log stream: %w", err)
+	}
+	defer stream.Close()
+
+	var logEntries []map[string]any
+	scanner := bufio.NewScanner(stream)
+	lineNum := 0
+	for scanner.Scan() {
+		lineNum++
+		line := scanner.Text()
+
+		// Skip empty lines
+		if strings.TrimSpace(line) == "" {
+			continue
+		}
+
+		// Check if line is JSON
+		if !strings.HasPrefix(line, "{") {
+			return nil, fmt.Errorf("non-JSON line at line %d: %s", lineNum, line)
+		}
+
+		var logEntry map[string]any
+		if err := json.Unmarshal([]byte(line), &logEntry); err != nil {
+			return nil, fmt.Errorf("invalid JSON at line %d: %w", lineNum, err)
+		}
+
+		logEntries = append(logEntries, logEntry)
+	}
+
+	if err := scanner.Err(); err != nil {
+		return nil, fmt.Errorf("error reading logs: %w", err)
+	}
+
+	return logEntries, nil
+}
+
+// FindArchiveBatches finds "WAL archive batch prepared" log entries and returns the parsed data.
+// Each returned map contains the structured log fields.
+func FindArchiveBatches(logEntries []map[string]any) []map[string]any {
+	var batches []map[string]any
+
+	for _, logEntry := range logEntries {
+		// Check if this is a "WAL archive batch prepared" message
+		if msg, ok := logEntry["msg"].(string); ok && msg == "WAL archive batch prepared" {
+			batches = append(batches, logEntry)
+		}
+	}
+
+	return batches
+}
+
+// FindArchiveBatchCompletions finds "WAL archive batch completed" log entries and returns the parsed data.
+// Each returned map contains the structured log fields.
+func FindArchiveBatchCompletions(logEntries []map[string]any) []map[string]any {
+	var batches []map[string]any
+
+	for _, logEntry := range logEntries {
+		// Check if this is a "WAL archive batch completed" message
+		if msg, ok := logEntry["msg"].(string); ok && msg == "WAL archive batch completed" {
+			batches = append(batches, logEntry)
+		}
+	}
+
+	return batches
+}

test/e2e/internal/objectstore/minio.go

@@ -300,7 +300,7 @@ func newMinioPVC(namespace, name string) *corev1.PersistentVolumeClaim {
 }
 
 // NewMinioArchive creates a new Archive configured to use the Minio object store.
-func NewMinioArchive(namespace, name, minioOSName string) *pluginPgbackrestV1.Archive {
+func NewMinioArchive(namespace, name, minioOSName string, maxParallel int) *pluginPgbackrestV1.Archive {
 	return &pluginPgbackrestV1.Archive{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "Archive",
@@ -312,6 +312,9 @@ func NewMinioArchive(namespace, name, minioOSName string) *pluginPgbackrestV1.Ar
 		},
 		Spec: pluginPgbackrestV1.ArchiveSpec{
 			Configuration: pgbackrestApi.PgbackrestConfiguration{
+				Wal: &pgbackrestApi.WalBackupConfiguration{
+					MaxParallel: maxParallel,
+				},
 				Repositories: []pgbackrestApi.PgbackrestRepository{
 					{
 						PgbackrestCredentials: pgbackrestApi.PgbackrestCredentials{

test/e2e/internal/tests/backup/fixtures.go

@@ -68,7 +68,7 @@ func (s s3BackupPluginBackupPluginRestore) createBackupRestoreTestResources(
 	result := backupRestoreTestResources{}
 
 	result.ObjectStoreResources = objectstore.NewMinioObjectStoreResources(namespace, minio)
-	result.Archive = objectstore.NewMinioArchive(namespace, archiveName, minio)
+	result.Archive = objectstore.NewMinioArchive(namespace, archiveName, minio, 1)
 	result.SrcCluster = newSrcClusterWithPlugin(namespace)
 	result.SrcBackup = newSrcPluginBackup(namespace)
 	result.DstCluster = newDstClusterWithPlugin(namespace)