Merge branch 'replication_rewrite' (in fact it's a 90% rewrite)

zfs/diff.go

@@ -5,6 +5,7 @@ import (
 	"crypto/sha512"
 	"encoding/hex"
 	"fmt"
+	"io"
 	"os/exec"
 	"sort"
 )
@@ -245,6 +246,19 @@ func IsPlaceholder(p *DatasetPath, placeholderPropertyValue string) (isPlacehold
 	return
 }
 
+// for nonexistent FS, isPlaceholder == false && err == nil
+func ZFSIsPlaceholderFilesystem(p *DatasetPath) (isPlaceholder bool, err error) {
+	props, err := zfsGet(p.ToString(), []string{ZREPL_PLACEHOLDER_PROPERTY_NAME}, sourceLocal)
+	if err == io.ErrUnexpectedEOF {
+		// interpret this as an early exit of the zfs binary due to the fs not existing
+		return false, nil
+	} else if err != nil {
+		return false, err
+	}
+	isPlaceholder, _ = IsPlaceholder(p, props.Get(ZREPL_PLACEHOLDER_PROPERTY_NAME))
+	return
+}
+
 func ZFSCreatePlaceholderFilesystem(p *DatasetPath) (err error) {
 	v := PlaceholderPropertyValue(p)
 	cmd := exec.Command(ZFS_BINARY, "create",

zfs/mapping.go

@@ -10,26 +10,56 @@ type DatasetFilter interface {
 }
 
 func ZFSListMapping(filter DatasetFilter) (datasets []*DatasetPath, err error) {
+	res, err := ZFSListMappingProperties(filter, nil)
+	if err != nil {
+		return nil, err
+	}
+	datasets = make([]*DatasetPath, len(res))
+	for i, r := range res {
+		datasets[i] = r.Path
+	}
+	return datasets, nil
+}
+
+type ZFSListMappingPropertiesResult struct {
+	Path *DatasetPath
+	// Guaranteed to have the same length as properties in the originating call
+	Fields []string
+}
+
+// properties must not contain 'name'
+func ZFSListMappingProperties(filter DatasetFilter, properties []string) (datasets []ZFSListMappingPropertiesResult, err error) {
 
 	if filter == nil {
 		panic("filter must not be nil")
 	}
 
+	for _, p := range properties {
+		if p == "name" {
+			panic("properties must not contain 'name'")
+		}
+	}
+	newProps := make([]string, len(properties)+1)
+	newProps[0] = "name"
+	copy(newProps[1:], properties)
+	properties = newProps
+
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	rchan := make(chan ZFSListResult)
-	go ZFSListChan(ctx, rchan, []string{"name"}, "-r", "-t", "filesystem,volume")
 
-	datasets = make([]*DatasetPath, 0)
+	go ZFSListChan(ctx, rchan, properties, "-r", "-t", "filesystem,volume")
+
+	datasets = make([]ZFSListMappingPropertiesResult, 0)
 	for r := range rchan {
 
-		if r.err != nil {
-			err = r.err
+		if r.Err != nil {
+			err = r.Err
 			return
 		}
 
 		var path *DatasetPath
-		if path, err = NewDatasetPath(r.fields[0]); err != nil {
+		if path, err = NewDatasetPath(r.Fields[0]); err != nil {
 			return
 		}
 
@@ -38,7 +68,10 @@ func ZFSListMapping(filter DatasetFilter) (datasets []*DatasetPath, err error) {
 			return nil, fmt.Errorf("error calling filter: %s", filterErr)
 		}
 		if pass {
-			datasets = append(datasets, path)
+			datasets = append(datasets, ZFSListMappingPropertiesResult{
+				Path:   path,
+				Fields: r.Fields[1:],
+			})
 		}
 
 	}

zfs/prometheus.go

@@ -3,10 +3,10 @@ package zfs
 import "github.com/prometheus/client_golang/prometheus"
 
 var prom struct {
-	ZFSListFilesystemVersionDuration    *prometheus.HistogramVec
-	ZFSDestroyFilesystemVersionDuration *prometheus.HistogramVec
-	ZFSSnapshotDuration                 *prometheus.HistogramVec
-	ZFSBookmarkDuration                 *prometheus.HistogramVec
+	ZFSListFilesystemVersionDuration *prometheus.HistogramVec
+	ZFSSnapshotDuration              *prometheus.HistogramVec
+	ZFSBookmarkDuration              *prometheus.HistogramVec
+	ZFSDestroyDuration               *prometheus.HistogramVec
 }
 
 func init() {
@@ -16,12 +16,6 @@ func init() {
 		Name:      "list_filesystem_versions_duration",
 		Help:      "Seconds it took for listing the versions of a given filesystem",
 	}, []string{"filesystem"})
-	prom.ZFSDestroyFilesystemVersionDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
-		Namespace: "zrepl",
-		Subsystem: "zfs",
-		Name:      "destroy_filesystem_version_duration",
-		Help:      "Seconds it took to destroy a version of a given filesystem",
-	}, []string{"filesystem", "version_type"})
 	prom.ZFSSnapshotDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
 		Namespace: "zrepl",
 		Subsystem: "zfs",
@@ -34,20 +28,26 @@ func init() {
 		Name:      "bookmark_duration",
 		Help:      "Duration it took to bookmark a given snapshot",
 	}, []string{"filesystem"})
+	prom.ZFSDestroyDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "zrepl",
+		Subsystem: "zfs",
+		Name:      "destroy_duration",
+		Help:      "Duration it took to destroy a dataset",
+	}, []string{"dataset_type", "filesystem"})
 }
 
 func PrometheusRegister(registry prometheus.Registerer) error {
 	if err := registry.Register(prom.ZFSListFilesystemVersionDuration); err != nil {
 		return err
 	}
-	if err := registry.Register(prom.ZFSDestroyFilesystemVersionDuration); err != nil {
-		return err
-	}
 	if err := registry.Register(prom.ZFSBookmarkDuration); err != nil {
 		return err
 	}
 	if err := registry.Register(prom.ZFSSnapshotDuration); err != nil {
 		return err
 	}
+	if err := registry.Register(prom.ZFSDestroyDuration); err != nil {
+		return err
+	}
 	return nil
 }

