zfs: generalize dry send information for normal sends and with resume token

This is in preparation for resumable send & recv, thus we just don't use
the ResumeToken field for the time being.

endpoint/endpoint.go

@@ -79,16 +79,17 @@ func (p *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 	}
 
 	if r.DryRun {
-		size, err := zfs.ZFSSendDry(r.Filesystem, r.From, r.To)
+		si, err := zfs.ZFSSendDry(r.Filesystem, r.From, r.To, "")
-		if err == zfs.BookmarkSizeEstimationNotSupported {
-			return &pdu.SendRes{ExpectedSize: 0}, nil, nil
-		}
 		if err != nil {
 			return nil, nil, err
 		}
-		return &pdu.SendRes{ExpectedSize: size}, nil, nil
+		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
+		}
+		return &pdu.SendRes{ExpectedSize: expSize}, nil, nil
 	} else {
-		stream, err := zfs.ZFSSend(r.Filesystem, r.From, r.To)
+		stream, err := zfs.ZFSSend(r.Filesystem, r.From, r.To, "")
 		if err != nil {
 			return nil, nil, err
 		}

zfs/zfs.go

@@ -286,8 +286,12 @@ func absVersion(fs, v string) (full string, err error) {
 	return fmt.Sprintf("%s%s", fs, v), nil
 }
 
-// from may be "", in which case a full ZFS send is done
+func buildCommonSendArgs(fs string, from, to string, token string) ([]string, error) {
-func ZFSSend(fs string, from, to string) (stream io.ReadCloser, err error) {
+	args := make([]string, 0, 3)
+	if token != "" {
+		args = append(args, "-t", token)
+		return args, nil
+	}
 
 	toV, err := absVersion(fs, to)
 	if err != nil {
@@ -302,77 +306,155 @@ func ZFSSend(fs string, from, to string) (stream io.ReadCloser, err error) {
 		}
 	}
 
-	args := make([]string, 0)
-	args = append(args, "send")
-
 	if fromV == "" { // Initial
 		args = append(args, toV)
 	} else {
 		args = append(args, "-i", fromV, toV)
 	}
+	return args, nil
+}
+
+// if token != "", then send -t token is used
+// otherwise send [-i from] to is used
+// (if from is "" a full ZFS send is done)
+func ZFSSend(fs string, from, to string, token string) (stream io.ReadCloser, err error) {
+
+	args := make([]string, 0)
+	args = append(args, "send")
+
+	sargs, err := buildCommonSendArgs(fs, from, to, token)
+	if err != nil {
+		return nil, err
+	}
+	args = append(args, sargs...)
 
 	stream, err = util.RunIOCommand(ZFS_BINARY, args...)
 
 	return
 }
 
-var BookmarkSizeEstimationNotSupported error = fmt.Errorf("size estimation is not supported for bookmarks")
+
+type DrySendType string
+
+const (
+	DrySendTypeFull        DrySendType = "full"
+	DrySendTypeIncremental DrySendType = "incremental"
+)
+
+func DrySendTypeFromString(s string) (DrySendType, error) {
+	switch s {
+	case string(DrySendTypeFull): return DrySendTypeFull, nil
+	case string(DrySendTypeIncremental): return DrySendTypeIncremental, nil
+	default:
+		return "", fmt.Errorf("unknown dry send type %q", s)
+	}
+}
+
+type DrySendInfo struct {
+	Type DrySendType
+	Filesystem string // parsed from To field
+	From, To string // direct copy from ZFS output
+	SizeEstimate int64 // -1 if size estimate is not possible
+}
+
+var sendDryRunInfoLineRegex = regexp.MustCompile(`^(full|incremental)(\t(\S+))?\t(\S+)\t([0-9]+)$`)
+
+// see test cases for example output
+func (s *DrySendInfo) unmarshalZFSOutput(output []byte) (err error) {
+	lines := strings.Split(string(output), "\n")
+	for _, l := range lines {
+		regexMatched, err := s.unmarshalInfoLine(l)
+		if err != nil {
+			return fmt.Errorf("line %q: %s", l, err)
+		}
+		if !regexMatched {
+			continue
+		}
+		return nil
+	}
+	return fmt.Errorf("no match for info line (regex %s)", sendDryRunInfoLineRegex)
+}
+
+
+// unmarshal info line, looks like this:
+//   full	zroot/test/a@1	5389768
+//   incremental	zroot/test/a@1	zroot/test/a@2	5383936
+// => see test cases
+func (s *DrySendInfo) unmarshalInfoLine(l string) (regexMatched bool, err error) {
+
+	m := sendDryRunInfoLineRegex.FindStringSubmatch(l)
+	if m == nil {
+		return false, nil
+	}
+	s.Type, err = DrySendTypeFromString(m[1])
+	if err != nil {
+		return true, err
+	}
+
+	s.From = m[3]
+	s.To = m[4]
+	toFS, _, _ , err := DecomposeVersionString(s.To)
+	if err != nil {
+		return true, fmt.Errorf("'to' is not a valid filesystem version: %s", err)
+	}
+	s.Filesystem = toFS
+
+	s.SizeEstimate, err = strconv.ParseInt(m[5], 10, 64)
+	if err != nil {
+		return true, fmt.Errorf("cannot not parse size: %s", err)
+	}
+
+	return true, nil
+}
 
 // from may be "", in which case a full ZFS send is done
 // May return BookmarkSizeEstimationNotSupported as err if from is a bookmark.
-func ZFSSendDry(fs string, from, to string) (size int64, err error) {
+func ZFSSendDry(fs string, from, to string, token string) (_ *DrySendInfo, err error) {
 
-	toV, err := absVersion(fs, to)
+	if strings.Contains(from, "#") {
-	if err != nil {
-		return 0, err
-	}
-
-	fromV := ""
-	if from != "" {
-		fromV, err = absVersion(fs, from)
-		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
+		 fromAbs, err := absVersion(fs, from)
+		 if err != nil {
+		 	return nil, fmt.Errorf("error building abs version for 'from': %s", err)
+		 }
+		 toAbs, err := absVersion(fs, to)
+		 if err != nil {
+		 	return nil, fmt.Errorf("error building abs version for 'to': %s", err)
+		 }
+		 return &DrySendInfo{
+			Type: DrySendTypeIncremental,
+			Filesystem: fs,
+			From: fromAbs,
+			To: toAbs,
+			SizeEstimate: -1}, nil
 	}
 
 	args := make([]string, 0)
 	args = append(args, "send", "-n", "-v", "-P")
-
+	sargs, err := buildCommonSendArgs(fs, from, to, token)
-	if fromV == "" {
+	if err != nil {
-		args = append(args, toV)
+		return nil, err
-	} else {
-		args = append(args, "-i", fromV, toV)
 	}
+	args = append(args, sargs...)
 
 	cmd := exec.Command(ZFS_BINARY, args...)
 	output, err := cmd.CombinedOutput()
 	if err != nil {
-		return 0, err
+		return nil, err
 	}
-	o := string(output)
+	var si DrySendInfo
-	lines := strings.Split(o, "\n")
+	if err := si.unmarshalZFSOutput(output); err != nil {
-	if len(lines) < 2 {
+		return nil, fmt.Errorf("could not parse zfs send -n output: %s", err)
-		return 0, errors.New("zfs send -n did not return the expected number of lines")
 	}
-	fields := strings.Fields(lines[1])
+	return &si, nil
-	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 {
@@ -415,6 +497,32 @@ func ZFSRecv(fs string, stream io.Reader, additionalArgs ...string) (err error)
 	return nil
 }
 
+type ClearResumeTokenError struct {
+	ZFSOutput []byte
+	CmdError error
+}
+
+func (e ClearResumeTokenError) Error() string {
+	return fmt.Sprintf("could not clear resume token: %q", string(e.ZFSOutput))
+}
+
+// always returns *ClearResumeTokenError
+func ZFSRecvClearResumeToken(fs string) (err error) {
+	if err := validateZFSFilesystem(fs); err != nil {
+		return err
+	}
+
+	cmd := exec.Command(ZFS_BINARY, "recv", "-A", fs)
+	o, err := cmd.CombinedOutput()
+	if err != nil {
+		if bytes.Contains(o, []byte("does not have any resumable receive state to abort")) {
+			return nil
+		}
+		return &ClearResumeTokenError{o, err}
+	}
+	return nil
+}
+
 func ZFSRecvWriter(fs *DatasetPath, additionalArgs ...string) (io.WriteCloser, error) {
 
 	args := make([]string, 0)

zfs/zfs_test.go

@@ -68,3 +68,134 @@ func TestZFSPropertySource(t *testing.T) {
 	}
 
 }
+
+func TestDrySendInfo(t *testing.T) {
+
+	// # full send
+	// $ zfs send -Pnv -t 1-9baebea70-b8-789c636064000310a500c4ec50360710e72765a52697303030419460caa7a515a79680647ce0f26c48f2499525a9c5405ac3c90fabfe92fcf4d2cc140686b30972c7850efd0cd24092e704cbe725e6a632305415e5e797e803cd2ad14f743084b805001b201795
+	fullSend := `
+resume token contents:
+nvlist version: 0
+	object = 0x2
+	offset = 0x4c0000
+	bytes = 0x4e4228
+	toguid = 0x52f9c212c71e60cd
+	toname = zroot/test/a@1
+full	zroot/test/a@1	5389768
+`
+
+	// # incremental send with token
+	// $ zfs send -nvP -t 1-ef01e717e-e0-789c636064000310a501c49c50360710a715e5e7a69766a63040c1d904b9e342877e062900d9ec48eaf293b252934b181898a0ea30e4d3d28a534b40323e70793624f9a4ca92d46220fdc1ce0fabfe927c882bc46c8a0a9f71ad3baf8124cf0996cf4bcc4d6560a82acacf2fd1079a55a29fe86004710b00d8ae1f93
+	incSend := `
+resume token contents:
+nvlist version: 0
+	fromguid = 0x52f9c212c71e60cd
+	object = 0x2
+	offset = 0x4c0000
+	bytes = 0x4e3ef0
+	toguid = 0xcfae0ae671723c16
+	toname = zroot/test/a@2
+incremental	zroot/test/a@1	zroot/test/a@2	5383936
+`
+
+	// # incremental send with token + bookmarmk
+	// $ zfs send -nvP -t 1-ef01e717e-e0-789c636064000310a501c49c50360710a715e5e7a69766a63040c1d904b9e342877e062900d9ec48eaf293b252934b181898a0ea30e4d3d28a534b40323e70793624f9a4ca92d46220fdc1ce0fabfe927c882bc46c8a0a9f71ad3baf8124cf0996cf4bcc4d6560a82acacf2fd1079a55a29fe86004710b00d8ae1f93
+	incSendBookmark := `
+resume token contents:
+nvlist version: 0
+	fromguid = 0x52f9c212c71e60cd
+	object = 0x2
+	offset = 0x4c0000
+	bytes = 0x4e3ef0
+	toguid = 0xcfae0ae671723c16
+	toname = zroot/test/a@2
+incremental	zroot/test/a#1	zroot/test/a@2	5383312
+
+`
+
+	// incremental send without token
+	// $ sudo zfs send -nvP -i @1 zroot/test/a@2
+	incNoToken := `
+incremental	1	zroot/test/a@2	10511856
+size	10511856
+`
+	// full send without token
+	// $ sudo zfs send -nvP  zroot/test/a@3
+	fullNoToken := `
+full	zroot/test/a@3	10518512
+size	10518512
+`
+
+	type tc struct {
+		name string
+		in string
+		exp *DrySendInfo
+		expErr bool
+	}
+
+	tcs := []tc{
+		{
+			name: "fullSend", in: fullSend,
+			exp: &DrySendInfo{
+				Type: DrySendTypeFull,
+				From: "",
+				To: "zroot/test/a@1",
+				SizeEstimate: 5389768,
+			},
+		},
+		{
+			name: "incSend", in: incSend,
+			exp: &DrySendInfo{
+				Type:         DrySendTypeIncremental,
+				From:         "zroot/test/a@1",
+				To:           "zroot/test/a@2",
+				SizeEstimate: 5383936,
+			},
+		},
+		{
+			name: "incSendBookmark", in: incSendBookmark,
+			exp: &DrySendInfo{
+				Type:         DrySendTypeIncremental,
+				From:         "zroot/test/a#1",
+				To:           "zroot/test/a@2",
+				SizeEstimate: 5383312,
+			},
+		},
+		{
+			name: "incNoToken", in: incNoToken,
+			//exp: &DrySendInfo{
+			//	Type:         DrySendTypeIncremental,
+			//	From:         "1", // yes, this is actually correct on ZoL 0.7.11
+			//	To:           "zroot/test/a@2",
+			//	SizeEstimate: 10511856,
+			//},
+			expErr: true,
+		},
+		{
+			name: "fullNoToken", in: fullNoToken,
+			exp: &DrySendInfo{
+				Type: DrySendTypeFull,
+				From: "",
+				To: "zroot/test/a@3",
+				SizeEstimate: 10518512,
+			},
+		},
+	}
+
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+
+			in := tc.in[1:] // strip first newline
+			var si DrySendInfo
+			err := si.unmarshalZFSOutput([]byte(in))
+			if tc.expErr {
+				assert.Error(t, err)
+			}
+			t.Logf("%#v", &si)
+			if tc.exp != nil {
+				assert.Equal(t, tc.exp, &si)
+			}
+
+		})
+	}
+}