mirror of
https://github.com/zrepl/zrepl.git
synced 2025-02-02 03:29:13 +01:00
pruned unused lines and comments ;)
This commit is contained in:
InsanePrawn
parent
3d2688e959
commit
1265cc7934
@ -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
|
||||
})
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user