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
46 changed files with 340 additions and 1726 deletions
+13 -15
View File
@@ -24,12 +24,7 @@ commands:
apt-update-and-install-common-deps: apt-update-and-install-common-deps:
steps: steps:
- run: sudo apt-get update - run: sudo apt update && sudo apt install gawk make
- 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: restore-cache-gomod:
steps: steps:
@@ -155,7 +150,7 @@ parameters:
release_docker_baseimage_tag: release_docker_baseimage_tag:
type: string type: string
default: "1.19" default: "1.17"
workflows: workflows:
version: 2 version: 2
@@ -165,19 +160,19 @@ workflows:
jobs: jobs:
- quickcheck-docs - quickcheck-docs
- quickcheck-go: &quickcheck-go-smoketest - quickcheck-go: &quickcheck-go-smoketest
name: quickcheck-go-amd64-linux-1.19 name: quickcheck-go-amd64-linux-1.17
goversion: &latest-go-release "1.19" goversion: &latest-go-release "1.17"
goos: linux goos: linux
goarch: amd64 goarch: amd64
- test-go-on-latest-go-release: - test-go-on-latest-go-release:
goversion: *latest-go-release goversion: *latest-go-release
- quickcheck-go: - quickcheck-go:
requires: requires:
- quickcheck-go-amd64-linux-1.19 #quickcheck-go-smoketest.name - quickcheck-go-amd64-linux-1.17 #quickcheck-go-smoketest.name
matrix: &quickcheck-go-matrix matrix: &quickcheck-go-matrix
alias: quickcheck-go-matrix alias: quickcheck-go-matrix
parameters: parameters:
goversion: [*latest-go-release, "1.18"] goversion: [*latest-go-release, "1.12"]
goos: ["linux", "freebsd"] goos: ["linux", "freebsd"]
goarch: ["amd64", "arm64"] goarch: ["amd64", "arm64"]
exclude: exclude:
@@ -185,6 +180,10 @@ workflows:
- goversion: *latest-go-release - goversion: *latest-go-release
goos: linux goos: linux
goarch: amd64 goarch: amd64
# not supported by Go 1.12
- goversion: "1.12"
goos: freebsd
goarch: arm64
release: release:
when: << pipeline.parameters.do_release >> when: << pipeline.parameters.do_release >>
@@ -250,7 +249,7 @@ jobs:
goarch: goarch:
type: string type: string
docker: docker:
- image: cimg/go:<<parameters.goversion>> - image: circleci/golang:<<parameters.goversion>>
environment: environment:
GOOS: <<parameters.goos>> GOOS: <<parameters.goos>>
GOARCH: <<parameters.goarch>> GOARCH: <<parameters.goarch>>
@@ -258,13 +257,12 @@ jobs:
steps: steps:
- checkout - checkout
- install-godep
- restore-cache-gomod - restore-cache-gomod
- run: go mod download - run: go mod download
- run: cd build && go mod download - run: cd build && go mod download
- save-cache-gomod - save-cache-gomod
- install-godep
- run: make formatcheck - run: make formatcheck
- run: make generate-platform-test-list - run: make generate-platform-test-list
- run: make zrepl-bin test-platform-bin - run: make zrepl-bin test-platform-bin
@@ -288,7 +286,7 @@ jobs:
goversion: goversion:
type: string type: string
docker: docker:
- image: cimg/go:<<parameters.goversion>> - image: circleci/golang:<<parameters.goversion>>
steps: steps:
- checkout - checkout
- restore-cache-gomod - restore-cache-gomod
+1 -7
View File
@@ -28,7 +28,7 @@ GO_BUILDFLAGS := $(GO_MOD_READONLY) $(GO_EXTRA_BUILDFLAGS)
GO_BUILD := $(GO_ENV_VARS) $(GO) build $(GO_BUILDFLAGS) -ldflags $(GO_LDFLAGS) GO_BUILD := $(GO_ENV_VARS) $(GO) build $(GO_BUILDFLAGS) -ldflags $(GO_LDFLAGS)
GOLANGCI_LINT := golangci-lint GOLANGCI_LINT := golangci-lint
GOCOVMERGE := gocovmerge GOCOVMERGE := gocovmerge
RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.19 RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.17
RELEASE_DOCKER_BASEIMAGE ?= golang:$(RELEASE_DOCKER_BASEIMAGE_TAG) RELEASE_DOCKER_BASEIMAGE ?= golang:$(RELEASE_DOCKER_BASEIMAGE_TAG)
ifneq ($(GOARM),) ifneq ($(GOARM),)
@@ -134,13 +134,7 @@ endif
deb-docker: deb-docker:
docker build -t zrepl_debian_pkg --pull -f packaging/deb/Dockerfile . 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) \ docker run --rm -i -v $(CURDIR):/build/src -u $$(id -u):$$(id -g) \
--ulimit nofile=1024:1024 \
zrepl_debian_pkg \ zrepl_debian_pkg \
make deb GOOS=$(GOOS) GOARCH=$(GOARCH) GOARM=$(GOARM) make deb GOOS=$(GOOS) GOARCH=$(GOARCH) GOARM=$(GOARM)
+6 -3
View File
@@ -4,10 +4,13 @@ go 1.12
require ( require (
github.com/alvaroloes/enumer v1.1.1 github.com/alvaroloes/enumer v1.1.1
github.com/golangci/golangci-lint v1.50.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/spf13/jwalterweatherman v1.1.0 // indirect github.com/spf13/jwalterweatherman v1.1.0 // indirect
github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad
golang.org/x/tools v0.2.0 golang.org/x/tools v0.0.0-20210105210202-9ed45478a130
google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0 // indirect google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0 // indirect
google.golang.org/protobuf v1.28.0 google.golang.org/protobuf v1.25.0
) )
-1128
View File
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -57,7 +57,7 @@ func (f *zabsFilterFlags) registerZabsFilterFlags(s *pflag.FlagSet, verb string)
for v := range endpoint.AbstractionTypesAll { for v := range endpoint.AbstractionTypesAll {
variants = append(variants, string(v)) variants = append(variants, string(v))
} }
sort.Strings(variants) variants = sort.StringSlice(variants)
variantsJoined := strings.Join(variants, "|") variantsJoined := strings.Join(variants, "|")
s.Var(&f.Types, "type", fmt.Sprintf("only %s holds of the specified type [default: all] [comma-separated list of %s]", verb, variantsJoined)) s.Var(&f.Types, "type", fmt.Sprintf("only %s holds of the specified type [default: all] [comma-separated list of %s]", verb, variantsJoined))
-2
View File
@@ -227,7 +227,6 @@ type SnapshottingPeriodic struct {
Prefix string `yaml:"prefix"` Prefix string `yaml:"prefix"`
Interval *PositiveDuration `yaml:"interval"` Interval *PositiveDuration `yaml:"interval"`
Hooks HookList `yaml:"hooks,optional"` Hooks HookList `yaml:"hooks,optional"`
TimestampFormat string `yaml:"timestamp_format,optional,default=dense"`
} }
type CronSpec struct { type CronSpec struct {
@@ -260,7 +259,6 @@ type SnapshottingCron struct {
Prefix string `yaml:"prefix"` Prefix string `yaml:"prefix"`
Cron CronSpec `yaml:"cron"` Cron CronSpec `yaml:"cron"`
Hooks HookList `yaml:"hooks,optional"` Hooks HookList `yaml:"hooks,optional"`
TimestampFormat string `yaml:"timestamp_format,optional,default=dense"`
} }
type SnapshottingManual struct { type SnapshottingManual struct {
-71
View File
@@ -35,16 +35,8 @@ jobs:
snapshotting: snapshotting:
type: periodic type: periodic
prefix: zrepl_ prefix: zrepl_
timestamp_format: dense
interval: 10m interval: 10m
` `
cron := `
snapshotting:
type: cron
prefix: zrepl_
timestamp_format: human
cron: "10 * * * *"
`
periodicDaily := ` periodicDaily := `
snapshotting: snapshotting:
@@ -99,15 +91,6 @@ jobs:
assert.Equal(t, "periodic", snp.Type) assert.Equal(t, "periodic", snp.Type)
assert.Equal(t, 24*time.Hour, snp.Interval.Duration()) assert.Equal(t, 24*time.Hour, snp.Interval.Duration())
assert.Equal(t, "zrepl_", snp.Prefix) 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) { t.Run("hooks", func(t *testing.T) {
@@ -120,57 +103,3 @@ 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
})
}
+1 -2
View File
@@ -44,8 +44,7 @@ func TestSampleConfigsAreParsedWithoutErrors(t *testing.T) {
} }
// template must be a template/text template with a single '{{ . }}' as placeholder for val // template must be a template/text template with a single '{{ . }}' as placeholder for val
// //nolint[:deadcode,unused]
//nolint:deadcode,unused
func testValidConfigTemplate(t *testing.T, tmpl string, val string) *Config { func testValidConfigTemplate(t *testing.T, tmpl string, val string) *Config {
tmp, err := template.New("master").Parse(tmpl) tmp, err := template.New("master").Parse(tmpl)
if err != nil { if err != nil {
+3 -2
View File
@@ -5,7 +5,7 @@
// //
// This package also provides all supported hook type implementations and abstractions around them. // This package also provides all supported hook type implementations and abstractions around them.
// //
// # Use For Other Kinds Of ExpectStepReports // Use For Other Kinds Of ExpectStepReports
// //
// This package REQUIRES REFACTORING before it can be used for other activities than snapshots, e.g. pre- and post-replication: // This package REQUIRES REFACTORING before it can be used for other activities than snapshots, e.g. pre- and post-replication:
// //
@@ -15,7 +15,7 @@
// The hook implementations should move out of this package. // The hook implementations should move out of this package.
// However, there is a lot of tight coupling which to untangle isn't worth it ATM. // However, there is a lot of tight coupling which to untangle isn't worth it ATM.
// //
// # How This Package Is Used By Package Snapper // How This Package Is Used By Package Snapper
// //
// Deserialize a config.List using ListFromConfig(). // Deserialize a config.List using ListFromConfig().
// Then it MUST filter the list to only contain hooks for a particular filesystem using // Then it MUST filter the list to only contain hooks for a particular filesystem using
@@ -30,4 +30,5 @@
// Command hooks make it available in the environment variable ZREPL_DRYRUN. // Command hooks make it available in the environment variable ZREPL_DRYRUN.
// //
// Plan.Report() can be called while Plan.Run() is executing to give an overview of plan execution progress (future use in "zrepl status"). // Plan.Report() can be called while Plan.Run() is executing to give an overview of plan execution progress (future use in "zrepl status").
//
package hooks package hooks
@@ -17,7 +17,6 @@ import (
"github.com/zrepl/zrepl/zfs" "github.com/zrepl/zrepl/zfs"
) )
// Hook to implement the following recommmendation from MySQL docs
// https://dev.mysql.com/doc/mysql-backup-excerpt/5.7/en/backup-methods.html // https://dev.mysql.com/doc/mysql-backup-excerpt/5.7/en/backup-methods.html
// //
// Making Backups Using a File System Snapshot: // Making Backups Using a File System Snapshot:
@@ -31,8 +30,6 @@ import (
// Unmount the snapshot. // Unmount the snapshot.
// //
// Similar snapshot capabilities may be available in other file systems, such as LVM or ZFS. // Similar snapshot capabilities may be available in other file systems, such as LVM or ZFS.
//
type MySQLLockTables struct { type MySQLLockTables struct {
errIsFatal bool errIsFatal bool
connector sqldriver.Connector connector sqldriver.Connector
+7 -4
View File
@@ -1,6 +1,6 @@
// package trace provides activity tracing via ctx through Tasks and Spans // package trace provides activity tracing via ctx through Tasks and Spans
// //
// # Basic Concepts // Basic Concepts
// //
// Tracing can be used to identify where a piece of code spends its time. // Tracing can be used to identify where a piece of code spends its time.
// //
@@ -10,9 +10,11 @@
// to tech-savvy users (albeit not developers). // to tech-savvy users (albeit not developers).
// //
// This package provides the concept of Tasks and Spans to express what activity is happening within an application: // This package provides the concept of Tasks and Spans to express what activity is happening within an application:
//
// - Neither task nor span is really tangible but instead contained within the context.Context tree // - Neither task nor span is really tangible but instead contained within the context.Context tree
// - Tasks represent concurrent activity (i.e. goroutines). // - Tasks represent concurrent activity (i.e. goroutines).
// - Spans represent a semantic stack trace within a task. // - Spans represent a semantic stack trace within a task.
//
// As a consequence, whenever a context is propagated across goroutine boundary, you need to create a child task: // As a consequence, whenever a context is propagated across goroutine boundary, you need to create a child task:
// //
// go func(ctx context.Context) { // go func(ctx context.Context) {
@@ -37,7 +39,6 @@
// } // }
// //
// In combination: // In combination:
//
// ctx, endTask = WithTask(ctx, "copy-dirs") // ctx, endTask = WithTask(ctx, "copy-dirs")
// defer endTask() // defer endTask()
// for i := range dirs { // for i := range dirs {
@@ -64,7 +65,8 @@
// //
// Recovering from endSpan() or endTask() panics will corrupt the trace stack and lead to corrupt tracefile output. // Recovering from endSpan() or endTask() panics will corrupt the trace stack and lead to corrupt tracefile output.
// //
// # Best Practices For Naming Tasks And Spans //
// Best Practices For Naming Tasks And Spans
// //
// Tasks should always have string constants as names, and must not contain the `#` character. WHy? // Tasks should always have string constants as names, and must not contain the `#` character. WHy?
// First, the visualization by chrome://tracing draws a horizontal bar for each task in the trace. // First, the visualization by chrome://tracing draws a horizontal bar for each task in the trace.
@@ -72,7 +74,8 @@
// Note that the `#NUM` suffix will be reused if a task has ended, in order to avoid an // Note that the `#NUM` suffix will be reused if a task has ended, in order to avoid an
// infinite number of horizontal bars in the visualization. // infinite number of horizontal bars in the visualization.
// //
// # Chrome-compatible Tracefile Support //
// Chrome-compatible Tracefile Support
// //
// The activity trace generated by usage of WithTask and WithSpan can be rendered to a JSON output file // The activity trace generated by usage of WithTask and WithSpan can be rendered to a JSON output file
// that can be loaded into chrome://tracing . // that can be loaded into chrome://tracing .
@@ -11,6 +11,8 @@ import (
// use like this: // use like this:
// //
// defer WithSpanFromStackUpdateCtx(&existingCtx)() // defer WithSpanFromStackUpdateCtx(&existingCtx)()
//
//
func WithSpanFromStackUpdateCtx(ctx *context.Context) DoneFunc { func WithSpanFromStackUpdateCtx(ctx *context.Context) DoneFunc {
childSpanCtx, end := WithSpan(*ctx, getMyCallerOrPanic()) childSpanCtx, end := WithSpan(*ctx, getMyCallerOrPanic())
*ctx = childSpanCtx *ctx = childSpanCtx
+25 -5
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"
) )
@@ -22,7 +21,6 @@ func cronFromConfig(fsf zfs.DatasetFilter, in config.SnapshottingCron) (*Cron, e
} }
planArgs := planArgs{ planArgs := planArgs{
prefix: in.Prefix, prefix: in.Prefix,
timestampFormat: in.TimestampFormat,
hooks: hooksList, hooks: hooksList,
} }
return &Cron{config: in, fsf: fsf, planArgs: planArgs}, nil return &Cron{config: in, fsf: fsf, planArgs: planArgs}, nil
@@ -44,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 {
@@ -84,6 +103,7 @@ func (s *Cron) Run(ctx context.Context, snapshotsTaken chan<- struct{}) {
} }
}() }()
} }
}
} }
+1 -18
View File
@@ -4,7 +4,6 @@ import (
"context" "context"
"fmt" "fmt"
"sort" "sort"
"strconv"
"strings" "strings"
"time" "time"
@@ -16,7 +15,6 @@ import (
type planArgs struct { type planArgs struct {
prefix string prefix string
timestampFormat string
hooks *hooks.List hooks *hooks.List
} }
@@ -60,21 +58,6 @@ type snapProgress struct {
runResults hooks.PlanReport 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) { func (plan *plan) execute(ctx context.Context, dryRun bool) (ok bool) {
hookMatchCount := make(map[hooks.Hook]int, len(*plan.args.hooks)) hookMatchCount := make(map[hooks.Hook]int, len(*plan.args.hooks))
@@ -85,7 +68,7 @@ func (plan *plan) execute(ctx context.Context, dryRun bool) (ok bool) {
anyFsHadErr := false anyFsHadErr := false
// TODO channel programs -> allow a little jitter? // TODO channel programs -> allow a little jitter?
for fs, progress := range plan.snaps { for fs, progress := range plan.snaps {
suffix := plan.formatNow(plan.args.timestampFormat) suffix := time.Now().In(time.UTC).Format("20060102_150405_000")
snapname := fmt.Sprintf("%s%s", plan.args.prefix, suffix) snapname := fmt.Sprintf("%s%s", plan.args.prefix, suffix)
ctx := logging.WithInjectedField(ctx, "fs", fs.ToString()) ctx := logging.WithInjectedField(ctx, "fs", fs.ToString())
+15 -10
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"
) )
@@ -37,7 +36,6 @@ func periodicFromConfig(g *config.Global, fsf zfs.DatasetFilter, in *config.Snap
fsf: fsf, fsf: fsf,
planArgs: planArgs{ planArgs: planArgs{
prefix: in.Prefix, prefix: in.Prefix,
timestampFormat: in.TimestampFormat,
hooks: hookList, hooks: hookList,
}, },
// ctx and log is set in Run() // ctx and log is set in Run()
@@ -173,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 {
@@ -240,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) {
+2 -2
View File
@@ -1235,7 +1235,7 @@
] ]
}, },
"timezone": "", "timezone": "",
"title": "zrepl", "title": "zrepl 0.3",
"uid": "etQuvBnGz", "uid": "etQuvBnGz",
"version": 8 "version": 7
} }
+1 -5
View File
@@ -9,9 +9,6 @@ ExecStart=/usr/local/bin/zrepl --config /etc/zrepl/zrepl.yml daemon
RuntimeDirectory=zrepl zrepl/stdinserver RuntimeDirectory=zrepl zrepl/stdinserver
RuntimeDirectoryMode=0700 RuntimeDirectoryMode=0700
# Make Go produce coredumps
Environment=GOTRACEBACK='crash'
ProtectSystem=strict ProtectSystem=strict
#PrivateDevices=yes # TODO ZFS needs access to /dev/zfs, could we limit this? #PrivateDevices=yes # TODO ZFS needs access to /dev/zfs, could we limit this?
ProtectKernelTunables=yes ProtectKernelTunables=yes
@@ -30,8 +27,7 @@ ProtectHome=read-only
# SystemCallFilter # SystemCallFilter
# ~@privileged doesn't work with Ubuntu 18.04 ssh # ~@privileged doesn't work with Ubuntu 18.04 ssh
SystemCallFilter=~ @mount @cpu-emulation @keyring @module @obsolete @raw-io @debug @clock @resources 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] [Install]
WantedBy=multi-user.target WantedBy=multi-user.target
+5 -41
View File
@@ -16,42 +16,13 @@ Changelog
The changelog summarizes bugfixes that are deemed relevant for users and package maintainers. The changelog summarizes bugfixes that are deemed relevant for users and package maintainers.
Developers should consult the git commit log or GitHub issue tracker. Developers should consult the git commit log or GitHub issue tracker.
Next Release 0.6 (Unreleased)
------------ ----------------
The plan for the next release is to revisit how zrepl does snapshot management. * `Feature Wishlist on GitHub <https://github.com/zrepl/zrepl/discussions/547>`_
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| :ref:`Schedule-based snapshotting<job-snapshotting--cron>` using ``cron`` syntax instead of an interval.
* |feature| Configurable initial replication policy. * |feature| Add ``ZREPL_DESTROY_MAX_BATCH_SIZE`` env var (default 0=unlimited).
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. * |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. * The send options introduced in zrepl 0.4 allow users to specify additional zfs send flags for zrepl to use.
@@ -63,7 +34,7 @@ There will be beta/RC releases to give users a chance to evaluate.
* However, this means that the ``zrepl status`` UI no longer indicates whether a replication step uses encrypted sends or not. * 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. The setting is still effective though.
* |break| convert Prometheus metric ``zrepl_version_daemon`` to ``zrepl_start_time`` metric * |break| |feature| convert Prometheus metric ``zrepl_version_daemon`` to ``zrepl_start_time`` metric
* The metric still reports the zrepl version in a label. * 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. But the metric *value* is now the Unix timestamp at the time the daemon was started.
@@ -72,13 +43,6 @@ There will be beta/RC releases to give users a chance to evaluate.
* |bugfix| transient zrepl status error: ``Post "http://unix/status": EOF`` * |bugfix| transient zrepl status error: ``Post "http://unix/status": EOF``
* |bugfix| don't treat receive-side bookmarks as a replication conflict. * |bugfix| don't treat receive-side bookmarks as a replication conflict.
This facilitates chaining of replication jobs. See :issue:`490`. 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 0.5
--- ---
+1 -1
View File
@@ -15,7 +15,7 @@ Conflict Resolution Options
... ...
.. _conflict_resolution-initial_replication: .. _conflict_resolution-initial_replication-option-send_all_snapshots:
``initial_replication`` option ``initial_replication`` option
-21
View File
@@ -62,9 +62,6 @@ The ``periodic`` and ``cron`` snapshotting types share some common options and b
type: periodic type: periodic
prefix: zrepl_ prefix: zrepl_
interval: 10m 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: ... hooks: ...
pruning: ... pruning: ...
@@ -94,9 +91,6 @@ The snapshotter uses the ``prefix`` to identify which snapshots it created.
# (second, optional) minute hour day-of-month month day-of-week # (second, optional) minute hour day-of-month month day-of-week
# This example takes snapshots daily at 3:00. # This example takes snapshots daily at 3:00.
cron: "0 3 * * *" 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: ... pruning: ...
In ``cron`` mode, the snapshotter takes snaphots at fixed points in time. In ``cron`` mode, the snapshotter takes snaphots at fixed points in time.
@@ -104,21 +98,6 @@ 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. 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. 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 ``manual`` Snapshotting
----------------------- -----------------------
+1 -1
View File
@@ -168,7 +168,7 @@ func (s AbstractionTypeSet) String() string {
for i := range s { for i := range s {
sts = append(sts, string(i)) sts = append(sts, string(i))
} }
sort.Strings(sts) sts = sort.StringSlice(sts)
return strings.Join(sts, ",") return strings.Join(sts, ",")
} }
@@ -23,7 +23,7 @@ func replicationCursorBookmarkNameImpl(fs string, guid uint64, jobid string) (st
var ErrV1ReplicationCursor = fmt.Errorf("bookmark name is a v1-replication cursor") var ErrV1ReplicationCursor = fmt.Errorf("bookmark name is a v1-replication cursor")
// err != nil always means that the bookmark is not a valid replication bookmark //err != nil always means that the bookmark is not a valid replication bookmark
// //
// Returns ErrV1ReplicationCursor as error if the bookmark is a v1 replication cursor // Returns ErrV1ReplicationCursor as error if the bookmark is a v1 replication cursor
func ParseReplicationCursorBookmarkName(fullname string) (uint64, JobID, error) { func ParseReplicationCursorBookmarkName(fullname string) (uint64, JobID, error) {
+1 -1
View File
@@ -797,7 +797,7 @@ func (a *attempt) errorReport() *errorReport {
r.byClass[class] = errs r.byClass[class] = errs
} }
for _, err := range r.flattened { for _, err := range r.flattened {
if neterr, ok := err.Err.(net.Error); ok && neterr.Timeout() { if neterr, ok := err.Err.(net.Error); ok && neterr.Temporary() {
putClass(err, errorClassTemporaryConnectivityRelated) putClass(err, errorClassTemporaryConnectivityRelated)
continue continue
} }
@@ -13,7 +13,7 @@ func init() {
} }
} }
//nolint:deadcode,unused //nolint[:deadcode,unused]
func debug(format string, args ...interface{}) { func debug(format string, args ...interface{}) {
if debugEnabled { if debugEnabled {
fmt.Fprintf(os.Stderr, "repl: driver: %s\n", fmt.Sprintf(format, args...)) fmt.Fprintf(os.Stderr, "repl: driver: %s\n", fmt.Sprintf(format, args...))
@@ -22,7 +22,7 @@ func debug(format string, args ...interface{}) {
type debugFunc func(format string, args ...interface{}) type debugFunc func(format string, args ...interface{})
//nolint:deadcode,unused //nolint[:deadcode,unused]
func debugPrefix(prefixFormat string, prefixFormatArgs ...interface{}) debugFunc { func debugPrefix(prefixFormat string, prefixFormatArgs ...interface{}) debugFunc {
prefix := fmt.Sprintf(prefixFormat, prefixFormatArgs...) prefix := fmt.Sprintf(prefixFormat, prefixFormatArgs...)
return func(format string, args ...interface{}) { return func(format string, args ...interface{}) {
+1 -1
View File
@@ -14,7 +14,7 @@ func init() {
} }
} }
//nolint:deadcode,unused //nolint[:deadcode,unused]
func debug(format string, args ...interface{}) { func debug(format string, args ...interface{}) {
if debugEnabled { if debugEnabled {
fmt.Fprintf(os.Stderr, "rpc/dataconn: %s\n", fmt.Sprintf(format, args...)) fmt.Fprintf(os.Stderr, "rpc/dataconn: %s\n", fmt.Sprintf(format, args...))
@@ -24,9 +24,6 @@ func (e HeartbeatTimeout) Error() string {
return "heartbeat timeout" return "heartbeat timeout"
} }
// This function is deprecated in net.Error and since this
// function is not involved in .Accept() code path, nothing
// really needs this method to be here.
func (e HeartbeatTimeout) Temporary() bool { return true } func (e HeartbeatTimeout) Temporary() bool { return true }
func (e HeartbeatTimeout) Timeout() bool { return true } func (e HeartbeatTimeout) Timeout() bool { return true }
@@ -13,7 +13,7 @@ func init() {
} }
} }
//nolint:deadcode,unused //nolint[:deadcode,unused]
func debug(format string, args ...interface{}) { func debug(format string, args ...interface{}) {
if debugEnabled { if debugEnabled {
fmt.Fprintf(os.Stderr, "rpc/dataconn/heartbeatconn: %s\n", fmt.Sprintf(format, args...)) fmt.Fprintf(os.Stderr, "rpc/dataconn/heartbeatconn: %s\n", fmt.Sprintf(format, args...))
@@ -22,8 +22,10 @@
// The fix contained in the commit this message was committed with resets the deadline whenever // The fix contained in the commit this message was committed with resets the deadline whenever
// a heartbeat is received from the server. // a heartbeat is received from the server.
// //
//
// How to run this integration test: // How to run this integration test:
// //
//
// Terminal 1: // Terminal 1:
// $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go -mode server -addr 127.0.0.1:12345 // $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go -mode server -addr 127.0.0.1:12345
// rpc/dataconn/heartbeatconn: send heartbeat // rpc/dataconn/heartbeatconn: send heartbeat
@@ -39,6 +41,8 @@
// rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>) // rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
// rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout // rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
// ... // ...
//
// You should observe
package main package main
import ( import (
@@ -5,10 +5,12 @@
// ./microbenchmark -appmode server | pv -r > /dev/null // ./microbenchmark -appmode server | pv -r > /dev/null
// ./microbenchmark -appmode client -direction recv < /dev/zero // ./microbenchmark -appmode client -direction recv < /dev/zero
// //
//
// Without the overhead of pipes (just protocol performance, mostly useful with perf bc no bw measurement) // Without the overhead of pipes (just protocol performance, mostly useful with perf bc no bw measurement)
// //
// ./microbenchmark -appmode client -direction recv -devnoopWriter -devnoopReader // ./microbenchmark -appmode client -direction recv -devnoopWriter -devnoopReader
// ./microbenchmark -appmode server -devnoopReader -devnoopWriter // ./microbenchmark -appmode server -devnoopReader -devnoopWriter
//
package main package main
import ( import (
+11 -7
View File
@@ -29,7 +29,7 @@ func WithLogger(ctx context.Context, log Logger) context.Context {
return context.WithValue(ctx, contextKeyLogger, log) return context.WithValue(ctx, contextKeyLogger, log)
} }
//nolint:deadcode,unused //nolint[:deadcode,unused]
func getLog(ctx context.Context) Logger { func getLog(ctx context.Context) Logger {
log, ok := ctx.Value(contextKeyLogger).(Logger) log, ok := ctx.Value(contextKeyLogger).(Logger)
if !ok { if !ok {
@@ -181,19 +181,23 @@ func (e *ReadStreamError) Error() string {
var _ net.Error = &ReadStreamError{} var _ net.Error = &ReadStreamError{}
func (e ReadStreamError) Timeout() bool { func (e ReadStreamError) netErr() net.Error {
if netErr, ok := e.Err.(net.Error); ok { if netErr, ok := e.Err.(net.Error); ok {
return netErr
}
return nil
}
func (e ReadStreamError) Timeout() bool {
if netErr := e.netErr(); netErr != nil {
return netErr.Timeout() return netErr.Timeout()
} }
return false return false
} }
// This function is deprecated in net.Error and since this
// function is not involved in .Accept() code path, nothing
// really needs this method to be here.
func (e ReadStreamError) Temporary() bool { func (e ReadStreamError) Temporary() bool {
if te, ok := e.Err.(interface{ Temporary() bool }); ok { if netErr := e.netErr(); netErr != nil {
return te.Temporary() return netErr.Temporary()
} }
return false return false
} }
-5
View File
@@ -232,12 +232,7 @@ var _ net.Error = (*closeStateErrConnectionClosed)(nil)
func (e *closeStateErrConnectionClosed) Error() string { func (e *closeStateErrConnectionClosed) Error() string {
return "connection closed" return "connection closed"
} }
func (e *closeStateErrConnectionClosed) Timeout() bool { return false } func (e *closeStateErrConnectionClosed) Timeout() bool { return false }
// This function is deprecated in net.Error and since this
// function is not involved in .Accept() code path, nothing
// really needs this method to be here.
func (e *closeStateErrConnectionClosed) Temporary() bool { return false } func (e *closeStateErrConnectionClosed) Temporary() bool { return false }
func (s *closeState) CloseEntry() error { func (s *closeState) CloseEntry() error {
+1 -1
View File
@@ -13,7 +13,7 @@ func init() {
} }
} }
//nolint:deadcode,unused //nolint[:deadcode,unused]
func debug(format string, args ...interface{}) { func debug(format string, args ...interface{}) {
if debugEnabled { if debugEnabled {
fmt.Fprintf(os.Stderr, "rpc/dataconn/stream: %s\n", fmt.Sprintf(format, args...)) fmt.Fprintf(os.Stderr, "rpc/dataconn/stream: %s\n", fmt.Sprintf(format, args...))
@@ -1,5 +1,25 @@
// Package netadaptor implements an adaptor from // Package netadaptor implements an adaptor from
// transport.AuthenticatedListener to net.Listener. // transport.AuthenticatedListener to net.Listener.
//
// In contrast to transport.AuthenticatedListener,
// net.Listener is commonly expected (e.g. by net/http.Server.Serve),
// to return errors that fulfill the Temporary interface:
// interface Temporary { Temporary() bool }
// Common behavior of packages consuming net.Listener is to return
// from the serve-loop if an error returned by Accept is not Temporary,
// i.e., does not implement the interface or is !net.Error.Temporary().
//
// The zrepl transport infrastructure was written with the
// idea that Accept() may return any kind of error, and the consumer
// would just log the error and continue calling Accept().
// We have to adapt these listeners' behavior to the expectations
// of net/http.Server.
//
// Hence, Listener does not return an error at all but blocks the
// caller of Accept() until we get a (successfully authenticated)
// connection without errors from the transport.
// Accept errors returned from the transport are logged as errors
// to the logger passed on initialization.
package netadaptor package netadaptor
import ( import (
+2 -2
View File
@@ -188,8 +188,8 @@ func (c *Client) WaitForConnectivity(ctx context.Context) error {
data, dataErr := c.dataClient.ReqPing(ctx, &req) data, dataErr := c.dataClient.ReqPing(ctx, &req)
// dataClient uses transport.Connecter, which doesn't expose WaitForReady(true) // dataClient uses transport.Connecter, which doesn't expose WaitForReady(true)
// => we need to mask dial timeouts // => we need to mask dial timeouts
if err, ok := dataErr.(interface{ Timeout() bool }); ok && err.Timeout() { if err, ok := dataErr.(interface{ Temporary() bool }); ok && err.Temporary() {
// Rate-limit pings here in case Timeout() == true is a mis-classification // Rate-limit pings here in case Temporary() is a mis-classification
// or returns immediately (this is a tight loop in that case) // or returns immediately (this is a tight loop in that case)
// TODO keep this in lockstep with controlClient // TODO keep this in lockstep with controlClient
// => don't use FailFast for control, but check that both control and data worked // => don't use FailFast for control, but check that both control and data worked
+1 -1
View File
@@ -14,7 +14,7 @@ func init() {
} }
} }
//nolint:deadcode,unused //nolint[:deadcode,unused]
func debug(format string, args ...interface{}) { func debug(format string, args ...interface{}) {
if debugEnabled { if debugEnabled {
fmt.Fprintf(os.Stderr, "rpc: %s\n", fmt.Sprintf(format, args...)) fmt.Fprintf(os.Stderr, "rpc: %s\n", fmt.Sprintf(format, args...))
+4 -2
View File
@@ -3,7 +3,7 @@
// The zrepl documentation refers to the client as the // The zrepl documentation refers to the client as the
// `active side` and to the server as the `passive side`. // `active side` and to the server as the `passive side`.
// //
// # Design Considerations // Design Considerations
// //
// zrepl has non-standard requirements to remote procedure calls (RPC): // zrepl has non-standard requirements to remote procedure calls (RPC):
// whereas the coordination of replication (the planning phase) mostly // whereas the coordination of replication (the planning phase) mostly
@@ -35,7 +35,7 @@
// //
// Hence, this package attempts to combine the best of both worlds: // Hence, this package attempts to combine the best of both worlds:
// //
// # GRPC for Coordination and Dataconn for Bulk Data Transfer // GRPC for Coordination and Dataconn for Bulk Data Transfer
// //
// This package's Client uses its transport.Connecter to maintain // This package's Client uses its transport.Connecter to maintain
// separate control and data connections to the Server. // separate control and data connections to the Server.
@@ -107,6 +107,8 @@
// | Handler | // | Handler |
// +------------+ // +------------+
// (usually endpoint.{Sender,Receiver}) // (usually endpoint.{Sender,Receiver})
//
//
package rpc package rpc
// edit trick for the ASCII art above: // edit trick for the ASCII art above:
+1 -1
View File
@@ -9,7 +9,7 @@ import (
type Logger = logger.Logger type Logger = logger.Logger
// All fields must be non-nil /// All fields must be non-nil
type Loggers struct { type Loggers struct {
General Logger General Logger
Control Logger Control Logger
+2 -41
View File
@@ -33,47 +33,8 @@ var _ net.Error = &HandshakeError{}
func (e HandshakeError) Error() string { return e.msg } func (e HandshakeError) Error() string { return e.msg }
// When a net.Listener.Accept() returns an error, the server must // Like with net.OpErr (Go issue 6163), a client failing to handshake
// decide whether to retry calling Accept() or not. // should be a temporary Accept error toward the Listener .
// On some platforms (e.g., Linux), Accept() can return errors
// related to the specific protocol connection that was supposed
// to be returned as asocket FD. Obviously, we want to ignore,
// maybe log, those errors and retry Accept() immediately to
// serve other connections.
// But there are also conditions where we get Accept() errors because
// the process has run out of file descriptors. In that case, retrying
// won't help. We need to close some file descriptor to make progress.
// Note that there could be lots of open file descriptors because we
// have accepted, and not yet closed, lots of connections in the past.
// And then, of course there can be errors where we just want
// to return, e.g., if there's a programming error and we're getting
// an EBADFD or whatever.
//
// So, the serve loops in net/http.Server.Serve() or gRPC's server.Serve()
// must inspect the error and decide what to do.
// The vehicle for this is the
//
// interface { Temporary() bool }
//
// Behavior in both of the aforementioned Serve() loops:
//
// - if the error doesn't implement the interface, stop serving and return
// - `Temporary() == true`: retry with back-off
// - `Temporary() == false`: stop serving and return
//
// So, to make this package's HandshakeListener work with these
// Serve() loops, we return Temporary() == true if the handshake fails.
// In the aforementioned categories, that's the case of a per-connection
// protocol error.
//
// Note: the net.Error interface has deprecated the Temporary() method
// in go.dev/issue/45729, but there is no replacement for users of .Accept().
// Existing users of .Accept() continue to check for the interface.
// So, we need to continue supporting Temporary() until there's a different
// mechanism for serve loops to decide whether to retry or not.
// The following mailing list post proposes to eliminate the retries
// completely, but it seems like the effort has stalled.
// https://groups.google.com/g/golang-nuts/c/-JcZzOkyqYI/m/xwaZzjCgAwAJ
func (e HandshakeError) Temporary() bool { func (e HandshakeError) Temporary() bool {
if e.isAcceptError { if e.isAcceptError {
return true return true
+1
View File
@@ -9,6 +9,7 @@
// defer a.l.Unlock().Lock() // defer a.l.Unlock().Lock()
// fssesDone.Wait() // fssesDone.Wait()
// }() // }()
//
package chainlock package chainlock
import "sync" import "sync"
+1
View File
@@ -28,4 +28,5 @@
// f(Config{}) // crashes // f(Config{}) // crashes
// //
// f Config{ CriticalSetting: &nodefault.Bool{B: false}} // doesn't crash // f Config{ CriticalSetting: &nodefault.Bool{B: false}} // doesn't crash
//
package nodefault package nodefault
@@ -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()
}
}
}
+1
View File
@@ -53,6 +53,7 @@ var componentValidChar = regexp.MustCompile(`^[0-9a-zA-Z-_\.: ]+$`)
// characters: // characters:
// //
// [-_.: ] // [-_.: ]
//
func ComponentNamecheck(datasetPathComponent string) error { func ComponentNamecheck(datasetPathComponent string) error {
if len(datasetPathComponent) == 0 { if len(datasetPathComponent) == 0 {
return fmt.Errorf("path component must not be empty") return fmt.Errorf("path component must not be empty")
+4 -25
View File
@@ -343,6 +343,7 @@ const (
type SendStream struct { type SendStream struct {
cmd *zfscmd.Cmd cmd *zfscmd.Cmd
kill context.CancelFunc
stdoutReader io.ReadCloser // not *os.File for mocking during platformtest stdoutReader io.ReadCloser // not *os.File for mocking during platformtest
stderrBuf *circlog.CircularLog stderrBuf *circlog.CircularLog
@@ -415,29 +416,8 @@ func (s *SendStream) killAndWait() error {
debug("sendStream: killAndWait enter") debug("sendStream: killAndWait enter")
defer debug("sendStream: killAndWait leave") defer debug("sendStream: killAndWait leave")
// ensure this function is only called once
if s.state != sendStreamOpen {
panic(s.state)
}
// send SIGKILL // send SIGKILL
// In an earlier version, we used the starting context.Context's cancel function s.kill()
// 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. // Close our read-end of the pipe.
// //
@@ -983,10 +963,10 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*SendStream, e
stream := &SendStream{ stream := &SendStream{
cmd: cmd, cmd: cmd,
kill: cancel,
stdoutReader: stdoutReader, stdoutReader: stdoutReader,
stderrBuf: stderrBuf, stderrBuf: stderrBuf,
} }
_ = cancel // the SendStream.killAndWait() will kill the process
return stream, nil return stream, nil
} }
@@ -1042,10 +1022,8 @@ func (s *DrySendInfo) unmarshalZFSOutput(output []byte) (err error) {
} }
// unmarshal info line, looks like this: // unmarshal info line, looks like this:
//
// full zroot/test/a@1 5389768 // full zroot/test/a@1 5389768
// incremental zroot/test/a@1 zroot/test/a@2 5383936 // incremental zroot/test/a@1 zroot/test/a@2 5383936
//
// => see test cases // => see test cases
func (s *DrySendInfo) unmarshalInfoLine(l string) (regexMatched bool, err error) { func (s *DrySendInfo) unmarshalInfoLine(l string) (regexMatched bool, err error) {
@@ -1877,6 +1855,7 @@ var ErrBookmarkCloningNotSupported = fmt.Errorf("bookmark cloning feature is not
// unless a bookmark with the name `bookmark` exists and has the same idenitty (zfs.FilesystemVersionEqualIdentity) // unless a bookmark with the name `bookmark` exists and has the same idenitty (zfs.FilesystemVersionEqualIdentity)
// //
// v must be validated by the caller // v must be validated by the caller
//
func ZFSBookmark(ctx context.Context, fs string, v FilesystemVersion, bookmark string) (bm FilesystemVersion, err error) { func ZFSBookmark(ctx context.Context, fs string, v FilesystemVersion, bookmark string) (bm FilesystemVersion, err error) {
bm = FilesystemVersion{ bm = FilesystemVersion{
+1 -1
View File
@@ -13,7 +13,7 @@ func init() {
} }
} }
//nolint:deadcode,unused //nolint[:deadcode,unused]
func debug(format string, args ...interface{}) { func debug(format string, args ...interface{}) {
if debugEnabled { if debugEnabled {
fmt.Fprintf(os.Stderr, "zfs: %s\n", fmt.Sprintf(format, args...)) fmt.Fprintf(os.Stderr, "zfs: %s\n", fmt.Sprintf(format, args...))