APPS-1243 Improve list backups performance

go.mod

@@ -5,7 +5,7 @@ go 1.22
 require (
 	github.com/aerospike/aerospike-client-go/v7 v7.6.0
 	github.com/aerospike/aerospike-management-lib v1.4.0
-	github.com/aerospike/backup-go v0.0.0-20240730103930-5ef82c03f86f
+	github.com/aerospike/backup-go v0.0.0-20240801154156-cd3f281d358d
 	github.com/aws/aws-sdk-go-v2 v1.30.3
 	github.com/aws/aws-sdk-go-v2/config v1.27.27
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.58.2

go.sum

@@ -222,6 +222,8 @@ github.com/aerospike/backup-go v0.0.0-20240730084743-64d81fa5d7a4 h1:YvY6Gyedc3h
 github.com/aerospike/backup-go v0.0.0-20240730084743-64d81fa5d7a4/go.mod h1:WfFo03sY4w5qu3IALjqNQaPZVe+0p7GqYPG+qifkm+8=
 github.com/aerospike/backup-go v0.0.0-20240730103930-5ef82c03f86f h1:7gVhto19JTcDKBG2cMnEP3I+ktr+05ny32NjtaLvBH4=
 github.com/aerospike/backup-go v0.0.0-20240730103930-5ef82c03f86f/go.mod h1:WfFo03sY4w5qu3IALjqNQaPZVe+0p7GqYPG+qifkm+8=
+github.com/aerospike/backup-go v0.0.0-20240801154156-cd3f281d358d h1:4EbDTn8ub8gl70oRwAOQiGJ+xQIx1FrzU5nbQcxxE3g=
+github.com/aerospike/backup-go v0.0.0-20240801154156-cd3f281d358d/go.mod h1:WfFo03sY4w5qu3IALjqNQaPZVe+0p7GqYPG+qifkm+8=
 github.com/aws/aws-sdk-go-v2 v1.26.1 h1:5554eUqIYVWpU0YmeeYZ0wU64H2VLBs8TlhRB2L+EkA=
 github.com/aws/aws-sdk-go-v2 v1.26.1/go.mod h1:ffIFB97e2yNsv4aTSGkqtHnppsIJzw7G7BReUZ3jCXM=
 github.com/aws/aws-sdk-go-v2 v1.27.0 h1:7bZWKoXhzI+mMR/HjdMx8ZCC5+6fY0lS5tr0bbgiLlo=

internal/server/restore_handlers.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/aerospike/backup/pkg/model"
+	"github.com/aerospike/backup/pkg/service"
 )
 
 // @Summary     Trigger an asynchronous full restore operation.
@@ -38,7 +39,7 @@ func (ws *HTTPServer) restoreFullHandler(w http.ResponseWriter, r *http.Request)
 			Dir:            request.SourceStorage.Path,
 		}
 
