adminer/zrepl_patched
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

endpoint: refactor, fix stale holds on initial replication failure, zfs-abstractions subcmd, more efficient ZFS queries

Browse Source 
The motivation for this recatoring are based on two independent issues:

- @JMoVS found that the changes merged as part of #259 slowed his OS X
  based installation down significantly.
  Analysis of the zfs command logging introduced in #296 showed that
  `zfs holds` took most of the execution time, and they pointed out
  that not all of those `zfs holds` invocations were actually necessary.
  I.e.: zrepl was inefficient about retrieving information from ZFS.

- @InsanePrawn found that failures on initial replication would lead
  to step holds accumulating on the sending side, i.e. they would never
  be cleaned up in the HintMostRecentCommonAncestor RPC handler.
  That was because we only sent that RPC if there was a most recent
  common ancestor detected during replication planning.
  @InsanePrawn prototyped an implementation of a `zrepl zfs-abstractions release`
  command to mitigate the situation.
  As part of that development work and back-and-forth with @problame,
  it became evident that the abstractions that #259 built on top of
  zfs in package endpoint (step holds, replication cursor,
  last-received-hold), were not well-represented for re-use in the
  `zrepl zfs-abstractions release` subocommand prototype.

This commit refactors package endpoint to address both of these issues:

- endpoint abstractions now share an interface `Abstraction` that, among
  other things, provides a uniform `Destroy()` method.
  However, that method should not be destroyed directly but instead
  the package-level `BatchDestroy` function should be used in order
  to allow for a migration to zfs channel programs in the future.

- endpoint now has a query facitilty (`ListAbstractions`) which is
  used to find on-disk
    - step holds and bookmarks
    - replication cursors (v1, v2)
    - last-received-holds
  By describing the query in a struct, we can centralized the retrieval
  of information via the ZFS CLI and only have to be clever once.
  We are "clever" in the following ways:
  - When asking for hold-based abstractions, we only run `zfs holds` on
    snapshot that have `userrefs` > 0
    - To support this functionality, add field `UserRefs` to zfs.FilesystemVersion
      and retrieve it anywhere we retrieve zfs.FilesystemVersion from ZFS.
  - When asking only for bookmark-based abstractions, we only run
    `zfs list -t bookmark`, not with snapshots.
  - Currently unused (except for CLI) per-filesystem concurrent lookup
  - Option to only include abstractions with CreateTXG in a specified range

- refactor `endpoint`'s various ZFS info  retrieval methods to use
  `ListAbstractions`

- rename the `zrepl holds list` command to `zrepl zfs-abstractions list`
- make `zrepl zfs-abstractions list` consume endpoint.ListAbstractions

- Add a `ListStale` method which, given a query template,
  lists stale holds and bookmarks.
  - it uses replication cursor has different modes
- the new `zrepl zfs-abstractions release-{all,stale}` commands can be used
  to remove abstractions of package endpoint

- Adjust HintMostRecentCommonAncestor RPC for stale-holds cleanup:
    - send it also if no most recent common ancestor exists between sender and receiver
    - have the sender clean up its abstractions when it receives the RPC
      with no most recent common ancestor, using `ListStale`
    - Due to changed semantics, bump the protocol version.

- Adjust HintMostRecentCommonAncestor RPC for performance problems
  encountered by @JMoVS
    - by default, per (job,fs)-combination, only consider cleaning
      step holds in the createtxg range
      `[last replication cursor,conservatively-estimated-receive-side-version)`
    - this behavior ensures resumability at cost proportional to the
      time that replication was donw
    - however, as explained in a comment, we might leak holds if
      the zrepl daemon stops running
    - that  trade-off is acceptable because in the presumably rare
      this might happen the user has two tools at their hand:
    - Tool 1: run `zrepl zfs-abstractions release-stale`
    - Tool 2: use env var `ZREPL_ENDPOINT_SENDER_HINT_MOST_RECENT_STEP_HOLD_CLEANUP_MODE`
      to adjust the lower bound of the createtxg range (search for it in the code).
      The env var can also be used to disable hold-cleanup on the
      send-side entirely.

supersedes closes #293
supersedes closes #282
fixes #280
fixes #278

Additionaly, we fixed a couple of bugs:

- zfs: fix half-nil error reporting of dataset-does-not-exist for ZFSListChan and ZFSBookmark

- endpoint: Sender's `HintMostRecentCommonAncestor` handler would not
  check whether access to the specified filesystem was allowed.