zfs/replication_history.go

@@ -0,0 +1,58 @@
+package zfs
+
+import (
+	"fmt"
+	"github.com/pkg/errors"
+	"strconv"
+)
+
+const ReplicationCursorBookmarkName = "zrepl_replication_cursor"
+
+// may return nil for both values, indicating there is no cursor
+func ZFSGetReplicationCursor(fs *DatasetPath) (*FilesystemVersion, error) {
+	versions, err := ZFSListFilesystemVersions(fs, nil)
+	if err != nil {
+		return nil, err
+	}
+	for _, v := range versions {
+		if v.Type == Bookmark && v.Name == ReplicationCursorBookmarkName {
+			return &v, nil
+		}
+	}
+	return nil, nil
+}
+
+func ZFSSetReplicationCursor(fs *DatasetPath, snapname string) (guid uint64, err error) {
+	snapPath := fmt.Sprintf("%s@%s", fs.ToString(), snapname)
+	propsSnap, err := zfsGet(snapPath, []string{"createtxg", "guid"}, sourceAny)
+	if err != nil {
+		return 0, err
+	}
+	snapGuid, err := strconv.ParseUint(propsSnap.Get("guid"), 10, 64)
+	bookmarkPath := fmt.Sprintf("%s#%s", fs.ToString(), ReplicationCursorBookmarkName)
+	propsBookmark, err := zfsGet(bookmarkPath, []string{"createtxg"}, sourceAny)
+	_, bookmarkNotExistErr := err.(*DatasetDoesNotExist)
+	if err != nil && !bookmarkNotExistErr {
+		return 0, err
+	}
+	if err == nil {
+		bookmarkTxg, err := strconv.ParseUint(propsBookmark.Get("createtxg"), 10, 64)
+		if err != nil {
+			return 0, errors.Wrap(err, "cannot parse bookmark createtxg")
+		}
+		snapTxg, err := strconv.ParseUint(propsSnap.Get("createtxg"), 10, 64)
+		if err != nil {
+			return 0, errors.Wrap(err, "cannot parse snapshot createtxg")
+		}
+		if snapTxg < bookmarkTxg {
+			return 0, errors.New("replication cursor can only be advanced, not set back")
+		}
+		if err := ZFSDestroy(bookmarkPath); err != nil { // FIXME make safer by using new temporary bookmark, then rename, possible with channel programs
+			return 0, err
+		}
+	}
+	if err := ZFSBookmark(fs, snapname, ReplicationCursorBookmarkName); err != nil {
+		return 0, err
+	}
+	return snapGuid, nil
+}

