diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 1467588..902ffd5 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -6,16 +6,12 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/zrepl/zrepl/config" "github.com/zrepl/zrepl/daemon/filters" - "github.com/zrepl/zrepl/daemon/job/reset" "github.com/zrepl/zrepl/daemon/job/wakeup" "github.com/zrepl/zrepl/daemon/logging" "github.com/zrepl/zrepl/daemon/pruner" "github.com/zrepl/zrepl/daemon/snapper" "github.com/zrepl/zrepl/endpoint" - "github.com/zrepl/zrepl/util/envconst" "github.com/zrepl/zrepl/zfs" - "sync" - "time" ) type SnapJob struct { @@ -27,19 +23,10 @@ type SnapJob struct { promPruneSecs *prometheus.HistogramVec // labels: prune_side - tasksMtx sync.Mutex - tasks snap_activeSideTasks -} - - -type snap_activeSideTasks struct { - state ActiveSideState - pruner *pruner.Pruner - - prunerCancel context.CancelFunc } + func (j *SnapJob) Name() string { return j.name } func (j *SnapJob) GetPruner(ctx context.Context, sender *endpoint.Sender) (*pruner.Pruner) { @@ -48,20 +35,6 @@ func (j *SnapJob) GetPruner(ctx context.Context, sender *endpoint.Sender) (*prun } -func (j *SnapJob) updateTasks(u func(*snap_activeSideTasks)) snap_activeSideTasks { - j.tasksMtx.Lock() - defer j.tasksMtx.Unlock() - var copy snap_activeSideTasks - copy = j.tasks - if u == nil { - return copy - } - u(©) - j.tasks = copy - return copy -} - - func (j *SnapJob) Type() Type { return TypeSnap } func (j *SnapJob) RunPeriodic(ctx context.Context, wakeUpCommon chan <- struct{}) { @@ -107,12 +80,12 @@ type SnapJobStatus struct { } func (j *SnapJob) Status() *Status { - tasks := j.updateTasks(nil) + //tasks := j.updateTasks(nil) s := &SnapJobStatus{} t := j.Type() - if tasks.pruner != nil { - s.Pruning = tasks.pruner.Report() + if j.pruner != nil { + s.Pruning = j.pruner.Report() } return &Status{Type: t, JobSpecific: s} } @@ -142,98 +115,17 @@ outer: } invocationCount++ invLog := log.WithField("invocation", invocationCount) - j.do(WithLogger(ctx, invLog)) + j.doPrune(WithLogger(ctx, invLog)) } } -func (j *SnapJob) do(ctx context.Context) { - +func (j *SnapJob) doPrune(ctx context.Context) { log := GetLogger(ctx) ctx = logging.WithSubsystemLoggers(ctx, log) - - // allow cancellation of an invocation (this function) - ctx, cancelThisRun := context.WithCancel(ctx) - defer cancelThisRun() - go func() { - select { - case <-reset.Wait(ctx): - log.Info("reset received, cancelling current invocation") - cancelThisRun() - case <-ctx.Done(): - } - }() - - // The code after this watchdog goroutine is sequential and transitions the state from - // ActiveSideReplicating -> ActiveSidePruneSender -> ActiveSidePruneReceiver -> ActiveSideDone - // If any of those sequential tasks 'gets stuck' (livelock, no progress), the watchdog will eventually - // cancel its context. - // If the task is written to support context cancellation, it will return immediately (in permanent error state), - // and the sequential code above transitions to the next state. - go func() { - - wdto := envconst.Duration("ZREPL_JOB_WATCHDOG_TIMEOUT", 10*time.Minute) - jitter := envconst.Duration("ZREPL_JOB_WATCHDOG_JITTER", 1*time.Second) - // shadowing! - log := log.WithField("watchdog_timeout", wdto.String()) - - log.Debug("starting watchdog") - defer log.Debug("watchdog stopped") - - t := time.NewTicker(wdto) - defer t.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-t.C: // fall - } - - j.updateTasks(func(tasks *snap_activeSideTasks) { - // Since cancelling a task will cause the sequential code to transition to the next state immediately, - // we cannot check for its progress right then (no fallthrough). - // Instead, we return (not continue because we are in a closure) and give the new state another - // ZREPL_JOB_WATCHDOG_TIMEOUT interval to try make some progress. - - log.WithField("state", tasks.state).Debug("watchdog firing") - - const WATCHDOG_ENVCONST_NOTICE = " (adjust ZREPL_JOB_WATCHDOG_TIMEOUT env variable if inappropriate)" - - switch tasks.state { - case ActiveSidePruneSender: - log.WithField("prune_sender_progress", "TEST DEBUG 123"). - Debug("check pruner_sender progress") - if tasks.pruner.Progress.CheckTimeout(wdto, jitter) { - log.Error("pruner_sender did not make progress, cancelling" + WATCHDOG_ENVCONST_NOTICE) - tasks.prunerCancel() - return - } - case ActiveSideDone: - // ignore, ctx will be Done() in a few milliseconds and the watchdog will exit - default: - log.WithField("state", tasks.state). - Error("watchdog implementation error: unknown active side state") - } - }) - - } - }() - - ctx, localCancel := context.WithCancel(ctx) - sender := endpoint.NewSender(j.FSFilter()) - tasks := j.updateTasks(func(tasks *snap_activeSideTasks) { - tasks.pruner = j.GetPruner(ctx, sender) - tasks.prunerCancel = localCancel - tasks.state = ActiveSidePruneSender - }) - - log.Info("start pruning sender") - tasks.pruner.Prune() - log.Info("finished pruning sender") - localCancel() - j.updateTasks(func(tasks *snap_activeSideTasks) { - tasks.state = ActiveSideDone - }) - + sender := endpoint.NewSender(j.FSFilter()) + j.pruner = j.GetPruner(ctx, sender) + log.Info("start pruning") + j.pruner.Prune() + log.Info("finished pruning") }