zrepl/daemon/snapper/cron.go

package snapper

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/pkg/errors"

	"github.com/zrepl/zrepl/config"
	"github.com/zrepl/zrepl/daemon/hooks"
	"github.com/zrepl/zrepl/util/suspendresumesafetimer"
	"github.com/zrepl/zrepl/zfs"
)

func cronFromConfig(fsf zfs.DatasetFilter, in config.SnapshottingCron) (*Cron, error) {

	hooksList, err := hooks.ListFromConfig(&in.Hooks)
	if err != nil {
		return nil, errors.Wrap(err, "hook config error")
	}
	planArgs := planArgs{
		prefix:          in.Prefix,
		timestampFormat: in.TimestampFormat,
		hooks:           hooksList,
	}
	return &Cron{config: in, fsf: fsf, planArgs: planArgs}, nil
}

type Cron struct {
	config   config.SnapshottingCron
	fsf      zfs.DatasetFilter
	planArgs planArgs

	mtx sync.RWMutex

	running                 bool
	wakeupTime              time.Time // zero value means uninit
	lastError               error
	lastPlan                *plan
	wakeupWhileRunningCount int
}

func (s *Cron) Run(ctx context.Context, snapshotsTaken chan<- struct{}) {

	for {
		now := time.Now()
		s.mtx.Lock()
		s.wakeupTime = s.config.Cron.Schedule.Next(now)
		s.mtx.Unlock()

		ctxDone := suspendresumesafetimer.SleepUntil(ctx, s.wakeupTime)
		if ctxDone != nil {
			return
		}

		getLogger(ctx).Debug("cron timer fired")
		s.mtx.Lock()
		if s.running {
			getLogger(ctx).Warn("snapshotting triggered according to cron rules but previous snapshotting is not done; not taking a snapshot this time")
			s.wakeupWhileRunningCount++
			s.mtx.Unlock()
			continue
		}
		s.lastError = nil
		s.lastPlan = nil
		s.wakeupWhileRunningCount = 0
		s.running = true
		s.mtx.Unlock()
		go func() {
			err := s.do(ctx)
			s.mtx.Lock()
			s.lastError = err
			s.running = false
			s.mtx.Unlock()

			select {
			case snapshotsTaken <- struct{}{}:
			default:
				if snapshotsTaken != nil {
					getLogger(ctx).Warn("callback channel is full, discarding snapshot update event")
				}
			}
		}()
	}

}

func (s *Cron) do(ctx context.Context) error {
	fss, err := zfs.ZFSListMapping(ctx, s.fsf)
	if err != nil {
		return errors.Wrap(err, "cannot list filesystems")
	}
	p := makePlan(s.planArgs, fss)

	s.mtx.Lock()
	s.lastPlan = p
	s.lastError = nil
	s.mtx.Unlock()

	ok := p.execute(ctx, false)
	if !ok {
		return errors.New("one or more snapshots could not be created, check logs for details")
	} else {
		return nil
	}
}

type CronState string

const (
	CronStateRunning CronState = "running"
	CronStateWaiting CronState = "waiting"
)

type CronReport struct {
	State      CronState
	WakeupTime time.Time
	Errors     []string
	Progress   []*ReportFilesystem
}

func (s *Cron) Report() Report {

	s.mtx.Lock()
	defer s.mtx.Unlock()

	r := CronReport{}

	r.WakeupTime = s.wakeupTime

	if s.running {
		r.State = CronStateRunning
	} else {
		r.State = CronStateWaiting
	}

	if s.lastError != nil {
		r.Errors = append(r.Errors, s.lastError.Error())
	}
	if s.wakeupWhileRunningCount > 0 {
		r.Errors = append(r.Errors, fmt.Sprintf("cron frequency is too high; snapshots were not taken %d times", s.wakeupWhileRunningCount))
	}

	r.Progress = nil
	if s.lastPlan != nil {
		r.Progress = s.lastPlan.report()
	}

	return Report{Type: TypeCron, Cron: &r}
}