extern/zrepl
1
0
Fork 0
mirror of https://github.com/zrepl/zrepl.git synced 2025-06-20 01:37:45 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

job local: refactor + use Task API

Browse Source 
refs #10
This commit is contained in:
Christian Schwarz 2017-12-26 22:48:42 +01:00
parent 63fa7a67e9
commit d13c6e3fc3
1 changed files with 30 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
cmd/config_job_local.go
View File

@ -21,7 +21,8 @@ type LocalJob struct {
PruneRHS PrunePolicy PruneRHS PrunePolicy
Debug JobDebugSettings Debug JobDebugSettings
snapperTask *Task snapperTask *Task
replTask *Task mainTask *Task
handlerTask *Task
pruneRHSTask *Task pruneRHSTask *Task
pruneLHSTask *Task pruneLHSTask *Task
} }
@ -85,13 +86,13 @@ func (j *LocalJob) JobName() string {
func (j *LocalJob) JobStart(ctx context.Context) { func (j *LocalJob) JobStart(ctx context.Context) {
log := ctx.Value(contextKeyLog).(Logger) rootLog := ctx.Value(contextKeyLog).(Logger)
defer log.Info("exiting")
j.snapperTask = NewTask("snapshot", log) j.snapperTask = NewTask("snapshot", rootLog)
j.replTask = NewTask("repl", log) j.mainTask = NewTask("main", rootLog)
j.pruneRHSTask = NewTask("prune_rhs", log) j.handlerTask = NewTask("handler", rootLog)
j.pruneLHSTask = NewTask("prune_lhs", log) j.pruneRHSTask = NewTask("prune_rhs", rootLog)
j.pruneLHSTask = NewTask("prune_lhs", rootLog)
local := rpc.NewLocalRPC() local := rpc.NewLocalRPC()
// Allow access to any dataset since we control what mapping // Allow access to any dataset since we control what mapping
@ -99,7 +100,7 @@ func (j *LocalJob) JobStart(ctx context.Context) {
// All local datasets will be passed to its Map() function, // All local datasets will be passed to its Map() function,
// but only those for which a mapping exists will actually be pulled. // but only those for which a mapping exists will actually be pulled.
// We can pay this small performance penalty for now. // We can pay this small performance penalty for now.
handler := NewHandler(log.WithField(logTaskField, "handler"), localPullACL{}, NewPrefixFilter(j.SnapshotPrefix)) handler := NewHandler(j.handlerTask.Log(), localPullACL{}, NewPrefixFilter(j.SnapshotPrefix))
registerEndpoints(local, handler) registerEndpoints(local, handler)
@ -112,43 +113,35 @@ func (j *LocalJob) JobStart(ctx context.Context) {
plhs, err := j.Pruner(j.pruneLHSTask, PrunePolicySideLeft, false) plhs, err := j.Pruner(j.pruneLHSTask, PrunePolicySideLeft, false)
if err != nil { if err != nil {
log.WithError(err).Error("error creating lhs pruner") rootLog.WithError(err).Error("error creating lhs pruner")
return return
} }
prhs, err := j.Pruner(j.pruneRHSTask, PrunePolicySideRight, false) prhs, err := j.Pruner(j.pruneRHSTask, PrunePolicySideRight, false)
if err != nil { if err != nil {
log.WithError(err).Error("error creating rhs pruner") rootLog.WithError(err).Error("error creating rhs pruner")
return return
} }
makeCtx := func(parent context.Context, taskName string) (ctx context.Context) {
return context.WithValue(parent, contextKeyLog, log.WithField(logTaskField, taskName))
}
var snapCtx, plCtx, prCtx, pullCtx context.Context
snapCtx = makeCtx(ctx, "autosnap")
plCtx = makeCtx(ctx, "prune_lhs")
prCtx = makeCtx(ctx, "prune_rhs")
pullCtx = makeCtx(ctx, "repl")
didSnaps := make(chan struct{}) didSnaps := make(chan struct{})
go snapper.Run(snapCtx, didSnaps) go snapper.Run(ctx, didSnaps)
outer: outer:
for { for {
select { select {
case <-ctx.Done(): case <-ctx.Done():
j.mainTask.Log().WithError(ctx.Err()).Info("context")
break outer break outer
case <-didSnaps: case <-didSnaps:
log.Debug("finished taking snapshots") j.mainTask.Log().Debug("finished taking snapshots")
log.Info("starting replication procedure") j.mainTask.Log().Info("starting replication procedure")
} }
{ j.mainTask.Log().Debug("replicating from lhs to rhs")
log := pullCtx.Value(contextKeyLog).(Logger) j.mainTask.Enter("replicate")
log.Debug("replicating from lhs to rhs") puller := Puller{j.mainTask, local, j.Mapping, j.InitialReplPolicy}
puller := Puller{j.replTask, local, j.Mapping, j.InitialReplPolicy}
puller.Pull() puller.Pull()
j.mainTask.Finish()
// use a ctx as soon as Pull gains ctx support // use a ctx as soon as Pull gains ctx support
select { select {
@ -156,21 +149,20 @@ outer:
break outer break outer
default: default:
} }
}
var wg sync.WaitGroup var wg sync.WaitGroup
log.Info("pruning lhs") j.mainTask.Log().Info("pruning lhs")
wg.Add(1) wg.Add(1)
go func() { go func() {
plhs.Run(plCtx) plhs.Run(ctx)
wg.Done() wg.Done()
}() }()
log.Info("pruning rhs") j.mainTask.Log().Info("pruning rhs")
wg.Add(1) wg.Add(1)
go func() { go func() {
prhs.Run(prCtx) prhs.Run(ctx)
wg.Done() wg.Done()
}() }()
@ -178,8 +170,6 @@ outer:
} }
log.WithError(ctx.Err()).Info("context")
} }
func (j *LocalJob) JobStatus(ctxt context.Context) (*JobStatus, error) { func (j *LocalJob) JobStatus(ctxt context.Context) (*JobStatus, error) {
@ -187,7 +177,7 @@ func (j *LocalJob) JobStatus(ctxt context.Context) (*JobStatus, error) {
j.snapperTask.Status(), j.snapperTask.Status(),
j.pruneLHSTask.Status(), j.pruneLHSTask.Status(),
j.pruneRHSTask.Status(), j.pruneRHSTask.Status(),
j.replTask.Status(), j.mainTask.Status(),
}}, nil }}, nil
} }