Compare commits

..

3 Commits

Author SHA1 Message Date
Christian Schwarz 83e8bd8b8f fixups
cc @asomers
2021-10-09 15:37:03 +02:00
Christian Schwarz fc9c9b184e implement test server and client to diagnose certificate issues
cc @asomers
2021-10-09 15:28:00 +02:00
Christian Schwarz 6f11e92801 stderr debugging for #517 (tls: bad certificate)
refs #517

cc @asomers
2021-09-25 22:11:59 +02:00
67 changed files with 599 additions and 1599 deletions
+6 -4
View File
@@ -150,7 +150,7 @@ parameters:
release_docker_baseimage_tag:
type: string
default: "1.17"
default: "1.16"
workflows:
version: 2
@@ -160,15 +160,15 @@ 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.16
goversion: &latest-go-release "1.16"
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.16 #quickcheck-go-smoketest.name
matrix: &quickcheck-go-matrix
alias: quickcheck-go-matrix
parameters:
@@ -231,6 +231,8 @@ jobs:
- install-docdep
- run: make docs
- store_artifacts:
path: artifacts
- download-and-install-minio-client
- upload-minio:
src: artifacts
+1 -3
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)
GOLANGCI_LINT := golangci-lint
GOCOVMERGE := gocovmerge
RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.17
RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.16
RELEASE_DOCKER_BASEIMAGE ?= golang:$(RELEASE_DOCKER_BASEIMAGE_TAG)
ifneq ($(GOARM),)
@@ -189,8 +189,6 @@ GO_SUPPORTS_ILLUMOS := $(shell $(GO) version | gawk -F '.' '/^go version /{split
bins-all:
$(MAKE) $(BINS_ALL_TARGETS) GOOS=freebsd GOARCH=amd64
$(MAKE) $(BINS_ALL_TARGETS) GOOS=freebsd GOARCH=386
$(MAKE) $(BINS_ALL_TARGETS) GOOS=freebsd GOARCH=arm GOARM=7
$(MAKE) $(BINS_ALL_TARGETS) GOOS=freebsd GOARCH=arm64
$(MAKE) $(BINS_ALL_TARGETS) GOOS=linux GOARCH=amd64
$(MAKE) $(BINS_ALL_TARGETS) GOOS=linux GOARCH=arm64
$(MAKE) $(BINS_ALL_TARGETS) GOOS=linux GOARCH=arm GOARM=7
+1 -1
View File
@@ -95,7 +95,7 @@ Downstream packagers can read the changelog to determine whether they want to pu
### Additional Notes to Distro Package Maintainers
* Run the platform tests (Docs -> Usage -> Platform Tests) **on a test system** to validate that zrepl's abstractions on top of ZFS work with the system ZFS.
* Use `sudo make test-platform-bin && sudo make test-platform` **on a test system** to validate that zrepl's abstractions on top of ZFS work with the system ZFS.
* Ship a default config that adheres to your distro's `hier` and logging system.
* Ship a service manager file and _please_ try to upstream it to this repository.
* `dist/systemd` contains a Systemd unit template.
-1
View File
@@ -1,4 +1,3 @@
//go:build tools
// +build tools
package main
+1 -9
View File
@@ -2,22 +2,14 @@ package viewmodel
import (
"fmt"
"math"
)
func ByteCountBinaryUint(b uint64) string {
if b > math.MaxInt64 {
panic(b)
}
return ByteCountBinary(int64(b))
}
func ByteCountBinary(b int64) string {
const unit = 1024
if b < unit {
return fmt.Sprintf("%d B", b)
}
div, exp := unit, 0
div, exp := int64(unit), 0
for n := b / unit; n >= unit; n /= unit {
div *= unit
exp++
@@ -4,7 +4,7 @@ import "time"
type byteProgressMeasurement struct {
time time.Time
val uint64
val int64
}
type bytesProgressHistory struct {
@@ -13,7 +13,7 @@ type bytesProgressHistory struct {
lastChange time.Time
}
func (p *bytesProgressHistory) Update(currentVal uint64) (bytesPerSecondAvg int64, changeCount int) {
func (p *bytesProgressHistory) Update(currentVal int64) (bytesPerSecondAvg int64, changeCount int) {
if p.last == nil {
p.last = &byteProgressMeasurement{
@@ -33,12 +33,7 @@ func (p *bytesProgressHistory) Update(currentVal uint64) (bytesPerSecondAvg int6
return 0, 0
}
var deltaV int64
if currentVal >= p.last.val {
deltaV = int64(currentVal - p.last.val)
} else {
deltaV = -int64(p.last.val - currentVal)
}
deltaV := currentVal - p.last.val
deltaT := time.Since(p.last.time)
rate := float64(deltaV) / deltaT.Seconds()
+7 -12
View File
@@ -216,7 +216,7 @@ func drawJob(t *stringbuilder.B, name string, v *job.Status, history *bytesProgr
}
}
func printFilesystemStatus(t *stringbuilder.B, rep *report.FilesystemReport, maxFS int) {
func printFilesystemStatus(t *stringbuilder.B, rep *report.FilesystemReport, active bool, maxFS int) {
expected, replicated, containsInvalidSizeEstimates := rep.BytesSum()
sizeEstimationImpreciseNotice := ""
@@ -227,20 +227,15 @@ func printFilesystemStatus(t *stringbuilder.B, rep *report.FilesystemReport, max
sizeEstimationImpreciseNotice = " (step lacks size estimation)"
}
userVisisbleCurrentStep, userVisibleTotalSteps := rep.CurrentStep, len(rep.Steps)
if len(rep.Steps) > 0 {
userVisisbleCurrentStep = rep.CurrentStep + 1 // CurrentStep is an index that starts at 0
}
status := fmt.Sprintf("%s (step %d/%d, %s/%s)%s",
strings.ToUpper(string(rep.State)),
userVisisbleCurrentStep, userVisibleTotalSteps,
ByteCountBinaryUint(replicated), ByteCountBinaryUint(expected),
rep.CurrentStep, len(rep.Steps),
ByteCountBinary(replicated), ByteCountBinary(expected),
sizeEstimationImpreciseNotice,
)
activeIndicator := " "
if rep.BlockedOn == report.FsBlockedOnNothing &&
(rep.State == report.FilesystemPlanning || rep.State == report.FilesystemStepping) {
if active {
activeIndicator = "*"
}
t.AddIndent(1)
@@ -363,12 +358,12 @@ func renderReplicationReport(t *stringbuilder.B, rep *report.Report, history *by
rate, changeCount := history.Update(replicated)
eta := time.Duration(0)
if rate > 0 {
eta = time.Duration((float64(expected)-float64(replicated))/float64(rate)) * time.Second
eta = time.Duration((expected-replicated)/rate) * time.Second
}
t.Write("Progress: ")
t.DrawBar(50, replicated, expected, changeCount)
t.Write(fmt.Sprintf(" %s / %s @ %s/s", ByteCountBinaryUint(replicated), ByteCountBinaryUint(expected), ByteCountBinary(rate)))
t.Write(fmt.Sprintf(" %s / %s @ %s/s", ByteCountBinary(replicated), ByteCountBinary(expected), ByteCountBinary(rate)))
if eta != 0 {
t.Write(fmt.Sprintf(" (%s remaining)", humanizeDuration(eta)))
}
@@ -390,7 +385,7 @@ func renderReplicationReport(t *stringbuilder.B, rep *report.Report, history *by
}
}
for _, fs := range latest.Filesystems {
printFilesystemStatus(t, fs, maxFSLen)
printFilesystemStatus(t, fs, false, maxFSLen) // FIXME bring 'active' flag back
}
}
@@ -99,11 +99,11 @@ func RightPad(str string, length int, pad string) string {
}
// changeCount = 0 indicates stall / no progress
func (w *B) DrawBar(length int, bytes, totalBytes uint64, changeCount int) {
func (w *B) DrawBar(length int, bytes, totalBytes int64, changeCount int) {
const arrowPositions = `>\|/`
var completedLength int
if totalBytes > 0 {
completedLength = int(uint64(length) * bytes / totalBytes)
completedLength = int(int64(length) * bytes / totalBytes)
if completedLength > length {
completedLength = length
}
+1 -9
View File
@@ -139,11 +139,9 @@ var testPlaceholder = &cli.Subcommand{
func runTestPlaceholder(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
var checkDPs []*zfs.DatasetPath
var datasetWasExplicitArgument bool
// all actions first
if testPlaceholderArgs.all {
datasetWasExplicitArgument = false
out, err := zfs.ZFSList(ctx, []string{"name"})
if err != nil {
return errors.Wrap(err, "could not list ZFS filesystems")
@@ -156,7 +154,6 @@ func runTestPlaceholder(ctx context.Context, subcommand *cli.Subcommand, args []
checkDPs = append(checkDPs, dp)
}
} else {
datasetWasExplicitArgument = true
dp, err := zfs.NewDatasetPath(testPlaceholderArgs.ds)
if err != nil {
return err
@@ -174,12 +171,7 @@ func runTestPlaceholder(ctx context.Context, subcommand *cli.Subcommand, args []
return errors.Wrap(err, "cannot get placeholder state")
}
if !ph.FSExists {
if datasetWasExplicitArgument {
return errors.Errorf("filesystem %q does not exist", ph.FS)
} else {
// got deleted between ZFSList and ZFSGetFilesystemPlaceholderState
continue
}
panic("placeholder state inconsistent: filesystem " + ph.FS + " must exist in this context")
}
is := "yes"
if !ph.IsPlaceholder {
+1 -7
View File
@@ -86,7 +86,7 @@ type SendOptions struct {
BackupProperties bool `yaml:"backup_properties,optional,default=false"`
LargeBlocks bool `yaml:"large_blocks,optional,default=false"`
Compressed bool `yaml:"compressed,optional,default=false"`
EmbeddedData bool `yaml:"embedded_data,optional,default=false"`
EmbeddedData bool `yaml:"embbeded_data,optional,default=false"`
Saved bool `yaml:"saved,optional,default=false"`
BandwidthLimit *BandwidthLimit `yaml:"bandwidth_limit,optional,fromdefaults"`
@@ -101,8 +101,6 @@ type RecvOptions struct {
Properties *PropertyRecvOptions `yaml:"properties,fromdefaults"`
BandwidthLimit *BandwidthLimit `yaml:"bandwidth_limit,optional,fromdefaults"`
Placeholder *PlaceholderRecvOptions `yaml:"placeholder,fromdefaults"`
}
var _ yaml.Unmarshaler = &datasizeunit.Bits{}
@@ -132,10 +130,6 @@ type PropertyRecvOptions struct {
Override map[zfsprop.Property]string `yaml:"override,optional"`
}
type PlaceholderRecvOptions struct {
Encryption string `yaml:"encryption,default=unspecified"`
}
type PushJob struct {
ActiveJob `yaml:",inline"`
Snapshotting SnapshottingEnum `yaml:"snapshotting"`
@@ -9,7 +9,4 @@ jobs:
key: "/etc/zrepl/backups.key"
client_cns:
- "prod"
recv:
placeholder:
encryption: inherit # use 'off' if sender uses send.encrypted
root_fs: "storage/zrepl/sink"
+4 -6
View File
@@ -8,7 +8,6 @@ import (
"io"
"net"
"net/http"
"os"
"time"
"github.com/pkg/errors"
@@ -125,9 +124,8 @@ func (j *controlJob) Run(ctx context.Context) {
s := Status{
Jobs: jobs,
Global: GlobalStatus{
ZFSCmds: globalZFS,
Envconst: envconstReport,
OsEnviron: os.Environ(),
ZFSCmds: globalZFS,
Envconst: envconstReport,
}}
return s, nil
}})
@@ -158,8 +156,8 @@ func (j *controlJob) Run(ctx context.Context) {
server := http.Server{
Handler: mux,
// control socket is local, 1s timeout should be more than sufficient, even on a loaded system
WriteTimeout: envconst.Duration("ZREPL_DAEMON_CONTROL_SERVER_WRITE_TIMEOUT", 1*time.Second),
ReadTimeout: envconst.Duration("ZREPL_DAEMON_CONTROL_SERVER_READ_TIMEOUT", 1*time.Second),
WriteTimeout: 1 * time.Second,
ReadTimeout: 1 * time.Second,
}
outer:
+2 -3
View File
@@ -160,9 +160,8 @@ type Status struct {
}
type GlobalStatus struct {
ZFSCmds *zfscmd.Report
Envconst *envconst.Report
OsEnviron []string
ZFSCmds *zfscmd.Report
Envconst *envconst.Report
}
func (s *jobs) status() map[string]*job.Status {
+2 -14
View File
@@ -42,7 +42,6 @@ type ActiveSide struct {
promPruneSecs *prometheus.HistogramVec // labels: prune_side
promBytesReplicated *prometheus.CounterVec // labels: filesystem
promReplicationErrors prometheus.Gauge
promLastSuccessful prometheus.Gauge
tasksMtx sync.Mutex
tasks activeSideTasks
@@ -322,6 +321,7 @@ func activeSide(g *config.Global, in *config.ActiveJob, configJob interface{}) (
Help: "number of bytes replicated from sender to receiver per filesystem",
ConstLabels: prometheus.Labels{"zrepl_job": j.name.String()},
}, []string{"filesystem"})
j.promReplicationErrors = prometheus.NewGauge(prometheus.GaugeOpts{
Namespace: "zrepl",
Subsystem: "replication",
@@ -329,13 +329,6 @@ func activeSide(g *config.Global, in *config.ActiveJob, configJob interface{}) (
Help: "number of filesystems that failed replication in the latest replication attempt, or -1 if the job failed before enumerating the filesystems",
ConstLabels: prometheus.Labels{"zrepl_job": j.name.String()},
})
j.promLastSuccessful = prometheus.NewGauge(prometheus.GaugeOpts{
Namespace: "zrepl",
Subsystem: "replication",
Name: "last_successful",
Help: "timestamp of last successful replication",
ConstLabels: prometheus.Labels{"zrepl_job": j.name.String()},
})
j.connecter, err = fromconfig.ConnecterFromConfig(g, in.Connect)
if err != nil {
@@ -367,7 +360,6 @@ func (j *ActiveSide) RegisterMetrics(registerer prometheus.Registerer) {
registerer.MustRegister(j.promPruneSecs)
registerer.MustRegister(j.promBytesReplicated)
registerer.MustRegister(j.promReplicationErrors)
registerer.MustRegister(j.promLastSuccessful)
}
func (j *ActiveSide) Name() string { return j.name.String() }
@@ -502,11 +494,7 @@ func (j *ActiveSide) do(ctx context.Context) {
repCancel() // always cancel to free up context resources
replicationReport := j.tasks.replicationReport()
var numErrors = replicationReport.GetFailedFilesystemsCountInLatestAttempt()
j.promReplicationErrors.Set(float64(numErrors))
if numErrors == 0 {
j.promLastSuccessful.SetToCurrentTime()
}
j.promReplicationErrors.Set(float64(replicationReport.GetFailedFilesystemsCountInLatestAttempt()))
endSpan()
}
-12
View File
@@ -72,16 +72,6 @@ func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoi
return rc, errors.Wrap(err, "cannot build bandwith limit config")
}
placeholderEncryption, err := endpoint.PlaceholderCreationEncryptionPropertyString(recvOpts.Placeholder.Encryption)
if err != nil {
options := []string{}
for _, v := range endpoint.PlaceholderCreationEncryptionPropertyValues() {
options = append(options, endpoint.PlaceholderCreationEncryptionProperty(v).String())
}
return rc, errors.Errorf("placeholder encryption value %q is invalid, must be one of %s",
recvOpts.Placeholder.Encryption, options)
}
rc = endpoint.ReceiverConfig{
JobID: jobID,
RootWithoutClientComponent: rootFs,
@@ -91,8 +81,6 @@ func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoi
OverrideProperties: recvOpts.Properties.Override,
BandwidthLimit: bwlim,
PlaceholderEncryption: placeholderEncryption,
}
if err := rc.Validate(); err != nil {
return rc, errors.Wrap(err, "cannot build receiver config")
+3 -7
View File
@@ -10,7 +10,6 @@ import (
"github.com/prometheus/client_golang/prometheus"
"github.com/zrepl/zrepl/daemon/logging/trace"
"github.com/zrepl/zrepl/util/bandwidthlimit"
"github.com/zrepl/zrepl/util/nodefault"
"github.com/zrepl/zrepl/config"
@@ -147,7 +146,7 @@ type alwaysUpToDateReplicationCursorHistory struct {
target pruner.Target
}
var _ pruner.Sender = (*alwaysUpToDateReplicationCursorHistory)(nil)
var _ pruner.History = (*alwaysUpToDateReplicationCursorHistory)(nil)
func (h alwaysUpToDateReplicationCursorHistory) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) {
fsvReq := &pdu.ListFilesystemVersionsReq{
@@ -180,11 +179,8 @@ func (j *SnapJob) doPrune(ctx context.Context) {
sender := endpoint.NewSender(endpoint.SenderConfig{
JobID: j.name,
FSF: j.fsfilter,
// FIXME the following config fields are irrelevant for SnapJob
// because the endpoint is only used as pruner.Target.
// However, the implementation requires them to be set.
Encrypt: &nodefault.Bool{B: true},
BandwidthLimit: bandwidthlimit.NoLimitConfig(),
// FIXME encryption setting is irrelevant for SnapJob because the endpoint is only used as pruner.Target
Encrypt: &nodefault.Bool{B: true},
})
j.prunerMtx.Lock()
j.pruner = j.prunerFactory.BuildLocalPruner(ctx, sender, alwaysUpToDateReplicationCursorHistory{sender})
+1 -1
View File
@@ -246,7 +246,7 @@ func WithTask(ctx context.Context, taskName string) (context.Context, DoneFunc)
// the debugString can be quite long and panic won't print it completely
fmt.Fprintf(os.Stderr, "going to panic due to activeChildTasks:\n%s\n", this.debugString())
}
panic(errors.WithMessagef(ErrTaskStillHasActiveChildTasks, "end task: %v active child tasks (run daemon with env var %s=1 for more details)\n", this.activeChildTasks, debugEnabledEnvVar))
panic(errors.WithMessagef(ErrTaskStillHasActiveChildTasks, "end task: %v active child tasks\n", this.activeChildTasks))
}
// support idempotent task ends
+1 -3
View File
@@ -7,9 +7,7 @@ import (
"github.com/zrepl/zrepl/util/envconst"
)
const debugEnabledEnvVar = "ZREPL_TRACE_DEBUG_ENABLED"
var debugEnabled = envconst.Bool(debugEnabledEnvVar, false)
var debugEnabled = envconst.Bool("ZREPL_TRACE_DEBUG_ENABLED", false)
func debug(format string, args ...interface{}) {
if !debugEnabled {
+12 -20
View File
@@ -19,20 +19,12 @@ import (
"github.com/zrepl/zrepl/util/envconst"
)
// The sender in the replication setup.
// The pruner uses the Sender to determine which of the Target's filesystems need to be pruned.
// Also, it asks the Sender about the replication cursor of each filesystem
// to enable the 'not_replicated' pruning rule.
//
// Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
type Sender interface {
type History interface {
ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
}
// The pruning target, i.e., on which snapshots are destroyed.
// This can be a replication sender or receiver.
//
// Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
type Target interface {
ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
@@ -56,7 +48,7 @@ func GetLogger(ctx context.Context) Logger {
type args struct {
ctx context.Context
target Target
sender Sender
receiver History
rules []pruning.KeepRule
retryWait time.Duration
considerSnapAtCursorReplicated bool
@@ -140,12 +132,12 @@ func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus
return f, nil
}
func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, sender Sender) *Pruner {
func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, receiver History) *Pruner {
p := &Pruner{
args: args{
context.WithValue(ctx, contextKeyPruneSide, "sender"),
target,
sender,
receiver,
f.senderRules,
f.retryWait,
f.considerSnapAtCursorReplicated,
@@ -156,12 +148,12 @@ func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, se
return p
}
func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, sender Sender) *Pruner {
func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, receiver History) *Pruner {
p := &Pruner{
args: args{
context.WithValue(ctx, contextKeyPruneSide, "receiver"),
target,
sender,
receiver,
f.receiverRules,
f.retryWait,
false, // senseless here anyways
@@ -172,12 +164,12 @@ func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target,
return p
}
func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, history Sender) *Pruner {
func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, receiver History) *Pruner {
p := &Pruner{
args: args{
context.WithValue(ctx, contextKeyPruneSide, "local"),
target,
history,
receiver,
f.keepRules,
f.retryWait,
false, // considerSnapAtCursorReplicated is not relevant for local pruning
@@ -351,9 +343,9 @@ func (s snapshot) Date() time.Time { return s.date }
func doOneAttempt(a *args, u updater) {
ctx, target, sender := a.ctx, a.target, a.sender
ctx, target, receiver := a.ctx, a.target, a.receiver
sfssres, err := sender.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
sfssres, err := receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
if err != nil {
u(func(p *Pruner) {
p.state = PlanErr
@@ -418,7 +410,7 @@ tfss_loop:
rcReq := &pdu.ReplicationCursorReq{
Filesystem: tfs.Path,
}
rc, err := sender.ReplicationCursor(ctx, rcReq)
rc, err := receiver.ReplicationCursor(ctx, rcReq)
if err != nil {
pfsPlanErrAndLog(err, "cannot get replication cursor bookmark")
continue tfss_loop
@@ -464,7 +456,7 @@ tfss_loop:
})
}
if preCursor {
pfsPlanErrAndLog(fmt.Errorf("prune target has no snapshot that corresponds to sender replication cursor bookmark"), "")
pfsPlanErrAndLog(fmt.Errorf("replication cursor not found in prune target filesystem versions"), "")
continue tfss_loop
}
+18 -37
View File
@@ -16,44 +16,18 @@ 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.5
---
* |feature| :ref:`Bandwidth limiting <job-send-recv-options--bandwidth-limit>` (Thanks, Prominic.NET, Inc.)
* |feature| zrepl status: use a ``*`` to indicate which filesystem is currently replicating
* |feature| include daemon environment variables in zrepl status (currently only in ``--raw``)
* |bugfix| **fix encrypt-on-receive + placeholders use case** (:issue:`504`)
* Before this fix, **plain sends** to a receiver with an encrypted ``root_fs`` **could be received unencrypted** if zrepl needed to create placeholders on the receiver.
* Existing zrepl users should :ref:`read the docs <job-recv-options--placeholder>` and check ``zfs get -r encryption,zrepl:placeholder PATH_TO_ROOTFS`` on the receiver.
* Thanks to `@mologie <https://github.com/mologie>`_ and `@razielgn <https://github.com/razielgn>`_ for reporting and testing!
* |bugfix| Rename mis-spelled :ref:`send option <job-send-options>` ``embbeded_data`` to ``embedded_data``.
* |bugfix| zrepl status: replication step numbers should start at 1
* |bugfix| incorrect bandwidth averaging in ``zrepl status``.
* |bugfix| FreeBSD with OpenZFS 2.0: zrepl would wait indefinitely for zfs send to exit on timeouts.
* |bugfix| fix ``strconv.ParseInt: value out of range`` bug (and use the control RPCs).
* |docs| improve description of multiple pruning rules.
* |docs| document :ref:`platform tests <usage-platform-tests>`.
* |docs| quickstart: make users aware that prune rules apply to all snapshots.
* |maint| some platformtests were broken.
* |maint| FreeBSD: release armv7 and arm64 binaries.
* |maint| apt repo: update instructions due to ``apt-key`` deprecation.
Note to all users: please read up on the following OpenZFS bugs, as you might be affected:
* `ZFS send/recv with ashift 9->12 leads to data corruption <https://github.com/openzfs/zfs/issues/12762>`_.
* Various bugs with encrypted send/recv (`Leadership meeting notes <https://openzfs.topicbox.com/groups/developer/T24bdaa2886c6cbf5-Mc039a11c3f1507ea0664817b/december-openzfs-leadership-meeting>`_)
Finally, I'd like to point you to the `GitHub discussion <https://github.com/zrepl/zrepl/discussions/547>`_ about which bugfixes and features should be prioritized in zrepl 0.6 and beyond!
.. NOTE::
| zrepl is a spare-time project primarily developed by `Christian Schwarz <https://cschwarz.com>`_.
| You can support maintenance and feature development through one of the following services:
| |Donate via Patreon| |Donate via GitHub Sponsors| |Donate via Liberapay| |Donate via PayPal|
| Note that PayPal processing fees are relatively high for small donations.
| For SEPA wire transfer and **commercial support**, please `contact Christian directly <https://cschwarz.com>`_.
We use the following annotations for classifying changes:
* |break_config| Change that breaks the config.
As a package maintainer, make sure to warn your users about config breakage somehow.
* |break| Change that breaks interoperability or persistent state representation with previous releases.
As a package maintainer, make sure to warn your users about config breakage somehow.
Note that even updating the package on both sides might not be sufficient, e.g. if persistent state needs to be migrated to a new format.
* |mig| Migration that must be run by the user.
* |feature| Change that introduces new functionality.
* |bugfix| Change that fixes a bug, no regressions or incompatibilities expected.
* |docs| Change to the documentation.
* |maint| Maintenance changes.
0.4.0
-----
@@ -79,6 +53,13 @@ The following bugfix in 0.3.1 :issue:`caused problems for some users <400>`:
* |bugfix| pruning: ``grid``: add all snapshots that do not match the regex to the rule's destroy list.
.. NOTE::
| zrepl is a spare-time project primarily developed by `Christian Schwarz <https://cschwarz.com>`_.
| You can support maintenance and feature development through one of the following services:
| |Donate via Patreon| |Donate via GitHub Sponsors| |Donate via Liberapay| |Donate via PayPal|
| Note that PayPal processing fees are relatively high for small donations.
| For SEPA wire transfer and **commercial support**, please `contact Christian directly <https://cschwarz.com>`_.
0.3.1
-----
+11 -12
View File
@@ -67,9 +67,8 @@ Policy ``not_replicated``
...
``not_replicated`` keeps all snapshots that have not been replicated to the receiving side.
It only makes sense to specify this rule for the ``keep_sender``.
The reason is that, by definition, all snapshots on the receiver have already been replicated to there from the sender.
To determine whether a sender-side snapshot has already been replicated, zrepl uses the :ref:`replication cursor bookmark <replication-cursor-and-last-received-hold>` which corresponds to the most recent successfully replicated snapshot.
It only makes sense to specify this rule on a sender (source or push job).
The state required to evaluate this rule is stored in the :ref:`replication cursor bookmark <replication-cursor-and-last-received-hold>` on the sending side.
.. _prune-keep-retention-grid:
@@ -107,7 +106,7 @@ The following procedure happens during pruning:
#. All subsequent buckets are placed adjacent to their predecessor bucket.
#. Now each snapshot on the axis either falls into one bucket or it is older than our rightmost bucket.
Buckets are left-inclusive and right-exclusive which means that a snapshot on the edge of bucket will always 'fall into the right one'.
#. Snapshots older than the rightmost bucket are **not kept** by the grid specification.
#. Snapshots older than the rightmost bucket **not kept** by this gridspec.
#. For each bucket, we only keep the ``keep`` oldest snapshots.
The syntax to describe the bucket list is as follows:
@@ -125,11 +124,11 @@ The syntax to describe the bucket list is as follows:
::
Assume the following grid specification:
Assume the following grid spec:
grid: 1x1h(keep=all) | 2x2h | 1x3h
This grid specification produces the following constellation of buckets:
This grid spec produces the following constellation of buckets:
0h 1h 2h 3h 4h 5h 6h 7h 8h 9h
| | | | | | | | | |
@@ -149,13 +148,13 @@ The syntax to describe the bucket list is as follows:
|-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
| keep=all| keep=1 | keep=1 | keep=1 |
| | | | |
| a b c | d e f g h i j k l m n o p |q r s t u v w x y z |A B C D
| a b c| d e f g h i j k l m n o p |q r s t u v w x y z |A B C D
We obtain the following mapping of snapshots to buckets:
The result is the following mapping of snapshots to buckets:
Bucket1: a,b,c
Bucket2: d,e,f,g,h,i
Bucket3: j,k,l,m,n,o,p
Bucket1: a, b, c
Bucket2: d,e,f,g,h,i,j
Bucket3: k,l,m,n,o,p
Bucket4: q,r,s,t,u,v,w,x,y,z
No bucket: A,B,C,D
@@ -169,7 +168,7 @@ The syntax to describe the bucket list is as follows:
| | | | | | | | | |
|-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
| | | | |
| a b c | i | p | z |
| a b c| j p | z |
.. _prune-keep-last-n:
+5 -42
View File
@@ -38,10 +38,10 @@ See the `upstream man page <https://openzfs.github.io/openzfs-docs/man/8/zfs-sen
- Specific to zrepl, :ref:`see below <job-send-options-encrypted>`.
* - ``bandwidth_limit``
-
- Specific to zrepl, :ref:`see below <job-send-recv-options--bandwidth-limit>`.
- Specific to zrepl, :ref:`see below <job-send-recv-options-bandwidth-limit>`.
* - ``raw``
- ``-w``
- Use ``encrypted`` to only allow encrypted sends. Mixed sends are not supported.
- Use ``encrypted`` to only allow encrypted sends.
* - ``send_properties``
- ``-p``
- **Be careful**, read the :ref:`note on property replication below <job-note-property-replication>`.
@@ -54,7 +54,7 @@ See the `upstream man page <https://openzfs.github.io/openzfs-docs/man/8/zfs-sen
* - ``compressed``
- ``-c``
-
* - ``embedded_data``
* - ``embbeded_data``
- ``-e``
-
* - ``saved``
@@ -141,16 +141,9 @@ Recv Options
override: {
"org.openzfs.systemd:ignore": "on"
}
bandwidth_limit: ...
placeholder:
encryption: unspecified | off | inherit
bandwidth_limit: ... # see below
...
Jump to
:ref:`properties <job-recv-options--inherit-and-override>` ,
:ref:`bandwidth_limit <job-send-recv-options--bandwidth-limit>` , and
:ref:`bandwidth_limit <job-recv-options--placeholder>`.
.. _job-recv-options--inherit-and-override:
``properties``
@@ -224,40 +217,10 @@ and property replication is enabled, the receiver must :ref:`inherit the followi
* ``keyformat``
* ``encryption``
.. _job-recv-options--placeholder:
Placeholders
------------
During replication, zrepl :ref:`creates placeholder datasets <replication-placeholder-property>` on the receiving side if the sending side's ``filesystems`` filter creates gaps in the dataset hierarchy.
This is generally fully transparent to the user.
However, with OpenZFS Native Encryption, placeholders require zrepl user attention.
Specifically, the problem is that, when zrepl attempts to create the placeholder dataset on the receiver, and that placeholder's parent dataset is encrypted, ZFS wants to inherit encryption to the placeholder.
This is relevant to two use cases that zrepl supports:
1. **encrypted-send-to-untrusted-receiver** In this use case, the sender sends an :ref:`encrypted send stream <job-send-options-encrypted>` and the receiver doesn't have the key loaded.
2. **send-plain-encrypt-on-receive** The receive-side ``root_fs`` dataset is encrypted, and the senders are unencrypted.
The key of ``root_fs`` is loaded, and the goal is that the plain sends (e.g., from production) are encrypted on-the-fly during receive, with ``root_fs``'s key.
For **encrypted-send-to-untrusted-receiver**, the placeholder datasets need to be created with ``-o encryption=off``.
Without it, creation would fail with an error, indicating that the placeholder's parent dataset's key needs to be loaded.
But we don't trust the receiver, so we can't expect that to ever happen.
However, for **send-plain-encrypt-on-receive**, we cannot set ``-o encryption=off``.
The reason is that if we did, any of the (non-placeholder) child datasets below the placeholder would inherit ``encryption=off``, thereby silently breaking our encrypt-on-receive use case.
So, to cover this use case, we need to create placeholders without specifying ``-o encryption``.
This will make ``zfs create`` inherit the encryption mode from the parent dataset, and thereby transitively from ``root_fs``.
The zrepl config provides the `recv.placeholder.encryption` knob to control this behavior.
In ``undefined`` mode (default), placeholder creation bails out and asks the user to configure a behavior.
In ``off`` mode, the placeholder is created with ``encryption=off``, i.e., **encrypted-send-to-untrusted-rceiver** use case.
In ``inherit`` mode, the placeholder is created without specifying ``-o encryption`` at all, i.e., the **send-plain-encrypt-on-receive** use case.
Common Options
~~~~~~~~~~~~~~
.. _job-send-recv-options--bandwidth-limit:
.. _job-send-recv-options-bandwidth-limit:
Bandwidth Limit (send & recv)
-----------------------------
+2 -3
View File
@@ -25,10 +25,10 @@ zrepl - ZFS replication
Progress: [=========================\----] 246.7 MiB / 264.7 MiB @ 11.5 MiB/s
zroot STEPPING (step 1/2, 624 B/1.2 KiB) next: @a => @b
zroot/ROOT DONE (step 2/2, 1.2 KiB/1.2 KiB)
* zroot/ROOT/default STEPPING (step 1/2, 123.4 MiB/129.3 MiB) next: @a => @b
zroot/ROOT/default STEPPING (step 1/2, 123.4 MiB/129.3 MiB) next: @a => @b
zroot/tmp STEPPING (step 1/2, 29.9 KiB/44.2 KiB) next: @a => @b
zroot/usr STEPPING (step 1/2, 624 B/1.2 KiB) next: @a => @b
* zroot/usr/home STEPPING (step 1/2, 123.3 MiB/135.3 MiB) next: @a => @b
zroot/usr/home STEPPING (step 1/2, 123.3 MiB/135.3 MiB) next: @a => @b
zroot/var STEPPING (step 1/2, 624 B/1.2 KiB) next: @a => @b
zroot/var/audit DONE (step 2/2, 1.2 KiB/1.2 KiB)
zroot/var/crash DONE (step 2/2, 1.2 KiB/1.2 KiB)
@@ -66,7 +66,6 @@ Main Features
* [x] Large blocks send & receive
* [x] Embedded data send & receive
* [x] Resume state send & receive
* [x] Bandwidth limiting
* **Automatic snapshot management**
+9 -20
View File
@@ -9,29 +9,18 @@ The fingerprint of the signing key is ``E101 418F D3D6 FBCB 9D65 A62D 7086 99FC
It is available at `<https://zrepl.cschwarz.com/apt/apt-key.asc>`_ .
Please open an issue in on GitHub if you encounter any issues with the repository.
The following snippet configure the repository for your Debian or Ubuntu release:
::
(
set -ex
zrepl_apt_key_url=https://zrepl.cschwarz.com/apt/apt-key.asc
zrepl_apt_key_dst=/usr/share/keyrings/zrepl.gpg
zrepl_apt_repo_file=/etc/apt/sources.list.d/zrepl.list
# Install dependencies for subsequent commands
sudo apt update && sudo apt install curl gnupg lsb-release
# Deploy the zrepl apt key.
curl -fsSL "$zrepl_apt_key_url" | tee | gpg --dearmor | sudo tee "$zrepl_apt_key_dst" > /dev/null
# Add the zrepl apt repo.
ARCH="$(dpkg --print-architecture)"
CODENAME="$(lsb_release -i -s | tr '[:upper:]' '[:lower:]') $(lsb_release -c -s | tr '[:upper:]' '[:lower:]')"
echo "Using Distro and Codename: $CODENAME"
echo "deb [arch=$ARCH signed-by=$zrepl_apt_key_dst] https://zrepl.cschwarz.com/apt/$CODENAME main" | sudo tee /etc/apt/sources.list.d/zrepl.list
# Update apt repos.
sudo apt update && sudo apt install curl gnupg lsb-release; \
ARCH="$(dpkg --print-architecture)"; \
CODENAME="$(lsb_release -i -s | tr '[:upper:]' '[:lower:]') $(lsb_release -c -s | tr '[:upper:]' '[:lower:]')"; \
echo "Using Distro and Codename: $CODENAME"; \
(curl https://zrepl.cschwarz.com/apt/apt-key.asc | sudo apt-key add -) && \
(echo "deb [arch=$ARCH] https://zrepl.cschwarz.com/apt/$CODENAME main" | sudo tee /etc/apt/sources.list.d/zrepl.list) && \
sudo apt update
)
.. NOTE::
@@ -93,7 +93,6 @@ Enable the zrepl daemon to start automatically at boot:
sysrc zrepl_enable="YES"
Now jump to :ref:`the summary <installation-freebsd-jail-summary>` below.
Plugin
######
@@ -135,18 +134,7 @@ Now ``zrepl`` can be started.
service zrepl start
Now jump to :ref:`the summary <installation-freebsd-jail-summary>` below.
.. _installation-freebsd-jail-summary:
Summary
-------
Congratulations, you have a working jail!
.. NOTE::
With FreeBSD 13's transition to OpenZFS 2.0, please ensure that your jail's FreeBSD version matches the one in the kernel module.
If you are getting cryptic errors such as
``cannot receive new filesystem stream: invalid backup stream``
the instructions posted `here <https://github.com/zrepl/zrepl/issues/500#issuecomment-966215205>`_ might help.
-13
View File
@@ -51,19 +51,6 @@ We hope that you have found a configuration that fits your use case.
Use ``zrepl configcheck`` once again to make sure the config is correct (output indicates that everything is fine).
Then restart the zrepl daemon on all systems involved in the replication, likely using ``service zrepl restart`` or ``systemctl restart zrepl``.
.. WARNING::
Please :ref:`read up carefully <prune>` on the pruning rules before applying the config.
In particular, note that most example configs apply to all snapshots, not just zrepl-created snapshots.
Use the following keep rule on sender and receiver to prevent this:
::
- type: regex
negate: true
regex: "^zrepl_.*" # <- the 'prefix' specified in snapshotting.prefix
Watch it Work
=============
+1 -71
View File
@@ -77,74 +77,4 @@ Systemd Unit File
A systemd service definition template is available in :repomasterlink:`dist/systemd`.
Note that some of the options only work on recent versions of systemd.
Any help & improvements are very welcome, see :issue:`145`.
============
Ops Runbooks
============
.. toctree::
usage/runbooks/migrating_sending_side_to_new_zpool.rst
.. _usage-platform-tests:
==============
Platform Tests
==============
Along with the main ``zrepl`` binary, we release the ``platformtest`` binaries.
The zrepl platform tests are an integration test suite that is complementary to the pure Go unit tests.
Any test that needs to interact with ZFS is a platform test.
The platform need to run as root.
For each test, we create a fresh dummy zpool backed by a file-based vdev.
The file path, and a root mountpoint for the dummy zpool, must be specified on the command line:
::
mkdir -p /tmp/zreplplatformtest
./platformtest \
-poolname 'zreplplatformtest' \ # <- name must contain zreplplatformtest
-imagepath /tmp/zreplplatformtest.img \ # <- zrepl will create the file
-mountpoint /tmp/zreplplatformtest # <- must exist
.. WARNING::
``platformtest`` will unconditionally overwrite the file at `imagepath`
and unconditionally ``zpool destroy $poolname``.
So, don't use a production poolname, and consider running the test in a VM.
It'll be a lot faster as well because the underlying operations, ``zfs list`` in particular, will be faster.
While the platformtests are running, there will be a log of log output.
After all tests have run, it prints a summary with a list of tests, grouped by result type (success, failure, skipped):
::
PASSING TESTS:
github.com/zrepl/zrepl/platformtest/tests.BatchDestroy
github.com/zrepl/zrepl/platformtest/tests.CreateReplicationCursor
github.com/zrepl/zrepl/platformtest/tests.GetNonexistent
github.com/zrepl/zrepl/platformtest/tests.HoldsWork
...
github.com/zrepl/zrepl/platformtest/tests.SendStreamNonEOFReadErrorHandling
github.com/zrepl/zrepl/platformtest/tests.UndestroyableSnapshotParsing
SKIPPED TESTS:
github.com/zrepl/zrepl/platformtest/tests.SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden__EncryptionSupported_false
FAILED TESTS: []
If there is a failure, or a skipped test that you believe should be passing, re-run the test suite, capture stderr & stdout to a text file, and create an issue on GitHub.
To run a specific test case, or a subset of tests matched by regex, use the ``-run REGEX`` command line flag.
To stop test execution at the first failing test, and prevent cleanup of the dummy zpool, use the ``-failure.stop-and-keep-pool`` flag.
To build the platformtests yourself, use ``make test-platform-bin``.
There's also the ``make test-platform`` target to run the platform tests with a default command line.
Any help & improvements are very welcome, see :issue:`145`.
@@ -1,36 +0,0 @@
Migrating Sending Side
~~~~~~~~~~~~~~~~~~~~~~
**Objective**:
Move sending-side zpool to new hardware.
Make the move fully transparent to the sending-side jobs.
After the move is done, all sending-side zrepl jobs should continue to work as if the move had not happened.
In particular, incremental replication should be able to pick up where it left before the move.
Suppose we want to migrate all data from one zpool ``oldpool`` to another zpool ``newpool``.
A possible reason might be that we want to change RAID levels, ``ashift``, or just migrate over to next-gen hardware.
If the pool names are different, zrepl's matching between sender and receiver dataset will break becase the receive-side dataset names contain ``oldpool``.
To avoid this, we will need the name of the new pool to match that of the old pool.
The following steps will accomplish this:
1. Stop zrepl.
2. Create the new pool: ``zpool create newpool ...``
3. Take a snapshot of the old pool so that you have something that you can ``zfs send``.
For example, run ``zfs snapshot -r oldpool@migration_oldpool_newpool``.
4. Send all of the oldpool's datasets to the new pool:
``zfs send -R oldpool@migration_oldpool_newpool | zfs recv -F newpool``
5. Export the old pool: ``zpool export oldpool``
6. Export the new pool: ``zpool export newpool``
7. (Optional) Change the name of the old pool to something that does not conflict with the new pool.
We are going to use the name ``oldoldpool`` in this example.
Use ``zpool import`` with no arguments to see the pool id.
Then ``zpool import <id> oldoldpool && zpool export oldoldpool``.
8. Import the new pool, while changing the name to match the old pool: ``zpool import newpool oldpool``
9. Start zrepl again and wake up the relevant jobs.
10. Use ``zrepl status`` or you monitoring to ensure that replication works.
The best test is an end-to-end test where you write some junk data on a sender dataset and wait until a snapshot with that data appears on the receiving side.
11. Once you are confident that replication is working, you may dispose of the old pool.
Note that, depending on pruning rules, it will not be possible to switch back to the old pool seamlessly, i.e., without a full re-replication.
+43 -132
View File
@@ -159,11 +159,12 @@ func sendArgsFromPDUAndValidateExistsAndGetVersion(ctx context.Context, fs strin
return version, nil
}
func (s *Sender) sendMakeArgs(ctx context.Context, r *pdu.SendReq) (sendArgs zfs.ZFSSendArgsValidated, _ error) {
func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
defer trace.WithSpanFromStackUpdateCtx(&ctx)()
_, err := s.filterCheckFS(r.Filesystem)
if err != nil {
return sendArgs, err
return nil, nil, err
}
switch r.Encrypted {
case pdu.Tri_DontCare:
@@ -171,16 +172,16 @@ func (s *Sender) sendMakeArgs(ctx context.Context, r *pdu.SendReq) (sendArgs zfs
// ok, fallthrough outer
case pdu.Tri_False:
if s.config.Encrypt.B {
return sendArgs, errors.New("only encrypted sends allowed (send -w + encryption!= off), but unencrypted send requested")
return nil, nil, errors.New("only encrypted sends allowed (send -w + encryption!= off), but unencrypted send requested")
}
// fallthrough outer
case pdu.Tri_True:
if !s.config.Encrypt.B {
return sendArgs, errors.New("only unencrypted sends allowed, but encrypted send requested")
return nil, nil, errors.New("only unencrypted sends allowed, but encrypted send requested")
}
// fallthrough outer
default:
return sendArgs, fmt.Errorf("unknown pdu.Tri variant %q", r.Encrypted)
return nil, nil, fmt.Errorf("unknown pdu.Tri variant %q", r.Encrypted)
}
sendArgsUnvalidated := zfs.ZFSSendArgsUnvalidated{
@@ -200,19 +201,30 @@ func (s *Sender) sendMakeArgs(ctx context.Context, r *pdu.SendReq) (sendArgs zfs
},
}
sendArgs, err = sendArgsUnvalidated.Validate(ctx)
sendArgs, err := sendArgsUnvalidated.Validate(ctx)
if err != nil {
return sendArgs, errors.Wrap(err, "validate send arguments")
return nil, nil, errors.Wrap(err, "validate send arguments")
}
return sendArgs, nil
}
func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
defer trace.WithSpanFromStackUpdateCtx(&ctx)()
sendArgs, err := s.sendMakeArgs(ctx, r)
si, err := zfs.ZFSSendDry(ctx, sendArgs)
if err != nil {
return nil, nil, err
return nil, nil, errors.Wrap(err, "zfs send dry failed")
}
// From now on, assume that sendArgs has been validated by ZFSSendDry
// (because validation involves shelling out, it's actually a little expensive)
var expSize int64 = 0 // protocol says 0 means no estimate
if si.SizeEstimate != -1 { // but si returns -1 for no size estimate
expSize = si.SizeEstimate
}
res := &pdu.SendRes{
ExpectedSize: expSize,
UsedResumeToken: r.ResumeToken != "",
}
if r.DryRun {
return res, nil, nil
}
// create holds or bookmarks of `From` and `To` to guarantee one of the following:
@@ -310,37 +322,9 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
// apply rate limit
sendStream = s.bwLimit.WrapReadCloser(sendStream)
res := &pdu.SendRes{
ExpectedSize: 0,
UsedResumeToken: r.ResumeToken != "",
}
return res, sendStream, nil
}
func (s *Sender) SendDry(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, error) {
defer trace.WithSpanFromStackUpdateCtx(&ctx)()
sendArgs, err := s.sendMakeArgs(ctx, r)
if err != nil {
return nil, err
}
si, err := zfs.ZFSSendDry(ctx, sendArgs)
if err != nil {
return nil, errors.Wrap(err, "zfs send dry failed")
}
// From now on, assume that sendArgs has been validated by ZFSSendDry
// (because validation involves shelling out, it's actually a little expensive)
res := &pdu.SendRes{
ExpectedSize: si.SizeEstimate,
UsedResumeToken: r.ResumeToken != "",
}
return res, nil
}
func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
defer trace.WithSpanFromStackUpdateCtx(&ctx)()
@@ -472,20 +456,8 @@ type ReceiverConfig struct {
OverrideProperties map[zfsprop.Property]string
BandwidthLimit bandwidthlimit.Config
PlaceholderEncryption PlaceholderCreationEncryptionProperty
}
//go:generate enumer -type=PlaceholderCreationEncryptionProperty -transform=kebab -trimprefix=PlaceholderCreationEncryptionProperty
type PlaceholderCreationEncryptionProperty int
// Note: the constant names, transformed through enumer, are part of the config format!
const (
PlaceholderCreationEncryptionPropertyUnspecified PlaceholderCreationEncryptionProperty = 1 << iota
PlaceholderCreationEncryptionPropertyInherit
PlaceholderCreationEncryptionPropertyOff
)
func (c *ReceiverConfig) copyIn() {
c.RootWithoutClientComponent = c.RootWithoutClientComponent.Copy()
@@ -525,10 +497,6 @@ func (c *ReceiverConfig) Validate() error {
return errors.Wrap(err, "`BandwidthLimit` field invalid")
}
if !c.PlaceholderEncryption.IsAPlaceholderCreationEncryptionProperty() {
return errors.Errorf("`PlaceholderEncryption` field is invalid")
}
return nil
}
@@ -541,8 +509,6 @@ type Receiver struct {
bwLimit bandwidthlimit.Wrapper
recvParentCreationMtx *chainlock.L
Test_OverrideClientIdentityFunc func() string // for use by platformtest
}
func NewReceiver(config ReceiverConfig) *Receiver {
@@ -580,15 +546,9 @@ func (s *Receiver) clientRootFromCtx(ctx context.Context) *zfs.DatasetPath {
return s.conf.RootWithoutClientComponent.Copy()
}
var clientIdentity string
if s.Test_OverrideClientIdentityFunc != nil {
clientIdentity = s.Test_OverrideClientIdentityFunc()
} else {
var ok bool
clientIdentity, ok = ctx.Value(ClientIdentityKey).(string) // no shadow
if !ok {
panic("ClientIdentityKey context value must be set")
}
clientIdentity, ok := ctx.Value(ClientIdentityKey).(string)
if !ok {
panic("ClientIdentityKey context value must be set")
}
clientRoot, err := clientRoot(s.conf.RootWithoutClientComponent, clientIdentity)
@@ -629,8 +589,7 @@ func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemR
if rphs, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, s.conf.RootWithoutClientComponent); err != nil {
return nil, errors.Wrap(err, "cannot determine whether root_fs exists")
} else if !rphs.FSExists {
getLogger(ctx).WithField("root_fs", s.conf.RootWithoutClientComponent).Error("root_fs does not exist")
return nil, errors.Errorf("root_fs does not exist")
return nil, errors.New("root_fs does not exist")
}
root := s.clientRootFromCtx(ctx)
@@ -734,38 +693,6 @@ func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, io
return nil, nil, fmt.Errorf("receiver does not implement Send()")
}
func (s *Receiver) SendDry(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, error) {
defer trace.WithSpanFromStackUpdateCtx(&ctx)()
return nil, fmt.Errorf("receiver does not implement SendDry()")
}
func (s *Receiver) receive_GetPlaceholderCreationEncryptionValue(client_root, path *zfs.DatasetPath) (zfs.FilesystemPlaceholderCreateEncryptionValue, error) {
if !s.conf.PlaceholderEncryption.IsAPlaceholderCreationEncryptionProperty() {
panic(s.conf.PlaceholderEncryption)
}
if client_root.Equal(path) && s.conf.PlaceholderEncryption == PlaceholderCreationEncryptionPropertyUnspecified {
// If our Receiver is configured to append a client component to s.conf.RootWithoutClientComponent
// then that dataset is always going to be a placeholder.
// We don't want to burden users with the concept of placeholders if their `filesystems` filter on the sender
// doesn't introduce any gaps.
// Since the dataset hierarchy up to and including that client component dataset is still fully controlled by us,
// using `inherit` is going to make it work in all expected use cases.
return zfs.FilesystemPlaceholderCreateEncryptionInherit, nil
}
switch s.conf.PlaceholderEncryption {
case PlaceholderCreationEncryptionPropertyUnspecified:
return 0, fmt.Errorf("placeholder filesystem encryption handling is unspecified in receiver config")
case PlaceholderCreationEncryptionPropertyInherit:
return zfs.FilesystemPlaceholderCreateEncryptionInherit, nil
case PlaceholderCreationEncryptionPropertyOff:
return zfs.FilesystemPlaceholderCreateEncryptionOff, nil
default:
panic(s.conf.PlaceholderEncryption)
}
}
func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error) {
defer trace.WithSpanFromStackUpdateCtx(&ctx)()
@@ -806,18 +733,15 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
if v.Path.Equal(lp) {
return false
}
l := getLogger(ctx).
WithField("placeholder_fs", v.Path.ToString()).
WithField("receive_fs", lp.ToString())
ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, v.Path)
l.WithField("placeholder_state", fmt.Sprintf("%#v", ph)).
getLogger(ctx).
WithField("fs", v.Path.ToString()).
WithField("placeholder_state", fmt.Sprintf("%#v", ph)).
WithField("err", fmt.Sprintf("%s", err)).
WithField("errType", fmt.Sprintf("%T", err)).
Debug("get placeholder state for filesystem")
Debug("placeholder state for filesystem")
if err != nil {
visitErr = errors.Wrapf(err, "cannot get placeholder state of %s", v.Path.ToString())
visitErr = err
return false
}
@@ -828,34 +752,21 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
} else {
visitErr = fmt.Errorf("root_fs %q does not exist", s.conf.RootWithoutClientComponent.ToString())
}
l.WithError(visitErr).Error("placeholders are only created automatically below root_fs")
getLogger(ctx).WithError(visitErr).Error("placeholders are only created automatically below root_fs")
return false
}
// compute the value lazily so that users who don't rely on
// placeholders can use the default value PlaceholderCreationEncryptionPropertyUnspecified
placeholderEncryption, err := s.receive_GetPlaceholderCreationEncryptionValue(root, v.Path)
l := getLogger(ctx).WithField("placeholder_fs", v.Path)
l.Debug("create placeholder filesystem")
err := zfs.ZFSCreatePlaceholderFilesystem(ctx, v.Path, v.Parent.Path)
if err != nil {
l.WithError(err).Error("cannot create placeholder filesystem") // logger already contains path
visitErr = errors.Wrapf(err, "cannot create placeholder filesystem %s", v.Path.ToString())
l.WithError(err).Error("cannot create placeholder filesystem")
visitErr = err
return false
}
l := l.WithField("encryption", placeholderEncryption)
l.Debug("creating placeholder filesystem")
err = zfs.ZFSCreatePlaceholderFilesystem(ctx, v.Path, v.Parent.Path, placeholderEncryption)
if err != nil {
l.WithError(err).Error("cannot create placeholder filesystem") // logger already contains path
visitErr = errors.Wrapf(err, "cannot create placeholder filesystem %s", v.Path.ToString())
return false
}
l.Info("created placeholder filesystem")
return true
} else {
l.Debug("filesystem exists")
return true // leave this fs as is
}
getLogger(ctx).WithField("filesystem", v.Path.ToString()).Debug("exists")
return true // leave this fs as is
})
}()
getLogger(ctx).WithField("visitErr", visitErr).Debug("complete tree-walk")
@@ -1,62 +0,0 @@
// Code generated by "enumer -type=PlaceholderCreationEncryptionProperty -transform=kebab -trimprefix=PlaceholderCreationEncryptionProperty"; DO NOT EDIT.
//
package endpoint
import (
"fmt"
)
const (
_PlaceholderCreationEncryptionPropertyName_0 = "unspecifiedinherit"
_PlaceholderCreationEncryptionPropertyName_1 = "off"
)
var (
_PlaceholderCreationEncryptionPropertyIndex_0 = [...]uint8{0, 11, 18}
_PlaceholderCreationEncryptionPropertyIndex_1 = [...]uint8{0, 3}
)
func (i PlaceholderCreationEncryptionProperty) String() string {
switch {
case 1 <= i && i <= 2:
i -= 1
return _PlaceholderCreationEncryptionPropertyName_0[_PlaceholderCreationEncryptionPropertyIndex_0[i]:_PlaceholderCreationEncryptionPropertyIndex_0[i+1]]
case i == 4:
return _PlaceholderCreationEncryptionPropertyName_1
default:
return fmt.Sprintf("PlaceholderCreationEncryptionProperty(%d)", i)
}
}
var _PlaceholderCreationEncryptionPropertyValues = []PlaceholderCreationEncryptionProperty{1, 2, 4}
var _PlaceholderCreationEncryptionPropertyNameToValueMap = map[string]PlaceholderCreationEncryptionProperty{
_PlaceholderCreationEncryptionPropertyName_0[0:11]: 1,
_PlaceholderCreationEncryptionPropertyName_0[11:18]: 2,
_PlaceholderCreationEncryptionPropertyName_1[0:3]: 4,
}
// PlaceholderCreationEncryptionPropertyString retrieves an enum value from the enum constants string name.
// Throws an error if the param is not part of the enum.
func PlaceholderCreationEncryptionPropertyString(s string) (PlaceholderCreationEncryptionProperty, error) {
if val, ok := _PlaceholderCreationEncryptionPropertyNameToValueMap[s]; ok {
return val, nil
}
return 0, fmt.Errorf("%s does not belong to PlaceholderCreationEncryptionProperty values", s)
}
// PlaceholderCreationEncryptionPropertyValues returns all values of the enum
func PlaceholderCreationEncryptionPropertyValues() []PlaceholderCreationEncryptionProperty {
return _PlaceholderCreationEncryptionPropertyValues
}
// IsAPlaceholderCreationEncryptionProperty returns "true" if the value is listed in the enum definition. "false" otherwise
func (i PlaceholderCreationEncryptionProperty) IsAPlaceholderCreationEncryptionProperty() bool {
for _, v := range _PlaceholderCreationEncryptionPropertyValues {
if i == v {
return true
}
}
return false
}
+2
View File
@@ -8,6 +8,7 @@ require (
github.com/gdamore/tcell/v2 v2.2.0
github.com/gitchander/permutation v0.0.0-20181107151852-9e56b92e9909
github.com/go-logfmt/logfmt v0.4.0
github.com/go-playground/universal-translator v0.17.0 // indirect
github.com/go-playground/validator v9.31.0+incompatible
github.com/go-playground/validator/v10 v10.4.1
github.com/go-sql-driver/mysql v1.4.1-0.20190907122137-b2c03bcae3d4
@@ -43,6 +44,7 @@ require (
golang.org/x/net v0.0.0-20210119194325-5f4716e94777
golang.org/x/sync v0.0.0-20190423024810-112230192c58
golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c
golang.org/x/text v0.3.5 // indirect
golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135
gonum.org/v1/gonum v0.7.0 // indirect
google.golang.org/genproto v0.0.0-20210122163508-8081c04a3579 // indirect
-1
View File
@@ -37,7 +37,6 @@ func main() {
flag.StringVar(&args.CreateArgs.Mountpoint, "mountpoint", "", "")
flag.BoolVar(&args.StopAndKeepPoolOnFail, "failure.stop-and-keep-pool", false, "if a test case fails, stop test execution and keep pool as it was when the test failed")
flag.StringVar(&args.Run, "run", "", "")
flag.DurationVar(&platformtest.ZpoolExportTimeout, "zfs.zpool-export-timeout", platformtest.ZpoolExportTimeout, "")
flag.Parse()
if err := HarnessRun(args); err != nil {
+2 -22
View File
@@ -5,17 +5,12 @@ import (
"fmt"
"os"
"path/filepath"
"runtime"
"strings"
"time"
"github.com/pkg/errors"
"github.com/zrepl/zrepl/zfs"
)
var ZpoolExportTimeout time.Duration = 500 * time.Millisecond
type Zpool struct {
args ZpoolCreateArgs
}
@@ -98,23 +93,8 @@ func (p *Zpool) Name() string { return p.args.PoolName }
func (p *Zpool) Destroy(ctx context.Context, e Execer) error {
exportDeadline := time.Now().Add(ZpoolExportTimeout)
for {
if time.Now().After(exportDeadline) {
return errors.Errorf("could not zpool export (got 'pool is busy'): %s", p.args.PoolName)
}
err := e.RunExpectSuccessNoOutput(ctx, "zpool", "export", p.args.PoolName)
if err == nil {
break
}
if strings.Contains(err.Error(), "pool is busy") {
runtime.Gosched()
continue
}
if err != nil {
return errors.Wrapf(err, "export pool %q", p.args.PoolName)
}
if err := e.RunExpectSuccessNoOutput(ctx, "zpool", "export", p.args.PoolName); err != nil {
return errors.Wrapf(err, "export pool %q", p.args.PoolName)
}
if err := os.Remove(p.args.ImagePath); err != nil {
-8
View File
@@ -24,9 +24,6 @@ var Cases = []Case{BatchDestroy,
ReplicationIsResumableFullSend__both_GuaranteeResumability,
ReplicationIsResumableFullSend__initial_GuaranteeIncrementalReplication_incremental_GuaranteeIncrementalReplication,
ReplicationIsResumableFullSend__initial_GuaranteeResumability_incremental_GuaranteeIncrementalReplication,
ReplicationPlaceholderEncryption__EncryptOnReceiverUseCase__WorksIfConfiguredWithInherit,
ReplicationPlaceholderEncryption__UnspecifiedIsOkForClientIdentityPlaceholder,
ReplicationPlaceholderEncryption__UnspecifiedLeadsToFailureAtRuntimeWhenCreatingPlaceholders,
ReplicationPropertyReplicationWorks,
ReplicationReceiverErrorWhileStillSending,
ReplicationStepCompletedLostBehavior__GuaranteeIncrementalReplication,
@@ -37,10 +34,5 @@ var Cases = []Case{BatchDestroy,
SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden__EncryptionSupported_true,
SendArgsValidationResumeTokenDifferentFilesystemForbidden,
SendArgsValidationResumeTokenEncryptionMismatchForbidden,
SendStreamCloseAfterBlockedOnPipeWrite,
SendStreamCloseAfterEOFRead,
SendStreamMultipleCloseAfterEOF,
SendStreamMultipleCloseBeforeEOF,
SendStreamNonEOFReadErrorHandling,
UndestroyableSnapshotParsing,
}
-6
View File
@@ -10,7 +10,6 @@ import (
"github.com/stretchr/testify/require"
"github.com/zrepl/zrepl/daemon/filters"
"github.com/zrepl/zrepl/platformtest"
"github.com/zrepl/zrepl/util/limitio"
"github.com/zrepl/zrepl/zfs"
@@ -175,8 +174,3 @@ func datasetToStringSortedTrimPrefix(prefix *zfs.DatasetPath, paths []*zfs.Datas
sort.Strings(pstrs)
return pstrs
}
func mustAddToSFilter(ctx *platformtest.Context, f *filters.DatasetMapFilter, fs string) {
err := f.Add(fs, "ok")
require.NoError(ctx, err)
}
+14 -222
View File
@@ -79,7 +79,6 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
AppendClientIdentity: false,
RootWithoutClientComponent: mustDatasetPath(i.rfsRoot),
BandwidthLimit: bandwidthlimit.NoLimitConfig(),
PlaceholderEncryption: endpoint.PlaceholderCreationEncryptionPropertyUnspecified,
}
if i.receiverConfigHook != nil {
i.receiverConfigHook(&receiverConfig)
@@ -810,22 +809,13 @@ type NeverEndingSender struct {
*endpoint.Sender
}
func (s *NeverEndingSender) SendDry(ctx context.Context, req *pdu.SendReq) (r *pdu.SendRes, err error) {
r, _, err = s.sendImpl(ctx, req, true)
return r, err
}
func (s *NeverEndingSender) Send(ctx context.Context, req *pdu.SendReq) (r *pdu.SendRes, stream io.ReadCloser, _ error) {
return s.sendImpl(ctx, req, false)
}
func (s *NeverEndingSender) sendImpl(ctx context.Context, req *pdu.SendReq, dry bool) (r *pdu.SendRes, stream io.ReadCloser, _ error) {
stream = nil
r = &pdu.SendRes{
UsedResumeToken: false,
ExpectedSize: 1 << 30,
}
if dry {
if req.DryRun {
return r, stream, nil
}
dz, err := os.Open("/dev/zero")
@@ -904,9 +894,13 @@ func ReplicationFailingInitialParentProhibitsChildReplication(ctx *platformtest.
fsAA := ctx.RootDataset + "/sender/aa"
sfilter := filters.NewDatasetMapFilter(3, true)
mustAddToSFilter(ctx, sfilter, fsA)
mustAddToSFilter(ctx, sfilter, fsAChild)
mustAddToSFilter(ctx, sfilter, fsAA)
mustAddToSFilter := func(fs string) {
err := sfilter.Add(fs, "ok")
require.NoError(ctx, err)
}
mustAddToSFilter(fsA)
mustAddToSFilter(fsAChild)
mustAddToSFilter(fsAA)
rfsRoot := ctx.RootDataset + "/receiver"
mockRecvErr := fmt.Errorf("yifae4ohPhaquaes0hohghiep9oufie4roo7quoWooluaj2ee8")
@@ -962,7 +956,6 @@ func ReplicationPropertyReplicationWorks(ctx *platformtest.Context) {
+ "sender/a/child"
+ "sender/a/child@1"
+ "receiver"
R zfs create -p "${ROOTDS}/receiver/${ROOTDS}/sender"
`)
sjid := endpoint.MustMakeJobID("sender-job")
@@ -972,8 +965,12 @@ func ReplicationPropertyReplicationWorks(ctx *platformtest.Context) {
fsAChild := ctx.RootDataset + "/sender/a/child"
sfilter := filters.NewDatasetMapFilter(2, true)
mustAddToSFilter(ctx, sfilter, fsA)
mustAddToSFilter(ctx, sfilter, fsAChild)
mustAddToSFilter := func(fs string) {
err := sfilter.Add(fs, "ok")
require.NoError(ctx, err)
}
mustAddToSFilter(fsA)
mustAddToSFilter(fsAChild)
rfsRoot := ctx.RootDataset + "/receiver"
type testPropExpectation struct {
@@ -1101,208 +1098,3 @@ func ReplicationPropertyReplicationWorks(ctx *platformtest.Context) {
}
}
}
func ReplicationPlaceholderEncryption__UnspecifiedLeadsToFailureAtRuntimeWhenCreatingPlaceholders(ctx *platformtest.Context) {
platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
CREATEROOT
+ "sender"
+ "sender/a"
+ "sender/a/child"
+ "receiver"
R zfs snapshot -r ${ROOTDS}/sender@initial
`)
sjid := endpoint.MustMakeJobID("sender-job")
rjid := endpoint.MustMakeJobID("receiver-job")
childfs := ctx.RootDataset + "/sender/a/child"
sfilter := filters.NewDatasetMapFilter(3, true)
mustAddToSFilter(ctx, sfilter, childfs)
rfsRoot := ctx.RootDataset + "/receiver"
rep := replicationInvocation{
sjid: sjid,
rjid: rjid,
sfilter: sfilter,
rfsRoot: rfsRoot,
guarantee: pdu.ReplicationConfigProtectionWithKind(pdu.ReplicationGuaranteeKind_GuaranteeResumability),
receiverConfigHook: func(rc *endpoint.ReceiverConfig) {
rc.PlaceholderEncryption = endpoint.PlaceholderCreationEncryptionPropertyUnspecified
},
}
r := rep.Do(ctx)
ctx.Logf("\n%s", pretty.Sprint(r))
require.Len(ctx, r.Attempts, 1)
attempt := r.Attempts[0]
require.Nil(ctx, attempt.PlanError)
require.Len(ctx, attempt.Filesystems, 1)
afs := attempt.Filesystems[0]
require.Equal(ctx, childfs, afs.Info.Name)
require.Equal(ctx, 1, len(afs.Steps))
require.Equal(ctx, 0, afs.CurrentStep)
require.Equal(ctx, report.FilesystemSteppingErrored, afs.State)
childfsFirstComponent := strings.Split(childfs, "/")[0]
require.Contains(ctx, afs.StepError.Err, "cannot create placeholder filesystem "+rfsRoot+"/"+childfsFirstComponent+": placeholder filesystem encryption handling is unspecified in receiver config")
}
type ClientIdentityReceiver struct {
clientIdentity string
*endpoint.Receiver
}
func (r *ClientIdentityReceiver) Receive(ctx context.Context, req *pdu.ReceiveReq, stream io.ReadCloser) (*pdu.ReceiveRes, error) {
ctx = context.WithValue(ctx, endpoint.ClientIdentityKey, r.clientIdentity)
return r.Receiver.Receive(ctx, req, stream)
}
func ReplicationPlaceholderEncryption__UnspecifiedIsOkForClientIdentityPlaceholder(ctx *platformtest.Context) {
platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
CREATEROOT
+ "receiver"
`)
sjid := endpoint.MustMakeJobID("sender-job")
rjid := endpoint.MustMakeJobID("receiver-job")
sfilter := filters.NewDatasetMapFilter(1, true)
// hacky...
comps := strings.Split(ctx.RootDataset, "/")
require.GreaterOrEqual(ctx, len(comps), 2)
pool := comps[0]
require.Contains(ctx, pool, "zreplplatformtest", "don't want to cause accidents")
poolchild := pool + "/" + comps[1]
err := zfs.ZFSSnapshot(ctx, mustDatasetPath(pool), "testsnap", false)
require.NoError(ctx, err)
err = zfs.ZFSSnapshot(ctx, mustDatasetPath(poolchild), "testsnap", false)
require.NoError(ctx, err)
mustAddToSFilter(ctx, sfilter, pool)
mustAddToSFilter(ctx, sfilter, poolchild)
clientIdentity := "testclientid"
rfsRoot := ctx.RootDataset + "/receiver"
rep := replicationInvocation{
sjid: sjid,
rjid: rjid,
sfilter: sfilter,
rfsRoot: rfsRoot,
guarantee: pdu.ReplicationConfigProtectionWithKind(pdu.ReplicationGuaranteeKind_GuaranteeResumability),
receiverConfigHook: func(rc *endpoint.ReceiverConfig) {
rc.PlaceholderEncryption = endpoint.PlaceholderCreationEncryptionPropertyUnspecified
rc.AppendClientIdentity = true
},
interceptReceiver: func(r *endpoint.Receiver) logic.Receiver {
r.Test_OverrideClientIdentityFunc = func() string { return clientIdentity }
return r
},
}
r := rep.Do(ctx)
ctx.Logf("\n%s", pretty.Sprint(r))
require.Len(ctx, r.Attempts, 1)
attempt := r.Attempts[0]
require.Nil(ctx, attempt.PlanError)
require.Len(ctx, attempt.Filesystems, 2)
filesystemsByName := make(map[string]*report.FilesystemReport)
for _, fs := range attempt.Filesystems {
filesystemsByName[fs.Info.Name] = fs
}
require.Len(ctx, filesystemsByName, len(attempt.Filesystems))
afs, ok := filesystemsByName[pool]
require.True(ctx, ok)
require.Nil(ctx, afs.PlanError)
require.Nil(ctx, afs.StepError)
require.Equal(ctx, report.FilesystemDone, afs.State)
afs, ok = filesystemsByName[poolchild]
require.True(ctx, ok)
require.Nil(ctx, afs.PlanError)
require.Nil(ctx, afs.StepError)
require.Equal(ctx, report.FilesystemDone, afs.State)
mustGetFilesystemVersion(ctx, rfsRoot+"/"+clientIdentity+"/"+pool+"@testsnap")
mustGetFilesystemVersion(ctx, rfsRoot+"/"+clientIdentity+"/"+poolchild+"@testsnap")
}
func replicationPlaceholderEncryption__EncryptOnReceiverUseCase__impl(ctx *platformtest.Context, placeholderEncryption endpoint.PlaceholderCreationEncryptionProperty) {
platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
CREATEROOT
+ "sender"
+ "sender/a"
+ "sender/a/child"
+ "receiver" encrypted
R zfs snapshot -r ${ROOTDS}/sender@initial
`)
sjid := endpoint.MustMakeJobID("sender-job")
rjid := endpoint.MustMakeJobID("receiver-job")
childfs := ctx.RootDataset + "/sender/a/child"
sfilter := filters.NewDatasetMapFilter(3, true)
mustAddToSFilter(ctx, sfilter, childfs)
rfsRoot := ctx.RootDataset + "/receiver"
rep := replicationInvocation{
sjid: sjid,
rjid: rjid,
sfilter: sfilter,
rfsRoot: rfsRoot,
guarantee: pdu.ReplicationConfigProtectionWithKind(pdu.ReplicationGuaranteeKind_GuaranteeResumability),
receiverConfigHook: func(rc *endpoint.ReceiverConfig) {
rc.PlaceholderEncryption = placeholderEncryption
rc.AppendClientIdentity = false
},
}
r := rep.Do(ctx)
ctx.Logf("\n%s", pretty.Sprint(r))
require.Len(ctx, r.Attempts, 1)
attempt := r.Attempts[0]
require.Equal(ctx, report.AttemptDone, attempt.State)
require.Len(ctx, attempt.Filesystems, 1)
afs := attempt.Filesystems[0]
require.Equal(ctx, childfs, afs.Info.Name)
require.Equal(ctx, 1, len(afs.Steps))
rfs := mustDatasetPath(rfsRoot + "/" + childfs)
mustGetFilesystemVersion(ctx, rfs.ToString()+"@initial")
}
func ReplicationPlaceholderEncryption__EncryptOnReceiverUseCase__WorksIfConfiguredWithInherit(ctx *platformtest.Context) {
placeholderEncryption := endpoint.PlaceholderCreationEncryptionPropertyInherit
replicationPlaceholderEncryption__EncryptOnReceiverUseCase__impl(ctx, placeholderEncryption)
childfs := ctx.RootDataset + "/sender/a/child"
rfsRoot := ctx.RootDataset + "/receiver"
rfs := mustDatasetPath(rfsRoot + "/" + childfs)
// The leaf child dataset should be inhering from rfsRoot.
// If we had replicated with PlaceholderCreationEncryptionPropertyOff
// then it would be unencrypted and inherit from the placeholder.
props, err := zfs.ZFSGet(ctx, rfs, []string{"encryptionroot"})
require.NoError(ctx, err)
require.Equal(ctx, rfsRoot, props.Get("encryptionroot"))
}
-186
View File
@@ -1,186 +0,0 @@
package tests
import (
"fmt"
"io"
"io/ioutil"
"os"
"os/exec"
"path"
"syscall"
"time"
"github.com/stretchr/testify/require"
"golang.org/x/sys/unix"
"github.com/zrepl/zrepl/platformtest"
"github.com/zrepl/zrepl/util/nodefault"
"github.com/zrepl/zrepl/zfs"
)
func sendStreamTest(ctx *platformtest.Context) *zfs.SendStream {
platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
DESTROYROOT
CREATEROOT
+ "sender"
`)
fs := fmt.Sprintf("%s/sender", ctx.RootDataset)
fsmpo, err := zfs.ZFSGetMountpoint(ctx, fs)
require.NoError(ctx, err)
writeDummyData(path.Join(fsmpo.Mountpoint, "dummy.data"), 1<<26)
mustSnapshot(ctx, fs+"@1")
snap := fsversion(ctx, fs, "@1")
snapSendArg := snap.ToSendArgVersion()
sendArgs, err := zfs.ZFSSendArgsUnvalidated{
FS: fs,
From: nil,
To: &snapSendArg,
ZFSSendFlags: zfs.ZFSSendFlags{
Encrypted: &nodefault.Bool{B: false},
},
}.Validate(ctx)
require.NoError(ctx, err)
sendStream, err := zfs.ZFSSend(ctx, sendArgs)
require.NoError(ctx, err)
return sendStream
}
func SendStreamCloseAfterBlockedOnPipeWrite(ctx *platformtest.Context) {
sendStream := sendStreamTest(ctx)
// let the pipe buffer fill and the zfs process block uninterruptibly
ctx.Logf("waiting for pipe write to block")
time.Sleep(5 * time.Second) // XXX need a platform-neutral way to detect that the pipe is full and the writer is blocked
ctx.Logf("closing send stream")
err := sendStream.Close() // this is what this test case is about
ctx.Logf("close error: %T %s", err, err)
require.NoError(ctx, err)
exitErrZfsError := sendStream.TestOnly_ExitErr()
require.Contains(ctx, exitErrZfsError.Error(), "signal")
require.Error(ctx, exitErrZfsError)
exitErr, ok := exitErrZfsError.WaitErr.(*exec.ExitError)
require.True(ctx, ok)
if exitErr.Exited() {
// some ZFS impls (FreeBSD 12) behaves that way
return
}
// ProcessState is only available after exit
// => use as proxy that the process was wait()ed upon and is gone
ctx.Logf("%#v", exitErr.ProcessState)
require.NotNil(ctx, exitErr.ProcessState)
// and let's verify that the process got killed, so that we know it was the call to .Close() above
waitStatus := exitErr.ProcessState.Sys().(syscall.WaitStatus)
ctx.Logf("wait status: %#v", waitStatus)
ctx.Logf("exit status: %v", waitStatus.ExitStatus())
require.True(ctx, waitStatus.Signaled())
switch waitStatus.Signal() {
case unix.SIGKILL:
fallthrough
case unix.SIGPIPE:
// ok
default:
ctx.Errorf("%T %s\n%v", waitStatus.Signal(), waitStatus.Signal(), waitStatus.Signal())
ctx.FailNow()
}
}
func SendStreamCloseAfterEOFRead(ctx *platformtest.Context) {
sendStream := sendStreamTest(ctx)
_, err := io.Copy(ioutil.Discard, sendStream)
require.NoError(ctx, err)
var buf [128]byte
n, err := sendStream.Read(buf[:])
require.Zero(ctx, n)
require.Equal(ctx, io.EOF, err)
err = sendStream.Close()
require.NoError(ctx, err)
n, err = sendStream.Read(buf[:])
require.Zero(ctx, n)
require.Equal(ctx, os.ErrClosed, err, "same read error should be returned")
}
func SendStreamMultipleCloseAfterEOF(ctx *platformtest.Context) {
sendStream := sendStreamTest(ctx)
_, err := io.Copy(ioutil.Discard, sendStream)
require.NoError(ctx, err)
var buf [128]byte
n, err := sendStream.Read(buf[:])
require.Zero(ctx, n)
require.Equal(ctx, io.EOF, err)
err = sendStream.Close()
require.NoError(ctx, err)
err = sendStream.Close()
require.Equal(ctx, os.ErrClosed, err)
}
func SendStreamMultipleCloseBeforeEOF(ctx *platformtest.Context) {
sendStream := sendStreamTest(ctx)
err := sendStream.Close()
require.NoError(ctx, err)
err = sendStream.Close()
require.Equal(ctx, os.ErrClosed, err)
}
type failingReadCloser struct {
err error
}
var _ io.ReadCloser = &failingReadCloser{}
func (c *failingReadCloser) Read(p []byte) (int, error) { return 0, c.err }
func (c *failingReadCloser) Close() error { return c.err }
func SendStreamNonEOFReadErrorHandling(ctx *platformtest.Context) {
sendStream := sendStreamTest(ctx)
var buf [128]byte
n, err := sendStream.Read(buf[:])
require.Equal(ctx, len(buf), n)
require.NoError(ctx, err)
var mockError = fmt.Errorf("taeghaefow4piesahwahjocu7ul5tiachaiLipheijae8ooZ8Pies8shohGee9feeTeirai5aiFeiyaecai4kiaLoh4azeih0tea")
mock := &failingReadCloser{err: mockError}
orig := sendStream.TestOnly_ReplaceStdoutReader(mock)
n, err = sendStream.Read(buf[:])
require.Equal(ctx, 0, n)
require.Equal(ctx, mockError, err)
if sendStream.TestOnly_ReplaceStdoutReader(orig) != mock {
panic("incorrect test impl")
}
err = sendStream.Close()
require.NoError(ctx, err) // if we can't kill the child then this will be a flaky test, but let's assume we can kill the child
err = sendStream.Close()
require.Equal(ctx, os.ErrClosed, err)
}
+5 -21
View File
@@ -151,8 +151,6 @@ type fs struct {
l *chainlock.L
blockedOn report.FsBlockedOn
// ordering relationship that must be maintained for initial replication
initialRepOrd struct {
parents, children []*fs
@@ -160,9 +158,8 @@ type fs struct {
}
planning struct {
waitingForStepQueue bool
done bool
err *timedError
done bool
err *timedError
}
// valid iff planning.done && planning.err == nil
@@ -340,9 +337,8 @@ func (a *attempt) doGlobalPlanning(ctx context.Context, prev *attempt) map[*fs]*
for _, pfs := range pfss {
fs := &fs{
fs: pfs,
l: a.l,
blockedOn: report.FsBlockedOnNothing,
fs: pfs,
l: a.l,
}
fs.initialRepOrd.parentDidUpdate = make(chan struct{}, 1)
a.fss = append(a.fss, fs)
@@ -452,10 +448,6 @@ func (a *attempt) doFilesystems(ctx context.Context, prevs map[*fs]*fs) {
ctx, endTask := trace.WithTaskAndSpan(ctx, "repl-fs", f.report().Info.Name)
defer endTask()
f.do(ctx, stepQueue, prevs[f])
f.l.HoldWhile(func() {
// every return from f means it's unblocked...
f.blockedOn = report.FsBlockedOnNothing
})
}(f)
}
a.l.DropWhile(func() {
@@ -494,16 +486,11 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
var psteps []Step
var errTime time.Time
var err error
f.blockedOn = report.FsBlockedOnPlanningStepQueue
f.l.DropWhile(func() {
// TODO hacky
// choose target time that is earlier than any snapshot, so fs planning is always prioritized
targetDate := time.Unix(0, 0)
defer pq.WaitReady(ctx, f, targetDate)()
f.l.HoldWhile(func() {
// transition before we call PlanFS
f.blockedOn = report.FsBlockedOnNothing
})
psteps, err = f.fs.PlanFS(ctx) // no shadow
errTime = time.Now() // no shadow
})
@@ -558,7 +545,6 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
f.planning.done = true
// wait for parents' initial replication
f.blockedOn = report.FsBlockedOnParentInitialRepl
var parents []string
for _, p := range f.initialRepOrd.parents {
parents = append(parents, p.fs.ReportInfo().Name)
@@ -627,6 +613,7 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
select {
case <-ctx.Done():
f.planned.stepErr = newTimedError(ctx.Err(), time.Now())
return
case <-f.initialRepOrd.parentDidUpdate:
// loop
}
@@ -644,9 +631,7 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
f.l.DropWhile(func() {
// wait for parallel replication
targetDate := s.step.TargetDate()
f.l.HoldWhile(func() { f.blockedOn = report.FsBlockedOnReplStepQueue })
defer pq.WaitReady(ctx, f, targetDate)()
f.l.HoldWhile(func() { f.blockedOn = report.FsBlockedOnNothing })
// do the step
ctx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("%#v", s.step.ReportInfo()))
defer endSpan()
@@ -740,7 +725,6 @@ func (f *fs) report() *report.FilesystemReport {
r := &report.FilesystemReport{
Info: f.fs.ReportInfo(),
State: state,
BlockedOn: f.blockedOn,
PlanError: f.planning.err.IntoReportError(),
StepError: f.planned.stepErr.IntoReportError(),
Steps: make([]*report.StepReport, len(f.planned.steps)),
@@ -14,12 +14,11 @@ import (
"github.com/stretchr/testify/assert"
"github.com/zrepl/zrepl/daemon/logging/trace"
"github.com/zrepl/zrepl/util/zreplcircleci"
)
// FIXME: this test relies on timing and is thus rather flaky
// (relies on scheduler responsiveness of < 500ms)
func TestPqNotconcurrent(t *testing.T) {
zreplcircleci.SkipOnCircleCI(t, "because it relies on scheduler responsiveness < 500ms")
ctx, end := trace.WithTaskFromStack(context.Background())
defer end()
var ctr uint32
@@ -91,8 +90,6 @@ func (r record) String() string {
// Hence, perform some statistics on the wakeup times and assert that the mean wakeup
// times for each step are close together.
func TestPqConcurrent(t *testing.T) {
zreplcircleci.SkipOnCircleCI(t, "because it relies on scheduler responsiveness < 500ms")
ctx, end := trace.WithTaskFromStack(context.Background())
defer end()
+152 -134
View File
@@ -518,7 +518,8 @@ type SendReq struct {
// encoded in the ResumeToken. Otherwise, the Sender MUST return an error.
ResumeToken string `protobuf:"bytes,4,opt,name=ResumeToken,proto3" json:"ResumeToken,omitempty"`
Encrypted Tri `protobuf:"varint,5,opt,name=Encrypted,proto3,enum=Tri" json:"Encrypted,omitempty"`
ReplicationConfig *ReplicationConfig `protobuf:"bytes,6,opt,name=ReplicationConfig,proto3" json:"ReplicationConfig,omitempty"`
DryRun bool `protobuf:"varint,6,opt,name=DryRun,proto3" json:"DryRun,omitempty"`
ReplicationConfig *ReplicationConfig `protobuf:"bytes,7,opt,name=ReplicationConfig,proto3" json:"ReplicationConfig,omitempty"`
}
func (x *SendReq) Reset() {
@@ -588,6 +589,13 @@ func (x *SendReq) GetEncrypted() Tri {
return Tri_DontCare
}
func (x *SendReq) GetDryRun() bool {
if x != nil {
return x.DryRun
}
return false
}
func (x *SendReq) GetReplicationConfig() *ReplicationConfig {
if x != nil {
return x.ReplicationConfig
@@ -758,11 +766,12 @@ type SendRes struct {
unknownFields protoimpl.UnknownFields
// Whether the resume token provided in the request has been used or not.
// If the SendReq.ResumeToken == "", this field MUST be false.
UsedResumeToken bool `protobuf:"varint,1,opt,name=UsedResumeToken,proto3" json:"UsedResumeToken,omitempty"`
// If the SendReq.ResumeToken == "", this field has no meaning.
UsedResumeToken bool `protobuf:"varint,2,opt,name=UsedResumeToken,proto3" json:"UsedResumeToken,omitempty"`
// Expected stream size determined by dry run, not exact.
// 0 indicates that for the given SendReq, no size estimate could be made.
ExpectedSize uint64 `protobuf:"varint,2,opt,name=ExpectedSize,proto3" json:"ExpectedSize,omitempty"`
ExpectedSize int64 `protobuf:"varint,3,opt,name=ExpectedSize,proto3" json:"ExpectedSize,omitempty"`
Properties []*Property `protobuf:"bytes,4,rep,name=Properties,proto3" json:"Properties,omitempty"`
}
func (x *SendRes) Reset() {
@@ -804,13 +813,20 @@ func (x *SendRes) GetUsedResumeToken() bool {
return false
}
func (x *SendRes) GetExpectedSize() uint64 {
func (x *SendRes) GetExpectedSize() int64 {
if x != nil {
return x.ExpectedSize
}
return 0
}
func (x *SendRes) GetProperties() []*Property {
if x != nil {
return x.Properties
}
return nil
}
type SendCompletedReq struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
@@ -1427,7 +1443,7 @@ var file_pdu_proto_rawDesc = []byte{
0x72, 0x65, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x29, 0x0a, 0x0b, 0x56, 0x65, 0x72, 0x73, 0x69,
0x6f, 0x6e, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0c, 0x0a, 0x08, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68,
0x6f, 0x74, 0x10, 0x00, 0x12, 0x0c, 0x0a, 0x08, 0x42, 0x6f, 0x6f, 0x6b, 0x6d, 0x61, 0x72, 0x6b,
0x10, 0x01, 0x22, 0xfd, 0x01, 0x0a, 0x07, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x12, 0x1e,
0x10, 0x01, 0x22, 0x95, 0x02, 0x0a, 0x07, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x12, 0x1e,
0x0a, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01,
0x28, 0x09, 0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x12, 0x26,
0x0a, 0x04, 0x46, 0x72, 0x6f, 0x6d, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46,
@@ -1439,119 +1455,122 @@ var file_pdu_proto_rawDesc = []byte{
0x0b, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x22, 0x0a, 0x09,
0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0e, 0x32,
0x04, 0x2e, 0x54, 0x72, 0x69, 0x52, 0x09, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64,
0x12, 0x40, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43,
0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x52, 0x65,
0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52,
0x12, 0x16, 0x0a, 0x06, 0x44, 0x72, 0x79, 0x52, 0x75, 0x6e, 0x18, 0x06, 0x20, 0x01, 0x28, 0x08,
0x52, 0x06, 0x44, 0x72, 0x79, 0x52, 0x75, 0x6e, 0x12, 0x40, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c,
0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x07, 0x20,
0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61,
0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x22, 0x51, 0x0a, 0x11, 0x52, 0x65,
0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12,
0x3c, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20,
0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x50, 0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f,
0x6e, 0x52, 0x0a, 0x70, 0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x8f, 0x01,
0x0a, 0x1b, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e,
0x66, 0x69, 0x67, 0x50, 0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x33, 0x0a,
0x07, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x19,
0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61, 0x72,
0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e, 0x64, 0x52, 0x07, 0x49, 0x6e, 0x69, 0x74, 0x69,
0x61, 0x6c, 0x12, 0x3b, 0x0a, 0x0b, 0x49, 0x6e, 0x63, 0x72, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61,
0x6c, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x19, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
0x61, 0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69,
0x6e, 0x64, 0x52, 0x0b, 0x49, 0x6e, 0x63, 0x72, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x22,
0x34, 0x0a, 0x08, 0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x79, 0x12, 0x12, 0x0a, 0x04, 0x4e,
0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x12,
0x14, 0x0a, 0x05, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05,
0x56, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x82, 0x01, 0x0a, 0x07, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65,
0x73, 0x12, 0x28, 0x0a, 0x0f, 0x55, 0x73, 0x65, 0x64, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54,
0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0f, 0x55, 0x73, 0x65, 0x64,
0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x22, 0x0a, 0x0c, 0x45,
0x78, 0x70, 0x65, 0x63, 0x74, 0x65, 0x64, 0x53, 0x69, 0x7a, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28,
0x03, 0x52, 0x0c, 0x45, 0x78, 0x70, 0x65, 0x63, 0x74, 0x65, 0x64, 0x53, 0x69, 0x7a, 0x65, 0x12,
0x29, 0x0a, 0x0a, 0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x18, 0x04, 0x20,
0x03, 0x28, 0x0b, 0x32, 0x09, 0x2e, 0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x79, 0x52, 0x0a,
0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x22, 0x3e, 0x0a, 0x10, 0x53, 0x65,
0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x12, 0x2a,
0x0a, 0x0b, 0x4f, 0x72, 0x69, 0x67, 0x69, 0x6e, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x18, 0x02, 0x20,
0x01, 0x28, 0x0b, 0x32, 0x08, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x52, 0x0b, 0x4f,
0x72, 0x69, 0x67, 0x69, 0x6e, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x22, 0x12, 0x0a, 0x10, 0x53, 0x65,
0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73, 0x22, 0xbe,
0x01, 0x0a, 0x0a, 0x52, 0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x52, 0x65, 0x71, 0x12, 0x1e, 0x0a,
0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28,
0x09, 0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x12, 0x22, 0x0a,
0x02, 0x54, 0x6f, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c, 0x65,
0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x02, 0x54,
0x6f, 0x12, 0x2a, 0x0a, 0x10, 0x43, 0x6c, 0x65, 0x61, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65,
0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x10, 0x43, 0x6c, 0x65,
0x61, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x40, 0x0a,
0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66,
0x69, 0x67, 0x22, 0x51, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x3c, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x74, 0x65,
0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x52, 0x65,
0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x50,
0x72, 0x6f, 0x74, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x0a, 0x70, 0x72, 0x6f, 0x74, 0x65,
0x63, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x8f, 0x01, 0x0a, 0x1b, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x50, 0x72, 0x6f, 0x74, 0x65,
0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x33, 0x0a, 0x07, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x6c,
0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x19, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61,
0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e,
0x64, 0x52, 0x07, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x6c, 0x12, 0x3b, 0x0a, 0x0b, 0x49, 0x6e,
0x63, 0x72, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32,
0x19, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61,
0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e, 0x64, 0x52, 0x0b, 0x49, 0x6e, 0x63, 0x72,
0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x22, 0x34, 0x0a, 0x08, 0x50, 0x72, 0x6f, 0x70, 0x65,
0x72, 0x74, 0x79, 0x12, 0x12, 0x0a, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28,
0x09, 0x52, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x56, 0x61, 0x6c, 0x75, 0x65,
0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x57, 0x0a,
0x07, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x73, 0x12, 0x28, 0x0a, 0x0f, 0x55, 0x73, 0x65, 0x64,
0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28,
0x08, 0x52, 0x0f, 0x55, 0x73, 0x65, 0x64, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b,
0x65, 0x6e, 0x12, 0x22, 0x0a, 0x0c, 0x45, 0x78, 0x70, 0x65, 0x63, 0x74, 0x65, 0x64, 0x53, 0x69,
0x7a, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0c, 0x45, 0x78, 0x70, 0x65, 0x63, 0x74,
0x65, 0x64, 0x53, 0x69, 0x7a, 0x65, 0x22, 0x3e, 0x0a, 0x10, 0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f,
0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x12, 0x2a, 0x0a, 0x0b, 0x4f, 0x72,
0x69, 0x67, 0x69, 0x6e, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32,
0x08, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x52, 0x0b, 0x4f, 0x72, 0x69, 0x67, 0x69,
0x6e, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x22, 0x12, 0x0a, 0x10, 0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f,
0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73, 0x22, 0xbe, 0x01, 0x0a, 0x0a, 0x52,
0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x52, 0x65, 0x71, 0x12, 0x1e, 0x0a, 0x0a, 0x46, 0x69, 0x6c,
0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x46,
0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x12, 0x22, 0x0a, 0x02, 0x54, 0x6f, 0x18,
0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74,
0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x02, 0x54, 0x6f, 0x12, 0x2a, 0x0a,
0x10, 0x43, 0x6c, 0x65, 0x61, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65,
0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x10, 0x43, 0x6c, 0x65, 0x61, 0x72, 0x52, 0x65,
0x73, 0x75, 0x6d, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x40, 0x0a, 0x11, 0x52, 0x65, 0x70,
0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x04,
0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69,
0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x22, 0x0c, 0x0a, 0x0a, 0x52,
0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x52, 0x65, 0x73, 0x22, 0x67, 0x0a, 0x13, 0x44, 0x65, 0x73,
0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x71,
0x69, 0x67, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69,
0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x11, 0x52, 0x65,
0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x22,
0x0c, 0x0a, 0x0a, 0x52, 0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x52, 0x65, 0x73, 0x22, 0x67, 0x0a,
0x13, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74,
0x73, 0x52, 0x65, 0x71, 0x12, 0x1e, 0x0a, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74,
0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79,
0x73, 0x74, 0x65, 0x6d, 0x12, 0x30, 0x0a, 0x09, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74,
0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79,
0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x53, 0x6e, 0x61,
0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x22, 0x5a, 0x0a, 0x12, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f,
0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x12, 0x2e, 0x0a, 0x08,
0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12,
0x2e, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69,
0x6f, 0x6e, 0x52, 0x08, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x12, 0x14, 0x0a, 0x05,
0x45, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x45, 0x72, 0x72,
0x6f, 0x72, 0x22, 0x44, 0x0a, 0x13, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61,
0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x73, 0x12, 0x2d, 0x0a, 0x07, 0x52, 0x65, 0x73,
0x75, 0x6c, 0x74, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x44, 0x65, 0x73,
0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x52,
0x07, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73, 0x22, 0x36, 0x0a, 0x14, 0x52, 0x65, 0x70, 0x6c,
0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x71,
0x12, 0x1e, 0x0a, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01,
0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d,
0x12, 0x30, 0x0a, 0x09, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x18, 0x02, 0x20,
0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d,
0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f,
0x74, 0x73, 0x22, 0x5a, 0x0a, 0x12, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61,
0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x12, 0x2e, 0x0a, 0x08, 0x53, 0x6e, 0x61, 0x70,
0x73, 0x68, 0x6f, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x46, 0x69, 0x6c,
0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x08,
0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x12, 0x14, 0x0a, 0x05, 0x45, 0x72, 0x72, 0x6f,
0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x22, 0x44,
0x0a, 0x13, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f,
0x74, 0x73, 0x52, 0x65, 0x73, 0x12, 0x2d, 0x0a, 0x07, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73,
0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79,
0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x52, 0x07, 0x52, 0x65, 0x73,
0x75, 0x6c, 0x74, 0x73, 0x22, 0x36, 0x0a, 0x14, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74,
0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x12, 0x1e, 0x0a, 0x0a,
0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
0x52, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x22, 0x54, 0x0a, 0x14,
0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f,
0x72, 0x52, 0x65, 0x73, 0x12, 0x14, 0x0a, 0x04, 0x47, 0x75, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01,
0x28, 0x04, 0x48, 0x00, 0x52, 0x04, 0x47, 0x75, 0x69, 0x64, 0x12, 0x1c, 0x0a, 0x08, 0x4e, 0x6f,
0x74, 0x65, 0x78, 0x69, 0x73, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x48, 0x00, 0x52, 0x08,
0x4e, 0x6f, 0x74, 0x65, 0x78, 0x69, 0x73, 0x74, 0x42, 0x08, 0x0a, 0x06, 0x52, 0x65, 0x73, 0x75,
0x6c, 0x74, 0x22, 0x23, 0x0a, 0x07, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71, 0x12, 0x18, 0x0a,
0x07, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07,
0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x1d, 0x0a, 0x07, 0x50, 0x69, 0x6e, 0x67, 0x52,
0x65, 0x73, 0x12, 0x12, 0x0a, 0x04, 0x45, 0x63, 0x68, 0x6f, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
0x52, 0x04, 0x45, 0x63, 0x68, 0x6f, 0x2a, 0x28, 0x0a, 0x03, 0x54, 0x72, 0x69, 0x12, 0x0c, 0x0a,
0x08, 0x44, 0x6f, 0x6e, 0x74, 0x43, 0x61, 0x72, 0x65, 0x10, 0x00, 0x12, 0x09, 0x0a, 0x05, 0x46,
0x61, 0x6c, 0x73, 0x65, 0x10, 0x01, 0x12, 0x08, 0x0a, 0x04, 0x54, 0x72, 0x75, 0x65, 0x10, 0x02,
0x2a, 0x86, 0x01, 0x0a, 0x18, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e,
0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e, 0x64, 0x12, 0x14, 0x0a,
0x10, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x49, 0x6e, 0x76, 0x61, 0x6c, 0x69,
0x64, 0x10, 0x00, 0x12, 0x19, 0x0a, 0x15, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65,
0x52, 0x65, 0x73, 0x75, 0x6d, 0x61, 0x62, 0x69, 0x6c, 0x69, 0x74, 0x79, 0x10, 0x01, 0x12, 0x23,
0x0a, 0x1f, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x49, 0x6e, 0x63, 0x72, 0x65,
0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
0x6e, 0x10, 0x02, 0x12, 0x14, 0x0a, 0x10, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65,
0x4e, 0x6f, 0x74, 0x68, 0x69, 0x6e, 0x67, 0x10, 0x03, 0x32, 0x8f, 0x03, 0x0a, 0x0b, 0x52, 0x65,
0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1a, 0x0a, 0x04, 0x50, 0x69, 0x6e,
0x67, 0x12, 0x08, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71, 0x1a, 0x08, 0x2e, 0x50, 0x69,
0x6e, 0x67, 0x52, 0x65, 0x73, 0x12, 0x39, 0x0a, 0x0f, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c,
0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x73, 0x12, 0x12, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x46,
0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x52, 0x65, 0x71, 0x1a, 0x12, 0x2e, 0x4c,
0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x52, 0x65, 0x73,
0x12, 0x50, 0x0a, 0x16, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74,
0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x1a, 0x2e, 0x4c, 0x69, 0x73,
0x22, 0x54, 0x0a, 0x14, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43,
0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x12, 0x14, 0x0a, 0x04, 0x47, 0x75, 0x69, 0x64,
0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x48, 0x00, 0x52, 0x04, 0x47, 0x75, 0x69, 0x64, 0x12, 0x1c,
0x0a, 0x08, 0x4e, 0x6f, 0x74, 0x65, 0x78, 0x69, 0x73, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08,
0x48, 0x00, 0x52, 0x08, 0x4e, 0x6f, 0x74, 0x65, 0x78, 0x69, 0x73, 0x74, 0x42, 0x08, 0x0a, 0x06,
0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x22, 0x23, 0x0a, 0x07, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65,
0x71, 0x12, 0x18, 0x0a, 0x07, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x01, 0x20, 0x01,
0x28, 0x09, 0x52, 0x07, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x1d, 0x0a, 0x07, 0x50,
0x69, 0x6e, 0x67, 0x52, 0x65, 0x73, 0x12, 0x12, 0x0a, 0x04, 0x45, 0x63, 0x68, 0x6f, 0x18, 0x01,
0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x45, 0x63, 0x68, 0x6f, 0x2a, 0x28, 0x0a, 0x03, 0x54, 0x72,
0x69, 0x12, 0x0c, 0x0a, 0x08, 0x44, 0x6f, 0x6e, 0x74, 0x43, 0x61, 0x72, 0x65, 0x10, 0x00, 0x12,
0x09, 0x0a, 0x05, 0x46, 0x61, 0x6c, 0x73, 0x65, 0x10, 0x01, 0x12, 0x08, 0x0a, 0x04, 0x54, 0x72,
0x75, 0x65, 0x10, 0x02, 0x2a, 0x86, 0x01, 0x0a, 0x18, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61,
0x74, 0x69, 0x6f, 0x6e, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x4b, 0x69, 0x6e,
0x64, 0x12, 0x14, 0x0a, 0x10, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x49, 0x6e,
0x76, 0x61, 0x6c, 0x69, 0x64, 0x10, 0x00, 0x12, 0x19, 0x0a, 0x15, 0x47, 0x75, 0x61, 0x72, 0x61,
0x6e, 0x74, 0x65, 0x65, 0x52, 0x65, 0x73, 0x75, 0x6d, 0x61, 0x62, 0x69, 0x6c, 0x69, 0x74, 0x79,
0x10, 0x01, 0x12, 0x23, 0x0a, 0x1f, 0x47, 0x75, 0x61, 0x72, 0x61, 0x6e, 0x74, 0x65, 0x65, 0x49,
0x6e, 0x63, 0x72, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x61, 0x6c, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
0x61, 0x74, 0x69, 0x6f, 0x6e, 0x10, 0x02, 0x12, 0x14, 0x0a, 0x10, 0x47, 0x75, 0x61, 0x72, 0x61,
0x6e, 0x74, 0x65, 0x65, 0x4e, 0x6f, 0x74, 0x68, 0x69, 0x6e, 0x67, 0x10, 0x03, 0x32, 0xf0, 0x02,
0x0a, 0x0b, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1a, 0x0a,
0x04, 0x50, 0x69, 0x6e, 0x67, 0x12, 0x08, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71, 0x1a,
0x08, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x73, 0x12, 0x39, 0x0a, 0x0f, 0x4c, 0x69, 0x73,
0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x73, 0x12, 0x12, 0x2e, 0x4c,
0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x52, 0x65, 0x71,
0x1a, 0x12, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65,
0x6d, 0x52, 0x65, 0x73, 0x12, 0x50, 0x0a, 0x16, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65,
0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x1a,
0x2e, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56,
0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x71, 0x1a, 0x1a, 0x2e, 0x4c, 0x69, 0x73,
0x74, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69,
0x6f, 0x6e, 0x73, 0x52, 0x65, 0x71, 0x1a, 0x1a, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x46, 0x69, 0x6c,
0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52,
0x65, 0x73, 0x12, 0x3e, 0x0a, 0x10, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61,
0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x12, 0x14, 0x2e, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79,
0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x71, 0x1a, 0x14, 0x2e, 0x44,
0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52,
0x65, 0x73, 0x12, 0x41, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x12, 0x15, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x1a, 0x15,
0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73,
0x6f, 0x72, 0x52, 0x65, 0x73, 0x12, 0x1d, 0x0a, 0x07, 0x53, 0x65, 0x6e, 0x64, 0x44, 0x72, 0x79,
0x12, 0x08, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x1a, 0x08, 0x2e, 0x53, 0x65, 0x6e,
0x64, 0x52, 0x65, 0x73, 0x12, 0x35, 0x0a, 0x0d, 0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70,
0x6c, 0x65, 0x74, 0x65, 0x64, 0x12, 0x11, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70,
0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x1a, 0x11, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x43,
0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73, 0x42, 0x07, 0x5a, 0x05, 0x2e,
0x3b, 0x70, 0x64, 0x75, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
0x6f, 0x6e, 0x73, 0x52, 0x65, 0x73, 0x12, 0x3e, 0x0a, 0x10, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f,
0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x12, 0x14, 0x2e, 0x44, 0x65, 0x73,
0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x71,
0x1a, 0x14, 0x2e, 0x44, 0x65, 0x73, 0x74, 0x72, 0x6f, 0x79, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68,
0x6f, 0x74, 0x73, 0x52, 0x65, 0x73, 0x12, 0x41, 0x0a, 0x11, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63,
0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x12, 0x15, 0x2e, 0x52, 0x65,
0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52,
0x65, 0x71, 0x1a, 0x15, 0x2e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e,
0x43, 0x75, 0x72, 0x73, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x12, 0x35, 0x0a, 0x0d, 0x53, 0x65, 0x6e,
0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x12, 0x11, 0x2e, 0x53, 0x65, 0x6e,
0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x1a, 0x11, 0x2e,
0x53, 0x65, 0x6e, 0x64, 0x43, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73,
0x42, 0x07, 0x5a, 0x05, 0x2e, 0x3b, 0x70, 0x64, 0x75, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f,
0x33,
}
var (
@@ -1606,31 +1625,30 @@ var file_pdu_proto_depIdxs = []int32{
11, // 7: ReplicationConfig.protection:type_name -> ReplicationConfigProtection
1, // 8: ReplicationConfigProtection.Initial:type_name -> ReplicationGuaranteeKind
1, // 9: ReplicationConfigProtection.Incremental:type_name -> ReplicationGuaranteeKind
9, // 10: SendCompletedReq.OriginalReq:type_name -> SendReq
8, // 11: ReceiveReq.To:type_name -> FilesystemVersion
10, // 12: ReceiveReq.ReplicationConfig:type_name -> ReplicationConfig
8, // 13: DestroySnapshotsReq.Snapshots:type_name -> FilesystemVersion
8, // 14: DestroySnapshotRes.Snapshot:type_name -> FilesystemVersion
19, // 15: DestroySnapshotsRes.Results:type_name -> DestroySnapshotRes
23, // 16: Replication.Ping:input_type -> PingReq
3, // 17: Replication.ListFilesystems:input_type -> ListFilesystemReq
6, // 18: Replication.ListFilesystemVersions:input_type -> ListFilesystemVersionsReq
18, // 19: Replication.DestroySnapshots:input_type -> DestroySnapshotsReq
21, // 20: Replication.ReplicationCursor:input_type -> ReplicationCursorReq
9, // 21: Replication.SendDry:input_type -> SendReq
12, // 10: SendRes.Properties:type_name -> Property
9, // 11: SendCompletedReq.OriginalReq:type_name -> SendReq
8, // 12: ReceiveReq.To:type_name -> FilesystemVersion
10, // 13: ReceiveReq.ReplicationConfig:type_name -> ReplicationConfig
8, // 14: DestroySnapshotsReq.Snapshots:type_name -> FilesystemVersion
8, // 15: DestroySnapshotRes.Snapshot:type_name -> FilesystemVersion
19, // 16: DestroySnapshotsRes.Results:type_name -> DestroySnapshotRes
23, // 17: Replication.Ping:input_type -> PingReq
3, // 18: Replication.ListFilesystems:input_type -> ListFilesystemReq
6, // 19: Replication.ListFilesystemVersions:input_type -> ListFilesystemVersionsReq
18, // 20: Replication.DestroySnapshots:input_type -> DestroySnapshotsReq
21, // 21: Replication.ReplicationCursor:input_type -> ReplicationCursorReq
14, // 22: Replication.SendCompleted:input_type -> SendCompletedReq
24, // 23: Replication.Ping:output_type -> PingRes
4, // 24: Replication.ListFilesystems:output_type -> ListFilesystemRes
7, // 25: Replication.ListFilesystemVersions:output_type -> ListFilesystemVersionsRes
20, // 26: Replication.DestroySnapshots:output_type -> DestroySnapshotsRes
22, // 27: Replication.ReplicationCursor:output_type -> ReplicationCursorRes
13, // 28: Replication.SendDry:output_type -> SendRes
15, // 29: Replication.SendCompleted:output_type -> SendCompletedRes
23, // [23:30] is the sub-list for method output_type
16, // [16:23] is the sub-list for method input_type
16, // [16:16] is the sub-list for extension type_name
16, // [16:16] is the sub-list for extension extendee
0, // [0:16] is the sub-list for field type_name
15, // 28: Replication.SendCompleted:output_type -> SendCompletedRes
23, // [23:29] is the sub-list for method output_type
17, // [17:23] is the sub-list for method input_type
17, // [17:17] is the sub-list for extension type_name
17, // [17:17] is the sub-list for extension extendee
0, // [0:17] is the sub-list for field type_name
}
func init() { file_pdu_proto_init() }
+8 -5
View File
@@ -8,7 +8,6 @@ service Replication {
returns (ListFilesystemVersionsRes);
rpc DestroySnapshots(DestroySnapshotsReq) returns (DestroySnapshotsRes);
rpc ReplicationCursor(ReplicationCursorReq) returns (ReplicationCursorRes);
rpc SendDry(SendReq) returns (SendRes);
rpc SendCompleted(SendCompletedReq) returns (SendCompletedRes);
// for Send and Recv, see package rpc
}
@@ -61,7 +60,9 @@ message SendReq {
string ResumeToken = 4;
Tri Encrypted = 5;
ReplicationConfig ReplicationConfig = 6;
bool DryRun = 6;
ReplicationConfig ReplicationConfig = 7;
}
message ReplicationConfig {
@@ -88,12 +89,14 @@ message Property {
message SendRes {
// Whether the resume token provided in the request has been used or not.
// If the SendReq.ResumeToken == "", this field MUST be false.
bool UsedResumeToken = 1;
// If the SendReq.ResumeToken == "", this field has no meaning.
bool UsedResumeToken = 2;
// Expected stream size determined by dry run, not exact.
// 0 indicates that for the given SendReq, no size estimate could be made.
uint64 ExpectedSize = 2;
int64 ExpectedSize = 3;
repeated Property Properties = 4;
}
message SendCompletedReq {
-36
View File
@@ -23,7 +23,6 @@ type ReplicationClient interface {
ListFilesystemVersions(ctx context.Context, in *ListFilesystemVersionsReq, opts ...grpc.CallOption) (*ListFilesystemVersionsRes, error)
DestroySnapshots(ctx context.Context, in *DestroySnapshotsReq, opts ...grpc.CallOption) (*DestroySnapshotsRes, error)
ReplicationCursor(ctx context.Context, in *ReplicationCursorReq, opts ...grpc.CallOption) (*ReplicationCursorRes, error)
SendDry(ctx context.Context, in *SendReq, opts ...grpc.CallOption) (*SendRes, error)
SendCompleted(ctx context.Context, in *SendCompletedReq, opts ...grpc.CallOption) (*SendCompletedRes, error)
}
@@ -80,15 +79,6 @@ func (c *replicationClient) ReplicationCursor(ctx context.Context, in *Replicati
return out, nil
}
func (c *replicationClient) SendDry(ctx context.Context, in *SendReq, opts ...grpc.CallOption) (*SendRes, error) {
out := new(SendRes)
err := c.cc.Invoke(ctx, "/Replication/SendDry", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *replicationClient) SendCompleted(ctx context.Context, in *SendCompletedReq, opts ...grpc.CallOption) (*SendCompletedRes, error) {
out := new(SendCompletedRes)
err := c.cc.Invoke(ctx, "/Replication/SendCompleted", in, out, opts...)
@@ -107,7 +97,6 @@ type ReplicationServer interface {
ListFilesystemVersions(context.Context, *ListFilesystemVersionsReq) (*ListFilesystemVersionsRes, error)
DestroySnapshots(context.Context, *DestroySnapshotsReq) (*DestroySnapshotsRes, error)
ReplicationCursor(context.Context, *ReplicationCursorReq) (*ReplicationCursorRes, error)
SendDry(context.Context, *SendReq) (*SendRes, error)
SendCompleted(context.Context, *SendCompletedReq) (*SendCompletedRes, error)
mustEmbedUnimplementedReplicationServer()
}
@@ -131,9 +120,6 @@ func (UnimplementedReplicationServer) DestroySnapshots(context.Context, *Destroy
func (UnimplementedReplicationServer) ReplicationCursor(context.Context, *ReplicationCursorReq) (*ReplicationCursorRes, error) {
return nil, status.Errorf(codes.Unimplemented, "method ReplicationCursor not implemented")
}
func (UnimplementedReplicationServer) SendDry(context.Context, *SendReq) (*SendRes, error) {
return nil, status.Errorf(codes.Unimplemented, "method SendDry not implemented")
}
func (UnimplementedReplicationServer) SendCompleted(context.Context, *SendCompletedReq) (*SendCompletedRes, error) {
return nil, status.Errorf(codes.Unimplemented, "method SendCompleted not implemented")
}
@@ -240,24 +226,6 @@ func _Replication_ReplicationCursor_Handler(srv interface{}, ctx context.Context
return interceptor(ctx, in, info, handler)
}
func _Replication_SendDry_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(SendReq)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(ReplicationServer).SendDry(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/Replication/SendDry",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(ReplicationServer).SendDry(ctx, req.(*SendReq))
}
return interceptor(ctx, in, info, handler)
}
func _Replication_SendCompleted_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(SendCompletedReq)
if err := dec(in); err != nil {
@@ -303,10 +271,6 @@ var Replication_ServiceDesc = grpc.ServiceDesc{
MethodName: "ReplicationCursor",
Handler: _Replication_ReplicationCursor_Handler,
},
{
MethodName: "SendDry",
Handler: _Replication_SendDry_Handler,
},
{
MethodName: "SendCompleted",
Handler: _Replication_SendCompleted_Handler,
+7 -7
View File
@@ -41,7 +41,6 @@ type Sender interface {
// any next call to the parent github.com/zrepl/zrepl/replication.Endpoint.
// If the send request is for dry run the io.ReadCloser will be nil
Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error)
SendDry(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, error)
SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error)
ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
}
@@ -159,7 +158,7 @@ type Step struct {
encrypt tri
resumeToken string // empty means no resume token shall be used
expectedSize uint64 // 0 means no size estimate present / possible
expectedSize int64 // 0 means no size estimate present / possible
// byteCounter is nil initially, and set later in Step.doReplication
// => concurrent read of that pointer from Step.ReportInfo must be protected
@@ -190,7 +189,7 @@ func (s *Step) Step(ctx context.Context) error {
func (s *Step) ReportInfo() *report.StepInfo {
// get current byteCounter value
var byteCounter uint64
var byteCounter int64
s.byteCounterMtx.Lock()
if s.byteCounter != nil {
byteCounter = s.byteCounter.Count()
@@ -547,10 +546,10 @@ func (s *Step) updateSizeEstimate(ctx context.Context) error {
log := getLogger(ctx)
sr := s.buildSendRequest()
sr := s.buildSendRequest(true)
log.Debug("initiate dry run send request")
sres, err := s.sender.SendDry(ctx, sr)
sres, _, err := s.sender.Send(ctx, sr)
if err != nil {
log.WithError(err).Error("dry run send request failed")
return err
@@ -564,7 +563,7 @@ func (s *Step) updateSizeEstimate(ctx context.Context) error {
return nil
}
func (s *Step) buildSendRequest() (sr *pdu.SendReq) {
func (s *Step) buildSendRequest(dryRun bool) (sr *pdu.SendReq) {
fs := s.parent.Path
sr = &pdu.SendReq{
Filesystem: fs,
@@ -572,6 +571,7 @@ func (s *Step) buildSendRequest() (sr *pdu.SendReq) {
To: s.to,
Encrypted: s.encrypt.ToPDU(),
ResumeToken: s.resumeToken,
DryRun: dryRun,
ReplicationConfig: s.parent.policy.ReplicationConfig,
}
return sr
@@ -582,7 +582,7 @@ func (s *Step) doReplication(ctx context.Context) error {
fs := s.parent.Path
log := getLogger(ctx).WithField("filesystem", fs)
sr := s.buildSendRequest()
sr := s.buildSendRequest(false)
log.Debug("initiate send request")
sres, stream, err := s.sender.Send(ctx, sr)
+4 -16
View File
@@ -62,23 +62,11 @@ const (
FilesystemDone FilesystemState = "done"
)
type FsBlockedOn string
const (
FsBlockedOnNothing FsBlockedOn = "nothing"
FsBlockedOnPlanningStepQueue FsBlockedOn = "plan-queue"
FsBlockedOnParentInitialRepl FsBlockedOn = "parent-initial-repl"
FsBlockedOnReplStepQueue FsBlockedOn = "repl-queue"
)
type FilesystemReport struct {
Info *FilesystemInfo
State FilesystemState
// Always valid.
BlockedOn FsBlockedOn
// Valid in State = FilesystemPlanningErrored
PlanError *TimedError
// Valid in State = FilesystemSteppingErrored
@@ -109,11 +97,11 @@ type StepInfo struct {
From, To string
Resumed bool
Encrypted EncryptedEnum
BytesExpected uint64
BytesReplicated uint64
BytesExpected int64
BytesReplicated int64
}
func (a *AttemptReport) BytesSum() (expected, replicated uint64, containsInvalidSizeEstimates bool) {
func (a *AttemptReport) BytesSum() (expected, replicated int64, containsInvalidSizeEstimates bool) {
for _, fs := range a.Filesystems {
e, r, fsContainsInvalidEstimate := fs.BytesSum()
containsInvalidSizeEstimates = containsInvalidSizeEstimates || fsContainsInvalidEstimate
@@ -123,7 +111,7 @@ func (a *AttemptReport) BytesSum() (expected, replicated uint64, containsInvalid
return expected, replicated, containsInvalidSizeEstimates
}
func (f *FilesystemReport) BytesSum() (expected, replicated uint64, containsInvalidSizeEstimates bool) {
func (f *FilesystemReport) BytesSum() (expected, replicated int64, containsInvalidSizeEstimates bool) {
for _, step := range f.Steps {
expected += step.Info.BytesExpected
replicated += step.Info.BytesReplicated
+13 -3
View File
@@ -114,6 +114,12 @@ func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, i
if err != nil {
return nil, nil, err
}
putWireOnReturn := true
defer func() {
if putWireOnReturn {
c.putWire(conn)
}
}()
if err := c.send(ctx, conn, EndpointSend, req, nil); err != nil {
return nil, nil, err
@@ -125,10 +131,14 @@ func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, i
}
var stream io.ReadCloser
stream, err = conn.ReadStream(ZFSStream, true) // no shadow
if err != nil {
return nil, nil, err
if !req.DryRun {
putWireOnReturn = false
stream, err = conn.ReadStream(ZFSStream, true) // no shadow
if err != nil {
return nil, nil, err
}
}
return &res, stream, nil
}
+4 -10
View File
@@ -172,15 +172,6 @@ func (s *Server) serveConnRequest(ctx context.Context, endpoint string, c *strea
return
}
res, sendStream, handlerErr = s.h.Send(ctx, &req) // SHADOWING
// ensure that we always close the sendStream
if sendStream != nil {
defer func() {
err := sendStream.Close()
if err != nil {
s.log.WithError(err).Error("cannot close send stream")
}
}()
}
case EndpointRecv:
var req pdu.ReceiveReq
if err := proto.Unmarshal(reqStructured, &req); err != nil {
@@ -248,9 +239,12 @@ func (s *Server) serveConnRequest(ctx context.Context, endpoint string, c *strea
if sendStream != nil {
err := c.SendStream(ctx, sendStream, ZFSStream)
closeErr := sendStream.Close()
if closeErr != nil {
s.log.WithError(closeErr).Error("cannot close send stream")
}
if err != nil {
s.log.WithError(err).Error("cannot write send stream")
}
// sendStream.Close() done via defer above
}
}
@@ -1,4 +1,3 @@
//go:build illumos || solaris
// +build illumos solaris
package timeoutconn
@@ -14,7 +14,6 @@ import (
"github.com/stretchr/testify/require"
"github.com/zrepl/zrepl/util/socketpair"
"github.com/zrepl/zrepl/util/zreplcircleci"
)
func TestReadTimeout(t *testing.T) {
@@ -82,8 +81,6 @@ func TestWriteTimeout(t *testing.T) {
}
func TestNoPartialReadsDueToDeadline(t *testing.T) {
zreplcircleci.SkipOnCircleCI(t, "needs predictable low scheduling latency")
a, b, err := socketpair.SocketPair()
require.NoError(t, err)
defer a.Close()
@@ -154,7 +151,6 @@ func (c *partialWriteMockConn) Write(p []byte) (int, error) {
}
func TestPartialWriteMockConn(t *testing.T) {
zreplcircleci.SkipOnCircleCI(t, "because it relies on scheduler responsiveness < 50ms")
mc := newPartialWriteMockConn(100*time.Millisecond, 5)
buf := []byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
begin := time.Now()
+2 -2
View File
@@ -1,5 +1,5 @@
//go:build !illumos && !solaris
// +build !illumos,!solaris
// +build !illumos
// +build !solaris
package timeoutconn
-7
View File
@@ -108,13 +108,6 @@ func (c *Client) Receive(ctx context.Context, req *pdu.ReceiveReq, stream io.Rea
return c.dataClient.ReqRecv(ctx, req, stream)
}
func (c *Client) SendDry(ctx context.Context, in *pdu.SendReq) (*pdu.SendRes, error) {
ctx, endSpan := trace.WithSpan(ctx, "rpc.client.SendDry")
defer endSpan()
return c.controlClient.SendDry(ctx, in)
}
func (c *Client) ListFilesystems(ctx context.Context, in *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {
ctx, endSpan := trace.WithSpan(ctx, "rpc.client.ListFilesystems")
defer endSpan()
+3
View File
@@ -7,6 +7,7 @@ package transportmux
import (
"context"
"os"
"sync/atomic"
"syscall"
@@ -15,6 +16,7 @@ import (
"net"
"time"
"github.com/kr/pretty"
"github.com/zrepl/zrepl/daemon/logging"
"github.com/zrepl/zrepl/logger"
"github.com/zrepl/zrepl/transport"
@@ -151,6 +153,7 @@ func Demux(ctx context.Context, rawListener transport.AuthenticatedListener, lab
rawConn, err := rawListener.Accept(ctx)
if err != nil {
fmt.Fprintf(os.Stderr, "transportmux.Demux: rawListener.Accept() returned error: %T %s\n%s\n", err, err, pretty.Sprint(err))
if ctx.Err() != nil {
return
}
+5 -1
View File
@@ -10,9 +10,12 @@ import (
"fmt"
"io"
"net"
"os"
"strings"
"time"
"unicode/utf8"
"github.com/kr/pretty"
)
type HandshakeMessage struct {
@@ -93,6 +96,7 @@ func (m *HandshakeMessage) Encode() ([]byte, error) {
func (m *HandshakeMessage) DecodeReader(r io.Reader, maxLen int) error {
var lenAndSpace [11]byte
if _, err := io.ReadFull(r, lenAndSpace[:]); err != nil {
fmt.Fprintf(os.Stderr, "HandshakeMessage.DecodeReader error: %T\n%s", err, pretty.Sprint(err))
return hsIOErr(err, "error reading protocol banner length: %s", err)
}
if !utf8.Valid(lenAndSpace[:]) {
@@ -152,7 +156,7 @@ func (m *HandshakeMessage) DecodeReader(r io.Reader, maxLen int) error {
func DoHandshakeCurrentVersion(conn net.Conn, deadline time.Time) *HandshakeError {
// current protocol version is hardcoded here
return DoHandshakeVersion(conn, deadline, 6)
return DoHandshakeVersion(conn, deadline, 5)
}
const HandshakeMessageMaxLen = 16 * 4096
@@ -2,7 +2,9 @@ package versionhandshake
import (
"context"
"fmt"
"net"
"os"
"time"
"github.com/zrepl/zrepl/transport"
@@ -23,6 +25,7 @@ func (c HandshakeConnecter) Connect(ctx context.Context) (transport.Wire, error)
dl = time.Now().Add(c.timeout)
}
if err := DoHandshakeCurrentVersion(conn, dl); err != nil {
fmt.Fprintf(os.Stderr, "HandshakeConnecter error: %T\n\t%s\n\t%v\n\t%#v\n\n", err, err, err, err)
conn.Close()
return nil, err
}
+3
View File
@@ -3,7 +3,9 @@ package tls
import (
"context"
"crypto/tls"
"fmt"
"net"
"os"
"github.com/pkg/errors"
@@ -48,6 +50,7 @@ func TLSConnecterFromConfig(in *config.TLSConnect) (*TLSConnecter, error) {
func (c *TLSConnecter) Connect(dialCtx context.Context) (transport.Wire, error) {
conn, err := c.dialer.DialContext(dialCtx, "tcp", c.Address)
if err != nil {
fmt.Fprintf(os.Stderr, "tls connecter error %T\n\t%s\n\t%v\n\t%#v\n\n", err, err, err, err)
return nil, err
}
tcpConn := conn.(*net.TCPConn)
+140
View File
@@ -0,0 +1,140 @@
package main
import (
"context"
"flag"
"io"
"log"
"os"
"strings"
"time"
"github.com/zrepl/zrepl/config"
"github.com/zrepl/zrepl/transport/tls"
)
var servConf = config.TLSServe{
ServeCommon: config.ServeCommon{
Type: "tls",
},
HandshakeTimeout: 10 * time.Second,
}
var clientConf = config.TLSConnect{
ConnectCommon: config.ConnectCommon{
Type: "",
},
Address: "",
Ca: "",
Cert: "",
Key: "",
ServerCN: "",
DialTimeout: 10 * time.Second,
}
var ca string
var mode string
func main() {
flag.StringVar(&mode, "mode", "", "server|client")
flag.StringVar(&ca, "ca", "", "path")
flag.StringVar(&servConf.Listen, "serve.listen", "", "")
flag.StringVar(&servConf.Cert, "serve.cert", "", "path")
flag.StringVar(&servConf.Key, "serve.key", "", "path")
var clientCN string
flag.StringVar(&clientCN, "serve.client_cn", "", "")
flag.StringVar(&clientConf.Address, "client.address", "", "")
flag.StringVar(&clientConf.Cert, "client.cert", "", "path")
flag.StringVar(&clientConf.Key, "client.key", "", "path")
flag.StringVar(&clientConf.ServerCN, "client.server_cn", "", "")
flag.Parse()
servConf.ClientCNs = append(servConf.ClientCNs, clientCN)
servConf.Ca = ca
clientConf.Ca = ca
switch mode {
case "server":
server()
case "client":
client()
default:
panic(mode)
}
}
func server() {
servFactory, err := tls.TLSListenerFactoryFromConfig(nil, &servConf)
if err != nil {
panic(err)
}
listener, err := servFactory()
if err != nil {
panic(err)
}
ctx := context.Background()
for {
conn, err := listener.Accept(ctx)
if err != nil {
log.Printf("accept error: %s", err)
continue
}
go func() {
defer conn.Close()
log.Printf("handling connection %s", conn.RemoteAddr())
_, err = io.Copy(conn, strings.NewReader("here is the server\n"))
if err != nil {
log.Printf("%s: respond to client error: %s", conn.RemoteAddr(), err)
return
}
err = conn.CloseWrite()
if err != nil {
log.Printf("%s: failed to close write connection: err", conn.RemoteAddr(), err)
}
log.Printf("%s: waiting for client to close connection", conn.RemoteAddr())
_, err = io.Copy(io.Discard, conn)
if err != nil {
log.Printf("%s: error draining client connection: %s", conn.RemoteAddr(), err)
return
}
log.Printf("%s: done", conn.RemoteAddr())
return
}()
}
}
func client() {
connecter, err := tls.TLSConnecterFromConfig(&clientConf)
if err != nil {
panic(err)
}
ctx := context.Background()
conn, err := connecter.Connect(ctx)
if err != nil {
panic(err)
}
defer conn.Close()
_, err = io.Copy(os.Stdout, conn)
if err != nil {
panic(err)
}
}
+3
View File
@@ -4,8 +4,10 @@ import (
"context"
"crypto/tls"
"fmt"
"os"
"time"
"github.com/kr/pretty"
"github.com/pkg/errors"
"github.com/zrepl/zrepl/config"
@@ -68,6 +70,7 @@ type tlsAuthListener struct {
func (l tlsAuthListener) Accept(ctx context.Context) (*transport.AuthConn, error) {
tcpConn, tlsConn, cn, err := l.ClientAuthListener.Accept()
if err != nil {
fmt.Fprintf(os.Stderr, "tlsAuthListener.Accept: l.ClientAuthListener.Accept returned error %T %s\n%s\n", err, err, pretty.Sprint(err))
return nil, err
}
if _, ok := l.clientCNs[cn]; !ok {
+5 -8
View File
@@ -9,7 +9,7 @@ import (
// its interface and counting the bytes written to during copying.
type ReadCloser interface {
io.ReadCloser
Count() uint64
Count() int64
}
// NewReadCloser wraps rc.
@@ -19,11 +19,11 @@ func NewReadCloser(rc io.ReadCloser) ReadCloser {
type readCloser struct {
rc io.ReadCloser
count uint64
count int64
}
func (r *readCloser) Count() uint64 {
return atomic.LoadUint64(&r.count)
func (r *readCloser) Count() int64 {
return atomic.LoadInt64(&r.count)
}
var _ io.ReadCloser = &readCloser{}
@@ -34,9 +34,6 @@ func (r *readCloser) Close() error {
func (r *readCloser) Read(p []byte) (int, error) {
n, err := r.rc.Read(p)
if n < 0 {
panic("expecting n >= 0")
}
atomic.AddUint64(&r.count, uint64(n))
atomic.AddInt64(&r.count, int64(n))
return n, err
}
-18
View File
@@ -73,24 +73,6 @@ func Int64(varname string, def int64) (d int64) {
return d
}
func Uint64(varname string, def uint64) (d uint64) {
var err error
if v, ok := cache.Load(varname); ok {
return v.(uint64)
}
e := os.Getenv(varname)
if e == "" {
d = def
} else {
d, err = strconv.ParseUint(e, 10, 64)
if err != nil {
panic(err)
}
}
cache.Store(varname, d)
return d
}
func Bool(varname string, def bool) (d bool) {
var err error
if v, ok := cache.Load(varname); ok {
-1
View File
@@ -1,4 +1,3 @@
//go:build freebsd
// +build freebsd
package tcpsock
-1
View File
@@ -1,4 +1,3 @@
//go:build linux
// +build linux
package tcpsock
@@ -1,4 +1,3 @@
//go:build !linux && !freebsd
// +build !linux,!freebsd
package tcpsock
-13
View File
@@ -1,13 +0,0 @@
package zreplcircleci
import (
"fmt"
"os"
"testing"
)
func SkipOnCircleCI(t *testing.T, reasonFmt string, args ...interface{}) {
if os.Getenv("CIRCLECI") != "" {
t.Skipf("This test is skipped in CircleCI. Reason: %s", fmt.Sprintf(reasonFmt, args...))
}
}
@@ -1,51 +0,0 @@
// Code generated by "enumer -type=FilesystemPlaceholderCreateEncryptionValue -trimprefix=FilesystemPlaceholderCreateEncryption"; DO NOT EDIT.
//
package zfs
import (
"fmt"
)
const _FilesystemPlaceholderCreateEncryptionValueName = "InheritOff"
var _FilesystemPlaceholderCreateEncryptionValueIndex = [...]uint8{0, 7, 10}
func (i FilesystemPlaceholderCreateEncryptionValue) String() string {
i -= 1
if i < 0 || i >= FilesystemPlaceholderCreateEncryptionValue(len(_FilesystemPlaceholderCreateEncryptionValueIndex)-1) {
return fmt.Sprintf("FilesystemPlaceholderCreateEncryptionValue(%d)", i+1)
}
return _FilesystemPlaceholderCreateEncryptionValueName[_FilesystemPlaceholderCreateEncryptionValueIndex[i]:_FilesystemPlaceholderCreateEncryptionValueIndex[i+1]]
}
var _FilesystemPlaceholderCreateEncryptionValueValues = []FilesystemPlaceholderCreateEncryptionValue{1, 2}
var _FilesystemPlaceholderCreateEncryptionValueNameToValueMap = map[string]FilesystemPlaceholderCreateEncryptionValue{
_FilesystemPlaceholderCreateEncryptionValueName[0:7]: 1,
_FilesystemPlaceholderCreateEncryptionValueName[7:10]: 2,
}
// FilesystemPlaceholderCreateEncryptionValueString retrieves an enum value from the enum constants string name.
// Throws an error if the param is not part of the enum.
func FilesystemPlaceholderCreateEncryptionValueString(s string) (FilesystemPlaceholderCreateEncryptionValue, error) {
if val, ok := _FilesystemPlaceholderCreateEncryptionValueNameToValueMap[s]; ok {
return val, nil
}
return 0, fmt.Errorf("%s does not belong to FilesystemPlaceholderCreateEncryptionValue values", s)
}
// FilesystemPlaceholderCreateEncryptionValueValues returns all values of the enum
func FilesystemPlaceholderCreateEncryptionValueValues() []FilesystemPlaceholderCreateEncryptionValue {
return _FilesystemPlaceholderCreateEncryptionValueValues
}
// IsAFilesystemPlaceholderCreateEncryptionValue returns "true" if the value is listed in the enum definition. "false" otherwise
func (i FilesystemPlaceholderCreateEncryptionValue) IsAFilesystemPlaceholderCreateEncryptionValue() bool {
for _, v := range _FilesystemPlaceholderCreateEncryptionValueValues {
if i == v {
return true
}
}
return false
}
+4 -51
View File
@@ -80,15 +80,7 @@ func ZFSGetFilesystemPlaceholderState(ctx context.Context, p *DatasetPath) (stat
return state, nil
}
//go:generate enumer -type=FilesystemPlaceholderCreateEncryptionValue -trimprefix=FilesystemPlaceholderCreateEncryption
type FilesystemPlaceholderCreateEncryptionValue int
const (
FilesystemPlaceholderCreateEncryptionInherit FilesystemPlaceholderCreateEncryptionValue = 1 << iota
FilesystemPlaceholderCreateEncryptionOff
)
func ZFSCreatePlaceholderFilesystem(ctx context.Context, fs *DatasetPath, parent *DatasetPath, encryption FilesystemPlaceholderCreateEncryptionValue) (err error) {
func ZFSCreatePlaceholderFilesystem(ctx context.Context, fs *DatasetPath, parent *DatasetPath) (err error) {
if fs.Length() == 1 {
return fmt.Errorf("cannot create %q: pools cannot be created with zfs create", fs.ToString())
}
@@ -98,19 +90,11 @@ func ZFSCreatePlaceholderFilesystem(ctx context.Context, fs *DatasetPath, parent
"-o", fmt.Sprintf("%s=%s", PlaceholderPropertyName, placeholderPropertyOn),
"-o", "mountpoint=none",
}
if !encryption.IsAFilesystemPlaceholderCreateEncryptionValue() {
panic(encryption)
}
switch encryption {
case FilesystemPlaceholderCreateEncryptionInherit:
// no-op
case FilesystemPlaceholderCreateEncryptionOff:
if parentEncrypted, err := ZFSGetEncryptionEnabled(ctx, parent.ToString()); err != nil {
return errors.Wrap(err, "cannot determine encryption support")
} else if parentEncrypted {
cmdline = append(cmdline, "-o", "encryption=off")
default:
panic(encryption)
}
cmdline = append(cmdline, fs.ToString())
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, cmdline...)
@@ -164,34 +148,3 @@ func ZFSMigrateHashBasedPlaceholderToCurrent(ctx context.Context, fs *DatasetPat
}
return &report, nil
}
func ZFSListPlaceholderFilesystemsWithAdditionalProps(ctx context.Context, root string, additionalProps []string) (map[string]*ZFSProperties, error) {
props := []string{PlaceholderPropertyName}
if len(additionalProps) > 0 {
props = append(props, additionalProps...)
}
propsByFS, err := zfsGetRecursive(ctx, root, -1, []string{"filesystem", "volume"}, props, SourceAny)
if err != nil {
return nil, errors.Wrapf(err, "cannot get placeholder filesystems under %q", root)
}
filtered := make(map[string]*ZFSProperties)
for fs, props := range propsByFS {
details := props.GetDetails(PlaceholderPropertyName)
if details.Source != SourceLocal {
continue
}
fsp, err := NewDatasetPath(fs)
if err != nil {
return nil, errors.Wrapf(err, "zfs get returned invalid dataset path %q", fs)
}
if !isLocalPlaceholderPropertyValuePlaceholder(fsp, details.Value) {
continue
}
filtered[fs] = props
}
return filtered, nil
}
+81 -188
View File
@@ -7,7 +7,6 @@ import (
"encoding/json"
"fmt"
"io"
"math"
"os"
"os/exec"
"regexp"
@@ -15,6 +14,7 @@ import (
"strconv"
"strings"
"sync"
"time"
"github.com/pkg/errors"
"github.com/prometheus/client_golang/prometheus"
@@ -334,122 +334,60 @@ func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
return stdoutReader, stdoutWriter, nil
}
type sendStreamState int
const (
sendStreamOpen sendStreamState = iota
sendStreamClosed
)
type SendStream struct {
cmd *zfscmd.Cmd
kill context.CancelFunc
stdoutReader io.ReadCloser // not *os.File for mocking during platformtest
stderrBuf *circlog.CircularLog
cmd *zfscmd.Cmd
kill context.CancelFunc
mtx sync.Mutex
state sendStreamState
exitErr *ZFSError
closeMtx sync.Mutex
stdoutReader *os.File
stderrBuf *circlog.CircularLog
opErr error
}
func (s *SendStream) Read(p []byte) (n int, _ error) {
s.mtx.Lock()
defer s.mtx.Unlock()
switch s.state {
case sendStreamClosed:
return 0, os.ErrClosed
case sendStreamOpen:
n, readErr := s.stdoutReader.Read(p)
if readErr != nil {
debug("sendStream: read: readErr=%T %s", readErr, readErr)
if readErr == io.EOF {
// io.EOF must be bubbled up as is so that consumers can handle it properly.
return n, readErr
}
// Assume that the error is not retryable.
// Try to kill now so that we can return a nice *ZFSError with captured stderr.
// If the kill doesn't work, it doesn't matter because the caller must by contract call Close() anyways.
killErr := s.killAndWait()
debug("sendStream: read: killErr=%T %s", killErr, killErr)
if killErr == nil {
s.state = sendStreamClosed
return n, s.exitErr // return the nice error
} else {
// we remain open so that we retry
return n, readErr // return the normal error
}
}
return n, readErr
default:
panic("unreachable")
func (s *SendStream) Read(p []byte) (n int, err error) {
s.closeMtx.Lock()
opErr := s.opErr
s.closeMtx.Unlock()
if opErr != nil {
return 0, opErr
}
n, err = s.stdoutReader.Read(p)
if err != nil {
debug("sendStream: read err: %T %s", err, err)
// TODO we assume here that any read error is permanent
// which is most likely the case for a local zfs send
kwerr := s.killAndWait(err)
debug("sendStream: killAndWait n=%v err= %T %s", n, kwerr, kwerr)
// TODO we assume here that any read error is permanent
return n, kwerr
}
return n, err
}
func (s *SendStream) Close() error {
debug("sendStream: close called")
s.mtx.Lock()
defer s.mtx.Unlock()
switch s.state {
case sendStreamOpen:
err := s.killAndWait()
if err != nil {
return err
} else {
s.state = sendStreamClosed
return nil
}
case sendStreamClosed:
return os.ErrClosed
default:
panic("unreachable")
}
return s.killAndWait(nil)
}
// returns nil iff the child process is gone (has been successfully waited upon)
// in that case, s.exitErr is set
func (s *SendStream) killAndWait() error {
func (s *SendStream) killAndWait(precedingReadErr error) error {
debug("sendStream: killAndWait enter")
defer debug("sendStream: killAndWait leave")
if precedingReadErr == io.EOF {
// give the zfs process a little bit of time to terminate itself
// if it holds this deadline, exitErr will be nil
time.AfterFunc(200*time.Millisecond, s.kill)
} else {
s.kill()
}
// send SIGKILL
s.kill()
// allow async kills from Close(), that's why we only take the mutex here
s.closeMtx.Lock()
defer s.closeMtx.Unlock()
// Close our read-end of the pipe.
//
// We must do this before .Wait() because in some (not all) versions/build configs of ZFS,
// `zfs send` uses a separate kernel thread (taskq) to write the send stream (function `dump_bytes`).
// The `zfs send` thread then waits uinterruptably for the taskq thread to finish the write.
// And signalling the `zfs send` thread doesn't propagate to the taskq thread.
// So we end up in a state where we .Wait() forever.
// (See https://github.com/openzfs/zfs/issues/12500 and
// https://github.com/zrepl/zrepl/issues/495#issuecomment-902530043)
//
// By closing our read end of the pipe before .Wait(), we unblock the taskq thread if there is any.
// If there is no separate taskq thread, the SIGKILL to `zfs end` would suffice and be most precise,
// but due to the circumstances above, there is no other portable & robust way.
//
// However, the fallout from closing the pipe is that (in non-taskq builds) `zfs sends` will get a SIGPIPE.
// And on Linux, that SIGPIPE appears to win over the previously issued SIGKILL.
// And thus, on Linux, the `zfs send` will be killed by the default SIGPIPE handler.
// We can observe this in the WaitStatus below.
// This behavior is slightly annoying because the *exec.ExitError's message ("signal: broken pipe")
// isn't as clear as ("signal: killed").
// However, it seems like we just have to live with that. (covered by platformtest)
var closePipeErr error
if s.stdoutReader != nil {
closePipeErr = s.stdoutReader.Close()
if closePipeErr == nil {
// avoid double-closes in case waiting below doesn't work
// and someone attempts Close again
s.stdoutReader = nil
} else {
return closePipeErr
}
if s.opErr != nil {
return s.opErr
}
waitErr := s.cmd.Wait()
@@ -464,30 +402,39 @@ func (s *SendStream) killAndWait() error {
}
}
// invariant: at this point, the child is gone and we cleaned up everything related to the SendStream
// now, after we know the program exited do we close the pipe
var closePipeErr error
if s.stdoutReader != nil {
closePipeErr = s.stdoutReader.Close()
if closePipeErr == nil {
// avoid double-closes in case anything below doesn't work
// and someone calls Close again
s.stdoutReader = nil
} else {
return closePipeErr
}
}
// we managed to tear things down, no let's give the user some pretty *ZFSError
if exitErr != nil {
// zfs send exited with an error or was killed by a signal.
s.exitErr = &ZFSError{
s.opErr = &ZFSError{
Stderr: []byte(s.stderrBuf.String()),
WaitErr: exitErr,
}
} else {
// zfs send exited successfully (we know that since waitErr was either nil or wasn't an *exec.ExitError)
s.exitErr = nil
s.opErr = precedingReadErr
}
return nil
}
// detect the edge where we're called from s.Read
// after the pipe EOFed and zfs send exited without errors
// this is actually the "hot" / nice path
if exitErr == nil && precedingReadErr == io.EOF {
return precedingReadErr
}
func (s *SendStream) TestOnly_ReplaceStdoutReader(f io.ReadCloser) (prev io.ReadCloser) {
prev = s.stdoutReader
s.stdoutReader = f
return prev
return s.opErr
}
func (s *SendStream) TestOnly_ExitErr() *ZFSError { return s.exitErr }
// NOTE: When updating this struct, make sure to update funcs Validate ValidateCorrespondsToResumeToken
type ZFSSendArgVersion struct {
RelName string
@@ -961,7 +908,7 @@ type DrySendInfo struct {
Type DrySendType
Filesystem string // parsed from To field
From, To string // direct copy from ZFS output
SizeEstimate uint64 // 0 if size estimate is not possible
SizeEstimate int64 // -1 if size estimate is not possible
}
var (
@@ -1034,10 +981,11 @@ func (s *DrySendInfo) unmarshalInfoLine(l string) (regexMatched bool, err error)
// see https://github.com/zrepl/zrepl/issues/289
fields["size"] = "0"
}
s.SizeEstimate, err = strconv.ParseUint(fields["size"], 10, 64)
s.SizeEstimate, err = strconv.ParseInt(fields["size"], 10, 64)
if err != nil {
return true, fmt.Errorf("cannot not parse size: %s", err)
}
return true, nil
}
@@ -1047,7 +995,6 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
if sendArgs.From != nil && strings.Contains(sendArgs.From.RelName, "#") {
/* TODO:
* XXX feature check & support this as well
* ZFS at the time of writing does not support dry-run send because size-estimation
* uses fromSnap's deadlist. However, for a bookmark, that deadlist no longer exists.
* Redacted send & recv will bring this functionality, see
@@ -1066,7 +1013,7 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
Filesystem: sendArgs.FS,
From: fromAbs,
To: toAbs,
SizeEstimate: 0}, nil
SizeEstimate: -1}, nil
}
args := make([]string, 0)
@@ -1086,19 +1033,6 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
if err := si.unmarshalZFSOutput(output); err != nil {
return nil, fmt.Errorf("could not parse zfs send -n output: %s", err)
}
// There is a bug in OpenZFS where it estimates the size incorrectly.
// - zrepl: https://github.com/zrepl/zrepl/issues/463
// - resulting upstream bug: https://github.com/openzfs/zfs/issues/12265
//
// The wrong estimates are easy to detect because they are absurdly large.
// NB: we're doing the workaround for this late so that the test cases are not affected.
sizeEstimateThreshold := envconst.Uint64("ZREPL_ZFS_SEND_SIZE_ESTIMATE_INCORRECT_THRESHOLD", math.MaxInt64)
if sizeEstimateThreshold != 0 && si.SizeEstimate >= sizeEstimateThreshold {
debug("size estimate exceeds threshold %v, working around it: %#v %q", sizeEstimateThreshold, si, args)
si.SizeEstimate = 0
}
return &si, nil
}
@@ -1567,18 +1501,8 @@ func (s PropertySource) zfsGetSourceFieldPrefixes() []string {
return prefixes
}
func zfsGetRecursive(ctx context.Context, path string, depth int, dstypes []string, props []string, allowedSources PropertySource) (map[string]*ZFSProperties, error) {
args := []string{"get", "-Hp", "-o", "name,property,value,source"}
if depth != 0 {
args = append(args, "-r")
if depth != -1 {
args = append(args, "-d", fmt.Sprintf("%d", depth))
}
}
if len(dstypes) > 0 {
args = append(args, "-t", strings.Join(dstypes, ","))
}
args = append(args, strings.Join(props, ","), path)
func zfsGet(ctx context.Context, path string, props []string, allowedSources PropertySource) (*ZFSProperties, error) {
args := []string{"get", "-Hp", "-o", "property,value,source", strings.Join(props, ","), path}
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
stdout, err := cmd.Output()
if err != nil {
@@ -1598,67 +1522,36 @@ func zfsGetRecursive(ctx context.Context, path string, depth int, dstypes []stri
}
o := string(stdout)
lines := strings.Split(o, "\n")
propsByFS := make(map[string]*ZFSProperties)
if len(lines) < 1 || // account for newlines
len(lines)-1 != len(props) {
return nil, fmt.Errorf("zfs get did not return the number of expected property values")
}
res := &ZFSProperties{
make(map[string]PropertyValue, len(lines)),
}
allowedPrefixes := allowedSources.zfsGetSourceFieldPrefixes()
for _, line := range lines[:len(lines)-1] { // last line is an empty line due to how strings.Split works
for _, line := range lines[:len(lines)-1] {
fields := strings.FieldsFunc(line, func(r rune) bool {
return r == '\t'
})
if len(fields) != 4 {
return nil, fmt.Errorf("zfs get did not return name,property,value,source tuples")
if len(fields) != 3 {
return nil, fmt.Errorf("zfs get did not return property,value,source tuples")
}
for _, p := range allowedPrefixes {
// prefix-match so that SourceAny (= "") works
if strings.HasPrefix(fields[3], p) {
source, err := parsePropertySource(fields[3])
if strings.HasPrefix(fields[2], p) {
source, err := parsePropertySource(fields[2])
if err != nil {
return nil, errors.Wrap(err, "parse property source")
}
fsProps, ok := propsByFS[fields[0]]
if !ok {
fsProps = &ZFSProperties{
make(map[string]PropertyValue),
}
}
if _, ok := fsProps.m[fields[1]]; ok {
return nil, errors.Errorf("duplicate property %q for dataset %q", fields[1], fields[0])
}
fsProps.m[fields[1]] = PropertyValue{
Value: fields[2],
res.m[fields[0]] = PropertyValue{
Value: fields[1],
Source: source,
}
propsByFS[fields[0]] = fsProps
break
}
}
}
// validate we got expected output
for fs, fsProps := range propsByFS {
if len(fsProps.m) != len(props) {
return nil, errors.Errorf("zfs get did not return all requested values for dataset %q\noutput was:\n%s", fs, o)
}
}
return propsByFS, nil
}
func zfsGet(ctx context.Context, path string, props []string, allowedSources PropertySource) (*ZFSProperties, error) {
propMap, err := zfsGetRecursive(ctx, path, 0, nil, props, allowedSources)
if err != nil {
return nil, err
}
if len(propMap) == 0 {
// XXX callers expect to always get a result here
// They will observe props.Get("propname") == ""
// We should change .Get to return a tuple, or an error, or whatever.
return &ZFSProperties{make(map[string]PropertyValue)}, nil
}
if len(propMap) != 1 {
return nil, errors.Errorf("zfs get unexpectedly returned properties for multiple datasets")
}
res, ok := propMap[path]
if !ok {
return nil, errors.Errorf("zfs get returned properties for a different dataset that requested")
}
return res, nil
}
-1
View File
@@ -1,4 +1,3 @@
//go:build !linux
// +build !linux
package zfs
-4
View File
@@ -209,7 +209,3 @@ func (c *Cmd) Runtime() time.Duration {
}
return c.waitReturnedAt.Sub(c.startedAt)
}
func (c *Cmd) TestOnly_ExecCmd() *exec.Cmd {
return c.cmd
}