From 46456a88870934506ede4b67c3dfaa2f2afcee14 Mon Sep 17 00:00:00 2001 From: Gareth George Date: Fri, 22 Dec 2023 07:55:17 +0000 Subject: [PATCH] feat: implement garbage collection of old operations --- internal/oplog/oplog.go | 5 +- internal/orchestrator/orchestrator.go | 3 + internal/orchestrator/taskcollectgarbage.go | 118 ++++++++++++++++++++ internal/orchestrator/taskforget.go | 4 - proto/v1/operations.proto | 2 +- webui/src/components/ActivityBar.tsx | 4 +- webui/src/components/OperationList.tsx | 17 +-- webui/src/components/OperationTree.tsx | 26 ++++- webui/src/components/SnapshotBrowser.tsx | 16 ++- webui/src/state/oplog.ts | 99 ++++++++++------ webui/src/views/AddRepoModal.tsx | 6 +- webui/src/views/App.tsx | 1 - webui/src/views/PlanView.tsx | 2 +- 13 files changed, 237 insertions(+), 66 deletions(-) create mode 100644 internal/orchestrator/taskcollectgarbage.go diff --git a/internal/oplog/oplog.go b/internal/oplog/oplog.go index f641a50c..600dfe14 100644 --- a/internal/oplog/oplog.go +++ b/internal/oplog/oplog.go @@ -248,11 +248,8 @@ func (o *OpLog) nextOperationId(b *bolt.Bucket, unixTimeMs int64) (int64, error) func (o *OpLog) addOperationHelper(tx *bolt.Tx, op *v1.Operation) error { b := tx.Bucket(OpLogBucket) if op.Id == 0 { - if op.UnixTimeStartMs == 0 { - return fmt.Errorf("operation must have a start time") - } var err error - op.Id, err = o.nextOperationId(b, op.UnixTimeStartMs) + op.Id, err = o.nextOperationId(b, time.Now().UnixMilli()) if err != nil { return fmt.Errorf("create next operation ID: %w", err) } diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index fcb6a7e7..7175fd20 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -127,6 +127,9 @@ func (o *Orchestrator) ApplyConfig(cfg *v1.Config) error { zap.L().Info("Applied config to orchestrator, task queue reset. Rescheduling planned tasks now.") // Requeue tasks that are affected by the config change. + o.ScheduleTask(&CollectGarbageTask{ + orchestrator: o, + }, TaskPriorityDefault) for _, plan := range cfg.Plans { t, err := NewScheduledBackupTask(o, plan) if err != nil { diff --git a/internal/orchestrator/taskcollectgarbage.go b/internal/orchestrator/taskcollectgarbage.go new file mode 100644 index 00000000..f0ff54fc --- /dev/null +++ b/internal/orchestrator/taskcollectgarbage.go @@ -0,0 +1,118 @@ +package orchestrator + +import ( + "context" + "fmt" + "time" + + v1 "github.com/garethgeorge/restora/gen/go/v1" + "go.uber.org/zap" +) + +const ( + gcStartupDelay = 5 * time.Second + gcInterval = 24 * time.Hour + // keep operations that are eligible for gc for 30 days OR up to a limit of 100 for any one plan. + // an operation is eligible for gc if: + // - it has no snapshot associated with it + // - it has a forgotten snapshot associated with it + gcHistoryAge = 30 * 24 * time.Hour + gcHistoryMaxCount = 100 +) + +type CollectGarbageTask struct { + orchestrator *Orchestrator // owning orchestrator + firstRun bool +} + +var _ Task = &CollectGarbageTask{} + +func (t *CollectGarbageTask) Name() string { + return "collect garbage" +} + +func (t *CollectGarbageTask) Next(now time.Time) *time.Time { + if !t.firstRun { + t.firstRun = true + runAt := now.Add(gcStartupDelay) + return &runAt + } + + runAt := now.Add(gcInterval) + return &runAt +} + +func (t *CollectGarbageTask) Run(ctx context.Context) error { + oplog := t.orchestrator.OpLog + + // pass 1: identify forgotten snapshots. + snapshotIsForgotten := make(map[string]bool) + if err := oplog.ForAll(func(op *v1.Operation) error { + if snapshotOp, ok := op.Op.(*v1.Operation_OperationIndexSnapshot); ok { + if snapshotOp.OperationIndexSnapshot.Forgot { + snapshotIsForgotten[snapshotOp.OperationIndexSnapshot.Snapshot.Id] = true + } + } + return nil + }); err != nil { + return fmt.Errorf("identifying forgotten snapshots: %w", err) + } + + // pass 2: identify operations that are gc eligible + // - any operation that has no snapshot associated with it + // - any operation that has a forgotten snapshot associated with it + operationsByPlan := make(map[string][]gcOpInfo) + if err := oplog.ForAll(func(op *v1.Operation) error { + if op.SnapshotId == "" || snapshotIsForgotten[op.SnapshotId] { + operationsByPlan[op.PlanId] = append(operationsByPlan[op.PlanId], gcOpInfo{ + id: op.Id, + timestamp: op.UnixTimeStartMs, + }) + } + return nil + }); err != nil { + return fmt.Errorf("identifying gc eligible operations: %w", err) + } + + var gcOps []int64 + curTime := curTimeMillis() + for _, opInfos := range operationsByPlan { + if len(opInfos) >= gcHistoryMaxCount { + for _, opInfo := range opInfos[:len(opInfos)-gcHistoryMaxCount] { + gcOps = append(gcOps, opInfo.id) + } + opInfos = opInfos[len(opInfos)-gcHistoryMaxCount:] + } + + // check if each operation timestamp is old. + for _, opInfo := range opInfos { + if curTime-opInfo.timestamp > gcHistoryAge.Milliseconds() { + gcOps = append(gcOps, opInfo.id) + } + } + } + + // pass 3: remove gc eligible operations + if err := oplog.Delete(gcOps...); err != nil { + return fmt.Errorf("removing gc eligible operations: %w", err) + } + + zap.L().Info("collecting garbage", + zap.Int("forgotten_snapshots", len(snapshotIsForgotten)), + zap.Any("operations_removed", len(gcOps))) + + return nil +} + +func (t *CollectGarbageTask) Cancel(withStatus v1.OperationStatus) error { + return nil +} + +func (t *CollectGarbageTask) OperationId() int64 { + return 0 +} + +type gcOpInfo struct { + id int64 // operation ID + timestamp int64 // unix time milliseconds +} diff --git a/internal/orchestrator/taskforget.go b/internal/orchestrator/taskforget.go index 1f0974dc..3e934949 100644 --- a/internal/orchestrator/taskforget.go +++ b/internal/orchestrator/taskforget.go @@ -99,10 +99,6 @@ func (t *ForgetTask) Run(ctx context.Context) error { continue } } - // Soft delete the operation (can be recovered if necessary, todo: implement recovery). - if e := t.orch.OpLog.Delete(op.Id); err != nil { - err = multierror.Append(err, fmt.Errorf("delete operation %v: %w", op.Id, e)) - } } if len(forgot) > 0 { diff --git a/proto/v1/operations.proto b/proto/v1/operations.proto index 7e7ce6d3..8ebd4169 100644 --- a/proto/v1/operations.proto +++ b/proto/v1/operations.proto @@ -12,7 +12,7 @@ message OperationList { } message Operation { - // required, primary ID of the operation. + // required, primary ID of the operation. ID is sequential based on creation time of the operation. int64 id = 1; // required, repo id if associated with a repo string repo_id = 2; diff --git a/webui/src/components/ActivityBar.tsx b/webui/src/components/ActivityBar.tsx index cbf5996e..83c22c57 100644 --- a/webui/src/components/ActivityBar.tsx +++ b/webui/src/components/ActivityBar.tsx @@ -35,7 +35,7 @@ export const ActivityBar = () => { } }); - return {details.map(details => { - return <>{details.displayName} in progress for plan {details.op.planId} to {details.op.repoId} for {formatDuration(details.details.duration)} + return {details.map((details, idx) => { + return {details.displayName} in progress for plan {details.op.planId} to {details.op.repoId} for {formatDuration(details.details.duration)} })} } \ No newline at end of file diff --git a/webui/src/components/OperationList.tsx b/webui/src/components/OperationList.tsx index 6933127d..c963d48b 100644 --- a/webui/src/components/OperationList.tsx +++ b/webui/src/components/OperationList.tsx @@ -86,13 +86,13 @@ export const OperationList = ({ }; subscribeToOperations(lis); - backupCollector.subscribe(() => { + backupCollector.subscribe(_.debounce(() => { let backups = backupCollector.getAll(); backups.sort((a, b) => { return b.startTimeMs - a.startTimeMs; }); setBackups(backups); - }); + }, 50)); getOperations(req) .then((ops) => { @@ -106,7 +106,10 @@ export const OperationList = ({ }; }, [JSON.stringify(req)]); } else { - backups = useBackups || []; + backups = [...(useBackups || [])]; + backups.sort((a, b) => { + return b.startTimeMs - a.startTimeMs; + }); } if (backups.length === 0) { @@ -277,14 +280,14 @@ export const OperationRow = ({ children: <> Removed snapshots: 
{forgetOp.forget?.map((f) => (
-                  <>
+                  

                     {"removed snapshot " + normalizeSnapshotId(f.id!) + " taken at " + formatTime(f.unixTimeMs!)} 

-                  
+                  

                 ))}
Policy: , diff --git a/webui/src/components/OperationTree.tsx b/webui/src/components/OperationTree.tsx index 7b1ac97e..01c9c5f8 100644 --- a/webui/src/components/OperationTree.tsx +++ b/webui/src/components/OperationTree.tsx @@ -58,13 +58,13 @@ export const OperationTree = ({ }; subscribeToOperations(lis); - backupCollector.subscribe(() => { + backupCollector.subscribe(_.debounce(() => { let backups = backupCollector.getAll(); backups.sort((a, b) => { return b.startTimeMs - a.startTimeMs; }); setBackups(backups); - }); + }, 50)); getOperations(req) .then((ops) => { @@ -79,7 +79,7 @@ export const OperationTree = ({ }, [JSON.stringify(req)]); const treeData = useMemo(() => { - return buildTreeYear(backups); + return buildTreePlan(backups); }, [backups]); if (backups.length === 0) { @@ -199,6 +199,26 @@ export const OperationTree = ({ ); }; +const buildTreePlan = (operations: BackupInfo[]): OpTreeNode[] => { + const grouped = _.groupBy(operations, (op) => { + return op.planId; + }); + + const entries: OpTreeNode[] = _.map(grouped, (value, key) => { + return { + key: "p" + key, + title: "" + key, + children: buildTreeYear(value), + }; + }); + entries.sort(sortByKey); + + if (entries.length === 1) { + return entries[0].children!; + } + return entries; +}; + const buildTreeYear = (operations: BackupInfo[]): OpTreeNode[] => { const grouped = _.groupBy(operations, (op) => { return localISOTime(op.displayTime).substring(0, 4); diff --git a/webui/src/components/SnapshotBrowser.tsx b/webui/src/components/SnapshotBrowser.tsx index 438f6fe8..8f87a7dd 100644 --- a/webui/src/components/SnapshotBrowser.tsx +++ b/webui/src/components/SnapshotBrowser.tsx @@ -1,5 +1,5 @@ import React, { useEffect, useMemo, useState } from "react"; -import { Button, Dropdown, Form, Input, Modal, Space, Tree } from "antd"; +import { Button, Dropdown, Form, Input, Modal, Space, Spin, Tree } from "antd"; import type { DataNode, EventDataNode } from "antd/es/tree"; import { ListSnapshotFilesResponse, @@ -109,7 +109,7 @@ export const SnapshotBrowser = ({ }, { pathPrefix: "/api" } ); - + setTreeData((treeData) => { let toUpdate: DataNode | null = null; for (const node of treeData) { @@ -118,14 +118,14 @@ export const SnapshotBrowser = ({ break; } } - + if (!toUpdate) { return treeData; } - + const toUpdateCopy = { ...toUpdate }; toUpdateCopy.children = respToNodes(resp); - + return treeData.map((node) => { const didUpdate = replaceKeyInTree(node, key as string, toUpdateCopy); if (didUpdate) { @@ -136,6 +136,10 @@ export const SnapshotBrowser = ({ }); }; + if (treeData.length === 0) { + return ; + } + return ( {}; +const restoreFlow = (repoId: string, snapshotId: string, path: string) => { }; diff --git a/webui/src/state/oplog.ts b/webui/src/state/oplog.ts index 83519008..9ef33e04 100644 --- a/webui/src/state/oplog.ts +++ b/webui/src/state/oplog.ts @@ -94,12 +94,13 @@ export interface BackupInfo { startTimeMs: number; endTimeMs: number; status: OperationStatus; - operations: Operation[]; + operations: Operation[]; // operations ordered by their unixTimeStartMs (not ID) repoId?: string; planId?: string; snapshotId?: string; backupLastStatus?: BackupProgressEntry; snapshotInfo?: ResticSnapshot; + forgotten: boolean; } // BackupInfoCollector maps multiple operations to single aggregate 'BackupInfo' objects. @@ -112,31 +113,57 @@ export class BackupInfoCollector { private backupByOpId: { [key: string]: BackupInfo } = {}; private backupBySnapshotId: { [key: string]: BackupInfo } = {}; - private mergeBackups(existing: BackupInfo, newInfo: BackupInfo) { - if (existing.id > newInfo.id) { - existing.id = newInfo.id; - } - existing.startTimeMs = Math.min(existing.startTimeMs, newInfo.startTimeMs); - existing.endTimeMs = Math.max(existing.endTimeMs, newInfo.endTimeMs); - existing.displayTime = new Date(existing.startTimeMs); - existing.displayType = DisplayType.SNAPSHOT; - if (newInfo.startTimeMs >= existing.startTimeMs && newInfo.status !== OperationStatus.STATUS_SYSTEM_CANCELLED) { // don't overwrite with cancelled status since that operation will be hidden. - existing.status = newInfo.status; // use the latest status - } - existing.operations = _.uniqBy( - [...newInfo.operations, ...existing.operations], - (o) => o.id! - ); - existing.operations.sort((a, b) => { + private createBackup(operations: Operation[]): BackupInfo { + // deduplicate and sort operations. + operations.sort((a, b) => { return parseInt(b.unixTimeStartMs!) - parseInt(a.unixTimeStartMs!); }); - if (newInfo.backupLastStatus) { - existing.backupLastStatus = newInfo.backupLastStatus; + + // use the lowest ID of all operations as the ID of the backup, this will be the first created operation. + const id = operations.reduce((prev, curr) => { + return prev < curr.id! ? prev : curr.id!; + }, operations[0].id!); + + const startTimeMs = parseInt(operations[0].unixTimeStartMs!); + const endTimeMs = parseInt(operations[operations.length - 1].unixTimeEndMs!); + const displayTime = new Date(startTimeMs); + let displayType = DisplayType.SNAPSHOT; + if (operations.length === 1) { + displayType = getTypeForDisplay(operations[0]); } - if (newInfo.snapshotInfo) { - existing.snapshotInfo = newInfo.snapshotInfo; + + // use the latest status that is not cancelled. + let statusIdx = operations.length - 1; + let status = OperationStatus.STATUS_SYSTEM_CANCELLED; + while (statusIdx > 0 || shouldHideStatus(status)) { + status = operations[statusIdx].status!; + statusIdx--; } - return existing; + + let backupLastStatus = undefined; + let snapshotInfo = undefined; + let forgotten = false; + for (const op of operations) { + if (op.operationBackup) { + backupLastStatus = op.operationBackup.lastStatus; + } else if (op.operationIndexSnapshot) { + snapshotInfo = op.operationIndexSnapshot.snapshot; + forgotten = op.operationIndexSnapshot.forgot || false; + } + } + + return { + id, + startTimeMs, + endTimeMs, + displayTime, + displayType, + status, + operations, + backupLastStatus, + snapshotInfo, + forgotten, + }; } private operationToBackup(op: Operation): BackupInfo { @@ -167,21 +194,29 @@ export class BackupInfoCollector { return b; } - private addHelper(op: Operation) { + private addHelper(op: Operation): BackupInfo { if (op.snapshotId) { delete this.backupByOpId[op.id!]; - let newInfo = this.operationToBackup(op); const existing = this.backupBySnapshotId[op.snapshotId!]; + let operations: Operation[]; if (existing) { - this.mergeBackups(existing, newInfo); - return existing; + operations = [...existing.operations]; + const opIdx = operations.findIndex((o) => o.id === op.id); + if (opIdx > -1) { + operations[opIdx] = op; + } else { + operations.push(op); + } } else { - this.backupBySnapshotId[op.snapshotId] = newInfo; - return newInfo; + operations = [op]; } + + const newInfo = this.createBackup(operations); + this.backupBySnapshotId[op.snapshotId!] = newInfo; + return newInfo; } else { - const newInfo = this.operationToBackup(op); + const newInfo = this.createBackup([op]); this.backupByOpId[op.id!] = newInfo; return newInfo; } @@ -212,15 +247,11 @@ export class BackupInfoCollector { return backupInfos; } - public addOperationNoNotify(op: Operation) { - this.addHelper(op); - } - public getAll(): BackupInfo[] { const arr = []; arr.push(...Object.values(this.backupByOpId)); arr.push(...Object.values(this.backupBySnapshotId)); - return arr; + return arr.filter((b) => !b.forgotten && !shouldHideStatus(b.status)); } public subscribe( diff --git a/webui/src/views/AddRepoModal.tsx b/webui/src/views/AddRepoModal.tsx index 0baa95ce..33681f48 100644 --- a/webui/src/views/AddRepoModal.tsx +++ b/webui/src/views/AddRepoModal.tsx @@ -77,9 +77,9 @@ export const AddRepoModal = ({ showModal(null); alertsApi.success( "Deleted repo " + - template.id + - " from config but files remain. To release storage delete the files manually. URI: " + - template.uri + template.id + + " from config but files remain. To release storage delete the files manually. URI: " + + template.uri ); } catch (e: any) { alertsApi.error("Operation failed: " + e.message, 15); diff --git a/webui/src/views/App.tsx b/webui/src/views/App.tsx index 72b2139b..016031e5 100644 --- a/webui/src/views/App.tsx +++ b/webui/src/views/App.tsx @@ -68,7 +68,6 @@ export const App: React.FC = () => { {uiBuildVersion} - diff --git a/webui/src/views/PlanView.tsx b/webui/src/views/PlanView.tsx index 868b2647..a7bd426a 100644 --- a/webui/src/views/PlanView.tsx +++ b/webui/src/views/PlanView.tsx @@ -69,7 +69,7 @@ export const PlanView = ({ plan }: React.PropsWithChildren<{ plan: Plan }>) => { <> -

{plan.id}

+ {plan.id}