This commit is contained in:
Christian Schwarz
2020-03-26 23:43:17 +01:00
parent 96e188d7c4
commit e0b5bd75f8
46 changed files with 3190 additions and 1342 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zfs/holds.go
+3 -178
View File
@@ -6,8 +6,6 @@ import (
"context"
"fmt"
"os"
"sort"
"strconv"
"strings"
"syscall"
@@ -36,12 +34,9 @@ func ValidHoldTag(tag string) error {
}
// Idemptotent: does not return an error if the tag already exists
func ZFSHold(ctx context.Context, fs string, v ZFSSendArgVersion, tag string) error {
if err := v.ValidateInMemory(fs); err != nil {
return errors.Wrap(err, "invalid version")
}
func ZFSHold(ctx context.Context, fs string, v FilesystemVersion, tag string) error {
if !v.IsSnapshot() {
return errors.Errorf("can only hold snapshots, got %s", v.RelName)
return errors.Errorf("can only hold snapshots, got %s", v.RelName())
}
if err := validateNotEmpty("tag", tag); err != nil {
@@ -133,177 +128,7 @@ func ZFSRelease(ctx context.Context, tag string, snaps ...string) error {
debug("zfs release: no such tag lines=%v otherLines=%v", noSuchTagLines, otherLines)
}
if len(otherLines) > 0 {
return fmt.Errorf("unknown zfs error while releasing hold with tag %q: unidentified stderr lines\n%s", tag, strings.Join(otherLines, "\n"))
return fmt.Errorf("unknown zfs error while releasing hold with tag %q:\n%s", tag, strings.Join(otherLines, "\n"))
}
return nil
}
// Idempotent: if the hold doesn't exist, this is not an error
func ZFSReleaseAllOlderAndIncludingGUID(ctx context.Context, fs string, snapOrBookmarkGuid uint64, tag string) error {
return doZFSReleaseAllOlderAndIncOrExcludingGUID(ctx, fs, snapOrBookmarkGuid, tag, true)
}
// Idempotent: if the hold doesn't exist, this is not an error
func ZFSReleaseAllOlderThanGUID(ctx context.Context, fs string, snapOrBookmarkGuid uint64, tag string) error {
return doZFSReleaseAllOlderAndIncOrExcludingGUID(ctx, fs, snapOrBookmarkGuid, tag, false)
}
type zfsReleaseAllOlderAndIncOrExcludingGUIDZFSListLine struct {
entityType EntityType
name string
createtxg uint64
guid uint64
userrefs uint64 // always 0 for bookmarks
}
func doZFSReleaseAllOlderAndIncOrExcludingGUID(ctx context.Context, fs string, snapOrBookmarkGuid uint64, tag string, includeGuid bool) error {
// TODO channel program support still unreleased but
// might be a huge performance improvement
// https://github.com/zfsonlinux/zfs/pull/7902/files
if err := validateZFSFilesystem(fs); err != nil {
return errors.Wrap(err, "`fs` is not a valid filesystem path")
}
if tag == "" {
return fmt.Errorf("`tag` must not be empty`")
}
output, err := zfscmd.CommandContext(ctx,
"zfs", "list", "-o", "type,name,createtxg,guid,userrefs",
"-H", "-t", "snapshot,bookmark", "-r", "-d", "1", fs).CombinedOutput()
if err != nil {
return &ZFSError{output, errors.Wrap(err, "cannot list snapshots and their userrefs")}
}
lines, err := doZFSReleaseAllOlderAndIncOrExcludingGUIDParseListOutput(output)
if err != nil {
return errors.Wrap(err, "unexpected ZFS output")
}
releaseSnaps, err := doZFSReleaseAllOlderAndIncOrExcludingGUIDFindSnapshots(snapOrBookmarkGuid, includeGuid, lines)
if err != nil {
return err
}
if len(releaseSnaps) == 0 {
return nil
}
return ZFSRelease(ctx, tag, releaseSnaps...)
}
func doZFSReleaseAllOlderAndIncOrExcludingGUIDParseListOutput(output []byte) ([]zfsReleaseAllOlderAndIncOrExcludingGUIDZFSListLine, error) {
scan := bufio.NewScanner(bytes.NewReader(output))
var lines []zfsReleaseAllOlderAndIncOrExcludingGUIDZFSListLine
for scan.Scan() {
const numCols = 5
comps := strings.SplitN(scan.Text(), "\t", numCols)
if len(comps) != numCols {
return nil, fmt.Errorf("not %d columns\n%s", numCols, output)
}
dstype := comps[0]
name := comps[1]
var entityType EntityType
switch dstype {
case "snapshot":
entityType = EntityTypeSnapshot
case "bookmark":
entityType = EntityTypeBookmark
default:
return nil, fmt.Errorf("column 0 is %q, expecting \"snapshot\" or \"bookmark\"", dstype)
}
createtxg, err := strconv.ParseUint(comps[2], 10, 64)
if err != nil {
return nil, fmt.Errorf("cannot parse createtxg %q: %s\n%s", comps[2], err, output)
}
guid, err := strconv.ParseUint(comps[3], 10, 64)
if err != nil {
return nil, fmt.Errorf("cannot parse guid %q: %s\n%s", comps[3], err, output)
}
var userrefs uint64
switch entityType {
case EntityTypeBookmark:
if comps[4] != "-" {
return nil, fmt.Errorf("entity type \"bookmark\" should have userrefs=\"-\", got %q", comps[4])
}
userrefs = 0
case EntityTypeSnapshot:
userrefs, err = strconv.ParseUint(comps[4], 10, 64) // shadow
if err != nil {
return nil, fmt.Errorf("cannot parse userrefs %q: %s\n%s", comps[4], err, output)
}
default:
panic(entityType)
}
lines = append(lines, zfsReleaseAllOlderAndIncOrExcludingGUIDZFSListLine{
entityType: entityType,
name: name,
createtxg: createtxg,
guid: guid,
userrefs: userrefs,
})
}
return lines, nil
}
func doZFSReleaseAllOlderAndIncOrExcludingGUIDFindSnapshots(snapOrBookmarkGuid uint64, includeGuid bool, lines []zfsReleaseAllOlderAndIncOrExcludingGUIDZFSListLine) (releaseSnaps []string, err error) {
// sort lines by createtxg,(snap < bookmark)
// we cannot do this using zfs list -s because `type` is not a
sort.Slice(lines, func(i, j int) (less bool) {
if lines[i].createtxg == lines[j].createtxg {
iET := func(t EntityType) int {
switch t {
case EntityTypeSnapshot:
return 0
case EntityTypeBookmark:
return 1
default:
panic("unexpected entity type " + t.String())
}
}
return iET(lines[i].entityType) < iET(lines[j].entityType)
}
return lines[i].createtxg < lines[j].createtxg
})
// iterate over snapshots oldest to newest and collect snapshots that have holds and
// are older than (inclusive or exclusive, depends on includeGuid) a snapshot or bookmark
// with snapOrBookmarkGuid
foundGuid := false
for _, line := range lines {
if line.guid == snapOrBookmarkGuid {
foundGuid = true
}
if line.userrefs > 0 {
if !foundGuid || (foundGuid && includeGuid) {
// only snapshots have userrefs > 0, no need to check entityType
releaseSnaps = append(releaseSnaps, line.name)
}
}
if foundGuid {
// The secondary key in sorting (snap < bookmark) guarantees that we
// A) either found the snapshot with snapOrBookmarkGuid
// B) or no snapshot with snapGuid exists, but one or more bookmarks of it exists
// In the case of A, we already added the snapshot to releaseSnaps if includeGuid requests it,
// and can ignore possible subsequent bookmarks of the snapshot.
// In the case of B, there is nothing to add to releaseSnaps.
break
}
}
if !foundGuid {
return nil, fmt.Errorf("cannot find snapshot or bookmark with guid %v", snapOrBookmarkGuid)
}
return releaseSnaps, nil
}
zfs/holds_test.go
-38
View File
@@ -1,38 +0,0 @@
package zfs
import (
"testing"
"github.com/kr/pretty"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestDoZFSReleaseAllOlderAndIncOrExcludingGUIDFindSnapshots(t *testing.T) {
// what we test here: sort bookmark #3 before @3
// => assert that the function doesn't stop at the first guid match
// (which might be a bookmark, depending on zfs list ordering)
// but instead considers the entire stride of bookmarks and snapshots with that guid
//
// also, throw in unordered createtxg for good measure
list, err := doZFSReleaseAllOlderAndIncOrExcludingGUIDParseListOutput(
[]byte("snapshot\tfoo@1\t1\t1013001\t1\n" +
"snapshot\tfoo@2\t2\t2013002\t1\n" +
"bookmark\tfoo#3\t3\t7013003\t-\n" +
"snapshot\tfoo@6\t6\t5013006\t1\n" +
"snapshot\tfoo@3\t3\t7013003\t1\n" +
"snapshot\tfoo@4\t3\t6013004\t1\n" +
""),
)
require.NoError(t, err)
t.Log(pretty.Sprint(list))
require.Equal(t, 6, len(list))
require.Equal(t, EntityTypeBookmark, list[2].entityType)
releaseSnaps, err := doZFSReleaseAllOlderAndIncOrExcludingGUIDFindSnapshots(7013003, true, list)
t.Logf("releasedSnaps = %#v", releaseSnaps)
assert.NoError(t, err)
assert.Equal(t, []string{"foo@1", "foo@2", "foo@3"}, releaseSnaps)
}
zfs/mapping.go
+1 -1
View File
@@ -59,7 +59,7 @@ func ZFSListMappingProperties(ctx context.Context, filter DatasetFilter, propert
defer cancel()
rchan := make(chan ZFSListResult)
go ZFSListChan(ctx, rchan, properties, "-r", "-t", "filesystem,volume")
go ZFSListChan(ctx, rchan, properties, nil, "-r", "-t", "filesystem,volume")
datasets = make([]ZFSListMappingPropertiesResult, 0)
for r := range rchan {
zfs/versions.go
+142 -44
View File
@@ -20,6 +20,30 @@ const (
Snapshot VersionType = "snapshot"
)
type VersionTypeSet map[VersionType]bool
var (
AllVersionTypes = VersionTypeSet{
Bookmark: true,
Snapshot: true,
}
Bookmarks = VersionTypeSet{
Bookmark: true,
}
Snapshots = VersionTypeSet{
Snapshot: true,
}
)
func (s VersionTypeSet) zfsListTFlagRepr() string {
var types []string
for t := range s {
types = append(types, t.String())
}
return strings.Join(types, ",")
}
func (s VersionTypeSet) String() string { return s.zfsListTFlagRepr() }
func (t VersionType) DelimiterChar() string {
switch t {
case Bookmark:
@@ -55,6 +79,7 @@ func DecomposeVersionString(v string) (fs string, versionType VersionType, name
}
}
// The data in a FilesystemVersion is guaranteed to stem from a ZFS CLI invocation.
type FilesystemVersion struct {
Type VersionType
@@ -70,11 +95,26 @@ type FilesystemVersion struct {
// The time the dataset was created
Creation time.Time
// userrefs field (snapshots only)
UserRefs OptionUint64
}
func (v FilesystemVersion) String() string {
type OptionUint64 struct {
Value uint64
Valid bool
}
func (v FilesystemVersion) GetCreateTXG() uint64 { return v.CreateTXG }
func (v FilesystemVersion) GetGUID() uint64 { return v.Guid }
func (v FilesystemVersion) GetGuid() uint64 { return v.Guid }
func (v FilesystemVersion) GetName() string { return v.Name }
func (v FilesystemVersion) IsSnapshot() bool { return v.Type == Snapshot }
func (v FilesystemVersion) IsBookmark() bool { return v.Type == Bookmark }
func (v FilesystemVersion) RelName() string {
return fmt.Sprintf("%s%s", v.Type.DelimiterChar(), v.Name)
}
func (v FilesystemVersion) String() string { return v.RelName() }
func (v FilesystemVersion) ToAbsPath(p *DatasetPath) string {
var b bytes.Buffer
@@ -84,24 +124,89 @@ func (v FilesystemVersion) ToAbsPath(p *DatasetPath) string {
return b.String()
}
type FilesystemVersionFilter interface {
Filter(t VersionType, name string) (accept bool, err error)
func (v FilesystemVersion) FullPath(fs string) string {
return fmt.Sprintf("%s%s", fs, v.RelName())
}
type closureFilesystemVersionFilter struct {
cb func(t VersionType, name string) (accept bool, err error)
func (v FilesystemVersion) ToSendArgVersion() ZFSSendArgVersion {
return ZFSSendArgVersion{
RelName: v.RelName(),
GUID: v.Guid,
}
}
func (f *closureFilesystemVersionFilter) Filter(t VersionType, name string) (accept bool, err error) {
return f.cb(t, name)
type ParseFilesystemVersionArgs struct {
fullname string
guid, createtxg, creation, userrefs string
}
func FilterFromClosure(cb func(t VersionType, name string) (accept bool, err error)) FilesystemVersionFilter {
return &closureFilesystemVersionFilter{cb}
func ParseFilesystemVersion(args ParseFilesystemVersionArgs) (v FilesystemVersion, err error) {
_, v.Type, v.Name, err = DecomposeVersionString(args.fullname)
if err != nil {
return v, err
}
if v.Guid, err = strconv.ParseUint(args.guid, 10, 64); err != nil {
err = errors.Wrapf(err, "cannot parse GUID %q", args.guid)
return v, err
}
if v.CreateTXG, err = strconv.ParseUint(args.createtxg, 10, 64); err != nil {
err = errors.Wrapf(err, "cannot parse CreateTXG %q", args.createtxg)
return v, err
}
creationUnix, err := strconv.ParseInt(args.creation, 10, 64)
if err != nil {
err = errors.Wrapf(err, "cannot parse creation date %q", args.creation)
return v, err
} else {
v.Creation = time.Unix(creationUnix, 0)
}
switch v.Type {
case Bookmark:
if args.userrefs != "-" {
return v, errors.Errorf("expecting %q for bookmark property userrefs, got %q", "-", args.userrefs)
}
v.UserRefs = OptionUint64{Valid: false}
case Snapshot:
if v.UserRefs.Value, err = strconv.ParseUint(args.userrefs, 10, 64); err != nil {
err = errors.Wrapf(err, "cannot parse userrefs %q", args.userrefs)
return v, err
}
v.UserRefs.Valid = true
default:
panic(v.Type)
}
return v, nil
}
// returned versions are sorted by createtxg
func ZFSListFilesystemVersions(fs *DatasetPath, filter FilesystemVersionFilter) (res []FilesystemVersion, err error) {
type ListFilesystemVersionsOptions struct {
// the prefix of the version name, without the delimiter char
// empty means any prefix matches
ShortnamePrefix string
// which types should be returned
// nil or len(0) means any prefix matches
Types VersionTypeSet
}
func (o *ListFilesystemVersionsOptions) typesFlagArgs() string {
if len(o.Types) == 0 {
return AllVersionTypes.zfsListTFlagRepr()
} else {
return o.Types.zfsListTFlagRepr()
}
}
func (o *ListFilesystemVersionsOptions) matches(v FilesystemVersion) bool {
return (len(o.Types) == 0 || o.Types[v.Type]) && strings.HasPrefix(v.Name, o.ShortnamePrefix)
}
// returned versions are sorted by createtxg FIXME drop sort by createtxg requirement
func ZFSListFilesystemVersions(fs *DatasetPath, options ListFilesystemVersionsOptions) (res []FilesystemVersion, err error) {
listResults := make(chan ZFSListResult)
promTimer := prometheus.NewTimer(prom.ZFSListFilesystemVersionDuration.WithLabelValues(fs.ToString()))
@@ -110,9 +215,10 @@ func ZFSListFilesystemVersions(fs *DatasetPath, filter FilesystemVersionFilter)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go ZFSListChan(ctx, listResults,
[]string{"name", "guid", "createtxg", "creation"},
[]string{"name", "guid", "createtxg", "creation", "userrefs"},
fs,
"-r", "-d", "1",
"-t", "bookmark,snapshot",
"-t", options.typesFlagArgs(),
"-s", "createtxg", fs.ToString())
res = make([]FilesystemVersion, 0)
@@ -126,44 +232,36 @@ func ZFSListFilesystemVersions(fs *DatasetPath, filter FilesystemVersionFilter)
}
line := listResult.Fields
var v FilesystemVersion
_, v.Type, v.Name, err = DecomposeVersionString(line[0])
args := ParseFilesystemVersionArgs{
fullname: line[0],
guid: line[1],
createtxg: line[2],
creation: line[3],
userrefs: line[4],
}
v, err := ParseFilesystemVersion(args)
if err != nil {
return nil, err
}
if v.Guid, err = strconv.ParseUint(line[1], 10, 64); err != nil {
err = errors.Wrap(err, "cannot parse GUID")
return
}
if v.CreateTXG, err = strconv.ParseUint(line[2], 10, 64); err != nil {
err = errors.Wrap(err, "cannot parse CreateTXG")
return
}
creationUnix, err := strconv.ParseInt(line[3], 10, 64)
if err != nil {
err = fmt.Errorf("cannot parse creation date '%s': %s", line[3], err)
return nil, err
} else {
v.Creation = time.Unix(creationUnix, 0)
}
accept := true
if filter != nil {
accept, err = filter.Filter(v.Type, v.Name)
if err != nil {
err = fmt.Errorf("error executing filter: %s", err)
return nil, err
}
}
if accept {
if options.matches(v) {
res = append(res, v)
}
}
return
}
func ZFSGetFilesystemVersion(ctx context.Context, ds string) (v FilesystemVersion, _ error) {
props, err := zfsGet(ctx, ds, []string{"createtxg", "guid", "creation", "userrefs"}, sourceAny)
if err != nil {
return v, err
}
return ParseFilesystemVersion(ParseFilesystemVersionArgs{
fullname: ds,
createtxg: props.Get("createtxg"),
guid: props.Get("guid"),
creation: props.Get("creation"),
userrefs: props.Get("userrefs"),
})
}
zfs/zfs.go
+72 -93
View File
@@ -221,9 +221,12 @@ type ZFSListResult struct {
// If no error occurs, it is just closed.
// If the operation is cancelled via context, the channel is just closed.
//
// If notExistHint is not nil and zfs exits with an error,
// the stderr is attempted to be interpreted as a *DatasetDoesNotExist error.
//
// However, if callers do not drain `out` or cancel via `ctx`, the process will leak either running because
// IO is pending or as a zombie.
func ZFSListChan(ctx context.Context, out chan ZFSListResult, properties []string, zfsArgs ...string) {
func ZFSListChan(ctx context.Context, out chan ZFSListResult, properties []string, notExistHint *DatasetPath, zfsArgs ...string) {
defer close(out)
args := make([]string, 0, 4+len(zfsArgs))
@@ -272,11 +275,19 @@ func ZFSListChan(ctx context.Context, out chan ZFSListResult, properties []strin
}
}
if err := cmd.Wait(); err != nil {
if err, ok := err.(*exec.ExitError); ok {
sendResult(nil, &ZFSError{
Stderr: stderrBuf.Bytes(),
WaitErr: err,
})
if _, ok := err.(*exec.ExitError); ok {
var enotexist *DatasetDoesNotExist
if notExistHint != nil {
enotexist = tryDatasetDoesNotExist(notExistHint.ToString(), stderrBuf.Bytes())
}
if enotexist != nil {
sendResult(nil, enotexist)
} else {
sendResult(nil, &ZFSError{
Stderr: stderrBuf.Bytes(),
WaitErr: err,
})
}
} else {
sendResult(nil, &ZFSError{WaitErr: err})
}
@@ -308,7 +319,7 @@ func absVersion(fs string, v *ZFSSendArgVersion) (full string, err error) {
// a must already be validated
//
// SECURITY SENSITIVE because Raw must be handled correctly
func (a ZFSSendArgs) buildCommonSendArgs() ([]string, error) {
func (a ZFSSendArgsUnvalidated) buildCommonSendArgs() ([]string, error) {
args := make([]string, 0, 3)
// ResumeToken takes precedence, we assume that it has been validated to reflect
@@ -519,6 +530,9 @@ type ZFSSendArgVersion struct {
GUID uint64
}
func (v ZFSSendArgVersion) GetGuid() uint64 { return v.GUID }
func (v ZFSSendArgVersion) ToSendArgVersion() ZFSSendArgVersion { return v }
func (v ZFSSendArgVersion) ValidateInMemory(fs string) error {
if fs == "" {
panic(fs)
@@ -552,26 +566,26 @@ func (v ZFSSendArgVersion) mustValidateInMemory(fs string) {
}
// fs must be not empty
func (a ZFSSendArgVersion) ValidateExistsAndGetCheckedProps(ctx context.Context, fs string) (ZFSPropCreateTxgAndGuidProps, error) {
func (a ZFSSendArgVersion) ValidateExistsAndGetVersion(ctx context.Context, fs string) (v FilesystemVersion, _ error) {
if err := a.ValidateInMemory(fs); err != nil {
return ZFSPropCreateTxgAndGuidProps{}, nil
return v, nil
}
realProps, err := ZFSGetCreateTXGAndGuid(ctx, a.FullPath(fs))
realVersion, err := ZFSGetFilesystemVersion(ctx, a.FullPath(fs))
if err != nil {
return ZFSPropCreateTxgAndGuidProps{}, err
return v, err
}
if realProps.Guid != a.GUID {
return ZFSPropCreateTxgAndGuidProps{}, fmt.Errorf("`GUID` field does not match real dataset's GUID: %q != %q", realProps.Guid, a.GUID)
if realVersion.Guid != a.GUID {
return v, fmt.Errorf("`GUID` field does not match real dataset's GUID: %q != %q", realVersion.Guid, a.GUID)
}
return realProps, nil
return realVersion, nil
}
func (a ZFSSendArgVersion) ValidateExists(ctx context.Context, fs string) error {
_, err := a.ValidateExistsAndGetCheckedProps(ctx, fs)
_, err := a.ValidateExistsAndGetVersion(ctx, fs)
return err
}
@@ -613,7 +627,7 @@ func (n *NilBool) String() string {
}
// When updating this struct, check Validate and ValidateCorrespondsToResumeToken (POTENTIALLY SECURITY SENSITIVE)
type ZFSSendArgs struct {
type ZFSSendArgsUnvalidated struct {
FS string
From, To *ZFSSendArgVersion // From may be nil
Encrypted *NilBool
@@ -622,6 +636,12 @@ type ZFSSendArgs struct {
ResumeToken string // if not nil, must match what is specified in From, To (covered by ValidateCorrespondsToResumeToken)
}
type ZFSSendArgsValidated struct {
ZFSSendArgsUnvalidated
FromVersion *FilesystemVersion
ToVersion FilesystemVersion
}
type zfsSendArgsValidationContext struct {
encEnabled *NilBool
}
@@ -636,16 +656,16 @@ const (
)
type ZFSSendArgsValidationError struct {
Args ZFSSendArgs
Args ZFSSendArgsUnvalidated
What ZFSSendArgsValidationErrorCode
Msg error
}
func newValidationError(sendArgs ZFSSendArgs, what ZFSSendArgsValidationErrorCode, cause error) *ZFSSendArgsValidationError {
func newValidationError(sendArgs ZFSSendArgsUnvalidated, what ZFSSendArgsValidationErrorCode, cause error) *ZFSSendArgsValidationError {
return &ZFSSendArgsValidationError{sendArgs, what, cause}
}
func newGenericValidationError(sendArgs ZFSSendArgs, cause error) *ZFSSendArgsValidationError {
func newGenericValidationError(sendArgs ZFSSendArgsUnvalidated, cause error) *ZFSSendArgsValidationError {
return &ZFSSendArgsValidationError{sendArgs, ZFSSendArgsGenericValidationError, cause}
}
@@ -657,49 +677,57 @@ func (e ZFSSendArgsValidationError) Error() string {
// - Make sure that if ResumeToken != "", it reflects the same operation as the other parameters would.
//
// This function is not pure because GUIDs are checked against the local host's datasets.
func (a ZFSSendArgs) Validate(ctx context.Context) error {
func (a ZFSSendArgsUnvalidated) Validate(ctx context.Context) (v ZFSSendArgsValidated, _ error) {
if dp, err := NewDatasetPath(a.FS); err != nil || dp.Length() == 0 {
return newGenericValidationError(a, fmt.Errorf("`FS` must be a valid non-zero dataset path"))
return v, newGenericValidationError(a, fmt.Errorf("`FS` must be a valid non-zero dataset path"))
}
if a.To == nil {
return newGenericValidationError(a, fmt.Errorf("`To` must not be nil"))
return v, newGenericValidationError(a, fmt.Errorf("`To` must not be nil"))
}
if err := a.To.ValidateExists(ctx, a.FS); err != nil {
return newGenericValidationError(a, errors.Wrap(err, "`To` invalid"))
toVersion, err := a.To.ValidateExistsAndGetVersion(ctx, a.FS)
if err != nil {
return v, newGenericValidationError(a, errors.Wrap(err, "`To` invalid"))
}
var fromVersion *FilesystemVersion
if a.From != nil {
if err := a.From.ValidateExists(ctx, a.FS); err != nil {
return newGenericValidationError(a, errors.Wrap(err, "`From` invalid"))
fromV, err := a.From.ValidateExistsAndGetVersion(ctx, a.FS)
if err != nil {
return v, newGenericValidationError(a, errors.Wrap(err, "`From` invalid"))
}
fromVersion = &fromV
// fallthrough
}
if err := a.Encrypted.Validate(); err != nil {
return newGenericValidationError(a, errors.Wrap(err, "`Raw` invalid"))
return v, newGenericValidationError(a, errors.Wrap(err, "`Raw` invalid"))
}
valCtx := &zfsSendArgsValidationContext{}
fsEncrypted, err := ZFSGetEncryptionEnabled(ctx, a.FS)
if err != nil {
return newValidationError(a, ZFSSendArgsFSEncryptionCheckFail,
return v, newValidationError(a, ZFSSendArgsFSEncryptionCheckFail,
errors.Wrapf(err, "cannot check whether filesystem %q is encrypted", a.FS))
}
valCtx.encEnabled = &NilBool{fsEncrypted}
if a.Encrypted.B && !fsEncrypted {
return newValidationError(a, ZFSSendArgsEncryptedSendRequestedButFSUnencrypted,
return v, newValidationError(a, ZFSSendArgsEncryptedSendRequestedButFSUnencrypted,
errors.Errorf("encrypted send requested, but filesystem %q is not encrypted", a.FS))
}
if a.ResumeToken != "" {
if err := a.validateCorrespondsToResumeToken(ctx, valCtx); err != nil {
return newValidationError(a, ZFSSendArgsResumeTokenMismatch, err)
return v, newValidationError(a, ZFSSendArgsResumeTokenMismatch, err)
}
}
return nil
return ZFSSendArgsValidated{
ZFSSendArgsUnvalidated: a,
FromVersion: fromVersion,
ToVersion: toVersion,
}, nil
}
type ZFSSendArgsResumeTokenMismatchError struct {
@@ -729,7 +757,7 @@ func (c ZFSSendArgsResumeTokenMismatchErrorCode) fmt(format string, args ...inte
// This is SECURITY SENSITIVE and requires exhaustive checking of both side's values
// An attacker requesting a Send with a crafted ResumeToken may encode different parameters in the resume token than expected:
// for example, they may specify another file system (e.g. the filesystem with secret data) or request unencrypted send instead of encrypted raw send.
func (a ZFSSendArgs) validateCorrespondsToResumeToken(ctx context.Context, valCtx *zfsSendArgsValidationContext) error {
func (a ZFSSendArgsUnvalidated) validateCorrespondsToResumeToken(ctx context.Context, valCtx *zfsSendArgsValidationContext) error {
if a.ResumeToken == "" {
return nil // nothing to do
@@ -802,7 +830,7 @@ var ErrEncryptedSendNotSupported = fmt.Errorf("raw sends which are required for
// (if from is "" a full ZFS send is done)
//
// Returns ErrEncryptedSendNotSupported if encrypted send is requested but not supported by CLI
func ZFSSend(ctx context.Context, sendArgs ZFSSendArgs) (*ReadCloserCopier, error) {
func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*ReadCloserCopier, error) {
args := make([]string, 0)
args = append(args, "send")
@@ -819,10 +847,6 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgs) (*ReadCloserCopier, erro
}
}
if err := sendArgs.Validate(ctx); err != nil {
return nil, err // do not wrap, part of API, tested by platformtest
}
sargs, err := sendArgs.buildCommonSendArgs()
if err != nil {
return nil, err
@@ -956,11 +980,7 @@ func (s *DrySendInfo) unmarshalInfoLine(l string) (regexMatched bool, err error)
// to may be "", in which case a full ZFS send is done
// May return BookmarkSizeEstimationNotSupported as err if from is a bookmark.
func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgs) (_ *DrySendInfo, err error) {
if err := sendArgs.Validate(ctx); err != nil {
return nil, errors.Wrap(err, "cannot validate send args")
}
func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendInfo, err error) {
if sendArgs.From != nil && strings.Contains(sendArgs.From.RelName, "#") {
/* TODO:
@@ -1064,21 +1084,15 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier
if opts.RollbackAndForceRecv {
// destroy all snapshots before `recv -F` because `recv -F`
// does not perform a rollback unless `send -R` was used (which we assume hasn't been the case)
var snaps []FilesystemVersion
{
vs, err := ZFSListFilesystemVersions(fsdp, nil)
if err != nil {
return fmt.Errorf("cannot list versions for rollback for forced receive: %s", err)
}
for _, v := range vs {
if v.Type == Snapshot {
snaps = append(snaps, v)
}
}
sort.Slice(snaps, func(i, j int) bool {
return snaps[i].CreateTXG < snaps[j].CreateTXG
})
snaps, err := ZFSListFilesystemVersions(fsdp, ListFilesystemVersionsOptions{
Types: Snapshots,
})
if err != nil {
return fmt.Errorf("cannot list versions for rollback for forced receive: %s", err)
}
sort.Slice(snaps, func(i, j int) bool {
return snaps[i].CreateTXG < snaps[j].CreateTXG
})
// bookmarks are rolled back automatically
if len(snaps) > 0 {
// use rollback to efficiently destroy all but the earliest snapshot
@@ -1356,7 +1370,7 @@ type DatasetDoesNotExist struct {
func (d *DatasetDoesNotExist) Error() string { return fmt.Sprintf("dataset %q does not exist", d.Path) }
func tryDatasetDoesNotExist(expectPath string, stderr []byte) error {
func tryDatasetDoesNotExist(expectPath string, stderr []byte) *DatasetDoesNotExist {
if sm := zfsGetDatasetDoesNotExistRegexp.FindSubmatch(stderr); sm != nil {
if string(sm[1]) == expectPath {
return &DatasetDoesNotExist{expectPath}
@@ -1450,41 +1464,6 @@ func zfsGet(ctx context.Context, path string, props []string, allowedSources zfs
return res, nil
}
type ZFSPropCreateTxgAndGuidProps struct {
CreateTXG, Guid uint64
}
func ZFSGetCreateTXGAndGuid(ctx context.Context, ds string) (ZFSPropCreateTxgAndGuidProps, error) {
props, err := zfsGetNumberProps(ctx, ds, []string{"createtxg", "guid"}, sourceAny)
if err != nil {
return ZFSPropCreateTxgAndGuidProps{}, err
}
return ZFSPropCreateTxgAndGuidProps{
CreateTXG: props["createtxg"],
Guid: props["guid"],
}, nil
}
// returns *DatasetDoesNotExist if the dataset does not exist
func zfsGetNumberProps(ctx context.Context, ds string, props []string, src zfsPropertySource) (map[string]uint64, error) {
sps, err := zfsGet(ctx, ds, props, sourceAny)
if err != nil {
if _, ok := err.(*DatasetDoesNotExist); ok {
return nil, err // pass through as is
}
return nil, errors.Wrap(err, "zfs: set replication cursor: get snapshot createtxg")
}
r := make(map[string]uint64, len(props))
for _, p := range props {
v, err := strconv.ParseUint(sps.Get(p), 10, 64)
if err != nil {
return nil, errors.Wrapf(err, "zfs get: parse number property %q", p)
}
r[p] = v
}
return r, nil
}
type DestroySnapshotsError struct {
RawLines []string
Filesystem string
@@ -1669,7 +1648,7 @@ func ZFSBookmark(ctx context.Context, fs string, v ZFSSendArgVersion, bookmark s
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "bookmark", snapname, bookmarkname)
stdio, err := cmd.CombinedOutput()
if err != nil {
if ddne := tryDatasetDoesNotExist(snapname, stdio); err != nil {
if ddne := tryDatasetDoesNotExist(snapname, stdio); ddne != nil {
return ddne
} else if zfsBookmarkExistsRegex.Match(stdio) {