Merge branch 'InsanePrawn-master' into 'master'

2019-03-20 19:45:00 +01:00
parent f2a6193735 283c2821cd
commit 6f7467e8d8
10 changed files with 380 additions and 50 deletions
@@ -45,6 +45,11 @@ func buildJob(c *config.Global, in config.JobEnum) (j Job, err error) {
 		if err != nil {
 			return cannotBuildJob(err, v.Name)
 		}
+	case *config.SnapJob:
+		j, err = snapJobFromConfig(c, v)
+		if err != nil {
+			return cannotBuildJob(err, v.Name)
+		}
 	case *config.PushJob:
 		m, err := modePushFromConfig(c, v)
 		if err != nil {
@@ -39,6 +39,7 @@ type Type string

 const (
 	TypeInternal Type = "internal"
+	TypeSnap Type = "snap"
 	TypePush Type = "push"
 	TypeSink Type = "sink"
 	TypePull Type  = "pull"
@@ -84,6 +85,10 @@ func (s *Status) UnmarshalJSON(in []byte) (err error) {
 		return fmt.Errorf("field '%s', not found", key)
 	}
 	switch s.Type {
+	case TypeSnap:
+		var st SnapJobStatus
+		err = json.Unmarshal(jobJSON, &st)
+		s.JobSpecific = &st
 	case TypePull: fallthrough
 	case TypePush:
 		var st ActiveSideStatus
@@ -0,0 +1,163 @@
+package job
+
+import (
+	"context"
+	"fmt"
+	"sort"
+
+	"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/replication/logic/pdu"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+type SnapJob struct {
+	name     string
+	fsfilter zfs.DatasetFilter
+	snapper  *snapper.PeriodicOrManual
+
+	prunerFactory *pruner.LocalPrunerFactory
+
+	promPruneSecs *prometheus.HistogramVec // labels: prune_side
+
+	pruner *pruner.Pruner
+}
+
+func (j *SnapJob) Name() string { return j.name }
+
+func (j *SnapJob) Type() Type { return TypeSnap }
+
+func snapJobFromConfig(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.NewLocalPrunerFactory(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.snapper.Run(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))
+	}
+}
+
+// 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 BuildLocalPruner
+	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.BuildLocalPruner(ctx, sender, alwaysUpToDateReplicationCursorHistory{sender})
+	log.Info("start pruning")
+	j.pruner.Prune()
+	log.Info("finished pruning")
+}
@@ -78,6 +78,12 @@ type PrunerFactory struct {
 	promPruneSecs *prometheus.HistogramVec
 }

+type LocalPrunerFactory struct {
+	keepRules     []pruning.KeepRule
+	retryWait     time.Duration
+	promPruneSecs *prometheus.HistogramVec
+}
+
 func checkContainsKeep1(rules []pruning.KeepRule) error {
 	if len(rules) == 0 {
 		return nil //No keep rules means keep all - ok
@@ -91,6 +97,26 @@ 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 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")
+	}
+	for _, r := range in.Keep {
+		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")
+		}
+	}
+	f := &LocalPrunerFactory{
+		keepRules:     rules,
+		retryWait:     envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10*time.Second),
+		promPruneSecs: promPruneSecs,
+	}
+	return f, nil
+}
+
 func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus.HistogramVec) (*PrunerFactory, error) {
 	keepRulesReceiver, err := pruning.RulesFromConfig(in.KeepReceiver)
 	if err != nil {
@@ -152,6 +178,22 @@ func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target,
 	return p
 }

+func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, receiver History) *Pruner {
+	p := &Pruner{
+		args: args{
+			ctx,
+			target,
+			receiver,
+			f.keepRules,
+			f.retryWait,
+			false, // considerSnapAtCursorReplicated is not relevant for local pruning
+			f.promPruneSecs.WithLabelValues("local"),
+		},
+		state: Plan,
+	}
+	return p
+}
+
 //go:generate enumer -type=State
 type State int