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

SQUASH THIS MERGE branch 'problame/zfs-command-logging-and-status' into problame/holds-release-and-hold-leak-fix-v2

Browse Source
This commit is contained in:
Christian Schwarz
2020-03-27 17:14:12 +01:00
parent 47d7bba192 4e13ea46de
commit 586a4ffac8
40 changed files with 633 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zfs/zfs.go
+80 -116
View File
@@ -21,6 +21,7 @@ import (
"github.com/zrepl/zrepl/util/circlog"
"github.com/zrepl/zrepl/util/envconst"
"github.com/zrepl/zrepl/zfs/zfscmd"
)
var (
@@ -164,7 +165,7 @@ func (e *ZFSError) Error() string {
var ZFS_BINARY string = "zfs"
func ZFSList(properties []string, zfsArgs ...string) (res [][]string, err error) {
func ZFSList(ctx context.Context, properties []string, zfsArgs ...string) (res [][]string, err error) {
args := make([]string, 0, 4+len(zfsArgs))
args = append(args,
@@ -172,13 +173,9 @@ func ZFSList(properties []string, zfsArgs ...string) (res [][]string, err error)
"-o", strings.Join(properties, ","))
args = append(args, zfsArgs...)
cmd := exec.Command(ZFS_BINARY, args...)
var stdout io.Reader
stderr := bytes.NewBuffer(make([]byte, 0, 1024))
cmd.Stderr = stderr
if stdout, err = cmd.StdoutPipe(); err != nil {
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
stdout, stderrBuf, err := cmd.StdoutPipeWithErrorBuf()
if err != nil {
return
}
@@ -205,7 +202,7 @@ func ZFSList(properties []string, zfsArgs ...string) (res [][]string, err error)
if waitErr := cmd.Wait(); waitErr != nil {
err := &ZFSError{
Stderr: stderr.Bytes(),
Stderr: stderrBuf.Bytes(),
WaitErr: waitErr,
}
return nil, err
@@ -244,15 +241,12 @@ func ZFSListChan(ctx context.Context, out chan ZFSListResult, properties []strin
}
}
cmd := exec.CommandContext(ctx, ZFS_BINARY, args...)
stdout, err := cmd.StdoutPipe()
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
stdout, stderrBuf, err := cmd.StdoutPipeWithErrorBuf()
if err != nil {
sendResult(nil, err)
return
}
// TODO bounded buffer
stderr := bytes.NewBuffer(make([]byte, 0, 1024))
cmd.Stderr = stderr
if err = cmd.Start(); err != nil {
sendResult(nil, err)
return
@@ -280,7 +274,7 @@ 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: stderr.Bytes(),
Stderr: stderrBuf.Bytes(),
WaitErr: err,
})
} else {
@@ -419,7 +413,7 @@ func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
}
type sendStream struct {
cmd *exec.Cmd
cmd *zfscmd.Cmd
kill context.CancelFunc
closeMtx sync.Mutex
@@ -567,7 +561,7 @@ func (a ZFSSendArgVersion) ValidateExistsAndGetVersion(ctx context.Context, fs s
return v, nil
}
realVersion, err := ZFSGetFilesystemVersion(a.FullPath(fs))
realVersion, err := ZFSGetFilesystemVersion(ctx, a.FullPath(fs))
if err != nil {
return v, err
}
@@ -849,7 +843,6 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*ReadCloserCop
args = append(args, sargs...)
ctx, cancel := context.WithCancel(ctx)
cmd := exec.CommandContext(ctx, ZFS_BINARY, args...)
// setup stdout with an os.Pipe to control pipe buffer size
stdoutReader, stdoutWriter, err := pipeWithCapacityHint(ZFSSendPipeCapacityHint)
@@ -857,11 +850,14 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*ReadCloserCop
cancel()
return nil, err
}
cmd.Stdout = stdoutWriter
stderrBuf := circlog.MustNewCircularLog(zfsSendStderrCaptureMaxSize)
cmd.Stderr = stderrBuf
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
cmd.SetStdio(zfscmd.Stdio{
Stdin: nil,
Stdout: stdoutWriter,
Stderr: stderrBuf,
})
if err := cmd.Start(); err != nil {
cancel()
@@ -869,6 +865,7 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*ReadCloserCop
stdoutReader.Close()
return nil, errors.Wrap(err, "cannot start zfs send command")
}
// close our writing-end of the pipe so that we don't wait for ourselves when reading from the reading end
stdoutWriter.Close()
stream := &sendStream{
@@ -1005,7 +1002,7 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
}
args = append(args, sargs...)
cmd := exec.Command(ZFS_BINARY, args...)
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
output, err := cmd.CombinedOutput()
if err != nil {
return nil, &ZFSError{output, err}
@@ -1099,11 +1096,11 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier
rollbackTarget := snaps[0]
rollbackTargetAbs := rollbackTarget.ToAbsPath(fsdp)
debug("recv: rollback to %q", rollbackTargetAbs)
if err := ZFSRollback(fsdp, rollbackTarget, "-r"); err != nil {
if err := ZFSRollback(ctx, fsdp, rollbackTarget, "-r"); err != nil {
return fmt.Errorf("cannot rollback %s to %s for forced receive: %s", fsdp.ToString(), rollbackTarget, err)
}
debug("recv: destroy %q", rollbackTargetAbs)
if err := ZFSDestroy(rollbackTargetAbs); err != nil {
if err := ZFSDestroy(ctx, rollbackTargetAbs); err != nil {
return fmt.Errorf("cannot destroy %s for forced receive: %s", rollbackTargetAbs, err)
}
}
@@ -1124,24 +1121,26 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier
ctx, cancelCmd := context.WithCancel(ctx)
defer cancelCmd()
cmd := exec.CommandContext(ctx, ZFS_BINARY, args...)
stderr := bytes.NewBuffer(make([]byte, 0, 1024))
cmd.Stderr = stderr
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
// TODO report bug upstream
// Setup an unused stdout buffer.
// Otherwise, ZoL v0.6.5.9-1 3.16.0-4-amd64 writes the following error to stderr and exits with code 1
// cannot receive new filesystem stream: invalid backup stream
stdout := bytes.NewBuffer(make([]byte, 0, 1024))
cmd.Stdout = stdout
stderr := bytes.NewBuffer(make([]byte, 0, 1024))
stdin, stdinWriter, err := pipeWithCapacityHint(ZFSRecvPipeCapacityHint)
if err != nil {
return err
}
cmd.Stdin = stdin
cmd.SetStdio(zfscmd.Stdio{
Stdin: stdin,
Stdout: stdout,
Stderr: stderr,
})
if err = cmd.Start(); err != nil {
stdinWriter.Close()
@@ -1151,7 +1150,7 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier
stdin.Close()
defer stdinWriter.Close()
pid := cmd.Process.Pid
pid := cmd.Process().Pid
debug := func(format string, args ...interface{}) {
debug("recv: pid=%v: %s", pid, fmt.Sprintf(format, args...))
}
@@ -1234,12 +1233,12 @@ func (e ClearResumeTokenError) Error() string {
}
// always returns *ClearResumeTokenError
func ZFSRecvClearResumeToken(fs string) (err error) {
func ZFSRecvClearResumeToken(ctx context.Context, fs string) (err error) {
if err := validateZFSFilesystem(fs); err != nil {
return err
}
cmd := exec.Command(ZFS_BINARY, "recv", "-A", fs)
cmd := zfscmd.CommandContext(ctx, 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")) {
@@ -1276,11 +1275,11 @@ func (p *ZFSProperties) appendArgs(args *[]string) (err error) {
return nil
}
func ZFSSet(fs *DatasetPath, props *ZFSProperties) (err error) {
return zfsSet(fs.ToString(), props)
func ZFSSet(ctx context.Context, fs *DatasetPath, props *ZFSProperties) (err error) {
return zfsSet(ctx, fs.ToString(), props)
}
func zfsSet(path string, props *ZFSProperties) (err error) {
func zfsSet(ctx context.Context, path string, props *ZFSProperties) (err error) {
args := make([]string, 0)
args = append(args, "set")
err = props.appendArgs(&args)
@@ -1289,18 +1288,11 @@ func zfsSet(path string, props *ZFSProperties) (err error) {
}
args = append(args, path)
cmd := exec.Command(ZFS_BINARY, args...)
stderr := bytes.NewBuffer(make([]byte, 0, 1024))
cmd.Stderr = stderr
if err = cmd.Start(); err != nil {
return err
}
if err = cmd.Wait(); err != nil {
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
stdio, err := cmd.CombinedOutput()
if err != nil {
err = &ZFSError{
Stderr: stderr.Bytes(),
Stderr: stdio,
WaitErr: err,
}
}
@@ -1308,12 +1300,12 @@ func zfsSet(path string, props *ZFSProperties) (err error) {
return
}
func ZFSGet(fs *DatasetPath, props []string) (*ZFSProperties, error) {
return zfsGet(fs.ToString(), props, sourceAny)
func ZFSGet(ctx context.Context, fs *DatasetPath, props []string) (*ZFSProperties, error) {
return zfsGet(ctx, fs.ToString(), props, sourceAny)
}
// The returned error includes requested filesystem and version as quoted strings in its error message
func ZFSGetGUID(fs string, version string) (g uint64, err error) {
func ZFSGetGUID(ctx context.Context, fs string, version string) (g uint64, err error) {
defer func(e *error) {
if *e != nil {
*e = fmt.Errorf("zfs get guid fs=%q version=%q: %s", fs, version, *e)
@@ -1329,7 +1321,7 @@ func ZFSGetGUID(fs string, version string) (g uint64, err error) {
return 0, errors.New("version does not start with @ or #")
}
path := fmt.Sprintf("%s%s", fs, version)
props, err := zfsGet(path, []string{"guid"}, sourceAny) // always local
props, err := zfsGet(ctx, path, []string{"guid"}, sourceAny) // always local
if err != nil {
return 0, err
}
@@ -1341,11 +1333,11 @@ type GetMountpointOutput struct {
Mountpoint string
}
func ZFSGetMountpoint(fs string) (*GetMountpointOutput, error) {
func ZFSGetMountpoint(ctx context.Context, fs string) (*GetMountpointOutput, error) {
if err := EntityNamecheck(fs, EntityTypeFilesystem); err != nil {
return nil, err
}
props, err := zfsGet(fs, []string{"mountpoint", "mounted"}, sourceAny)
props, err := zfsGet(ctx, fs, []string{"mountpoint", "mounted"}, sourceAny)
if err != nil {
return nil, err
}
@@ -1361,8 +1353,8 @@ func ZFSGetMountpoint(fs string) (*GetMountpointOutput, error) {
return o, nil
}
func ZFSGetRawAnySource(path string, props []string) (*ZFSProperties, error) {
return zfsGet(path, props, sourceAny)
func ZFSGetRawAnySource(ctx context.Context, path string, props []string) (*ZFSProperties, error) {
return zfsGet(ctx, path, props, sourceAny)
}
var zfsGetDatasetDoesNotExistRegexp = regexp.MustCompile(`^cannot open '([^)]+)': (dataset does not exist|no such pool or dataset)`) // verified in platformtest
@@ -1421,9 +1413,9 @@ func (s zfsPropertySource) zfsGetSourceFieldPrefixes() []string {
return prefixes
}
func zfsGet(path string, props []string, allowedSources zfsPropertySource) (*ZFSProperties, error) {
func zfsGet(ctx context.Context, 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...)
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
stdout, err := cmd.Output()
if err != nil {
if exitErr, ok := err.(*exec.ExitError); ok {
@@ -1531,7 +1523,7 @@ func tryParseDestroySnapshotsError(arg string, stderr []byte) *DestroySnapshotsE
}
}
func ZFSDestroy(arg string) (err error) {
func ZFSDestroy(ctx context.Context, arg string) (err error) {
var dstype, filesystem string
idx := strings.IndexAny(arg, "@#")
@@ -1550,28 +1542,21 @@ func ZFSDestroy(arg string) (err error) {
defer prometheus.NewTimer(prom.ZFSDestroyDuration.WithLabelValues(dstype, filesystem))
cmd := exec.Command(ZFS_BINARY, "destroy", arg)
var stderr bytes.Buffer
cmd.Stderr = &stderr
if err = cmd.Start(); err != nil {
return err
}
if err = cmd.Wait(); err != nil {
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "destroy", arg)
stdio, err := cmd.CombinedOutput()
if err != nil {
err = &ZFSError{
Stderr: stderr.Bytes(),
Stderr: stdio,
WaitErr: err,
}
if destroyOneOrMoreSnapshotsNoneExistedErrorRegexp.Match(stderr.Bytes()) {
if destroyOneOrMoreSnapshotsNoneExistedErrorRegexp.Match(stdio) {
err = &DatasetDoesNotExist{arg}
} else if match := destroyBookmarkDoesNotExist.FindStringSubmatch(stderr.String()); match != nil && match[1] == arg {
} else if match := destroyBookmarkDoesNotExist.FindStringSubmatch(string(stdio)); match != nil && match[1] == arg {
err = &DatasetDoesNotExist{arg}
} else if dsNotExistErr := tryDatasetDoesNotExist(filesystem, stderr.Bytes()); dsNotExistErr != nil {
} else if dsNotExistErr := tryDatasetDoesNotExist(filesystem, stdio); dsNotExistErr != nil {
err = dsNotExistErr
} else if dserr := tryParseDestroySnapshotsError(arg, stderr.Bytes()); dserr != nil {
} else if dserr := tryParseDestroySnapshotsError(arg, stdio); dserr != nil {
err = dserr
}
@@ -1581,15 +1566,15 @@ func ZFSDestroy(arg string) (err error) {
}
func ZFSDestroyIdempotent(path string) error {
err := ZFSDestroy(path)
func ZFSDestroyIdempotent(ctx context.Context, path string) error {
err := ZFSDestroy(ctx, path)
if _, ok := err.(*DatasetDoesNotExist); ok {
return nil
}
return err
}
func ZFSSnapshot(fs *DatasetPath, name string, recursive bool) (err error) {
func ZFSSnapshot(ctx context.Context, fs *DatasetPath, name string, recursive bool) (err error) {
promTimer := prometheus.NewTimer(prom.ZFSSnapshotDuration.WithLabelValues(fs.ToString()))
defer promTimer.ObserveDuration()
@@ -1598,18 +1583,12 @@ func ZFSSnapshot(fs *DatasetPath, name string, recursive bool) (err error) {
if err := EntityNamecheck(snapname, EntityTypeSnapshot); err != nil {
return errors.Wrap(err, "zfs snapshot")
}
cmd := exec.Command(ZFS_BINARY, "snapshot", snapname)
stderr := bytes.NewBuffer(make([]byte, 0, 1024))
cmd.Stderr = stderr
if err = cmd.Start(); err != nil {
return err
}
if err = cmd.Wait(); err != nil {
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "snapshot", snapname)
stdio, err := cmd.CombinedOutput()
if err != nil {
err = &ZFSError{
Stderr: stderr.Bytes(),
Stderr: stdio,
WaitErr: err,
}
}
@@ -1641,7 +1620,7 @@ var ErrBookmarkCloningNotSupported = fmt.Errorf("bookmark cloning feature is not
//
// does not destroy an existing bookmark, returns
//
func ZFSBookmark(fs string, v ZFSSendArgVersion, bookmark string) (err error) {
func ZFSBookmark(ctx context.Context, fs string, v ZFSSendArgVersion, bookmark string) (err error) {
promTimer := prometheus.NewTimer(prom.ZFSBookmarkDuration.WithLabelValues(fs))
defer promTimer.ObserveDuration()
@@ -1661,23 +1640,15 @@ func ZFSBookmark(fs string, v ZFSSendArgVersion, bookmark string) (err error) {
debug("bookmark: %q %q", snapname, bookmarkname)
cmd := exec.Command(ZFS_BINARY, "bookmark", snapname, bookmarkname)
stderr := bytes.NewBuffer(make([]byte, 0, 1024))
cmd.Stderr = stderr
if err = cmd.Start(); err != nil {
return err
}
if err = cmd.Wait(); err != nil {
if ddne := tryDatasetDoesNotExist(snapname, stderr.Bytes()); err != nil {
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "bookmark", snapname, bookmarkname)
stdio, err := cmd.CombinedOutput()
if err != nil {
if ddne := tryDatasetDoesNotExist(snapname, stdio); err != nil {
return ddne
} else if zfsBookmarkExistsRegex.Match(stderr.Bytes()) {
} else if zfsBookmarkExistsRegex.Match(stdio) {
// check if this was idempotent
bookGuid, err := ZFSGetGUID(fs, "#"+bookmark)
bookGuid, err := ZFSGetGUID(ctx, fs, "#"+bookmark)
if err != nil {
return errors.Wrap(err, "bookmark idempotency check") // guid error expressive enough
}
@@ -1688,13 +1659,13 @@ func ZFSBookmark(fs string, v ZFSSendArgVersion, bookmark string) (err error) {
}
return &BookmarkExists{
fs: fs, bookmarkOrigin: v, bookmark: bookmark,
zfsMsg: stderr.String(),
zfsMsg: string(stdio),
bookGuid: bookGuid,
}
} else {
return &ZFSError{
Stderr: stderr.Bytes(),
Stderr: stdio,
WaitErr: err,
}
}
@@ -1705,7 +1676,7 @@ func ZFSBookmark(fs string, v ZFSSendArgVersion, bookmark string) (err error) {
}
func ZFSRollback(fs *DatasetPath, snapshot FilesystemVersion, rollbackArgs ...string) (err error) {
func ZFSRollback(ctx context.Context, fs *DatasetPath, snapshot FilesystemVersion, rollbackArgs ...string) (err error) {
snapabs := snapshot.ToAbsPath(fs)
if snapshot.Type != Snapshot {
@@ -1716,18 +1687,11 @@ func ZFSRollback(fs *DatasetPath, snapshot FilesystemVersion, rollbackArgs ...st
args = append(args, rollbackArgs...)
args = append(args, snapabs)
cmd := exec.Command(ZFS_BINARY, args...)
stderr := bytes.NewBuffer(make([]byte, 0, 1024))
cmd.Stderr = stderr
if err = cmd.Start(); err != nil {
return err
}
if err = cmd.Wait(); err != nil {
cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
stdio, err := cmd.CombinedOutput()
if err != nil {
err = &ZFSError{
Stderr: stderr.Bytes(),
Stderr: stdio,
WaitErr: err,
}
}