feat: add collection pattern to delete empty volumes (#10129)

* feat: add collection pattern to delete empty volumes

Co-authored-by: Codex <noreply@openai.com>

* shell: match collection pattern with wildcard matcher

Use wildcard.MatchesWildcard in the shared collection-pattern helper,
matching command_volume_fix_replication's matchCollectionPattern. The
flag only advertises '*' and '?', which is exactly what the matcher
supports.

---------

Co-authored-by: Codex <noreply@openai.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/shell/command_volume_delete_empty.go

@@ -27,6 +27,7 @@ func (c *commandVolumeDeleteEmpty) Help() string {
 	return `delete empty volumes from all volume servers
 
 	volume.deleteEmpty -quietFor=24h -apply
+	volume.deleteEmpty -collectionPattern=important* -quietFor=24h -apply
 
 	This command deletes all empty volumes from one volume server.
 
@@ -41,6 +42,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 '?'")
 	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)")
@@ -67,7 +69,7 @@ func (c *commandVolumeDeleteEmpty) Do(args []string, commandEnv *CommandEnv, wri
 	eachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
 		for _, diskInfo := range dn.DiskInfos {
 			for _, v := range diskInfo.VolumeInfos {
-				if v.Size <= super_block.SuperBlockSize && v.ModifiedAtSecond > 0 && v.ModifiedAtSecond+quietSeconds < nowUnixSeconds {
+				if isEmptyVolumeDeleteCandidate(v, quietSeconds, nowUnixSeconds, *collectionPattern) {
 					if *applyBalancing {
 						log.Printf("deleting empty volume %d from %s", v.Id, dn.Id)
 						if deleteErr := deleteVolume(commandEnv.option.GrpcDialOption, needle.VolumeId(v.Id),
@@ -85,3 +87,10 @@ 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) &&
+		v.Size <= super_block.SuperBlockSize &&
+		v.ModifiedAtSecond > 0 &&
+		v.ModifiedAtSecond+quietSeconds < nowUnixSeconds
+}