Compare commits

..

1 Commits

Author SHA1 Message Date
Christian Schwarz ffb1d89a72 config: support zrepl's day and week units for snapshotting.interval
Originally, I had a patch that would replace all usages of
time.Duration in package config with the new config.Duration
types, but:
1. these are all timeouts/retry intervals that have default values.
   Most users don't touch them, and if they do, they don't need
   day or week units.
2. go-yaml's error reporting for yaml.Unmarshaler is inferior to
   built-in types (line numbers are missing, so the error would not have
   sufficient context)

fixes https://github.com/zrepl/zrepl/issues/486
2022-10-09 21:15:46 +02:00
6 changed files with 198 additions and 189 deletions
+2 -45
View File
@@ -6,8 +6,6 @@ import (
"log/syslog" "log/syslog"
"os" "os"
"reflect" "reflect"
"regexp"
"strconv"
"time" "time"
"github.com/pkg/errors" "github.com/pkg/errors"
@@ -190,13 +188,10 @@ func (i *PositiveDurationOrManual) UnmarshalYAML(u func(interface{}, bool) error
return fmt.Errorf("value must not be empty") return fmt.Errorf("value must not be empty")
default: default:
i.Manual = false i.Manual = false
i.Interval, err = time.ParseDuration(s) i.Interval, err = parsePositiveDuration(s)
if err != nil { if err != nil {
return err return err
} }
if i.Interval <= 0 {
return fmt.Errorf("value must be a positive duration, got %q", s)
}
} }
return nil return nil
} }
@@ -230,7 +225,7 @@ type SnapshottingEnum struct {
type SnapshottingPeriodic struct { type SnapshottingPeriodic struct {
Type string `yaml:"type"` Type string `yaml:"type"`
Prefix string `yaml:"prefix"` Prefix string `yaml:"prefix"`
Interval time.Duration `yaml:"interval,positive"` Interval *PositiveDuration `yaml:"interval"`
Hooks HookList `yaml:"hooks,optional"` Hooks HookList `yaml:"hooks,optional"`
} }
@@ -715,41 +710,3 @@ func ParseConfigBytes(bytes []byte) (*Config, error) {
} }
return c, nil return c, nil
} }
var durationStringRegex *regexp.Regexp = regexp.MustCompile(`^\s*(\d+)\s*(s|m|h|d|w)\s*$`)
func parsePositiveDuration(e string) (d time.Duration, err error) {
comps := durationStringRegex.FindStringSubmatch(e)
if len(comps) != 3 {
err = fmt.Errorf("does not match regex: %s %#v", e, comps)
return
}
durationFactor, err := strconv.ParseInt(comps[1], 10, 64)
if err != nil {
return 0, err
}
if durationFactor <= 0 {
return 0, errors.New("duration must be positive integer")
}
var durationUnit time.Duration
switch comps[2] {
case "s":
durationUnit = time.Second
case "m":
durationUnit = time.Minute
case "h":
durationUnit = time.Hour
case "d":
durationUnit = 24 * time.Hour
case "w":
durationUnit = 24 * 7 * time.Hour
default:
err = fmt.Errorf("contains unknown time unit '%s'", comps[2])
return
}
d = time.Duration(durationFactor) * durationUnit
return
}
+106
View File
@@ -0,0 +1,106 @@
package config
import (
"errors"
"fmt"
"regexp"
"strconv"
"time"
"github.com/kr/pretty"
"github.com/zrepl/yaml-config"
)
type Duration struct{ d time.Duration }
func (d Duration) Duration() time.Duration { return d.d }
var _ yaml.Unmarshaler = &Duration{}
func (d *Duration) UnmarshalYAML(unmarshal func(v interface{}, not_strict bool) error) error {
var s string
err := unmarshal(&s, false)
if err != nil {
return err
}
d.d, err = parseDuration(s)
if err != nil {
d.d = 0
return &yaml.TypeError{Errors: []string{fmt.Sprintf("cannot parse value %q: %s", s, err)}}
}
return nil
}
type PositiveDuration struct{ d Duration }
var _ yaml.Unmarshaler = &PositiveDuration{}
func (d PositiveDuration) Duration() time.Duration { return d.d.Duration() }
func (d *PositiveDuration) UnmarshalYAML(unmarshal func(v interface{}, not_strict bool) error) error {
err := d.d.UnmarshalYAML(unmarshal)
if err != nil {
return err
}
if d.d.Duration() <= 0 {
return fmt.Errorf("duration must be positive, got %s", d.d.Duration())
}
return nil
}
func parsePositiveDuration(e string) (time.Duration, error) {
d, err := parseDuration(e)
if err != nil {
return d, err
}
if d <= 0 {
return 0, errors.New("duration must be positive integer")
}
return d, err
}
var durationStringRegex *regexp.Regexp = regexp.MustCompile(`^\s*([\+-]?\d+)\s*(|s|m|h|d|w)\s*$`)
func parseDuration(e string) (d time.Duration, err error) {
comps := durationStringRegex.FindStringSubmatch(e)
if comps == nil {
err = fmt.Errorf("must match %s", durationStringRegex)
return
}
if len(comps) != 3 {
panic(pretty.Sprint(comps))
}
durationFactor, err := strconv.ParseInt(comps[1], 10, 64)
if err != nil {
return 0, err
}
var durationUnit time.Duration
switch comps[2] {
case "":
if durationFactor != 0 {
err = fmt.Errorf("missing time unit")
return
} else {
// It's the case where user specified '0'.
// We want to allow this, just like time.ParseDuration.
}
case "s":
durationUnit = time.Second
case "m":
durationUnit = time.Minute
case "h":
durationUnit = time.Hour
case "d":
durationUnit = 24 * time.Hour
case "w":
durationUnit = 24 * 7 * time.Hour
default:
err = fmt.Errorf("contains unknown time unit '%s'", comps[2])
return
}
d = time.Duration(durationFactor) * durationUnit
return
}
+16 -1
View File
@@ -38,6 +38,13 @@ jobs:
interval: 10m interval: 10m
` `
periodicDaily := `
snapshotting:
type: periodic
prefix: zrepl_
interval: 1d
`
hooks := ` hooks := `
snapshotting: snapshotting:
type: periodic type: periodic
@@ -74,7 +81,15 @@ jobs:
c = testValidConfig(t, fillSnapshotting(periodic)) c = testValidConfig(t, fillSnapshotting(periodic))
snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingPeriodic) snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingPeriodic)
assert.Equal(t, "periodic", snp.Type) assert.Equal(t, "periodic", snp.Type)
assert.Equal(t, 10*time.Minute, snp.Interval) assert.Equal(t, 10*time.Minute, snp.Interval.Duration())
assert.Equal(t, "zrepl_", snp.Prefix)
})
t.Run("periodicDaily", func(t *testing.T) {
c = testValidConfig(t, fillSnapshotting(periodicDaily))
snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingPeriodic)
assert.Equal(t, "periodic", snp.Type)
assert.Equal(t, 24*time.Hour, snp.Interval.Duration())
assert.Equal(t, "zrepl_", snp.Prefix) assert.Equal(t, "zrepl_", snp.Prefix)
}) })
+25 -4
View File
@@ -10,7 +10,6 @@ import (
"github.com/zrepl/zrepl/config" "github.com/zrepl/zrepl/config"
"github.com/zrepl/zrepl/daemon/hooks" "github.com/zrepl/zrepl/daemon/hooks"
"github.com/zrepl/zrepl/util/suspendresumesafetimer"
"github.com/zrepl/zrepl/zfs" "github.com/zrepl/zrepl/zfs"
) )
@@ -43,17 +42,38 @@ type Cron struct {
func (s *Cron) Run(ctx context.Context, snapshotsTaken chan<- struct{}) { func (s *Cron) Run(ctx context.Context, snapshotsTaken chan<- struct{}) {
t := time.NewTimer(0)
defer func() {
if !t.Stop() {
select {
case <-t.C:
default:
}
}
}()
for { for {
now := time.Now() now := time.Now()
s.mtx.Lock() s.mtx.Lock()
s.wakeupTime = s.config.Cron.Schedule.Next(now) s.wakeupTime = s.config.Cron.Schedule.Next(now)
s.mtx.Unlock() s.mtx.Unlock()
ctxDone := suspendresumesafetimer.SleepUntil(ctx, s.wakeupTime) // Re-arm the timer.
if ctxDone != nil { // Need to Stop before Reset, see docs.
return if !t.Stop() {
// Use non-blocking read from timer channel
// because, except for the first loop iteration,
// the channel is already drained
select {
case <-t.C:
default:
} }
}
t.Reset(s.wakeupTime.Sub(now))
select {
case <-ctx.Done():
return
case <-t.C:
getLogger(ctx).Debug("cron timer fired") getLogger(ctx).Debug("cron timer fired")
s.mtx.Lock() s.mtx.Lock()
if s.running { if s.running {
@@ -83,6 +103,7 @@ func (s *Cron) Run(ctx context.Context, snapshotsTaken chan<- struct{}) {
} }
}() }()
} }
}
} }
+17 -11
View File
@@ -15,7 +15,6 @@ import (
"github.com/zrepl/zrepl/daemon/hooks" "github.com/zrepl/zrepl/daemon/hooks"
"github.com/zrepl/zrepl/daemon/logging" "github.com/zrepl/zrepl/daemon/logging"
"github.com/zrepl/zrepl/util/envconst" "github.com/zrepl/zrepl/util/envconst"
"github.com/zrepl/zrepl/util/suspendresumesafetimer"
"github.com/zrepl/zrepl/zfs" "github.com/zrepl/zrepl/zfs"
) )
@@ -23,7 +22,7 @@ func periodicFromConfig(g *config.Global, fsf zfs.DatasetFilter, in *config.Snap
if in.Prefix == "" { if in.Prefix == "" {
return nil, errors.New("prefix must not be empty") return nil, errors.New("prefix must not be empty")
} }
if in.Interval <= 0 { if in.Interval.Duration() <= 0 {
return nil, errors.New("interval must be positive") return nil, errors.New("interval must be positive")
} }
@@ -33,7 +32,7 @@ func periodicFromConfig(g *config.Global, fsf zfs.DatasetFilter, in *config.Snap
} }
args := periodicArgs{ args := periodicArgs{
interval: in.Interval, interval: in.Interval.Duration(),
fsf: fsf, fsf: fsf,
planArgs: planArgs{ planArgs: planArgs{
prefix: in.Prefix, prefix: in.Prefix,
@@ -172,13 +171,16 @@ func periodicStateSyncUp(a periodicArgs, u updater) state {
u(func(s *Periodic) { u(func(s *Periodic) {
s.sleepUntil = syncPoint s.sleepUntil = syncPoint
}) })
ctxDone := suspendresumesafetimer.SleepUntil(a.ctx, syncPoint) t := time.NewTimer(time.Until(syncPoint))
if ctxDone != nil { defer t.Stop()
return onMainCtxDone(a.ctx, u) select {
} case <-t.C:
return u(func(s *Periodic) { return u(func(s *Periodic) {
s.state = Planning s.state = Planning
}).sf() }).sf()
case <-a.ctx.Done():
return onMainCtxDone(a.ctx, u)
}
} }
func periodicStatePlan(a periodicArgs, u updater) state { func periodicStatePlan(a periodicArgs, u updater) state {
@@ -239,13 +241,17 @@ func periodicStateWait(a periodicArgs, u updater) state {
logFunc("enter wait-state after error") logFunc("enter wait-state after error")
}) })
ctxDone := suspendresumesafetimer.SleepUntil(a.ctx, sleepUntil) t := time.NewTimer(time.Until(sleepUntil))
if ctxDone != nil { defer t.Stop()
return onMainCtxDone(a.ctx, u)
} select {
case <-t.C:
return u(func(snapper *Periodic) { return u(func(snapper *Periodic) {
snapper.state = Planning snapper.state = Planning
}).sf() }).sf()
case <-a.ctx.Done():
return onMainCtxDone(a.ctx, u)
}
} }
func listFSes(ctx context.Context, mf zfs.DatasetFilter) (fss []*zfs.DatasetPath, err error) { func listFSes(ctx context.Context, mf zfs.DatasetFilter) (fss []*zfs.DatasetPath, err error) {
@@ -1,96 +0,0 @@
package suspendresumesafetimer
import (
"context"
"time"
"github.com/zrepl/zrepl/util/envconst"
)
// The returned error is guaranteed to be the ctx.Err()
func SleepUntil(ctx context.Context, sleepUntil time.Time) error {
// We use .Round(0) to strip the monotonic clock reading from the time.Time
// returned by time.Now(). That will make the before/after check in the ticker
// for-loop compare wall-clock times instead of monotonic time.
// Comparing wall clock time is necessary because monotonic time does not progress
// while the system is suspended.
//
// Background
//
// A time.Time carries a wallclock timestamp and optionally a monotonic clock timestamp.
// time.Now() returns a time.Time that carries both.
// time.Time.Add() applies the same delta to both timestamps in the time.Time.
// x.Sub(y) will return the *monotonic* delta if both x and y carry a monotonic timestamp.
// time.Until(x) == x.Sub(now) where `now` will have a monotonic timestamp.
// So, time.Until(x) with an `x` that has monotonic timestamp will return monotonic delta.
//
// Why Do We Care?
//
// On systems that suspend/resume, wall clock time progresses during suspend but
// monotonic time does not.
//
// So, suppose the following sequence of events:
// x <== time.Now()
// System suspends for 1 hour
// delta <== time.Now().Sub(x)
// `delta` will be near 0 because time.Until() subtracts the monotonic
// timestamps, and monotonic time didn't progress during suspend.
//
// Now strip the timestamp using .Round(0)
// x <== time.Now().Round(0)
// System suspends for 1 hour
// delta <== time.Now().Sub(x)
// `delta` will be 1 hour because time.Sub() subtracted wallclock timestamps
// because x didn't have a monotonic timestamp because we stripped it using .Round(0).
//
//
sleepUntil = sleepUntil.Round(0)
// Set up a timer so that, if the system doesn't suspend/resume,
// we get a precise wake-up time from the native Go timer.
monotonicClockTimer := time.NewTimer(time.Until(sleepUntil))
defer func() {
if !monotonicClockTimer.Stop() {
// non-blocking read since we can come here when
// we've already drained the channel through
// case <-monotonicClockTimer.C
// in the `for` loop below.
select {
case <-monotonicClockTimer.C:
default:
}
}
}()
// Set up a ticker so that we're guaranteed to wake up periodically.
// We'll then get the current wall-clock time and check ourselves
// whether we're past the requested expiration time.
// Pick a 10 second check interval by default since it's rare that
// suspend/resume is done more frequently.
ticker := time.NewTicker(envconst.Duration("ZREPL_WALLCLOCKTIMER_MAX_DELAY", 10*time.Second))
defer ticker.Stop()
for {
select {
case <-monotonicClockTimer.C:
return nil
case <-ticker.C:
now := time.Now()
if now.Before(sleepUntil) {
// Continue waiting.
// Reset the monotonic timer to reset drift.
if !monotonicClockTimer.Stop() {
<-monotonicClockTimer.C
}
monotonicClockTimer.Reset(time.Until(sleepUntil))
continue
}
return nil
case <-ctx.Done():
return ctx.Err()
}
}
}