lee/backrest
1
0
Fork 0
mirror of https://github.com/garethgeorge/backrest.git synced 2026-09-03 14:55:37 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

feat: implement snapshot indexing

Browse Source
This commit is contained in:
garethgeorge
2023-11-15 22:26:52 -08:00
parent 985c0f7a20
commit 1c06d2ebe3
3 changed files with 135 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/orchestrator/orchestrator.go
+22 -19
View File
@@ -95,26 +95,28 @@ func (o *Orchestrator) GetPlan(planId string) (*v1.Plan, error) {
// Run is the main orchestration loop. Cancel the context to stop the loop.
func (o *Orchestrator) Run(mainCtx context.Context) error {
zap.L().Info("Starting orchestrator loop")
zap.L().Info("starting orchestrator loop")
o.mu.Lock()
o.configUpdates = make(chan *v1.Config)
o.mu.Unlock()
for mainCtx.Err() == nil {
for {
o.mu.Lock()
config := o.config
o.mu.Unlock()
o.runVersion(mainCtx, config)
zap.L().Info("Restarting orchestrator loop")
if o.runVersion(mainCtx, config) {
zap.L().Info("restarting orchestrator loop")
} else {
zap.L().Info("exiting orchestrator loop, context cancelled.")
break
}
}
zap.L().Info("Exited orchestrator loop, context cancelled.")
return nil
}
// runImmutable is a helper function for Run() that runs the orchestration loop with a single version of the config.
func (o *Orchestrator) runVersion(mainCtx context.Context, config *v1.Config) {
func (o *Orchestrator) runVersion(mainCtx context.Context, config *v1.Config) bool {
lock := sync.Mutex{}
ctx, cancel := context.WithCancel(mainCtx)
@@ -126,12 +128,12 @@ func (o *Orchestrator) runVersion(mainCtx context.Context, config *v1.Config) {
runAt := t.Next(curTime)
if runAt == nil {
zap.L().Debug("Task has no next run, not scheduling.", zap.String("task", t.Name()))
zap.L().Debug("task has no next run, not scheduling.", zap.String("task", t.Name()))
return
}
timer := time.NewTimer(runAt.Sub(curTime))
zap.L().Debug("Scheduling task", zap.String("task", t.Name()), zap.String("runAt", runAt.Format(time.RFC3339)))
zap.L().Info("scheduling task", zap.String("task", t.Name()), zap.String("runAt", runAt.Format(time.RFC3339)))
wg.Add(1)
go func() {
@@ -139,22 +141,22 @@ func (o *Orchestrator) runVersion(mainCtx context.Context, config *v1.Config) {
select {
case <-ctx.Done():
timer.Stop()
zap.L().Debug("Cancelling scheduled (but not running) task, orchestrator context is cancelled.", zap.String("task", t.Name()))
zap.L().Debug("cancelled scheduled (but not running) task, orchestrator context is cancelled.", zap.String("task", t.Name()))
return
case <-timer.C:
lock.Lock()
defer lock.Unlock()
zap.L().Debug("Running task", zap.String("task", t.Name()))
zap.L().Info("running task", zap.String("task", t.Name()))
// Task execution runs with mainCtx meaning config changes do not interrupt it, but cancelling the orchestration loop will.
if err := t.Run(mainCtx); err != nil {
zap.L().Error("Task failed", zap.String("task", t.Name()), zap.Error(err))
zap.L().Error("task failed", zap.String("task", t.Name()), zap.Error(err))
} else {
zap.L().Debug("Task finished", zap.String("task", t.Name()))
zap.L().Debug("task finished", zap.String("task", t.Name()))
}
if ctx.Err() != nil {
zap.L().Debug("Not attempting to reschedule task, orchestrator context is cancelled.", zap.String("task", t.Name()))
zap.L().Debug("not attempting to reschedule task, orchestrator context is cancelled.", zap.String("task", t.Name()))
return
}
@@ -167,7 +169,7 @@ func (o *Orchestrator) runVersion(mainCtx context.Context, config *v1.Config) {
for _, plan := range config.Plans {
t, err := NewScheduledBackupTask(o, plan)
if err != nil {
zap.L().Error("Failed to create backup task for plan", zap.String("plan", plan.Id), zap.Error(err))
zap.L().Error("failed to create backup task for plan", zap.String("plan", plan.Id), zap.Error(err))
}
execTask(t)
@@ -178,15 +180,16 @@ func (o *Orchestrator) runVersion(mainCtx context.Context, config *v1.Config) {
select {
case t := <-o.externTasks:
execTask(t)
case <-ctx.Done():
case <-mainCtx.Done():
zap.L().Info("orchestrator context cancelled, shutting down orchestrator")
cancel()
wg.Wait()
return
return false
case <-o.configUpdates:
zap.L().Info("Orchestrator received config change, waiting for in-progress operations")
zap.L().Info("orchestrator received config change, waiting for in-progress operations then restarting")
cancel()
wg.Wait()
return
return true
}
}
}
internal/orchestrator/repo.go
+43 -14
View File
@@ -10,6 +10,7 @@ import (
v1 "github.com/garethgeorge/resticui/gen/go/v1"
"github.com/garethgeorge/resticui/pkg/restic"
"go.uber.org/zap"
)
// RepoOrchestrator is responsible for managing a single repo.
@@ -19,8 +20,10 @@ type RepoOrchestrator struct {
repoConfig *v1.Repo
repo *restic.Repo
// TODO: decide if snapshot caching is a good idea. We gain performance but
// increase background memory use by a small amount at all times (probably on the order of 1MB).
snapshotsMu sync.Mutex // enable very fast snapshot access IF no update is required.
snapshotsAge time.Time
snapshotsResetTimer *time.Timer
snapshots []*restic.Snapshot
}
@@ -31,13 +34,23 @@ func newRepoOrchestrator(repoConfig *v1.Repo, repo *restic.Repo) *RepoOrchestrat
}
}
func (r *RepoOrchestrator) updateSnapshotsIfNeeded(ctx context.Context) error {
r.snapshotsMu.Lock()
defer r.snapshotsMu.Unlock()
if time.Since(r.snapshotsAge) > 10 * time.Minute {
r.snapshots = nil
func (r *RepoOrchestrator) updateSnapshotsIfNeeded(ctx context.Context, force bool) error {
if r.snapshots != nil {
return nil
}
if r.snapshotsResetTimer != nil {
if !r.snapshotsResetTimer.Stop() {
<-r.snapshotsResetTimer.C
}
}
r.snapshotsResetTimer = time.AfterFunc(10 * time.Minute, func() {
r.snapshotsMu.Lock()
defer r.snapshotsMu.Unlock()
r.snapshots = nil
})
if r.snapshots != nil {
return nil
}
@@ -45,7 +58,9 @@ func (r *RepoOrchestrator) updateSnapshotsIfNeeded(ctx context.Context) error {
r.mu.Lock()
defer r.mu.Unlock()
snapshots, err := r.repo.Snapshots(ctx, restic.WithPropagatedEnvVars(restic.EnvToPropagate...))
startTime := time.Now()
snapshots, err := r.repo.Snapshots(ctx, restic.WithPropagatedEnvVars(restic.EnvToPropagate...), restic.WithFlags("--latest", "1000"))
if err != nil {
return fmt.Errorf("failed to update snapshots: %w", err)
}
@@ -55,26 +70,29 @@ func (r *RepoOrchestrator) updateSnapshotsIfNeeded(ctx context.Context) error {
})
r.snapshots = snapshots
zap.L().Debug("Updated snapshots", zap.String("repo", r.repoConfig.Id), zap.Duration("duration", time.Since(startTime)))
return nil
}
func (r *RepoOrchestrator) Snapshots(ctx context.Context) ([]*restic.Snapshot, error) {
if err := r.updateSnapshotsIfNeeded(ctx); err != nil {
r.snapshotsMu.Lock()
defer r.snapshotsMu.Unlock()
if err := r.updateSnapshotsIfNeeded(ctx, false); err != nil {
return nil, err
}
r.snapshotsMu.Lock()
defer r.snapshotsMu.Unlock()
return r.snapshots, nil
}
func (r *RepoOrchestrator) SnapshotsForPlan(ctx context.Context, plan *v1.Plan) ([]*restic.Snapshot, error) {
if err := r.updateSnapshotsIfNeeded(ctx); err != nil {
return nil, err
}
r.snapshotsMu.Lock()
defer r.snapshotsMu.Unlock()
if err := r.updateSnapshotsIfNeeded(ctx, false); err != nil {
return nil, err
}
return filterSnapshotsForPlan(r.snapshots, plan), nil
}
@@ -87,6 +105,8 @@ func (r *RepoOrchestrator) Backup(ctx context.Context, plan *v1.Plan, progressCa
r.mu.Lock()
defer r.mu.Unlock()
startTime := time.Now()
var opts []restic.BackupOption
opts = append(opts, restic.WithBackupPaths(plan.Paths...))
opts = append(opts, restic.WithBackupExcludes(plan.Excludes...))
@@ -101,6 +121,15 @@ func (r *RepoOrchestrator) Backup(ctx context.Context, plan *v1.Plan, progressCa
if err != nil {
return nil, fmt.Errorf("failed to backup: %w", err)
}
// Reset snapshots since a new backup has been added.
r.snapshotsMu.Lock()
r.snapshots = nil
r.snapshotsMu.Unlock()
zap.L().Debug("Backup completed", zap.String("repo", r.repoConfig.Id), zap.Duration("duration", time.Since(startTime)))
return summary, nil
}
internal/orchestrator/tasks.go
+70 -5
View File
@@ -98,11 +98,13 @@ func backupHelper(ctx context.Context, orchestrator *Orchestrator, plan *v1.Plan
Op: backupOp,
}
return WithOperation(orchestrator.oplog, op, func() error {
startTime := time.Now()
err := WithOperation(orchestrator.oplog, op, func() error {
zap.L().Info("Starting backup", zap.String("plan", plan.Id))
repo, err := orchestrator.GetRepo(plan.Repo)
if err != nil {
return fmt.Errorf("failed to get repo %q: %w", plan.Repo, err)
return fmt.Errorf("couldn't get repo %q: %w", plan.Repo, err)
}
lastSent := time.Now() // debounce progress updates, these can endup being very frequent.
@@ -116,10 +118,10 @@ func backupHelper(ctx context.Context, orchestrator *Orchestrator, plan *v1.Plan
if err := orchestrator.oplog.Update(op); err != nil {
zap.S().Errorf("failed to update oplog with progress for backup: %v", err)
}
zap.L().Debug("Backup progress", zap.Float64("progress", entry.PercentDone))
zap.L().Debug("backup progress", zap.Float64("progress", entry.PercentDone))
})
if err != nil {
return fmt.Errorf("failed to backup repo %q: %w", plan.Repo, err)
return fmt.Errorf("repo.Backup for repo %q: %w", plan.Repo, err)
}
backupOp.OperationBackup.LastStatus = summary.ToProto()
@@ -127,9 +129,72 @@ func backupHelper(ctx context.Context, orchestrator *Orchestrator, plan *v1.Plan
return fmt.Errorf("update oplog with summary for backup: %v", err)
}
zap.L().Info("Backup complete", zap.String("plan", plan.Id))
zap.L().Info("backup complete", zap.String("plan", plan.Id), zap.Duration("duration", time.Since(startTime)))
return nil
})
if err != nil {
return fmt.Errorf("backup operation: %w", err)
}
// this could alternatively be a separate operation, but it probably makes sense to index snapshots immediately after a backup.
if err := indexSnapshotsHelper(ctx, orchestrator, plan); err != nil {
return fmt.Errorf("reindexing snapshots after backup operation: %w", err)
}
return nil
}
func indexSnapshotsHelper(ctx context.Context, orchestrator *Orchestrator, plan *v1.Plan) error {
repo, err := orchestrator.GetRepo(plan.Repo)
if err != nil {
return fmt.Errorf("couldn't get repo %q: %w", plan.Repo, err)
}
snapshots, err := repo.SnapshotsForPlan(ctx, plan)
if err != nil {
return fmt.Errorf("get snapshots for plan %q: %w", plan.Id, err)
}
startTime := time.Now()
alreadyIndexed := 0
opTime := curTimeMillis()
var indexOps []*v1.Operation
for _, snapshot := range snapshots {
opid, err := orchestrator.oplog.HasIndexedSnapshot(snapshot.Id)
if err != nil {
return fmt.Errorf("HasIndexSnapshot for snapshot %q: %w", snapshot.Id, err)
}
if opid < 0 {
continue
}
indexOps = append(indexOps, &v1.Operation{
RepoId: plan.Repo,
PlanId: plan.Id,
UnixTimeStartMs: opTime,
UnixTimeEndMs: opTime,
Status: v1.OperationStatus_STATUS_SUCCESS,
Op: &v1.Operation_OperationIndexSnapshot{
OperationIndexSnapshot: &v1.OperationIndexSnapshot{
Snapshot: snapshot.ToProto(),
},
},
})
}
if err := orchestrator.oplog.BulkAdd(indexOps); err != nil {
return fmt.Errorf("BulkAdd snapshot operations: %w", err)
}
zap.L().Debug("Indexed snapshots",
zap.String("plan", plan.Id),
zap.Duration("duration", time.Since(startTime)),
zap.Int("alreadyIndexed", alreadyIndexed),
zap.Int("newlyAdded", len(snapshots) - alreadyIndexed),
)
return err
}
// WithOperation is a utility that creates an operation to track the function's execution.