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