fix: store large log outputs in tar bundles of logs

internal/orchestrator/task.go

@@ -0,0 +1,120 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync/atomic"
+	"time"
+
+	v1 "github.com/garethgeorge/backrest/gen/go/v1"
+	"github.com/garethgeorge/backrest/internal/oplog"
+	"github.com/hashicorp/go-multierror"
+)
+
+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.
+	Cancel(withStatus v1.OperationStatus) error // informat the task that it's scheduled execution will be skipped (either STATUS_USER_CANCELLED or STATUS_SYSTEM_CANCELLED).
+	OperationId() int64                         // the id of the operation associated with this task (if any).
+}
+
+type TaskWithOperation struct {
+	orch    *Orchestrator
+	op      *v1.Operation
+	running atomic.Bool
+}
+
+func (t *TaskWithOperation) OperationId() int64 {
+	if t.op == nil {
+		return 0
+	}
+	return t.op.Id
+}
+
+func (t *TaskWithOperation) setOperation(op *v1.Operation) error {
+	if t.op != nil {
+		return errors.New("task already has an operation")
+	}
+	if err := t.orch.OpLog.Add(op); err != nil {
+		return fmt.Errorf("task failed to add operation to oplog: %v", err)
+	}
+	t.op = op
+	return nil
+}
+
+func (t *TaskWithOperation) runWithOpAndContext(ctx context.Context, do func(ctx context.Context, op *v1.Operation) error) error {
+	if t.op == nil {
+		return errors.New("task has no operation, a call to setOperation first is required")
+	}
+	if t.running.Load() {
+		return errors.New("task is already running")
+	}
+
+	t.running.Store(true)
+	defer t.running.Store(false)
+	defer func() {
+		t.op = nil
+	}()
+
+	return WithOperation(t.orch.OpLog, t.op, func() error {
+		return do(ctx, t.op)
+	})
+}
+
+// Cancel marks a task as cancelled. Note that, unintuitively, it is actually an error to call cancel on a running task.
+func (t *TaskWithOperation) Cancel(withStatus v1.OperationStatus) error {
+	if t.running.Load() {
+		return errors.New("cannot cancel a running task") // should never happen.
+	}
+	if t.op == nil {
+		return nil
+	}
+	t.op.Status = withStatus
+	t.op.UnixTimeEndMs = curTimeMillis()
+	if err := t.orch.OpLog.Update(t.op); err != nil {
+		return fmt.Errorf("failed to update operation %v in oplog: %w", t.op.Id, err)
+	}
+	t.op = nil
+	return nil
+}
+
+// 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 {
+	op.UnixTimeStartMs = curTimeMillis() // update the start time from the planned time to the actual time.
+	if op.Status == v1.OperationStatus_STATUS_PENDING || op.Status == v1.OperationStatus_STATUS_UNKNOWN {
+		op.Status = v1.OperationStatus_STATUS_INPROGRESS
+	}
+	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)
+		}
+	}
+	err := do()
+	if err != nil {
+		op.Status = v1.OperationStatus_STATUS_ERROR
+		op.DisplayMessage = err.Error()
+	}
+	op.UnixTimeEndMs = curTimeMillis()
+	if op.Status == v1.OperationStatus_STATUS_INPROGRESS {
+		op.Status = v1.OperationStatus_STATUS_SUCCESS
+	}
+	if e := oplog.Update(op); e != nil {
+		return multierror.Append(err, fmt.Errorf("failed to update operation in oplog: %w", e))
+	}
+	return err
+}
+
+func timeToUnixMillis(t time.Time) int64 {
+	return t.Unix()*1000 + int64(t.Nanosecond()/1000000)
+}
+
+func curTimeMillis() int64 {
+	return timeToUnixMillis(time.Now())
+}