package zfs import ( "context" "fmt" "os" "os/exec" "sort" "strings" "sync" "syscall" "github.com/zrepl/zrepl/util/envconst" "github.com/zrepl/zrepl/zfs/zfscmd" ) func ZFSDestroyFilesystemVersion(ctx context.Context, filesystem *DatasetPath, version *FilesystemVersion) (err error) { datasetPath := version.ToAbsPath(filesystem) // Sanity check... if !strings.ContainsAny(datasetPath, "@#") { return fmt.Errorf("sanity check failed: no @ or # character found in %q", datasetPath) } return ZFSDestroy(ctx, datasetPath) } var destroyerSingleton = destroyerImpl{} type DestroySnapOp struct { Filesystem string Name string ErrOut *error } func (o *DestroySnapOp) String() string { return fmt.Sprintf("destroy operation %s@%s", o.Filesystem, o.Name) } func ZFSDestroyFilesystemVersions(ctx context.Context, reqs []*DestroySnapOp) { doDestroy(ctx, reqs, destroyerSingleton) } func setDestroySnapOpErr(b []*DestroySnapOp, err error) { for _, r := range b { *r.ErrOut = err } } type destroyer interface { Destroy(ctx context.Context, args []string) error DestroySnapshotsCommaSyntaxSupported(context.Context) (bool, error) } func doDestroy(ctx context.Context, reqs []*DestroySnapOp, e destroyer) { var validated []*DestroySnapOp for _, req := range reqs { // Filesystem and Snapshot should not be empty // ZFS will generally fail because those are invalid destroy arguments, // but we'd rather apply defensive programming here (doing destroy after all) if req.Filesystem == "" { *req.ErrOut = fmt.Errorf("Filesystem must not be an empty string") } else if req.Name == "" { *req.ErrOut = fmt.Errorf("Name must not be an empty string") } else { validated = append(validated, req) } } reqs = validated commaSupported, err := e.DestroySnapshotsCommaSyntaxSupported(ctx) if err != nil { debug("destroy: comma syntax support detection failed: %s", err) setDestroySnapOpErr(reqs, err) return } if !commaSupported { doDestroySeq(ctx, reqs, e) } else { doDestroyBatched(ctx, reqs, e) } } func doDestroySeq(ctx context.Context, reqs []*DestroySnapOp, e destroyer) { for _, r := range reqs { *r.ErrOut = e.Destroy(ctx, []string{fmt.Sprintf("%s@%s", r.Filesystem, r.Name)}) } } func doDestroyBatched(ctx context.Context, reqs []*DestroySnapOp, d destroyer) { perFS := buildBatches(reqs) for _, fsbatch := range perFS { doDestroyBatchedRec(ctx, fsbatch, d) } } func buildBatches(reqs []*DestroySnapOp) [][]*DestroySnapOp { if len(reqs) == 0 { return nil } sorted := make([]*DestroySnapOp, len(reqs)) copy(sorted, reqs) sort.SliceStable(sorted, func(i, j int) bool { // by filesystem, then snap name fscmp := strings.Compare(sorted[i].Filesystem, sorted[j].Filesystem) if fscmp != 0 { return fscmp == -1 } return strings.Compare(sorted[i].Name, sorted[j].Name) == -1 }) // group by fs var perFS [][]*DestroySnapOp consumed := 0 maxBatchSize := envconst.Int("ZREPL_DESTROY_MAX_BATCH_SIZE", 0) for consumed < len(sorted) { batchConsumedUntil := consumed for ; batchConsumedUntil < len(sorted) && (maxBatchSize < 1 || batchConsumedUntil-consumed < maxBatchSize) && sorted[batchConsumedUntil].Filesystem == sorted[consumed].Filesystem; batchConsumedUntil++ { } perFS = append(perFS, sorted[consumed:batchConsumedUntil]) consumed = batchConsumedUntil } return perFS } // batch must be on same Filesystem, panics otherwise func tryBatch(ctx context.Context, batch []*DestroySnapOp, d destroyer) error { if len(batch) == 0 { return nil } batchFS := batch[0].Filesystem batchNames := make([]string, len(batch)) for i := range batchNames { batchNames[i] = batch[i].Name if batchFS != batch[i].Filesystem { panic("inconsistent batch") } } batchArg := fmt.Sprintf("%s@%s", batchFS, strings.Join(batchNames, ",")) return d.Destroy(ctx, []string{batchArg}) } // fsbatch must be on same filesystem func doDestroyBatchedRec(ctx context.Context, fsbatch []*DestroySnapOp, d destroyer) { if len(fsbatch) <= 1 { doDestroySeq(ctx, fsbatch, d) return } err := tryBatch(ctx, fsbatch, d) if err == nil { setDestroySnapOpErr(fsbatch, nil) return } if pe, ok := err.(*os.PathError); ok && pe.Err == syscall.E2BIG { // see TestExcessiveArgumentsResultInE2BIG // try halving batch size, assuming snapshots names are roughly the same length debug("batch destroy: E2BIG encountered: %s", err) doDestroyBatchedRec(ctx, fsbatch[0:len(fsbatch)/2], d) doDestroyBatchedRec(ctx, fsbatch[len(fsbatch)/2:], d) return } singleRun := fsbatch // the destroys that will be tried sequentially after "smart" error handling below if err, ok := err.(*DestroySnapshotsError); ok { // eliminate undestroyable datasets from batch and try it once again strippedBatch, remaining := make([]*DestroySnapOp, 0, len(fsbatch)), make([]*DestroySnapOp, 0, len(fsbatch)) for _, b := range fsbatch { isUndestroyable := false for _, undestroyable := range err.Undestroyable { if undestroyable == b.Name { isUndestroyable = true break } } if isUndestroyable { remaining = append(remaining, b) } else { strippedBatch = append(strippedBatch, b) } } err := tryBatch(ctx, strippedBatch, d) if err != nil { // run entire batch sequentially if the stripped one fails // (it shouldn't because we stripped erroneous datasets) singleRun = fsbatch // shadow } else { setDestroySnapOpErr(strippedBatch, nil) // these ones worked singleRun = remaining // shadow } // fallthrough } doDestroySeq(ctx, singleRun, d) } type destroyerImpl struct{} 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)) } // we know that we are only using this for snapshots, so also sanity check for an @ in args[0] if !strings.ContainsAny(args[0], "@") { panic(fmt.Sprintf("sanity check: expecting '@' in call to Destroy, got %q", args[0])) } return ZFSDestroy(ctx, args[0]) } var batchDestroyFeatureCheck struct { once sync.Once enable bool err error } func (d destroyerImpl) DestroySnapshotsCommaSyntaxSupported(ctx context.Context) (bool, error) { batchDestroyFeatureCheck.once.Do(func() { // "feature discovery" 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) batchDestroyFeatureCheck.err = err } def := strings.Contains(string(output), "@[%][,...]") batchDestroyFeatureCheck.enable = envconst.Bool("ZREPL_EXPERIMENTAL_ZFS_COMMA_SYNTAX_SUPPORTED", def) debug("destroy feature check complete %#v", &batchDestroyFeatureCheck) }) return batchDestroyFeatureCheck.enable, batchDestroyFeatureCheck.err }