filter volume by local or remote storage name (#10946)

* filter volume by local or remote storage name

Signed-off-by: lou <alex1988@outlook.com>

* fix SelectsEverything

Signed-off-by: lou <alex1988@outlook.com>

* keep the proto sync out of this change

The branch copied weed/pb/*.proto over their seaweed-volume and Java
counterparts and regenerated every .pb.go with a different protoc and
protoc-gen-go-grpc. DiskStatus.error arriving that way broke the Rust
build, and the rest is toolchain churn in files this change has nothing
to say about.

---------

Signed-off-by: lou <alex1988@outlook.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

seaweed-volume/proto/master.proto

@@ -27,6 +27,8 @@ service Seaweed {
   }
   rpc VolumeList (VolumeListRequest) returns (VolumeListResponse) {
   }
+  rpc VolumeListStream (VolumeListRequest) returns (stream VolumeListStreamResponse) {
+  }
   rpc LookupEcVolume (LookupEcVolumeRequest) returns (LookupEcVolumeResponse) {
   }
   rpc VacuumVolume (VacuumVolumeRequest) returns (VacuumVolumeResponse) {
@@ -406,12 +408,40 @@ message TopologyInfo {
   map<string, DiskInfo> diskInfos = 3;
 }
 message VolumeListRequest {
+  // Empty and zero take everything. Only the volumes and ec shards listed
+  // under a disk are selected; the topology and its disk counters are always
+  // reported in full.
+  string collection = 1;
+  uint32 volume_id = 2;
+  // The one collection the empty string cannot name. A named collection wins.
+  bool default_collection_only = 3;
+  // Empty and zero take everything. Wildcards are supported.
+  string remote_storage_name = 4;
+  bool local_volume_only = 5;
 }
 message VolumeListResponse {
   TopologyInfo topology_info = 1;
   uint64 volume_size_limit_mb = 2;
 }
 
+// VolumeListStream answers the same request as VolumeList without either end
+// holding every volume in the cluster at once. At 800k volumes the reply is
+// 36MB on the wire but 305MB as messages, which the master built in full
+// before sending any of it.
+message VolumeListStreamResponse {
+  // Sent once, first, listing no volumes: the topology, its disks and their
+  // counters. Every message after carries volumes for one of those disks.
+  VolumeListResponse header = 1;
+  // Which disk this batch is from. A disk arrives over as many batches as it
+  // takes, so append rather than assign.
+  string data_center = 2;
+  string rack = 3;
+  string data_node = 4;
+  string disk_type = 5;
+  repeated VolumeInformationMessage volume_infos = 6;
+  repeated VolumeEcShardInformationMessage ec_shard_infos = 7;
+}
+
 message LookupEcVolumeRequest {
   uint32 volume_id = 1;
 }

weed/pb/master.proto

@@ -415,6 +415,9 @@ message VolumeListRequest {
   uint32 volume_id = 2;
   // The one collection the empty string cannot name. A named collection wins.
   bool default_collection_only = 3;
+  // Empty and zero take everything. Wildcards are supported.
+  string remote_storage_name = 4;
+  bool local_volume_only = 5;
 }
 message VolumeListResponse {
   TopologyInfo topology_info = 1;

weed/pb/master_pb/master.pb.go

@@ -2788,8 +2788,11 @@ type VolumeListRequest struct {
 	VolumeId   uint32 `protobuf:"varint,2,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"`
 	// The one collection the empty string cannot name. A named collection wins.
 	DefaultCollectionOnly bool `protobuf:"varint,3,opt,name=default_collection_only,json=defaultCollectionOnly,proto3" json:"default_collection_only,omitempty"`
-	unknownFields         protoimpl.UnknownFields
-	sizeCache             protoimpl.SizeCache
+	// Empty and zero take everything. Wildcards are supported.
+	RemoteStorageName string `protobuf:"bytes,4,opt,name=remote_storage_name,json=remoteStorageName,proto3" json:"remote_storage_name,omitempty"`
+	LocalVolumeOnly   bool   `protobuf:"varint,5,opt,name=local_volume_only,json=localVolumeOnly,proto3" json:"local_volume_only,omitempty"`
+	unknownFields     protoimpl.UnknownFields
+	sizeCache         protoimpl.SizeCache
 }
 
 func (x *VolumeListRequest) Reset() {
@@ -2843,6 +2846,20 @@ func (x *VolumeListRequest) GetDefaultCollectionOnly() bool {
 	return false
 }
 
+func (x *VolumeListRequest) GetRemoteStorageName() string {
+	if x != nil {
+		return x.RemoteStorageName
+	}
+	return ""
+}
+
+func (x *VolumeListRequest) GetLocalVolumeOnly() bool {
+	if x != nil {
+		return x.LocalVolumeOnly
+	}
+	return false
+}
+
 type VolumeListResponse struct {
 	state             protoimpl.MessageState `protogen:"open.v1"`
 	TopologyInfo      *TopologyInfo          `protobuf:"bytes,1,opt,name=topology_info,json=topologyInfo,proto3" json:"topology_info,omitempty"`
@@ -5215,13 +5232,15 @@ const file_master_proto_rawDesc = "" +
 	"\tdiskInfos\x18\x03 \x03(\v2&.master_pb.TopologyInfo.DiskInfosEntryR\tdiskInfos\x1aQ\n" +
 	"\x0eDiskInfosEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12)\n" +
-	"\x05value\x18\x02 \x01(\v2\x13.master_pb.DiskInfoR\x05value:\x028\x01\"\x88\x01\n" +
+	"\x05value\x18\x02 \x01(\v2\x13.master_pb.DiskInfoR\x05value:\x028\x01\"\xe4\x01\n" +
 	"\x11VolumeListRequest\x12\x1e\n" +
 	"\n" +
 	"collection\x18\x01 \x01(\tR\n" +
 	"collection\x12\x1b\n" +
 	"\tvolume_id\x18\x02 \x01(\rR\bvolumeId\x126\n" +
-	"\x17default_collection_only\x18\x03 \x01(\bR\x15defaultCollectionOnly\"\x83\x01\n" +
+	"\x17default_collection_only\x18\x03 \x01(\bR\x15defaultCollectionOnly\x12.\n" +
+	"\x13remote_storage_name\x18\x04 \x01(\tR\x11remoteStorageName\x12*\n" +
+	"\x11local_volume_only\x18\x05 \x01(\bR\x0flocalVolumeOnly\"\x83\x01\n" +
 	"\x12VolumeListResponse\x12<\n" +
 	"\rtopology_info\x18\x01 \x01(\v2\x17.master_pb.TopologyInfoR\ftopologyInfo\x12/\n" +
 	"\x14volume_size_limit_mb\x18\x02 \x01(\x04R\x11volumeSizeLimitMb\"\xda\x02\n" +

weed/storage/erasure_coding/ec_volume_info.go

@@ -55,3 +55,15 @@ func (evi *EcVolumeInfo) ToVolumeEcShardInformationMessage() (ret *master_pb.Vol
 		EncodeTsNs:  evi.EncodeTsNs,
 	}
 }
+
+func (evi *EcVolumeInfo) GetCollection() string {
+	return evi.Collection
+}
+
+func (evi *EcVolumeInfo) GetVolumeId() needle.VolumeId {
+	return evi.VolumeId
+}
+
+func (evi *EcVolumeInfo) GetRemoteStorageName() string {
+	return ""
+}

weed/storage/volume_info.go

@@ -145,6 +145,18 @@ func (vi VolumeInfo) String() string {
 	return s
 }
 
+func (vi VolumeInfo) GetCollection() string {
+	return vi.Collection
+}
+
+func (vi VolumeInfo) GetVolumeId() needle.VolumeId {
+	return vi.Id
+}
+
+func (vi VolumeInfo) GetRemoteStorageName() string {
+	return vi.RemoteStorageName
+}
+
 func (vi VolumeInfo) ToVolumeInformationMessage() *master_pb.VolumeInformationMessage {
 	return &master_pb.VolumeInformationMessage{
 		Id:                uint32(vi.Id),

weed/topology/disk.go

@@ -426,7 +426,7 @@ func (d *Disk) ToDiskInfo(filter VolumeFilter) *master_pb.DiskInfo {
 		if !haveDiskId || v.DiskId < diskId {
 			diskId, haveDiskId = v.DiskId, true
 		}
-		if !filter.matches(v.Collection, v.Id) {
+		if !filter.matches(v) {
 			continue
 		}
 		volumeInfos = append(volumeInfos, v.ToVolumeInformationMessage())
@@ -460,7 +460,7 @@ func (d *Disk) ToDiskInfo(filter VolumeFilter) *master_pb.DiskInfo {
 	}
 	m.EcShardInfos = make([]*master_pb.VolumeEcShardInformationMessage, 0, ecCapacity)
 	for _, ecv := range ecShards {
-		if !filter.matches(ecv.Collection, ecv.VolumeId) {
+		if !filter.matches(ecv) {
 			continue
 		}
 		m.EcShardInfos = append(m.EcShardInfos, ecv.ToVolumeEcShardInformationMessage())

weed/topology/volume_filter.go

@@ -3,14 +3,16 @@ package topology
 import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 )
 
 // VolumeFilter narrows a topology listing to the volumes a caller asked about,
 // selecting only what is listed under a disk. A nil field filters nothing, and
 // only nil does: the empty collection is a real one.
 type VolumeFilter struct {
-	Collection *string
-	VolumeId   *needle.VolumeId
+	Collection        *string
+	remoteStorageName *string
+	VolumeId          *needle.VolumeId
 	// nothing selects the topology alone, for a listing whose volumes travel
 	// in messages of their own.
 	nothing bool
@@ -34,6 +36,14 @@ func NewVolumeFilter(req *master_pb.VolumeListRequest) VolumeFilter {
 		defaultCollection := ""
 		filter.Collection = &defaultCollection
 	}
+
+	switch {
+	case req.RemoteStorageName != "":
+		filter.remoteStorageName = new(req.RemoteStorageName)
+	case req.LocalVolumeOnly:
+		filter.remoteStorageName = new("")
+	}
+
 	if req.VolumeId != 0 {
 		volumeId := needle.VolumeId(req.VolumeId)
 		filter.VolumeId = &volumeId
@@ -43,17 +53,34 @@ func NewVolumeFilter(req *master_pb.VolumeListRequest) VolumeFilter {
 
 // SelectsEverything lets a caller size its result for the whole disk up front.
 func (f VolumeFilter) SelectsEverything() bool {
-	return !f.nothing && f.Collection == nil && f.VolumeId == nil
+	return !f.nothing && f.Collection == nil && f.VolumeId == nil && f.remoteStorageName == nil
 }
 
-func (f VolumeFilter) matches(collection string, id needle.VolumeId) bool {
+type volumeLike interface {
+	GetCollection() string
+	GetVolumeId() needle.VolumeId
+	GetRemoteStorageName() string
+}
+
+func (f VolumeFilter) matches(vi volumeLike) bool {
 	if f.nothing {
 		return false
 	}
-	if f.Collection != nil && *f.Collection != collection {
+	if f.Collection != nil && *f.Collection != vi.GetCollection() {
 		return false
 	}
-	if f.VolumeId != nil && *f.VolumeId != id {
+	if f.remoteStorageName != nil {
+		pattern, name := *f.remoteStorageName, vi.GetRemoteStorageName()
+		// the empty name is the local volumes, which only asking for the empty
+		// name selects; even * leaves them out.
+		if name == "" && pattern != "" {
+			return false
+		}
+		if !wildcard.MatchesWildcard(pattern, name) {
+			return false
+		}
+	}
+	if f.VolumeId != nil && *f.VolumeId != vi.GetVolumeId() {
 		return false
 	}
 	return true

weed/topology/volume_filter_test.go

@@ -95,6 +95,45 @@ func TestVolumeFilterSelects(t *testing.T) {
 	}
 }
 
+func TestVolumeFilterRemoteStorageNameWildcards(t *testing.T) {
+	topo := NewTopology("filter", nil, 32*1024*1024*1024, 5, false)
+	dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("10.0.0.2", 8080, 18080, "", "", map[string]uint32{"": 100})
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		{Id: 11, Collection: "c", Size: 100, Version: 3, DiskId: 3, RemoteStorageName: "s3.backup"},
+		{Id: 12, Collection: "c", Size: 100, Version: 3, DiskId: 3, RemoteStorageName: "s3.archive"},
+		{Id: 13, Collection: "c", Size: 100, Version: 3, DiskId: 3, RemoteStorageName: "gcs.cold"},
+		{Id: 14, Collection: "c", Size: 100, Version: 3, DiskId: 3},
+	}, dn)
+
+	for _, tc := range []struct {
+		name string
+		f    VolumeFilter
+		want []uint32
+	}{
+		// A nil name asks for every volume, tiered or local alike.
+		{"no remote storage asked", VolumeFilter{}, []uint32{11, 12, 13, 14}},
+		{"an exact name", VolumeFilter{remoteStorageName: new("s3.backup")}, []uint32{11}},
+		{"a wildcarded storage", VolumeFilter{remoteStorageName: new("s3.*")}, []uint32{11, 12}},
+		{"a wildcarded tier", VolumeFilter{remoteStorageName: new("*.cold")}, []uint32{13}},
+		{"every tiered volume", VolumeFilter{remoteStorageName: new("*")}, []uint32{11, 12, 13}},
+		// Not a wildcard, so it matches nothing rather than everything.
+		{"a name no volume is on", VolumeFilter{remoteStorageName: new("s3.other")}, nil},
+		// The local volumes are the ones with no remote storage name.
+		{"the local ones only", VolumeFilter{remoteStorageName: new("")}, []uint32{14}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			volumes, ecVolumes := listed(topo.ToTopologyInfo(tc.f))
+			if !equalIds(volumes, tc.want) {
+				t.Errorf("listed volumes %v, want %v", volumes, tc.want)
+			}
+			if len(ecVolumes) != 0 {
+				t.Errorf("listed ec volumes %v, want none", ecVolumes)
+			}
+		})
+	}
+}
+
 // A filter never changes which disks are reported or what they say about
 // themselves.
 func TestVolumeFilterKeepsTheTopology(t *testing.T) {

weed/topology/volume_stream.go

@@ -96,7 +96,7 @@ func (d *Disk) streamVolumes(filter VolumeFilter, batchSize int, newBatch func()
 		d.RLock()
 		for _, id := range ids[start:end] {
 			v, found := d.volumes[id]
-			if !found || !filter.matches(v.Collection, v.Id) {
+			if !found || !filter.matches(v) {
 				continue
 			}
 			batch.VolumeInfos = append(batch.VolumeInfos, v.ToVolumeInformationMessage())
@@ -121,7 +121,7 @@ func (d *Disk) streamEcShards(filter VolumeFilter, batchSize int, newBatch func(
 		end := min(start+batchSize, len(shards))
 		batch := newBatch()
 		for _, ecv := range shards[start:end] {
-			if !filter.matches(ecv.Collection, ecv.VolumeId) {
+			if !filter.matches(ecv) {
 				continue
 			}
 			batch.EcShardInfos = append(batch.EcShardInfos, ecv.ToVolumeEcShardInformationMessage())