zfs/resume_token.go

@@ -0,0 +1,121 @@
+package zfs
+
+import (
+	"context"
+	"errors"
+	"os/exec"
+	"regexp"
+	"strconv"
+	"time"
+)
+
+type ResumeToken struct {
+	HasFromGUID, HasToGUID bool
+	FromGUID, ToGUID       uint64
+	// no support for other fields
+}
+
+var resumeTokenNVListRE = regexp.MustCompile(`\t(\S+) = (.*)`)
+var resumeTokenContentsRE = regexp.MustCompile(`resume token contents:\nnvlist version: 0`)
+var resumeTokenIsCorruptRE = regexp.MustCompile(`resume token is corrupt`)
+
+var ResumeTokenCorruptError = errors.New("resume token is corrupt")
+var ResumeTokenDecodingNotSupported = errors.New("zfs binary does not allow decoding resume token or zrepl cannot scrape zfs output")
+var ResumeTokenParsingError = errors.New("zrepl cannot parse resume token values")
+
+// Abuse 'zfs send' to decode the resume token
+//
+// FIXME: implement nvlist unpacking in Go and read through libzfs_sendrecv.c
+func ParseResumeToken(ctx context.Context, token string) (*ResumeToken, error) {
+
+	// Example resume tokens:
+	//
+	// From a non-incremental send
+	// 1-bf31b879a-b8-789c636064000310a500c4ec50360710e72765a5269740f80cd8e4d3d28a534b18e00024cf86249f5459925acc802a8facbf243fbd3433858161f5ddb9ab1ae7c7466a20c97382e5f312735319180af2f3730cf58166953824c2cc0200cde81651
+
+	// From an incremental send
+	// 1-c49b979a2-e0-789c636064000310a501c49c50360710a715e5e7a69766a63040c1eabb735735ce8f8d5400b2d991d4e52765a5269740f82080219f96569c5ac2000720793624f9a4ca92d46206547964fd25f91057f09e37babb88c9bf5503499e132c9f97989bcac050909f9f63a80f34abc421096616007c881d4c
+
+	// Resulting output of zfs send -nvt <token>
+	//
+	//resume token contents:
+	//nvlist version: 0
+	//	fromguid = 0x595d9f81aa9dddab
+	//	object = 0x1
+	//	offset = 0x0
+	//	bytes = 0x0
+	//	toguid = 0x854f02a2dd32cf0d
+	//	toname = pool1/test@b
+	//cannot resume send: 'pool1/test@b' used in the initial send no longer exists
+
+	ctx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
+	defer cancel()
+	cmd := exec.CommandContext(ctx, ZFS_BINARY, "send", "-nvt", string(token))
+	output, err := cmd.CombinedOutput()
+	if err != nil {
+		if exitErr, ok := err.(*exec.ExitError); ok {
+			if !exitErr.Exited() {
+				return nil, err
+			}
+			// we abuse zfs send for decoding, the exit error may be due to
+			// a) the token being from a third machine
+			// b) it no longer exists on the machine where
+		} else {
+			return nil, err
+		}
+	}
+
+	if !resumeTokenContentsRE.Match(output) {
+		if resumeTokenIsCorruptRE.Match(output) {
+			return nil, ResumeTokenCorruptError
+		}
+		return nil, ResumeTokenDecodingNotSupported
+	}
+
+	matches := resumeTokenNVListRE.FindAllStringSubmatch(string(output), -1)
+	if matches == nil {
+		return nil, ResumeTokenDecodingNotSupported
+	}
+
+	rt := &ResumeToken{}
+
+	for _, m := range matches {
+		attr, val := m[1], m[2]
+		switch attr {
+		case "fromguid":
+			rt.FromGUID, err = strconv.ParseUint(val, 0, 64)
+			if err != nil {
+				return nil, ResumeTokenParsingError
+			}
+			rt.HasFromGUID = true
+		case "toguid":
+			rt.ToGUID, err = strconv.ParseUint(val, 0, 64)
+			if err != nil {
+				return nil, ResumeTokenParsingError
+			}
+			rt.HasToGUID = true
+		}
+	}
+
+	if !rt.HasToGUID {
+		return nil, ResumeTokenDecodingNotSupported
+	}
+
+	return rt, nil
+
+}
+
+func ZFSGetReceiveResumeToken(fs *DatasetPath) (string, error) {
+	const prop_receive_resume_token = "receive_resume_token"
+	props, err := ZFSGet(fs, []string{prop_receive_resume_token})
+	if err != nil {
+		return "", err
+	}
+	res := props.m[prop_receive_resume_token]
+	if res == "-" {
+		return "", nil
+	} else {
+		return res, nil
+	}
+
+}

