From 3d2688e95990d3a65dc51dcc6db19a16daa898b9 Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Tue, 20 Nov 2018 19:30:15 +0100 Subject: [PATCH 01/27] Ugly but working inital snapjob implementation --- config/config.go | 11 ++ config/samples/snap.yml | 14 ++ daemon/job/build_jobs.go | 9 + daemon/job/job.go | 3 + daemon/job/snapjob.go | 379 +++++++++++++++++++++++++++++++++++++++ daemon/pruner/pruner.go | 39 ++++ 6 files changed, 455 insertions(+) create mode 100644 config/samples/snap.yml create mode 100644 daemon/job/snapjob.go diff --git a/config/config.go b/config/config.go index eeea023..2c8c5dc 100644 --- a/config/config.go +++ b/config/config.go @@ -33,6 +33,7 @@ type JobEnum struct { func (j JobEnum) Name() string { var name string switch v := j.Ret.(type) { + case *SnapJob: name = v.Name case *PushJob: name = v.Name case *SinkJob: name = v.Name case *PullJob: name = v.Name @@ -58,6 +59,15 @@ type PassiveJob struct { Debug JobDebugSettings `yaml:"debug,optional"` } +type SnapJob struct { + Type string `yaml:"type"` + Name string `yaml:"name"` + Pruning PruningLocal `yaml:"pruning"` + Debug JobDebugSettings `yaml:"debug,optional"` + Snapshotting SnapshottingEnum `yaml:"snapshotting"` + Filesystems FilesystemsFilter `yaml:"filesystems"` +} + type PushJob struct { ActiveJob `yaml:",inline"` Snapshotting SnapshottingEnum `yaml:"snapshotting"` @@ -340,6 +350,7 @@ func enumUnmarshal(u func(interface{}, bool) error, types map[string]interface{} func (t *JobEnum) UnmarshalYAML(u func(interface{}, bool) error) (err error) { t.Ret, err = enumUnmarshal(u, map[string]interface{}{ + "snap": &SnapJob{}, "push": &PushJob{}, "sink": &SinkJob{}, "pull": &PullJob{}, diff --git a/config/samples/snap.yml b/config/samples/snap.yml new file mode 100644 index 0000000..6775b01 --- /dev/null +++ b/config/samples/snap.yml @@ -0,0 +1,14 @@ +jobs: +- name: snapjob + type: snap + filesystems: { + "tank/frequently_changed<": true, + } + snapshotting: + type: periodic + interval: 2m + prefix: zrepl_snapjob_ + pruning: + keep: + - type: last_n + count: 60 diff --git a/daemon/job/build_jobs.go b/daemon/job/build_jobs.go index a48a183..d8f2371 100644 --- a/daemon/job/build_jobs.go +++ b/daemon/job/build_jobs.go @@ -45,6 +45,15 @@ func buildJob(c *config.Global, in config.JobEnum) (j Job, err error) { if err != nil { return cannotBuildJob(err, v.Name) } + case *config.SnapJob: + m, err := modeSnapFromConfig(c, v) + if err != nil { + return cannotBuildJob(err, v.Name) + } + j, err = snap_activeSide(c, v, m) + if err != nil { + return cannotBuildJob(err, v.Name) + } case *config.PushJob: m, err := modePushFromConfig(c, v) if err != nil { diff --git a/daemon/job/job.go b/daemon/job/job.go index 5b3a684..6888fed 100644 --- a/daemon/job/job.go +++ b/daemon/job/job.go @@ -39,6 +39,7 @@ type Type string const ( TypeInternal Type = "internal" + TypeSnap = "snap" TypePush Type = "push" TypeSink Type = "sink" TypePull Type = "pull" @@ -84,6 +85,8 @@ func (s *Status) UnmarshalJSON(in []byte) (err error) { return fmt.Errorf("field '%s', not found", key) } switch s.Type { + case TypeSnap: + fallthrough case TypePull: fallthrough case TypePush: var st ActiveSideStatus diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go new file mode 100644 index 0000000..8515799 --- /dev/null +++ b/daemon/job/snapjob.go @@ -0,0 +1,379 @@ +package job + +import ( + "context" + "github.com/pkg/errors" +// "github.com/problame/go-streamrpc" + "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/pruning" + "github.com/zrepl/zrepl/daemon/snapper" + "github.com/zrepl/zrepl/daemon/transport/connecter" + "github.com/zrepl/zrepl/endpoint" +// "github.com/zrepl/zrepl/replication" + "github.com/zrepl/zrepl/util/envconst" + "github.com/zrepl/zrepl/zfs" + "sync" + "time" +) + +type snap_ActiveSide struct { + mode snap_activeMode + name string + clientFactory *connecter.ClientFactory + + prunerFactory *pruner.SinglePrunerFactory + + //promRepStateSecs *prometheus.HistogramVec // labels: state + promPruneSecs *prometheus.HistogramVec // labels: prune_side + //promBytesReplicated *prometheus.CounterVec // labels: filesystem + + tasksMtx sync.Mutex + tasks snap_activeSideTasks +} + + +type snap_activeSideTasks struct { + state ActiveSideState + + // valid for state ActiveSideReplicating, ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone + //replication *replication.Replication + //replicationCancel context.CancelFunc + + // valid for state ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone + //prunerSender, prunerReceiver *pruner.Pruner + pruner *pruner.Pruner + + // valid for state ActiveSidePruneReceiver, ActiveSideDone + prunerCancel context.CancelFunc +} + +func (a *snap_ActiveSide) Name() string { return a.name } + +func (a *snap_ActiveSide) GetPruner(ctx context.Context, sender *endpoint.Sender) (*pruner.Pruner) { + //p := &pruner.Pruner{ args: pruner.args{ctx, WithLogger(ctx, GetLogger(ctx).WithField("prune_side", "sender")), sender, sender, a.rules, envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10 * time.Second), true, a.promPruneSecs.WithLabelValues("sender")}, state: pruner.State.Plan,} + p := a.prunerFactory.BuildSinglePruner(ctx,sender,sender) + /*if err != nil { + return nil, errors.Wrap(err, "cannot build receiver pruning rules") + }*/ + return p +} + + +func (a *snap_ActiveSide) updateTasks(u func(*snap_activeSideTasks)) snap_activeSideTasks { + a.tasksMtx.Lock() + defer a.tasksMtx.Unlock() + var copy snap_activeSideTasks + copy = a.tasks + if u == nil { + return copy + } + u(©) + a.tasks = copy + return copy +} + + +type snap_activeMode interface { + Type() Type + RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) + //FSFilter() endpoint.FSFilter + FSFilter() zfs.DatasetFilter +} + + + +type modeSnap struct { +// fsfilter endpoint.FSFilter + fsfilter zfs.DatasetFilter + snapper *snapper.PeriodicOrManual +} + +func (m *modeSnap) Type() Type { return TypeSnap } + +func (m *modeSnap) RunPeriodic(ctx context.Context, wakeUpCommon chan <- struct{}) { + m.snapper.Run(ctx, wakeUpCommon) +} + +func (m *modeSnap) FSFilter() zfs.DatasetFilter { + return m.fsfilter +} + +func modeSnapFromConfig(g *config.Global, in *config.SnapJob) (*modeSnap, error) { + m := &modeSnap{} + fsf, err := filters.DatasetMapFilterFromConfig(in.Filesystems) + if err != nil { + return nil, errors.Wrap(err, "cannnot build filesystem filter") + } + m.fsfilter = fsf + + if m.snapper, err = snapper.FromConfig(g, fsf, in.Snapshotting); err != nil { + return nil, errors.Wrap(err, "cannot build snapper") + } + + return m, nil +} + + +func snap_activeSide(g *config.Global, in *config.SnapJob, mode *modeSnap) (j *snap_ActiveSide, err error) { + + j = &snap_ActiveSide{mode: mode} + j.name = in.Name +/*** j.promRepStateSecs = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "zrepl", + Subsystem: "replication", + Name: "state_time", + Help: "seconds spent during replication", + ConstLabels: prometheus.Labels{"zrepl_job":j.name}, + }, []string{"state"}) + j.promBytesReplicated = prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "zrepl", + Subsystem: "replication", + Name: "bytes_replicated", + Help: "number of bytes replicated from sender to receiver per filesystem", + ConstLabels: prometheus.Labels{"zrepl_job":j.name}, + }, []string{"filesystem"}) + + j.clientFactory, err = connecter.FromConfig(g, in.Connect) + if err != nil { + return nil, errors.Wrap(err, "cannot build client") + } +***/ + 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) + //j.rules = pruning.RulesFromConfig(in.Keep) + if err != nil { + return nil, errors.Wrap(err, "cannot build snapjob pruning rules") + } + return j, nil +} + +func (j *snap_ActiveSide) RegisterMetrics(registerer prometheus.Registerer) { + registerer.MustRegister(j.promPruneSecs) +} + +/*func (j *ActiveSide) Name() string { return j.name } + +type ActiveSideStatus struct { + Replication *replication.Report + PruningSender, PruningReceiver *pruner.Report +} +*/ + +func (j *snap_ActiveSide) Status() *Status { + tasks := j.updateTasks(nil) + + s := &ActiveSideStatus{} + t := j.mode.Type() + /*if tasks.replication != nil { + s.Replication = tasks.replication.Report() + } + if tasks.prunerSender != nil {*/ + if tasks.pruner != nil { + s.PruningSender = tasks.pruner.Report() + } + /*if tasks.prunerReceiver != nil { + s.PruningReceiver = tasks.prunerReceiver.Report() + }*/ + return &Status{Type: t, JobSpecific: s} +} + +func (j *snap_ActiveSide) 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.mode.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.do(WithLogger(ctx, invLog)) + } +} + +func (j *snap_ActiveSide) do(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 ActiveSideReplicating: + log.WithField("replication_progress", tasks.replication.Progress.String()). + Debug("check replication progress") + if tasks.replication.Progress.CheckTimeout(wdto, jitter) { + log.Error("replication did not make progress, cancelling" + WATCHDOG_ENVCONST_NOTICE) + tasks.replicationCancel() + return + }*/ + 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 ActiveSidePruneReceiver: + log.WithField("prune_receiver_progress", tasks.replication.Progress.String()). + Debug("check pruner_receiver progress") + if tasks.prunerReceiver.Progress.CheckTimeout(wdto, jitter) { + log.Error("pruner_receiver did not make progress, cancelling" + WATCHDOG_ENVCONST_NOTICE) + tasks.prunerReceiverCancel() + 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") + } + }) + + } + }() + +/* client, err := j.clientFactory.NewClient() + if err != nil { + log.WithError(err).Error("factory cannot instantiate streamrpc client") + } + defer client.Close(ctx) + + sender, receiver, err := j.mode.SenderReceiver(client) + { + select { + case <-ctx.Done(): + return + default: + } + ctx, repCancel := context.WithCancel(ctx) + tasks := j.updateTasks(func(tasks *activeSideTasks) { + // reset it + *tasks = activeSideTasks{} + tasks.replicationCancel = repCancel + tasks.replication = replication.NewReplication(j.promRepStateSecs, j.promBytesReplicated) + tasks.state = ActiveSideReplicating + }) + log.Info("start replication") + tasks.replication.Drive(ctx, sender, receiver) + repCancel() // always cancel to free up context resources + } + + { + select { + case <-ctx.Done(): + return + default: + } +*/ + ctx, localCancel := context.WithCancel(ctx) + sender := endpoint.NewSender(j.mode.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() +// } +/* { + select { + case <-ctx.Done(): + return + default: + } + ctx, receiverCancel := context.WithCancel(ctx) + tasks := j.updateTasks(func(tasks *activeSideTasks) { + tasks.prunerReceiver = j.prunerFactory.BuildReceiverPruner(ctx, receiver, sender) + tasks.prunerReceiverCancel = receiverCancel + tasks.state = ActiveSidePruneReceiver + }) + log.Info("start pruning receiver") + tasks.prunerReceiver.Prune() + log.Info("finished pruning receiver") + receiverCancel() + } +*/ + j.updateTasks(func(tasks *snap_activeSideTasks) { + tasks.state = ActiveSideDone + }) + +} + diff --git a/daemon/pruner/pruner.go b/daemon/pruner/pruner.go index d5705db..1d66d95 100644 --- a/daemon/pruner/pruner.go +++ b/daemon/pruner/pruner.go @@ -17,6 +17,7 @@ import ( "strings" "sync" "time" +// "github.com/zrepl/zrepl/replication" ) // Try to keep it compatible with gitub.com/zrepl/zrepl/replication.Endpoint @@ -82,6 +83,13 @@ type PrunerFactory struct { promPruneSecs *prometheus.HistogramVec } +type SinglePrunerFactory struct { + keepRules []pruning.KeepRule + retryWait time.Duration + considerSnapAtCursorReplicated bool + promPruneSecs *prometheus.HistogramVec +} + func checkContainsKeep1(rules []pruning.KeepRule) error { if len(rules) == 0 { return nil //No keep rules means keep all - ok @@ -95,6 +103,21 @@ func checkContainsKeep1(rules []pruning.KeepRule) error { return errors.New("sender keep rules must contain last_n or be empty so that the last snapshot is definitely kept") } +func NewSinglePrunerFactory(in config.PruningLocal, promPruneSecs *prometheus.HistogramVec) (*SinglePrunerFactory, error) { + rules, err := pruning.RulesFromConfig(in.Keep) + if err != nil { + return nil, errors.Wrap(err, "cannot build pruning rules") + } + considerSnapAtCursorReplicated := false + f := &SinglePrunerFactory{ + keepRules: rules, + retryWait: envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10 * time.Second), + considerSnapAtCursorReplicated: considerSnapAtCursorReplicated, + promPruneSecs: promPruneSecs, + } + return f, nil +} + func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus.HistogramVec) (*PrunerFactory, error) { keepRulesReceiver, err := pruning.RulesFromConfig(in.KeepReceiver) if err != nil { @@ -156,6 +179,22 @@ func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, return p } +func (f *SinglePrunerFactory) BuildSinglePruner(ctx context.Context, target Target, receiver History) *Pruner { + p := &Pruner{ + args: args{ + WithLogger(ctx, GetLogger(ctx).WithField("prune_side", "sender")), + target, + receiver, + f.keepRules, + f.retryWait, + f.considerSnapAtCursorReplicated, + f.promPruneSecs.WithLabelValues("sender"), + }, + state: Plan, + } + return p +} + //go:generate enumer -type=State type State int From 1265cc7934a7a02d9c048b90bf52d80fe5cb85f5 Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 01:34:50 +0100 Subject: [PATCH 02/27] pruned unused lines and comments ;) --- daemon/job/snapjob.go | 124 ---------------------------------------- daemon/pruner/pruner.go | 1 - 2 files changed, 125 deletions(-) diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 8515799..7998fe8 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -3,7 +3,6 @@ package job import ( "context" "github.com/pkg/errors" -// "github.com/problame/go-streamrpc" "github.com/prometheus/client_golang/prometheus" "github.com/zrepl/zrepl/config" "github.com/zrepl/zrepl/daemon/filters" @@ -11,11 +10,8 @@ import ( "github.com/zrepl/zrepl/daemon/job/wakeup" "github.com/zrepl/zrepl/daemon/logging" "github.com/zrepl/zrepl/daemon/pruner" - //"github.com/zrepl/zrepl/pruning" "github.com/zrepl/zrepl/daemon/snapper" - "github.com/zrepl/zrepl/daemon/transport/connecter" "github.com/zrepl/zrepl/endpoint" -// "github.com/zrepl/zrepl/replication" "github.com/zrepl/zrepl/util/envconst" "github.com/zrepl/zrepl/zfs" "sync" @@ -25,13 +21,10 @@ import ( type snap_ActiveSide struct { mode snap_activeMode name string - clientFactory *connecter.ClientFactory prunerFactory *pruner.SinglePrunerFactory - //promRepStateSecs *prometheus.HistogramVec // labels: state promPruneSecs *prometheus.HistogramVec // labels: prune_side - //promBytesReplicated *prometheus.CounterVec // labels: filesystem tasksMtx sync.Mutex tasks snap_activeSideTasks @@ -41,26 +34,15 @@ type snap_ActiveSide struct { type snap_activeSideTasks struct { state ActiveSideState - // valid for state ActiveSideReplicating, ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone - //replication *replication.Replication - //replicationCancel context.CancelFunc - - // valid for state ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone - //prunerSender, prunerReceiver *pruner.Pruner pruner *pruner.Pruner - // valid for state ActiveSidePruneReceiver, ActiveSideDone prunerCancel context.CancelFunc } func (a *snap_ActiveSide) Name() string { return a.name } func (a *snap_ActiveSide) GetPruner(ctx context.Context, sender *endpoint.Sender) (*pruner.Pruner) { - //p := &pruner.Pruner{ args: pruner.args{ctx, WithLogger(ctx, GetLogger(ctx).WithField("prune_side", "sender")), sender, sender, a.rules, envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10 * time.Second), true, a.promPruneSecs.WithLabelValues("sender")}, state: pruner.State.Plan,} p := a.prunerFactory.BuildSinglePruner(ctx,sender,sender) - /*if err != nil { - return nil, errors.Wrap(err, "cannot build receiver pruning rules") - }*/ return p } @@ -82,14 +64,12 @@ func (a *snap_ActiveSide) updateTasks(u func(*snap_activeSideTasks)) snap_active type snap_activeMode interface { Type() Type RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) - //FSFilter() endpoint.FSFilter FSFilter() zfs.DatasetFilter } type modeSnap struct { -// fsfilter endpoint.FSFilter fsfilter zfs.DatasetFilter snapper *snapper.PeriodicOrManual } @@ -124,26 +104,6 @@ func snap_activeSide(g *config.Global, in *config.SnapJob, mode *modeSnap) (j *s j = &snap_ActiveSide{mode: mode} j.name = in.Name -/*** j.promRepStateSecs = prometheus.NewHistogramVec(prometheus.HistogramOpts{ - Namespace: "zrepl", - Subsystem: "replication", - Name: "state_time", - Help: "seconds spent during replication", - ConstLabels: prometheus.Labels{"zrepl_job":j.name}, - }, []string{"state"}) - j.promBytesReplicated = prometheus.NewCounterVec(prometheus.CounterOpts{ - Namespace: "zrepl", - Subsystem: "replication", - Name: "bytes_replicated", - Help: "number of bytes replicated from sender to receiver per filesystem", - ConstLabels: prometheus.Labels{"zrepl_job":j.name}, - }, []string{"filesystem"}) - - j.clientFactory, err = connecter.FromConfig(g, in.Connect) - if err != nil { - return nil, errors.Wrap(err, "cannot build client") - } -***/ j.promPruneSecs = prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: "zrepl", Subsystem: "pruning", @@ -152,7 +112,6 @@ func snap_activeSide(g *config.Global, in *config.SnapJob, mode *modeSnap) (j *s ConstLabels: prometheus.Labels{"zrepl_job":j.name}, }, []string{"prune_side"}) j.prunerFactory, err = pruner.NewSinglePrunerFactory(in.Pruning, j.promPruneSecs) - //j.rules = pruning.RulesFromConfig(in.Keep) if err != nil { return nil, errors.Wrap(err, "cannot build snapjob pruning rules") } @@ -163,29 +122,14 @@ func (j *snap_ActiveSide) RegisterMetrics(registerer prometheus.Registerer) { registerer.MustRegister(j.promPruneSecs) } -/*func (j *ActiveSide) Name() string { return j.name } - -type ActiveSideStatus struct { - Replication *replication.Report - PruningSender, PruningReceiver *pruner.Report -} -*/ - func (j *snap_ActiveSide) Status() *Status { tasks := j.updateTasks(nil) s := &ActiveSideStatus{} t := j.mode.Type() - /*if tasks.replication != nil { - s.Replication = tasks.replication.Report() - } - if tasks.prunerSender != nil {*/ if tasks.pruner != nil { s.PruningSender = tasks.pruner.Report() } - /*if tasks.prunerReceiver != nil { - s.PruningReceiver = tasks.prunerReceiver.Report() - }*/ return &Status{Type: t, JobSpecific: s} } @@ -272,14 +216,6 @@ func (j *snap_ActiveSide) do(ctx context.Context) { const WATCHDOG_ENVCONST_NOTICE = " (adjust ZREPL_JOB_WATCHDOG_TIMEOUT env variable if inappropriate)" switch tasks.state { - /*case ActiveSideReplicating: - log.WithField("replication_progress", tasks.replication.Progress.String()). - Debug("check replication progress") - if tasks.replication.Progress.CheckTimeout(wdto, jitter) { - log.Error("replication did not make progress, cancelling" + WATCHDOG_ENVCONST_NOTICE) - tasks.replicationCancel() - return - }*/ case ActiveSidePruneSender: log.WithField("prune_sender_progress", "TEST DEBUG 123"). Debug("check pruner_sender progress") @@ -288,14 +224,6 @@ func (j *snap_ActiveSide) do(ctx context.Context) { tasks.prunerCancel() return } - /*case ActiveSidePruneReceiver: - log.WithField("prune_receiver_progress", tasks.replication.Progress.String()). - Debug("check pruner_receiver progress") - if tasks.prunerReceiver.Progress.CheckTimeout(wdto, jitter) { - log.Error("pruner_receiver did not make progress, cancelling" + WATCHDOG_ENVCONST_NOTICE) - tasks.prunerReceiverCancel() - return - }*/ case ActiveSideDone: // ignore, ctx will be Done() in a few milliseconds and the watchdog will exit default: @@ -307,39 +235,6 @@ func (j *snap_ActiveSide) do(ctx context.Context) { } }() -/* client, err := j.clientFactory.NewClient() - if err != nil { - log.WithError(err).Error("factory cannot instantiate streamrpc client") - } - defer client.Close(ctx) - - sender, receiver, err := j.mode.SenderReceiver(client) - { - select { - case <-ctx.Done(): - return - default: - } - ctx, repCancel := context.WithCancel(ctx) - tasks := j.updateTasks(func(tasks *activeSideTasks) { - // reset it - *tasks = activeSideTasks{} - tasks.replicationCancel = repCancel - tasks.replication = replication.NewReplication(j.promRepStateSecs, j.promBytesReplicated) - tasks.state = ActiveSideReplicating - }) - log.Info("start replication") - tasks.replication.Drive(ctx, sender, receiver) - repCancel() // always cancel to free up context resources - } - - { - select { - case <-ctx.Done(): - return - default: - } -*/ ctx, localCancel := context.WithCancel(ctx) sender := endpoint.NewSender(j.mode.FSFilter()) tasks := j.updateTasks(func(tasks *snap_activeSideTasks) { @@ -352,25 +247,6 @@ func (j *snap_ActiveSide) do(ctx context.Context) { tasks.pruner.Prune() log.Info("finished pruning sender") localCancel() -// } -/* { - select { - case <-ctx.Done(): - return - default: - } - ctx, receiverCancel := context.WithCancel(ctx) - tasks := j.updateTasks(func(tasks *activeSideTasks) { - tasks.prunerReceiver = j.prunerFactory.BuildReceiverPruner(ctx, receiver, sender) - tasks.prunerReceiverCancel = receiverCancel - tasks.state = ActiveSidePruneReceiver - }) - log.Info("start pruning receiver") - tasks.prunerReceiver.Prune() - log.Info("finished pruning receiver") - receiverCancel() - } -*/ j.updateTasks(func(tasks *snap_activeSideTasks) { tasks.state = ActiveSideDone }) diff --git a/daemon/pruner/pruner.go b/daemon/pruner/pruner.go index 1d66d95..b1f391f 100644 --- a/daemon/pruner/pruner.go +++ b/daemon/pruner/pruner.go @@ -17,7 +17,6 @@ import ( "strings" "sync" "time" -// "github.com/zrepl/zrepl/replication" ) // Try to keep it compatible with gitub.com/zrepl/zrepl/replication.Endpoint From 19d0916e3434412a3249ca79dd57e5d76fd7f200 Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 01:54:56 +0100 Subject: [PATCH 03/27] remove snapMode, rename snap_ActiveSide to SnapJob --- daemon/job/build_jobs.go | 6 +--- daemon/job/snapjob.go | 76 +++++++++++++++------------------------- 2 files changed, 29 insertions(+), 53 deletions(-) diff --git a/daemon/job/build_jobs.go b/daemon/job/build_jobs.go index d8f2371..5f716a6 100644 --- a/daemon/job/build_jobs.go +++ b/daemon/job/build_jobs.go @@ -46,11 +46,7 @@ func buildJob(c *config.Global, in config.JobEnum) (j Job, err error) { return cannotBuildJob(err, v.Name) } case *config.SnapJob: - m, err := modeSnapFromConfig(c, v) - if err != nil { - return cannotBuildJob(err, v.Name) - } - j, err = snap_activeSide(c, v, m) + j, err = snapJob(c, v) if err != nil { return cannotBuildJob(err, v.Name) } diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 7998fe8..82d2b28 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -18,9 +18,10 @@ import ( "time" ) -type snap_ActiveSide struct { - mode snap_activeMode +type SnapJob struct { name string + fsfilter zfs.DatasetFilter + snapper *snapper.PeriodicOrManual prunerFactory *pruner.SinglePrunerFactory @@ -39,70 +40,49 @@ type snap_activeSideTasks struct { prunerCancel context.CancelFunc } -func (a *snap_ActiveSide) Name() string { return a.name } +func (j *SnapJob) Name() string { return j.name } -func (a *snap_ActiveSide) GetPruner(ctx context.Context, sender *endpoint.Sender) (*pruner.Pruner) { - p := a.prunerFactory.BuildSinglePruner(ctx,sender,sender) +func (j *SnapJob) GetPruner(ctx context.Context, sender *endpoint.Sender) (*pruner.Pruner) { + p := j.prunerFactory.BuildSinglePruner(ctx,sender,sender) return p } -func (a *snap_ActiveSide) updateTasks(u func(*snap_activeSideTasks)) snap_activeSideTasks { - a.tasksMtx.Lock() - defer a.tasksMtx.Unlock() +func (j *SnapJob) updateTasks(u func(*snap_activeSideTasks)) snap_activeSideTasks { + j.tasksMtx.Lock() + defer j.tasksMtx.Unlock() var copy snap_activeSideTasks - copy = a.tasks + copy = j.tasks if u == nil { return copy } u(©) - a.tasks = copy + j.tasks = copy return copy } -type snap_activeMode interface { - Type() Type - RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) - FSFilter() zfs.DatasetFilter +func (j *SnapJob) Type() Type { return TypeSnap } + +func (j *SnapJob) RunPeriodic(ctx context.Context, wakeUpCommon chan <- struct{}) { + j.snapper.Run(ctx, wakeUpCommon) } - - -type modeSnap struct { - fsfilter zfs.DatasetFilter - snapper *snapper.PeriodicOrManual +func (j *SnapJob) FSFilter() zfs.DatasetFilter { + return j.fsfilter } -func (m *modeSnap) Type() Type { return TypeSnap } - -func (m *modeSnap) RunPeriodic(ctx context.Context, wakeUpCommon chan <- struct{}) { - m.snapper.Run(ctx, wakeUpCommon) -} - -func (m *modeSnap) FSFilter() zfs.DatasetFilter { - return m.fsfilter -} - -func modeSnapFromConfig(g *config.Global, in *config.SnapJob) (*modeSnap, error) { - m := &modeSnap{} +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") } - m.fsfilter = fsf + j.fsfilter = fsf - if m.snapper, err = snapper.FromConfig(g, fsf, in.Snapshotting); err != nil { + if j.snapper, err = snapper.FromConfig(g, fsf, in.Snapshotting); err != nil { return nil, errors.Wrap(err, "cannot build snapper") } - - return m, nil -} - - -func snap_activeSide(g *config.Global, in *config.SnapJob, mode *modeSnap) (j *snap_ActiveSide, err error) { - - j = &snap_ActiveSide{mode: mode} j.name = in.Name j.promPruneSecs = prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: "zrepl", @@ -118,22 +98,22 @@ func snap_activeSide(g *config.Global, in *config.SnapJob, mode *modeSnap) (j *s return j, nil } -func (j *snap_ActiveSide) RegisterMetrics(registerer prometheus.Registerer) { +func (j *SnapJob) RegisterMetrics(registerer prometheus.Registerer) { registerer.MustRegister(j.promPruneSecs) } -func (j *snap_ActiveSide) Status() *Status { +func (j *SnapJob) Status() *Status { tasks := j.updateTasks(nil) s := &ActiveSideStatus{} - t := j.mode.Type() + t := j.Type() if tasks.pruner != nil { s.PruningSender = tasks.pruner.Report() } return &Status{Type: t, JobSpecific: s} } -func (j *snap_ActiveSide) Run(ctx context.Context) { +func (j *SnapJob) Run(ctx context.Context) { log := GetLogger(ctx) ctx = logging.WithSubsystemLoggers(ctx, log) @@ -142,7 +122,7 @@ func (j *snap_ActiveSide) Run(ctx context.Context) { periodicDone := make(chan struct{}) ctx, cancel := context.WithCancel(ctx) defer cancel() - go j.mode.RunPeriodic(ctx, periodicDone) + go j.RunPeriodic(ctx, periodicDone) invocationCount := 0 outer: @@ -162,7 +142,7 @@ outer: } } -func (j *snap_ActiveSide) do(ctx context.Context) { +func (j *SnapJob) do(ctx context.Context) { log := GetLogger(ctx) ctx = logging.WithSubsystemLoggers(ctx, log) @@ -236,7 +216,7 @@ func (j *snap_ActiveSide) do(ctx context.Context) { }() ctx, localCancel := context.WithCancel(ctx) - sender := endpoint.NewSender(j.mode.FSFilter()) + sender := endpoint.NewSender(j.FSFilter()) tasks := j.updateTasks(func(tasks *snap_activeSideTasks) { tasks.pruner = j.GetPruner(ctx, sender) tasks.prunerCancel = localCancel From 58dcc0743045b050b3eff876ac7d9a6ba7fde2cb Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 02:08:39 +0100 Subject: [PATCH 04/27] Added SnapJobStatus --- daemon/job/snapjob.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 82d2b28..1467588 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -102,13 +102,17 @@ func (j *SnapJob) RegisterMetrics(registerer prometheus.Registerer) { registerer.MustRegister(j.promPruneSecs) } +type SnapJobStatus struct { + Pruning *pruner.Report +} + func (j *SnapJob) Status() *Status { tasks := j.updateTasks(nil) - s := &ActiveSideStatus{} + s := &SnapJobStatus{} t := j.Type() if tasks.pruner != nil { - s.PruningSender = tasks.pruner.Report() + s.Pruning = tasks.pruner.Report() } return &Status{Type: t, JobSpecific: s} } From 442d61918b2597a9137bfcb1baec83afdc97681a Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 02:42:13 +0100 Subject: [PATCH 05/27] remove most of the watchdog machinery --- daemon/job/snapjob.go | 130 ++++-------------------------------------- 1 file changed, 11 insertions(+), 119 deletions(-) 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") } From 141e49727c02443eb8a85cce99c6dabab7b741b0 Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 02:51:23 +0100 Subject: [PATCH 06/27] Missed a last reference to tasks --- daemon/job/snapjob.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 902ffd5..dc6c9bc 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -80,8 +80,6 @@ type SnapJobStatus struct { } func (j *SnapJob) Status() *Status { - //tasks := j.updateTasks(nil) - s := &SnapJobStatus{} t := j.Type() if j.pruner != nil { From 7de3c0a09a1ccd29c2422d80bb9fd1c7ebd34e63 Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 02:52:33 +0100 Subject: [PATCH 07/27] Removed the references to a pruning 'side' in the singlepruner logging code and the snapjob prometheus thing. --- daemon/job/snapjob.go | 4 ++-- daemon/pruner/pruner.go | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index dc6c9bc..45722e9 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -21,7 +21,7 @@ type SnapJob struct { prunerFactory *pruner.SinglePrunerFactory - promPruneSecs *prometheus.HistogramVec // labels: prune_side + promPruneSecs *prometheus.HistogramVec // no labels! pruner *pruner.Pruner } @@ -63,7 +63,7 @@ func snapJob(g *config.Global, in *config.SnapJob) (j *SnapJob, err error) { Name: "time", Help: "seconds spent in pruner", ConstLabels: prometheus.Labels{"zrepl_job":j.name}, - }, []string{"prune_side"}) + }, []string {}) j.prunerFactory, err = pruner.NewSinglePrunerFactory(in.Pruning, j.promPruneSecs) if err != nil { return nil, errors.Wrap(err, "cannot build snapjob pruning rules") diff --git a/daemon/pruner/pruner.go b/daemon/pruner/pruner.go index b1f391f..412c434 100644 --- a/daemon/pruner/pruner.go +++ b/daemon/pruner/pruner.go @@ -181,13 +181,13 @@ func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, func (f *SinglePrunerFactory) BuildSinglePruner(ctx context.Context, target Target, receiver History) *Pruner { p := &Pruner{ args: args{ - WithLogger(ctx, GetLogger(ctx).WithField("prune_side", "sender")), + ctx, target, receiver, f.keepRules, f.retryWait, f.considerSnapAtCursorReplicated, - f.promPruneSecs.WithLabelValues("sender"), + f.promPruneSecs.WithLabelValues(), }, state: Plan, } From dd11fc96db552bdef8e9771d153d1e750b8dbeae Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 03:27:39 +0100 Subject: [PATCH 08/27] Touchups in job.go --- daemon/job/job.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/daemon/job/job.go b/daemon/job/job.go index 6888fed..0c97f94 100644 --- a/daemon/job/job.go +++ b/daemon/job/job.go @@ -39,7 +39,7 @@ type Type string const ( TypeInternal Type = "internal" - TypeSnap = "snap" + TypeSnap Type = "snap" TypePush Type = "push" TypeSink Type = "sink" TypePull Type = "pull" @@ -86,7 +86,9 @@ func (s *Status) UnmarshalJSON(in []byte) (err error) { } switch s.Type { case TypeSnap: - fallthrough + var st SnapJobStatus + err = json.Unmarshal(jobJSON, &st) + s.JobSpecific = &st case TypePull: fallthrough case TypePush: var st ActiveSideStatus From e10dc129deac5165a77f0b8c8d2da165ad9d64d2 Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 03:39:03 +0100 Subject: [PATCH 09/27] Make getPruner() private --- daemon/job/snapjob.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 45722e9..f500520 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -29,7 +29,7 @@ type SnapJob struct { func (j *SnapJob) Name() string { return j.name } -func (j *SnapJob) GetPruner(ctx context.Context, sender *endpoint.Sender) (*pruner.Pruner) { +func (j *SnapJob) getPruner(ctx context.Context, sender *endpoint.Sender) (*pruner.Pruner) { p := j.prunerFactory.BuildSinglePruner(ctx,sender,sender) return p } @@ -121,7 +121,7 @@ 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) + j.pruner = j.getPruner(ctx, sender) log.Info("start pruning") j.pruner.Prune() log.Info("finished pruning") From c4e23862cd6e7a892fa1981f835405502501b1ec Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 04:06:13 +0100 Subject: [PATCH 10/27] Added status view for SnapJob. --- client/status.go | 65 ++++++++++++++++++++++++++++-------------------- 1 file changed, 38 insertions(+), 27 deletions(-) diff --git a/client/status.go b/client/status.go index bf882f7..f5e886e 100644 --- a/client/status.go +++ b/client/status.go @@ -298,7 +298,44 @@ func (t *tui) draw() { t.setIndent(1) t.newline() - if v.Type != job.TypePush && v.Type != job.TypePull { + if v.Type == job.TypePush || v.Type == job.TypePull { + activeStatus, ok := v.JobSpecific.(*job.ActiveSideStatus) + if !ok || activeStatus == nil { + t.printf("ActiveSideStatus is null") + t.newline() + continue + } + + t.printf("Replication:") + t.newline() + t.addIndent(1) + t.renderReplicationReport(activeStatus.Replication, t.getReplicationProgresHistory(k)) + t.addIndent(-1) + + t.printf("Pruning Sender:") + t.newline() + t.addIndent(1) + t.renderPrunerReport(activeStatus.PruningSender) + t.addIndent(-1) + + t.printf("Pruning Receiver:") + t.newline() + t.addIndent(1) + t.renderPrunerReport(activeStatus.PruningReceiver) + t.addIndent(-1) + } else if v.Type == job.TypeSnap { + snapStatus, ok := v.JobSpecific.(*job.SnapJobStatus) + if !ok || snapStatus == nil { + t.printf("SnapJobStatus is null") + t.newline() + continue + } + t.printf("Pruning snapshots:") + t.newline() + t.addIndent(1) + t.renderPrunerReport(snapStatus.Pruning) + t.addIndent(-1) + } else { t.printf("No status representation for job type '%s', dumping as YAML", v.Type) t.newline() asYaml, err := yaml.Marshal(v.JobSpecific) @@ -311,32 +348,6 @@ func (t *tui) draw() { t.newline() continue } - - pushStatus, ok := v.JobSpecific.(*job.ActiveSideStatus) - if !ok || pushStatus == nil { - t.printf("ActiveSideStatus is null") - t.newline() - continue - } - - t.printf("Replication:") - t.newline() - t.addIndent(1) - t.renderReplicationReport(pushStatus.Replication, t.getReplicationProgresHistory(k)) - t.addIndent(-1) - - t.printf("Pruning Sender:") - t.newline() - t.addIndent(1) - t.renderPrunerReport(pushStatus.PruningSender) - t.addIndent(-1) - - t.printf("Pruning Receiver:") - t.newline() - t.addIndent(1) - t.renderPrunerReport(pushStatus.PruningReceiver) - t.addIndent(-1) - } } termbox.Flush() From 22d9830baadb7b54b934a567fd50d1c30af54c5b Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 04:26:03 +0100 Subject: [PATCH 11/27] Fix prometheus with multiple jobs --- daemon/job/snapjob.go | 2 +- daemon/pruner/pruner.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index f500520..7ff6a19 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -63,7 +63,7 @@ func snapJob(g *config.Global, in *config.SnapJob) (j *SnapJob, err error) { Name: "time", Help: "seconds spent in pruner", ConstLabels: prometheus.Labels{"zrepl_job":j.name}, - }, []string {}) + }, []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") diff --git a/daemon/pruner/pruner.go b/daemon/pruner/pruner.go index 412c434..9446dd1 100644 --- a/daemon/pruner/pruner.go +++ b/daemon/pruner/pruner.go @@ -187,7 +187,7 @@ func (f *SinglePrunerFactory) BuildSinglePruner(ctx context.Context, target Targ f.keepRules, f.retryWait, f.considerSnapAtCursorReplicated, - f.promPruneSecs.WithLabelValues(), + f.promPruneSecs.WithLabelValues("local"), }, state: Plan, } From d0f898751fed62aa21d2aeff3c370b1f10005cba Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 14:02:21 +0100 Subject: [PATCH 12/27] Gofmt snapjob.go --- daemon/job/snapjob.go | 37 +++++++++++++++++-------------------- 1 file changed, 17 insertions(+), 20 deletions(-) diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 7ff6a19..173d7db 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -15,9 +15,9 @@ import ( ) type SnapJob struct { - name string - fsfilter zfs.DatasetFilter - snapper *snapper.PeriodicOrManual + name string + fsfilter zfs.DatasetFilter + snapper *snapper.PeriodicOrManual prunerFactory *pruner.SinglePrunerFactory @@ -26,19 +26,17 @@ type SnapJob struct { 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) 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) RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) { + j.snapper.Run(ctx, wakeUpCommon) } func (j *SnapJob) FSFilter() zfs.DatasetFilter { @@ -47,22 +45,22 @@ func (j *SnapJob) FSFilter() zfs.DatasetFilter { 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 + 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") - } + 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}, + ConstLabels: prometheus.Labels{"zrepl_job": j.name}, }, []string{"prune_side"}) j.prunerFactory, err = pruner.NewSinglePrunerFactory(in.Pruning, j.promPruneSecs) if err != nil { @@ -126,4 +124,3 @@ func (j *SnapJob) doPrune(ctx context.Context) { j.pruner.Prune() log.Info("finished pruning") } - From b79ad3ddc3570497822c083c135641ec3cca816b Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 14:17:38 +0100 Subject: [PATCH 13/27] Honour PruneKeepNotReplicated.KeepSnashotAtCursor in SinglePrunerFactory --- daemon/pruner/pruner.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/daemon/pruner/pruner.go b/daemon/pruner/pruner.go index 9446dd1..1e7fc8e 100644 --- a/daemon/pruner/pruner.go +++ b/daemon/pruner/pruner.go @@ -108,6 +108,13 @@ func NewSinglePrunerFactory(in config.PruningLocal, promPruneSecs *prometheus.Hi return nil, errors.Wrap(err, "cannot build pruning rules") } considerSnapAtCursorReplicated := false + for _, r := range in.Keep { + knr, ok := r.Ret.(*config.PruneKeepNotReplicated) + if !ok { + continue + } + considerSnapAtCursorReplicated = considerSnapAtCursorReplicated || !knr.KeepSnapshotAtCursor + } f := &SinglePrunerFactory{ keepRules: rules, retryWait: envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10 * time.Second), From e9564a7e5c96865d62bfa5c636fdb38f9dda7c54 Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 14:35:40 +0100 Subject: [PATCH 14/27] Inlined a couple legacy leftover functions from the mode copypasta --- daemon/job/snapjob.go | 19 +++---------------- 1 file changed, 3 insertions(+), 16 deletions(-) diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 173d7db..2e0034f 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -28,21 +28,8 @@ type SnapJob struct { 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) @@ -95,7 +82,7 @@ func (j *SnapJob) Run(ctx context.Context) { periodicDone := make(chan struct{}) ctx, cancel := context.WithCancel(ctx) defer cancel() - go j.RunPeriodic(ctx, periodicDone) + go j.snapper.Run(ctx, periodicDone) invocationCount := 0 outer: @@ -118,8 +105,8 @@ outer: 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) + sender := endpoint.NewSender(j.fsfilter) + j.pruner = j.prunerFactory.BuildSinglePruner(ctx, sender, sender) log.Info("start pruning") j.pruner.Prune() log.Info("finished pruning") From 3cef76d4633285fc62bdbc2ddf98731a69580767 Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 14:37:03 +0100 Subject: [PATCH 15/27] Refactor snapJob() to snapJobFromConfig() --- daemon/job/build_jobs.go | 2 +- daemon/job/snapjob.go | 2 +- daemon/job/snapjob.go_prefmt | 129 +++++++++++++++++++++++++++++++++++ 3 files changed, 131 insertions(+), 2 deletions(-) create mode 100644 daemon/job/snapjob.go_prefmt 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") +} + From d977796f18e4fbb3099e915e09a010c2743977b4 Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 16:59:46 +0100 Subject: [PATCH 16/27] Add SnapJob docs --- docs/configuration/jobs.rst | 41 +++++++++++++++++++++++++++++++++--- docs/configuration/prune.rst | 33 ++++++++++++++++++++++++++--- 2 files changed, 68 insertions(+), 6 deletions(-) diff --git a/docs/configuration/jobs.rst b/docs/configuration/jobs.rst index a7bdcae..ae25e2f 100644 --- a/docs/configuration/jobs.rst +++ b/docs/configuration/jobs.rst @@ -17,7 +17,7 @@ Overview & Terminology A *job* is the unit of activity tracked by the zrepl daemon and configured in the |mainconfig|. Every job has a unique ``name``, a ``type`` and type-dependent fields which are documented on this page. -Replication always happens between a pair of jobs: one is the **active side**, and one the **passive side**. +Aside from ``snap`` jobs, which are special jobs only for snapshotting and pruning, replication always happens between a pair of jobs: one is the **active side**, and one the **passive side**. The active side executes the replication logic whereas the passive side responds to requests after checking the active side's permissions. For communication, the active side connects to the passive side using a :ref:`transport ` and starts issuing remote procedure calls (RPCs). @@ -35,6 +35,12 @@ The following table shows how different job types can be combined to achieve bot | Local replication | | ``push`` + ``sink`` in one config | * Backup FreeBSD boot pool | | | | with :ref:`local transport ` | | +-----------------------+--------------+----------------------------------+-----------------------------------------------+ +| SnapJob | ``snap`` | N/A | * Data requires versioning but no backups | +| | | | * Combined with a ``push`` or ``pull`` job: | +| | | | | +| | | | * High frequency local snapshotting | +| | | | * Low frequency replication | ++-----------------------+--------------+----------------------------------+-----------------------------------------------+ How the Active Side Works ~~~~~~~~~~~~~~~~~~~~~~~~~ @@ -114,7 +120,7 @@ Regardless of whether that keep rule is used, the bookmark ensures that replicat Taking Snaphots --------------- -The ``push`` and ``source`` jobs can automatically take periodic snapshots of the filesystems matched by the ``filesystems`` filter field. +The ``push``, ``source`` and ``snap`` jobs can automatically take periodic snapshots of the filesystems matched by the ``filesystems`` filter field. The snapshot names are composed of a user-defined prefix followed by a UTC date formatted like ``20060102_150405_000``. We use UTC because it will avoid name conflicts when switching time zones or between summer and winter time. @@ -136,7 +142,7 @@ For ``push`` jobs, replication is automatically triggered after all filesystems There is also a ``manual`` snapshotting type, which covers the following use cases: -* Existing infrastructure for automatic snapshots: you only want to use zrepl for replication. +* Separate ``snap`` job or external infrastructures for automatic snapshots: you want to use this zrepl job for replication. * Run scripts before and after taking snapshots (like locking database tables). We are working on better integration for this use case: see :issue:`74`. @@ -255,6 +261,35 @@ Job Type ``source`` Example config: :sampleconf:`/source.yml` +.. _job-snap: + +Job Type ``snap`` +----------------- + +Job type that only takes care of local snapshotting and pruning. + +.. list-table:: + :widths: 20 80 + :header-rows: 1 + + * - Parameter + - Comment + * - ``type`` + - = ``snap`` + * - ``name`` + - unique name of the job + * - ``filesystems`` + - |filter-spec| for filesystems to be snapshotted + * - ``snapshotting`` + - |snapshotting-spec| + * - ``pruning`` + - |pruning-spec| + +Example config: :sampleconf:`/snap.yml` + + + + .. _replication-local: Local replication diff --git a/docs/configuration/prune.rst b/docs/configuration/prune.rst index ef40ce3..63b8042 100644 --- a/docs/configuration/prune.rst +++ b/docs/configuration/prune.rst @@ -10,10 +10,12 @@ Typically, the requirements to temporal resolution and maximum retention time di For example, when using zrepl to back up a busy database server, you will want high temporal resolution (snapshots every 10 min) for the last 24h in case of administrative disasters, but cannot afford to store them for much longer because you might have high turnover volume in the database. On the receiving side, you may have more disk space available, or need to comply with other backup retention policies. -zrepl uses a set of **keep rules** to determine which snapshots shall be kept per filesystem. +zrepl uses a set of **keep rules** per replication side to determine which snapshots shall be kept per filesystem. **A snapshot that is not kept by any rule is destroyed.** The keep rules are **evaluated on the active side** (:ref:`push ` or :ref:`pull job `) of the replication setup, for both active and passive side, after replication completed or was determined to have failed permanently. + + Example Configuration: :: @@ -53,14 +55,39 @@ Example Configuration: It is currently not possible to define pruning on a source job. The source job creates snapshots, which means that extended replication downtime will fill up the source's zpool with snapshots, since pruning is directed by the corresponding active side (pull job). - If this is a potential risk for you, consider using :ref:`push mode `. + If this is a potential risk for you, consider using :ref:`push mode ` or adding a pruning-only :ref:`snap job ` to thin out extremely old snapshots in case the pull job doesn't prune for a very long time. + + +.. _prune-local-vs-twosided: + +Local vs two-sided ``pruning`` Syntax +------------------------------------- + +Since ``snap`` jobs work locally only, their ``pruning`` field takes a simple ``keep`` instead of ``keep_sender`` and ``keep_receiver``. + +:: + + jobs: + - type: snap + pruning: + keep: + - type: not_replicated + ... + + - type: push + pruning: + keep_sender: + - type: not_replicated + ... + keep_receiver: + ... + .. _prune-keep-not-replicated: Policy ``not_replicated`` ------------------------- - :: jobs: From 160a3b6d32cdd6872e98bba5caba22400dc76aea Mon Sep 17 00:00:00 2001 From: InsanePrawn Date: Wed, 21 Nov 2018 21:44:48 +0100 Subject: [PATCH 17/27] more gofmt, drop snapjob.go_prefmt after it was accidentally added --- daemon/job/snapjob.go | 4 +- daemon/job/snapjob.go_prefmt | 129 ----------------------------------- 2 files changed, 2 insertions(+), 131 deletions(-) delete mode 100644 daemon/job/snapjob.go_prefmt diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 138c660..15feb42 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -82,7 +82,7 @@ func (j *SnapJob) Run(ctx context.Context) { periodicDone := make(chan struct{}) ctx, cancel := context.WithCancel(ctx) defer cancel() - go j.snapper.Run(ctx, periodicDone) + go j.snapper.Run(ctx, periodicDone) invocationCount := 0 outer: @@ -106,7 +106,7 @@ func (j *SnapJob) doPrune(ctx context.Context) { log := GetLogger(ctx) ctx = logging.WithSubsystemLoggers(ctx, log) sender := endpoint.NewSender(j.fsfilter) - j.pruner = j.prunerFactory.BuildSinglePruner(ctx, sender, sender) + j.pruner = j.prunerFactory.BuildSinglePruner(ctx, sender, sender) log.Info("start pruning") j.pruner.Prune() log.Info("finished pruning") diff --git a/daemon/job/snapjob.go_prefmt b/daemon/job/snapjob.go_prefmt deleted file mode 100644 index 7ff6a19..0000000 --- a/daemon/job/snapjob.go_prefmt +++ /dev/null @@ -1,129 +0,0 @@ -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") -} - From 5c5e8c0bafc38eca85f1dd5194fc6210d60b1d5c Mon Sep 17 00:00:00 2001 From: Jakob Berger Date: Tue, 22 Jan 2019 16:46:34 +0100 Subject: [PATCH 18/27] Documentation changes mostly as requested --- docs/configuration/jobs.rst | 5 ++-- docs/configuration/prune.rst | 56 ++++++++++++++++++------------------ 2 files changed, 31 insertions(+), 30 deletions(-) diff --git a/docs/configuration/jobs.rst b/docs/configuration/jobs.rst index ae25e2f..f601c16 100644 --- a/docs/configuration/jobs.rst +++ b/docs/configuration/jobs.rst @@ -17,7 +17,7 @@ Overview & Terminology A *job* is the unit of activity tracked by the zrepl daemon and configured in the |mainconfig|. Every job has a unique ``name``, a ``type`` and type-dependent fields which are documented on this page. -Aside from ``snap`` jobs, which are special jobs only for snapshotting and pruning, replication always happens between a pair of jobs: one is the **active side**, and one the **passive side**. +Replication always happens between a pair of jobs: one is the **active side**, and one the **passive side**. The active side executes the replication logic whereas the passive side responds to requests after checking the active side's permissions. For communication, the active side connects to the passive side using a :ref:`transport ` and starts issuing remote procedure calls (RPCs). @@ -142,9 +142,10 @@ For ``push`` jobs, replication is automatically triggered after all filesystems There is also a ``manual`` snapshotting type, which covers the following use cases: -* Separate ``snap`` job or external infrastructures for automatic snapshots: you want to use this zrepl job for replication. +* Existing infrastructure for automatic snapshots: you only want to use this zrepl job for replication. * Run scripts before and after taking snapshots (like locking database tables). We are working on better integration for this use case: see :issue:`74`. +* Handling snapshotting through a separate ``snap`` job. Note that you will have to trigger replication manually using the ``zrepl signal wakeup JOB`` subcommand in that case. diff --git a/docs/configuration/prune.rst b/docs/configuration/prune.rst index 63b8042..98104b6 100644 --- a/docs/configuration/prune.rst +++ b/docs/configuration/prune.rst @@ -10,7 +10,7 @@ Typically, the requirements to temporal resolution and maximum retention time di For example, when using zrepl to back up a busy database server, you will want high temporal resolution (snapshots every 10 min) for the last 24h in case of administrative disasters, but cannot afford to store them for much longer because you might have high turnover volume in the database. On the receiving side, you may have more disk space available, or need to comply with other backup retention policies. -zrepl uses a set of **keep rules** per replication side to determine which snapshots shall be kept per filesystem. +zrepl uses a set of **keep rules** per sending and receiving side to determine which snapshots shall be kept per filesystem. **A snapshot that is not kept by any rule is destroyed.** The keep rules are **evaluated on the active side** (:ref:`push ` or :ref:`pull job `) of the replication setup, for both active and passive side, after replication completed or was determined to have failed permanently. @@ -51,37 +51,15 @@ Example Configuration: You might have **existing snapshots** of filesystems affected by pruning which you want to keep, i.e. not be destroyed by zrepl. Make sure to actually add the necessary ``regex`` keep rules on both sides, like with ``manual`` in the example above. +.. _prune-attention-issue-102: + .. ATTENTION:: It is currently not possible to define pruning on a source job. The source job creates snapshots, which means that extended replication downtime will fill up the source's zpool with snapshots, since pruning is directed by the corresponding active side (pull job). - If this is a potential risk for you, consider using :ref:`push mode ` or adding a pruning-only :ref:`snap job ` to thin out extremely old snapshots in case the pull job doesn't prune for a very long time. - - -.. _prune-local-vs-twosided: - -Local vs two-sided ``pruning`` Syntax -------------------------------------- - -Since ``snap`` jobs work locally only, their ``pruning`` field takes a simple ``keep`` instead of ``keep_sender`` and ``keep_receiver``. - -:: - - jobs: - - type: snap - pruning: - keep: - - type: not_replicated - ... - - - type: push - pruning: - keep_sender: - - type: not_replicated - ... - keep_receiver: - ... - + If this is a potential risk for you, consider using :ref:`push mode `. + As a :ref:`temporary workaround` for :issue:`102` you can define a pruning-only :ref:`snap job ` to thin out extremely old snapshots in case the active side doesn't prune for a very long time. + **Note**: Since jobs run in parallel, it is possible for a ``snap`` job to prune snapshots that are queued to be replicated or destroyed by the remote ``pull`` job. .. _prune-keep-not-replicated: @@ -191,4 +169,26 @@ Policy ``regex`` Like all other regular expression fields in prune policies, zrepl uses Go's `regexp.Regexp `_ Perl-compatible regular expressions (`Syntax `_). The optional `negate` boolean field inverts the semantics: Use it if you want to keep all snapshots that *do not* match the given regex. +.. _prune-workaround-issue-102: +Workaround for Source-side snapshot pruning +------------------------------------------- + +Using the ``snap`` job type as a workaround for :issue:`102` as mentioned :ref:`here `: + +:: + + jobs: + - type: snap + pruning: + keep: + - type: grid + ... + + - type: pull + pruning: + keep_sender: + - type: not_replicated + ... + keep_receiver: + ... From 3543fbbb650fb0e09a73368232d6e18b8734437f Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Fri, 15 Mar 2019 21:54:46 +0100 Subject: [PATCH 19/27] docs: clarify language & example of source-side pruning workaround --- docs/configuration/prune.rst | 75 ++++++++++++++++++++++-------------- 1 file changed, 47 insertions(+), 28 deletions(-) diff --git a/docs/configuration/prune.rst b/docs/configuration/prune.rst index 98104b6..3d5d6e8 100644 --- a/docs/configuration/prune.rst +++ b/docs/configuration/prune.rst @@ -51,17 +51,6 @@ Example Configuration: You might have **existing snapshots** of filesystems affected by pruning which you want to keep, i.e. not be destroyed by zrepl. Make sure to actually add the necessary ``regex`` keep rules on both sides, like with ``manual`` in the example above. -.. _prune-attention-issue-102: - -.. ATTENTION:: - - It is currently not possible to define pruning on a source job. - The source job creates snapshots, which means that extended replication downtime will fill up the source's zpool with snapshots, since pruning is directed by the corresponding active side (pull job). - If this is a potential risk for you, consider using :ref:`push mode `. - As a :ref:`temporary workaround` for :issue:`102` you can define a pruning-only :ref:`snap job ` to thin out extremely old snapshots in case the active side doesn't prune for a very long time. - **Note**: Since jobs run in parallel, it is possible for a ``snap`` job to prune snapshots that are queued to be replicated or destroyed by the remote ``pull`` job. - - .. _prune-keep-not-replicated: Policy ``not_replicated`` @@ -169,26 +158,56 @@ Policy ``regex`` Like all other regular expression fields in prune policies, zrepl uses Go's `regexp.Regexp `_ Perl-compatible regular expressions (`Syntax `_). The optional `negate` boolean field inverts the semantics: Use it if you want to keep all snapshots that *do not* match the given regex. -.. _prune-workaround-issue-102: -Workaround for Source-side snapshot pruning -------------------------------------------- +Source-side snapshot pruning +---------------------------- -Using the ``snap`` job type as a workaround for :issue:`102` as mentioned :ref:`here `: +A :ref:`source jobs` takes snapshots on the system it runs on. +The corresponding :ref:`pull job ` on the replication target connects to the source job and replicates the snapshots. +Afterwards, the pull job coordinates pruning on both sender (the source job side) and receiver (the pull job side). + +There is no built-in way to define and execute pruning on the source side independently of the pull side. +The source job will continue taking snapshots which will not be pruned until the pull side connects. +This means that **extended replication downtime will fill up the source's zpool with snapshots**. + +If the above is a conceivable situation for you, consider using :ref:`push mode `, where pruning happens on the same side where snapshots are taken. + +Workaround using ``snap`` job +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +As a workaround (see GitHub :issue:`102` for development progress), a pruning-only :ref:`snap job ` can be defined on the source side: +The snap job is in charge of snapshot creation & destruction, whereas the source job's role is reduced to just serving snapshots. +However, since, jobs are run independently, it is possible that the snap job will prune snapshots that are queued for replication / destruction by the remote pull job that connects to the source job. +There is no possible configuration to avoid this problem with the current version of zrepl. + +Example configuration: :: - jobs: - - type: snap - pruning: - keep: - - type: grid - ... + # source side + jobs: + - type: snap + snapshotting: + type: periodic + pruning: + keep: + # source side pruning rules go here + ... - - type: pull - pruning: - keep_sender: - - type: not_replicated - ... - keep_receiver: - ... + - type: source + snapshotting: + type: manual + root_fs: ... + + # pull side + jobs: + - type: pull + pruning: + keep_sender: + # let the source-side snap job do the pruning + - type: regex + regex: ".*" + ... + keep_receiver: + # feel free to prune on the pull side as desired + ... From 34052d98d6c2cf30b6df03423f03c59b441d1403 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Fri, 15 Mar 2019 21:55:49 +0100 Subject: [PATCH 20/27] docs: move snap job below all replication-related job types --- docs/configuration/jobs.rst | 31 ++++++++++++++----------------- 1 file changed, 14 insertions(+), 17 deletions(-) diff --git a/docs/configuration/jobs.rst b/docs/configuration/jobs.rst index 515e46e..3468c98 100644 --- a/docs/configuration/jobs.rst +++ b/docs/configuration/jobs.rst @@ -262,12 +262,23 @@ Job Type ``source`` Example config: :sampleconf:`/source.yml` -.. _job-snap: -Job Type ``snap`` +.. _replication-local: + +Local replication ----------------- -Job type that only takes care of local snapshotting and pruning. +If you have the need for local replication (most likely between two local storage pools), you can use the :ref:`local transport type ` to connect a local push job to a local sink job. + +Example config: :sampleconf:`/local.yml`. + + +.. _job-snap: + +Job Type ``snap`` (snapshot & prune only) +----------------------------------------- + +Job type that only takes snapshots and pruning on the local machine. .. list-table:: :widths: 20 80 @@ -287,17 +298,3 @@ Job type that only takes care of local snapshotting and pruning. - |pruning-spec| Example config: :sampleconf:`/snap.yml` - - - - -.. _replication-local: - -Local replication ------------------ - -If you have the need for local replication (most likely between two local storage pools), you can use the :ref:`local transport type ` to connect a local push job to a local sink job. - -Example config: :sampleconf:`/local.yml`. - - From e8c0d206ea4d9573c55bde469a87218063d6625b Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Sun, 17 Mar 2019 20:54:47 +0100 Subject: [PATCH 21/27] docs: fix nitpicks --- docs/configuration/jobs.rst | 2 +- docs/configuration/prune.rst | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/configuration/jobs.rst b/docs/configuration/jobs.rst index cadf392..4e63fac 100644 --- a/docs/configuration/jobs.rst +++ b/docs/configuration/jobs.rst @@ -279,7 +279,7 @@ Example config: :sampleconf:`/local.yml`. Job Type ``snap`` (snapshot & prune only) ----------------------------------------- -Job type that only takes snapshots and pruning on the local machine. +Job type that only takes snapshots and performs pruning on the local machine. .. list-table:: :widths: 20 80 diff --git a/docs/configuration/prune.rst b/docs/configuration/prune.rst index 3d5d6e8..a725f91 100644 --- a/docs/configuration/prune.rst +++ b/docs/configuration/prune.rst @@ -178,7 +178,7 @@ Workaround using ``snap`` job As a workaround (see GitHub :issue:`102` for development progress), a pruning-only :ref:`snap job ` can be defined on the source side: The snap job is in charge of snapshot creation & destruction, whereas the source job's role is reduced to just serving snapshots. However, since, jobs are run independently, it is possible that the snap job will prune snapshots that are queued for replication / destruction by the remote pull job that connects to the source job. -There is no possible configuration to avoid this problem with the current version of zrepl. +Symptoms of such race conditions are spurious replication and destroy errors. Example configuration: @@ -198,7 +198,7 @@ Example configuration: snapshotting: type: manual root_fs: ... - + # pull side jobs: - type: pull From d8d9e3491409fc784afd5b3092a77168d8880e59 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Sun, 17 Mar 2019 20:57:34 +0100 Subject: [PATCH 22/27] pruner: single: remove unused member considerSnapAtCursorReplicated --- daemon/pruner/pruner.go | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/daemon/pruner/pruner.go b/daemon/pruner/pruner.go index f5de9bf..d44cbeb 100644 --- a/daemon/pruner/pruner.go +++ b/daemon/pruner/pruner.go @@ -82,9 +82,8 @@ type PrunerFactory struct { } type SinglePrunerFactory struct { - keepRules []pruning.KeepRule - retryWait time.Duration - considerSnapAtCursorReplicated bool + keepRules []pruning.KeepRule + retryWait time.Duration promPruneSecs *prometheus.HistogramVec } @@ -106,18 +105,16 @@ func NewSinglePrunerFactory(in config.PruningLocal, promPruneSecs *prometheus.Hi if err != nil { return nil, errors.Wrap(err, "cannot build pruning rules") } - considerSnapAtCursorReplicated := false for _, r := range in.Keep { - knr, ok := r.Ret.(*config.PruneKeepNotReplicated) - if !ok { - continue + if _, ok := r.Ret.(*config.PruneKeepNotReplicated); ok { + // rule NotReplicated for a local pruner doesn't make sense + // because no replication happens with that job type + return nil, fmt.Errorf("single-site pruner cannot support `not_replicated` keep rule") } - considerSnapAtCursorReplicated = considerSnapAtCursorReplicated || !knr.KeepSnapshotAtCursor } f := &SinglePrunerFactory{ - keepRules: rules, - retryWait: envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10 * time.Second), - considerSnapAtCursorReplicated: considerSnapAtCursorReplicated, + keepRules: rules, + retryWait: envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10*time.Second), promPruneSecs: promPruneSecs, } return f, nil @@ -192,7 +189,7 @@ func (f *SinglePrunerFactory) BuildSinglePruner(ctx context.Context, target Targ receiver, f.keepRules, f.retryWait, - f.considerSnapAtCursorReplicated, + false, // considerSnapAtCursorReplicated is not relevant for local pruning f.promPruneSecs.WithLabelValues("local"), }, state: Plan, From 5cd2593f521fd5ee475a3ac6aff0c44f8690825f Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Sun, 17 Mar 2019 21:06:19 +0100 Subject: [PATCH 23/27] job: snap: workaround for replication cursor requirement --- daemon/job/snapjob.go | 52 ++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 51 insertions(+), 1 deletion(-) diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 15feb42..0f47d5e 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -2,6 +2,9 @@ package job import ( "context" + "fmt" + "sort" + "github.com/pkg/errors" "github.com/prometheus/client_golang/prometheus" "github.com/zrepl/zrepl/config" @@ -11,6 +14,7 @@ import ( "github.com/zrepl/zrepl/daemon/pruner" "github.com/zrepl/zrepl/daemon/snapper" "github.com/zrepl/zrepl/endpoint" + "github.com/zrepl/zrepl/replication/logic/pdu" "github.com/zrepl/zrepl/zfs" ) @@ -102,11 +106,57 @@ outer: } } +// Adaptor that implements pruner.History around a pruner.Target. +// The ReplicationCursor method is Get-op only and always returns +// the filesystem's most recent version's GUID. +// +// TODO: +// This is a work-around for the current package daemon/pruner +// and package pruning.Snapshot limitation: they require the +// `Replicated` getter method be present, but obviously, +// a local job like SnapJob can't deliver on that. +// But the pruner.Pruner gives up on an FS if no replication +// cursor is present, which is why this pruner returns the +// most recent filesystem version. +type alwaysUpToDateReplicationCursorHistory struct { + // the Target passed as Target to BuildSinglePruner + target pruner.Target +} + +var _ pruner.History = (*alwaysUpToDateReplicationCursorHistory)(nil) + +func (h alwaysUpToDateReplicationCursorHistory) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) { + if req.GetGet() == nil { + return nil, fmt.Errorf("unsupported ReplicationCursor request: SnapJob only supports GETting a (faked) cursor") + } + fsvReq := &pdu.ListFilesystemVersionsReq{ + Filesystem: req.GetFilesystem(), + } + res, err := h.target.ListFilesystemVersions(ctx, fsvReq) + if err != nil { + return nil, err + } + fsvs := res.GetVersions() + if len(fsvs) <= 0 { + return &pdu.ReplicationCursorRes{Result: &pdu.ReplicationCursorRes_Notexist{Notexist: true}}, nil + } + // always return must recent version + sort.Slice(fsvs, func(i, j int) bool { + return fsvs[i].CreateTXG < fsvs[j].CreateTXG + }) + mostRecent := fsvs[len(fsvs)-1] + return &pdu.ReplicationCursorRes{Result: &pdu.ReplicationCursorRes_Guid{Guid: mostRecent.GetGuid()}}, nil +} + +func (h alwaysUpToDateReplicationCursorHistory) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) { + return h.target.ListFilesystems(ctx, req) +} + func (j *SnapJob) doPrune(ctx context.Context) { log := GetLogger(ctx) ctx = logging.WithSubsystemLoggers(ctx, log) sender := endpoint.NewSender(j.fsfilter) - j.pruner = j.prunerFactory.BuildSinglePruner(ctx, sender, sender) + j.pruner = j.prunerFactory.BuildSinglePruner(ctx, sender, alwaysUpToDateReplicationCursorHistory{sender}) log.Info("start pruning") j.pruner.Prune() log.Info("finished pruning") From b25da7b9b08a958232e7ba535d095994df7c22ea Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Sun, 17 Mar 2019 21:07:42 +0100 Subject: [PATCH 24/27] job: snap: comment fix --- daemon/job/snapjob.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 0f47d5e..d081b91 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -25,7 +25,7 @@ type SnapJob struct { prunerFactory *pruner.SinglePrunerFactory - promPruneSecs *prometheus.HistogramVec // no labels! + promPruneSecs *prometheus.HistogramVec // labels: prune_side pruner *pruner.Pruner } From 158d1175e33b9dc7e6105d76d658b553cc3583d8 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Sun, 17 Mar 2019 21:18:25 +0100 Subject: [PATCH 25/27] rename SinglePruner to LocalPruner --- daemon/job/snapjob.go | 8 ++++---- daemon/pruner/pruner.go | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index d081b91..b08a17d 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -23,7 +23,7 @@ type SnapJob struct { fsfilter zfs.DatasetFilter snapper *snapper.PeriodicOrManual - prunerFactory *pruner.SinglePrunerFactory + prunerFactory *pruner.LocalPrunerFactory promPruneSecs *prometheus.HistogramVec // labels: prune_side @@ -53,7 +53,7 @@ func snapJobFromConfig(g *config.Global, in *config.SnapJob) (j *SnapJob, err er Help: "seconds spent in pruner", ConstLabels: prometheus.Labels{"zrepl_job": j.name}, }, []string{"prune_side"}) - j.prunerFactory, err = pruner.NewSinglePrunerFactory(in.Pruning, j.promPruneSecs) + j.prunerFactory, err = pruner.NewLocalPrunerFactory(in.Pruning, j.promPruneSecs) if err != nil { return nil, errors.Wrap(err, "cannot build snapjob pruning rules") } @@ -119,7 +119,7 @@ outer: // cursor is present, which is why this pruner returns the // most recent filesystem version. type alwaysUpToDateReplicationCursorHistory struct { - // the Target passed as Target to BuildSinglePruner + // the Target passed as Target to BuildLocalPruner target pruner.Target } @@ -156,7 +156,7 @@ func (j *SnapJob) doPrune(ctx context.Context) { log := GetLogger(ctx) ctx = logging.WithSubsystemLoggers(ctx, log) sender := endpoint.NewSender(j.fsfilter) - j.pruner = j.prunerFactory.BuildSinglePruner(ctx, sender, alwaysUpToDateReplicationCursorHistory{sender}) + j.pruner = j.prunerFactory.BuildLocalPruner(ctx, sender, alwaysUpToDateReplicationCursorHistory{sender}) log.Info("start pruning") j.pruner.Prune() log.Info("finished pruning") diff --git a/daemon/pruner/pruner.go b/daemon/pruner/pruner.go index d44cbeb..1bd41e0 100644 --- a/daemon/pruner/pruner.go +++ b/daemon/pruner/pruner.go @@ -81,7 +81,7 @@ type PrunerFactory struct { promPruneSecs *prometheus.HistogramVec } -type SinglePrunerFactory struct { +type LocalPrunerFactory struct { keepRules []pruning.KeepRule retryWait time.Duration promPruneSecs *prometheus.HistogramVec @@ -100,7 +100,7 @@ func checkContainsKeep1(rules []pruning.KeepRule) error { return errors.New("sender keep rules must contain last_n or be empty so that the last snapshot is definitely kept") } -func NewSinglePrunerFactory(in config.PruningLocal, promPruneSecs *prometheus.HistogramVec) (*SinglePrunerFactory, error) { +func NewLocalPrunerFactory(in config.PruningLocal, promPruneSecs *prometheus.HistogramVec) (*LocalPrunerFactory, error) { rules, err := pruning.RulesFromConfig(in.Keep) if err != nil { return nil, errors.Wrap(err, "cannot build pruning rules") @@ -112,7 +112,7 @@ func NewSinglePrunerFactory(in config.PruningLocal, promPruneSecs *prometheus.Hi return nil, fmt.Errorf("single-site pruner cannot support `not_replicated` keep rule") } } - f := &SinglePrunerFactory{ + f := &LocalPrunerFactory{ keepRules: rules, retryWait: envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10*time.Second), promPruneSecs: promPruneSecs, @@ -181,7 +181,7 @@ func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, return p } -func (f *SinglePrunerFactory) BuildSinglePruner(ctx context.Context, target Target, receiver History) *Pruner { +func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, receiver History) *Pruner { p := &Pruner{ args: args{ ctx, From dd673bf923e71756781a464c5c23e0deb7bd6d36 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Sun, 17 Mar 2019 21:29:09 +0100 Subject: [PATCH 26/27] docs: condense snap job overview table row --- docs/configuration/jobs.rst | 34 ++++++++++++++++------------------ docs/configuration/prune.rst | 1 + 2 files changed, 17 insertions(+), 18 deletions(-) diff --git a/docs/configuration/jobs.rst b/docs/configuration/jobs.rst index 4e63fac..63fccb7 100644 --- a/docs/configuration/jobs.rst +++ b/docs/configuration/jobs.rst @@ -23,24 +23,22 @@ For communication, the active side connects to the passive side using a :ref:`tr The following table shows how different job types can be combined to achieve both push and pull mode setups: -+-----------------------+--------------+----------------------------------+-----------------------------------------------+ -| Setup name | active side | passive side | use case | -+=======================+==============+==================================+===============================================+ -| Push mode | ``push`` | ``sink`` | * Laptop backup | -| | | | * NAS behind NAT to offsite | -+-----------------------+--------------+----------------------------------+-----------------------------------------------+ -| Pull mode | ``pull`` | ``source`` | * Central backup-server for many nodes | -| | | | * Remote server to NAS behind NAT | -+-----------------------+--------------+----------------------------------+-----------------------------------------------+ -| Local replication | | ``push`` + ``sink`` in one config | * Backup FreeBSD boot pool | -| | | with :ref:`local transport ` | | -+-----------------------+--------------+----------------------------------+-----------------------------------------------+ -| SnapJob | ``snap`` | N/A | * Data requires versioning but no backups | -| | | | * Combined with a ``push`` or ``pull`` job: | -| | | | | -| | | | * High frequency local snapshotting | -| | | | * Low frequency replication | -+-----------------------+--------------+----------------------------------+-----------------------------------------------+ ++-----------------------+--------------+----------------------------------+------------------------------------------------------------------------------------+ +| Setup name | active side | passive side | use case | ++=======================+==============+==================================+====================================================================================+ +| Push mode | ``push`` | ``sink`` | * Laptop backup | +| | | | * NAS behind NAT to offsite | ++-----------------------+--------------+----------------------------------+------------------------------------------------------------------------------------+ +| Pull mode | ``pull`` | ``source`` | * Central backup-server for many nodes | +| | | | * Remote server to NAS behind NAT | ++-----------------------+--------------+----------------------------------+------------------------------------------------------------------------------------+ +| Local replication | | ``push`` + ``sink`` in one config | * Backup FreeBSD boot pool | +| | | with :ref:`local transport ` | | ++-----------------------+--------------+----------------------------------+------------------------------------------------------------------------------------+ +| Snap & prune-only | ``snap`` | N/A | * | Snapshots & pruning but no replication | +| | | | | required | +| | | | * Workaround for :ref:`source-side pruning ` | ++-----------------------+--------------+----------------------------------+------------------------------------------------------------------------------------+ How the Active Side Works ~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/docs/configuration/prune.rst b/docs/configuration/prune.rst index a725f91..d6ab343 100644 --- a/docs/configuration/prune.rst +++ b/docs/configuration/prune.rst @@ -158,6 +158,7 @@ Policy ``regex`` Like all other regular expression fields in prune policies, zrepl uses Go's `regexp.Regexp `_ Perl-compatible regular expressions (`Syntax `_). The optional `negate` boolean field inverts the semantics: Use it if you want to keep all snapshots that *do not* match the given regex. +.. _prune-workaround-source-side-pruning: Source-side snapshot pruning ---------------------------- From 283c2821cde30589d788443b9c3e96e5e59b0282 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Mon, 18 Mar 2019 14:53:12 +0100 Subject: [PATCH 27/27] docs: add SnapJob to 0.1 changelog --- docs/changelog.rst | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/changelog.rst b/docs/changelog.rst index 672cd61..996d0d7 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -92,6 +92,7 @@ Changes * |break_config| ``keep_bookmarks`` parameter of the ``grid`` keep rule has been removed * |feature| ``zrepl status`` for live-updating replication progress (it's really cool!) +* |feature| :ref:`Snapshot- & pruning-only job type ` (for local snapshot management) * |feature| :issue:`67`: Expose `Prometheus `_ metrics via HTTP (:ref:`config docs `) * |break_config| Logging outlet types must be specified using the ``type`` instead of ``outlet`` key