diff --git a/daemon/job/build_jobs.go b/daemon/job/build_jobs.go index 5f716a6..c3eafd4 100644 --- a/daemon/job/build_jobs.go +++ b/daemon/job/build_jobs.go @@ -46,7 +46,7 @@ func buildJob(c *config.Global, in config.JobEnum) (j Job, err error) { return cannotBuildJob(err, v.Name) } case *config.SnapJob: - j, err = snapJob(c, v) + j, err = snapJobFromConfig(c, v) if err != nil { return cannotBuildJob(err, v.Name) } diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 2e0034f..138c660 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -30,7 +30,7 @@ func (j *SnapJob) Name() string { return j.name } func (j *SnapJob) Type() Type { return TypeSnap } -func snapJob(g *config.Global, in *config.SnapJob) (j *SnapJob, err error) { +func snapJobFromConfig(g *config.Global, in *config.SnapJob) (j *SnapJob, err error) { j = &SnapJob{} fsf, err := filters.DatasetMapFilterFromConfig(in.Filesystems) if err != nil { diff --git a/daemon/job/snapjob.go_prefmt b/daemon/job/snapjob.go_prefmt new file mode 100644 index 0000000..7ff6a19 --- /dev/null +++ b/daemon/job/snapjob.go_prefmt @@ -0,0 +1,129 @@ +package job + +import ( + "context" + "github.com/pkg/errors" + "github.com/prometheus/client_golang/prometheus" + "github.com/zrepl/zrepl/config" + "github.com/zrepl/zrepl/daemon/filters" + "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/zfs" +) + +type SnapJob struct { + name string + fsfilter zfs.DatasetFilter + snapper *snapper.PeriodicOrManual + + prunerFactory *pruner.SinglePrunerFactory + + promPruneSecs *prometheus.HistogramVec // no labels! + + pruner *pruner.Pruner +} + + +func (j *SnapJob) Name() string { return j.name } + +func (j *SnapJob) getPruner(ctx context.Context, sender *endpoint.Sender) (*pruner.Pruner) { + p := j.prunerFactory.BuildSinglePruner(ctx,sender,sender) + return p +} + + +func (j *SnapJob) Type() Type { return TypeSnap } + +func (j *SnapJob) RunPeriodic(ctx context.Context, wakeUpCommon chan <- struct{}) { + j.snapper.Run(ctx, wakeUpCommon) +} + +func (j *SnapJob) FSFilter() zfs.DatasetFilter { + return j.fsfilter +} + +func snapJob(g *config.Global, in *config.SnapJob) (j *SnapJob, err error) { + j = &SnapJob{} + fsf, err := filters.DatasetMapFilterFromConfig(in.Filesystems) + if err != nil { + return nil, errors.Wrap(err, "cannnot build filesystem filter") + } + j.fsfilter = fsf + + if j.snapper, err = snapper.FromConfig(g, fsf, in.Snapshotting); err != nil { + return nil, errors.Wrap(err, "cannot build snapper") + } + j.name = in.Name + j.promPruneSecs = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "zrepl", + Subsystem: "pruning", + Name: "time", + Help: "seconds spent in pruner", + ConstLabels: prometheus.Labels{"zrepl_job":j.name}, + }, []string{"prune_side"}) + j.prunerFactory, err = pruner.NewSinglePrunerFactory(in.Pruning, j.promPruneSecs) + if err != nil { + return nil, errors.Wrap(err, "cannot build snapjob pruning rules") + } + return j, nil +} + +func (j *SnapJob) RegisterMetrics(registerer prometheus.Registerer) { + registerer.MustRegister(j.promPruneSecs) +} + +type SnapJobStatus struct { + Pruning *pruner.Report +} + +func (j *SnapJob) Status() *Status { + s := &SnapJobStatus{} + t := j.Type() + if j.pruner != nil { + s.Pruning = j.pruner.Report() + } + return &Status{Type: t, JobSpecific: s} +} + +func (j *SnapJob) Run(ctx context.Context) { + log := GetLogger(ctx) + ctx = logging.WithSubsystemLoggers(ctx, log) + + defer log.Info("job exiting") + + periodicDone := make(chan struct{}) + ctx, cancel := context.WithCancel(ctx) + defer cancel() + go j.RunPeriodic(ctx, periodicDone) + + invocationCount := 0 +outer: + for { + log.Info("wait for wakeups") + select { + case <-ctx.Done(): + log.WithError(ctx.Err()).Info("context") + break outer + + case <-wakeup.Wait(ctx): + case <-periodicDone: + } + invocationCount++ + invLog := log.WithField("invocation", invocationCount) + j.doPrune(WithLogger(ctx, invLog)) + } +} + +func (j *SnapJob) doPrune(ctx context.Context) { + log := GetLogger(ctx) + ctx = logging.WithSubsystemLoggers(ctx, log) + sender := endpoint.NewSender(j.FSFilter()) + j.pruner = j.getPruner(ctx, sender) + log.Info("start pruning") + j.pruner.Prune() + log.Info("finished pruning") +} +