zfs/resume_token_test.go

@@ -0,0 +1,66 @@
+package zfs_test
+
+import (
+	"context"
+	"github.com/stretchr/testify/assert"
+	"github.com/zrepl/zrepl/zfs"
+	"testing"
+)
+
+type ResumeTokenTest struct {
+	Msg         string
+	Token       string
+	ExpectToken *zfs.ResumeToken
+	ExpectError error
+}
+
+func (rtt *ResumeTokenTest) Test(t *testing.T) {
+	t.Log(rtt.Msg)
+	res, err := zfs.ParseResumeToken(context.TODO(), rtt.Token)
+
+	if rtt.ExpectError != nil {
+		assert.EqualValues(t, rtt.ExpectError, err)
+		return
+	}
+	if rtt.ExpectToken != nil {
+		assert.Nil(t, err)
+		assert.EqualValues(t, rtt.ExpectToken, res)
+		return
+	}
+}
+
+func TestParseResumeToken(t *testing.T) {
+
+	t.SkipNow() // FIXME not compatible with docker
+
+	tbl := []ResumeTokenTest{
+		{
+			Msg:   "normal send (non-incremental)",
+			Token: `1-bf31b879a-b8-789c636064000310a500c4ec50360710e72765a5269740f80cd8e4d3d28a534b18e00024cf86249f5459925acc802a8facbf243fbd3433858161f5ddb9ab1ae7c7466a20c97382e5f312735319180af2f3730cf58166953824c2cc0200cde81651`,
+			ExpectToken: &zfs.ResumeToken{
+				HasToGUID: true,
+				ToGUID:    0x595d9f81aa9dddab,
+			},
+		},
+		{
+			Msg:   "normal send (incremental)",
+			Token: `1-c49b979a2-e0-789c636064000310a501c49c50360710a715e5e7a69766a63040c1eabb735735ce8f8d5400b2d991d4e52765a5269740f82080219f96569c5ac2000720793624f9a4ca92d46206547964fd25f91057f09e37babb88c9bf5503499e132c9f97989bcac050909f9f63a80f34abc421096616007c881d4c`,
+			ExpectToken: &zfs.ResumeToken{
+				HasToGUID:   true,
+				ToGUID:      0x854f02a2dd32cf0d,
+				HasFromGUID: true,
+				FromGUID:    0x595d9f81aa9dddab,
+			},
+		},
+		{
+			Msg:         "corrupted token",
+			Token:       `1-bf31b879a-b8-789c636064000310a500c4ec50360710e72765a5269740f80cd8e4d3d28a534b18e00024cf86249f5459925acc802a8facbf243fbd3433858161f5ddb9ab1ae7c7466a20c97382e5f312735319180af2f3730cf58166953824c2cc0200cd12345`,
+			ExpectError: zfs.ResumeTokenCorruptError,
+		},
+	}
+
+	for _, test := range tbl {
+		test.Test(t)
+	}
+
+}

zfs/versions.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/prometheus/client_golang/prometheus"
+	"io"
 	"strconv"
 	"strings"
 	"time"
@@ -33,6 +34,26 @@ func (t VersionType) String() string {
 	return string(t)
 }
 
