mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-28 03:46:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

shell: parse every collection filter the same way (#10955)

Browse Source 
* worker: move the collection filter parser into weed/util/wildcard

The parser sits beside the volume-list filtering it was written for, in
weed/plugin/worker, which imports weed/shell — so the shell commands that
parse the same filter three other ways can never call it. Move it down to
weed/util/wildcard, next to the comma-separated wildcard helper it already
replaced, leaving the behavior unchanged.

* shell: parse every collection filter the same way

The shell parsed a collection filter three ways: compileCollectionPattern
compiled one regex for ec.encode, ec.decode, volume.balance and the tier
commands; volume.list and volume.deleteEmpty matched a single wildcard; and
volume.tier.move, volume.fix.replication and volume.configure.replication
called filepath.Match on their own. None of them took a list, so
"ec.encode -collection=a,b" selected nothing, the same way the admin UI did.

They all go through the shared matcher now: a comma-separated list of names,
"*" and "?" wildcards, "_default" for the collection with no name, and regex
entries. The one thing that stays per-command is what an empty value means -
every collection for -collectionPattern, the unnamed collection for the ec
and tier -collection flag - so compileCollectionPattern keeps that mapping.

The matchers are compiled once per command instead of once per volume, and a
regex entry now has to match the whole name unless it anchors itself, so
-collection=bucket no longer picks up mybucket2.

* shell: keep dots in collection names, and commas inside a regex

A dot no longer marks an entry as a regex, so a collection named "my.bucket"
matches itself and not "my-bucket" - the difference decides which volumes
volume.deleteEmpty and volume.tier.move touch. A dot still counts when it is
quantified, so "bucket.*" stays a prefix regex.

The comma split also leaves alone the commas inside a character class or a
repetition count, so "bucket[0-9]{1,3}" stays one entry instead of becoming
two broken fragments.

* shell: let a regex entry match its own spelling

A collection named after regex syntax, say "logs(2024)", was unreachable:
the entry compiled to a pattern that matches "logs2024" instead. Match the
entry verbatim as well, so naming a collection always selects it, whatever
characters it holds.

* shell: reject a collection filter that names no collection

A value of "," parsed to no entries and then matched every collection, so a
typo widened ec.encode or volume.deleteEmpty to the whole cluster. Only a
genuinely empty filter means "all collections"; anything else has to name one.

* shell: keep commas inside a regex group out of the entry split

The split already left alone the commas inside a character class or a
repetition count, but not the ones inside a group, so "bucket(foo,bar)"
was cut into two fragments that no longer compile.

* shell: cover escaping a collection name that is not a regex

A name like "logs(2024" does not parse as a regex on its own; escaping it,
"logs\(2024", reaches it. Pin that so the escape hatch does not regress.

* shell: split entries only on commas inside a closed regex construct

An unmatched "{" or "[" made the splitter swallow every comma after it, so
"foo{bar,videos" became one entry that matches neither collection - the
silent no-op this filter work exists to remove. A construct now has to close
before its commas stop separating entries.

* shell: skip character classes while scanning a regex group

A ")" inside a class is a literal, so "(a[)],b)" ended its group early and
split into two fragments that no longer compile.

* shell: cover escaping a comma inside a collection name

A comma separates entries, so a name holding one is reached by escaping it.

* shell: follow the regexp parser when scanning a character class

A "]" leading a class is a member of it, and a POSIX class such as
"[:alpha:]" carries its own "]", so stopping at the first one cut a valid
filter like "(a[]),],b)" into fragments and rejected it.
This commit is contained in:
Chris Lu
2026-08-25 18:03:52 -07:00
committed by GitHub
parent 368b2035b2
commit 627b5e9d59
31 changed files with 506 additions and 292 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/ec/ec_decode.go
+3 -3
View File
@@ -3,7 +3,6 @@ package ec
import (
"context"
"fmt"
"regexp"
"strings"
"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -15,6 +14,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/seaweedfs/seaweedfs/weed/util"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
@@ -357,12 +357,12 @@ func collectEcShards(env *Env, nodeToShardsInfo map[pb.ServerAddress]*erasure_co
return targetNodeLocation, err
}
func CollectEcShardIds(topoInfo *master_pb.TopologyInfo, collectionRegex *regexp.Regexp, diskType types.DiskType) (vids []needle.VolumeId) {
func CollectEcShardIds(topoInfo *master_pb.TopologyInfo, collectionMatcher *wildcard.CollectionMatcher, diskType types.DiskType) (vids []needle.VolumeId) {
vidMap := make(map[uint32]bool)
EachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
if diskInfo, found := dn.DiskInfos[string(diskType)]; found {
for _, v := range diskInfo.EcShardInfos {
if collectionRegex.MatchString(v.Collection) {
if collectionMatcher.Matches(v.Collection) {
vidMap[v.Id] = true
}
}
weed/ec/ec_encode.go
+4 -4
View File
@@ -5,7 +5,6 @@ import (
"errors"
"fmt"
"io"
"regexp"
"slices"
"sort"
"strconv"
@@ -25,6 +24,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/seaweedfs/seaweedfs/weed/storage/volume_replica"
"github.com/seaweedfs/seaweedfs/weed/util"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
"google.golang.org/grpc"
)
@@ -927,7 +927,7 @@ func generateEcShards(grpcDialOption grpc.DialOption, volumeId needle.VolumeId,
}
func SelectVolumeIdsFromTopology(topologyInfo *master_pb.TopologyInfo, volumeSizeLimitMb uint64, collectionRegex *regexp.Regexp, sourceDiskType *types.DiskType, quietSeconds int64, nowUnixSeconds int64, fullPercentage float64, verbose bool) (vids []needle.VolumeId, matchedCollections []string) {
func SelectVolumeIdsFromTopology(topologyInfo *master_pb.TopologyInfo, volumeSizeLimitMb uint64, collectionMatcher *wildcard.CollectionMatcher, sourceDiskType *types.DiskType, quietSeconds int64, nowUnixSeconds int64, fullPercentage float64, verbose bool) (vids []needle.VolumeId, matchedCollections []string) {
// Statistics for verbose mode
var (
totalVolumes int
@@ -957,11 +957,11 @@ func SelectVolumeIdsFromTopology(topologyInfo *master_pb.TopologyInfo, volumeSiz
}
// check collection against regex pattern
if !collectionRegex.MatchString(v.Collection) {
if !collectionMatcher.Matches(v.Collection) {
wrongCollection++
if verbose {
fmt.Printf("skip volume %d on %s: collection doesn't match pattern (pattern: %s, actual: %s)\n",
v.Id, dn.Id, collectionRegex.String(), v.Collection)
v.Id, dn.Id, collectionMatcher.String(), v.Collection)
}
continue
}
weed/ec/ec_encode_test.go
+4 -5
View File
@@ -1,7 +1,6 @@
package ec
import (
"regexp"
"testing"
"time"
@@ -10,6 +9,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
"github.com/stretchr/testify/assert"
)
@@ -126,8 +126,7 @@ func TestSelectVolumeIdsFromTopology(t *testing.T) {
}
volumeSizeLimitMb := uint64(1000)
collectionPattern := ".*"
collectionRegex, _ := regexp.Compile(collectionPattern)
collectionMatcher, _ := wildcard.CompileCollectionMatcher(".*")
// ec.encode -force
// force means we ignore the check for 4+ volume servers.
@@ -177,7 +176,7 @@ func TestSelectVolumeIdsFromTopology(t *testing.T) {
// In the provided topology, FreeVolumeCount is 1 ("free:1"), so it is < 2.
// So ALL volumes should be skipped due to insufficient free disk space.
vids, _ := SelectVolumeIdsFromTopology(topologyInfo, volumeSizeLimitMb, collectionRegex, nil, quietSeconds, nowUnixSeconds, fullPercentage, verbose)
vids, _ := SelectVolumeIdsFromTopology(topologyInfo, volumeSizeLimitMb, collectionMatcher, nil, quietSeconds, nowUnixSeconds, fullPercentage, verbose)
assert.Equal(t, 0, len(vids), "Should select 0 volumes because FreeVolumeCount is 1 (less than 2)")
@@ -197,7 +196,7 @@ func TestSelectVolumeIdsFromTopology(t *testing.T) {
// So expected volumes: 10, 11, 12.
vids, _ = SelectVolumeIdsFromTopology(topologyInfo, volumeSizeLimitMb, collectionRegex, nil, quietSeconds, nowUnixSeconds, fullPercentage, verbose)
vids, _ = SelectVolumeIdsFromTopology(topologyInfo, volumeSizeLimitMb, collectionMatcher, nil, quietSeconds, nowUnixSeconds, fullPercentage, verbose)
expectedVids := []needle.VolumeId{10, 11, 12}
assert.Equal(t, len(expectedVids), len(vids), "Should select 3 volumes")
weed/plugin/worker/collection_filter.go
-89
View File
@@ -1,89 +0,0 @@
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 matching:
// - CollectionFilterAll: pool every collection together (default).
// - CollectionFilterEach: run detection separately per collection.
//
// 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
}
weed/plugin/worker/collection_filter_test.go
-48
View File
@@ -1,48 +0,0 @@
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")
}
}
weed/plugin/worker/volume_metrics.go
+2 -1
View File
@@ -12,6 +12,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
workertypes "github.com/seaweedfs/seaweedfs/weed/worker/types"
"google.golang.org/grpc"
)
@@ -136,7 +137,7 @@ func buildVolumeMetrics(
return nil, nil, nil, err
}
collectionMatcher, err := CompileCollectionMatcher(collectionFilter)
collectionMatcher, err := wildcard.CompileCollectionMatcher(collectionFilter)
if err != nil {
return nil, nil, nil, &configError{err: err}
}
weed/plugin/worker/volume_metrics_test.go
+3 -2
View File
@@ -4,6 +4,7 @@ import (
"testing"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
)
func makeTestVolumeListResponse(volumes ...*master_pb.VolumeInformationMessage) *master_pb.VolumeListResponse {
@@ -53,7 +54,7 @@ func TestBuildVolumeMetricsAllCollections(t *testing.T) {
&master_pb.VolumeInformationMessage{Id: 1, Collection: "photos", Size: 100},
&master_pb.VolumeInformationMessage{Id: 2, Collection: "videos", Size: 200},
)
metrics, _, _, err := buildVolumeMetrics(resp, string(CollectionFilterAll))
metrics, _, _, err := buildVolumeMetrics(resp, string(wildcard.CollectionFilterAll))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
@@ -68,7 +69,7 @@ func TestBuildVolumeMetricsEachCollection(t *testing.T) {
&master_pb.VolumeInformationMessage{Id: 2, Collection: "videos", Size: 200},
)
// EACH_COLLECTION passes all volumes through; filtering happens in the handler
metrics, _, _, err := buildVolumeMetrics(resp, string(CollectionFilterEach))
metrics, _, _, err := buildVolumeMetrics(resp, string(wildcard.CollectionFilterEach))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
weed/shell/command_ec_common.go
+10 -13
View File
@@ -2,7 +2,7 @@ package shell
import (
"context"
"regexp"
"strings"
"time"
"github.com/seaweedfs/seaweedfs/weed/ec"
@@ -13,6 +13,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
)
@@ -141,17 +142,13 @@ 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 a regex pattern for collection matching.
// Empty patterns match empty collections only.
// The special keyword CollectionDefault ("_default") matches empty collections.
func compileCollectionPattern(pattern string) (*regexp.Regexp, error) {
if pattern == "" {
// empty pattern matches empty collection
return regexp.Compile("^$")
// 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.
func compileCollectionPattern(pattern string) (*wildcard.CollectionMatcher, error) {
if strings.TrimSpace(pattern) == "" {
pattern = CollectionDefault
}
if pattern == CollectionDefault {
// CollectionDefault keyword matches empty collection
return regexp.Compile("^$")
}
return regexp.Compile(pattern)
return wildcard.CompileCollectionMatcher(pattern)
}
weed/shell/command_ec_common_test.go
+39
View File
@@ -0,0 +1,39 @@
package shell
import "testing"
// The ec and tier commands read an empty -collection as the collection with no
// name, unlike -collectionPattern, where empty means every collection.
func TestCompileCollectionPattern(t *testing.T) {
tests := []struct {
pattern string
matches []string
misses []string
}{
{pattern: "", matches: []string{""}, misses: []string{"pictures"}},
{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*", matches: []string{"pictures", "pictures-backup"}, misses: []string{"clips"}},
{pattern: "^pictures", matches: []string{"pictures", "pictures-backup"}, misses: []string{"clips"}},
{pattern: "*", matches: []string{"", "pictures"}},
}
for _, tt := range tests {
matcher, err := compileCollectionPattern(tt.pattern)
if err != nil {
t.Fatalf("compileCollectionPattern(%q): %v", tt.pattern, err)
}
for _, collection := range tt.matches {
if !matcher.Matches(collection) {
t.Errorf("pattern %q should match collection %q", tt.pattern, collection)
}
}
for _, collection := range tt.misses {
if matcher.Matches(collection) {
t.Errorf("pattern %q should not match collection %q", tt.pattern, collection)
}
}
}
}
weed/shell/command_ec_decode.go
+3 -3
View File
@@ -55,7 +55,7 @@ func (c *commandEcDecode) HasTag(CommandTag) bool {
func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
decodeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
volumeId := decodeCommand.Int("volumeId", 0, "the volume id")
collection := decodeCommand.String("collection", "", "the collection name")
collection := decodeCommand.String("collection", "", "comma-separated collection names, wildcards, or regex patterns; empty matches the collection with no name")
diskTypeStr := decodeCommand.String("diskType", "", "source disk type where EC shards are stored (hdd, ssd, or empty for default hdd)")
checkMinFreeSpace := decodeCommand.Bool("checkMinFreeSpace", true, "check min free space when selecting the decode target")
batchSize := decodeCommand.Int("batchSize", DefaultEcBatchSize, "decode up to this many volumes per topology refresh (0 = one snapshot for all volumes)")
@@ -90,11 +90,11 @@ func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
}
// apply to all volumes in the collection
collectionRegex, err := compileCollectionPattern(*collection)
collectionMatcher, err := compileCollectionPattern(*collection)
if err != nil {
return fmt.Errorf("invalid collection pattern '%s': %v", *collection, err)
}
volumeIds := ec.CollectEcShardIds(topologyInfo, collectionRegex, diskType)
volumeIds := ec.CollectEcShardIds(topologyInfo, collectionMatcher, diskType)
fmt.Printf("ec decode volumes: %v\n", volumeIds)
batches := chunkVolumeIds(volumeIds, *batchSize)
for i, batch := range batches {
weed/shell/command_ec_encode.go
+10 -7
View File
@@ -47,10 +47,13 @@ func (c *commandEcEncode) Help() string {
If you only have less than 4 volume servers, with erasure coding, at least you can afford to
have 4 corrupted shard files.
The -collection parameter supports regular expressions for pattern matching:
- Use exact match: ec.encode -collection="^mybucket$"
- Match multiple buckets: ec.encode -collection="bucket.*"
- Match all collections: ec.encode -collection=".*"
The -collection parameter is a comma-separated list of collection names, with
"*" and "?" wildcards, and regex patterns:
- One collection: ec.encode -collection="mybucket"
- 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"
Options:
-verbose: show detailed reasons why volumes are not selected for encoding
@@ -88,7 +91,7 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
encodeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
volumeId := encodeCommand.Int("volumeId", 0, "the volume id")
volumeIdsStr := encodeCommand.String("volumeIds", "", "comma-separated volume ids")
collection := encodeCommand.String("collection", "", "collection name or regex pattern")
collection := encodeCommand.String("collection", "", "comma-separated collection names, wildcards, or regex patterns; empty matches the collection with no name")
fullPercentage := encodeCommand.Float64("fullPercent", 95, "the volume reaches the percentage of max volume size")
quietPeriod := encodeCommand.Duration("quietFor", time.Hour, "select volumes without no writes for this period")
maxParallelization := encodeCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible")
@@ -186,7 +189,7 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
}
func collectVolumeIdsForEcEncode(commandEnv *CommandEnv, collectionPattern string, sourceDiskType *types.DiskType, fullPercentage float64, quietPeriod time.Duration, verbose bool) (vids []needle.VolumeId, matchedCollections []string, err error) {
// compile regex pattern for collection matching
collectionRegex, err := compileCollectionPattern(collectionPattern)
collectionMatcher, err := compileCollectionPattern(collectionPattern)
if err != nil {
return nil, nil, fmt.Errorf("invalid collection pattern '%s': %v", collectionPattern, err)
}
@@ -202,6 +205,6 @@ func collectVolumeIdsForEcEncode(commandEnv *CommandEnv, collectionPattern strin
fmt.Printf("collect volumes with collection pattern '%s', quiet for: %d seconds and %.1f%% full\n", collectionPattern, quietSeconds, fullPercentage)
vids, matchedCollections = ec.SelectVolumeIdsFromTopology(topologyInfo, volumeSizeLimitMb, collectionRegex, sourceDiskType, quietSeconds, nowUnixSeconds, fullPercentage, verbose)
vids, matchedCollections = ec.SelectVolumeIdsFromTopology(topologyInfo, volumeSizeLimitMb, collectionMatcher, sourceDiskType, quietSeconds, nowUnixSeconds, fullPercentage, verbose)
return
}
weed/shell/command_volume_balance.go
+14 -15
View File
@@ -7,7 +7,6 @@ import (
"fmt"
"io"
"os"
"regexp"
"strings"
"sync"
"time"
@@ -22,6 +21,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/seaweedfs/seaweedfs/weed/topology/balancer"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
@@ -131,7 +131,7 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
*allowedVolumeBy["ACTIVE"] = true
balanceCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
verbose := balanceCommand.Bool("v", false, "verbose mode")
collection := balanceCommand.String("collection", "ALL_COLLECTIONS", "collection name, or use \"ALL_COLLECTIONS\" across collections, \"EACH_COLLECTION\" for each collection")
collection := balanceCommand.String("collection", "ALL_COLLECTIONS", "comma-separated collection names, wildcards, or regex patterns, or \"ALL_COLLECTIONS\" across collections, \"EACH_COLLECTION\" for each collection")
dc := balanceCommand.String("dataCenter", "", "only apply the balancing for this dataCenter")
racks := balanceCommand.String("racks", "", "only apply the balancing for this racks")
nodes := balanceCommand.String("nodes", "", "only apply the balancing for this nodes")
@@ -198,7 +198,7 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
volumeReplicas, _ := collectVolumeReplicaLocations(topologyInfo)
diskTypes := collectVolumeDiskTypes(topologyInfo)
if *collection == "EACH_COLLECTION" {
if *collection == string(wildcard.CollectionFilterEach) {
collections, err := ListCollectionNames(commandEnv, true, false)
if err != nil {
return err
@@ -212,18 +212,18 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
return err
}
}
} else if *collection == "ALL_COLLECTIONS" {
// Pass nil pattern for all collections
if err = c.balanceVolumeServers(diskTypes, volumeReplicas, volumeServers, nil, *collection); err != nil {
} else if *collection == string(wildcard.CollectionFilterAll) || *collection == "*" {
// Pass nil matcher for all collections
if err = c.balanceVolumeServers(diskTypes, volumeReplicas, volumeServers, nil, string(wildcard.CollectionFilterAll)); err != nil {
return err
}
} else {
// Compile user-provided pattern
collectionPattern, err := compileCollectionPattern(*collection)
collectionMatcher, err := compileCollectionPattern(*collection)
if err != nil {
return fmt.Errorf("invalid collection pattern '%s': %v", *collection, err)
}
if err = c.balanceVolumeServers(diskTypes, volumeReplicas, volumeServers, collectionPattern, *collection); err != nil {
if err = c.balanceVolumeServers(diskTypes, volumeReplicas, volumeServers, collectionMatcher, *collection); err != nil {
return err
}
}
@@ -231,25 +231,24 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
return nil
}
func (c *commandVolumeBalance) balanceVolumeServers(diskTypes []types.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, collectionPattern *regexp.Regexp, collectionName string) error {
func (c *commandVolumeBalance) balanceVolumeServers(diskTypes []types.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, collectionMatcher *wildcard.CollectionMatcher, collectionName string) error {
for _, diskType := range diskTypes {
if c.volumesPerExec > 0 && c.movedCount >= c.volumesPerExec {
break
}
if err := c.balanceVolumeServersByDiskType(diskType, volumeReplicas, nodes, collectionPattern, collectionName); err != nil {
if err := c.balanceVolumeServersByDiskType(diskType, volumeReplicas, nodes, collectionMatcher, collectionName); err != nil {
return err
}
}
return nil
}
func (c *commandVolumeBalance) balanceVolumeServersByDiskType(diskType types.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, collectionPattern *regexp.Regexp, collectionName string) error {
func (c *commandVolumeBalance) balanceVolumeServersByDiskType(diskType types.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, collectionMatcher *wildcard.CollectionMatcher, collectionName string) error {
for _, n := range nodes {
n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool {
if collectionName != "ALL_COLLECTIONS" {
if collectionPattern != nil {
// Use regex pattern matching
if !collectionPattern.MatchString(v.Collection) {
if collectionName != string(wildcard.CollectionFilterAll) {
if collectionMatcher != nil {
if !collectionMatcher.Matches(v.Collection) {
return false
}
} else {
weed/shell/command_volume_configure_replication.go
+14 -20
View File
@@ -6,7 +6,6 @@ import (
"flag"
"fmt"
"io"
"path/filepath"
"github.com/seaweedfs/seaweedfs/weed/pb"
@@ -15,6 +14,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
)
func init() {
@@ -45,7 +45,7 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
configureReplicationCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
volumeIdInt := configureReplicationCommand.Int("volumeId", 0, "the volume id")
replicationString := configureReplicationCommand.String("replication", "", "the intended replication value")
collectionPattern := configureReplicationCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
collectionPattern := configureReplicationCommand.String("collectionPattern", "", "comma-separated collection names, with '*' and '?' wildcards; empty matches all")
if err = configureReplicationCommand.Parse(args); err != nil {
return nil
}
@@ -70,7 +70,10 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
}
vid := needle.VolumeId(*volumeIdInt)
volumeFilter := getVolumeFilter(replicaPlacement, uint32(vid), *collectionPattern)
volumeFilter, err := getVolumeFilter(replicaPlacement, uint32(vid), *collectionPattern)
if err != nil {
return err
}
// find all data nodes with volumes that needs replication change
eachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
@@ -108,27 +111,18 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
return err
}
func getVolumeFilter(replicaPlacement *super_block.ReplicaPlacement, volumeId uint32, collectionPattern string) func(message *master_pb.VolumeInformationMessage) bool {
func getVolumeFilter(replicaPlacement *super_block.ReplicaPlacement, volumeId uint32, collectionPattern string) (func(message *master_pb.VolumeInformationMessage) bool, error) {
replicaPlacementInt32 := uint32(replicaPlacement.Byte())
if volumeId > 0 {
return func(v *master_pb.VolumeInformationMessage) bool {
return v.Id == volumeId && v.ReplicaPlacement != replicaPlacementInt32
}
}, nil
}
collectionMatcher, err := wildcard.CompileCollectionMatcher(collectionPattern)
if err != nil {
return nil, err
}
return func(v *master_pb.VolumeInformationMessage) bool {
var collectionMatched bool
if collectionPattern == "" {
// Empty pattern matches all collections
collectionMatched = true
} else if collectionPattern == CollectionDefault {
collectionMatched = v.Collection == ""
} else {
m, err := filepath.Match(collectionPattern, v.Collection)
if err != nil {
return false
}
collectionMatched = m
}
return collectionMatched && v.ReplicaPlacement != replicaPlacementInt32
}
return collectionMatcher.Matches(v.Collection) && v.ReplicaPlacement != replicaPlacementInt32
}, nil
}
weed/shell/command_volume_delete_empty.go
+10 -4
View File
@@ -11,6 +11,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
)
func init() {
@@ -43,7 +44,7 @@ func (c *commandVolumeDeleteEmpty) Do(args []string, commandEnv *CommandEnv, wri
volDeleteCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
quietPeriod := volDeleteCommand.Duration("quietFor", 24*time.Hour, "select empty volumes with no recent writes, avoid newly created ones")
collectionPattern := volDeleteCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
collectionPattern := volDeleteCommand.String("collectionPattern", "", "comma-separated collection names, with '*' and '?' wildcards; empty matches all")
applyBalancing := volDeleteCommand.Bool("apply", false, "apply to delete empty volumes")
// TODO: remove this alias
applyBalancingAlias := volDeleteCommand.Bool("force", false, "apply to delete empty volumes (alias for -apply)")
@@ -64,13 +65,18 @@ func (c *commandVolumeDeleteEmpty) Do(args []string, commandEnv *CommandEnv, wri
return err
}
collectionMatcher, err := wildcard.CompileCollectionMatcher(*collectionPattern)
if err != nil {
return err
}
quietSeconds := int64(*quietPeriod / time.Second)
nowUnixSeconds := time.Now().Unix()
eachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
for _, diskInfo := range dn.DiskInfos {
for _, v := range diskInfo.VolumeInfos {
if isEmptyVolumeDeleteCandidate(v, quietSeconds, nowUnixSeconds, *collectionPattern) {
if isEmptyVolumeDeleteCandidate(v, quietSeconds, nowUnixSeconds, collectionMatcher) {
if *applyBalancing {
log.Printf("deleting empty volume %d from %s", v.Id, dn.Id)
if deleteErr := deleteVolume(context.Background(), commandEnv.option.GrpcDialOption, needle.VolumeId(v.Id),
@@ -89,8 +95,8 @@ func (c *commandVolumeDeleteEmpty) Do(args []string, commandEnv *CommandEnv, wri
return
}
func isEmptyVolumeDeleteCandidate(v *master_pb.VolumeInformationMessage, quietSeconds, nowUnixSeconds int64, collectionPattern string) bool {
return matchesVolumeCollectionPattern(collectionPattern, v.Collection) &&
func isEmptyVolumeDeleteCandidate(v *master_pb.VolumeInformationMessage, quietSeconds, nowUnixSeconds int64, collectionMatcher *wildcard.CollectionMatcher) bool {
return collectionMatcher.Matches(v.Collection) &&
v.Size <= super_block.SuperBlockSize &&
v.ModifiedAtSecond > 0 &&
v.ModifiedAtSecond+quietSeconds < nowUnixSeconds
weed/shell/command_volume_delete_empty_test.go
+8 -1
View File
@@ -4,6 +4,7 @@ import (
"testing"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
)
func TestIsEmptyVolumeDeleteCandidateCollectionPattern(t *testing.T) {
@@ -26,13 +27,19 @@ func TestIsEmptyVolumeDeleteCandidateCollectionPattern(t *testing.T) {
{name: "wildcard rejects collection", pattern: "important*", collection: "other-logs", want: false},
{name: "default pattern matches empty collection", pattern: CollectionDefault, collection: "", want: true},
{name: "default pattern rejects named collection", pattern: CollectionDefault, collection: "important-logs", want: false},
{name: "list matches a listed collection", pattern: "other-logs,important-logs", collection: "important-logs", want: true},
{name: "list rejects an unlisted collection", pattern: "other-logs,important-logs", collection: "audit-logs", want: false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
v := *quietEmptyVolume
v.Collection = tt.collection
if got := isEmptyVolumeDeleteCandidate(&v, quietSeconds, now, tt.pattern); got != tt.want {
matcher, err := wildcard.CompileCollectionMatcher(tt.pattern)
if err != nil {
t.Fatalf("CompileCollectionMatcher(%q): %v", tt.pattern, err)
}
if got := isEmptyVolumeDeleteCandidate(&v, quietSeconds, now, matcher); got != tt.want {
t.Fatalf("isEmptyVolumeDeleteCandidate(collection=%q, pattern=%q) = %v, want %v",
tt.collection, tt.pattern, got, tt.want)
}
weed/shell/command_volume_fix_replication.go
+7 -15
View File
@@ -29,7 +29,7 @@ func init() {
}
type commandVolumeFixReplication struct {
collectionPattern *string
collectionMatcher *wildcard.CollectionMatcher
// TODO: move parameter flags here so we don't shuffle them around via function calls.
}
@@ -71,7 +71,7 @@ func (c *commandVolumeFixReplication) HasTag(tag CommandTag) bool {
func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
volFixReplicationCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
c.collectionPattern = volFixReplicationCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
collectionPattern := volFixReplicationCommand.String("collectionPattern", "", "comma-separated collection names, with '*' and '?' wildcards; empty matches all")
applyChanges := volFixReplicationCommand.Bool("apply", false, "apply the fix")
// TODO: remove this alias
applyChangesAlias := volFixReplicationCommand.Bool("force", false, "apply the fix (alias for -apply)")
@@ -87,6 +87,10 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
return nil
}
if c.collectionMatcher, err = wildcard.CompileCollectionMatcher(*collectionPattern); err != nil {
return err
}
handleDeprecatedForceFlag(writer, volFixReplicationCommand, applyChangesAlias, applyChanges)
infoAboutSimulationMode(writer, *applyChanges, "-apply")
commandEnv.noLock = !*applyChanges
@@ -127,7 +131,7 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
replica := replicas[0]
// Filter here so the termination counter matches what gets fixed; else -apply loops forever.
if !c.matchCollectionPattern(replica.info.Collection) {
if !c.collectionMatcher.Matches(replica.info.Collection) {
continue
}
@@ -330,18 +334,6 @@ func checkOneVolume(a *VolumeReplica, b *VolumeReplica, writer io.Writer, comman
return
}
// matchCollectionPattern reports whether collection matches -collectionPattern:
// empty matches everything, CollectionDefault matches the unnamed collection.
func (c *commandVolumeFixReplication) matchCollectionPattern(collection string) bool {
if *c.collectionPattern == "" {
return true
}
if *c.collectionPattern == CollectionDefault {
return collection == ""
}
return wildcard.MatchesWildcard(*c.collectionPattern, collection)
}
// deleteOneVolume trims one replica from each of the given volumes, and
// reports how many replicas it actually deleted.
func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, writer io.Writer, applyChanges bool, doCheck bool, volumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, selectOneVolumeFn SelectOneVolumeFunc) (deleted int, err error) {
weed/shell/command_volume_fix_replication_collection_test.go
+15 -6
View File
@@ -1,8 +1,12 @@
package shell
import "testing"
import (
"testing"
func TestMatchCollectionPattern(t *testing.T) {
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
)
func TestFixReplicationCollectionPattern(t *testing.T) {
tests := []struct {
name string
pattern string
@@ -19,14 +23,19 @@ func TestMatchCollectionPattern(t *testing.T) {
{name: "prefix wildcard mismatch", pattern: "smart*", collection: "jfs-hdfs-test", expected: false},
{name: "single char wildcard match", pattern: "vol?", collection: "vol1", expected: true},
{name: "single char wildcard mismatch", pattern: "vol?", collection: "vol42", expected: false},
{name: "list matches a listed collection", pattern: "smart-highlevel-test,jfs-hdfs-test", collection: "jfs-hdfs-test", expected: true},
{name: "list rejects an unlisted collection", pattern: "smart-highlevel-test,jfs-hdfs-test", collection: "other", expected: false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
pattern := tt.pattern
c := &commandVolumeFixReplication{collectionPattern: &pattern}
if got := c.matchCollectionPattern(tt.collection); got != tt.expected {
t.Errorf("matchCollectionPattern(pattern=%q, collection=%q) = %v, want %v",
matcher, err := wildcard.CompileCollectionMatcher(tt.pattern)
if err != nil {
t.Fatalf("CompileCollectionMatcher(%q): %v", tt.pattern, err)
}
c := &commandVolumeFixReplication{collectionMatcher: matcher}
if got := c.collectionMatcher.Matches(tt.collection); got != tt.expected {
t.Errorf("collection pattern %q against collection %q = %v, want %v",
tt.pattern, tt.collection, got, tt.expected)
}
})
weed/shell/command_volume_fix_replication_parallel_test.go
+1 -1
View File
@@ -120,7 +120,7 @@ func TestFixUnderReplicatedVolumesInParallel(t *testing.T) {
volumeIds = append(volumeIds, vid)
}
c := &commandVolumeFixReplication{collectionPattern: new(string)}
c := &commandVolumeFixReplication{}
fixedVolumes, err := c.fixUnderReplicatedVolumes(nil, io.Discard, false, volumeIds, volumeReplicas, allLocations, 0, 0, 8, 1)
if err != nil {
t.Fatalf("fixUnderReplicatedVolumes: %v", err)
weed/shell/command_volume_list.go
+7 -13
View File
@@ -24,7 +24,7 @@ func init() {
}
type commandVolumeList struct {
collectionPattern *string
collectionMatcher *wildcard.CollectionMatcher
dataCenter *string
rack *string
dataNode *string
@@ -54,7 +54,7 @@ func (c *commandVolumeList) Do(args []string, commandEnv *CommandEnv, writer io.
volumeListCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
verbosityLevel := volumeListCommand.Int("v", 5, "verbose mode: 0, 1, 2, 3, 4, 5")
c.collectionPattern = volumeListCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
collectionPattern := volumeListCommand.String("collectionPattern", "", "comma-separated collection names, with '*' and '?' wildcards; empty matches all")
c.readonly = volumeListCommand.Bool("readonly", false, "show only readonly volumes")
c.writable = volumeListCommand.Bool("writable", false, "show only writable volumes")
c.volumeId = volumeListCommand.Uint64("volumeId", 0, "show only volume id")
@@ -66,6 +66,10 @@ func (c *commandVolumeList) Do(args []string, commandEnv *CommandEnv, writer io.
return nil
}
if c.collectionMatcher, err = wildcard.CompileCollectionMatcher(*collectionPattern); err != nil {
return err
}
// collect topology information
var topologyInfo *master_pb.TopologyInfo
topologyInfo, c.volumeSizeLimitMb, err = collectTopologyInfo(commandEnv, 0)
@@ -319,7 +323,7 @@ func (c *commandVolumeList) isNotMatchDiskInfo(readOnly bool, collection string,
if *c.writable && (readOnly || volumeSize == -1 || (c.volumeSizeLimitMb > 0 && uint64(volumeSize) >= c.volumeSizeLimitMb*util.MiByte)) {
return true
}
if !matchesVolumeCollectionPattern(*c.collectionPattern, collection) {
if !c.collectionMatcher.Matches(collection) {
return true
}
if *c.volumeId > 0 && *c.volumeId != uint64(volumeId) {
@@ -328,16 +332,6 @@ func (c *commandVolumeList) isNotMatchDiskInfo(readOnly bool, collection string,
return false
}
func matchesVolumeCollectionPattern(pattern, collection string) bool {
if pattern == "" {
return true
}
if pattern == CollectionDefault {
return collection == ""
}
return wildcard.MatchesWildcard(pattern, collection)
}
func (c *commandVolumeList) writeDiskInfo(writer io.Writer, t *master_pb.DiskInfo, verbosityLevel int, outNodeInfo func()) statistics {
var s statistics
diskType := t.Type
weed/shell/command_volume_list_test.go
-2
View File
@@ -160,7 +160,6 @@ func TestWriteDataNodeInfo_SplitsCollapsedDisksByPhysicalDiskId(t *testing.T) {
c := &commandVolumeList{}
fs := flag.NewFlagSet("volume.list", flag.ContinueOnError)
c.collectionPattern = fs.String("collection", "", "")
c.dataCenter = fs.String("dataCenter", "", "")
c.rack = fs.String("rack", "", "")
c.dataNode = fs.String("dataNode", "", "")
@@ -205,7 +204,6 @@ func TestWriteTopologyInfo_PrintsParentHeadersOnce(t *testing.T) {
c := &commandVolumeList{}
fs := flag.NewFlagSet("volume.list", flag.ContinueOnError)
c.collectionPattern = fs.String("collection", "", "")
c.dataCenter = fs.String("dataCenter", "", "")
c.rack = fs.String("rack", "", "")
c.dataNode = fs.String("dataNode", "", "")
weed/shell/command_volume_tier_compact.go
+5 -6
View File
@@ -62,7 +62,7 @@ func (c *commandVolumeTierCompact) Do(args []string, commandEnv *CommandEnv, wri
tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
volumeId := tierCommand.Int("volumeId", 0, "the volume id")
collection := tierCommand.String("collection", "", "the collection name (supports regex)")
collection := tierCommand.String("collection", "", "comma-separated collection names, wildcards, or regex patterns; empty matches the collection with no name")
garbageThreshold := tierCommand.Float64("garbageThreshold", 0.3, "compact when garbage ratio exceeds this value")
if err = tierCommand.Parse(args); err != nil {
return nil
@@ -120,14 +120,13 @@ func (c *commandVolumeTierCompact) Do(args []string, commandEnv *CommandEnv, wri
}
func findRemoteVolumeInTopology(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, collectionPattern string) (remoteVolumeInfo, bool, error) {
// when collectionPattern is provided, compile and use as regex filter
var matchesCollection func(string) bool
if collectionPattern != "" {
collectionRegex, err := compileCollectionPattern(collectionPattern)
collectionMatcher, err := compileCollectionPattern(collectionPattern)
if err != nil {
return remoteVolumeInfo{}, false, fmt.Errorf("invalid collection pattern '%s': %v", collectionPattern, err)
}
matchesCollection = collectionRegex.MatchString
matchesCollection = collectionMatcher.Matches
} else {
matchesCollection = func(string) bool { return true }
}
@@ -161,7 +160,7 @@ func findRemoteVolumeInTopology(topoInfo *master_pb.TopologyInfo, vid needle.Vol
}
func collectRemoteVolumesWithInfo(topoInfo *master_pb.TopologyInfo, collectionPattern string) ([]remoteVolumeInfo, error) {
collectionRegex, err := compileCollectionPattern(collectionPattern)
collectionMatcher, err := compileCollectionPattern(collectionPattern)
if err != nil {
return nil, fmt.Errorf("invalid collection pattern '%s': %v", collectionPattern, err)
}
@@ -174,7 +173,7 @@ func collectRemoteVolumesWithInfo(topoInfo *master_pb.TopologyInfo, collectionPa
if v.RemoteStorageName == "" {
continue
}
if !collectionRegex.MatchString(v.Collection) {
if !collectionMatcher.Matches(v.Collection) {
continue
}
if seen[v.Id] {
weed/shell/command_volume_tier_download.go
+3 -3
View File
@@ -55,7 +55,7 @@ func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, wr
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")
if err = tierCommand.Parse(args); err != nil {
return nil
}
@@ -95,7 +95,7 @@ func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, wr
func collectRemoteVolumes(topoInfo *master_pb.TopologyInfo, collectionPattern string) (vids []needle.VolumeId, err error) {
// compile regex pattern for collection matching
collectionRegex, err := compileCollectionPattern(collectionPattern)
collectionMatcher, err := compileCollectionPattern(collectionPattern)
if err != nil {
return nil, fmt.Errorf("invalid collection pattern '%s': %v", collectionPattern, err)
}
@@ -104,7 +104,7 @@ func collectRemoteVolumes(topoInfo *master_pb.TopologyInfo, collectionPattern st
eachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
for _, diskInfo := range dn.DiskInfos {
for _, v := range diskInfo.VolumeInfos {
if collectionRegex.MatchString(v.Collection) && v.RemoteStorageName != "" {
if collectionMatcher.Matches(v.Collection) && v.RemoteStorageName != "" {
vidMap[v.Id] = true
}
}
weed/shell/command_volume_tier_move.go
+12 -20
View File
@@ -6,7 +6,6 @@ import (
"flag"
"fmt"
"io"
"path/filepath"
"sync"
"time"
@@ -18,6 +17,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
@@ -66,8 +66,9 @@ func (c *commandVolumeTierMove) Help() string {
replication setting. Otherwise, the volume's existing replication is preserved.
Note:
Use -collectionPattern="_default" to match only the default collection (volumes with no collection name).
Empty collectionPattern matches all collections.
-collectionPattern is a comma-separated list of collection names, with "*" and "?"
wildcards, and regex patterns. Use "_default" to match only the default collection
(volumes with no collection name). An empty pattern matches all collections.
`
}
@@ -79,7 +80,7 @@ func (c *commandVolumeTierMove) HasTag(CommandTag) bool {
func (c *commandVolumeTierMove) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
collectionPattern := tierCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
collectionPattern := tierCommand.String("collectionPattern", "", "comma-separated collection names, with '*' and '?' wildcards; empty matches all")
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")
source := tierCommand.String("fromDiskType", "", "the source disk type")
@@ -561,6 +562,11 @@ func (c *commandVolumeTierMove) ensureReplicationFulfilled(commandEnv *CommandEn
func collectVolumeIdsForTierChange(topologyInfo *master_pb.TopologyInfo, volumeSizeLimitMb uint64, sourceTier types.DiskType, sourceDataCenter string, collectionPattern string, fullPercentage float64, quietPeriod time.Duration) (vids []needle.VolumeId, err error) {
collectionMatcher, err := wildcard.CompileCollectionMatcher(collectionPattern)
if err != nil {
return nil, err
}
quietSeconds := int64(quietPeriod / time.Second)
nowUnixSeconds := time.Now().Unix()
@@ -573,22 +579,8 @@ func collectVolumeIdsForTierChange(topologyInfo *master_pb.TopologyInfo, volumeS
}
for _, diskInfo := range dn.DiskInfos {
for _, v := range diskInfo.VolumeInfos {
// check collection name pattern
if collectionPattern != "" {
var matched bool
if collectionPattern == CollectionDefault {
matched = v.Collection == ""
} else {
var matchErr error
matched, matchErr = filepath.Match(collectionPattern, v.Collection)
if matchErr != nil {
err = fmt.Errorf("collection pattern %q failed to match: %w", collectionPattern, matchErr)
return
}
}
if !matched {
continue
}
if !collectionMatcher.Matches(v.Collection) {
continue
}
if v.ModifiedAtSecond+quietSeconds < nowUnixSeconds && types.ToDiskType(v.DiskType) == sourceTier {
weed/shell/common.go
+2 -2
View File
@@ -2,6 +2,7 @@ package shell
import (
"github.com/seaweedfs/seaweedfs/weed/util"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
)
var (
@@ -10,8 +11,7 @@ var (
// Default number of volumes EC encode/decode process per batch.
DefaultEcBatchSize = 10
// CollectionDefault is the special keyword to match empty collection names.
// Use "_default" to avoid collision with a literal collection named "default".
CollectionDefault = "_default"
CollectionDefault = wildcard.CollectionDefault
)
// ErrorWaitGroup lives in weed/util so packages outside the shell can share it.
weed/util/wildcard/collection_filter.go
+246
View File
@@ -0,0 +1,246 @@
package wildcard
import (
"fmt"
"regexp"
"strings"
)
// CollectionFilterMode controls how collections are interpreted during
// 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 a comma-separated list of collection patterns.
type CollectionFilterMode string
const (
CollectionFilterAll CollectionFilterMode = "ALL_COLLECTIONS"
CollectionFilterEach CollectionFilterMode = "EACH_COLLECTION"
// CollectionDefault is the entry that matches the empty-named collection.
// "_default" avoids colliding with a collection literally named "default".
CollectionDefault = "_default"
)
// regexMetaCharacters mark a filter entry as a regex; "*" and "?" stay wildcards.
// "." is not one of them: a collection named "my.bucket" is a name, not a
// pattern that would also match "my-bucket". A "." only counts as regex syntax
// when it is quantified, as in "bucket.*".
const regexMetaCharacters = `^$+()[]{}|\`
// CollectionMatcher matches a volume collection against a collection_filter value.
type CollectionMatcher struct {
filter string
matchEmpty bool
literals []string
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, CollectionDefault for the empty-named collection, or a regex
// when it carries regex syntax. A regex entry must match the whole name unless
// it anchors itself with "^" or "$", and always matches its own spelling too,
// so a collection named "logs(2024)" stays reachable by name.
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{filter: trimmed}
for _, entry := range splitCollectionEntries(trimmed) {
entry = strings.TrimSpace(entry)
if entry == "" {
continue
}
if entry == CollectionDefault {
matcher.matchEmpty = true
continue
}
if !isRegexEntry(entry) {
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.literals = append(matcher.literals, entry)
matcher.regexes = append(matcher.regexes, compiled)
}
if !matcher.matchEmpty && len(matcher.wildcards) == 0 && len(matcher.regexes) == 0 {
// Only a genuinely empty filter means "every collection". A value like ","
// is a typo, and matching everything on it would encode or delete far more
// than the operator asked for.
return nil, fmt.Errorf("collection_filter %q has no collection names", trimmed)
}
return matcher, nil
}
// splitCollectionEntries splits a filter on the commas that separate entries.
// A comma inside a closed regex construct - a character class, a repetition
// count, or a group - belongs to that construct. An unclosed one is a literal
// brace or bracket in a name, so the commas after it still separate entries.
func splitCollectionEntries(filter string) []string {
entries := make([]string, 0, 4)
start := 0
for i := 0; i < len(filter); i++ {
switch filter[i] {
case '\\':
i++
case '[':
if end := classEnd(filter, i); end > i {
i = end
}
case '{':
if end := repeatCountEnd(filter, i); end > i {
i = end
}
case '(':
if end := groupEnd(filter, i); end > i {
i = end
}
case ',':
entries = append(entries, filter[start:i])
start = i + 1
}
}
return append(entries, filter[start:])
}
// classEnd returns the index of the "]" closing the character class opened at
// start, or -1 when it is never closed. It follows the regexp parser: a "]"
// leading the class is a member of it, and a POSIX class such as "[:alpha:]"
// carries a "]" of its own.
func classEnd(filter string, start int) int {
i := start + 1
if i < len(filter) && filter[i] == '^' {
i++
}
if i < len(filter) && filter[i] == ']' {
i++
}
for ; i < len(filter); i++ {
switch filter[i] {
case '\\':
i++
case '[':
if end := posixClassEnd(filter, i); end > i {
i = end
}
case ']':
return i
}
}
return -1
}
// posixClassEnd returns the index of the "]" closing the POSIX class opened at
// start, or -1 when start does not open one.
func posixClassEnd(filter string, start int) int {
if !strings.HasPrefix(filter[start:], "[:") {
return -1
}
end := strings.Index(filter[start+2:], ":]")
if end < 0 {
return -1
}
return start + 2 + end + 1
}
// repeatCountEnd returns the index of the "}" closing the repetition count
// opened at start, or -1 when what follows is not a count.
func repeatCountEnd(filter string, start int) int {
for i := start + 1; i < len(filter); i++ {
if filter[i] == '}' {
return i
}
if filter[i] != ',' && (filter[i] < '0' || filter[i] > '9') {
return -1
}
}
return -1
}
// groupEnd returns the index of the ")" closing the group opened at start, or
// -1 when it is never closed.
func groupEnd(filter string, start int) int {
depth := 0
for i := start; i < len(filter); i++ {
switch filter[i] {
case '\\':
i++
case '[':
// a ")" inside a character class is a literal, not the group's end
if end := classEnd(filter, i); end > i {
i = end
}
case '(':
depth++
case ')':
if depth--; depth == 0 {
return i
}
}
}
return -1
}
// isRegexEntry reports whether an entry carries regex syntax rather than being a
// plain collection name with optional "*" and "?" wildcards.
func isRegexEntry(entry string) bool {
if strings.ContainsAny(entry, regexMetaCharacters) {
return true
}
for i := 0; i+1 < len(entry); i++ {
if entry[i] == '.' && (entry[i+1] == '*' || entry[i+1] == '?') {
return true
}
}
return false
}
// 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
}
if m.matchEmpty && collection == "" {
return true
}
for _, literal := range m.literals {
if literal == collection {
return true
}
}
for _, pattern := range m.wildcards {
if MatchesWildcard(pattern, collection) {
return true
}
}
for _, re := range m.regexes {
if re.MatchString(collection) {
return true
}
}
return false
}
// String returns the filter this matcher was compiled from.
func (m *CollectionMatcher) String() string {
if m == nil {
return ""
}
return m.filter
}
weed/util/wildcard/collection_filter_test.go
+74
View File
@@ -0,0 +1,74 @@
package wildcard
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"}},
// A dot is part of a collection name unless it is quantified.
{filter: "my.bucket", matches: []string{"my.bucket"}, misses: []string{"my-bucket", "myxbucket"}},
{filter: "my.bucket,videos", matches: []string{"my.bucket", "videos"}, misses: []string{"my-bucket"}},
// A name made of regex syntax is still reachable by its own spelling.
{filter: "logs(2024)", matches: []string{"logs(2024)", "logs2024"}, misses: []string{"logs"}},
// Escaping reaches a name whose regex syntax does not parse on its own.
{filter: `logs\(2024`, matches: []string{"logs(2024"}, misses: []string{"logs2024", "logs"}},
// A comma inside a character class or a repetition count is not a separator.
{filter: "bucket[0-9]{1,3}", matches: []string{"bucket1", "bucket123"}, misses: []string{"bucket", "bucketx", "bucket1234"}},
{filter: "[a,b]", matches: []string{"a", "b", ","}, misses: []string{"ab", "c"}},
{filter: "bucket[0-9]{1,3},videos", matches: []string{"bucket7", "videos"}, misses: []string{"bucketx"}},
{filter: "bucket(foo,bar)", matches: []string{"bucketfoo,bar", "bucket(foo,bar)"}, misses: []string{"bucketfoo"}},
{filter: "logs(2024),videos", matches: []string{"logs(2024)", "logs2024", "videos"}, misses: []string{"logs"}},
{filter: "(a[)],b)", matches: []string{"a),b"}, misses: []string{"a", "b"}},
{filter: "(a[]),],b)", matches: []string{"a],b", "a),b"}, misses: []string{"a,b"}},
{filter: "x[[:alpha:],b]y", matches: []string{"xay", "xby", "x,y"}, misses: []string{"xy"}},
// A comma separates entries unless it is escaped, so a name holding one is
// reachable as "\(a\,b\)".
{filter: `\(a\,b\)`, matches: []string{"(a,b)"}, misses: []string{"(a", "b)"}},
// An unclosed brace is a literal in a name, so the comma after it still separates.
{filter: "foo{bar,videos", matches: []string{"foo{bar", "videos"}, misses: []string{"foo"}},
{filter: "foo{2,videos", matches: []string{"foo{2", "videos"}, misses: []string{"foo"}},
{filter: CollectionDefault, matches: []string{""}, misses: []string{"photos"}},
{filter: "photos," + CollectionDefault, matches: []string{"photos", ""}, misses: []string{"videos"}},
}
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) {
for _, filter := range []string{"photos,[invalid", ",", ", ,"} {
if _, err := CompileCollectionMatcher(filter); err == nil {
t.Errorf("filter %q should not compile", filter)
}
}
}
weed/worker/tasks/balance/plugin_handler.go
+2 -1
View File
@@ -16,6 +16,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
workertypes "github.com/seaweedfs/seaweedfs/weed/worker/types"
"google.golang.org/grpc"
"google.golang.org/protobuf/proto"
@@ -313,7 +314,7 @@ func (h *VolumeBalanceHandler) Detect(
var results []*workertypes.TaskDetectionResult
var hasMore bool
if pluginworker.CollectionFilterMode(collectionFilter) == pluginworker.CollectionFilterEach {
if wildcard.CollectionFilterMode(collectionFilter) == wildcard.CollectionFilterEach {
// Group metrics by collection in a single pass (O(N) instead of O(C*N))
metricsByCollection := make(map[string][]*workertypes.VolumeHealthMetrics)
for _, m := range metrics {
weed/worker/tasks/ec_balance/detection.go
+2 -3
View File
@@ -9,7 +9,6 @@ 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"
@@ -49,7 +48,7 @@ func Detection(
return nil, false, fmt.Errorf("topology info not available")
}
allowedCollections, err := pluginworker.CompileCollectionMatcher(ecConfig.CollectionFilter)
allowedCollections, err := wildcard.CompileCollectionMatcher(ecConfig.CollectionFilter)
if err != nil {
return nil, false, err
}
@@ -154,7 +153,7 @@ 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, allowedCollections *pluginworker.CollectionMatcher) (*ecbalancer.Topology, int, func(collection string, vid uint32) (int, int)) {
func buildBalancerTopology(topoInfo *master_pb.TopologyInfo, config *Config, allowedCollections *wildcard.CollectionMatcher) (*ecbalancer.Topology, int, func(collection string, vid uint32) (int, int)) {
topo := ecbalancer.NewTopology()
type volRatioKey struct {
weed/worker/tasks/ec_balance/detection_test.go
+2 -2
View File
@@ -6,9 +6,9 @@ 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/util/wildcard"
"github.com/seaweedfs/seaweedfs/weed/worker/types"
)
@@ -100,7 +100,7 @@ func TestBuildBalancerTopologyGroupsByHost(t *testing.T) {
func TestBuildBalancerTopologyCollectionFilter(t *testing.T) {
config := NewDefaultConfig()
config.CollectionFilter = "other" // does not match the volume's collection
allowed, err := pluginworker.CompileCollectionMatcher(config.CollectionFilter)
allowed, err := wildcard.CompileCollectionMatcher(config.CollectionFilter)
if err != nil {
t.Fatalf("CompileCollectionMatcher: %v", err)
}
weed/worker/tasks/erasure_coding/detection.go
+2 -2
View File
@@ -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,7 @@ 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, err := pluginworker.CompileCollectionMatcher(ecConfig.CollectionFilter)
allowedCollections, err := wildcard.CompileCollectionMatcher(ecConfig.CollectionFilter)
if err != nil {
return nil, false, err
}
weed/worker/tasks/erasure_coding/plugin_handler.go
+2 -1
View File
@@ -14,6 +14,7 @@ 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"
@@ -295,7 +296,7 @@ func emitErasureCodingDetectionDecisionTrace(
quietThreshold := time.Duration(taskConfig.QuietForSeconds) * time.Second
minSizeBytes := uint64(taskConfig.MinSizeMB) * 1024 * 1024
allowedCollections, err := pluginworker.CompileCollectionMatcher(taskConfig.CollectionFilter)
allowedCollections, err := wildcard.CompileCollectionMatcher(taskConfig.CollectionFilter)
if err != nil {
return err
}