From b0ab4c4e1efc48f40d74ed3ca04571d24ac5dfdf Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 16 Oct 2026 21:07:51 -0700 Subject: [PATCH] shell: make every -collection flag parse the same way (#11794) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * shell: make every -collection flag parse the same way The filter vocabulary (comma-separated names, wildcards, regexes, "_default" for the empty-named collection) already existed in wildcard.CompileCollectionMatcher, but several commands still did their own exact-match comparison or sent the raw flag value to servers. fs.mergeVolumes used a literal string compare, so -collection=_default silently produced an empty merge plan; it now uses the shared matcher while keeping "*" as the match-all default. volume.fsck and volume.mark filter through the matcher too, and volume.vacuum resolves the filter client-side into per-collection and per-volume requests, since no Collection value scopes the VacuumVolume RPC to the unnamed collection alone. Commands that put one concrete collection name on the wire normalize "_default" to "" (volume.grow, fs.configure, mq.topic.compact, collection.delete, ec.rebuild, ec.balance), and the tier -volumeId commands resolve the volume's actual collection from topology first so patterns and named-collection requests pass the volume server's collection check. * shell: spell the default-collection keyword via CollectionDefault * shell: address review findings on collection filter handling volume.fsck keeps a set of every volume the master knows before any filtering, so a filer entry whose volume was excluded by -collection or -volumeId is skipped instead of reported "volume not found" and deleted by -reallyDeleteFilerEntries. The collection filter also no longer scopes the filer walk to /buckets/: fs.configure can bind a collection to any path prefix, and a narrower walk could mark live chunks as orphans under -reallyDeleteFromVolume. ec.rebuild and ec.balance resolve -collection through the new wildcard.ExpandCollectionFilter, which passes literal names through even when the master does not list them — ec.rebuild must be able to recover a collection whose shards all lost their index and never registered. fs.configure and mq.topic.compact now reject _default outright: an empty Collection means "unset" in both the path-rule merge and the filer's storage-option fallback chain, so the keyword would have been silently ignored. The unnamed-collection vacuum sweep keeps the master's low-disk-space exception: a volume whose every replica is read-only is probed with VacuumVolumeCheck and stays eligible when a replica reports DiskSpaceLow, instead of being dropped outright. --- weed/shell/command_collection_delete.go | 8 +- weed/shell/command_ec_balance.go | 14 +- weed/shell/command_ec_common.go | 33 +++- weed/shell/command_ec_common_test.go | 2 +- weed/shell/command_ec_encode.go | 2 +- weed/shell/command_ec_rebuild.go | 14 +- weed/shell/command_fs_configure.go | 6 + weed/shell/command_fs_merge_volumes.go | 18 +- weed/shell/command_fs_merge_volumes_test.go | 50 ++++- weed/shell/command_mq_topic_compact.go | 8 + weed/shell/command_volume_fsck.go | 32 +++- weed/shell/command_volume_grow.go | 3 +- weed/shell/command_volume_mark.go | 15 +- weed/shell/command_volume_mark_test.go | 2 +- weed/shell/command_volume_tier_compact.go | 33 ++++ weed/shell/command_volume_tier_download.go | 16 +- weed/shell/command_volume_tier_move.go | 2 +- weed/shell/command_volume_tier_upload.go | 24 ++- weed/shell/command_volume_vacuum.go | 187 ++++++++++++++++++- weed/shell/command_volume_vacuum_test.go | 106 ++++++++++- weed/shell/common.go | 10 + weed/util/wildcard/collection_filter.go | 41 ++++ weed/util/wildcard/collection_filter_test.go | 38 +++- 23 files changed, 593 insertions(+), 71 deletions(-) diff --git a/weed/shell/command_collection_delete.go b/weed/shell/command_collection_delete.go index 889da7f99..f80c3d9a5 100644 --- a/weed/shell/command_collection_delete.go +++ b/weed/shell/command_collection_delete.go @@ -35,7 +35,7 @@ func (c *commandCollectionDelete) HasTag(CommandTag) bool { func (c *commandCollectionDelete) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) { colDeleteCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) - collectionName := colDeleteCommand.String("collection", "", "collection to delete. Use '_default_' for the empty-named collection.") + collectionName := colDeleteCommand.String("collection", "", "collection to delete. Use '"+CollectionDefault+"' for the empty-named collection.") applyBalancing := colDeleteCommand.Bool("apply", false, "apply the collection") // TODO: remove this alias applyBalancingAlias := colDeleteCommand.Bool("force", false, "apply the collection (alias for -apply)") @@ -55,9 +55,13 @@ func (c *commandCollectionDelete) Do(args []string, commandEnv *CommandEnv, writ return fmt.Errorf("empty collection name is not allowed") } - if *collectionName == "_default_" { + // The legacy "_default_" spelling stays accepted for compatibility; + // CollectionDefault is the shared keyword every other -collection flag + // uses for the empty-named collection. + if *collectionName == CollectionDefault+"_" { *collectionName = "" } + *collectionName = resolveCollectionName(*collectionName) if !*applyBalancing { fmt.Fprintf(writer, "collection '%s' will be deleted. Use -apply to apply the change.\n", *collectionName) diff --git a/weed/shell/command_ec_balance.go b/weed/shell/command_ec_balance.go index 0ce9127a0..27cf0eaf9 100644 --- a/weed/shell/command_ec_balance.go +++ b/weed/shell/command_ec_balance.go @@ -7,6 +7,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util/wildcard" ) func init() { @@ -42,7 +43,7 @@ func (c *commandEcBalance) HasTag(CommandTag) bool { func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) { balanceCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) - collection := balanceCommand.String("collection", "EACH_COLLECTION", "collection name, or \"EACH_COLLECTION\" for each collection") + collection := balanceCommand.String("collection", string(wildcard.CollectionFilterEach), "comma-separated collection names, wildcards, or regex patterns, \""+CollectionDefault+"\" for the collection with no name, or \""+string(wildcard.CollectionFilterEach)+"\" for each collection") dc := balanceCommand.String("dataCenter", "", "only apply the balancing for this dataCenter") shardReplicaPlacement := balanceCommand.String("shardReplicaPlacement", "", "replica placement for EC shards, or master default if empty") diskTypeStr := balanceCommand.String("diskType", "", "the disk type for EC shards (hdd, ssd, or empty for default hdd)") @@ -71,14 +72,9 @@ func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.W return } - var collections []string - if *collection == "EACH_COLLECTION" { - collections, err = ListCollectionNames(commandEnv, false, true) - if err != nil { - return err - } - } else { - collections = append(collections, *collection) + collections, err := resolveEcCollections(commandEnv, *collection) + if err != nil { + return err } glog.V(1).Infof("balanceEcVolumes collections %+v\n", len(collections)) diff --git a/weed/shell/command_ec_common.go b/weed/shell/command_ec_common.go index 7922f876b..ea7a9fd9e 100644 --- a/weed/shell/command_ec_common.go +++ b/weed/shell/command_ec_common.go @@ -2,6 +2,7 @@ package shell import ( "context" + "fmt" "strings" "time" @@ -142,13 +143,37 @@ func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplic return ec.EcBalance(commandEnv.ecEnv(), collections, dc, ecReplicaPlacement, diskType, maxParallelization, ioBytePerSecond, applyBalancing, excludeNodes, volumeIds, nil) } -// compileCollectionPattern compiles the -collection value shared by the ec and -// tier commands: a comma-separated list of collection names, wildcards, and -// regex patterns. An empty value matches the empty-named collection only, the -// same as CollectionDefault. +// compileCollectionPattern compiles the -collection value shared by commands +// where an empty value means the collection with no name: a comma-separated +// list of collection names, wildcards, and regex patterns, or CollectionDefault +// for the empty-named collection. func compileCollectionPattern(pattern string) (*wildcard.CollectionMatcher, error) { if strings.TrimSpace(pattern) == "" { pattern = CollectionDefault } return wildcard.CompileCollectionMatcher(pattern) } + +// resolveEcCollections expands the ec commands' -collection value into +// collection names. An empty value means the empty-named collection, the same +// as CollectionDefault. Literal names pass through even when the master does +// not list them — the listing only contains collections with registered +// volumes, and ec.rebuild exists to bring back a collection whose volumes all +// lost their index. Wildcard and regex entries resolve against the listing. +func resolveEcCollections(commandEnv *CommandEnv, filter string) (collections []string, err error) { + if strings.TrimSpace(filter) == "" { + filter = CollectionDefault + } + names, err := ListCollectionNames(commandEnv, false, true) + if err != nil { + return nil, err + } + collections, err = wildcard.ExpandCollectionFilter(filter, names) + if err != nil { + return nil, fmt.Errorf("invalid collection pattern %q: %w", filter, err) + } + if len(collections) == 0 { + return nil, fmt.Errorf("no collections match %q", filter) + } + return collections, nil +} diff --git a/weed/shell/command_ec_common_test.go b/weed/shell/command_ec_common_test.go index a44780081..22a0c1a1a 100644 --- a/weed/shell/command_ec_common_test.go +++ b/weed/shell/command_ec_common_test.go @@ -14,7 +14,7 @@ func TestCompileCollectionPattern(t *testing.T) { {pattern: CollectionDefault, matches: []string{""}, misses: []string{"pictures"}}, {pattern: "pictures", matches: []string{"pictures"}, misses: []string{"", "pictures-backup"}}, {pattern: "pictures,videos", matches: []string{"pictures", "videos"}, misses: []string{"", "clips"}}, - {pattern: "pictures,_default", matches: []string{"pictures", ""}, misses: []string{"clips"}}, + {pattern: "pictures," + CollectionDefault, matches: []string{"pictures", ""}, misses: []string{"clips"}}, {pattern: "pictures*", matches: []string{"pictures", "pictures-backup"}, misses: []string{"clips"}}, {pattern: "^pictures", matches: []string{"pictures", "pictures-backup"}, misses: []string{"clips"}}, {pattern: "*", matches: []string{"", "pictures"}}, diff --git a/weed/shell/command_ec_encode.go b/weed/shell/command_ec_encode.go index 7ae07cfcf..0effe470a 100644 --- a/weed/shell/command_ec_encode.go +++ b/weed/shell/command_ec_encode.go @@ -62,7 +62,7 @@ func (c *commandEcEncode) Help() string { - Several collections: ec.encode -collection="mybucket,otherbucket" - Match by prefix: ec.encode -collection="bucket*" - Match all collections: ec.encode -collection="*" - - The empty-named collection: ec.encode -collection="_default" + - The empty-named collection: ec.encode -collection="` + CollectionDefault + `" Options: -verbose: show detailed reasons why volumes are not selected for encoding diff --git a/weed/shell/command_ec_rebuild.go b/weed/shell/command_ec_rebuild.go index 1237e1a16..51f810a56 100644 --- a/weed/shell/command_ec_rebuild.go +++ b/weed/shell/command_ec_rebuild.go @@ -10,6 +10,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/ec" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util/wildcard" ) func init() { @@ -73,7 +74,7 @@ func (c *commandEcRebuild) HasTag(CommandTag) bool { func (c *commandEcRebuild) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) { fixCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) - collection := fixCommand.String("collection", "EACH_COLLECTION", "collection name, or \"EACH_COLLECTION\" for each collection") + collection := fixCommand.String("collection", string(wildcard.CollectionFilterEach), "comma-separated collection names, wildcards, or regex patterns, \""+CollectionDefault+"\" for the collection with no name, or \""+string(wildcard.CollectionFilterEach)+"\" for each collection") volumeIdsStr := fixCommand.String("volumeIds", "", "optional comma-separated list of volume ID to process; defaults to all volumes in the collection") maxParallelization := fixCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible") applyChanges := fixCommand.Bool("apply", false, "apply the changes") @@ -98,14 +99,9 @@ func (c *commandEcRebuild) Do(args []string, commandEnv *CommandEnv, writer io.W return err } - var collections []string - if *collection == "EACH_COLLECTION" { - collections, err = ListCollectionNames(commandEnv, false, true) - if err != nil { - return err - } - } else { - collections = []string{*collection} + collections, err := resolveEcCollections(commandEnv, *collection) + if err != nil { + return err } var volumeIds []needle.VolumeId diff --git a/weed/shell/command_fs_configure.go b/weed/shell/command_fs_configure.go index 3edd2a5b4..29be9b87a 100644 --- a/weed/shell/command_fs_configure.go +++ b/weed/shell/command_fs_configure.go @@ -90,6 +90,12 @@ func (c *commandFsConfigure) Do(args []string, commandEnv *CommandEnv, writer io if *locationPrefix != "" { infoAboutSimulationMode(writer, *apply, "-apply") + // A path rule cannot express "the collection with no name": an empty + // Collection means "unset" and merges as inherit, so the keyword would + // be silently ignored. + if *collection == CollectionDefault { + return fmt.Errorf("path rules cannot select the collection with no name; omit -collection to inherit") + } locConf := &filer_pb.FilerConf_PathConf{ LocationPrefix: *locationPrefix, Collection: *collection, diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 59aceecc2..99b614094 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/weed/shell/command_fs_merge_volumes.go @@ -18,6 +18,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/util/wildcard" "github.com/seaweedfs/seaweedfs/weed/wdclient" "golang.org/x/exp/maps" "google.golang.org/protobuf/proto" @@ -66,7 +67,7 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer dirArg := fsMergeVolumesCommand.String("dir", "/", "base directory to find and update files") fromVolumeArg := fsMergeVolumesCommand.Uint("fromVolumeId", 0, "move chunks with this volume id") toVolumeArg := fsMergeVolumesCommand.String("toVolumeId", "", "change chunks to this volume id, or distribute across a comma-separated list of volume ids") - collectionArg := fsMergeVolumesCommand.String("collection", "*", "Name of collection to merge") + collectionArg := fsMergeVolumesCommand.String("collection", "*", "comma-separated collection names, wildcards, or regex patterns; \""+CollectionDefault+"\" for the collection with no name") apply := fsMergeVolumesCommand.Bool("apply", false, "applying the metadata changes") if err = fsMergeVolumesCommand.Parse(args); err != nil { return err @@ -532,12 +533,17 @@ func parseTargetVolumeIds(arg string) ([]needle.VolumeId, error) { } func (c *commandFsMergeVolumes) createMergePlan(collection string, toVolumeIds []needle.VolumeId, fromVolumeId needle.VolumeId) (*mergePlan, error) { + collectionMatcher, err := compileCollectionPattern(collection) + if err != nil { + return nil, fmt.Errorf("invalid collection pattern %q: %w", collection, err) + } + // When the user names both endpoints, honor that exact direction. The // heuristic below only ever merges a smaller volume into a larger one, so // an explicit "merge larger into smaller" request would otherwise yield an // empty plan and silently do nothing. if fromVolumeId != 0 && len(toVolumeIds) > 0 { - return c.createDirectedMergePlan(collection, fromVolumeId, toVolumeIds) + return c.createDirectedMergePlan(collectionMatcher, collection, fromVolumeId, toVolumeIds) } plan := newMergePlan(c.volumeSizeLimit) @@ -549,7 +555,7 @@ func (c *commandFsMergeVolumes) createMergePlan(collection string, toVolumeIds [ l := len(volumeIds) for i := 0; i < l; i++ { volume := c.volumes[volumeIds[i]] - if volume.GetReadOnly() || c.getVolumeSize(volume) == 0 || (collection != "*" && collection != volume.GetCollection()) { + if volume.GetReadOnly() || c.getVolumeSize(volume) == 0 || !collectionMatcher.Matches(volume.GetCollection()) { if fromVolumeId != 0 && volumeIds[i] == fromVolumeId || slices.Contains(toVolumeIds, volumeIds[i]) { if volume.GetReadOnly() { @@ -607,7 +613,7 @@ func (c *commandFsMergeVolumes) createMergePlan(collection string, toVolumeIds [ // createDirectedMergePlan honors the exact direction the user named, skipping // the heuristic planner's smaller-into-larger ordering. -func (c *commandFsMergeVolumes) createDirectedMergePlan(collection string, from needle.VolumeId, toIds []needle.VolumeId) (*mergePlan, error) { +func (c *commandFsMergeVolumes) createDirectedMergePlan(collectionMatcher *wildcard.CollectionMatcher, collection string, from needle.VolumeId, toIds []needle.VolumeId) (*mergePlan, error) { if slices.Contains(toIds, from) { return nil, fmt.Errorf("no volume id changes, %d is both source and target", from) } @@ -619,8 +625,8 @@ func (c *commandFsMergeVolumes) createDirectedMergePlan(collection string, from if volume.GetReadOnly() { return nil, fmt.Errorf("volume %d is readonly", vid) } - if collection != "*" && collection != volume.GetCollection() { - return nil, fmt.Errorf("volume %d is not in collection %q", vid, collection) + if !collectionMatcher.Matches(volume.GetCollection()) { + return nil, fmt.Errorf("volume %d does not match collection filter %q", vid, collection) } // Merging into an empty target is valid (e.g. a freshly vacuumed // volume); only an empty source has nothing to move. diff --git a/weed/shell/command_fs_merge_volumes_test.go b/weed/shell/command_fs_merge_volumes_test.go index 2fe85923a..d02dd60b7 100644 --- a/weed/shell/command_fs_merge_volumes_test.go +++ b/weed/shell/command_fs_merge_volumes_test.go @@ -108,7 +108,7 @@ func TestCreateMergePlan_DirectedRejectsIneligible(t *testing.T) { from: &master_pb.VolumeInformationMessage{Id: 87, Size: 100, Collection: "other"}, to: &master_pb.VolumeInformationMessage{Id: 83, Size: 100, Collection: "other"}, coll: "wanted", - wantErr: "volume 87 is not in collection", + wantErr: "volume 87 does not match collection filter", }, } for _, tc := range cases { @@ -233,6 +233,54 @@ func TestParseTargetVolumeIds(t *testing.T) { } } +// The collection filter speaks the shared vocabulary: "_default" selects the +// empty-named collection, a name or pattern only its matches, and "*" (or no +// flag) every collection. +func TestCreateMergePlan_CollectionFilter(t *testing.T) { + mb := uint64(1024 * 1024) + mk := func(id uint32, sizeMB uint64, collection string) *master_pb.VolumeInformationMessage { + return &master_pb.VolumeInformationMessage{Id: id, Size: sizeMB * mb, Collection: collection} + } + newCmd := func() *commandFsMergeVolumes { + return newMergeCmd(250000, mk(1, 10, ""), mk(2, 80, ""), mk(3, 10, "x"), mk(4, 80, "x")) + } + + // "_default" merges only the unnamed-collection volumes. + plan, err := newCmd().createMergePlan(CollectionDefault, nil, 0) + if err != nil { + t.Fatalf("%s: %v", CollectionDefault, err) + } + if got := plan.targets[needle.VolumeId(1)]; len(got) != 1 || got[0] != needle.VolumeId(2) { + t.Fatalf("expected 1->2, got %v", plan.targets) + } + if plan.isSource(needle.VolumeId(3)) { + t.Fatalf("named-collection volume must not merge under _default, got %v", plan.targets) + } + + // A literal name still selects its collection. + plan, err = newCmd().createMergePlan("x", nil, 0) + if err != nil { + t.Fatalf("x: %v", err) + } + if got := plan.targets[needle.VolumeId(3)]; len(got) != 1 || got[0] != needle.VolumeId(4) { + t.Fatalf("expected 3->4, got %v", plan.targets) + } + if plan.isSource(needle.VolumeId(1)) { + t.Fatalf("unnamed volume must not merge under x, got %v", plan.targets) + } + + // "*" keeps covering every collection. + plan, err = newCmd().createMergePlan("*", nil, 0) + if err != nil { + t.Fatalf("*: %v", err) + } + for src, dest := range map[needle.VolumeId]needle.VolumeId{1: 2, 3: 4} { + if got := plan.targets[src]; len(got) != 1 || got[0] != dest { + t.Fatalf("expected %d->%d, got %v", src, dest, plan.targets) + } + } +} + // A volume reporting more deleted bytes than it holds must read as empty. func TestGetVolumeSize_ClampsDeletedOverSize(t *testing.T) { c := newMergeCmd(250000) diff --git a/weed/shell/command_mq_topic_compact.go b/weed/shell/command_mq_topic_compact.go index 79d8a45f8..ab3552022 100644 --- a/weed/shell/command_mq_topic_compact.go +++ b/weed/shell/command_mq_topic_compact.go @@ -2,6 +2,7 @@ package shell import ( "flag" + "fmt" "io" "time" @@ -55,6 +56,13 @@ func (c *commandMqTopicCompact) Do(args []string, commandEnv *CommandEnv, writer return err } + // The AssignVolumeRequest collection falls back to the path rule and the + // filer default when empty, so the empty-named collection cannot be + // requested explicitly. + if *collection == CollectionDefault { + return fmt.Errorf("%s is not a usable collection name here; omit -collection to inherit the configured collection", CollectionDefault) + } + storagePreference := &operation.StoragePreference{ Replication: *replication, Collection: *collection, diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index ddbdaa3c5..63981282b 100644 --- a/weed/shell/command_volume_fsck.go +++ b/weed/shell/command_volume_fsck.go @@ -34,6 +34,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/types" "github.com/seaweedfs/seaweedfs/weed/util" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" + "github.com/seaweedfs/seaweedfs/weed/util/wildcard" "golang.org/x/sync/errgroup" ) @@ -55,6 +56,8 @@ type commandVolumeFsck struct { writer io.Writer bucketsPath string collection *string + collectionMatcher *wildcard.CollectionMatcher + knownVolumeIds map[uint32]bool volumeIds map[uint32]bool scopedFilerPath string tempFolder string @@ -109,7 +112,7 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. c.verbose = fsckCommand.Bool("v", false, "verbose mode") c.skipEcVolumes = fsckCommand.Bool("skipEcVolumes", false, "skip erasure coded volumes") c.findMissingChunksInFiler = fsckCommand.Bool("findMissingChunksInFiler", false, "see \"help volume.fsck\"") - c.collection = fsckCommand.String("collection", "", "the collection name") + c.collection = fsckCommand.String("collection", "", "comma-separated collection names, wildcards, or regex patterns; empty or \"*\" matches all, \""+CollectionDefault+"\" the collection with no name") volumeIds := fsckCommand.String("volumeId", "", "comma separated the volume id") applyPurging := fsckCommand.Bool("reallyDeleteFromVolume", false, " after detection, delete missing data from volumes / delete missing file entries from filer. Currently this only works with default filerGroup.") c.forcePurging = fsckCommand.Bool("forcePurging", false, "delete missing data from volumes in one replica used together with applyPurging") @@ -147,6 +150,10 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. c.env = commandEnv c.writer = writer + if c.collectionMatcher, err = wildcard.CompileCollectionMatcher(*c.collection); err != nil { + return fmt.Errorf("invalid collection pattern %q: %w", *c.collection, err) + } + c.bucketsPath, err = readFilerBucketsPath(commandEnv) if err != nil { return fmt.Errorf("read filer buckets path: %w", err) @@ -170,6 +177,16 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. return fmt.Errorf("failed to collect all volume locations: %w", err) } + // Every volume the master knows about, before any filtering: a filer + // entry pointing at a volume that was filtered out of this run is + // excluded, not missing — it must never be reported or purged. + c.knownVolumeIds = make(map[uint32]bool) + for _, vidMap := range dataNodeVolumeIdToVInfo { + for vid := range vidMap { + c.knownVolumeIds[vid] = true + } + } + // If the operator specified -volumeId, reject unknown ids up front instead // of silently filtering them out and reporting "no orphan data". A silent // success on a missing volume looks identical to a clean volume and hides @@ -222,7 +239,7 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. continue } } - if *c.collection != "" && vinfo.collection != *c.collection { + if !c.collectionMatcher.Matches(vinfo.collection) { delete(volumeIdToVInfo, volumeId) continue } @@ -375,6 +392,11 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m continue } } + // A volume the master knows but this run filtered out is not a + // missing volume: its files are healthy and stay untouched. + if c.knownVolumeIds[i.vid] { + continue + } fmt.Fprintf(c.writer, "%d,%x%08x %s volume not found\n", i.vid, i.fileKey, i.cookie, i.path) if purgeAbsent { fmt.Fprintf(c.writer, "deleting path %s after volume not found\n", i.path) @@ -966,7 +988,11 @@ func (c *commandVolumeFsck) getCollectFilerFilePath() string { func (c *commandVolumeFsck) resolveScopedFilerPath(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo) string { if *c.collection != "" { - return fmt.Sprintf("%s/%s", c.bucketsPath, *c.collection) + // fs.configure can bind a collection to any path prefix, so files + // referencing the filtered volumes may live anywhere. A bucket-dir + // walk could miss them and report their live chunks as orphans; + // only a full walk is safe for a collection filter. + return "/" } if len(c.volumeIds) == 0 { return "/" diff --git a/weed/shell/command_volume_grow.go b/weed/shell/command_volume_grow.go index 240506deb..b1f11d869 100644 --- a/weed/shell/command_volume_grow.go +++ b/weed/shell/command_volume_grow.go @@ -39,7 +39,7 @@ func (c *commandGrow) Do(args []string, commandEnv *CommandEnv, writer io.Writer volumeVacuumCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) growCount := volumeVacuumCommand.Uint("count", 2, "") - collection := volumeVacuumCommand.String("collection", "", "grow this collection") + collection := volumeVacuumCommand.String("collection", "", "grow this collection; \""+CollectionDefault+"\" for the collection with no name") dataCenter := volumeVacuumCommand.String("dataCenter", "", "grow volumes only from the specified data center") rack := volumeVacuumCommand.String("rack", "", "grow volumes only from the specified rack") dataNode := volumeVacuumCommand.String("dataNode", "", "grow volumes only from the specified data node") @@ -51,6 +51,7 @@ func (c *commandGrow) Do(args []string, commandEnv *CommandEnv, writer io.Writer if *collection == "" { return fmt.Errorf("collection option is required") } + *collection = resolveCollectionName(*collection) t, _, err := collectTopologyInfo(commandEnv, 0) if err != nil { return err diff --git a/weed/shell/command_volume_mark.go b/weed/shell/command_volume_mark.go index f8c630af8..443a7a943 100644 --- a/weed/shell/command_volume_mark.go +++ b/weed/shell/command_volume_mark.go @@ -9,6 +9,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/util/wildcard" "github.com/seaweedfs/seaweedfs/weed/storage/needle" ) @@ -47,7 +48,7 @@ func (c *commandVolumeMark) Do(args []string, commandEnv *CommandEnv, writer io. volMarkCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) volumeIdInt := volMarkCommand.Int("volumeId", 0, "the volume id") nodeStr := volMarkCommand.String("node", "", "the volume server :") - collection := volMarkCommand.String("collection", "", "the collection name") + collection := volMarkCommand.String("collection", "", "comma-separated collection names, wildcards, or regex patterns; \""+CollectionDefault+"\" for the collection with no name") writable := volMarkCommand.Bool("writable", false, "volume mark writable") readonly := volMarkCommand.Bool("readonly", false, "volume mark readonly") readonlyCanDelete := volMarkCommand.Bool("readonlyCanDelete", false, "volume mark readonly but still accepting deletes") @@ -137,11 +138,9 @@ func collectVolumeMarkTargetsByCollection(topoInfo *master_pb.TopologyInfo, coll if collection == "" { return nil, fmt.Errorf("collection is required") } - - // _default targets volumes that belong to no named collection. - matchCollection := collection - if matchCollection == CollectionDefault { - matchCollection = "" + collectionMatcher, err := wildcard.CompileCollectionMatcher(collection) + if err != nil { + return nil, fmt.Errorf("invalid collection pattern %q: %w", collection, err) } var targets []volumeMarkTarget @@ -152,7 +151,7 @@ func collectVolumeMarkTargetsByCollection(topoInfo *master_pb.TopologyInfo, coll sourceVolumeServer := pb.NewServerAddressFromDataNode(dn) for _, diskInfo := range dn.GetDiskInfos() { for _, v := range diskInfo.GetVolumeInfos() { - if v.GetCollection() == matchCollection { + if collectionMatcher.Matches(v.GetCollection()) { targets = append(targets, volumeMarkTarget{ volumeId: needle.VolumeId(v.GetId()), sourceVolumeServer: sourceVolumeServer, @@ -162,7 +161,7 @@ func collectVolumeMarkTargetsByCollection(topoInfo *master_pb.TopologyInfo, coll } }) if len(targets) == 0 { - return nil, fmt.Errorf("collection %s has no volumes", collection) + return nil, fmt.Errorf("no volumes match collection %s", collection) } return targets, nil } diff --git a/weed/shell/command_volume_mark_test.go b/weed/shell/command_volume_mark_test.go index 8251ca926..47a19cbb3 100644 --- a/weed/shell/command_volume_mark_test.go +++ b/weed/shell/command_volume_mark_test.go @@ -122,7 +122,7 @@ func TestCollectVolumeMarkTargetsByCollectionRejectsEmptyCollection(t *testing.T func TestCollectVolumeMarkTargetsByCollectionReportsMissingCollection(t *testing.T) { _, err := collectVolumeMarkTargetsByCollection(&master_pb.TopologyInfo{}, "missing") - if err == nil || !strings.Contains(err.Error(), "collection missing has no volumes") { + if err == nil || !strings.Contains(err.Error(), "no volumes match collection missing") { t.Fatalf("expected missing collection error, got %v", err) } } diff --git a/weed/shell/command_volume_tier_compact.go b/weed/shell/command_volume_tier_compact.go index 80c3d2865..282b43734 100644 --- a/weed/shell/command_volume_tier_compact.go +++ b/weed/shell/command_volume_tier_compact.go @@ -126,6 +126,39 @@ func (c *commandVolumeTierCompact) Do(args []string, commandEnv *CommandEnv, wri return nil } +// findVolumeCollection resolves the collection a volume actually lives in, +// honoring the -collection filter: an empty pattern matches any collection, a +// non-empty one compiles like the rest of the tier commands. The volume server +// verifies the collection name on tier requests, so callers need the real +// name, not the pattern. +func findVolumeCollection(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, collectionPattern string) (collection string, found bool, err error) { + var matchesCollection func(string) bool + if collectionPattern != "" { + collectionMatcher, compileErr := compileCollectionPattern(collectionPattern) + if compileErr != nil { + return "", false, fmt.Errorf("invalid collection pattern '%s': %v", collectionPattern, compileErr) + } + matchesCollection = collectionMatcher.Matches + } else { + matchesCollection = func(string) bool { return true } + } + + eachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) { + if found { + return + } + for _, diskInfo := range dn.DiskInfos { + for _, v := range diskInfo.VolumeInfos { + if needle.VolumeId(v.Id) == vid && matchesCollection(v.Collection) { + collection, found = v.Collection, true + return + } + } + } + }) + return +} + func findRemoteVolumeInTopology(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, collectionPattern string) (remoteVolumeInfo, bool, error) { var matchesCollection func(string) bool if collectionPattern != "" { diff --git a/weed/shell/command_volume_tier_download.go b/weed/shell/command_volume_tier_download.go index 79668744c..12f4fead3 100644 --- a/weed/shell/command_volume_tier_download.go +++ b/weed/shell/command_volume_tier_download.go @@ -80,7 +80,7 @@ func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, wr // volumeId is provided if vid != 0 { - return doVolumeTierDownload(commandEnv, writer, *collection, vid, *concurrency) + return doVolumeTierDownload(commandEnv, writer, topologyInfo, *collection, vid, *concurrency) } // apply to all volumes in the collection @@ -91,7 +91,7 @@ func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, wr } fmt.Printf("tier download volumes: %v\n", volumeIds) for _, vid := range volumeIds { - if err = doVolumeTierDownload(commandEnv, writer, *collection, vid, *concurrency); err != nil { + if err = doVolumeTierDownload(commandEnv, writer, topologyInfo, *collection, vid, *concurrency); err != nil { return err } } @@ -124,13 +124,23 @@ func collectRemoteVolumes(topoInfo *master_pb.TopologyInfo, collectionPattern st return } -func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId, concurrency int) (err error) { +func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, topoInfo *master_pb.TopologyInfo, collectionPattern string, vid needle.VolumeId, concurrency int) (err error) { // find volume location locations, found := commandEnv.MasterClient.GetLocationsClone(uint32(vid)) if !found { return fmt.Errorf("volume %d not found", vid) } + // The volume server verifies the request's collection against the volume's + // own, so the pattern must resolve to the actual collection here. + collection, found, err := findVolumeCollection(topoInfo, vid, collectionPattern) + if err != nil { + return err + } + if !found { + return fmt.Errorf("volume %d not found in collections matching %q", vid, collectionPattern) + } + // All replicas point at the same remote object; only the final download may delete // it. Every earlier replica keeps it so the survivors are not left dangling. // TODO parallelize this diff --git a/weed/shell/command_volume_tier_move.go b/weed/shell/command_volume_tier_move.go index 0d0ac97be..11a8481a9 100644 --- a/weed/shell/command_volume_tier_move.go +++ b/weed/shell/command_volume_tier_move.go @@ -67,7 +67,7 @@ func (c *commandVolumeTierMove) Help() string { Note: -collectionPattern is a comma-separated list of collection names, with "*" and "?" - wildcards, and regex patterns. Use "_default" to match only the default collection + wildcards, and regex patterns. Use "` + CollectionDefault + `" to match only the default collection (volumes with no collection name). An empty pattern matches all collections. ` diff --git a/weed/shell/command_volume_tier_upload.go b/weed/shell/command_volume_tier_upload.go index 304676d8c..962d55f47 100644 --- a/weed/shell/command_volume_tier_upload.go +++ b/weed/shell/command_volume_tier_upload.go @@ -89,7 +89,7 @@ func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writ tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) volumeId := tierCommand.Int("volumeId", 0, "the volume id") - collection := tierCommand.String("collection", "", "the collection name") + collection := tierCommand.String("collection", "", "comma-separated collection names, wildcards, or regex patterns; empty matches the collection with no name") fullPercentage := tierCommand.Float64("fullPercent", 95, "the volume reaches the percentage of max volume size") quietPeriod := tierCommand.Duration("quietFor", 24*time.Hour, "select volumes without no writes for this period") dest := tierCommand.String("dest", "", "the target tier name") @@ -137,20 +137,30 @@ func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writ return nil } -func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId, dest string, keepLocalDatFile bool, concurrency int) (err error) { +func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collectionPattern string, vid needle.VolumeId, dest string, keepLocalDatFile bool, concurrency int) (err error) { // find volume location topoInfo, _, err := collectTopologyInfo(commandEnv, 0) if err != nil { return fmt.Errorf("collect topology info: %v", err) } + // The volume server verifies the request's collection against the volume's + // own, so the pattern must resolve to the actual collection here. + collection, found, err := findVolumeCollection(topoInfo, vid, collectionPattern) + if err != nil { + return err + } + if !found { + if collectionPattern == "" { + return fmt.Errorf("volume %d not found", vid) + } + return fmt.Errorf("volume %d not found in collections matching %s", vid, collectionPattern) + } + existingLocations := collectVolumeTierUploadLocations(topoInfo, vid, collection, writer) if len(existingLocations) == 0 { - if collection == "" { - return fmt.Errorf("volume %d not found", vid) - } - return fmt.Errorf("volume %d not found in collection %s", vid, collection) + return fmt.Errorf("volume %d has no replicas in collection %s", vid, collection) } _, err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, existingLocations, false, false) @@ -192,7 +202,7 @@ func collectVolumeTierUploadLocations(topoInfo *master_pb.TopologyInfo, vid need eachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) { for _, disk := range dn.DiskInfos { for _, vi := range disk.VolumeInfos { - if needle.VolumeId(vi.Id) == vid && (collection == "" || vi.Collection == collection) { + if needle.VolumeId(vi.Id) == vid && vi.Collection == collection { fmt.Fprintf(writer, "find volume %d from Url:%s, GrpcPort:%d, DC:%s\n", vid, dn.Id, dn.GrpcPort, string(dc)) loc := wdclient.Location{ Url: dn.Id, diff --git a/weed/shell/command_volume_vacuum.go b/weed/shell/command_volume_vacuum.go index a3531d35b..c1ee55d0c 100644 --- a/weed/shell/command_volume_vacuum.go +++ b/weed/shell/command_volume_vacuum.go @@ -5,11 +5,18 @@ import ( "flag" "fmt" "io" + "slices" "sort" "strconv" "strings" + "time" + "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/util/wildcard" + "google.golang.org/grpc" ) func init() { @@ -42,12 +49,17 @@ func (c *commandVacuum) Do(args []string, commandEnv *CommandEnv, writer io.Writ volumeVacuumCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) garbageThreshold := volumeVacuumCommand.Float64("garbageThreshold", 0.3, "vacuum when garbage is more than this limit") - collection := volumeVacuumCommand.String("collection", "", "vacuum this collection") + collection := volumeVacuumCommand.String("collection", "", "comma-separated collection names, wildcards, or regex patterns; empty or \"*\" matches all, \""+CollectionDefault+"\" the collection with no name") volumeIds := volumeVacuumCommand.String("volumeId", "", "comma-separated list of volume IDs") if err = volumeVacuumCommand.Parse(args); err != nil { return nil } + collectionMatcher, err := wildcard.CompileCollectionMatcher(*collection) + if err != nil { + return fmt.Errorf("invalid collection pattern %q: %w", *collection, err) + } + if err = commandEnv.confirmIsLocked(args); err != nil { return } @@ -107,19 +119,41 @@ func (c *commandVacuum) Do(args []string, commandEnv *CommandEnv, writer io.Writ } else if topo != nil { // The sweep says nothing about the volumes it leaves alone, so an // operator on a full disk sees the command return and nothing change. - if skipped := readOnlyVolumesAboveThreshold(topo, *collection, *garbageThreshold); len(skipped) > 0 { + if skipped := readOnlyVolumesAboveThreshold(topo, collectionMatcher, *garbageThreshold); len(skipped) > 0 { fmt.Fprintf(writer, "%d read-only volume(s) hold garbage above %g and are skipped by the sweep: %v\n", len(skipped), *garbageThreshold, skipped) fmt.Fprintf(writer, "vacuum them explicitly with -volumeId\n") } } - for _, volumeId := range volumeIdInts { - err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { - _, err = client.VacuumVolume(context.Background(), &master_pb.VacuumVolumeRequest{ - GarbageThreshold: float32(*garbageThreshold), - VolumeId: volumeId, - Collection: *collection, + // The VacuumVolume RPC takes one exact collection name per call and reads + // "" as "all collections", so a filter has to be resolved here: one request + // per matched named collection, and per-volume requests for the unnamed + // collection, which no Collection value selects on its own. + var requests []*master_pb.VacuumVolumeRequest + if collectionMatcher == nil { + for _, volumeId := range volumeIdInts { + requests = append(requests, &master_pb.VacuumVolumeRequest{ + VolumeId: volumeId, }) + } + } else { + if topo == nil { + return fmt.Errorf("cannot resolve collection pattern %q: topology unavailable", *collection) + } + if *volumeIds != "" { + requests, err = vacuumRequestsForVolumeIds(topo, collectionMatcher, volumeIdInts) + } else { + requests, err = vacuumRequestsForPattern(commandEnv.option.GrpcDialOption, writer, topo, collectionMatcher) + } + if err != nil { + return err + } + } + + for _, req := range requests { + req.GarbageThreshold = float32(*garbageThreshold) + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { + _, err = client.VacuumVolume(context.Background(), req) return err }) if err != nil { @@ -130,6 +164,139 @@ func (c *commandVacuum) Do(args []string, commandEnv *CommandEnv, writer io.Writ return nil } +// vidReplica pairs a volume replica's read-only state with the server that +// reports it, so the unnamed-collection sweep can reproduce the master's +// low-disk-space exception client-side. +type vidReplica struct { + server pb.ServerAddress + readOnly bool +} + +// vacuumRequestsForPattern resolves a collection filter to the sweep requests +// that cover it: one per matched named collection, which keeps the server's +// read-only skip, plus per-volume requests for the unnamed collection — no +// Collection value scopes the RPC to it alone. +func vacuumRequestsForPattern(grpcDialOption grpc.DialOption, writer io.Writer, topo *master_pb.TopologyInfo, collectionMatcher *wildcard.CollectionMatcher) ([]*master_pb.VacuumVolumeRequest, error) { + collections := make(map[string]bool) + defaultReplicas := make(map[uint32][]vidReplica) + eachDataNode(topo, func(_ DataCenterId, _ RackId, dn *master_pb.DataNodeInfo) { + for _, disk := range dn.DiskInfos { + for _, v := range disk.VolumeInfos { + if !collectionMatcher.Matches(v.Collection) { + continue + } + collections[v.Collection] = true + if v.Collection == "" { + defaultReplicas[v.Id] = append(defaultReplicas[v.Id], vidReplica{ + server: pb.NewServerAddressFromDataNode(dn), + readOnly: v.ReadOnly, + }) + } + } + } + }) + if len(collections) == 0 { + return nil, fmt.Errorf("no volumes found in collections matching the filter") + } + + var requests []*master_pb.VacuumVolumeRequest + for name := range collections { + if name != "" { + requests = append(requests, &master_pb.VacuumVolumeRequest{Collection: name}) + } + } + // The unnamed collection has no scoping name, so each of its volumes gets + // a per-volume request — which also vacuums any other collection sharing + // the id. A volume whose every replica is read-only is skipped unless a + // replica reports low disk space, the sweep's own exception. + var defaultVids []uint32 + for vid, replicas := range defaultReplicas { + writable := false + for _, r := range replicas { + if !r.readOnly { + writable = true + break + } + } + if !writable && !anyReplicaDiskSpaceLow(grpcDialOption, replicas, vid, writer) { + continue + } + defaultVids = append(defaultVids, vid) + } + sort.Slice(defaultVids, func(i, j int) bool { return defaultVids[i] < defaultVids[j] }) + for _, vid := range defaultVids { + requests = append(requests, &master_pb.VacuumVolumeRequest{VolumeId: vid}) + } + return requests, nil +} + +// anyReplicaDiskSpaceLow probes read-only replicas for the low-disk condition +// under which the master's sweep still vacuums a read-only volume. +func anyReplicaDiskSpaceLow(grpcDialOption grpc.DialOption, replicas []vidReplica, vid uint32, writer io.Writer) bool { + for _, r := range replicas { + diskSpaceLow := false + err := operation.WithVolumeServerClient(false, r.server, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + resp, checkErr := client.VacuumVolumeCheck(ctx, &volume_server_pb.VacuumVolumeCheckRequest{VolumeId: vid}) + if checkErr != nil { + return checkErr + } + diskSpaceLow = resp.DiskSpaceLow + return nil + }) + if err != nil { + fmt.Fprintf(writer, "volume %d on %s: cannot check disk space: %v\n", vid, r.server, err) + continue + } + if diskSpaceLow { + return true + } + } + return false +} + +// vacuumRequestsForVolumeIds scopes each requested volume to the collections +// it lives in that the filter matches, so -volumeId and a pattern narrow each +// other the same way the server narrows an exact -collection name. +func vacuumRequestsForVolumeIds(topo *master_pb.TopologyInfo, collectionMatcher *wildcard.CollectionMatcher, volumeIdInts []uint32) ([]*master_pb.VacuumVolumeRequest, error) { + collectionsByVid := make(map[uint32]map[string]bool) + eachDataNode(topo, func(_ DataCenterId, _ RackId, dn *master_pb.DataNodeInfo) { + for _, disk := range dn.DiskInfos { + for _, v := range disk.VolumeInfos { + if collectionsByVid[v.Id] == nil { + collectionsByVid[v.Id] = make(map[string]bool) + } + collectionsByVid[v.Id][v.Collection] = true + } + } + }) + + var requests []*master_pb.VacuumVolumeRequest + for _, vid := range volumeIdInts { + var matched []string + for name := range collectionsByVid[vid] { + if collectionMatcher.Matches(name) { + matched = append(matched, name) + } + } + if len(matched) == 0 { + return nil, fmt.Errorf("volume %d is not in a collection matching the filter", vid) + } + // The unnamed collection has no scoping name; a volume-id-only request + // covers it (and any other collection the id is shared into). + if slices.Contains(matched, "") { + requests = append(requests, &master_pb.VacuumVolumeRequest{VolumeId: vid}) + continue + } + sort.Strings(matched) + for _, name := range matched { + requests = append(requests, &master_pb.VacuumVolumeRequest{VolumeId: vid, Collection: name}) + } + } + return requests, nil +} + // readOnlyVolumesAboveThreshold lists the volumes a sweep leaves alone: any // replica read-only (that is what the sweep checks), in the collection when one // is given, and some replica with a garbage ratio at or above the threshold. @@ -138,13 +305,13 @@ func (c *commandVacuum) Do(args []string, commandEnv *CommandEnv, writer io.Writ // it can only understate, and a converted index that reports deletes without // their sizes is left out because its ratio is not knowable here. This is a // hint; the volume server's own check decides. -func readOnlyVolumesAboveThreshold(topo *master_pb.TopologyInfo, collection string, garbageThreshold float64) []uint32 { +func readOnlyVolumesAboveThreshold(topo *master_pb.TopologyInfo, collectionMatcher *wildcard.CollectionMatcher, garbageThreshold float64) []uint32 { readOnly := make(map[uint32]bool) garbage := make(map[uint32]float64) // the highest ratio any replica reports eachDataNode(topo, func(_ DataCenterId, _ RackId, dn *master_pb.DataNodeInfo) { for _, disk := range dn.DiskInfos { for _, v := range disk.VolumeInfos { - if collection != "" && v.Collection != collection { + if !collectionMatcher.Matches(v.Collection) { continue } if v.ReadOnly { diff --git a/weed/shell/command_volume_vacuum_test.go b/weed/shell/command_volume_vacuum_test.go index 3b32d21b8..dba71cf93 100644 --- a/weed/shell/command_volume_vacuum_test.go +++ b/weed/shell/command_volume_vacuum_test.go @@ -1,10 +1,12 @@ package shell import ( + "io" "reflect" "testing" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/util/wildcard" ) func vacuumTestTopology(nodes ...*master_pb.DataNodeInfo) *master_pb.TopologyInfo { @@ -42,17 +44,115 @@ func TestReadOnlyVolumesAboveThreshold(t *testing.T) { vacuumTestNode(readOnlyHalfGarbage, readOnlyOtherCollection, mixedWritableReplica), ) - got := readOnlyVolumesAboveThreshold(topo, "a", 0.3) + matcherA, err := wildcard.CompileCollectionMatcher("a") + if err != nil { + t.Fatalf("compile matcher: %v", err) + } + + got := readOnlyVolumesAboveThreshold(topo, matcherA, 0.3) if want := []uint32{1, 6, 7}; !reflect.DeepEqual(got, want) { t.Fatalf("collection a, threshold 0.3: got %v, want %v", got, want) } - got = readOnlyVolumesAboveThreshold(topo, "", 0.3) + got = readOnlyVolumesAboveThreshold(topo, nil, 0.3) if want := []uint32{1, 4, 6, 7}; !reflect.DeepEqual(got, want) { t.Fatalf("all collections, threshold 0.3: got %v, want %v", got, want) } - if got := readOnlyVolumesAboveThreshold(topo, "a", 0.95); len(got) != 0 { + if got := readOnlyVolumesAboveThreshold(topo, matcherA, 0.95); len(got) != 0 { t.Fatalf("threshold 0.95: got %v, want none", got) } } + +func mustMatcher(t *testing.T, filter string) *wildcard.CollectionMatcher { + t.Helper() + m, err := wildcard.CompileCollectionMatcher(filter) + if err != nil { + t.Fatalf("compile %q: %v", filter, err) + } + return m +} + +func TestVacuumRequestsForPattern(t *testing.T) { + node := vacuumTestNode( + &master_pb.VolumeInformationMessage{Id: 1, Collection: "a", Size: 100}, + &master_pb.VolumeInformationMessage{Id: 2, Collection: "", Size: 100}, + // read-only volumes stay out of the unnamed-collection sweep, the + // same as the server skips them on a named-collection sweep — the + // unreachable address keeps the low-disk probe a fast miss + &master_pb.VolumeInformationMessage{Id: 3, Collection: "", Size: 100, ReadOnly: true}, + &master_pb.VolumeInformationMessage{Id: 4, Collection: "b", Size: 100}, + ) + node.Id = "localhost:1" + topo := vacuumTestTopology(node) + + // A named collection sweeps as a single collection-scoped request. + reqs, err := vacuumRequestsForPattern(nil, io.Discard, topo, mustMatcher(t, "a")) + if err != nil { + t.Fatalf("pattern a: %v", err) + } + if len(reqs) != 1 || reqs[0].Collection != "a" || reqs[0].VolumeId != 0 { + t.Fatalf("pattern a: got %v, want one collection-a sweep request", reqs) + } + + // Patterns cover several named collections. + reqs, err = vacuumRequestsForPattern(nil, io.Discard, topo, mustMatcher(t, "a,b")) + if err != nil { + t.Fatalf("pattern a,b: %v", err) + } + names := map[string]bool{} + for _, r := range reqs { + names[r.Collection] = true + } + if len(reqs) != 2 || !names["a"] || !names["b"] { + t.Fatalf("pattern a,b: got %v, want sweeps for a and b", reqs) + } + + // The unnamed collection resolves to per-volume requests — no Collection + // value selects it alone, and the read-only volume is skipped because the + // probe cannot confirm low disk space. + reqs, err = vacuumRequestsForPattern(nil, io.Discard, topo, mustMatcher(t, CollectionDefault)) + if err != nil { + t.Fatalf("%s: %v", CollectionDefault, err) + } + if len(reqs) != 1 || reqs[0].VolumeId != 2 || reqs[0].Collection != "" { + t.Fatalf("%s: got %v, want one per-volume request for volume 2", CollectionDefault, reqs) + } + + // A filter hitting nothing is an error, not a silent no-op. + if _, err = vacuumRequestsForPattern(nil, io.Discard, topo, mustMatcher(t, "missing")); err == nil { + t.Fatal("expected an error for a filter matching no collection") + } +} + +func TestVacuumRequestsForVolumeIds(t *testing.T) { + topo := vacuumTestTopology( + vacuumTestNode( + &master_pb.VolumeInformationMessage{Id: 1, Collection: "a"}, + &master_pb.VolumeInformationMessage{Id: 2, Collection: ""}, + ), + ) + + // A matching name scopes the request to that collection. + reqs, err := vacuumRequestsForVolumeIds(topo, mustMatcher(t, "a"), []uint32{1}) + if err != nil { + t.Fatalf("vid 1 in a: %v", err) + } + if len(reqs) != 1 || reqs[0].VolumeId != 1 || reqs[0].Collection != "a" { + t.Fatalf("vid 1 in a: got %v, want {VolumeId:1, Collection:a}", reqs) + } + + // The unnamed collection sends a volume-id-only request. + reqs, err = vacuumRequestsForVolumeIds(topo, mustMatcher(t, CollectionDefault), []uint32{2}) + if err != nil { + t.Fatalf("vid 2 in default: %v", err) + } + if len(reqs) != 1 || reqs[0].VolumeId != 2 || reqs[0].Collection != "" { + t.Fatalf("vid 2 in default: got %v, want {VolumeId:2}", reqs) + } + + // A volume outside the filter is an error. + if _, err = vacuumRequestsForVolumeIds(topo, mustMatcher(t, "a"), []uint32{2}); err == nil { + t.Fatal("expected an error for a volume outside the filter") + } +} diff --git a/weed/shell/common.go b/weed/shell/common.go index 7ba5f22b5..76d0b28ad 100644 --- a/weed/shell/common.go +++ b/weed/shell/common.go @@ -14,6 +14,16 @@ var ( CollectionDefault = wildcard.CollectionDefault ) +// resolveCollectionName maps the shared "_default" keyword to the empty-named +// collection for commands whose -collection flag names one collection rather +// than filtering by a pattern. +func resolveCollectionName(name string) string { + if name == CollectionDefault { + return "" + } + return name +} + // ErrorWaitGroup lives in weed/util so packages outside the shell can share it. type ErrorWaitGroup = util.ErrorWaitGroup type ErrorWaitGroupTask = util.ErrorWaitGroupTask diff --git a/weed/util/wildcard/collection_filter.go b/weed/util/wildcard/collection_filter.go index 7a00043f6..ad6434666 100644 --- a/weed/util/wildcard/collection_filter.go +++ b/weed/util/wildcard/collection_filter.go @@ -211,6 +211,47 @@ func isRegexEntry(entry string) bool { return false } +// ExpandCollectionFilter resolves a collection_filter into concrete names +// drawn from candidates — the same names CompileCollectionMatcher would +// match — with one difference: a literal entry (no wildcard or regex syntax) +// is returned even when absent from candidates, since a command may target a +// collection the listing does not contain yet. CollectionDefault resolves to +// the empty-named collection. Empty, "*", and the mode sentinels return all +// candidates. +func ExpandCollectionFilter(filter string, candidates []string) ([]string, error) { + trimmed := strings.TrimSpace(filter) + mode := CollectionFilterMode(trimmed) + if trimmed == "" || trimmed == "*" || mode == CollectionFilterAll || mode == CollectionFilterEach { + return candidates, nil + } + matcher, err := CompileCollectionMatcher(trimmed) + if err != nil { + return nil, err + } + seen := make(map[string]bool, len(candidates)) + var out []string + add := func(name string) { + if !seen[name] { + seen[name] = true + out = append(out, name) + } + } + for _, entry := range splitCollectionEntries(trimmed) { + entry = strings.TrimSpace(entry) + if entry == CollectionDefault { + add("") + } else if !isRegexEntry(entry) && !strings.ContainsAny(entry, "*?") { + add(entry) + } + } + for _, name := range candidates { + if matcher.Matches(name) { + add(name) + } + } + return out, nil +} + // Matches reports whether a collection passes the filter. A nil matcher accepts everything. func (m *CollectionMatcher) Matches(collection string) bool { if m == nil { diff --git a/weed/util/wildcard/collection_filter_test.go b/weed/util/wildcard/collection_filter_test.go index d11401924..d76cb41d9 100644 --- a/weed/util/wildcard/collection_filter_test.go +++ b/weed/util/wildcard/collection_filter_test.go @@ -1,6 +1,9 @@ package wildcard -import "testing" +import ( + "reflect" + "testing" +) func TestCompileCollectionMatcher(t *testing.T) { cases := []struct { @@ -72,3 +75,36 @@ func TestCompileCollectionMatcherInvalidEntry(t *testing.T) { } } } + +func TestExpandCollectionFilter(t *testing.T) { + listed := []string{"photos", "photos-backup", "videos"} + cases := []struct { + filter string + want []string + }{ + // Match-all forms return the candidates. + {"", listed}, + {"*", listed}, + {string(CollectionFilterEach), listed}, + // A literal name passes through even when unlisted — the listing only + // covers collections with registered volumes. + {"photos", []string{"photos"}}, + {"archived", []string{"archived"}}, + {CollectionDefault, []string{""}}, + {"photos," + CollectionDefault, []string{"photos", ""}}, + // Patterns resolve only against listed names; a literal in the same + // list still passes through. + {"photos*", []string{"photos", "photos-backup"}}, + {"photos*,archived", []string{"archived", "photos", "photos-backup"}}, + {"^videos$", []string{"videos"}}, + } + for _, c := range cases { + got, err := ExpandCollectionFilter(c.filter, listed) + if err != nil { + t.Fatalf("ExpandCollectionFilter(%q): %v", c.filter, err) + } + if !reflect.DeepEqual(got, c.want) { + t.Errorf("ExpandCollectionFilter(%q) = %v, want %v", c.filter, got, c.want) + } + } +}