feat: support marking volumes by collection (#9585)

* feat: add collection.mark shell command

Add collection.mark to mark all existing normal volume replicas in a collection as readonly or writable. The command runs in preview mode by default and requires -apply to execute changes. It reuses existing volume mark RPCs, supports default collection aliases, skips EC shards, and adds unit tests for option parsing and target collection logic.

* Revert "feat: add collection.mark shell command"

This reverts commit 50c2bbf94c.

* feat: support marking volumes by collection

Add a -collection option to volume.mark so operators can mark every normal volume replica in a collection using existing topology data and volume mark RPCs.

The change keeps the single-volume path unchanged and adds tests for collection target selection, EC shard exclusion, and argument validation.

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

* volume.mark: reuse eachDataNode for collection traversal

* volume.mark: continue past per-volume failures and report progress

Collection marking aborted on the first failed RPC, leaving the
collection half-marked with no record of which volumes succeeded.
Mark every reachable volume, print per-volume progress to the writer,
and return an aggregated error naming the failures.

* volume.mark: let -collection _default target the unnamed collection

Other volume commands use the _default sentinel to match volumes with
no named collection; volume.mark could not reach them at all. Map
_default to the empty collection name in the filter.

---------

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

weed/shell/command_volume_mark.go

@@ -1,11 +1,13 @@
 package shell
 
 import (
+	"errors"
 	"flag"
 	"fmt"
 	"io"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 )
@@ -22,9 +24,12 @@ func (c *commandVolumeMark) Name() string {
 }
 
 func (c *commandVolumeMark) Help() string {
-	return `Mark volume writable or readonly from one volume server
+	return `Mark volume writable or readonly from one volume server, or all volume replicas in one collection
 
 	volume.mark -node <volume server host:port> -volumeId <volume id> -writable or -readonly
+	volume.mark -collection <collection> -writable or -readonly
+
+	Use -collection ` + CollectionDefault + ` to target volumes that belong to no named collection.
 `
 }
 
@@ -37,11 +42,23 @@ func (c *commandVolumeMark) Do(args []string, commandEnv *CommandEnv, writer io.
 	volMarkCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeIdInt := volMarkCommand.Int("volumeId", 0, "the volume id")
 	nodeStr := volMarkCommand.String("node", "", "the volume server <host>:<port>")
+	collection := volMarkCommand.String("collection", "", "the collection name")
 	writable := volMarkCommand.Bool("writable", false, "volume mark writable")
 	readonly := volMarkCommand.Bool("readonly", false, "volume mark readonly")
 	if err = volMarkCommand.Parse(args); err != nil {
 		return nil
 	}
+	collectionSet, nodeSet, volumeIdSet := false, false, false
+	volMarkCommand.Visit(func(f *flag.Flag) {
+		switch f.Name {
+		case "collection":
+			collectionSet = true
+		case "node":
+			nodeSet = true
+		case "volumeId":
+			volumeIdSet = true
+		}
+	})
 	markWritable := false
 	if (*writable && *readonly) || (!*writable && !*readonly) {
 		return fmt.Errorf("use -readonly or -writable")
@@ -49,13 +66,91 @@ func (c *commandVolumeMark) Do(args []string, commandEnv *CommandEnv, writer io.
 		markWritable = true
 	}
 
+	if collectionSet {
+		if *collection == "" {
+			return fmt.Errorf("collection is required")
+		}
+		if nodeSet || volumeIdSet {
+			return fmt.Errorf("cannot use -collection with -node or -volumeId")
+		}
+	} else if !nodeSet || !volumeIdSet || *nodeStr == "" || *volumeIdInt == 0 {
+		return fmt.Errorf("use -node and -volumeId, or -collection")
+	}
+
 	if err = commandEnv.confirmIsLocked(args); err != nil {
 		return
 	}
 
+	if collectionSet {
+		topologyInfo, _, err := collectTopologyInfo(commandEnv, 0)
+		if err != nil {
+			return err
+		}
+		targets, err := collectVolumeMarkTargetsByCollection(topologyInfo, *collection)
+		if err != nil {
+			return err
+		}
+		state := "readonly"
+		if markWritable {
+			state = "writable"
+		}
+		var failures []error
+		for _, target := range targets {
+			if err := markVolumeWritable(commandEnv.option.GrpcDialOption, target.volumeId, target.sourceVolumeServer, markWritable, true); err != nil {
+				failures = append(failures, fmt.Errorf("mark volume %d on %s: %w", target.volumeId, target.sourceVolumeServer, err))
+				fmt.Fprintf(writer, "volume %d on %s: %v\n", target.volumeId, target.sourceVolumeServer, err)
+				continue
+			}
+			fmt.Fprintf(writer, "volume %d on %s marked %s\n", target.volumeId, target.sourceVolumeServer, state)
+		}
+		if len(failures) > 0 {
+			return fmt.Errorf("marked %d of %d volumes %s, %d failed: %w", len(targets)-len(failures), len(targets), state, len(failures), errors.Join(failures...))
+		}
+		return nil
+	}
+
 	sourceVolumeServer := pb.ServerAddress(*nodeStr)
 
 	volumeId := needle.VolumeId(*volumeIdInt)
 
 	return markVolumeWritable(commandEnv.option.GrpcDialOption, volumeId, sourceVolumeServer, markWritable, true)
 }
+
+type volumeMarkTarget struct {
+	volumeId           needle.VolumeId
+	sourceVolumeServer pb.ServerAddress
+}
+
+func collectVolumeMarkTargetsByCollection(topoInfo *master_pb.TopologyInfo, collection string) ([]volumeMarkTarget, error) {
+	if collection == "" {
+		return nil, fmt.Errorf("collection is required")
+	}
+
+	// _default targets volumes that belong to no named collection.
+	matchCollection := collection
+	if matchCollection == CollectionDefault {
+		matchCollection = ""
+	}
+
+	var targets []volumeMarkTarget
+	eachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
+		if dn == nil {
+			return
+		}
+		sourceVolumeServer := pb.NewServerAddressFromDataNode(dn)
+		for _, diskInfo := range dn.GetDiskInfos() {
+			for _, v := range diskInfo.GetVolumeInfos() {
+				if v.GetCollection() == matchCollection {
+					targets = append(targets, volumeMarkTarget{
+						volumeId:           needle.VolumeId(v.GetId()),
+						sourceVolumeServer: sourceVolumeServer,
+					})
+				}
+			}
+		}
+	})
+	if len(targets) == 0 {
+		return nil, fmt.Errorf("collection %s has no volumes", collection)
+	}
+	return targets, nil
+}

weed/shell/command_volume_mark_test.go

@@ -0,0 +1,167 @@
+package shell
+
+import (
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+)
+
+func TestCollectVolumeMarkTargetsByCollection(t *testing.T) {
+	topo := &master_pb.TopologyInfo{
+		DataCenterInfos: []*master_pb.DataCenterInfo{
+			{
+				RackInfos: []*master_pb.RackInfo{
+					{
+						DataNodeInfos: []*master_pb.DataNodeInfo{
+							{
+								Id:       "node-a:8080",
+								Address:  "10.0.0.1:8080",
+								GrpcPort: 19080,
+								DiskInfos: map[string]*master_pb.DiskInfo{
+									"hdd": {
+										VolumeInfos: []*master_pb.VolumeInformationMessage{
+											{Id: 1, Collection: "photos"},
+											{Id: 2, Collection: "docs"},
+										},
+									},
+									"ssd": {
+										VolumeInfos: []*master_pb.VolumeInformationMessage{
+											{Id: 3, Collection: "photos"},
+										},
+									},
+								},
+							},
+							{
+								Id: "node-b:8080",
+								DiskInfos: map[string]*master_pb.DiskInfo{
+									"hdd": {
+										VolumeInfos: []*master_pb.VolumeInformationMessage{
+											{Id: 1, Collection: "photos"},
+										},
+										EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{
+											{Id: 4, Collection: "photos"},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	targets, err := collectVolumeMarkTargetsByCollection(topo, "photos")
+	if err != nil {
+		t.Fatalf("collectVolumeMarkTargetsByCollection: %v", err)
+	}
+
+	got := map[string]int{}
+	for _, target := range targets {
+		got[string(target.sourceVolumeServer)+"#"+target.volumeId.String()]++
+	}
+	want := map[string]int{
+		string(pb.ServerAddress("10.0.0.1:8080.19080")) + "#" + needle.VolumeId(1).String(): 1,
+		string(pb.ServerAddress("10.0.0.1:8080.19080")) + "#" + needle.VolumeId(3).String(): 1,
+		string(pb.ServerAddress("node-b:8080")) + "#" + needle.VolumeId(1).String():         1,
+	}
+	if len(got) != len(want) {
+		t.Fatalf("got %v targets, want %v", got, want)
+	}
+	for key, wantCount := range want {
+		if got[key] != wantCount {
+			t.Fatalf("target %s count %d, want %d; all targets: %v", key, got[key], wantCount, got)
+		}
+	}
+}
+
+func TestCollectVolumeMarkTargetsByCollectionMatchesDefaultCollection(t *testing.T) {
+	topo := &master_pb.TopologyInfo{
+		DataCenterInfos: []*master_pb.DataCenterInfo{
+			{
+				RackInfos: []*master_pb.RackInfo{
+					{
+						DataNodeInfos: []*master_pb.DataNodeInfo{
+							{
+								Id: "node-a:8080",
+								DiskInfos: map[string]*master_pb.DiskInfo{
+									"hdd": {
+										VolumeInfos: []*master_pb.VolumeInformationMessage{
+											{Id: 1, Collection: ""},
+											{Id: 2, Collection: "photos"},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	targets, err := collectVolumeMarkTargetsByCollection(topo, CollectionDefault)
+	if err != nil {
+		t.Fatalf("collectVolumeMarkTargetsByCollection: %v", err)
+	}
+	if len(targets) != 1 || targets[0].volumeId != needle.VolumeId(1) {
+		t.Fatalf("expected only volume 1 from the default collection, got %v", targets)
+	}
+}
+
+func TestCollectVolumeMarkTargetsByCollectionRejectsEmptyCollection(t *testing.T) {
+	_, err := collectVolumeMarkTargetsByCollection(&master_pb.TopologyInfo{}, "")
+	if err == nil || !strings.Contains(err.Error(), "collection is required") {
+		t.Fatalf("expected collection required error, got %v", err)
+	}
+}
+
+func TestCollectVolumeMarkTargetsByCollectionReportsMissingCollection(t *testing.T) {
+	_, err := collectVolumeMarkTargetsByCollection(&master_pb.TopologyInfo{}, "missing")
+	if err == nil || !strings.Contains(err.Error(), "collection missing has no volumes") {
+		t.Fatalf("expected missing collection error, got %v", err)
+	}
+}
+
+func TestVolumeMarkRejectsCollectionWithNodeOrVolumeId(t *testing.T) {
+	err := (&commandVolumeMark{}).Do([]string{
+		"-collection", "photos",
+		"-node", "127.0.0.1:8080",
+		"-readonly",
+	}, nil, io.Discard)
+	if err == nil || !strings.Contains(err.Error(), "cannot use -collection with -node or -volumeId") {
+		t.Fatalf("expected collection conflict error, got %v", err)
+	}
+
+	err = (&commandVolumeMark{}).Do([]string{
+		"-collection", "photos",
+		"-volumeId", "1",
+		"-readonly",
+	}, nil, io.Discard)
+	if err == nil || !strings.Contains(err.Error(), "cannot use -collection with -node or -volumeId") {
+		t.Fatalf("expected collection conflict error, got %v", err)
+	}
+
+	err = (&commandVolumeMark{}).Do([]string{
+		"-collection", "photos",
+		"-volumeId", "0",
+		"-readonly",
+	}, nil, io.Discard)
+	if err == nil || !strings.Contains(err.Error(), "cannot use -collection with -node or -volumeId") {
+		t.Fatalf("expected collection conflict error, got %v", err)
+	}
+}
+
+func TestVolumeMarkRejectsEmptyCollectionFlag(t *testing.T) {
+	err := (&commandVolumeMark{}).Do([]string{
+		"-collection", "",
+		"-readonly",
+	}, nil, io.Discard)
+	if err == nil || !strings.Contains(err.Error(), "collection is required") {
+		t.Fatalf("expected collection required error, got %v", err)
+	}
+}