2017-09-10 16:13:05 +02:00
|
|
|
package cmd
|
|
|
|
|
|
|
|
import (
|
|
|
|
"time"
|
|
|
|
|
2017-09-13 23:27:18 +02:00
|
|
|
"context"
|
2017-09-10 16:13:05 +02:00
|
|
|
"github.com/mitchellh/mapstructure"
|
|
|
|
"github.com/pkg/errors"
|
2017-09-11 15:45:10 +02:00
|
|
|
"github.com/zrepl/zrepl/rpc"
|
|
|
|
"github.com/zrepl/zrepl/util"
|
2017-09-10 16:13:05 +02:00
|
|
|
)
|
|
|
|
|
|
|
|
type PullJob struct {
|
2017-09-13 23:27:18 +02:00
|
|
|
Name string
|
|
|
|
Connect RWCConnecter
|
|
|
|
Interval time.Duration
|
|
|
|
Mapping *DatasetMapFilter
|
|
|
|
// constructed from mapping during parsing
|
|
|
|
pruneFilter *DatasetMapFilter
|
|
|
|
SnapshotPrefix string
|
2017-09-10 16:13:05 +02:00
|
|
|
InitialReplPolicy InitialReplPolicy
|
|
|
|
Prune PrunePolicy
|
2017-09-11 15:45:10 +02:00
|
|
|
Debug JobDebugSettings
|
2017-12-26 19:36:27 +01:00
|
|
|
|
|
|
|
task *Task
|
2017-09-10 16:13:05 +02:00
|
|
|
}
|
|
|
|
|
2017-09-17 18:20:05 +02:00
|
|
|
func parsePullJob(c JobParsingContext, name string, i map[string]interface{}) (j *PullJob, err error) {
|
2017-09-10 16:13:05 +02:00
|
|
|
|
|
|
|
var asMap struct {
|
|
|
|
Connect map[string]interface{}
|
2017-09-13 23:27:18 +02:00
|
|
|
Interval string
|
2017-09-10 16:13:05 +02:00
|
|
|
Mapping map[string]string
|
|
|
|
InitialReplPolicy string `mapstructure:"initial_repl_policy"`
|
|
|
|
Prune map[string]interface{}
|
|
|
|
SnapshotPrefix string `mapstructure:"snapshot_prefix"`
|
2017-09-11 15:45:10 +02:00
|
|
|
Debug map[string]interface{}
|
2017-09-10 16:13:05 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
if err = mapstructure.Decode(i, &asMap); err != nil {
|
|
|
|
err = errors.Wrap(err, "mapstructure error")
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
j = &PullJob{Name: name}
|
|
|
|
|
|
|
|
j.Connect, err = parseSSHStdinserverConnecter(asMap.Connect)
|
|
|
|
if err != nil {
|
|
|
|
err = errors.Wrap(err, "cannot parse 'connect'")
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2017-10-05 15:12:50 +02:00
|
|
|
if j.Interval, err = parsePostitiveDuration(asMap.Interval); err != nil {
|
2017-09-13 23:27:18 +02:00
|
|
|
err = errors.Wrap(err, "cannot parse 'interval'")
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2017-09-10 16:13:05 +02:00
|
|
|
j.Mapping, err = parseDatasetMapFilter(asMap.Mapping, false)
|
|
|
|
if err != nil {
|
|
|
|
err = errors.Wrap(err, "cannot parse 'mapping'")
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2017-09-13 23:27:18 +02:00
|
|
|
if j.pruneFilter, err = j.Mapping.InvertedFilter(); err != nil {
|
|
|
|
err = errors.Wrap(err, "cannot automatically invert 'mapping' for prune job")
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2017-09-10 16:13:05 +02:00
|
|
|
j.InitialReplPolicy, err = parseInitialReplPolicy(asMap.InitialReplPolicy, DEFAULT_INITIAL_REPL_POLICY)
|
|
|
|
if err != nil {
|
|
|
|
err = errors.Wrap(err, "cannot parse 'initial_repl_policy'")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2017-09-13 23:27:18 +02:00
|
|
|
if j.SnapshotPrefix, err = parseSnapshotPrefix(asMap.SnapshotPrefix); err != nil {
|
2017-09-10 16:13:05 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
if j.Prune, err = parsePrunePolicy(asMap.Prune); err != nil {
|
|
|
|
err = errors.Wrap(err, "cannot parse prune policy")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2017-09-11 15:45:10 +02:00
|
|
|
if err = mapstructure.Decode(asMap.Debug, &j.Debug); err != nil {
|
|
|
|
err = errors.Wrap(err, "cannot parse 'debug'")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2017-09-10 16:13:05 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
func (j *PullJob) JobName() string {
|
|
|
|
return j.Name
|
|
|
|
}
|
|
|
|
|
2017-09-13 23:27:18 +02:00
|
|
|
func (j *PullJob) JobStart(ctx context.Context) {
|
2017-09-11 15:45:10 +02:00
|
|
|
|
2017-09-16 21:12:26 +02:00
|
|
|
log := ctx.Value(contextKeyLog).(Logger)
|
2017-09-23 17:52:29 +02:00
|
|
|
defer log.Info("exiting")
|
2017-12-26 19:36:27 +01:00
|
|
|
j.task = NewTask("main", log)
|
|
|
|
log = j.task.Log()
|
2017-09-16 21:12:26 +02:00
|
|
|
|
2017-09-13 23:27:18 +02:00
|
|
|
ticker := time.NewTicker(j.Interval)
|
|
|
|
|
|
|
|
start:
|
|
|
|
|
2017-09-23 17:52:29 +02:00
|
|
|
log.Info("connecting")
|
2017-09-11 15:45:10 +02:00
|
|
|
rwc, err := j.Connect.Connect()
|
2017-09-10 16:13:05 +02:00
|
|
|
if err != nil {
|
2017-09-23 17:52:29 +02:00
|
|
|
log.WithError(err).Error("error connecting")
|
2017-09-13 23:27:18 +02:00
|
|
|
return
|
2017-09-10 16:13:05 +02:00
|
|
|
}
|
2017-09-11 15:45:10 +02:00
|
|
|
|
|
|
|
rwc, err = util.NewReadWriteCloserLogger(rwc, j.Debug.Conn.ReadDump, j.Debug.Conn.WriteDump)
|
|
|
|
if err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
client := rpc.NewClient(rwc)
|
|
|
|
if j.Debug.RPC.Log {
|
|
|
|
client.SetLogger(log, true)
|
|
|
|
}
|
|
|
|
|
2017-09-23 17:52:29 +02:00
|
|
|
log.Info("starting pull")
|
2017-09-13 23:27:18 +02:00
|
|
|
|
2017-09-23 11:24:36 +02:00
|
|
|
pullLog := log.WithField(logTaskField, "pull")
|
2017-12-26 19:36:27 +01:00
|
|
|
puller := Puller{j.task, client, pullLog, j.Mapping, j.InitialReplPolicy}
|
2017-12-24 15:05:07 +01:00
|
|
|
if err = puller.doPull(); err != nil {
|
2017-09-23 17:52:29 +02:00
|
|
|
log.WithError(err).Error("error doing pull")
|
2017-09-13 23:27:18 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
closeRPCWithTimeout(log, client, time.Second*10, "")
|
|
|
|
|
2017-09-23 17:52:29 +02:00
|
|
|
log.Info("starting prune")
|
2017-09-23 11:24:36 +02:00
|
|
|
prunectx := context.WithValue(ctx, contextKeyLog, log.WithField(logTaskField, "prune"))
|
2017-12-26 19:36:27 +01:00
|
|
|
pruner, err := j.Pruner(j.task, PrunePolicySideDefault, false)
|
2017-09-16 21:12:26 +02:00
|
|
|
if err != nil {
|
2017-09-23 17:52:29 +02:00
|
|
|
log.WithError(err).Error("error creating pruner")
|
2017-09-16 21:12:26 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2017-09-13 23:27:18 +02:00
|
|
|
pruner.Run(prunectx)
|
2017-09-23 17:52:29 +02:00
|
|
|
log.Info("finish prune")
|
2017-09-13 23:27:18 +02:00
|
|
|
|
2017-09-23 17:52:29 +02:00
|
|
|
log.Info("wait for next interval")
|
2017-09-13 23:27:18 +02:00
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
2017-09-23 17:52:29 +02:00
|
|
|
log.WithError(ctx.Err()).Info("context")
|
2017-09-13 23:27:18 +02:00
|
|
|
return
|
|
|
|
case <-ticker.C:
|
|
|
|
goto start
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
2017-12-24 15:34:41 +01:00
|
|
|
func (j *PullJob) JobStatus(ctxt context.Context) (*JobStatus, error) {
|
2017-12-26 19:36:27 +01:00
|
|
|
return &JobStatus{Tasks: []*TaskStatus{j.task.Status()}}, nil
|
2017-12-24 15:34:41 +01:00
|
|
|
}
|
|
|
|
|
2017-12-26 19:36:27 +01:00
|
|
|
func (j *PullJob) Pruner(task *Task, side PrunePolicySide, dryRun bool) (p Pruner, err error) {
|
2017-09-16 21:12:26 +02:00
|
|
|
p = Pruner{
|
2017-12-26 19:36:27 +01:00
|
|
|
task,
|
2017-09-16 21:12:26 +02:00
|
|
|
time.Now(),
|
|
|
|
dryRun,
|
|
|
|
j.pruneFilter,
|
|
|
|
j.SnapshotPrefix,
|
|
|
|
j.Prune,
|
|
|
|
}
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2017-09-13 23:27:18 +02:00
|
|
|
func (j *PullJob) doRun(ctx context.Context) {
|
|
|
|
|
2017-09-10 16:13:05 +02:00
|
|
|
}
|