Compare commits
14 Commits
go1.19-update
...
v0.6.0
| Author | SHA1 | Date | |
|---|---|---|---|
| 2658695a35 | |||
| 1ac1635b3d | |||
| 4a2806f6d1 | |||
| 0a264b9b41 | |||
| a3379d6785 | |||
| 6260b75031 | |||
| 3ffb69bfb0 | |||
| 1da8f848f2 | |||
| 6ed4626df9 | |||
| c07f9ec62e | |||
| fd5b0e6831 | |||
| a4cea1b4f3 | |||
| c0b52b92d5 | |||
| 12018b3685 |
+15
-13
@@ -24,7 +24,12 @@ commands:
|
||||
|
||||
apt-update-and-install-common-deps:
|
||||
steps:
|
||||
- run: sudo apt update && sudo apt install gawk make
|
||||
- run: sudo apt-get update
|
||||
- run: sudo apt-get install -y gawk make
|
||||
# CircleCI doesn't update its cimg/go images.
|
||||
# So, need to update manually to get up-to-date trust chains.
|
||||
# The need for this was required for cimg/go:1.12, but let's future proof this here and now.
|
||||
- run: sudo apt-get install -y git ca-certificates
|
||||
|
||||
restore-cache-gomod:
|
||||
steps:
|
||||
@@ -150,7 +155,7 @@ parameters:
|
||||
|
||||
release_docker_baseimage_tag:
|
||||
type: string
|
||||
default: "1.17"
|
||||
default: "1.19"
|
||||
|
||||
workflows:
|
||||
version: 2
|
||||
@@ -160,19 +165,19 @@ workflows:
|
||||
jobs:
|
||||
- quickcheck-docs
|
||||
- quickcheck-go: &quickcheck-go-smoketest
|
||||
name: quickcheck-go-amd64-linux-1.17
|
||||
goversion: &latest-go-release "1.17"
|
||||
name: quickcheck-go-amd64-linux-1.19
|
||||
goversion: &latest-go-release "1.19"
|
||||
goos: linux
|
||||
goarch: amd64
|
||||
- test-go-on-latest-go-release:
|
||||
goversion: *latest-go-release
|
||||
- quickcheck-go:
|
||||
requires:
|
||||
- quickcheck-go-amd64-linux-1.17 #quickcheck-go-smoketest.name
|
||||
- quickcheck-go-amd64-linux-1.19 #quickcheck-go-smoketest.name
|
||||
matrix: &quickcheck-go-matrix
|
||||
alias: quickcheck-go-matrix
|
||||
parameters:
|
||||
goversion: [*latest-go-release, "1.12"]
|
||||
goversion: [*latest-go-release, "1.18"]
|
||||
goos: ["linux", "freebsd"]
|
||||
goarch: ["amd64", "arm64"]
|
||||
exclude:
|
||||
@@ -180,10 +185,6 @@ workflows:
|
||||
- goversion: *latest-go-release
|
||||
goos: linux
|
||||
goarch: amd64
|
||||
# not supported by Go 1.12
|
||||
- goversion: "1.12"
|
||||
goos: freebsd
|
||||
goarch: arm64
|
||||
|
||||
release:
|
||||
when: << pipeline.parameters.do_release >>
|
||||
@@ -249,7 +250,7 @@ jobs:
|
||||
goarch:
|
||||
type: string
|
||||
docker:
|
||||
- image: circleci/golang:<<parameters.goversion>>
|
||||
- image: cimg/go:<<parameters.goversion>>
|
||||
environment:
|
||||
GOOS: <<parameters.goos>>
|
||||
GOARCH: <<parameters.goarch>>
|
||||
@@ -257,12 +258,13 @@ jobs:
|
||||
steps:
|
||||
- checkout
|
||||
|
||||
- install-godep
|
||||
|
||||
- restore-cache-gomod
|
||||
- run: go mod download
|
||||
- run: cd build && go mod download
|
||||
- save-cache-gomod
|
||||
|
||||
- install-godep
|
||||
- run: make formatcheck
|
||||
- run: make generate-platform-test-list
|
||||
- run: make zrepl-bin test-platform-bin
|
||||
@@ -286,7 +288,7 @@ jobs:
|
||||
goversion:
|
||||
type: string
|
||||
docker:
|
||||
- image: circleci/golang:<<parameters.goversion>>
|
||||
- image: cimg/go:<<parameters.goversion>>
|
||||
steps:
|
||||
- checkout
|
||||
- restore-cache-gomod
|
||||
|
||||
@@ -28,7 +28,7 @@ GO_BUILDFLAGS := $(GO_MOD_READONLY) $(GO_EXTRA_BUILDFLAGS)
|
||||
GO_BUILD := $(GO_ENV_VARS) $(GO) build $(GO_BUILDFLAGS) -ldflags $(GO_LDFLAGS)
|
||||
GOLANGCI_LINT := golangci-lint
|
||||
GOCOVMERGE := gocovmerge
|
||||
RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.17
|
||||
RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.19
|
||||
RELEASE_DOCKER_BASEIMAGE ?= golang:$(RELEASE_DOCKER_BASEIMAGE_TAG)
|
||||
|
||||
ifneq ($(GOARM),)
|
||||
@@ -134,7 +134,13 @@ endif
|
||||
|
||||
deb-docker:
|
||||
docker build -t zrepl_debian_pkg --pull -f packaging/deb/Dockerfile .
|
||||
# Use a small open file limit to make fakeroot work. If we don't
|
||||
# specify it, docker daemon will use its file limit. I don't know
|
||||
# what changed (Docker, its systemd service, its Go version). But I
|
||||
# observed fakeroot iterating close(i) up to i > 1000000, which costs
|
||||
# a good amount of CPU time and makes the build slow.
|
||||
docker run --rm -i -v $(CURDIR):/build/src -u $$(id -u):$$(id -g) \
|
||||
--ulimit nofile=1024:1024 \
|
||||
zrepl_debian_pkg \
|
||||
make deb GOOS=$(GOOS) GOARCH=$(GOARCH) GOARM=$(GOARM)
|
||||
|
||||
|
||||
+3
-6
@@ -4,13 +4,10 @@ go 1.12
|
||||
|
||||
require (
|
||||
github.com/alvaroloes/enumer v1.1.1
|
||||
github.com/golangci/golangci-lint v1.35.2
|
||||
github.com/golangci/misspell v0.3.4 // indirect
|
||||
github.com/golangci/revgrep v0.0.0-20180812185044-276a5c0a1039 // indirect
|
||||
github.com/spf13/afero v1.2.2 // indirect
|
||||
github.com/golangci/golangci-lint v1.50.1
|
||||
github.com/spf13/jwalterweatherman v1.1.0 // indirect
|
||||
github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad
|
||||
golang.org/x/tools v0.0.0-20210105210202-9ed45478a130
|
||||
golang.org/x/tools v0.2.0
|
||||
google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0 // indirect
|
||||
google.golang.org/protobuf v1.25.0
|
||||
google.golang.org/protobuf v1.28.0
|
||||
)
|
||||
|
||||
+1128
File diff suppressed because it is too large
Load Diff
+11
-52
@@ -6,8 +6,6 @@ import (
|
||||
"log/syslog"
|
||||
"os"
|
||||
"reflect"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"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")
|
||||
default:
|
||||
i.Manual = false
|
||||
i.Interval, err = time.ParseDuration(s)
|
||||
i.Interval, err = parsePositiveDuration(s)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if i.Interval <= 0 {
|
||||
return fmt.Errorf("value must be a positive duration, got %q", s)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -228,10 +223,11 @@ type SnapshottingEnum struct {
|
||||
}
|
||||
|
||||
type SnapshottingPeriodic struct {
|
||||
Type string `yaml:"type"`
|
||||
Prefix string `yaml:"prefix"`
|
||||
Interval time.Duration `yaml:"interval,positive"`
|
||||
Hooks HookList `yaml:"hooks,optional"`
|
||||
Type string `yaml:"type"`
|
||||
Prefix string `yaml:"prefix"`
|
||||
Interval *PositiveDuration `yaml:"interval"`
|
||||
Hooks HookList `yaml:"hooks,optional"`
|
||||
TimestampFormat string `yaml:"timestamp_format,optional,default=dense"`
|
||||
}
|
||||
|
||||
type CronSpec struct {
|
||||
@@ -260,10 +256,11 @@ func (s *CronSpec) UnmarshalYAML(unmarshal func(v interface{}, not_strict bool)
|
||||
}
|
||||
|
||||
type SnapshottingCron struct {
|
||||
Type string `yaml:"type"`
|
||||
Prefix string `yaml:"prefix"`
|
||||
Cron CronSpec `yaml:"cron"`
|
||||
Hooks HookList `yaml:"hooks,optional"`
|
||||
Type string `yaml:"type"`
|
||||
Prefix string `yaml:"prefix"`
|
||||
Cron CronSpec `yaml:"cron"`
|
||||
Hooks HookList `yaml:"hooks,optional"`
|
||||
TimestampFormat string `yaml:"timestamp_format,optional,default=dense"`
|
||||
}
|
||||
|
||||
type SnapshottingManual struct {
|
||||
@@ -715,41 +712,3 @@ func ParseConfigBytes(bytes []byte) (*Config, error) {
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -35,8 +35,23 @@ jobs:
|
||||
snapshotting:
|
||||
type: periodic
|
||||
prefix: zrepl_
|
||||
timestamp_format: dense
|
||||
interval: 10m
|
||||
`
|
||||
cron := `
|
||||
snapshotting:
|
||||
type: cron
|
||||
prefix: zrepl_
|
||||
timestamp_format: human
|
||||
cron: "10 * * * *"
|
||||
`
|
||||
|
||||
periodicDaily := `
|
||||
snapshotting:
|
||||
type: periodic
|
||||
prefix: zrepl_
|
||||
interval: 1d
|
||||
`
|
||||
|
||||
hooks := `
|
||||
snapshotting:
|
||||
@@ -74,10 +89,27 @@ jobs:
|
||||
c = testValidConfig(t, fillSnapshotting(periodic))
|
||||
snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingPeriodic)
|
||||
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, "dense", snp.TimestampFormat)
|
||||
})
|
||||
|
||||
t.Run("cron", func(t *testing.T) {
|
||||
c = testValidConfig(t, fillSnapshotting(cron))
|
||||
snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingCron)
|
||||
assert.Equal(t, "cron", snp.Type)
|
||||
assert.Equal(t, "zrepl_", snp.Prefix)
|
||||
assert.Equal(t, "human", snp.TimestampFormat)
|
||||
})
|
||||
|
||||
t.Run("hooks", func(t *testing.T) {
|
||||
c = testValidConfig(t, fillSnapshotting(hooks))
|
||||
hs := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingPeriodic).Hooks
|
||||
@@ -88,3 +120,57 @@ jobs:
|
||||
})
|
||||
|
||||
}
|
||||
|
||||
func TestSnapshottingTimestampDefaults(t *testing.T) {
|
||||
tmpl := `
|
||||
jobs:
|
||||
- name: foo
|
||||
type: push
|
||||
connect:
|
||||
type: local
|
||||
listener_name: foo
|
||||
client_identity: bar
|
||||
filesystems: {"<": true}
|
||||
%s
|
||||
pruning:
|
||||
keep_sender:
|
||||
- type: last_n
|
||||
count: 10
|
||||
keep_receiver:
|
||||
- type: last_n
|
||||
count: 10
|
||||
`
|
||||
|
||||
periodic := `
|
||||
snapshotting:
|
||||
type: periodic
|
||||
prefix: zrepl_
|
||||
interval: 10m
|
||||
`
|
||||
cron := `
|
||||
snapshotting:
|
||||
type: cron
|
||||
prefix: zrepl_
|
||||
cron: "10 * * * *"
|
||||
`
|
||||
|
||||
fillSnapshotting := func(s string) string { return fmt.Sprintf(tmpl, s) }
|
||||
var c *Config
|
||||
|
||||
t.Run("periodic", func(t *testing.T) {
|
||||
c = testValidConfig(t, fillSnapshotting(periodic))
|
||||
snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingPeriodic)
|
||||
assert.Equal(t, "periodic", snp.Type)
|
||||
assert.Equal(t, 10*time.Minute, snp.Interval.Duration())
|
||||
assert.Equal(t, "zrepl_", snp.Prefix)
|
||||
assert.Equal(t, "dense", snp.TimestampFormat) // default was set correctly
|
||||
})
|
||||
|
||||
t.Run("cron", func(t *testing.T) {
|
||||
c = testValidConfig(t, fillSnapshotting(cron))
|
||||
snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingCron)
|
||||
assert.Equal(t, "cron", snp.Type)
|
||||
assert.Equal(t, "zrepl_", snp.Prefix)
|
||||
assert.Equal(t, "dense", snp.TimestampFormat) // default was set correctly
|
||||
})
|
||||
}
|
||||
|
||||
+35
-55
@@ -10,6 +10,7 @@ import (
|
||||
|
||||
"github.com/zrepl/zrepl/config"
|
||||
"github.com/zrepl/zrepl/daemon/hooks"
|
||||
"github.com/zrepl/zrepl/util/suspendresumesafetimer"
|
||||
"github.com/zrepl/zrepl/zfs"
|
||||
)
|
||||
|
||||
@@ -20,8 +21,9 @@ func cronFromConfig(fsf zfs.DatasetFilter, in config.SnapshottingCron) (*Cron, e
|
||||
return nil, errors.Wrap(err, "hook config error")
|
||||
}
|
||||
planArgs := planArgs{
|
||||
prefix: in.Prefix,
|
||||
hooks: hooksList,
|
||||
prefix: in.Prefix,
|
||||
timestampFormat: in.TimestampFormat,
|
||||
hooks: hooksList,
|
||||
}
|
||||
return &Cron{config: in, fsf: fsf, planArgs: planArgs}, nil
|
||||
}
|
||||
@@ -42,67 +44,45 @@ type Cron 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 {
|
||||
now := time.Now()
|
||||
s.mtx.Lock()
|
||||
s.wakeupTime = s.config.Cron.Schedule.Next(now)
|
||||
s.mtx.Unlock()
|
||||
|
||||
// Re-arm the timer.
|
||||
// Need to Stop before Reset, see docs.
|
||||
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():
|
||||
ctxDone := suspendresumesafetimer.SleepUntil(ctx, s.wakeupTime)
|
||||
if ctxDone != nil {
|
||||
return
|
||||
case <-t.C:
|
||||
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")
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
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")
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
+20
-3
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -14,8 +15,9 @@ import (
|
||||
)
|
||||
|
||||
type planArgs struct {
|
||||
prefix string
|
||||
hooks *hooks.List
|
||||
prefix string
|
||||
timestampFormat string
|
||||
hooks *hooks.List
|
||||
}
|
||||
|
||||
type plan struct {
|
||||
@@ -58,6 +60,21 @@ type snapProgress struct {
|
||||
runResults hooks.PlanReport
|
||||
}
|
||||
|
||||
func (plan *plan) formatNow(format string) string {
|
||||
now := time.Now().UTC()
|
||||
switch strings.ToLower(format) {
|
||||
case "dense":
|
||||
format = "20060102_150405_000"
|
||||
case "human":
|
||||
format = "2006-01-02_15:04:05"
|
||||
case "iso-8601":
|
||||
format = "2006-01-02T15:04:05.000Z"
|
||||
case "unix-seconds":
|
||||
return strconv.FormatInt(now.Unix(), 10)
|
||||
}
|
||||
return now.Format(format)
|
||||
}
|
||||
|
||||
func (plan *plan) execute(ctx context.Context, dryRun bool) (ok bool) {
|
||||
|
||||
hookMatchCount := make(map[hooks.Hook]int, len(*plan.args.hooks))
|
||||
@@ -68,7 +85,7 @@ func (plan *plan) execute(ctx context.Context, dryRun bool) (ok bool) {
|
||||
anyFsHadErr := false
|
||||
// TODO channel programs -> allow a little jitter?
|
||||
for fs, progress := range plan.snaps {
|
||||
suffix := time.Now().In(time.UTC).Format("20060102_150405_000")
|
||||
suffix := plan.formatNow(plan.args.timestampFormat)
|
||||
snapname := fmt.Sprintf("%s%s", plan.args.prefix, suffix)
|
||||
|
||||
ctx := logging.WithInjectedField(ctx, "fs", fs.ToString())
|
||||
|
||||
+16
-21
@@ -15,6 +15,7 @@ import (
|
||||
"github.com/zrepl/zrepl/daemon/hooks"
|
||||
"github.com/zrepl/zrepl/daemon/logging"
|
||||
"github.com/zrepl/zrepl/util/envconst"
|
||||
"github.com/zrepl/zrepl/util/suspendresumesafetimer"
|
||||
"github.com/zrepl/zrepl/zfs"
|
||||
)
|
||||
|
||||
@@ -22,7 +23,7 @@ func periodicFromConfig(g *config.Global, fsf zfs.DatasetFilter, in *config.Snap
|
||||
if in.Prefix == "" {
|
||||
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")
|
||||
}
|
||||
|
||||
@@ -32,11 +33,12 @@ func periodicFromConfig(g *config.Global, fsf zfs.DatasetFilter, in *config.Snap
|
||||
}
|
||||
|
||||
args := periodicArgs{
|
||||
interval: in.Interval,
|
||||
interval: in.Interval.Duration(),
|
||||
fsf: fsf,
|
||||
planArgs: planArgs{
|
||||
prefix: in.Prefix,
|
||||
hooks: hookList,
|
||||
prefix: in.Prefix,
|
||||
timestampFormat: in.TimestampFormat,
|
||||
hooks: hookList,
|
||||
},
|
||||
// ctx and log is set in Run()
|
||||
}
|
||||
@@ -171,16 +173,13 @@ func periodicStateSyncUp(a periodicArgs, u updater) state {
|
||||
u(func(s *Periodic) {
|
||||
s.sleepUntil = syncPoint
|
||||
})
|
||||
t := time.NewTimer(time.Until(syncPoint))
|
||||
defer t.Stop()
|
||||
select {
|
||||
case <-t.C:
|
||||
return u(func(s *Periodic) {
|
||||
s.state = Planning
|
||||
}).sf()
|
||||
case <-a.ctx.Done():
|
||||
ctxDone := suspendresumesafetimer.SleepUntil(a.ctx, syncPoint)
|
||||
if ctxDone != nil {
|
||||
return onMainCtxDone(a.ctx, u)
|
||||
}
|
||||
return u(func(s *Periodic) {
|
||||
s.state = Planning
|
||||
}).sf()
|
||||
}
|
||||
|
||||
func periodicStatePlan(a periodicArgs, u updater) state {
|
||||
@@ -241,17 +240,13 @@ func periodicStateWait(a periodicArgs, u updater) state {
|
||||
logFunc("enter wait-state after error")
|
||||
})
|
||||
|
||||
t := time.NewTimer(time.Until(sleepUntil))
|
||||
defer t.Stop()
|
||||
|
||||
select {
|
||||
case <-t.C:
|
||||
return u(func(snapper *Periodic) {
|
||||
snapper.state = Planning
|
||||
}).sf()
|
||||
case <-a.ctx.Done():
|
||||
ctxDone := suspendresumesafetimer.SleepUntil(a.ctx, sleepUntil)
|
||||
if ctxDone != nil {
|
||||
return onMainCtxDone(a.ctx, u)
|
||||
}
|
||||
return u(func(snapper *Periodic) {
|
||||
snapper.state = Planning
|
||||
}).sf()
|
||||
}
|
||||
|
||||
func listFSes(ctx context.Context, mf zfs.DatasetFilter) (fss []*zfs.DatasetPath, err error) {
|
||||
|
||||
+2
-2
@@ -1235,7 +1235,7 @@
|
||||
]
|
||||
},
|
||||
"timezone": "",
|
||||
"title": "zrepl 0.3",
|
||||
"title": "zrepl",
|
||||
"uid": "etQuvBnGz",
|
||||
"version": 7
|
||||
"version": 8
|
||||
}
|
||||
|
||||
Vendored
+5
-1
@@ -9,6 +9,9 @@ ExecStart=/usr/local/bin/zrepl --config /etc/zrepl/zrepl.yml daemon
|
||||
RuntimeDirectory=zrepl zrepl/stdinserver
|
||||
RuntimeDirectoryMode=0700
|
||||
|
||||
# Make Go produce coredumps
|
||||
Environment=GOTRACEBACK='crash'
|
||||
|
||||
ProtectSystem=strict
|
||||
#PrivateDevices=yes # TODO ZFS needs access to /dev/zfs, could we limit this?
|
||||
ProtectKernelTunables=yes
|
||||
@@ -27,7 +30,8 @@ ProtectHome=read-only
|
||||
# SystemCallFilter
|
||||
# ~@privileged doesn't work with Ubuntu 18.04 ssh
|
||||
SystemCallFilter=~ @mount @cpu-emulation @keyring @module @obsolete @raw-io @debug @clock @resources
|
||||
|
||||
# Go1.19 added automatic RLIMIT_NOFILE changes, so, we need to allow that
|
||||
SystemCallFilter= setrlimit
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
|
||||
+41
-5
@@ -16,13 +16,42 @@ Changelog
|
||||
The changelog summarizes bugfixes that are deemed relevant for users and package maintainers.
|
||||
Developers should consult the git commit log or GitHub issue tracker.
|
||||
|
||||
0.6 (Unreleased)
|
||||
----------------
|
||||
Next Release
|
||||
------------
|
||||
|
||||
* `Feature Wishlist on GitHub <https://github.com/zrepl/zrepl/discussions/547>`_
|
||||
The plan for the next release is to revisit how zrepl does snapshot management.
|
||||
High-level goals:
|
||||
|
||||
- Make it easy to decouple snapshot management (snapshotting, pruning) from replication.
|
||||
- Ability to include/exclude snapshots from replication.
|
||||
This is useful for aforementioned decoupling, e.g., separate snapshot prefixes for local & remote replication.
|
||||
Also, it makes explicit that by default, zrepl replicates all snapshots, and that
|
||||
replication has no concept of "zrepl-created snapshots", which is a common misconception.
|
||||
- Use of ``zfs snapshot`` comma syntax or channel programs to take snapshots of multiple
|
||||
datasets atomically.
|
||||
- Provide an alternative to the ``grid`` pruning policy.
|
||||
Most likely something based on hourly/daily/weekly/monthly "trains" plus a count.
|
||||
- Ability to prune at the granularity of the **group** of snapshots created at a given
|
||||
time, as opposed to the individual snapshots within a dataset.
|
||||
Maybe this will be addressed by the alternative to the ``grid`` pruning policy,
|
||||
as it will likely be more predictable.
|
||||
|
||||
Those changes will likely come with some breakage in the config.
|
||||
However, I want to avoid breaking **use cases** that are satisfied by the current design.
|
||||
There will be beta/RC releases to give users a chance to evaluate.
|
||||
|
||||
0.6
|
||||
---
|
||||
|
||||
* |feature| :ref:`Schedule-based snapshotting<job-snapshotting--cron>` using ``cron`` syntax instead of an interval.
|
||||
* |feature| Add ``ZREPL_DESTROY_MAX_BATCH_SIZE`` env var (default 0=unlimited).
|
||||
* |feature| Configurable initial replication policy.
|
||||
When a filesystem is first replicated to a receiver, this control whether just the newest
|
||||
snapshot will be replicated vs. all existing snapshots. Learn more :ref:`in the docs <conflict_resolution-initial_replication>`.
|
||||
* |feature| Configurable timestamp format for snapshot names via :ref:`timestamp_format<job-snapshotting-timestamp_format>`
|
||||
(Thanks, `@ydylla <https://github.com/ydylla>`_).
|
||||
* |feature| Add ``ZREPL_DESTROY_MAX_BATCH_SIZE`` env var (default 0=unlimited)
|
||||
(Thanks, `@3nprob <https://github.com/3nprob>`_).
|
||||
* |feature| Add ``zrepl configcheck --skip-cert-check`` flag (Thanks, `@cole-h <https://github.com/cole-h>`_).
|
||||
* |bugfix| Fix resuming from interrupted replications that use ``send.raw`` on unencrypted datasets.
|
||||
|
||||
* The send options introduced in zrepl 0.4 allow users to specify additional zfs send flags for zrepl to use.
|
||||
@@ -34,7 +63,7 @@ Developers should consult the git commit log or GitHub issue tracker.
|
||||
* However, this means that the ``zrepl status`` UI no longer indicates whether a replication step uses encrypted sends or not.
|
||||
The setting is still effective though.
|
||||
|
||||
* |break| |feature| convert Prometheus metric ``zrepl_version_daemon`` to ``zrepl_start_time`` metric
|
||||
* |break| convert Prometheus metric ``zrepl_version_daemon`` to ``zrepl_start_time`` metric
|
||||
|
||||
* The metric still reports the zrepl version in a label.
|
||||
But the metric *value* is now the Unix timestamp at the time the daemon was started.
|
||||
@@ -43,6 +72,13 @@ Developers should consult the git commit log or GitHub issue tracker.
|
||||
* |bugfix| transient zrepl status error: ``Post "http://unix/status": EOF``
|
||||
* |bugfix| don't treat receive-side bookmarks as a replication conflict.
|
||||
This facilitates chaining of replication jobs. See :issue:`490`.
|
||||
* |bugfix| workaround for Go/gRPC problem on Illumos where zrepl would
|
||||
crash when using the ``local`` transport type (:issue:`598`).
|
||||
* |bugfix| fix active child tasks panic that cold occur during replication plannig (:issue:`193abbe`)
|
||||
* |bugfix| ``zrepl status`` off-by-one error in display of completed step count (:commit:`ce6701f`)
|
||||
* |bugfix| Allow using day & week units for ``snapshotting.interval`` (:commit:`ffb1d89`)
|
||||
* |docs| ``docs/overview`` improvements (Thanks, `@jtagcat <https://github.com/jtagcat>`_).
|
||||
* |maint| Update to Go 1.19.
|
||||
|
||||
0.5
|
||||
---
|
||||
|
||||
@@ -15,7 +15,7 @@ Conflict Resolution Options
|
||||
|
||||
...
|
||||
|
||||
.. _conflict_resolution-initial_replication-option-send_all_snapshots:
|
||||
.. _conflict_resolution-initial_replication:
|
||||
|
||||
|
||||
``initial_replication`` option
|
||||
|
||||
@@ -62,6 +62,9 @@ The ``periodic`` and ``cron`` snapshotting types share some common options and b
|
||||
type: periodic
|
||||
prefix: zrepl_
|
||||
interval: 10m
|
||||
# Timestamp format that is used as snapshot suffix.
|
||||
# Can be any of "dense" (default), "human", "iso-8601", "unix-seconds" or a custom Go time format (see https://go.dev/src/time/format.go)
|
||||
timestamp_format: dense
|
||||
hooks: ...
|
||||
pruning: ...
|
||||
|
||||
@@ -91,6 +94,9 @@ The snapshotter uses the ``prefix`` to identify which snapshots it created.
|
||||
# (second, optional) minute hour day-of-month month day-of-week
|
||||
# This example takes snapshots daily at 3:00.
|
||||
cron: "0 3 * * *"
|
||||
# Timestamp format that is used as snapshot suffix.
|
||||
# Can be any of "dense" (default), "human", "iso-8601", "unix-seconds" or a custom Go time format (see https://go.dev/src/time/format.go)
|
||||
timestamp_format: dense
|
||||
pruning: ...
|
||||
|
||||
In ``cron`` mode, the snapshotter takes snaphots at fixed points in time.
|
||||
@@ -98,6 +104,21 @@ See https://en.wikipedia.org/wiki/Cron for details on the syntax.
|
||||
zrepl uses the ``the github.com/robfig/cron/v3`` Go package for parsing.
|
||||
An optional field for "seconds" is supported to take snapshots at sub-minute frequencies.
|
||||
|
||||
.. _job-snapshotting-timestamp_format:
|
||||
|
||||
Timestamp Format
|
||||
~~~~~~~~~~~~~~~~
|
||||
|
||||
The ``cron`` and ``periodic`` snapshotter support configuring a custom timestamp format that is used as suffix for the snapshot name.
|
||||
It can be used by setting ``timestamp_format`` to any of the following values:
|
||||
|
||||
* ``dense`` (default) looks like ``20060102_150405_000``
|
||||
* ``human`` looks like ``2006-01-02_15:04:05``
|
||||
* ``iso-8601`` looks like ``2006-01-02T15:04:05.000Z``
|
||||
* ``unix-seconds`` looks like ``1136214245``
|
||||
* Any custom Go time format accepted by `time.Time#Format <https://go.dev/src/time/format.go>`_.
|
||||
|
||||
|
||||
``manual`` Snapshotting
|
||||
-----------------------
|
||||
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
+23
-3
@@ -343,7 +343,6 @@ const (
|
||||
|
||||
type SendStream struct {
|
||||
cmd *zfscmd.Cmd
|
||||
kill context.CancelFunc
|
||||
stdoutReader io.ReadCloser // not *os.File for mocking during platformtest
|
||||
stderrBuf *circlog.CircularLog
|
||||
|
||||
@@ -416,8 +415,29 @@ func (s *SendStream) killAndWait() error {
|
||||
debug("sendStream: killAndWait enter")
|
||||
defer debug("sendStream: killAndWait leave")
|
||||
|
||||
// ensure this function is only called once
|
||||
if s.state != sendStreamOpen {
|
||||
panic(s.state)
|
||||
}
|
||||
|
||||
// send SIGKILL
|
||||
s.kill()
|
||||
// In an earlier version, we used the starting context.Context's cancel function
|
||||
// for this, but in Go > 1.19, doing so will cause .Wait() to return the
|
||||
// context cancel error instead of the *exec.ExitError.
|
||||
err := s.cmd.Process().Kill()
|
||||
if err != nil {
|
||||
if err == os.ErrProcessDone {
|
||||
// This can happen if
|
||||
// (1) the process has already been .Wait()ed, or
|
||||
// (2) some other goroutine cancels the context, likely further up
|
||||
// the context tree.
|
||||
// Case (1) can't happen to us because we only call
|
||||
// this function in sendStreamOpen state.
|
||||
// In Case (2), it's still our job to .Wait(), so, fallthrough.
|
||||
} else {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// Close our read-end of the pipe.
|
||||
//
|
||||
@@ -963,10 +983,10 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*SendStream, e
|
||||
|
||||
stream := &SendStream{
|
||||
cmd: cmd,
|
||||
kill: cancel,
|
||||
stdoutReader: stdoutReader,
|
||||
stderrBuf: stderrBuf,
|
||||
}
|
||||
_ = cancel // the SendStream.killAndWait() will kill the process
|
||||
|
||||
return stream, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user