From e482e67971c91e60c3e2e5b219fec146e9d994bc Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 25 Aug 2026 15:13:06 -0700 Subject: [PATCH] admin: accept a list of collections in the task collection filter (#10953) The collection filter was parsed twice with two syntaxes: the master-side volume listing compiled the whole string as one regex, while EC encode and EC balance detection split it on commas and matched each entry as a wildcard. A volume had to pass both, so "collection-a,collection-b" matched nothing (no collection is named that), and the ALL_COLLECTIONS sentinel, which the master side skips, dropped every volume at the task side. Parse it once, in one place: a comma-separated list where an entry is a name with optional * and ? wildcards, or a regex when it carries regex syntax. A regex entry now has to match the whole name unless it anchors itself, so listing a collection no longer picks up its longer namesakes. --- weed/command/scaffold/admin.toml | 3 +- weed/plugin/worker/collection_filter.go | 79 ++++++++++++++++++- weed/plugin/worker/collection_filter_test.go | 48 +++++++++++ weed/plugin/worker/volume_metrics.go | 15 +--- weed/plugin/worker/volume_metrics_test.go | 20 +++++ weed/worker/tasks/balance/plugin_handler.go | 2 +- weed/worker/tasks/ec_balance/config.go | 6 +- weed/worker/tasks/ec_balance/detection.go | 13 ++- .../worker/tasks/ec_balance/detection_test.go | 11 ++- .../ec_balance/multidisk_detection_test.go | 6 +- .../worker/tasks/ec_balance/plugin_handler.go | 2 +- weed/worker/tasks/erasure_coding/config.go | 6 +- weed/worker/tasks/erasure_coding/detection.go | 9 ++- .../detection_disk_type_test.go | 4 +- .../tasks/erasure_coding/detection_test.go | 47 +++++++++-- .../tasks/erasure_coding/plugin_handler.go | 10 ++- weed/worker/tasks/vacuum/plugin_handler.go | 2 +- 17 files changed, 235 insertions(+), 48 deletions(-) create mode 100644 weed/plugin/worker/collection_filter_test.go diff --git a/weed/command/scaffold/admin.toml b/weed/command/scaffold/admin.toml index c4046faea..8f947baf6 100644 --- a/weed/command/scaffold/admin.toml +++ b/weed/command/scaffold/admin.toml @@ -57,7 +57,8 @@ # scan_interval_seconds = 3600 # max_concurrent = 1 # min_size_mb = 30 -# only process volumes from this collection ("" = all collections) +# only process volumes from these collections, comma separated, "*" and "?" wildcards +# and regex patterns allowed ("" = all collections) # collection_filter = "" # disk tags preferred for shard placement # preferred_tags = ["fast", "ssd"] diff --git a/weed/plugin/worker/collection_filter.go b/weed/plugin/worker/collection_filter.go index 7721858c7..999af305b 100644 --- a/weed/plugin/worker/collection_filter.go +++ b/weed/plugin/worker/collection_filter.go @@ -1,14 +1,89 @@ package pluginworker +import ( + "fmt" + "regexp" + "strings" + + "github.com/seaweedfs/seaweedfs/weed/util/wildcard" +) + // CollectionFilterMode controls how collections are interpreted during -// detection. The two recognized sentinels short-circuit regex matching: +// detection. The two recognized sentinels short-circuit matching: // - CollectionFilterAll: pool every collection together (default). // - CollectionFilterEach: run detection separately per collection. // -// Any other non-empty value is treated as a regex. +// Any other non-empty value is a comma-separated list of collection patterns. type CollectionFilterMode string const ( CollectionFilterAll CollectionFilterMode = "ALL_COLLECTIONS" CollectionFilterEach CollectionFilterMode = "EACH_COLLECTION" ) + +// regexMetaCharacters mark a filter entry as a regex; "*" and "?" stay wildcards. +const regexMetaCharacters = `.^$+()[]{}|\` + +// CollectionMatcher matches a volume collection against a collection_filter value. +type CollectionMatcher struct { + wildcards []string + regexes []*regexp.Regexp +} + +// CompileCollectionMatcher parses a collection_filter into a matcher. A nil +// matcher accepts every collection: that is the empty filter, "*", and both +// mode sentinels. Any other value is a comma-separated list, and a collection +// passes when one entry matches it. An entry is a name, optionally with "*" and +// "?" wildcards, or a regex when it carries regex syntax. A regex entry must +// match the whole name unless it anchors itself with "^" or "$". +func CompileCollectionMatcher(filter string) (*CollectionMatcher, error) { + trimmed := strings.TrimSpace(filter) + mode := CollectionFilterMode(trimmed) + if trimmed == "" || trimmed == "*" || mode == CollectionFilterAll || mode == CollectionFilterEach { + return nil, nil + } + + matcher := &CollectionMatcher{} + for _, entry := range strings.Split(trimmed, ",") { + entry = strings.TrimSpace(entry) + if entry == "" { + continue + } + if !strings.ContainsAny(entry, regexMetaCharacters) { + matcher.wildcards = append(matcher.wildcards, entry) + continue + } + pattern := entry + if !strings.ContainsAny(entry, "^$") { + pattern = "^(?:" + entry + ")$" + } + compiled, err := regexp.Compile(pattern) + if err != nil { + return nil, fmt.Errorf("invalid collection_filter entry %q: %w", entry, err) + } + matcher.regexes = append(matcher.regexes, compiled) + } + + if len(matcher.wildcards) == 0 && len(matcher.regexes) == 0 { + return nil, nil + } + return matcher, nil +} + +// Matches reports whether a collection passes the filter. A nil matcher accepts everything. +func (m *CollectionMatcher) Matches(collection string) bool { + if m == nil { + return true + } + for _, pattern := range m.wildcards { + if wildcard.MatchesWildcard(pattern, collection) { + return true + } + } + for _, re := range m.regexes { + if re.MatchString(collection) { + return true + } + } + return false +} diff --git a/weed/plugin/worker/collection_filter_test.go b/weed/plugin/worker/collection_filter_test.go new file mode 100644 index 000000000..471358301 --- /dev/null +++ b/weed/plugin/worker/collection_filter_test.go @@ -0,0 +1,48 @@ +package pluginworker + +import "testing" + +func TestCompileCollectionMatcher(t *testing.T) { + cases := []struct { + filter string + matches []string + misses []string + }{ + {filter: "", matches: []string{"", "photos"}}, + {filter: "*", matches: []string{"", "photos"}}, + {filter: string(CollectionFilterAll), matches: []string{"", "photos"}}, + {filter: string(CollectionFilterEach), matches: []string{"", "photos"}}, + {filter: "photos", matches: []string{"photos"}, misses: []string{"", "photos-backup", "myphotos"}}, + {filter: "collection-a,collection-b", matches: []string{"collection-a", "collection-b"}, misses: []string{"collection-c"}}, + {filter: " collection-a , collection-b ,", matches: []string{"collection-a", "collection-b"}, misses: []string{"collection-c"}}, + {filter: "photos*,videos", matches: []string{"photos", "photos-backup", "videos"}, misses: []string{"clips"}}, + {filter: "photo?", matches: []string{"photos"}, misses: []string{"photo", "photos-backup"}}, + // A regex entry matches the whole name unless it anchors itself. + {filter: "photos|videos", matches: []string{"photos", "videos"}, misses: []string{"photos-backup"}}, + {filter: "^photos", matches: []string{"photos", "photos-backup"}, misses: []string{"videos"}}, + {filter: "photos-.*,videos", matches: []string{"photos-backup", "videos"}, misses: []string{"photos"}}, + } + + for _, c := range cases { + matcher, err := CompileCollectionMatcher(c.filter) + if err != nil { + t.Fatalf("CompileCollectionMatcher(%q): %v", c.filter, err) + } + for _, collection := range c.matches { + if !matcher.Matches(collection) { + t.Errorf("filter %q should match collection %q", c.filter, collection) + } + } + for _, collection := range c.misses { + if matcher.Matches(collection) { + t.Errorf("filter %q should not match collection %q", c.filter, collection) + } + } + } +} + +func TestCompileCollectionMatcherInvalidEntry(t *testing.T) { + if _, err := CompileCollectionMatcher("photos,[invalid"); err == nil { + t.Fatal("expected an error for an unparsable entry") + } +} diff --git a/weed/plugin/worker/volume_metrics.go b/weed/plugin/worker/volume_metrics.go index 4988bc715..d40c7c2b3 100644 --- a/weed/plugin/worker/volume_metrics.go +++ b/weed/plugin/worker/volume_metrics.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "regexp" "sort" "strings" "time" @@ -137,15 +136,9 @@ func buildVolumeMetrics( return nil, nil, nil, err } - var collectionRegex *regexp.Regexp - trimmedFilter := strings.TrimSpace(collectionFilter) - filterMode := CollectionFilterMode(trimmedFilter) - if trimmedFilter != "" && filterMode != CollectionFilterAll && filterMode != CollectionFilterEach && trimmedFilter != "*" { - var err error - collectionRegex, err = regexp.Compile(trimmedFilter) - if err != nil { - return nil, nil, nil, &configError{err: fmt.Errorf("invalid collection_filter regex %q: %w", trimmedFilter, err)} - } + collectionMatcher, err := CompileCollectionMatcher(collectionFilter) + if err != nil { + return nil, nil, nil, &configError{err: err} } volumeSizeLimitBytes := uint64(response.VolumeSizeLimitMb) * 1024 * 1024 @@ -167,7 +160,7 @@ func buildVolumeMetrics( Host: pb.NewServerAddressFromDataNode(node).ToHost(), }) - if collectionRegex != nil && !collectionRegex.MatchString(volume.Collection) { + if !collectionMatcher.Matches(volume.Collection) { continue } diff --git a/weed/plugin/worker/volume_metrics_test.go b/weed/plugin/worker/volume_metrics_test.go index 05eeb008f..89d1c9bd3 100644 --- a/weed/plugin/worker/volume_metrics_test.go +++ b/weed/plugin/worker/volume_metrics_test.go @@ -95,6 +95,26 @@ func TestBuildVolumeMetricsRegexFilter(t *testing.T) { } } +func TestBuildVolumeMetricsCollectionList(t *testing.T) { + resp := makeTestVolumeListResponse( + &master_pb.VolumeInformationMessage{Id: 1, Collection: "photos", Size: 100}, + &master_pb.VolumeInformationMessage{Id: 2, Collection: "videos", Size: 200}, + &master_pb.VolumeInformationMessage{Id: 3, Collection: "clips", Size: 300}, + ) + metrics, _, _, err := buildVolumeMetrics(resp, "photos,videos") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(metrics) != 2 { + t.Fatalf("expected 2 metrics, got %d", len(metrics)) + } + for _, metric := range metrics { + if metric.Collection != "photos" && metric.Collection != "videos" { + t.Fatalf("unexpected collection %q", metric.Collection) + } + } +} + func TestBuildVolumeMetricsInvalidRegex(t *testing.T) { resp := makeTestVolumeListResponse( &master_pb.VolumeInformationMessage{Id: 1, Collection: "photos", Size: 100}, diff --git a/weed/worker/tasks/balance/plugin_handler.go b/weed/worker/tasks/balance/plugin_handler.go index f9cedab55..8ea04427e 100644 --- a/weed/worker/tasks/balance/plugin_handler.go +++ b/weed/worker/tasks/balance/plugin_handler.go @@ -85,7 +85,7 @@ func (h *VolumeBalanceHandler) Descriptor() *plugin_pb.JobTypeDescriptor { { Name: "collection_filter", Label: "Collection Filter", - Description: "Filter collections for balance detection. Use ALL_COLLECTIONS (default) to treat all volumes as one pool, EACH_COLLECTION to run detection separately per collection, or a regex pattern to match specific collections.", + Description: "Filter collections for balance detection. Use ALL_COLLECTIONS (default) to treat all volumes as one pool, EACH_COLLECTION to run detection separately per collection, or a comma-separated list of names, wildcards, or regex patterns.", Placeholder: "ALL_COLLECTIONS", FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING, Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT, diff --git a/weed/worker/tasks/ec_balance/config.go b/weed/worker/tasks/ec_balance/config.go index c9e410d6e..33780cdbe 100644 --- a/weed/worker/tasks/ec_balance/config.go +++ b/weed/worker/tasks/ec_balance/config.go @@ -125,9 +125,9 @@ func GetConfigSpec() base.ConfigSpec { DefaultValue: "", Required: false, DisplayName: "Collection Filter", - Description: "Only balance EC shards from specific collections", - HelpText: "Leave empty to balance all collections, or specify collection name/wildcard", - Placeholder: "my_collection", + Description: "Only balance EC shards from specific collections (comma-separated names, wildcards, or regex patterns)", + HelpText: "Leave empty to balance all collections, or list the collections to balance", + Placeholder: "pictures,videos", InputType: "text", CSSClasses: "form-control", }, diff --git a/weed/worker/tasks/ec_balance/detection.go b/weed/worker/tasks/ec_balance/detection.go index 192ed2788..87db29402 100644 --- a/weed/worker/tasks/ec_balance/detection.go +++ b/weed/worker/tasks/ec_balance/detection.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/pb/worker_pb" + pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/ecbalancer" "github.com/seaweedfs/seaweedfs/weed/storage/super_block" @@ -48,7 +49,12 @@ func Detection( return nil, false, fmt.Errorf("topology info not available") } - topo, nodeCount, volumeRatio := buildBalancerTopology(topoInfo, ecConfig) + allowedCollections, err := pluginworker.CompileCollectionMatcher(ecConfig.CollectionFilter) + if err != nil { + return nil, false, err + } + + topo, nodeCount, volumeRatio := buildBalancerTopology(topoInfo, ecConfig, allowedCollections) if nodeCount < ecConfig.MinServerCount { glog.V(1).Infof("EC balance: only %d servers, need at least %d", nodeCount, ecConfig.MinServerCount) return nil, false, nil @@ -148,9 +154,8 @@ func Detection( // per-volume ratio lookup built from each shard's heartbeat (0,0 when unreported, // e.g. always in OSS) which Plan prefers over the collection ratio for mixed-ratio // clusters. -func buildBalancerTopology(topoInfo *master_pb.TopologyInfo, config *Config) (*ecbalancer.Topology, int, func(collection string, vid uint32) (int, int)) { +func buildBalancerTopology(topoInfo *master_pb.TopologyInfo, config *Config, allowedCollections *pluginworker.CollectionMatcher) (*ecbalancer.Topology, int, func(collection string, vid uint32) (int, int)) { topo := ecbalancer.NewTopology() - allowedCollections := wildcard.CompileWildcardMatchers(config.CollectionFilter) type volRatioKey struct { collection string @@ -255,7 +260,7 @@ func buildBalancerTopology(topoInfo *master_pb.TopologyInfo, config *Config) (*e continue } for _, eci := range diskInfo.EcShardInfos { - if len(allowedCollections) > 0 && !wildcard.MatchesAnyWildcard(allowedCollections, eci.Collection) { + if !allowedCollections.Matches(eci.Collection) { continue } node.AddShards(eci.Id, eci.Collection, eci.DiskId, erasure_coding.ShardBits(eci.EcIndexBits)) diff --git a/weed/worker/tasks/ec_balance/detection_test.go b/weed/worker/tasks/ec_balance/detection_test.go index 0bfdec5b4..55facb3bf 100644 --- a/weed/worker/tasks/ec_balance/detection_test.go +++ b/weed/worker/tasks/ec_balance/detection_test.go @@ -6,6 +6,7 @@ import ( "testing" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/ecbalancer" "github.com/seaweedfs/seaweedfs/weed/worker/types" @@ -41,7 +42,7 @@ func ecTopo(node1Collection string) *master_pb.TopologyInfo { func TestBuildBalancerTopology(t *testing.T) { config := NewDefaultConfig() - topo, nodeCount, _ := buildBalancerTopology(ecTopo("col1"), config) + topo, nodeCount, _ := buildBalancerTopology(ecTopo("col1"), config, nil) if nodeCount != 2 { t.Fatalf("nodeCount = %d, want 2", nodeCount) } @@ -77,7 +78,7 @@ func TestBuildBalancerTopologyGroupsByHost(t *testing.T) { }}, } - topo, _, _ := buildBalancerTopology(topoInfo, NewDefaultConfig()) + topo, _, _ := buildBalancerTopology(topoInfo, NewDefaultConfig(), nil) moves := ecbalancer.Plan(topo, ecbalancer.Options{ImbalanceThreshold: 0.01}) host := func(nodeID string) string { h, _, _ := net.SplitHostPort(nodeID); return h } @@ -99,7 +100,11 @@ func TestBuildBalancerTopologyGroupsByHost(t *testing.T) { func TestBuildBalancerTopologyCollectionFilter(t *testing.T) { config := NewDefaultConfig() config.CollectionFilter = "other" // does not match the volume's collection - topo, nodeCount, _ := buildBalancerTopology(ecTopo("col1"), config) + allowed, err := pluginworker.CompileCollectionMatcher(config.CollectionFilter) + if err != nil { + t.Fatalf("CompileCollectionMatcher: %v", err) + } + topo, nodeCount, _ := buildBalancerTopology(ecTopo("col1"), config, allowed) if nodeCount != 2 { t.Fatalf("nodeCount = %d, want 2", nodeCount) } diff --git a/weed/worker/tasks/ec_balance/multidisk_detection_test.go b/weed/worker/tasks/ec_balance/multidisk_detection_test.go index e039787b8..eb7e6f886 100644 --- a/weed/worker/tasks/ec_balance/multidisk_detection_test.go +++ b/weed/worker/tasks/ec_balance/multidisk_detection_test.go @@ -295,13 +295,13 @@ func TestBuildBalancerTopologyNormalizesHddDiskType(t *testing.T) { } topoInfo := buildMasterTopology("c", 100, 50, specs) - if _, n, _ := buildBalancerTopology(topoInfo, &Config{DiskType: "hdd"}); n != 2 { + if _, n, _ := buildBalancerTopology(topoInfo, &Config{DiskType: "hdd"}, nil); n != 2 { t.Errorf("disk_type=hdd matched %d nodes on an all-HDD cluster, want 2 (hdd must map to the empty HDD key)", n) } - if _, n, _ := buildBalancerTopology(topoInfo, &Config{DiskType: ""}); n != 2 { + if _, n, _ := buildBalancerTopology(topoInfo, &Config{DiskType: ""}, nil); n != 2 { t.Errorf("disk_type=empty matched %d nodes, want 2 (all)", n) } - if _, n, _ := buildBalancerTopology(topoInfo, &Config{DiskType: "ssd"}); n != 0 { + if _, n, _ := buildBalancerTopology(topoInfo, &Config{DiskType: "ssd"}, nil); n != 0 { t.Errorf("disk_type=ssd matched %d nodes on an all-HDD cluster, want 0", n) } } diff --git a/weed/worker/tasks/ec_balance/plugin_handler.go b/weed/worker/tasks/ec_balance/plugin_handler.go index dc3b89468..d545dc5a4 100644 --- a/weed/worker/tasks/ec_balance/plugin_handler.go +++ b/weed/worker/tasks/ec_balance/plugin_handler.go @@ -81,7 +81,7 @@ func (h *ECBalanceHandler) Descriptor() *plugin_pb.JobTypeDescriptor { { Name: "collection_filter", Label: "Collection Filter", - Description: "Only balance EC shards in matching collections (wildcard supported).", + Description: "Only balance EC shards in matching collections. Comma-separated list of names, wildcards, or regex patterns.", Placeholder: "all collections", FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING, Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT, diff --git a/weed/worker/tasks/erasure_coding/config.go b/weed/worker/tasks/erasure_coding/config.go index 7079d1640..5f7c085fb 100644 --- a/weed/worker/tasks/erasure_coding/config.go +++ b/weed/worker/tasks/erasure_coding/config.go @@ -123,9 +123,9 @@ func GetConfigSpec() base.ConfigSpec { DefaultValue: "", Required: false, DisplayName: "Collection Filter", - Description: "Only process volumes from specific collections", - HelpText: "Leave empty to process all collections, or specify collection name", - Placeholder: "my_collection", + Description: "Only process volumes from specific collections (comma-separated names, wildcards, or regex patterns)", + HelpText: "Leave empty to process all collections, or list the collections to process", + Placeholder: "pictures,videos", InputType: "text", CSSClasses: "form-control", }, diff --git a/weed/worker/tasks/erasure_coding/detection.go b/weed/worker/tasks/erasure_coding/detection.go index 63a602fdd..bd20737d9 100644 --- a/weed/worker/tasks/erasure_coding/detection.go +++ b/weed/worker/tasks/erasure_coding/detection.go @@ -14,10 +14,10 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" + pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/ecbalancer" "github.com/seaweedfs/seaweedfs/weed/storage/super_block" - "github.com/seaweedfs/seaweedfs/weed/util/wildcard" "github.com/seaweedfs/seaweedfs/weed/worker/tasks/base" workerutil "github.com/seaweedfs/seaweedfs/weed/worker/tasks/util" "github.com/seaweedfs/seaweedfs/weed/worker/types" @@ -69,7 +69,10 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste glog.Warningf("EC Detection: replica placement data-center digit (%d) is ignored for EC; only rack/node digits are honored", replicaPlacement.DiffDataCenterCount) } - allowedCollections := wildcard.CompileWildcardMatchers(ecConfig.CollectionFilter) + allowedCollections, err := pluginworker.CompileCollectionMatcher(ecConfig.CollectionFilter) + if err != nil { + return nil, false, err + } // Cluster node count for the min-node safety gate (mirrors the shell ec.encode // guard that refuses to encode when nodes < parity shards, so shards cannot be @@ -187,7 +190,7 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste } // Check collection filter if specified - if len(allowedCollections) > 0 && !wildcard.MatchesAnyWildcard(allowedCollections, metric.Collection) { + if !allowedCollections.Matches(metric.Collection) { skippedCollectionFilter++ continue } diff --git a/weed/worker/tasks/erasure_coding/detection_disk_type_test.go b/weed/worker/tasks/erasure_coding/detection_disk_type_test.go index e1fa35da9..8b47270c8 100644 --- a/weed/worker/tasks/erasure_coding/detection_disk_type_test.go +++ b/weed/worker/tasks/erasure_coding/detection_disk_type_test.go @@ -18,7 +18,7 @@ import ( func TestPlanECDestinationsPrefersSourceDiskType_FullCluster(t *testing.T) { // 14 nodes × 2 disks per node (hdd + ssd) — enough SSD slots alone // for a 10+4 layout with one-shard-per-(server,disk) diversity. - activeTopology := buildActiveTopology(t, erasure_coding.TotalShardsCount, []string{"hdd", "ssd"}, 100, 0) + activeTopology := buildActiveTopology(t, erasure_coding.TotalShardsCount, []string{"hdd", "ssd"}, 100, 0, "") metric := &types.VolumeHealthMetrics{ VolumeID: 1, @@ -53,7 +53,7 @@ func TestPlanECDestinationsPrefersSourceDiskType_FullCluster(t *testing.T) { func TestPlanECDestinationsSpillsToOtherDiskType_WhenPreferredScarce(t *testing.T) { // Start with every node carrying both HDD and SSD, then strip SSD // from all but the first node so the SSD pool is too small alone. - activeTopology := buildActiveTopology(t, erasure_coding.TotalShardsCount, []string{"hdd", "ssd"}, 100, 0) + activeTopology := buildActiveTopology(t, erasure_coding.TotalShardsCount, []string{"hdd", "ssd"}, 100, 0, "") topo := activeTopology.GetTopologyInfo() for _, dc := range topo.DataCenterInfos { for _, rack := range dc.RackInfos { diff --git a/weed/worker/tasks/erasure_coding/detection_test.go b/weed/worker/tasks/erasure_coding/detection_test.go index 18de939df..6634c41ea 100644 --- a/weed/worker/tasks/erasure_coding/detection_test.go +++ b/weed/worker/tasks/erasure_coding/detection_test.go @@ -16,7 +16,7 @@ import ( ) func TestPlanECDestinationsUsesPlanner(t *testing.T) { - activeTopology := buildActiveTopology(t, 7, []string{"hdd", "ssd"}, 100, 0) + activeTopology := buildActiveTopology(t, 7, []string{"hdd", "ssd"}, 100, 0, "") metric := &types.VolumeHealthMetrics{ VolumeID: 1, @@ -207,7 +207,7 @@ func TestCountExistingEcShardsForVolume(t *testing.T) { } func TestDetectionContextCancellation(t *testing.T) { - activeTopology := buildActiveTopology(t, 5, []string{"hdd", "ssd"}, 50, 0) + activeTopology := buildActiveTopology(t, 5, []string{"hdd", "ssd"}, 50, 0, "") clusterInfo := &types.ClusterInfo{ActiveTopology: activeTopology} metrics := buildVolumeMetricsForIDs(50) @@ -218,9 +218,44 @@ func TestDetectionContextCancellation(t *testing.T) { require.ErrorIs(t, err, context.Canceled) } +// The admin UI collection filter takes a list, and the two mode sentinels must +// not be read as a collection name. +func TestDetectionCollectionFilter(t *testing.T) { + for _, tc := range []struct { + filter string + want int + }{ + {filter: "", want: 3}, + {filter: "ALL_COLLECTIONS", want: 3}, + {filter: "collection-a", want: 3}, + {filter: "collection-a,collection-b", want: 3}, + {filter: " collection-b , collection-a ", want: 3}, + {filter: "collection-*", want: 3}, + {filter: "collection-b,collection-c", want: 0}, + {filter: "collection-ab", want: 0}, + } { + t.Run(tc.filter, func(t *testing.T) { + // A fresh topology per case: planned destinations reserve capacity. + activeTopology := buildActiveTopology(t, erasure_coding.TotalShardsCount, []string{"hdd"}, 20, 0, "collection-a") + clusterInfo := &types.ClusterInfo{ActiveTopology: activeTopology} + + metrics := buildVolumeMetricsForIDs(3) + for _, metric := range metrics { + metric.Collection = "collection-a" + } + + config := NewDefaultConfig() + config.CollectionFilter = tc.filter + results, _, err := Detection(context.Background(), metrics, clusterInfo, config, 0) + require.NoError(t, err) + assert.Len(t, results, tc.want) + }) + } +} + func TestDetectionMaxResultsHonorsLimit(t *testing.T) { // One node per shard so each shard gets its own disk (#9369). - activeTopology := buildActiveTopology(t, erasure_coding.TotalShardsCount, []string{"hdd"}, 20, 0) + activeTopology := buildActiveTopology(t, erasure_coding.TotalShardsCount, []string{"hdd"}, 20, 0, "") clusterInfo := &types.ClusterInfo{ActiveTopology: activeTopology} metrics := buildVolumeMetricsForIDs(3) @@ -285,7 +320,7 @@ func TestPlanECDestinationsSpreadsAcrossPhysicalDisks(t *testing.T) { } func TestPlanECDestinationsFailsWithInsufficientCapacity(t *testing.T) { - activeTopology := buildActiveTopology(t, 1, []string{"hdd"}, 1, 1) + activeTopology := buildActiveTopology(t, 1, []string{"hdd"}, 1, 1, "") metric := &types.VolumeHealthMetrics{ VolumeID: 2, @@ -411,7 +446,7 @@ func buildVolumeMetricsForIDs(count int) []*types.VolumeHealthMetrics { return metrics } -func buildActiveTopology(t *testing.T, nodeCount int, diskTypes []string, maxVolumeCount, usedVolumeCount int64) *topology.ActiveTopology { +func buildActiveTopology(t *testing.T, nodeCount int, diskTypes []string, maxVolumeCount, usedVolumeCount int64, collection string) *topology.ActiveTopology { t.Helper() activeTopology := topology.NewActiveTopology(10) @@ -427,7 +462,7 @@ func buildActiveTopology(t *testing.T, nodeCount int, diskTypes []string, maxVol for vid := 1; vid <= 200; vid++ { volumeInfos = append(volumeInfos, &master_pb.VolumeInformationMessage{ Id: uint32(vid), - Collection: "", + Collection: collection, DiskId: uint32(diskIndex), }) } diff --git a/weed/worker/tasks/erasure_coding/plugin_handler.go b/weed/worker/tasks/erasure_coding/plugin_handler.go index f62fea820..1ca7ef94a 100644 --- a/weed/worker/tasks/erasure_coding/plugin_handler.go +++ b/weed/worker/tasks/erasure_coding/plugin_handler.go @@ -14,7 +14,6 @@ import ( pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker" ecstorage "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/util" - "github.com/seaweedfs/seaweedfs/weed/util/wildcard" workertypes "github.com/seaweedfs/seaweedfs/weed/worker/types" "google.golang.org/grpc" "google.golang.org/protobuf/proto" @@ -78,7 +77,7 @@ func (h *ErasureCodingHandler) Descriptor() *plugin_pb.JobTypeDescriptor { { Name: "collection_filter", Label: "Collection Filter", - Description: "Only detect erasure coding opportunities in this collection when set.", + Description: "Only erasure code volumes in matching collections. Comma-separated list of names, wildcards, or regex patterns.", Placeholder: "all collections", FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING, Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT, @@ -296,7 +295,10 @@ func emitErasureCodingDetectionDecisionTrace( quietThreshold := time.Duration(taskConfig.QuietForSeconds) * time.Second minSizeBytes := uint64(taskConfig.MinSizeMB) * 1024 * 1024 - allowedCollections := wildcard.CompileWildcardMatchers(taskConfig.CollectionFilter) + allowedCollections, err := pluginworker.CompileCollectionMatcher(taskConfig.CollectionFilter) + if err != nil { + return err + } volumeGroups := make(map[uint32][]*workertypes.VolumeHealthMetrics) for _, metric := range metrics { @@ -334,7 +336,7 @@ func emitErasureCodingDetectionDecisionTrace( skippedTooSmall++ continue } - if len(allowedCollections) > 0 && !wildcard.MatchesAnyWildcard(allowedCollections, metric.Collection) { + if !allowedCollections.Matches(metric.Collection) { skippedCollectionFilter++ continue } diff --git a/weed/worker/tasks/vacuum/plugin_handler.go b/weed/worker/tasks/vacuum/plugin_handler.go index 849100d63..1ccd3cb7d 100644 --- a/weed/worker/tasks/vacuum/plugin_handler.go +++ b/weed/worker/tasks/vacuum/plugin_handler.go @@ -85,7 +85,7 @@ func (h *VacuumHandler) Descriptor() *plugin_pb.JobTypeDescriptor { { Name: "collection_filter", Label: "Collection Filter", - Description: "Filter collections for vacuum detection. Use ALL_COLLECTIONS (default) to treat all volumes as one pool, EACH_COLLECTION to run detection separately per collection, or a regex pattern to match specific collections.", + Description: "Filter collections for vacuum detection. Use ALL_COLLECTIONS (default) to treat all volumes as one pool, EACH_COLLECTION to run detection separately per collection, or a comma-separated list of names, wildcards, or regex patterns.", Placeholder: "ALL_COLLECTIONS", FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING, Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT,