diff --git a/internal/api/server.go b/internal/api/server.go index 51eca3fe..462c375d 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -135,6 +135,7 @@ func (s *Server) GetOperationEvents(_ *emptypb.Empty, stream v1.ResticUI_GetOper errorChan := make(chan error) defer close(errorChan) callback := func(eventType oplog.EventType, op *v1.Operation) { + zap.S().Debug("Sending an event") var eventTypeMapped v1.OperationEventType switch eventType { case oplog.EventTypeOpCreated: @@ -151,13 +152,14 @@ func (s *Server) GetOperationEvents(_ *emptypb.Empty, stream v1.ResticUI_GetOper Operation: op, } - if err := stream.Send(event); err != nil { - errorChan <- fmt.Errorf("failed to send event: %w", err) - } + go func() { + if err := stream.Send(event); err != nil { + errorChan <- fmt.Errorf("failed to send event: %w", err) + } + }() } s.oplog.Subscribe(&callback) defer s.oplog.Unsubscribe(&callback) - select { case <-stream.Context().Done(): return nil diff --git a/internal/config/config.go b/internal/config/config.go index 9880cc50..87018875 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -15,8 +15,8 @@ var ErrConfigNotFound = fmt.Errorf("config not found") var configDirFlag = flag.String("config_dir", "", "The directory to store the config file") var Default ConfigStore = &CachingValidatingStore{ - ConfigStore: &YamlFileStore{ - Path: path.Join(configDir(*configDirFlag), "config.yaml"), + ConfigStore: &JsonFileStore{ + Path: path.Join(configDir(*configDirFlag), "config.json"), }, } diff --git a/internal/config/yamlstore.go b/internal/config/jsonstore.go similarity index 58% rename from internal/config/yamlstore.go rename to internal/config/jsonstore.go index ac3ad47b..aa51581b 100644 --- a/internal/config/yamlstore.go +++ b/internal/config/jsonstore.go @@ -1,7 +1,6 @@ package config import ( - "encoding/json" "errors" "fmt" "os" @@ -11,17 +10,16 @@ import ( v1 "github.com/garethgeorge/resticui/gen/go/v1" "github.com/google/renameio" "google.golang.org/protobuf/encoding/protojson" - yaml "gopkg.in/yaml.v3" ) -type YamlFileStore struct { +type JsonFileStore struct { Path string mu sync.Mutex } -var _ ConfigStore = &YamlFileStore{} +var _ ConfigStore = &JsonFileStore{} -func (f *YamlFileStore) Get() (*v1.Config, error) { +func (f *JsonFileStore) Get() (*v1.Config, error) { f.mu.Lock() defer f.mu.Unlock() @@ -33,12 +31,6 @@ func (f *YamlFileStore) Get() (*v1.Config, error) { return nil, fmt.Errorf("failed to read config file: %w", err) } - - data, err = yamlToJson(data) - if err != nil { - return nil, fmt.Errorf("failed to parse YAML config: %w", err) - } - var config v1.Config if err = protojson.Unmarshal(data, &config); err != nil { @@ -52,7 +44,7 @@ func (f *YamlFileStore) Get() (*v1.Config, error) { return &config, nil } -func (f *YamlFileStore) Update(config *v1.Config) error { +func (f *JsonFileStore) Update(config *v1.Config) error { f.mu.Lock() defer f.mu.Unlock() @@ -65,11 +57,6 @@ func (f *YamlFileStore) Update(config *v1.Config) error { return fmt.Errorf("failed to marshal config: %w", err) } - data, err = jsonToYaml(data) - if err != nil { - return fmt.Errorf("failed to convert config to yaml: %w", err) - } - err = os.MkdirAll(filepath.Dir(f.Path), 0755) if err != nil { return fmt.Errorf("failed to create config directory: %w", err) @@ -82,23 +69,3 @@ func (f *YamlFileStore) Update(config *v1.Config) error { return nil } - -func jsonToYaml(data []byte) ([]byte, error) { - var config interface{} - err := json.Unmarshal(data, &config) - if err != nil { - return nil, fmt.Errorf("failed to unmarshal config: %w", err) - } - - return yaml.Marshal(config) -} - -func yamlToJson(data []byte) ([]byte, error) { - var config interface{} - err := yaml.Unmarshal(data, &config) - if err != nil { - return nil, fmt.Errorf("failed to unmarshal config: %w", err) - } - - return json.Marshal(config) -} \ No newline at end of file diff --git a/internal/config/validate.go b/internal/config/validate.go index 22ada8b5..731bc21d 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -71,6 +71,12 @@ func validatePlan(plan *v1.Plan, repos map[string]*v1.Repo) error { err = multierror.Append(err, fmt.Errorf("path is required")) } + for idx, p := range plan.Paths { + if p == "" { + err = multierror.Append(err, fmt.Errorf("path[%d] cannot be empty", idx)) + } + } + if plan.Repo == "" { err = multierror.Append(err,fmt.Errorf("repo is required")) } diff --git a/internal/database/oplog/oplog.go b/internal/database/oplog/oplog.go index 0cfaab2b..26957f65 100644 --- a/internal/database/oplog/oplog.go +++ b/internal/database/oplog/oplog.go @@ -12,6 +12,7 @@ import ( "github.com/garethgeorge/resticui/internal/database/indexutil" "github.com/garethgeorge/resticui/internal/database/serializationutil" bolt "go.etcd.io/bbolt" + "go.uber.org/zap" "google.golang.org/protobuf/proto" ) @@ -57,8 +58,40 @@ func NewOpLog(databasePath string) (*OpLog, error) { SystemBucket, OpLogBucket, RepoIndexBucket, PlanIndexBucket, IndexedSnapshotsSetBucket, } { if _, err := tx.CreateBucketIfNotExists(bucket); err != nil { - return fmt.Errorf("error creating bucket %s: %s", string(bucket), err) + return fmt.Errorf("creating bucket %s: %s", string(bucket), err) } + + // Validate the operation log on startup. + sysBucket := tx.Bucket(SystemBucket) + opLogBucket := tx.Bucket(OpLogBucket) + c := opLogBucket.Cursor() + if lastValidated := sysBucket.Get([]byte("last_validated")); lastValidated != nil { + c.Seek(lastValidated) + } + for k, v := c.First(); k != nil; k, v = c.Next() { + op := &v1.Operation{} + if err := proto.Unmarshal(v, op); err != nil { + zap.L().Error("error unmarshalling operation, there may be corruption in the oplog", zap.Error(err)) + continue + } + if op.Status == v1.OperationStatus_STATUS_INPROGRESS { + op.Status = v1.OperationStatus_STATUS_ERROR + op.DisplayMessage = "Operation timeout." + bytes, err := proto.Marshal(op) + if err != nil { + return fmt.Errorf("marshalling operation: %w", err) + } + if err := opLogBucket.Put(k, bytes); err != nil { + return fmt.Errorf("putting operation into bucket: %w", err) + } + } + } + if lastValidated, _ := c.Last(); lastValidated != nil { + if err := sysBucket.Put([]byte("last_validated"), lastValidated); err != nil { + return fmt.Errorf("checkpointing last_validated key: %w", err) + } + } + } return nil }); err != nil { @@ -88,12 +121,11 @@ func (o *OpLog) Add(op *v1.Operation) error { if err == nil { o.notifyHelper(EventTypeOpCreated, op) } - return err } -func (o *OpLog) BulkAdd(ops []*v1.Operation) { - o.db.Update(func(tx *bolt.Tx) error { +func (o *OpLog) BulkAdd(ops []*v1.Operation) error { + err := o.db.Update(func(tx *bolt.Tx) error { for _, op := range ops { if err := o.addOperationHelper(tx, op); err != nil { return err @@ -101,6 +133,12 @@ func (o *OpLog) BulkAdd(ops []*v1.Operation) { } return nil }) + if err == nil { + for _, op := range ops { + o.notifyHelper(EventTypeOpCreated, op) + } + } + return err } func (o *OpLog) addOperationHelper(tx *bolt.Tx, op *v1.Operation) error { @@ -239,6 +277,7 @@ func (o *OpLog) GetByRepo(repoId string, filter Filter) ([]*v1.Operation, error) var ops []*v1.Operation if err := o.db.View(func(tx *bolt.Tx) error { ids := indexutil.IndexSearchByteValue(tx.Bucket(RepoIndexBucket), []byte(repoId)).ToSlice() + ids = filter(ids) b := tx.Bucket(OpLogBucket) for _, id := range ids { @@ -260,6 +299,7 @@ func (o *OpLog) GetByPlan(planId string, filter Filter) ([]*v1.Operation, error) var ops []*v1.Operation if err := o.db.View(func(tx *bolt.Tx) error { ids := indexutil.IndexSearchByteValue(tx.Bucket(PlanIndexBucket), []byte(planId)).ToSlice() + ids = filter(ids) b := tx.Bucket(OpLogBucket) for _, id := range ids { diff --git a/internal/orchestrator/tasks.go b/internal/orchestrator/tasks.go index 2a606e69..7b86d196 100644 --- a/internal/orchestrator/tasks.go +++ b/internal/orchestrator/tasks.go @@ -93,7 +93,7 @@ func backupHelper(ctx context.Context, orchestrator *Orchestrator, plan *v1.Plan op := &v1.Operation{ PlanId: plan.Id, RepoId: plan.Repo, - UnixTimeStartMs: time.Now().Unix(), + UnixTimeStartMs: curTimeMillis(), Status: v1.OperationStatus_STATUS_INPROGRESS, Op: backupOp, } @@ -105,16 +105,28 @@ func backupHelper(ctx context.Context, orchestrator *Orchestrator, plan *v1.Plan return fmt.Errorf("failed to get repo %q: %w", plan.Repo, err) } - if _, err := repo.Backup(ctx, plan, func(entry *restic.BackupProgressEntry) { + lastSent := time.Now() // debounce progress updates, these can endup being very frequent. + summary, err := repo.Backup(ctx, plan, func(entry *restic.BackupProgressEntry) { + if time.Since(lastSent) < 200 * time.Millisecond { + return + } + lastSent = time.Now() + backupOp.OperationBackup.LastStatus = entry.ToProto() 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)) - }); err != nil { + }) + if err != nil { return fmt.Errorf("failed to backup repo %q: %w", plan.Repo, err) } + backupOp.OperationBackup.LastStatus = summary.ToProto() + if err := orchestrator.oplog.Update(op); err != nil { + return fmt.Errorf("update oplog with summary for backup: %v", err) + } + zap.L().Info("Backup complete", zap.String("plan", plan.Id)) return nil }) @@ -134,7 +146,7 @@ func WithOperation(oplog *oplog.OpLog, op *v1.Operation, do func() error) error op.Status = v1.OperationStatus_STATUS_ERROR op.DisplayMessage = err.Error() } - op.UnixTimeEndMs = time.Now().Unix() + op.UnixTimeEndMs = curTimeMillis() if op.Status == v1.OperationStatus_STATUS_INPROGRESS { op.Status = v1.OperationStatus_STATUS_SUCCESS } @@ -142,4 +154,9 @@ func WithOperation(oplog *oplog.OpLog, op *v1.Operation, do func() error) error return multierror.Append(err, fmt.Errorf("failed to update operation in oplog: %w", e)) } return err +} + +func curTimeMillis() int64 { + t := time.Now() + return t.Unix() * 1000 + int64(t.Nanosecond() / 1000000) } \ No newline at end of file diff --git a/webui/src/components/OperationList.tsx b/webui/src/components/OperationList.tsx new file mode 100644 index 00000000..9c175754 --- /dev/null +++ b/webui/src/components/OperationList.tsx @@ -0,0 +1,233 @@ +import React from "react"; +import { Operation, OperationStatus } from "../../gen/ts/v1/operations.pb"; +import { Col, Collapse, Empty, List, Progress, Row, Typography } from "antd"; +import { AlertOutlined, DatabaseOutlined } from "@ant-design/icons"; +import { BackupProgressEntry } from "../../gen/ts/v1/restic.pb"; + +export const OperationList = ({ + operations, +}: React.PropsWithoutRef<{ operations: Operation[] }>) => { + interface OpWrapper { + startTimeMs: number; + operation: Operation; + } + const ops = operations.map((operation) => { + return { + time: parseInt(operation.unixTimeStartMs!), + operation, + }; + }); + + ops.sort((a, b) => b.time - a.time); + + const elems = ops.map(({ operation }) => ( + + )); + + if (ops.length === 0) { + return ( + + ); + } + + return ( + ( + + )} + /> + ); +}; + +export const OperationRow = ({ + operation, +}: React.PropsWithoutRef<{ operation: Operation }>) => { + let contents: React.ReactNode; + + let color = "grey"; + if (operation.status === OperationStatus.STATUS_SUCCESS) { + color = "green"; + } else if (operation.status === OperationStatus.STATUS_ERROR) { + color = "red"; + } else if (operation.status === OperationStatus.STATUS_INPROGRESS) { + color = "blue"; + } + + if (operation.operationBackup) { + const backupOp = operation.operationBackup; + let desc = `Backup at ${formatTime(operation.unixTimeStartMs!)}`; + if (operation.status !== OperationStatus.STATUS_INPROGRESS) { + desc += ` and finished at ${formatTime(operation.unixTimeEndMs!)}`; + } else { + desc += " and is still running."; + } + + return ( + + } + description={ + <> + + ), + }, + ]} + /> + + } + /> + + ); + } else if (operation.operationIndexSnapshot) { + const snapshotOp = operation.operationIndexSnapshot; + return ( + + Snapshot at {formatTime(snapshotOp.snapshot!.unixTimeMs!)} + } + avatar={} + description={<>A snapshot. More info needed} + /> + + ); + } else if (operation.displayMessage) { + return ( + + Message} + avatar={} + description={operation.displayMessage} + /> + + ); + } +}; + +const formatTime = (time: number | string) => { + if (typeof time === "string") { + time = parseInt(time); + } + const d = new Date(); + d.setTime(time); + return d.toLocaleString(); +}; + +const BackupOperationStatus = ({ + status, +}: { + status?: BackupProgressEntry; +}) => { + if (!status) { + return <>No status yet.; + } + + if (status.status) { + const st = status.status; + const progress = + Math.round( + (parseInt(st.bytesDone!) / Math.max(parseInt(st.totalBytes!), 1)) * 1000 + ) / 10; + return ( + <> + +
+ + + Bytes Done/Total +
+ {formatBytes(st.bytesDone)}/{formatBytes(st.totalBytes)} + + + Files Done/Total +
+ {st.filesDone}/{st.totalFiles} + + + + ); + } else if (status.summary) { + const sum = status.summary; + return ( + <> + + Snapshot ID: + {sum.snapshotId} + + + + Files Added +
+ {sum.filesNew} + + + Files Changed +
+ {sum.filesChanged} + + + Files Unmodified +
+ {sum.filesChanged} + + + + + Bytes Added +
+ {formatBytes(sum.dataAdded)} + + + Total Bytes Processed +
+ {formatBytes(sum.totalBytesProcessed)} + + + Total Files Processed +
+ {sum.totalFilesProcessed} + + + + ); + } else { + console.error("GOT UNEXPECTED STATUS: ", status); + return <>No fields set. This shouldn't happen; + } +}; + +const formatBytes = (bytes?: number | string) => { + if (!bytes) { + return 0; + } + if (typeof bytes === "string") { + bytes = parseInt(bytes); + } + + const units = ["B", "KB", "MB", "GB", "TB", "PB"]; + let unit = 0; + while (bytes > 1024) { + bytes /= 1024; + unit++; + } + return `${Math.round(bytes * 100) / 100} ${units[unit]}`; +}; diff --git a/webui/src/state/oplog.ts b/webui/src/state/oplog.ts index 1aa612e3..116168a0 100644 --- a/webui/src/state/oplog.ts +++ b/webui/src/state/oplog.ts @@ -5,20 +5,140 @@ import { OperationEventType, OperationStatus, } from "../../gen/ts/v1/operations.pb"; -import { ResticUI } from "../../gen/ts/v1/service.pb"; +import { GetOperationsRequest, ResticUI } from "../../gen/ts/v1/service.pb"; import { EventEmitter } from "events"; +import { useAlertApi } from "../components/Alerts"; -export const operationEmitter = new EventEmitter(); +const subscribers: ((event: OperationEvent) => void)[] = []; // Start fetching and emitting operations. (async () => { - await ResticUI.GetOperationEvents( - {}, - (event: OperationEvent) => { - operationEmitter.emit("operation", event); + while (true) { + let nextConnWaitUntil = new Date().getTime() + 5000; + try { + await ResticUI.GetOperationEvents( + {}, + (event: OperationEvent) => { + console.log("operation event", event); + subscribers.forEach((subscriber) => subscriber(event)); + }, + { + pathPrefix: "/api", + } + ); + } catch (e: any) { + console.error("operations stream died with exception: ", e); + } + await new Promise((accept, _) => + setTimeout(accept, nextConnWaitUntil - new Date().getTime()) + ); + } +})(); + +export const getOperations = async ({ + planId, + repoId, + lastN, +}: GetOperationsRequest): Promise => { + const opList = await ResticUI.GetOperations( + { + planId, + repoId, + lastN, }, { pathPrefix: "/api", } ); -})(); + return opList.operations || []; +}; + +export const subscribeToOperations = ( + callback: (event: OperationEvent) => void +) => { + subscribers.push(callback); +}; + +export const unsubscribeFromOperations = ( + callback: (event: OperationEvent) => void +) => { + const index = subscribers.indexOf(callback); + if (index > -1) { + subscribers[index] = subscribers[subscribers.length - 1]; + subscribers.pop(); + } +}; + +export const buildOperationListListener = ( + req: GetOperationsRequest, + callback: (event: OperationEvent | null, list: Operation[]) => void +) => { + let operations: Operation[] = []; + + (async () => { + const opsFromServer = await getOperations(req); + operations = opsFromServer.filter( + (o) => !operations.find((op) => op.id === o.id) + ); + operations.sort((a, b) => { + return parseInt(a.id!) - parseInt(b.id!); + }); + + callback(null, operations); + })(); + + return (event: OperationEvent) => { + const op = event.operation!; + const type = event.type!; + if (!!req.planId && op.planId !== req.planId) { + return; + } + if (!!req.repoId && op.repoId !== req.repoId) { + return; + } + if (type === OperationEventType.EVENT_UPDATED) { + const index = operations.findIndex((o) => o.id === op.id); + if (index > -1) { + operations[index] = op; + } else { + operations.push(op); + operations.sort((a, b) => { + return parseInt(a.id!) - parseInt(b.id!); + }); + } + } else if (type === OperationEventType.EVENT_CREATED) { + operations.push(op); + } + + callback(event, operations); + }; +}; + +// OperationsStateTracker tracks the state of operations starting with an initial query +export class OperationListSubscriber { + private listener: ((event: OperationEvent) => void) | null = null; + private operations: Operation[] = []; + private eventEmitter = new EventEmitter(); + constructor(private req: GetOperationsRequest) { + this.listener = (event: OperationEvent) => { + this.eventEmitter.emit("changed"); + }; + subscribeToOperations(this.listener); + getOperations(req).then((ops) => { + this.operations = ops; + this.eventEmitter.emit("changed"); + }); + } + + getOperations() { + return this.operations; + } + + onChange(callback: () => void) { + this.eventEmitter.on("changed", callback); + } + + destroy() { + unsubscribeFromOperations(this.listener!); + } +} \ No newline at end of file diff --git a/webui/src/views/AddPlanModel.tsx b/webui/src/views/AddPlanModel.tsx index bccdaa15..8070fbf4 100644 --- a/webui/src/views/AddPlanModel.tsx +++ b/webui/src/views/AddPlanModel.tsx @@ -65,7 +65,7 @@ export const AddPlanModal = ({ showModal(null); alertsApi.success( - "Plan deleted from config, but not from restic repo. Snapshots will remain in storage until manually deleted." + "Plan deleted from config, but not from restic repo. Snapshots will remain in storage and operations will be tracked until manually deleted. Reusing a deleted plan ID is not recommended if backups have already been performed." ); } catch (e: any) { alertsApi.error("Operation failed: " + e.message, 15); @@ -115,13 +115,14 @@ export const AddPlanModal = ({ Cancel , template != null ? ( , template != null ? ( - - - ); -}; - -const OperationsPanel = ({ operations }: { operations: Operation[] }) => { - return ( - <>

Operations List

- {operations.map((op) => { - return ( -
-

{op.id}

-
- ); - })} + ); };