diff --git a/client/holds_list.go b/client/holds_list.go index 3f9b1a7..01b02a5 100644 --- a/client/holds_list.go +++ b/client/holds_list.go @@ -9,6 +9,7 @@ import ( "github.com/fatih/color" "github.com/pkg/errors" "github.com/spf13/pflag" + "github.com/zrepl/zrepl/cli" "github.com/zrepl/zrepl/endpoint" ) diff --git a/client/holds_release.go b/client/holds_release.go index b97c1f9..53c044f 100644 --- a/client/holds_release.go +++ b/client/holds_release.go @@ -9,6 +9,7 @@ import ( "github.com/fatih/color" "github.com/pkg/errors" "github.com/spf13/pflag" + "github.com/zrepl/zrepl/cli" "github.com/zrepl/zrepl/endpoint" ) diff --git a/client/migrate.go b/client/migrate.go index d899d0f..6e85a04 100644 --- a/client/migrate.go +++ b/client/migrate.go @@ -99,7 +99,7 @@ func doMigratePlaceholder0_1(sc *cli.Subcommand, args []string) error { } for _, fs := range wi.fss { fmt.Printf("\t%q ... ", fs.ToString()) - r, err := zfs.ZFSMigrateHashBasedPlaceholderToCurrent(fs, migratePlaceholder0_1Args.dryRun) + r, err := zfs.ZFSMigrateHashBasedPlaceholderToCurrent(ctx, fs, migratePlaceholder0_1Args.dryRun) if err != nil { fmt.Printf("error: %s\n", err) } else if !r.NeedsModification { @@ -264,7 +264,7 @@ func doMigrateReplicationCursorFS(ctx context.Context, v1CursorJobs []job.Job, f if migrateReplicationCursorArgs.dryRun { succ.Printf("DRY RUN\n") } else { - if err := zfs.ZFSDestroyFilesystemVersion(fs, oldCursor); err != nil { + if err := zfs.ZFSDestroyFilesystemVersion(ctx, fs, oldCursor); err != nil { return err } } diff --git a/client/status.go b/client/status.go index 6fef777..25ff014 100644 --- a/client/status.go +++ b/client/status.go @@ -81,7 +81,7 @@ type tui struct { indent int lock sync.Mutex //For report and error - report map[string]job.Status + report map[string]*job.Status err error jobFilter string @@ -219,7 +219,7 @@ func runStatus(s *cli.Subcommand, args []string) error { defer termbox.Close() update := func() { - m := make(map[string]job.Status) + var m daemon.Status err2 := jsonRequestResponse(httpc, daemon.ControlJobEndpointStatus, struct{}{}, @@ -228,7 +228,7 @@ func runStatus(s *cli.Subcommand, args []string) error { t.lock.Lock() t.err = err2 - t.report = m + t.report = m.Jobs t.lock.Unlock() t.draw() } diff --git a/client/testcmd.go b/client/testcmd.go index d430c6d..4120c2e 100644 --- a/client/testcmd.go +++ b/client/testcmd.go @@ -49,6 +49,7 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error { } conf := subcommand.Config() + ctx := context.Background() var confFilter config.FilesystemsFilter job, err := conf.Job(testFilterArgs.job) @@ -75,7 +76,7 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error { if testFilterArgs.input != "" { fsnames = []string{testFilterArgs.input} } else { - out, err := zfs.ZFSList([]string{"name"}) + out, err := zfs.ZFSList(ctx, []string{"name"}) if err != nil { return fmt.Errorf("could not list ZFS filesystems: %s", err) } @@ -139,10 +140,11 @@ var testPlaceholder = &cli.Subcommand{ func runTestPlaceholder(subcommand *cli.Subcommand, args []string) error { var checkDPs []*zfs.DatasetPath + ctx := context.Background() // all actions first if testPlaceholderArgs.all { - out, err := zfs.ZFSList([]string{"name"}) + out, err := zfs.ZFSList(ctx, []string{"name"}) if err != nil { return errors.Wrap(err, "could not list ZFS filesystems") } @@ -166,7 +168,7 @@ func runTestPlaceholder(subcommand *cli.Subcommand, args []string) error { fmt.Printf("IS_PLACEHOLDER\tDATASET\tzrepl:placeholder\n") for _, dp := range checkDPs { - ph, err := zfs.ZFSGetFilesystemPlaceholderState(dp) + ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, dp) if err != nil { return errors.Wrap(err, "cannot get placeholder state") } diff --git a/daemon/control.go b/daemon/control.go index f987ff9..7e84b6f 100644 --- a/daemon/control.go +++ b/daemon/control.go @@ -20,6 +20,7 @@ import ( "github.com/zrepl/zrepl/util/envconst" "github.com/zrepl/zrepl/version" "github.com/zrepl/zrepl/zfs" + "github.com/zrepl/zrepl/zfs/zfscmd" ) type controlJob struct { @@ -117,7 +118,9 @@ func (j *controlJob) Run(ctx context.Context) { mux.Handle(ControlJobEndpointStatus, // don't log requests to status endpoint, too spammy jsonResponder{log, func() (interface{}, error) { - s := j.jobs.status() + jobs := j.jobs.status() + globalZFS := zfscmd.GetReport() + s := Status{Jobs: jobs, Global: GlobalStatus{ZFSCmds: globalZFS}} return s, nil }}) diff --git a/daemon/daemon.go b/daemon/daemon.go index 265ddc4..dc2304f 100644 --- a/daemon/daemon.go +++ b/daemon/daemon.go @@ -20,6 +20,7 @@ import ( "github.com/zrepl/zrepl/daemon/logging" "github.com/zrepl/zrepl/logger" "github.com/zrepl/zrepl/version" + "github.com/zrepl/zrepl/zfs/zfscmd" ) func Run(conf *config.Config) error { @@ -81,6 +82,9 @@ func Run(conf *config.Config) error { jobs.start(ctx, job, true) } + // register global (=non job-local) metrics + zfscmd.RegisterMetrics(prometheus.DefaultRegisterer) + log.Info("starting daemon") // start regular jobs @@ -128,6 +132,15 @@ func (s *jobs) wait() <-chan struct{} { return ch } +type Status struct { + Jobs map[string]*job.Status + Global GlobalStatus +} + +type GlobalStatus struct { + ZFSCmds *zfscmd.Report +} + func (s *jobs) status() map[string]*job.Status { s.m.RLock() defer s.m.RUnlock() @@ -207,6 +220,7 @@ func (s *jobs) start(ctx context.Context, j job.Job, internal bool) { s.jobs[jobName] = j ctx = job.WithLogger(ctx, jobLog) + ctx = zfscmd.WithJobID(ctx, j.Name()) ctx, wakeup := wakeup.Context(ctx) ctx, resetFunc := reset.Context(ctx) s.wakeups[jobName] = wakeup diff --git a/daemon/logging/build_logging.go b/daemon/logging/build_logging.go index 7b46d59..1216a11 100644 --- a/daemon/logging/build_logging.go +++ b/daemon/logging/build_logging.go @@ -22,6 +22,7 @@ import ( "github.com/zrepl/zrepl/rpc/transportmux" "github.com/zrepl/zrepl/tlsconf" "github.com/zrepl/zrepl/transport" + "github.com/zrepl/zrepl/zfs/zfscmd" ) func OutletsFromConfig(in config.LoggingOutletEnumList) (*logger.Outlets, error) { @@ -79,6 +80,7 @@ const ( SubsysRPC Subsystem = "rpc" SubsysRPCControl Subsystem = "rpc.ctrl" SubsysRPCData Subsystem = "rpc.data" + SubsysZFSCmd Subsystem = "zfs.cmd" ) func WithSubsystemLoggers(ctx context.Context, log logger.Logger) context.Context { @@ -90,6 +92,7 @@ func WithSubsystemLoggers(ctx context.Context, log logger.Logger) context.Contex ctx = hooks.WithLogger(ctx, log.WithField(SubsysField, SubsysHooks)) ctx = transport.WithLogger(ctx, log.WithField(SubsysField, SubsysTransport)) ctx = transportmux.WithLogger(ctx, log.WithField(SubsysField, SubsysTransportMux)) + ctx = zfscmd.WithLogger(ctx, log.WithField(SubsysField, SubsysZFSCmd)) ctx = rpc.WithLoggers(ctx, rpc.Loggers{ General: log.WithField(SubsysField, SubsysRPC), diff --git a/daemon/snapper/snapper.go b/daemon/snapper/snapper.go index f7c4936..ec24785 100644 --- a/daemon/snapper/snapper.go +++ b/daemon/snapper/snapper.go @@ -277,7 +277,7 @@ func snapshot(a args, u updater) state { jobCallback := hooks.NewCallbackHookForFilesystem("snapshot", fs, func(_ context.Context) (err error) { l.Debug("create snapshot") - err = zfs.ZFSSnapshot(fs, snapname, false) // TODO propagate context to ZFSSnapshot + err = zfs.ZFSSnapshot(a.ctx, fs, snapname, false) // TODO propagate context to ZFSSnapshot if err != nil { l.WithError(err).Error("cannot create snapshot") } diff --git a/endpoint/endpoint.go b/endpoint/endpoint.go index b61dd72..031b7a7 100644 --- a/endpoint/endpoint.go +++ b/endpoint/endpoint.go @@ -537,7 +537,7 @@ func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemR fss := make([]*pdu.Filesystem, 0, len(filtered)) for _, a := range filtered { l := getLogger(ctx).WithField("fs", a) - ph, err := zfs.ZFSGetFilesystemPlaceholderState(a) + ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, a) if err != nil { l.WithError(err).Error("error getting placeholder state") return nil, errors.Wrapf(err, "cannot get placeholder state for fs %q", a) @@ -660,7 +660,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs if v.Path.Equal(lp) { return false } - ph, err := zfs.ZFSGetFilesystemPlaceholderState(v.Path) + ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, v.Path) getLogger(ctx). WithField("fs", v.Path.ToString()). WithField("placeholder_state", fmt.Sprintf("%#v", ph)). @@ -684,7 +684,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs } l := getLogger(ctx).WithField("placeholder_fs", v.Path) l.Debug("create placeholder filesystem") - err := zfs.ZFSCreatePlaceholderFilesystem(v.Path) + err := zfs.ZFSCreatePlaceholderFilesystem(ctx, v.Path) if err != nil { l.WithError(err).Error("cannot create placeholder filesystem") visitErr = err @@ -704,19 +704,19 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs // determine whether we need to rollback the filesystem / change its placeholder state var clearPlaceholderProperty bool var recvOpts zfs.RecvOptions - ph, err := zfs.ZFSGetFilesystemPlaceholderState(lp) + ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, lp) if err == nil && ph.FSExists && ph.IsPlaceholder { recvOpts.RollbackAndForceRecv = true clearPlaceholderProperty = true } if clearPlaceholderProperty { - if err := zfs.ZFSSetPlaceholder(lp, false); err != nil { + if err := zfs.ZFSSetPlaceholder(ctx, lp, false); err != nil { return nil, fmt.Errorf("cannot clear placeholder property for forced receive: %s", err) } } if req.ClearResumeToken && ph.FSExists { - if err := zfs.ZFSRecvClearResumeToken(lp.ToString()); err != nil { + if err := zfs.ZFSRecvClearResumeToken(ctx, lp.ToString()); err != nil { return nil, errors.Wrap(err, "cannot clear resume token") } } diff --git a/endpoint/endpoint_zfs_abstraction.go b/endpoint/endpoint_zfs_abstraction.go index c4d88d5..540917a 100644 --- a/endpoint/endpoint_zfs_abstraction.go +++ b/endpoint/endpoint_zfs_abstraction.go @@ -8,6 +8,7 @@ import ( "strings" "github.com/pkg/errors" + "github.com/zrepl/zrepl/util/envconst" "github.com/zrepl/zrepl/zfs" ) diff --git a/endpoint/endpoint_zfs_abstraction_cursor_and_last_received_hold.go b/endpoint/endpoint_zfs_abstraction_cursor_and_last_received_hold.go index b4b42ba..38ad2cb 100644 --- a/endpoint/endpoint_zfs_abstraction_cursor_and_last_received_hold.go +++ b/endpoint/endpoint_zfs_abstraction_cursor_and_last_received_hold.go @@ -9,6 +9,7 @@ import ( "github.com/kr/pretty" "github.com/pkg/errors" + "github.com/zrepl/zrepl/util/errorarray" "github.com/zrepl/zrepl/zfs" ) @@ -141,7 +142,7 @@ func MoveReplicationCursor(ctx context.Context, fs string, target ReplicationCur // idempotently create bookmark (guid is encoded in it, hence we'll most likely add a new one // cleanup the old one afterwards - err = zfs.ZFSBookmark(fs, target.ToSendArgVersion(), bookmarkname) + err = zfs.ZFSBookmark(ctx, fs, target.ToSendArgVersion(), bookmarkname) if err != nil { if err == zfs.ErrBookmarkCloningNotSupported { return nil, err // TODO go1.13 use wrapping @@ -370,7 +371,7 @@ func (c ReplicationCursorV1) String() string { return fmt.Sprintf("%s %s", c.Type, c.GetFullPath()) } func (c ReplicationCursorV1) Destroy(ctx context.Context) error { - if err := zfs.ZFSDestroyIdempotent(c.GetFullPath()); err != nil { + if err := zfs.ZFSDestroyIdempotent(ctx, c.GetFullPath()); err != nil { return errors.Wrapf(err, "destroy %s %s: zfs", c.Type, c.GetFullPath()) } return nil diff --git a/endpoint/endpoint_zfs_abstraction_step.go b/endpoint/endpoint_zfs_abstraction_step.go index 82deb00..3f5a866 100644 --- a/endpoint/endpoint_zfs_abstraction_step.go +++ b/endpoint/endpoint_zfs_abstraction_step.go @@ -6,6 +6,7 @@ import ( "regexp" "github.com/pkg/errors" + "github.com/zrepl/zrepl/util/errorarray" "github.com/zrepl/zrepl/zfs" ) @@ -86,7 +87,7 @@ func HoldStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID Job return errors.Wrap(err, "create step bookmark: determine bookmark name") } // idempotently create bookmark - err = zfs.ZFSBookmark(fs, v.ToSendArgVersion(), bmname) + err = zfs.ZFSBookmark(ctx, fs, v.ToSendArgVersion(), bmname) if err != nil { if err == zfs.ErrBookmarkCloningNotSupported { // TODO we could actually try to find a local snapshot that has the requested GUID @@ -131,7 +132,7 @@ func ReleaseStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID } // idempotently destroy bookmark - if err := zfs.ZFSDestroyIdempotent(bmname); err != nil { + if err := zfs.ZFSDestroyIdempotent(ctx, bmname); err != nil { return errors.Wrap(err, "step release: bookmark destroy: zfs") } diff --git a/endpoint/endpoint_zfs_helpers_types.go b/endpoint/endpoint_zfs_helpers_types.go index 6e0a0cb..b3f4eaf 100644 --- a/endpoint/endpoint_zfs_helpers_types.go +++ b/endpoint/endpoint_zfs_helpers_types.go @@ -6,6 +6,7 @@ import ( "fmt" "github.com/pkg/errors" + "github.com/zrepl/zrepl/zfs" ) @@ -34,7 +35,7 @@ func (b ListHoldsAndBookmarksOutputBookmark) GetFilesystemVersion() zfs.Filesyst } func (b ListHoldsAndBookmarksOutputBookmark) Destroy(ctx context.Context) error { - if err := zfs.ZFSDestroyIdempotent(b.GetFullPath()); err != nil { + if err := zfs.ZFSDestroyIdempotent(ctx, b.GetFullPath()); err != nil { return errors.Wrapf(err, "destroy %s: zfs", b) } return nil diff --git a/platformtest/platformtest_zpool.go b/platformtest/platformtest_zpool.go index c2cf509..1e3a099 100644 --- a/platformtest/platformtest_zpool.go +++ b/platformtest/platformtest_zpool.go @@ -45,7 +45,7 @@ func CreateOrReplaceZpool(ctx context.Context, e Execer, args ZpoolCreateArgs) ( } // export pool if it already exists (idempotence) - if _, err := zfs.ZFSGetRawAnySource(args.PoolName, []string{"name"}); err != nil { + if _, err := zfs.ZFSGetRawAnySource(ctx, args.PoolName, []string{"name"}); err != nil { if _, ok := err.(*zfs.DatasetDoesNotExist); ok { // we'll create it shortly } else { diff --git a/platformtest/tests/getNonexistent.go b/platformtest/tests/getNonexistent.go index 2b961fd..0520b65 100644 --- a/platformtest/tests/getNonexistent.go +++ b/platformtest/tests/getNonexistent.go @@ -17,14 +17,14 @@ func GetNonexistent(ctx *platformtest.Context) { `) // test raw - _, err := zfs.ZFSGetRawAnySource(fmt.Sprintf("%s/foo bar", ctx.RootDataset), []string{"name"}) + _, err := zfs.ZFSGetRawAnySource(ctx, fmt.Sprintf("%s/foo bar", ctx.RootDataset), []string{"name"}) if err != nil { panic(err) } // test nonexistent filesystem nonexistent := fmt.Sprintf("%s/nonexistent filesystem", ctx.RootDataset) - props, err := zfs.ZFSGetRawAnySource(nonexistent, []string{"name"}) + props, err := zfs.ZFSGetRawAnySource(ctx, nonexistent, []string{"name"}) if err == nil { panic(props) } @@ -37,7 +37,7 @@ func GetNonexistent(ctx *platformtest.Context) { // test nonexistent snapshot nonexistent = fmt.Sprintf("%s/foo bar@non existent", ctx.RootDataset) - props, err = zfs.ZFSGetRawAnySource(nonexistent, []string{"name"}) + props, err = zfs.ZFSGetRawAnySource(ctx, nonexistent, []string{"name"}) if err == nil { panic(props) } @@ -50,7 +50,7 @@ func GetNonexistent(ctx *platformtest.Context) { // test nonexistent bookmark nonexistent = fmt.Sprintf("%s/foo bar#non existent", ctx.RootDataset) - props, err = zfs.ZFSGetRawAnySource(nonexistent, []string{"name"}) + props, err = zfs.ZFSGetRawAnySource(ctx, nonexistent, []string{"name"}) if err == nil { panic(props) } diff --git a/platformtest/tests/helpers.go b/platformtest/tests/helpers.go index 0665672..7f7b612 100644 --- a/platformtest/tests/helpers.go +++ b/platformtest/tests/helpers.go @@ -14,8 +14,8 @@ import ( "github.com/zrepl/zrepl/zfs" ) -func sendArgVersion(fs, relName string) zfs.ZFSSendArgVersion { - guid, err := zfs.ZFSGetGUID(fs, relName) +func sendArgVersion(ctx *platformtest.Context, fs, relName string) zfs.ZFSSendArgVersion { + guid, err := zfs.ZFSGetGUID(ctx, fs, relName) if err != nil { panic(err) } @@ -25,8 +25,8 @@ func sendArgVersion(fs, relName string) zfs.ZFSSendArgVersion { } } -func fsversion(fs, relname string) zfs.FilesystemVersion { - v, err := zfs.ZFSGetFilesystemVersion(fs + relname) +func fsversion(ctx *platformtest.Context, fs, relname string) zfs.FilesystemVersion { + v, err := zfs.ZFSGetFilesystemVersion(ctx, fs+relname) if err != nil { panic(err) } @@ -41,7 +41,7 @@ func mustDatasetPath(fs string) *zfs.DatasetPath { return p } -func mustSnapshot(snap string) { +func mustSnapshot(ctx *platformtest.Context, snap string) { if err := zfs.EntityNamecheck(snap, zfs.EntityTypeSnapshot); err != nil { panic(err) } @@ -49,16 +49,16 @@ func mustSnapshot(snap string) { if len(comps) != 2 { panic(comps) } - err := zfs.ZFSSnapshot(mustDatasetPath(comps[0]), comps[1], false) + err := zfs.ZFSSnapshot(ctx, mustDatasetPath(comps[0]), comps[1], false) if err != nil { panic(err) } } -func mustGetFilesystemVersion(snapOrBookmark string) zfs.FilesystemVersion { - props, err := zfs.ZFSGetFilesystemVersion(snapOrBookmark) +func mustGetFilesystemVersion(ctx *platformtest.Context, snapOrBookmark string) zfs.FilesystemVersion { + v, err := zfs.ZFSGetFilesystemVersion(ctx, snapOrBookmark) check(err) - return props + return v } func check(err error) { @@ -95,7 +95,7 @@ type resumeSituation struct { func makeDummyDataSnapshots(ctx *platformtest.Context, sendFS string) (situation dummySnapshotSituation) { situation.sendFS = sendFS - sendFSMount, err := zfs.ZFSGetMountpoint(sendFS) + sendFSMount, err := zfs.ZFSGetMountpoint(ctx, sendFS) require.NoError(ctx, err) require.True(ctx, sendFSMount.Mounted) @@ -103,13 +103,13 @@ func makeDummyDataSnapshots(ctx *platformtest.Context, sendFS string) (situation situation.dummyDataLen = dummyLen writeDummyData(path.Join(sendFSMount.Mountpoint, "dummy_data"), dummyLen) - mustSnapshot(sendFS + "@a snapshot") - snapA := sendArgVersion(sendFS, "@a snapshot") + mustSnapshot(ctx, sendFS+"@a snapshot") + snapA := sendArgVersion(ctx, sendFS, "@a snapshot") situation.snapA = &snapA writeDummyData(path.Join(sendFSMount.Mountpoint, "dummy_data"), dummyLen) - mustSnapshot(sendFS + "@b snapshot") - snapB := sendArgVersion(sendFS, "@b snapshot") + mustSnapshot(ctx, sendFS+"@b snapshot") + snapB := sendArgVersion(ctx, sendFS, "@b snapshot") situation.snapB = &snapB return situation diff --git a/platformtest/tests/idempotentBookmark.go b/platformtest/tests/idempotentBookmark.go index d95b8b8..d3426c7 100644 --- a/platformtest/tests/idempotentBookmark.go +++ b/platformtest/tests/idempotentBookmark.go @@ -19,22 +19,22 @@ func IdempotentBookmark(ctx *platformtest.Context) { fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset) - asnap := sendArgVersion(fs, "@a snap") - anotherSnap := sendArgVersion(fs, "@another snap") + asnap := sendArgVersion(ctx, fs, "@a snap") + anotherSnap := sendArgVersion(ctx, fs, "@another snap") - err := zfs.ZFSBookmark(fs, asnap, "a bookmark") + err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark") if err != nil { panic(err) } // do it again, should be idempotent - err = zfs.ZFSBookmark(fs, asnap, "a bookmark") + err = zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark") if err != nil { panic(err) } // should fail for another snapshot - err = zfs.ZFSBookmark(fs, anotherSnap, "a bookmark") + err = zfs.ZFSBookmark(ctx, fs, anotherSnap, "a bookmark") if err == nil { panic(err) } @@ -43,12 +43,12 @@ func IdempotentBookmark(ctx *platformtest.Context) { } // destroy the snapshot - if err := zfs.ZFSDestroy(fmt.Sprintf("%s@a snap", fs)); err != nil { + if err := zfs.ZFSDestroy(ctx, fmt.Sprintf("%s@a snap", fs)); err != nil { panic(err) } // do it again, should fail with special error type - err = zfs.ZFSBookmark(fs, asnap, "a bookmark") + err = zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark") if err == nil { panic(err) } diff --git a/platformtest/tests/idempotentDestroy.go b/platformtest/tests/idempotentDestroy.go index b9540aa..53de8b1 100644 --- a/platformtest/tests/idempotentDestroy.go +++ b/platformtest/tests/idempotentDestroy.go @@ -18,8 +18,8 @@ func IdempotentDestroy(ctx *platformtest.Context) { `) fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset) - asnap := sendArgVersion(fs, "@a snap") - err := zfs.ZFSBookmark(fs, asnap, "a bookmark") + asnap := sendArgVersion(ctx, fs, "@a snap") + err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark") if err != nil { panic(err) } @@ -41,17 +41,17 @@ func IdempotentDestroy(ctx *platformtest.Context) { log.Printf("SUBBEGIN testing idempotent destroy %q for path %q", c.description, c.path) log.Println("destroy existing") - err = zfs.ZFSDestroy(c.path) + err = zfs.ZFSDestroy(ctx, c.path) if err != nil { panic(err) } log.Println("destroy again, non-idempotently, must error") - err = zfs.ZFSDestroy(c.path) + err = zfs.ZFSDestroy(ctx, c.path) if _, ok := err.(*zfs.DatasetDoesNotExist); !ok { panic(fmt.Sprintf("%T: %s", err, err)) } log.Println("destroy again, idempotently, must not error") - err = zfs.ZFSDestroyIdempotent(c.path) + err = zfs.ZFSDestroyIdempotent(ctx, c.path) if err != nil { panic(err) } @@ -62,12 +62,12 @@ func IdempotentDestroy(ctx *platformtest.Context) { } // also test idempotent destroy for cases where the parent dataset does not exist - err = zfs.ZFSDestroyIdempotent(fmt.Sprintf("%s/not foo bar@nonexistent snapshot", ctx.RootDataset)) + err = zfs.ZFSDestroyIdempotent(ctx, fmt.Sprintf("%s/not foo bar@nonexistent snapshot", ctx.RootDataset)) if err != nil { panic(err) } - err = zfs.ZFSDestroyIdempotent(fmt.Sprintf("%s/not foo bar#nonexistent bookmark", ctx.RootDataset)) + err = zfs.ZFSDestroyIdempotent(ctx, fmt.Sprintf("%s/not foo bar#nonexistent bookmark", ctx.RootDataset)) if err != nil { panic(err) } diff --git a/platformtest/tests/idempotentHold.go b/platformtest/tests/idempotentHold.go index d53b55d..8b33242 100644 --- a/platformtest/tests/idempotentHold.go +++ b/platformtest/tests/idempotentHold.go @@ -22,7 +22,7 @@ func IdempotentHold(ctx *platformtest.Context) { `) fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset) - v1 := fsversion(fs, "@1") + v1 := fsversion(ctx, fs, "@1") tag := "zrepl_platformtest" err := zfs.ZFSHold(ctx, fs, v1, tag) diff --git a/platformtest/tests/replicationCursor.go b/platformtest/tests/replicationCursor.go index d589dc8..8c51c01 100644 --- a/platformtest/tests/replicationCursor.go +++ b/platformtest/tests/replicationCursor.go @@ -5,6 +5,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/zrepl/zrepl/endpoint" "github.com/zrepl/zrepl/platformtest" "github.com/zrepl/zrepl/zfs" @@ -31,7 +32,7 @@ func ReplicationCursor(ctx *platformtest.Context) { } fs := ds.ToString() - snap := fsversion(fs, "@1 with space") + snap := fsversion(ctx, fs, "@1 with space") destroyed, err := endpoint.MoveReplicationCursor(ctx, fs, &snap, jobid) if err != nil { @@ -39,7 +40,7 @@ func ReplicationCursor(ctx *platformtest.Context) { } assert.Empty(ctx, destroyed) - snapProps, err := zfs.ZFSGetFilesystemVersion(snap.FullPath(fs)) + snapProps, err := zfs.ZFSGetFilesystemVersion(ctx, snap.FullPath(fs)) if err != nil { panic(err) } @@ -59,7 +60,7 @@ func ReplicationCursor(ctx *platformtest.Context) { cursor1BookmarkName, err := endpoint.ReplicationCursorBookmarkName(fs, snap.Guid, jobid) require.NoError(ctx, err) - snap2 := fsversion(fs, "@2 with space") + snap2 := fsversion(ctx, fs, "@2 with space") destroyed, err = endpoint.MoveReplicationCursor(ctx, fs, &snap2, jobid) require.NoError(ctx, err) require.Equal(ctx, 1, len(destroyed)) diff --git a/platformtest/tests/resumeTokenParsing.go b/platformtest/tests/resumeTokenParsing.go index 3ace583..eeb29c4 100644 --- a/platformtest/tests/resumeTokenParsing.go +++ b/platformtest/tests/resumeTokenParsing.go @@ -16,7 +16,7 @@ type resumeTokenTest struct { func (rtt *resumeTokenTest) Test(t *platformtest.Context) { - resumeSendSupported, err := zfs.ResumeSendSupported() + resumeSendSupported, err := zfs.ResumeSendSupported(t) if err != nil { t.Errorf("cannot determine whether resume supported: %T %s", err, err) t.FailNow() diff --git a/platformtest/tests/sendArgsValidation.go b/platformtest/tests/sendArgsValidation.go index da6858e..91ee6d9 100644 --- a/platformtest/tests/sendArgsValidation.go +++ b/platformtest/tests/sendArgsValidation.go @@ -23,7 +23,7 @@ func SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden(ctx *platformt `) fs := fmt.Sprintf("%s/send er", ctx.RootDataset) - props := mustGetFilesystemVersion(fs + "@a snap") + props := mustGetFilesystemVersion(ctx, fs+"@a snap") sendArgs, err := zfs.ZFSSendArgsUnvalidated{ FS: fs, @@ -63,7 +63,7 @@ func SendArgsValidationResumeTokenEncryptionMismatchForbidden(ctx *platformtest. if !supported { ctx.SkipNow() } - supported, err = zfs.ResumeSendSupported() + supported, err = zfs.ResumeSendSupported(ctx) check(err) if !supported { ctx.SkipNow() @@ -144,7 +144,7 @@ func SendArgsValidationResumeTokenDifferentFilesystemForbidden(ctx *platformtest if !supported { ctx.SkipNow() } - supported, err = zfs.ResumeSendSupported() + supported, err = zfs.ResumeSendSupported(ctx) check(err) if !supported { ctx.SkipNow() diff --git a/platformtest/tests/undestroyableSnapshotParsing.go b/platformtest/tests/undestroyableSnapshotParsing.go index 857dd85..8b6089e 100644 --- a/platformtest/tests/undestroyableSnapshotParsing.go +++ b/platformtest/tests/undestroyableSnapshotParsing.go @@ -20,7 +20,7 @@ func UndestroyableSnapshotParsing(t *platformtest.Context) { R zfs hold zrepl_platformtest "${ROOTDS}/foo bar@4 5 6" `) - err := zfs.ZFSDestroy(fmt.Sprintf("%s/foo bar@1 2 3,4 5 6,7 8 9", t.RootDataset)) + err := zfs.ZFSDestroy(t, fmt.Sprintf("%s/foo bar@1 2 3,4 5 6,7 8 9", t.RootDataset)) if err == nil { panic("expecting destroy error due to hold") } diff --git a/zfs/encryption.go b/zfs/encryption.go index 48e44c8..8da07fc 100644 --- a/zfs/encryption.go +++ b/zfs/encryption.go @@ -10,6 +10,7 @@ import ( "github.com/pkg/errors" "github.com/zrepl/zrepl/util/envconst" + "github.com/zrepl/zrepl/zfs/zfscmd" ) var encryptionCLISupport struct { @@ -21,7 +22,7 @@ var encryptionCLISupport struct { func EncryptionCLISupported(ctx context.Context) (bool, error) { encryptionCLISupport.once.Do(func() { // "feature discovery" - cmd := exec.Command("zfs", "load-key") + cmd := zfscmd.CommandContext(ctx, "zfs", "load-key") output, err := cmd.CombinedOutput() if ee, ok := err.(*exec.ExitError); !ok || ok && !ee.Exited() { encryptionCLISupport.err = errors.Wrap(err, "native encryption cli support feature check failed") @@ -50,7 +51,7 @@ func ZFSGetEncryptionEnabled(ctx context.Context, fs string) (enabled bool, err return false, err } - props, err := zfsGet(fs, []string{"encryption"}, sourceAny) + props, err := zfsGet(ctx, fs, []string{"encryption"}, sourceAny) if err != nil { return false, errors.Wrap(err, "cannot get `encryption` property") } diff --git a/zfs/holds.go b/zfs/holds.go index a85c022..af17f11 100644 --- a/zfs/holds.go +++ b/zfs/holds.go @@ -6,13 +6,13 @@ import ( "context" "fmt" "os" - "os/exec" "strings" "syscall" "github.com/pkg/errors" "github.com/zrepl/zrepl/util/envconst" + "github.com/zrepl/zrepl/zfs/zfscmd" ) // no need for feature tests, holds have been around forever @@ -43,7 +43,7 @@ func ZFSHold(ctx context.Context, fs string, v FilesystemVersion, tag string) er return err } fullPath := v.FullPath(fs) - output, err := exec.CommandContext(ctx, "zfs", "hold", tag, fullPath).CombinedOutput() + output, err := zfscmd.CommandContext(ctx, "zfs", "hold", tag, fullPath).CombinedOutput() if err != nil { if bytes.Contains(output, []byte("tag already exists on this dataset")) { goto success @@ -62,7 +62,7 @@ func ZFSHolds(ctx context.Context, fs, snap string) ([]string, error) { return nil, fmt.Errorf("`snap` must not be empty") } dp := fmt.Sprintf("%s@%s", fs, snap) - output, err := exec.CommandContext(ctx, "zfs", "holds", "-H", dp).CombinedOutput() + output, err := zfscmd.CommandContext(ctx, "zfs", "holds", "-H", dp).CombinedOutput() if err != nil { return nil, &ZFSError{output, errors.Wrap(err, "zfs holds failed")} } @@ -99,11 +99,13 @@ func ZFSRelease(ctx context.Context, tag string, snaps ...string) error { } args := []string{"release", tag} args = append(args, snaps[i:j]...) - output, err := exec.CommandContext(ctx, "zfs", args...).CombinedOutput() + output, err := zfscmd.CommandContext(ctx, "zfs", args...).CombinedOutput() if pe, ok := err.(*os.PathError); err != nil && ok && pe.Err == syscall.E2BIG { maxInvocationLen = maxInvocationLen / 2 continue } + // further error handling part of error scraper below + maxInvocationLen = maxInvocationLen + os.Getpagesize() i = j diff --git a/zfs/placeholder.go b/zfs/placeholder.go index 1fc95f1..0ffffa8 100644 --- a/zfs/placeholder.go +++ b/zfs/placeholder.go @@ -1,11 +1,12 @@ package zfs import ( - "bytes" + "context" "crypto/sha512" "encoding/hex" "fmt" - "os/exec" + + "github.com/zrepl/zrepl/zfs/zfscmd" ) const ( @@ -61,10 +62,10 @@ type FilesystemPlaceholderState struct { // is a placeholder. Note that the property source must be `local` for the returned value to be valid. // // For nonexistent FS, err == nil and state.FSExists == false -func ZFSGetFilesystemPlaceholderState(p *DatasetPath) (state *FilesystemPlaceholderState, err error) { +func ZFSGetFilesystemPlaceholderState(ctx context.Context, p *DatasetPath) (state *FilesystemPlaceholderState, err error) { state = &FilesystemPlaceholderState{FS: p.ToString()} state.FS = p.ToString() - props, err := zfsGet(p.ToString(), []string{PlaceholderPropertyName}, sourceLocal) + props, err := zfsGet(ctx, p.ToString(), []string{PlaceholderPropertyName}, sourceLocal) var _ error = (*DatasetDoesNotExist)(nil) // weak assertion on zfsGet's interface if _, ok := err.(*DatasetDoesNotExist); ok { return state, nil @@ -77,25 +78,19 @@ func ZFSGetFilesystemPlaceholderState(p *DatasetPath) (state *FilesystemPlacehol return state, nil } -func ZFSCreatePlaceholderFilesystem(p *DatasetPath) (err error) { +func ZFSCreatePlaceholderFilesystem(ctx context.Context, p *DatasetPath) (err error) { if p.Length() == 1 { return fmt.Errorf("cannot create %q: pools cannot be created with zfs create", p.ToString()) } - cmd := exec.Command(ZFS_BINARY, "create", + cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "create", "-o", fmt.Sprintf("%s=%s", PlaceholderPropertyName, placeholderPropertyOn), "-o", "mountpoint=none", p.ToString()) - stderr := bytes.NewBuffer(make([]byte, 0, 1024)) - cmd.Stderr = stderr - - if err = cmd.Start(); err != nil { - return err - } - - if err = cmd.Wait(); err != nil { + stdio, err := cmd.CombinedOutput() + if err != nil { err = &ZFSError{ - Stderr: stderr.Bytes(), + Stderr: stdio, WaitErr: err, } } @@ -103,14 +98,14 @@ func ZFSCreatePlaceholderFilesystem(p *DatasetPath) (err error) { return } -func ZFSSetPlaceholder(p *DatasetPath, isPlaceholder bool) error { +func ZFSSetPlaceholder(ctx context.Context, p *DatasetPath, isPlaceholder bool) error { props := NewZFSProperties() prop := placeholderPropertyOff if isPlaceholder { prop = placeholderPropertyOn } props.Set(PlaceholderPropertyName, prop) - return zfsSet(p.ToString(), props) + return zfsSet(ctx, p.ToString(), props) } type MigrateHashBasedPlaceholderReport struct { @@ -119,8 +114,8 @@ type MigrateHashBasedPlaceholderReport struct { } // fs must exist, will panic otherwise -func ZFSMigrateHashBasedPlaceholderToCurrent(fs *DatasetPath, dryRun bool) (*MigrateHashBasedPlaceholderReport, error) { - st, err := ZFSGetFilesystemPlaceholderState(fs) +func ZFSMigrateHashBasedPlaceholderToCurrent(ctx context.Context, fs *DatasetPath, dryRun bool) (*MigrateHashBasedPlaceholderReport, error) { + st, err := ZFSGetFilesystemPlaceholderState(ctx, fs) if err != nil { return nil, fmt.Errorf("error getting placeholder state: %s", err) } @@ -137,7 +132,7 @@ func ZFSMigrateHashBasedPlaceholderToCurrent(fs *DatasetPath, dryRun bool) (*Mig return &report, nil } - err = ZFSSetPlaceholder(fs, st.IsPlaceholder) + err = ZFSSetPlaceholder(ctx, fs, st.IsPlaceholder) if err != nil { return nil, fmt.Errorf("error re-writing placeholder property: %s", err) } diff --git a/zfs/resume_token.go b/zfs/resume_token.go index 80856ea..5eb60cf 100644 --- a/zfs/resume_token.go +++ b/zfs/resume_token.go @@ -13,6 +13,7 @@ import ( "github.com/pkg/errors" "github.com/zrepl/zrepl/util/envconst" + "github.com/zrepl/zrepl/zfs/zfscmd" ) // NOTE: Update ZFSSendARgs.Validate when changing fields (potentially SECURITY SENSITIVE) @@ -39,10 +40,10 @@ var resumeSendSupportedCheck struct { err error } -func ResumeSendSupported() (bool, error) { +func ResumeSendSupported(ctx context.Context) (bool, error) { resumeSendSupportedCheck.once.Do(func() { // "feature discovery" - cmd := exec.Command("zfs", "send") + cmd := zfscmd.CommandContext(ctx, "zfs", "send") output, err := cmd.CombinedOutput() if ee, ok := err.(*exec.ExitError); !ok || ok && !ee.Exited() { resumeSendSupportedCheck.err = errors.Wrap(err, "resumable send cli support feature check failed") @@ -86,7 +87,7 @@ func ResumeRecvSupported(ctx context.Context, fs *DatasetPath) (bool, error) { } if !sup.flagSupport.checked { - output, err := exec.CommandContext(ctx, "zfs", "receive").CombinedOutput() + output, err := zfscmd.CommandContext(ctx, ZFS_BINARY, "receive").CombinedOutput() upgradeWhile(func() { sup.flagSupport.checked = true if ee, ok := err.(*exec.ExitError); err != nil && (!ok || ok && !ee.Exited()) { @@ -124,7 +125,7 @@ func ResumeRecvSupported(ctx context.Context, fs *DatasetPath) (bool, error) { if poolSup, ok = sup.poolSupported[pool]; !ok || // shadow (!poolSup.supported && time.Since(poolSup.lastCheck) > resumeRecvPoolSupportRecheckTimeout) { - output, err := exec.CommandContext(ctx, "zpool", "get", "-H", "-p", "-o", "value", "feature@extensible_dataset", pool).CombinedOutput() + output, err := zfscmd.CommandContext(ctx, "zpool", "get", "-H", "-p", "-o", "value", "feature@extensible_dataset", pool).CombinedOutput() if err != nil { debug("resume recv pool support check result: %#v", sup.flagSupport) poolSup.supported = false @@ -155,7 +156,7 @@ func ResumeRecvSupported(ctx context.Context, fs *DatasetPath) (bool, error) { // FIXME: implement nvlist unpacking in Go and read through libzfs_sendrecv.c func ParseResumeToken(ctx context.Context, token string) (*ResumeToken, error) { - if supported, err := ResumeSendSupported(); err != nil { + if supported, err := ResumeSendSupported(ctx); err != nil { return nil, err } else if !supported { return nil, ResumeTokenDecodingNotSupported @@ -181,7 +182,7 @@ func ParseResumeToken(ctx context.Context, token string) (*ResumeToken, error) { // toname = pool1/test@b //cannot resume send: 'pool1/test@b' used in the initial send no longer exists - cmd := exec.CommandContext(ctx, ZFS_BINARY, "send", "-nvt", string(token)) + cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "send", "-nvt", string(token)) output, err := cmd.CombinedOutput() if err != nil { if exitErr, ok := err.(*exec.ExitError); ok { @@ -258,7 +259,7 @@ func ZFSGetReceiveResumeTokenOrEmptyStringIfNotSupported(ctx context.Context, fs return "", nil } const prop_receive_resume_token = "receive_resume_token" - props, err := ZFSGet(fs, []string{prop_receive_resume_token}) + props, err := ZFSGet(ctx, fs, []string{prop_receive_resume_token}) if err != nil { return "", err } diff --git a/zfs/versions.go b/zfs/versions.go index 8688890..89a0f8c 100644 --- a/zfs/versions.go +++ b/zfs/versions.go @@ -204,8 +204,8 @@ func ZFSListFilesystemVersions(fs *DatasetPath, filter FilesystemVersionFilter) return } -func ZFSGetFilesystemVersion(ds string) (v FilesystemVersion, _ error) { - props, err := zfsGet(ds, []string{"createtxg", "guid", "creation"}, sourceAny) +func ZFSGetFilesystemVersion(ctx context.Context, ds string) (v FilesystemVersion, _ error) { + props, err := zfsGet(ctx, ds, []string{"createtxg", "guid", "creation"}, sourceAny) if err != nil { return v, err } diff --git a/zfs/versions_destroy.go b/zfs/versions_destroy.go index 02539da..ecc1908 100644 --- a/zfs/versions_destroy.go +++ b/zfs/versions_destroy.go @@ -11,9 +11,10 @@ import ( "syscall" "github.com/zrepl/zrepl/util/envconst" + "github.com/zrepl/zrepl/zfs/zfscmd" ) -func ZFSDestroyFilesystemVersion(filesystem *DatasetPath, version *FilesystemVersion) (err error) { +func ZFSDestroyFilesystemVersion(ctx context.Context, filesystem *DatasetPath, version *FilesystemVersion) (err error) { datasetPath := version.ToAbsPath(filesystem) @@ -22,7 +23,7 @@ func ZFSDestroyFilesystemVersion(filesystem *DatasetPath, version *FilesystemVer return fmt.Errorf("sanity check failed: no @ or # character found in %q", datasetPath) } - return ZFSDestroy(datasetPath) + return ZFSDestroy(ctx, datasetPath) } var destroyerSingleton = destroyerImpl{} @@ -48,8 +49,8 @@ func setDestroySnapOpErr(b []*DestroySnapOp, err error) { } type destroyer interface { - Destroy(args []string) error - DestroySnapshotsCommaSyntaxSupported() (bool, error) + Destroy(ctx context.Context, args []string) error + DestroySnapshotsCommaSyntaxSupported(context.Context) (bool, error) } func doDestroy(ctx context.Context, reqs []*DestroySnapOp, e destroyer) { @@ -69,7 +70,7 @@ func doDestroy(ctx context.Context, reqs []*DestroySnapOp, e destroyer) { } reqs = validated - commaSupported, err := e.DestroySnapshotsCommaSyntaxSupported() + commaSupported, err := e.DestroySnapshotsCommaSyntaxSupported(ctx) if err != nil { debug("destroy: comma syntax support detection failed: %s", err) setDestroySnapOpErr(reqs, err) @@ -85,7 +86,7 @@ func doDestroy(ctx context.Context, reqs []*DestroySnapOp, e destroyer) { func doDestroySeq(ctx context.Context, reqs []*DestroySnapOp, e destroyer) { for _, r := range reqs { - *r.ErrOut = e.Destroy([]string{fmt.Sprintf("%s@%s", r.Filesystem, r.Name)}) + *r.ErrOut = e.Destroy(ctx, []string{fmt.Sprintf("%s@%s", r.Filesystem, r.Name)}) } } @@ -139,7 +140,7 @@ func tryBatch(ctx context.Context, batch []*DestroySnapOp, d destroyer) error { } } batchArg := fmt.Sprintf("%s@%s", batchFS, strings.Join(batchNames, ",")) - return d.Destroy([]string{batchArg}) + return d.Destroy(ctx, []string{batchArg}) } // fsbatch must be on same filesystem @@ -203,7 +204,7 @@ func doDestroyBatchedRec(ctx context.Context, fsbatch []*DestroySnapOp, d destro type destroyerImpl struct{} -func (d destroyerImpl) Destroy(args []string) error { +func (d destroyerImpl) Destroy(ctx context.Context, args []string) error { if len(args) != 1 { // we have no use case for this at the moment, so let's crash (safer than destroying something unexpectedly) panic(fmt.Sprintf("unexpected number of arguments: %v", args)) @@ -212,7 +213,7 @@ func (d destroyerImpl) Destroy(args []string) error { if !strings.ContainsAny(args[0], "@") { panic(fmt.Sprintf("sanity check: expecting '@' in call to Destroy, got %q", args[0])) } - return ZFSDestroy(args[0]) + return ZFSDestroy(ctx, args[0]) } var batchDestroyFeatureCheck struct { @@ -221,10 +222,10 @@ var batchDestroyFeatureCheck struct { err error } -func (d destroyerImpl) DestroySnapshotsCommaSyntaxSupported() (bool, error) { +func (d destroyerImpl) DestroySnapshotsCommaSyntaxSupported(ctx context.Context) (bool, error) { batchDestroyFeatureCheck.once.Do(func() { // "feature discovery" - cmd := exec.Command(ZFS_BINARY, "destroy") + cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "destroy") output, err := cmd.CombinedOutput() if _, ok := err.(*exec.ExitError); !ok { debug("destroy feature check failed: %T %s", err, err) diff --git a/zfs/versions_destroy_test.go b/zfs/versions_destroy_test.go index 112ae83..df73e99 100644 --- a/zfs/versions_destroy_test.go +++ b/zfs/versions_destroy_test.go @@ -25,11 +25,11 @@ type mockBatchDestroy struct { e2biglen int } -func (m *mockBatchDestroy) DestroySnapshotsCommaSyntaxSupported() (bool, error) { +func (m *mockBatchDestroy) DestroySnapshotsCommaSyntaxSupported(_ context.Context) (bool, error) { return !m.commaUnsupported, nil } -func (m *mockBatchDestroy) Destroy(args []string) error { +func (m *mockBatchDestroy) Destroy(ctx context.Context, args []string) error { defer m.mtx.Lock().Unlock() if len(args) != 1 { panic("unexpected use of Destroy") diff --git a/zfs/zfs.go b/zfs/zfs.go index 9234e8f..531b38b 100644 --- a/zfs/zfs.go +++ b/zfs/zfs.go @@ -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, } } diff --git a/zfs/zfs_test.go b/zfs/zfs_test.go index 81ad617..606ab54 100644 --- a/zfs/zfs_test.go +++ b/zfs/zfs_test.go @@ -1,6 +1,7 @@ package zfs import ( + "context" "testing" "github.com/stretchr/testify/assert" @@ -11,9 +12,11 @@ func TestZFSListHandlesProducesZFSErrorOnNonZeroExit(t *testing.T) { var err error + ctx := context.Background() + ZFS_BINARY = "./test_helpers/zfs_failer.sh" - _, err = ZFSList([]string{"fictionalprop"}, "nonexistent/dataset") + _, err = ZFSList(ctx, []string{"fictionalprop"}, "nonexistent/dataset") assert.Error(t, err) zfsError, ok := err.(*ZFSError) diff --git a/zfs/zfscmd/zfscmd.go b/zfs/zfscmd/zfscmd.go new file mode 100644 index 0000000..fec40e0 --- /dev/null +++ b/zfs/zfscmd/zfscmd.go @@ -0,0 +1,122 @@ +// Package zfscmd provides a wrapper around packate os/exec. +// Functionality provided by the wrapper: +// - logging start and end of command execution +// - status report of active commands +// - prometheus metrics of runtimes +package zfscmd + +import ( + "context" + "io" + "os" + "os/exec" + "strings" + "sync" + "time" + + "github.com/zrepl/zrepl/util/circlog" +) + +type Cmd struct { + cmd *exec.Cmd + ctx context.Context + mtx sync.RWMutex + startedAt, waitReturnedAt time.Time +} + +func CommandContext(ctx context.Context, name string, arg ...string) *Cmd { + cmd := exec.CommandContext(ctx, name, arg...) + return &Cmd{cmd: cmd, ctx: ctx} +} + +// err.(*exec.ExitError).Stderr will NOT be set +func (c *Cmd) CombinedOutput() (o []byte, err error) { + o, err = c.cmd.CombinedOutput() + return +} + +// err.(*exec.ExitError).Stderr will be set +func (c *Cmd) Output() (o []byte, err error) { + o, err = c.cmd.Output() + return +} + +// Careful: err.(*exec.ExitError).Stderr will not be set, even if you don't open an StderrPipe +func (c *Cmd) StdoutPipeWithErrorBuf() (p io.ReadCloser, errBuf *circlog.CircularLog, err error) { + p, err = c.cmd.StdoutPipe() + errBuf = circlog.MustNewCircularLog(1 << 15) + c.cmd.Stderr = errBuf + return p, errBuf, err +} + +type Stdio struct { + Stdin io.ReadCloser + Stdout io.Writer + Stderr io.Writer +} + +func (c *Cmd) SetStdio(stdio Stdio) { + c.cmd.Stdin = stdio.Stdin + c.cmd.Stderr = stdio.Stderr + c.cmd.Stdout = stdio.Stdout +} + +func (c *Cmd) String() string { + return strings.Join(c.cmd.Args, " ") // includes argv[0] if initialized with CommandContext, that's the only way we o it +} + +func (c *Cmd) log() Logger { + return getLogger(c.ctx).WithField("cmd", c.String()) +} + +func (c *Cmd) Start() (err error) { + startPreLogging(c, time.Now()) + + err = c.cmd.Start() + now := time.Now() + + c.mtx.Lock() + c.startedAt = now + c.mtx.Unlock() + + startPostReport(c, err, now) + startPostLogging(c, err, now) + return err +} + +// only call this after a successful call to .Start() +func (c *Cmd) Process() *os.Process { + if c.startedAt.IsZero() { + panic("calling Process() only allowed after successful call to Start()") + } + return c.cmd.Process +} + +func (c *Cmd) Wait() (err error) { + waitPreLogging(c, time.Now()) + + err = c.cmd.Wait() + now := time.Now() + + if !c.waitReturnedAt.IsZero() { + // ignore duplicate waits + return + } + + c.mtx.Lock() + c.waitReturnedAt = now + c.mtx.Unlock() + + waitPostReport(c, now) + waitPostLogging(c, err, now) + waitPostPrometheus(c, err, now) + return err +} + +// returns 0 if the command did not yet finish +func (c *Cmd) Runtime() time.Duration { + if c.waitReturnedAt.IsZero() { + return 0 + } + return c.waitReturnedAt.Sub(c.startedAt) +} diff --git a/zfs/zfscmd/zfscmd_context.go b/zfs/zfscmd/zfscmd_context.go new file mode 100644 index 0000000..b087652 --- /dev/null +++ b/zfs/zfscmd/zfscmd_context.go @@ -0,0 +1,39 @@ +package zfscmd + +import ( + "context" + + "github.com/zrepl/zrepl/logger" +) + +type contextKey int + +const ( + contextKeyLogger contextKey = iota + contextKeyJobID +) + +type Logger = logger.Logger + +func WithJobID(ctx context.Context, jobID string) context.Context { + return context.WithValue(ctx, contextKeyJobID, jobID) +} + +func getJobIDOrDefault(ctx context.Context, def string) string { + ret, ok := ctx.Value(contextKeyJobID).(string) + if !ok { + return def + } + return ret +} + +func WithLogger(ctx context.Context, log Logger) context.Context { + return context.WithValue(ctx, contextKeyLogger, log) +} + +func getLogger(ctx context.Context) Logger { + if l, ok := ctx.Value(contextKeyLogger).(Logger); ok { + return l + } + return logger.NewNullLogger() +} diff --git a/zfs/zfscmd/zfscmd_logging.go b/zfs/zfscmd/zfscmd_logging.go new file mode 100644 index 0000000..4215bd0 --- /dev/null +++ b/zfs/zfscmd/zfscmd_logging.go @@ -0,0 +1,40 @@ +package zfscmd + +import ( + "time" +) + +// Implementation Note: +// +// Pre-events logged with debug +// Post-event without error logged with info +// Post-events with error logged at error level + +func startPreLogging(c *Cmd, now time.Time) { + c.log().Debug("starting command") +} + +func startPostLogging(c *Cmd, err error, now time.Time) { + if err == nil { + c.log().Info("started command") + } else { + c.log().WithError(err).Error("cannot start command") + } +} + +func waitPreLogging(c *Cmd, now time.Time) { + c.log().Debug("start waiting") +} + +func waitPostLogging(c *Cmd, err error, now time.Time) { + log := c.log(). + WithField("total_time_s", c.Runtime().Seconds()). + WithField("systemtime_s", c.cmd.ProcessState.SystemTime().Seconds()). + WithField("usertime_s", c.cmd.ProcessState.UserTime().Seconds()) + + if err == nil { + log.Info("command exited without error") + } else { + log.WithError(err).Error("command exited with error") + } +} diff --git a/zfs/zfscmd/zfscmd_platform_test.bash b/zfs/zfscmd/zfscmd_platform_test.bash new file mode 100755 index 0000000..cc4ba3f --- /dev/null +++ b/zfs/zfscmd/zfscmd_platform_test.bash @@ -0,0 +1,7 @@ +#!/usr/bin/env bash +set -euo pipefail + +echo "to stderr" 1>&2 +echo "to stdout" + +exit "$1" \ No newline at end of file diff --git a/zfs/zfscmd/zfscmd_platform_test.go b/zfs/zfscmd/zfscmd_platform_test.go new file mode 100644 index 0000000..982e32e --- /dev/null +++ b/zfs/zfscmd/zfscmd_platform_test.go @@ -0,0 +1,60 @@ +package zfscmd + +import ( + "bytes" + "io" + "os/exec" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testBin = "./zfscmd_platform_test.bash" + +func TestCmdStderrBehaviorOutput(t *testing.T) { + + stdout, err := exec.Command(testBin, "0").Output() + require.NoError(t, err) + require.Equal(t, []byte("to stdout\n"), stdout) + + stdout, err = exec.Command(testBin, "1").Output() + assert.Equal(t, []byte("to stdout\n"), stdout) + require.Error(t, err) + ee, ok := err.(*exec.ExitError) + require.True(t, ok) + require.Equal(t, ee.Stderr, []byte("to stderr\n")) +} + +func TestCmdStderrBehaviorCombinedOutput(t *testing.T) { + + stdio, err := exec.Command(testBin, "0").CombinedOutput() + require.NoError(t, err) + require.Equal(t, "to stderr\nto stdout\n", string(stdio)) + + stdio, err = exec.Command(testBin, "1").CombinedOutput() + require.Equal(t, "to stderr\nto stdout\n", string(stdio)) + require.Error(t, err) + ee, ok := err.(*exec.ExitError) + require.True(t, ok) + require.Empty(t, ee.Stderr) // !!!! maybe not what one would expect +} + +func TestCmdStderrBehaviorStdoutPipe(t *testing.T) { + cmd := exec.Command(testBin, "1") + stdoutPipe, err := cmd.StdoutPipe() + require.NoError(t, err) + err = cmd.Start() + require.NoError(t, err) + defer cmd.Wait() + var stdout bytes.Buffer + _, err = io.Copy(&stdout, stdoutPipe) + require.NoError(t, err) + require.Equal(t, "to stdout\n", stdout.String()) + + err = cmd.Wait() + require.Error(t, err) + ee, ok := err.(*exec.ExitError) + require.True(t, ok) + require.Empty(t, ee.Stderr) // !!!!! probably not what one would expect if we only redirect stdout +} diff --git a/zfs/zfscmd/zfscmd_prometheus.go b/zfs/zfscmd/zfscmd_prometheus.go new file mode 100644 index 0000000..f68e998 --- /dev/null +++ b/zfs/zfscmd/zfscmd_prometheus.go @@ -0,0 +1,73 @@ +package zfscmd + +import ( + "time" + + "github.com/prometheus/client_golang/prometheus" +) + +var metrics struct { + totaltime *prometheus.HistogramVec + systemtime *prometheus.HistogramVec + usertime *prometheus.HistogramVec +} + +var timeLabels = []string{"jobid", "zfsbinary", "zfsverb"} +var timeBuckets = []float64{0.01, 0.1, 0.2, 0.5, 0.75, 1, 2, 5, 10, 60} + +func init() { + metrics.totaltime = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "zrepl", + Subsystem: "zfscmd", + Name: "runtime", + Help: "number of seconds that the command took from start until wait returned", + Buckets: timeBuckets, + }, timeLabels) + metrics.systemtime = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "zrepl", + Subsystem: "zfscmd", + Name: "systemtime", + Help: "https://golang.org/pkg/os/#ProcessState.SystemTime", + Buckets: timeBuckets, + }, timeLabels) + metrics.usertime = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "zrepl", + Subsystem: "zfscmd", + Name: "usertime", + Help: "https://golang.org/pkg/os/#ProcessState.UserTime", + Buckets: timeBuckets, + }, timeLabels) + +} + +func RegisterMetrics(r prometheus.Registerer) { + r.MustRegister(metrics.totaltime) + r.MustRegister(metrics.systemtime) + r.MustRegister(metrics.usertime) +} + +func waitPostPrometheus(c *Cmd, err error, now time.Time) { + + if len(c.cmd.Args) < 2 { + getLogger(c.ctx).WithField("args", c.cmd.Args). + Warn("prometheus: cannot turn zfs command into metric") + return + } + + // Note: do not start parsing other aspects + // of the ZFS command line. This is not the suitable layer + // for such a task. + + jobid := getJobIDOrDefault(c.ctx, "_nojobid") + + labelValues := []string{jobid, c.cmd.Args[0], c.cmd.Args[1]} + + metrics.totaltime. + WithLabelValues(labelValues...). + Observe(c.Runtime().Seconds()) + metrics.systemtime.WithLabelValues(labelValues...). + Observe(c.cmd.ProcessState.SystemTime().Seconds()) + metrics.usertime.WithLabelValues(labelValues...). + Observe(c.cmd.ProcessState.UserTime().Seconds()) + +} diff --git a/zfs/zfscmd/zfscmd_report.go b/zfs/zfscmd/zfscmd_report.go new file mode 100644 index 0000000..abd7b4f --- /dev/null +++ b/zfs/zfscmd/zfscmd_report.go @@ -0,0 +1,68 @@ +package zfscmd + +import ( + "fmt" + "sync" + "time" +) + +type Report struct { + Active []ActiveCommand +} + +type ActiveCommand struct { + Path string + Args []string + StartedAt time.Time +} + +func GetReport() *Report { + active.mtx.RLock() + defer active.mtx.RUnlock() + var activeCommands []ActiveCommand + for c := range active.cmds { + c.mtx.RLock() + activeCommands = append(activeCommands, ActiveCommand{ + Path: c.cmd.Path, + Args: c.cmd.Args, + StartedAt: c.startedAt, + }) + c.mtx.RUnlock() + } + return &Report{ + Active: activeCommands, + } +} + +var active struct { + mtx sync.RWMutex + cmds map[*Cmd]bool +} + +func init() { + active.cmds = make(map[*Cmd]bool) +} + +func startPostReport(c *Cmd, err error, now time.Time) { + if err != nil { + return + } + + active.mtx.Lock() + prev := active.cmds[c] + if prev { + panic("impl error: duplicate active command") + } + active.cmds[c] = true + active.mtx.Unlock() +} + +func waitPostReport(c *Cmd, now time.Time) { + active.mtx.Lock() + defer active.mtx.Unlock() + prev := active.cmds[c] + if !prev { + panic(fmt.Sprintf("impl error: onWaitDone must only be called on an active command: %s", c)) + } + delete(active.cmds, c) +}