lee/backrest
1
0
Fork 0
mirror of https://github.com/garethgeorge/backrest.git synced 2026-09-20 23:15:44 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

feat: display queued operations

Browse Source
This commit is contained in:
garethgeorge
2023-11-28 21:33:08 -08:00
parent 825ed12a09
commit 3d520f0848
15 changed files with 2897 additions and 871 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/resticui/resticui.go
+4 -1
View File
@@ -61,7 +61,10 @@ func main() {
zap.S().Fatalf("Error finding or installing restic: %v", err)
}
orchestrator := orchestrator.NewOrchestrator(resticPath, cfg, oplog)
orchestrator, err := orchestrator.NewOrchestrator(resticPath, cfg, oplog)
if err != nil {
zap.S().Fatalf("Error creating orchestrator: %v", err)
}
// Start orchestration loop. Only exits when ctx is cancelled.
go orchestrator.Run(ctx)
gen/go/v1/operations.pb.go
+25 -20
View File
@@ -73,12 +73,13 @@ func (OperationEventType) EnumDescriptor() ([]byte, []int) {
type OperationStatus int32
const (
OperationStatus_STATUS_UNKNOWN OperationStatus = 0
OperationStatus_STATUS_PENDING OperationStatus = 1
OperationStatus_STATUS_INPROGRESS OperationStatus = 2
OperationStatus_STATUS_SUCCESS OperationStatus = 3
OperationStatus_STATUS_ERROR OperationStatus = 4
OperationStatus_STATUS_CANCELLED OperationStatus = 5
OperationStatus_STATUS_UNKNOWN OperationStatus = 0 // used to indicate that the status is unknown.
OperationStatus_STATUS_PENDING OperationStatus = 1 // used to indicate that the operation is pending.
OperationStatus_STATUS_INPROGRESS OperationStatus = 2 // used to indicate that the operation is in progress.
OperationStatus_STATUS_SUCCESS OperationStatus = 3 // used to indicate that the operation completed successfully.
OperationStatus_STATUS_ERROR OperationStatus = 4 // used to indicate that the operation failed.
OperationStatus_STATUS_SYSTEM_CANCELLED OperationStatus = 5 // indicates operation cancelled by the system.
OperationStatus_STATUS_USER_CANCELLED OperationStatus = 6 // indicates operation cancelled by the user.
)
// Enum value maps for OperationStatus.
@@ -89,15 +90,17 @@ var (
2: "STATUS_INPROGRESS",
3: "STATUS_SUCCESS",
4: "STATUS_ERROR",
5: "STATUS_CANCELLED",
5: "STATUS_SYSTEM_CANCELLED",
6: "STATUS_USER_CANCELLED",
}
OperationStatus_value = map[string]int32{
"STATUS_UNKNOWN": 0,
"STATUS_PENDING": 1,
"STATUS_INPROGRESS": 2,
"STATUS_SUCCESS": 3,
"STATUS_ERROR": 4,
"STATUS_CANCELLED": 5,
"STATUS_UNKNOWN": 0,
"STATUS_PENDING": 1,
"STATUS_INPROGRESS": 2,
"STATUS_SUCCESS": 3,
"STATUS_ERROR": 4,
"STATUS_SYSTEM_CANCELLED": 5,
"STATUS_USER_CANCELLED": 6,
}
)
@@ -608,19 +611,21 @@ var file_v1_operations_proto_rawDesc = []byte{
0x4e, 0x54, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x11, 0x0a, 0x0d,
0x45, 0x56, 0x45, 0x4e, 0x54, 0x5f, 0x43, 0x52, 0x45, 0x41, 0x54, 0x45, 0x44, 0x10, 0x01, 0x12,
0x11, 0x0a, 0x0d, 0x45, 0x56, 0x45, 0x4e, 0x54, 0x5f, 0x55, 0x50, 0x44, 0x41, 0x54, 0x45, 0x44,
0x10, 0x02, 0x2a, 0x8c, 0x01, 0x0a, 0x0f, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e,
0x10, 0x02, 0x2a, 0xae, 0x01, 0x0a, 0x0f, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e,
0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53,
0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x54,
0x41, 0x54, 0x55, 0x53, 0x5f, 0x50, 0x45, 0x4e, 0x44, 0x49, 0x4e, 0x47, 0x10, 0x01, 0x12, 0x15,
0x0a, 0x11, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x49, 0x4e, 0x50, 0x52, 0x4f, 0x47, 0x52,
0x45, 0x53, 0x53, 0x10, 0x02, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f,
0x53, 0x55, 0x43, 0x43, 0x45, 0x53, 0x53, 0x10, 0x03, 0x12, 0x10, 0x0a, 0x0c, 0x53, 0x54, 0x41,
0x54, 0x55, 0x53, 0x5f, 0x45, 0x52, 0x52, 0x4f, 0x52, 0x10, 0x04, 0x12, 0x14, 0x0a, 0x10, 0x53,
0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x43, 0x41, 0x4e, 0x43, 0x45, 0x4c, 0x4c, 0x45, 0x44, 0x10,
0x05, 0x42, 0x2e, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f,
0x67, 0x61, 0x72, 0x65, 0x74, 0x68, 0x67, 0x65, 0x6f, 0x72, 0x67, 0x65, 0x2f, 0x72, 0x65, 0x73,
0x74, 0x69, 0x63, 0x75, 0x69, 0x2f, 0x67, 0x6f, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x76,
0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
0x54, 0x55, 0x53, 0x5f, 0x45, 0x52, 0x52, 0x4f, 0x52, 0x10, 0x04, 0x12, 0x1b, 0x0a, 0x17, 0x53,
0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x53, 0x59, 0x53, 0x54, 0x45, 0x4d, 0x5f, 0x43, 0x41, 0x4e,
0x43, 0x45, 0x4c, 0x4c, 0x45, 0x44, 0x10, 0x05, 0x12, 0x19, 0x0a, 0x15, 0x53, 0x54, 0x41, 0x54,
0x55, 0x53, 0x5f, 0x55, 0x53, 0x45, 0x52, 0x5f, 0x43, 0x41, 0x4e, 0x43, 0x45, 0x4c, 0x4c, 0x45,
0x44, 0x10, 0x06, 0x42, 0x2e, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f,
0x6d, 0x2f, 0x67, 0x61, 0x72, 0x65, 0x74, 0x68, 0x67, 0x65, 0x6f, 0x72, 0x67, 0x65, 0x2f, 0x72,
0x65, 0x73, 0x74, 0x69, 0x63, 0x75, 0x69, 0x2f, 0x67, 0x6f, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f,
0x2f, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
internal/oplog/oplog.go
+7 -3
View File
@@ -92,9 +92,13 @@ func NewOpLog(databasePath string) (*OpLog, error) {
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."
if op.Status == v1.OperationStatus_STATUS_PENDING || op.Status == v1.OperationStatus_STATUS_SYSTEM_CANCELLED {
// remove pending operations.
o.deleteOperationHelper(tx, op.Id)
continue
} else if op.Status == v1.OperationStatus_STATUS_INPROGRESS {
o.deleteOperationHelper(tx, op.Id)
}
if err := o.addOperationHelper(tx, op); err != nil {
internal/orchestrator/orchestrator.go
+16 -6
View File
@@ -31,11 +31,10 @@ type Orchestrator struct {
now func() time.Time
}
func NewOrchestrator(resticBin string, cfg *v1.Config, oplog *oplog.OpLog) *Orchestrator {
func NewOrchestrator(resticBin string, cfg *v1.Config, oplog *oplog.OpLog) (*Orchestrator, error) {
var o *Orchestrator
o = &Orchestrator{
config: cfg,
OpLog: oplog,
OpLog: oplog,
// repoPool created with a memory store to ensure the config is updated in an atomic operation with the repo pool's config value.
repoPool: newResticRepoPool(resticBin, &config.MemoryStore{Config: cfg}),
taskQueue: taskQueue{
@@ -47,7 +46,10 @@ func NewOrchestrator(resticBin string, cfg *v1.Config, oplog *oplog.OpLog) *Orch
},
},
}
return o
if err := o.ApplyConfig(cfg); err != nil {
return nil, fmt.Errorf("apply initial config: %w", err)
}
return o, nil
}
func (o *Orchestrator) ApplyConfig(cfg *v1.Config) error {
@@ -55,14 +57,22 @@ func (o *Orchestrator) ApplyConfig(cfg *v1.Config) error {
defer o.mu.Unlock()
o.config = cfg
zap.L().Debug("Applying config to orchestrator", zap.Any("config", cfg))
zap.L().Info("Applying config to orchestrator", zap.Any("config", cfg))
// Update the config provided to the repo pool.
if err := o.repoPool.configProvider.Update(cfg); err != nil {
return fmt.Errorf("failed to update repo pool config: %w", err)
}
o.taskQueue.Reset() // reset queued tasks, this may loose any ephemeral operations scheduled by RPC. Tasks in progress are not cancelled.
// reset queued tasks, this may loose any ephemeral operations scheduled by RPC. Tasks in progress are not cancelled.
removedTasks := o.taskQueue.Reset()
for _, t := range removedTasks {
if err := t.task.Cancel(v1.OperationStatus_STATUS_SYSTEM_CANCELLED); err != nil {
zap.L().Error("failed to cancel queued task", zap.String("task", t.task.Name()), zap.Error(err))
} else {
zap.L().Debug("queued task cancelled due to config change", zap.String("task", t.task.Name()))
}
}
// Requeue tasks that are affected by the config change.
for _, plan := range cfg.Plans {
internal/orchestrator/scheduledtaskheap.go
+5 -5
View File
@@ -37,14 +37,16 @@ func (t *taskQueue) Push(task scheduledTask) {
}
}
func (t *taskQueue) Reset() {
func (t *taskQueue) Reset() []*scheduledTask {
t.mu.Lock()
defer t.mu.Unlock()
oldTasks := t.heap.tasks
t.heap.tasks = nil
if t.notify != nil {
t.notify <- struct{}{}
}
return oldTasks
}
func (t *taskQueue) Dequeue(ctx context.Context) *scheduledTask {
@@ -71,10 +73,9 @@ func (t *taskQueue) Dequeue(ctx context.Context) *scheduledTask {
}
t.mu.Unlock()
timer := time.NewTimer(first.runAt.Sub(t.curTime()))
t.mu.Lock()
select {
case <-timer.C:
t.mu.Lock()
if t.heap.Len() == 0 {
break
}
@@ -83,11 +84,11 @@ func (t *taskQueue) Dequeue(ctx context.Context) *scheduledTask {
// task is not yet ready to run
break
}
heap.Pop(&t.heap) // remove the task from the heap
t.mu.Unlock()
return first
case <-t.notify: // new task was added, loop again to ensure we have the earliest task.
t.mu.Lock()
if !timer.Stop() {
<-timer.C
}
@@ -95,7 +96,6 @@ func (t *taskQueue) Dequeue(ctx context.Context) *scheduledTask {
if !timer.Stop() {
<-timer.C
}
t.mu.Unlock()
return nil
}
}
internal/orchestrator/tasks.go
+90 -56
View File
@@ -16,90 +16,115 @@ import (
)
type Task interface {
Name() string // huamn readable name for this task.
Next(now time.Time) *time.Time // when this task would like to be run.
Run(ctx context.Context) error // run the task.
Name() string // huamn readable name for this task.
Next(now time.Time) *time.Time // when this task would like to be run.
Run(ctx context.Context) error // run the task.
Cancel(withStatus v1.OperationStatus) error // cancel the task's execution with the given status (either STATUS_USER_CANCELLED or STATUS_SYSTEM_CANCELLED).
}
// BackupTask is a scheduled backup operation.
type ScheduledBackupTask struct {
type BackupTask struct {
name string
orchestrator *Orchestrator // owning orchestrator
plan *v1.Plan
schedule *cronexpr.Schedule
op *v1.Operation
scheduler func(curTime time.Time) *time.Time
cancel context.CancelFunc // nil unless operation is running.
}
var _ Task = &ScheduledBackupTask{}
var _ Task = &BackupTask{}
func NewScheduledBackupTask(orchestrator *Orchestrator, plan *v1.Plan) (*ScheduledBackupTask, error) {
func NewScheduledBackupTask(orchestrator *Orchestrator, plan *v1.Plan) (*BackupTask, error) {
sched, err := cronexpr.ParseInLocation(plan.Cron, time.Now().Location().String())
if err != nil {
return nil, fmt.Errorf("failed to parse schedule %q: %w", plan.Cron, err)
}
return &ScheduledBackupTask{
return &BackupTask{
name: fmt.Sprintf("backup for plan %q", plan.Id),
orchestrator: orchestrator,
plan: plan,
schedule: sched,
scheduler: func(curTime time.Time) *time.Time {
next := sched.Next(curTime)
return &next
},
}, nil
}
func (t *ScheduledBackupTask) Name() string {
return fmt.Sprintf("backup for plan %q", t.plan.Id)
}
func (t *ScheduledBackupTask) Next(now time.Time) *time.Time {
next := t.schedule.Next(now)
return &next
}
func (t *ScheduledBackupTask) Run(ctx context.Context) error {
return backupHelper(ctx, t.orchestrator, t.plan)
}
// OnetimeBackupTask is a single backup operation.
type OnetimeBackupTask struct {
orchestrator *Orchestrator
plan *v1.Plan
time *time.Time
}
func NewOneofBackupTask(orchestrator *Orchestrator, plan *v1.Plan, at time.Time) *OnetimeBackupTask {
return &OnetimeBackupTask{
func NewOneofBackupTask(orchestrator *Orchestrator, plan *v1.Plan, at time.Time) *BackupTask {
didOnce := false
return &BackupTask{
name: fmt.Sprintf("onetime backup for plan %q", plan.Id),
orchestrator: orchestrator,
plan: plan,
time: &at,
scheduler: func(curTime time.Time) *time.Time {
if didOnce {
return nil
}
didOnce = true
return &at
},
}
}
func (t *OnetimeBackupTask) Name() string {
return fmt.Sprintf("onetime backup for plan %q", t.plan.Id)
func (t *BackupTask) Name() string {
return t.name
}
func (t *OnetimeBackupTask) Next(now time.Time) *time.Time {
ret := t.time
t.time = nil
return ret
func (t *BackupTask) Next(now time.Time) *time.Time {
next := t.scheduler(now)
if next == nil {
return nil
}
t.op = &v1.Operation{
PlanId: t.plan.Id,
RepoId: t.plan.Repo,
UnixTimeStartMs: timeToUnixMillis(*next),
Status: v1.OperationStatus_STATUS_PENDING,
Op: &v1.Operation_OperationBackup{},
}
if err := t.orchestrator.OpLog.Add(t.op); err != nil {
zap.S().Errorf("task %v failed to add operation to oplog: %v", t.Name(), err)
return nil
}
return next
}
func (t *OnetimeBackupTask) Run(ctx context.Context) error {
return backupHelper(ctx, t.orchestrator, t.plan)
func (t *BackupTask) Run(ctx context.Context) error {
ctx, cancel := context.WithCancel(ctx)
t.cancel = cancel
err := backupHelper(ctx, t.orchestrator, t.plan, t.op)
t.op = nil
t.cancel = nil
return err
}
func (t *BackupTask) Cancel(status v1.OperationStatus) error {
if t.op == nil {
return nil
}
if t.cancel != nil && status == v1.OperationStatus_STATUS_USER_CANCELLED {
t.cancel() // try to interrupt the running operation.
}
t.op.Status = status
t.op.UnixTimeEndMs = curTimeMillis()
return t.orchestrator.OpLog.Update(t.op)
}
// backupHelper does a backup.
func backupHelper(ctx context.Context, orchestrator *Orchestrator, plan *v1.Plan) error {
func backupHelper(ctx context.Context, orchestrator *Orchestrator, plan *v1.Plan, op *v1.Operation) error {
backupOp := &v1.Operation_OperationBackup{
OperationBackup: &v1.OperationBackup{},
}
op := &v1.Operation{
PlanId: plan.Id,
RepoId: plan.Repo,
UnixTimeStartMs: curTimeMillis(),
Status: v1.OperationStatus_STATUS_INPROGRESS,
Op: backupOp,
}
startTime := time.Now()
op.Op = backupOp
op.UnixTimeStartMs = curTimeMillis()
err := WithOperation(orchestrator.OpLog, op, func() error {
zap.L().Info("Starting backup", zap.String("plan", plan.Id), zap.Int64("opId", op.Id))
@@ -119,7 +144,6 @@ 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))
})
if err != nil {
return fmt.Errorf("repo.Backup for repo %q: %w", plan.Repo, err)
@@ -131,7 +155,7 @@ func backupHelper(ctx context.Context, orchestrator *Orchestrator, plan *v1.Plan
return fmt.Errorf("expected a final backup progress entry, got nil")
}
zap.L().Info("backup complete", zap.String("plan", plan.Id), zap.Duration("duration", time.Since(startTime)))
zap.L().Info("Backup complete", zap.String("plan", plan.Id), zap.Duration("duration", time.Since(startTime)), zap.Any("summary", summary))
return nil
})
if err != nil {
@@ -204,10 +228,17 @@ func indexSnapshotsHelper(ctx context.Context, orchestrator *Orchestrator, plan
// WithOperation is a utility that creates an operation to track the function's execution.
// timestamps are automatically added and the status is automatically updated if an error occurs.
func WithOperation(oplog *oplog.OpLog, op *v1.Operation, do func() error) error {
if err := oplog.Add(op); err != nil {
return fmt.Errorf("failed to add operation to oplog: %w", err)
if op.Id != 0 {
if err := oplog.Update(op); err != nil {
return fmt.Errorf("failed to add operation to oplog: %w", err)
}
} else {
if err := oplog.Add(op); err != nil {
return fmt.Errorf("failed to add operation to oplog: %w", err)
}
}
if op.Status == v1.OperationStatus_STATUS_UNKNOWN {
if op.Status == v1.OperationStatus_STATUS_PENDING || op.Status == v1.OperationStatus_STATUS_UNKNOWN {
op.Status = v1.OperationStatus_STATUS_INPROGRESS
}
err := do()
@@ -225,11 +256,14 @@ func WithOperation(oplog *oplog.OpLog, op *v1.Operation, do func() error) error
return err
}
func curTimeMillis() int64 {
t := time.Now()
func timeToUnixMillis(t time.Time) int64 {
return t.Unix()*1000 + int64(t.Nanosecond()/1000000)
}
func curTimeMillis() int64 {
return timeToUnixMillis(time.Now())
}
func containsSnapshotOperation(ops []*v1.Operation) bool {
for _, op := range ops {
if _, ok := op.Op.(*v1.Operation_OperationIndexSnapshot); ok {
internal/resticinstaller/resticinstaller.go
+3
View File
@@ -13,6 +13,7 @@ import (
"sync"
"github.com/garethgeorge/resticui/internal/config"
"go.uber.org/zap"
)
var (
@@ -124,9 +125,11 @@ func FindOrInstallResticBinary() (string, error) {
}
didTryInstall = true
zap.S().Infof("Installing restic %v...", RequiredResticVersion)
if err := installResticIfNotExists(resticInstallPath); err != nil {
return "", fmt.Errorf("install restic: %w", err)
}
zap.S().Infof("Installed restic %v", RequiredResticVersion)
}
return resticInstallPath, nil
pkg/restic/limitwriter.go
+34
View File
@@ -0,0 +1,34 @@
package restic
import "io"
// limitWriter silently stops writing after 'limit' bytes.
type limitWriter struct {
written int64
limit int64
w io.Writer
}
var _ io.Writer = &limitWriter{}
func (w *limitWriter) Write(p []byte) (n int, err error) {
r := len(p)
if w.written >= w.limit {
return r, nil
}
if w.written+int64(len(p)) > w.limit {
p = p[:w.limit-w.written]
}
n, err = w.w.Write(p)
w.written += int64(n)
return r, err
}
func newLimitWriter(w io.Writer, limit int64) io.Writer {
return &limitWriter{
w: w,
limit: limit,
}
}
pkg/restic/restic.go
+2 -3
View File
@@ -98,11 +98,10 @@ func (r *Repo) Backup(ctx context.Context, progressCallback func(*BackupProgress
args = append(args, r.extraArgs...)
args = append(args, opt.paths...)
args = append(args, opt.extraArgs...)
output := bytes.NewBuffer(nil)
reader, writer := io.Pipe()
capture := io.MultiWriter(output, writer)
capture := io.MultiWriter(newLimitWriter(output, 1000), writer)
cmd := exec.CommandContext(ctx, r.cmd, args...)
cmd.Env = append(cmd.Env, r.buildEnv()...)
proto/v1/operations.proto
+7 -6
View File
@@ -47,12 +47,13 @@ enum OperationEventType {
}
enum OperationStatus {
STATUS_UNKNOWN = 0;
STATUS_PENDING = 1;
STATUS_INPROGRESS = 2;
STATUS_SUCCESS = 3;
STATUS_ERROR = 4;
STATUS_CANCELLED = 5;
STATUS_UNKNOWN = 0; // used to indicate that the status is unknown.
STATUS_PENDING = 1; // used to indicate that the operation is pending.
STATUS_INPROGRESS = 2; // used to indicate that the operation is in progress.
STATUS_SUCCESS = 3; // used to indicate that the operation completed successfully.
STATUS_ERROR = 4; // used to indicate that the operation failed.
STATUS_SYSTEM_CANCELLED = 5; // indicates operation cancelled by the system.
STATUS_USER_CANCELLED = 6; // indicates operation cancelled by the user.
}
message OperationBackup {
webui/gen/ts/v1/operations.pb.ts
+2 -1
View File
@@ -27,7 +27,8 @@ export enum OperationStatus {
STATUS_INPROGRESS = "STATUS_INPROGRESS",
STATUS_SUCCESS = "STATUS_SUCCESS",
STATUS_ERROR = "STATUS_ERROR",
STATUS_CANCELLED = "STATUS_CANCELLED",
STATUS_SYSTEM_CANCELLED = "STATUS_SYSTEM_CANCELLED",
STATUS_USER_CANCELLED = "STATUS_USER_CANCELLED",
}
export type OperationList = {
webui/package-lock.json
Generated
+2677 -758
View File
File diff suppressed because it is too large Load Diff
webui/src/components/OperationList.tsx
+2 -2
View File
@@ -139,12 +139,12 @@ export const OperationRow = ({
} else if (operation.operationBackup) {
const backupOp = operation.operationBackup;
let desc = `${formatTime(operation.unixTimeStartMs!)} - Backup`;
if (operation.status !== OperationStatus.STATUS_INPROGRESS) {
if (operation.status == OperationStatus.STATUS_SUCCESS) {
desc += ` completed in ${formatDuration(
parseInt(operation.unixTimeEndMs!) -
parseInt(operation.unixTimeStartMs!)
)}`;
} else {
} else if (operation.status === OperationStatus.STATUS_INPROGRESS) {
desc += " and is still running.";
}
webui/src/components/OperationTree.tsx
+16 -4
View File
@@ -4,6 +4,7 @@ import {
BackupInfoCollector,
EOperation,
getOperations,
shouldHideStatus,
subscribeToOperations,
toEop,
unsubscribeFromOperations,
@@ -56,7 +57,10 @@ export const OperationTree = ({
subscribeToOperations(lis);
backupCollector.subscribe(() => {
const backups = backupCollector.getAll();
let backups = backupCollector.getAll();
backups = backups.filter((b) => {
return !shouldHideStatus(b.status);
});
backups.sort((a, b) => {
return b.startTimeMs - a.startTimeMs;
});
@@ -73,7 +77,7 @@ export const OperationTree = ({
return () => {
unsubscribeFromOperations(lis);
};
}, [req]);
}, [JSON.stringify(req)]);
if (backups.length === 0) {
return (
@@ -128,12 +132,20 @@ export const OperationTree = ({
const b = node.backup;
const details: string[] = [];
if (b.status === OperationStatus.STATUS_PENDING) {
details.push("pending");
} else if (b.status === OperationStatus.STATUS_SYSTEM_CANCELLED) {
details.push("system cancel");
} else if (b.status === OperationStatus.STATUS_USER_CANCELLED) {
details.push("cancelled");
}
if (b.backupLastStatus) {
if (b.backupLastStatus.summary) {
const s = b.backupLastStatus.summary;
details.push(
`${formatBytes(s.totalBytesProcessed)} in ${formatDuration(
s.totalDuration!
s.totalDuration! * 1000.0 // convert to ms
)}`
);
} else if (b.backupLastStatus.status) {
@@ -245,7 +257,7 @@ const buildTreeLeaf = (operations: BackupInfo[]): OpTreeNode[] => {
case OperationStatus.STATUS_INPROGRESS:
iconColor = "blue";
break;
case OperationStatus.STATUS_CANCELLED:
case OperationStatus.STATUS_USER_CANCELLED:
iconColor = "orange";
break;
}
webui/src/state/oplog.ts
+7 -6
View File
@@ -76,14 +76,12 @@ export const buildOperationListListener = (
let operations: EOperation[] = [];
(async () => {
let opsFromServer = await getOperations(req);
operations = opsFromServer.filter(
(o) => !operations.find((op) => op.id === o.id)
);
const opsFromServer = await getOperations(req);
operations.push(...opsFromServer);
operations = _.uniqBy(opsFromServer, (o) => o.id!);
operations.sort((a, b) => {
return a.parsedTime! - b.parsedTime!;
});
callback(null, null, operations);
})();
@@ -115,7 +113,6 @@ export const buildOperationListListener = (
} else if (type === OperationEventType.EVENT_CREATED) {
operations.push(op);
}
callback(event.type || null, op, operations);
};
};
@@ -264,3 +261,7 @@ export class BackupInfoCollector {
}
}
}
export const shouldHideStatus = (status: OperationStatus) => {
return status === OperationStatus.STATUS_SYSTEM_CANCELLED;
};