-		jobID, err := ws.restoreService.Restore(requestInternal)
+		jobID, err := ws.restoreManager.Restore(requestInternal)
 		if err != nil {
 			http.Error(w, err.Error(), http.StatusBadRequest)
 			return
@@ -77,7 +78,7 @@ func (ws *HTTPServer) restoreIncrementalHandler(w http.ResponseWriter, r *http.R
 			RestoreRequest: request,
 			Dir:            request.SourceStorage.Path,
 		}
-		jobID, err := ws.restoreService.Restore(requestInternal)
+		jobID, err := ws.restoreManager.Restore(requestInternal)
 		if err != nil {
 			http.Error(w, err.Error(), http.StatusBadRequest)
 			return
@@ -113,7 +114,7 @@ func (ws *HTTPServer) restoreByTimeHandler(w http.ResponseWriter, r *http.Reques
 			http.Error(w, err.Error(), http.StatusBadRequest)
 			return
 		}
-		jobID, err := ws.restoreService.RestoreByTime(&request)
+		jobID, err := ws.restoreManager.RestoreByTime(&request)
 		if err != nil {
 			slog.Error("Restore by timestamp failed", "routine", request.Routine, "err", err)
 			http.Error(w, err.Error(), http.StatusBadRequest)
@@ -148,7 +149,7 @@ func (ws *HTTPServer) restoreStatusHandler(w http.ResponseWriter, r *http.Reques
 		return
 	}
 	w.Header().Set("Content-Type", "application/json")
-	status, err := ws.restoreService.JobStatus(jobID)
+	status, err := ws.restoreManager.JobStatus(service.RestoreJobID(jobID))
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusNotFound)
 	} else {
@@ -198,7 +199,7 @@ func (ws *HTTPServer) retrieveConfig(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	buf, err := ws.restoreService.RetrieveConfiguration(name, time.UnixMilli(timestamp))
+	buf, err := ws.restoreManager.RetrieveConfiguration(name, time.UnixMilli(timestamp))
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusNotFound)
 		return

internal/server/server.go

@@ -83,7 +83,7 @@ type HTTPServer struct {
 	rateLimiter    *IPRateLimiter
 	whiteList      *ipWhiteList
 	scheduler      quartz.Scheduler
-	restoreService service.RestoreService
+	restoreManager service.RestoreManager
 	backupBackends service.BackendsHolder
 	handlerHolder  service.BackupHandlerHolder
 }
@@ -111,7 +111,7 @@ func NewHTTPServer(
 		rateLimiter:    rateLimiter,
 		whiteList:      newIPWhiteList(serverConfig.GetRateOrDefault().GetWhiteListOrDefault()),
 		scheduler:      scheduler,
-		restoreService: service.NewRestoreMemory(backends, config, shared.NewRestoreGo()),
+		restoreManager: service.NewRestoreManager(backends, config, shared.NewRestoreGo()),
 		backupBackends: backends,
 		handlerHolder:  handlerHolder,
 	}

pkg/model/aerospike_cluster.go

@@ -34,7 +34,7 @@ type AerospikeCluster struct {
 	Credentials *Credentials `yaml:"credentials,omitempty" json:"credentials,omitempty"`
 	// The cluster TLS configuration.
 	TLS *TLS `yaml:"tls,omitempty" json:"tls,omitempty"`
-	//specifies the size of the Aerospike Connection Queue per node
+	// specifies the size of the Aerospike Connection Queue per node
 	ConnectionQueueSize *int `yaml:"connection-queue-size,omitempty" json:"connection-queue-size,omitempty" example:"100"`
 }
 

pkg/service/backup_backend.go

@@ -6,10 +6,13 @@ import (
 	"fmt"
 	"log/slog"
 	"path/filepath"
+	"sort"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/aerospike/backup/pkg/model"
+	"github.com/aerospike/backup/pkg/util"
 	"gopkg.in/yaml.v3"
 )
 
@@ -109,12 +112,65 @@ func (b *BackupBackend) FullBackupList(timebounds *model.TimeBounds) ([]model.Ba
 	return b.fromSubfolders(timebounds, b.fullBackupsPath)
 }
 
+func (b *BackupBackend) FindLastFullBackup(toTime time.Time) ([]model.BackupDetails, error) {
+	fullBackupList, err := b.FullBackupList(model.NewTimeBoundsTo(toTime))
+	if err != nil {
+		return nil, fmt.Errorf("cannot read full backup list: %v", err)
+	}
+
+	fullBackup := latestFullBackupBeforeTime(fullBackupList, toTime) // it's a list of namespaces
+	if len(fullBackup) == 0 {
+		return nil, fmt.Errorf("no full backup found at %s", toTime)
+	}
+	return fullBackup, nil
+}
+
+// latestFullBackupBeforeTime returns list of backups with same creation time, latest before upperBound.
+func latestFullBackupBeforeTime(allBackups []model.BackupDetails, upperBound time.Time) []model.BackupDetails {
+	var result []model.BackupDetails
+	var latestTime time.Time
+	for i := range allBackups {
+		current := &allBackups[i]
+		if current.Created.After(upperBound) {
+			continue
+		}
+
+		if len(result) == 0 || latestTime.Before(current.Created) {
+			latestTime = current.Created
+			result = []model.BackupDetails{*current}
+		} else if current.Created.Equal(latestTime) {
+			result = append(result, *current)
+		}
+	}
+	return result
+}
+
+func (b *BackupBackend) FindIncrementalBackupsForNamespace(bounds *model.TimeBounds, namespace string,
+) ([]model.BackupDetails, error) {
+	allIncrementalBackupList, err := b.IncrementalBackupList(bounds)
+	if err != nil {
+		return nil, err
+	}
+	var filteredIncrementalBackups []model.BackupDetails
+	for _, b := range allIncrementalBackupList {
+		if b.Namespace == namespace {
+			filteredIncrementalBackups = append(filteredIncrementalBackups, b)
+		}
+	}
+	// Sort in place
+	sort.Slice(filteredIncrementalBackups, func(i, j int) bool {
+		return filteredIncrementalBackups[i].Created.Before(filteredIncrementalBackups[j].Created)
+	})
+
+	return filteredIncrementalBackups, nil
+}
+
 func (b *BackupBackend) detailsFromPaths(timebounds *model.TimeBounds, useCache bool,
 	paths ...string) []model.BackupDetails {
 	// each path contains a backup of specific time
 	backupDetails := make([]model.BackupDetails, 0, len(paths))
 	for _, path := range paths {
-		namespaces, err := b.lsDir(filepath.Join(path, model.DataDirectory))
+		namespaces, err := b.lsDir(filepath.Join(path, model.DataDirectory), nil)
 		if err != nil {
 			slog.Warn("Cannot list backup dir", "path", path, "err", err)
 			continue
@@ -133,12 +189,18 @@ func (b *BackupBackend) detailsFromPaths(timebounds *model.TimeBounds, useCache
 	return backupDetails
 }
 
-func (b *BackupBackend) fromSubfolders(timebounds *model.TimeBounds,
-	backupFolder string) ([]model.BackupDetails, error) {
-	subfolders, err := b.lsDir(backupFolder)
+func (b *BackupBackend) fromSubfolders(timebounds *model.TimeBounds, backupFolder string,
+) ([]model.BackupDetails, error) {
+	var after *string
+	if timebounds.FromTime != nil {
+		after = util.Ptr(formatTime(*timebounds.FromTime))
+	}
+
+	subfolders, err := b.lsDir(backupFolder, after)
 	if err != nil {
 		return nil, err
 	}
+
 	return b.detailsFromPaths(timebounds, true, subfolders...), nil
 }
 

pkg/service/backup_list_reader.go

@@ -1,6 +1,10 @@
 package service
 
-import "github.com/aerospike/backup/pkg/model"
+import (
+	"time"
+
+	"github.com/aerospike/backup/pkg/model"
+)
 
 // BackupListReader allows to read list of existing backups
 type BackupListReader interface {
@@ -16,4 +20,10 @@ type BackupListReader interface {
 
 	// ReadClusterConfiguration return backed up cluster configuration as a compressed zip.
 	ReadClusterConfiguration(path string) ([]byte, error)
+
+	// FindLastFullBackup return last full backup prior to given time
+	FindLastFullBackup(toTime time.Time) ([]model.BackupDetails, error)
+
+	// FindIncrementalBackupsForNamespace return all incremental backups in given range, sorted by time
+	FindIncrementalBackupsForNamespace(bounds *model.TimeBounds, namespace string) ([]model.BackupDetails, error)
 }

pkg/service/jobs_holder.go

@@ -8,27 +8,29 @@ import (
 	"github.com/aerospike/backup/pkg/model"
 )
 
+type RestoreJobID int
+
 type JobsHolder struct {
 	sync.Mutex
-	restoreJobs map[int]*model.RestoreJobStatus
+	restoreJobs map[RestoreJobID]*model.RestoreJobStatus
 }
 
 func NewJobsHolder() *JobsHolder {
 	return &JobsHolder{
-		restoreJobs: make(map[int]*model.RestoreJobStatus),
+		restoreJobs: make(map[RestoreJobID]*model.RestoreJobStatus),
 	}
 }
 
-func (h *JobsHolder) newJob() int {
+func (h *JobsHolder) newJob() RestoreJobID {
 	// #nosec G404
-	jobID := rand.Int()
+	jobID := RestoreJobID(rand.Int())
 	h.Lock()
 	defer h.Unlock()
 	h.restoreJobs[jobID] = model.NewRestoreJobStatus()
 	return jobID
 }
 
-func (h *JobsHolder) getStatus(jobID int) (*model.RestoreJobStatus, error) {
+func (h *JobsHolder) getStatus(jobID RestoreJobID) (*model.RestoreJobStatus, error) {
 	h.Lock()
 	defer h.Unlock()
 	jobStatus, exists := h.restoreJobs[jobID]
@@ -39,7 +41,7 @@ func (h *JobsHolder) getStatus(jobID int) (*model.RestoreJobStatus, error) {
 	return &copyJob, nil
 }
 
-func (h *JobsHolder) increaseStats(jobID int, newStats *model.RestoreResult) {
+func (h *JobsHolder) increaseStats(jobID RestoreJobID, newStats *model.RestoreResult) {
 	h.Lock()
 	defer h.Unlock()
 	current, found := h.restoreJobs[jobID]
@@ -57,7 +59,7 @@ func (h *JobsHolder) increaseStats(jobID int, newStats *model.RestoreResult) {
 	}
 }
 
-func (h *JobsHolder) setDone(jobID int) {
+func (h *JobsHolder) setDone(jobID RestoreJobID) {
 	h.Lock()
 	defer h.Unlock()
 	current, found := h.restoreJobs[jobID]
@@ -66,7 +68,7 @@ func (h *JobsHolder) setDone(jobID int) {
 	}
 }
 
-func (h *JobsHolder) setFailed(jobID int, err error) {
+func (h *JobsHolder) setFailed(jobID RestoreJobID, err error) {
 	h.Lock()
 	defer h.Unlock()
 	current, found := h.restoreJobs[jobID]

pkg/service/os_accessor.go

@@ -7,6 +7,7 @@ import (
 	"log/slog"
 	"os"
 	"path/filepath"
+	"sort"
 	"strings"
 
 	"github.com/aerospike/backup/pkg/model"
@@ -78,23 +79,48 @@ func (o *OSDiskAccessor) write(filePath string, data []byte) error {
 	return os.WriteFile(filePath, data, 0600)
 }
 
-func (o *OSDiskAccessor) lsDir(path string) ([]string, error) {
-	content, err := os.ReadDir(path)
+func (o *OSDiskAccessor) lsDir(path string, after *string) ([]string, error) {
+	var result []string
+
+	err := filepath.WalkDir(path, func(p string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+
+		// Skip the root directory
+		if p == path {
+			return nil
+		}
+
+		// We're only interested in directories
+		if !d.IsDir() {
+			return nil
+		}
+
+		// Get the relative path
+		relPath, err := filepath.Rel(path, p)
+		if err != nil {
+			return fmt.Errorf("error getting relative path: %w", err)
+		}
+
+		// If 'after' is set, skip directories that come before to it lexicographically
+		if after != nil && relPath < *after {
+			return nil
+		}
+
+		result = append(result, p)
+		return filepath.SkipDir // Don't descend into this directory
+	})
+
 	if err != nil {
 		if os.IsNotExist(err) {
 			return []string{}, nil
 		}
-		return nil, err
+		return nil, fmt.Errorf("error walking directory: %w", err)
 	}
 
-	var onlyDirs []string
-	for _, c := range content {
-		if c.IsDir() {
-			fullPath := filepath.Join(path, c.Name())
-			onlyDirs = append(onlyDirs, fullPath)
-		}
-	}
-	return onlyDirs, nil
+	sort.Strings(result) // Ensure consistent ordering
+	return result, nil
 }
 
 func (o *OSDiskAccessor) lsFiles(path string) ([]string, error) {
@@ -124,7 +150,7 @@ func (o *OSDiskAccessor) DeleteFolder(pathToDelete string) error {
 	}
 
 	parentDir := filepath.Dir(pathToDelete)
-	lsDir, err := o.lsDir(parentDir)
+	lsDir, err := o.lsDir(parentDir, nil)
 	if err != nil {
 		return err
 	}