+func DecomposeVersionString(v string) (fs string, versionType VersionType, name string, err error) {
+	if len(v) < 3 {
+		err = errors.New(fmt.Sprintf("snapshot or bookmark name implausibly short: %s", v))
+		return
+	}
+
+	snapSplit := strings.SplitN(v, "@", 2)
+	bookmarkSplit := strings.SplitN(v, "#", 2)
+	if len(snapSplit)*len(bookmarkSplit) != 2 {
+		err = errors.New(fmt.Sprintf("dataset cannot be snapshot and bookmark at the same time: %s", v))
+		return
+	}
+
+	if len(snapSplit) == 2 {
+		return snapSplit[0], Snapshot, snapSplit[1], nil
+	} else {
+		return bookmarkSplit[0], Bookmark, bookmarkSplit[1], nil
+	}
+}
+
 type FilesystemVersion struct {
 	Type VersionType
 
@@ -63,7 +84,7 @@ func (v FilesystemVersion) ToAbsPath(p *DatasetPath) string {
 }
 
 type FilesystemVersionFilter interface {
-	Filter(fsv FilesystemVersion) (accept bool, err error)
+	Filter(t VersionType, name string) (accept bool, err error)
 }
 
 func ZFSListFilesystemVersions(fs *DatasetPath, filter FilesystemVersionFilter) (res []FilesystemVersion, err error) {
@@ -82,31 +103,21 @@ func ZFSListFilesystemVersions(fs *DatasetPath, filter FilesystemVersionFilter)
 
 	res = make([]FilesystemVersion, 0)
 	for listResult := range listResults {
-		if listResult.err != nil {
-			return nil, listResult.err
+		if listResult.Err != nil {
+			if listResult.Err == io.ErrUnexpectedEOF {
+				// Since we specified the fs on the command line, we'll treat this like the filesystem doesn't exist
+				return []FilesystemVersion{}, nil
+			}
+			return nil, listResult.Err
 		}
 
-		line := listResult.fields
-
-		if len(line[0]) < 3 {
-			err = errors.New(fmt.Sprintf("snapshot or bookmark name implausibly short: %s", line[0]))
-			return
-		}
-
-		snapSplit := strings.SplitN(line[0], "@", 2)
-		bookmarkSplit := strings.SplitN(line[0], "#", 2)
-		if len(snapSplit)*len(bookmarkSplit) != 2 {
-			err = errors.New(fmt.Sprintf("dataset cannot be snapshot and bookmark at the same time: %s", line[0]))
-			return
-		}
+		line := listResult.Fields
 
 		var v FilesystemVersion
-		if len(snapSplit) == 2 {
-			v.Name = snapSplit[1]
-			v.Type = Snapshot
-		} else {
-			v.Name = bookmarkSplit[1]
-			v.Type = Bookmark
+
+		_, v.Type, v.Name, err = DecomposeVersionString(line[0])
+		if err != nil {
+			return nil, err
 		}
 
 		if v.Guid, err = strconv.ParseUint(line[1], 10, 64); err != nil {
@@ -129,7 +140,7 @@ func ZFSListFilesystemVersions(fs *DatasetPath, filter FilesystemVersionFilter)
 
 		accept := true
 		if filter != nil {
-			accept, err = filter.Filter(v)
+			accept, err = filter.Filter(v.Type, v.Name)
 			if err != nil {
 				err = fmt.Errorf("error executing filter: %s", err)
 				return nil, err
@@ -143,10 +154,7 @@ func ZFSListFilesystemVersions(fs *DatasetPath, filter FilesystemVersionFilter)
 	return
 }
 
-func ZFSDestroyFilesystemVersion(filesystem *DatasetPath, version FilesystemVersion) (err error) {
-
-	promTimer := prometheus.NewTimer(prom.ZFSDestroyFilesystemVersionDuration.WithLabelValues(filesystem.ToString(), version.Type.String()))
-	defer promTimer.ObserveDuration()
+func ZFSDestroyFilesystemVersion(filesystem *DatasetPath, version *FilesystemVersion) (err error) {
 
 	datasetPath := version.ToAbsPath(filesystem)
 

zfs/zfs.go

@@ -14,6 +14,8 @@ import (
 	"github.com/problame/go-rwccmd"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/zrepl/zrepl/util"
+	"regexp"
+	"strconv"
 )
 
 type DatasetPath struct {
@@ -140,7 +142,7 @@ type ZFSError struct {
 }
 
 func (e ZFSError) Error() string {
-	return fmt.Sprintf("zfs exited with error: %s", e.WaitErr.Error())
+	return fmt.Sprintf("zfs exited with error: %s\nstderr:\n%s", e.WaitErr.Error(), e.Stderr)
 }
 
 var ZFS_BINARY string = "zfs"
@@ -195,13 +197,13 @@ func ZFSList(properties []string, zfsArgs ...string) (res [][]string, err error)
 }
 
 type ZFSListResult struct {
-	fields []string
-	err    error
+	Fields []string
+	Err    error
 }
 
 // ZFSListChan executes `zfs list` and sends the results to the `out` channel.
 // The `out` channel is always closed by ZFSListChan:
-// If an error occurs, it is closed after sending a result with the err field set.
+// If an error occurs, it is closed after sending a result with the Err field set.
 // If no error occurs, it is just closed.
 // If the operation is cancelled via context, the channel is just closed.
 //
@@ -256,15 +258,56 @@ func ZFSListChan(ctx context.Context, out chan ZFSListResult, properties []strin
 	return
 }
 
-func ZFSSend(fs *DatasetPath, from, to *FilesystemVersion) (stream io.Reader, err error) {
+func validateRelativeZFSVersion(s string) error {
+	if len(s) <= 1 {
+		return errors.New("version must start with a delimiter char followed by at least one character")
+	}
+	if !(s[0] == '#' || s[0] == '@') {
+		return errors.New("version name starts with invalid delimiter char")
+	}
+	// FIXME whitespace check...
+	return nil
+}
+
+func validateZFSFilesystem(fs string) error {
+	if len(fs) < 1 {
+		return errors.New("filesystem path must have length > 0")
+	}
+	return nil
+}
+
+func absVersion(fs, v string) (full string, err error) {
+	if err := validateZFSFilesystem(fs); err != nil {
+		return "", err
+	}
+	if err := validateRelativeZFSVersion(v); err != nil {
+		return "", err
+	}
+	return fmt.Sprintf("%s%s", fs, v), nil
+}
+
+func ZFSSend(fs string, from, to string) (stream io.ReadCloser, err error) {
+
+	fromV, err := absVersion(fs, from)
+	if err != nil {
+		return nil, err
+	}
+
+	toV := ""
+	if to != "" {
+		toV, err = absVersion(fs, to)
+		if err != nil {
+			return nil, err
+		}
+	}
 
 	args := make([]string, 0)
 	args = append(args, "send")
 
-	if to == nil { // Initial
-		args = append(args, from.ToAbsPath(fs))
+	if toV == "" { // Initial
+		args = append(args, fromV)
 	} else {
-		args = append(args, "-i", from.ToAbsPath(fs), to.ToAbsPath(fs))
+		args = append(args, "-i", fromV, toV)
 	}
 
 	stream, err = util.RunIOCommand(ZFS_BINARY, args...)
@@ -272,14 +315,74 @@ func ZFSSend(fs *DatasetPath, from, to *FilesystemVersion) (stream io.Reader, er
 	return
 }
 
-func ZFSRecv(fs *DatasetPath, stream io.Reader, additionalArgs ...string) (err error) {
+var BookmarkSizeEstimationNotSupported error = fmt.Errorf("size estimation is not supported for bookmarks")
+
+// May return BookmarkSizeEstimationNotSupported as err if from is a bookmark.
+func ZFSSendDry(fs string, from, to string) (size int64, err error) {
+
+	fromV, err := absVersion(fs, from)
+	if err != nil {
+		return 0, err
+	}
+
+	toV := ""
+	if to != "" {
+		toV, err = absVersion(fs, to)
+		if err != nil {
+			return 0, err
+		}
+	}
+
+	if strings.Contains(fromV, "#") {
+		/* TODO:
+		 * 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
+		 * 	https://github.com/openzfs/openzfs/pull/484
+		 */
+		 return 0, BookmarkSizeEstimationNotSupported
+	}
+
+	args := make([]string, 0)
+	args = append(args, "send", "-n", "-v", "-P")
+
+	if toV == "" { // Initial
+		args = append(args, fromV)
+	} else {
+		args = append(args, "-i", fromV, toV)
+	}
+
+	cmd := exec.Command(ZFS_BINARY, args...)
+	output, err := cmd.CombinedOutput()
+	if err != nil {
+		return 0, err
+	}
+	o := string(output)
+	lines := strings.Split(o, "\n")
+	if len(lines) < 2 {
+		return 0, errors.New("zfs send -n did not return the expected number of lines")
+	}
+	fields := strings.Fields(lines[1])
+	if len(fields) != 2 {
+		return 0, errors.New("zfs send -n returned unexpexted output")
+	}
+
+	size, err = strconv.ParseInt(fields[1], 10, 64)
+	return size, err
+}
+
+func ZFSRecv(fs string, stream io.Reader, additionalArgs ...string) (err error) {
+
+	if err := validateZFSFilesystem(fs); err != nil {
+		return err
+	}
 
 	args := make([]string, 0)
 	args = append(args, "recv")
 	if len(args) > 0 {
 		args = append(args, additionalArgs...)
 	}
-	args = append(args, fs.ToString())
+	args = append(args, fs)
 
 	cmd := exec.Command(ZFS_BINARY, args...)
 
@@ -310,6 +413,27 @@ func ZFSRecv(fs *DatasetPath, stream io.Reader, additionalArgs ...string) (err e
 	return nil
 }
 
+func ZFSRecvWriter(fs *DatasetPath, additionalArgs ...string) (io.WriteCloser, error) {
+
+	args := make([]string, 0)
+	args = append(args, "recv")
+	if len(args) > 0 {
+		args = append(args, additionalArgs...)
+	}
+	args = append(args, fs.ToString())
+
+	cmd, err := util.NewIOCommand(ZFS_BINARY, args, 1024)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = cmd.Start(); err != nil {
+		return nil, err
+	}
+
+	return cmd.Stdin, nil
+}
+
 type ZFSProperties struct {
 	m map[string]string
 }
@@ -322,6 +446,10 @@ func (p *ZFSProperties) Set(key, val string) {
 	p.m[key] = val
 }
 
+func (p *ZFSProperties) Get(key string) string {
+	return p.m[key]
+}
+
 func (p *ZFSProperties) appendArgs(args *[]string) (err error) {
 	for prop, val := range p.m {
 		if strings.Contains(prop, "=") {
@@ -333,14 +461,17 @@ func (p *ZFSProperties) appendArgs(args *[]string) (err error) {
 }
 
 func ZFSSet(fs *DatasetPath, props *ZFSProperties) (err error) {
+	return zfsSet(fs.ToString(), props)
+}
 
+func zfsSet(path string, props *ZFSProperties) (err error) {
 	args := make([]string, 0)
 	args = append(args, "set")
 	err = props.appendArgs(&args)
 	if err != nil {
 		return err
 	}
-	args = append(args, fs.ToString())
+	args = append(args, path)
 
 	cmd := exec.Command(ZFS_BINARY, args...)
 
@@ -361,8 +492,101 @@ func ZFSSet(fs *DatasetPath, props *ZFSProperties) (err error) {
 	return
 }
 
+func ZFSGet(fs *DatasetPath, props []string) (*ZFSProperties, error) {
+	return zfsGet(fs.ToString(), props, sourceAny)
+}
+
+var zfsGetDatasetDoesNotExistRegexp = regexp.MustCompile(`^cannot open '(\S+)': (dataset does not exist|no such pool or dataset)`)
+
+type DatasetDoesNotExist struct {
+	Path string
+}
+
+func (d *DatasetDoesNotExist) Error() string { return fmt.Sprintf("dataset %q does not exist", d.Path) }
+
+type zfsPropertySource uint
+
+const (
+	sourceLocal zfsPropertySource = 1 << iota
+	sourceDefault
+	sourceInherited
+	sourceNone
+	sourceTemporary
+
+	sourceAny zfsPropertySource = ^zfsPropertySource(0)
+)
+
+func (s zfsPropertySource) zfsGetSourceFieldPrefixes() []string {
+	prefixes := make([]string, 0, 5)
+	if s&sourceLocal != 0 {prefixes = append(prefixes, "local")}
+	if s&sourceDefault != 0 {prefixes = append(prefixes, "default")}
+	if s&sourceInherited != 0 {prefixes = append(prefixes, "inherited")}
+	if s&sourceNone != 0 {prefixes = append(prefixes, "-")}
+	if s&sourceTemporary != 0 { prefixes = append(prefixes, "temporary")}
+	return prefixes
+}
+
+func zfsGet(path string, props []string, allowedSources zfsPropertySource) (*ZFSProperties, error) {
+	args := []string{"get", "-Hp", "-o", "property,value,source", strings.Join(props, ","), path}
+	cmd := exec.Command(ZFS_BINARY, args...)
+	stdout, err := cmd.Output()
+	if err != nil {
+		if exitErr, ok := err.(*exec.ExitError); ok {
+			if exitErr.Exited() {
+				// screen-scrape output
+				if sm := zfsGetDatasetDoesNotExistRegexp.FindSubmatch(exitErr.Stderr); sm != nil {
+					if string(sm[1]) == path {
+						return nil, &DatasetDoesNotExist{path}
+					}
+				}
+			}
+		}
+		return nil, err
+	}
+	o := string(stdout)
+	lines := strings.Split(o, "\n")
+	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]string, len(lines)),
+	}
+	allowedPrefixes := allowedSources.zfsGetSourceFieldPrefixes()
+	for _, line := range lines[:len(lines)-1] {
+		fields := strings.FieldsFunc(line, func(r rune) bool {
+			return r == '\t'
+		})
+		if len(fields) != 3 {
+			return nil, fmt.Errorf("zfs get did not return property,value,source tuples")
+		}
+		for _, p := range allowedPrefixes {
+			if strings.HasPrefix(fields[2],p) {
+				res.m[fields[0]] = fields[1]
+				break
+			}
+		}
+	}
+	return res, nil
+}
+
 func ZFSDestroy(dataset string) (err error) {
 
+	var dstype, filesystem string
+	idx := strings.IndexAny(dataset, "@#")
+	if idx == -1 {
+		dstype = "filesystem"
+		filesystem = dataset
+	} else {
+		switch dataset[idx] {
+		case '@': dstype = "snapshot"
+		case '#': dstype = "bookmark"
+		}
+		filesystem = dataset[:idx]
+	}
+
+	defer prometheus.NewTimer(prom.ZFSDestroyDuration.WithLabelValues(dstype, filesystem))
+
 	cmd := exec.Command(ZFS_BINARY, "destroy", dataset)
 
 	stderr := bytes.NewBuffer(make([]byte, 0, 1024))

zfs/zfs_test.go

@@ -6,6 +6,8 @@ import (
 )
 
 func TestZFSListHandlesProducesZFSErrorOnNonZeroExit(t *testing.T) {
+	t.SkipNow() // FIXME ZFS_BINARY does not work if tests run in parallel
+
 	var err error
 
 	ZFS_BINARY = "./test_helpers/zfs_failer.sh"
@@ -28,3 +30,41 @@ func TestDatasetPathTrimNPrefixComps(t *testing.T) {
 	p.TrimNPrefixComps((1))
 	assert.True(t, p.Empty(), "empty trimming shouldn't do harm")
 }
+
+func TestZFSPropertySource(t *testing.T) {
+
+	tcs := []struct{
+		in zfsPropertySource
+		exp []string
+	}{
+		{
+			in: sourceAny,
+			exp: []string{"local", "default", "inherited", "-", "temporary"},
+		},
+		{
+			in: sourceTemporary,
+			exp: []string{"temporary"},
+		},
+		{
+			in: sourceLocal|sourceInherited,
+			exp: []string{"local", "inherited"},
+		},
+	}
+
+	toSet := func(in []string) map[string]struct{} {
+		m := make(map[string]struct{}, len(in))
+		for _, s := range in {
+			m[s] = struct{}{}
+		}
+		return m
+	}
+
+	for _, tc := range tcs {
+
+		res := tc.in.zfsGetSourceFieldPrefixes()
+		resSet := toSet(res)
+		expSet := toSet(tc.exp)
+		assert.Equal(t, expSet, resSet)
+	}
+
+}