From 13bf056a15a85309da369f4137c0e6fe7e497f8e Mon Sep 17 00:00:00 2001 From: Feng Shao <88640691+shaofeng66@users.noreply.github.com> Date: Thu, 10 Sep 2026 03:55:47 +0800 Subject: [PATCH] Mount req with collection (#11249) * volume mount req support specify collection * rust mirror change --- seaweed-volume/proto/volume_server.proto | 1 + seaweed-volume/src/server/grpc_server.rs | 14 +++++---- seaweed-volume/src/storage/store.rs | 29 ++++++++++++++++++- weed/pb/volume_server.proto | 1 + weed/pb/volume_server_pb/volume_server.pb.go | 19 ++++++++++-- .../volume_server_pb/volume_server_grpc.pb.go | 2 +- weed/server/volume_grpc_admin.go | 6 ++-- weed/server/volume_grpc_copy.go | 2 +- weed/server/volume_grpc_erasure_coding.go | 2 +- weed/server/volume_grpc_tier_download_test.go | 2 +- weed/storage/disk_location.go | 10 ++++++- weed/storage/store.go | 12 ++++---- weed/storage/store_consolidate_index_test.go | 4 +-- .../store_mark_readonly_can_delete_test.go | 2 +- weed/storage/store_mount_validation_test.go | 2 +- 15 files changed, 80 insertions(+), 28 deletions(-) diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index 961a800fe..39b8f1af9 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -236,6 +236,7 @@ message VolumeIncrementalCopyResponse { message VolumeMountRequest { uint32 volume_id = 1; + optional string collection = 2; } message VolumeMountResponse { } diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index c0d789b8c..059aedbe3 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -1450,7 +1450,7 @@ impl VolumeServer for VolumeGrpcService { let mut store = self.state.store.write().unwrap(); store - .mount_volume_by_id(vid) + .mount_volume_by_id(vid, req.collection.as_deref()) .map_err(|e| Status::internal(e.to_string()))?; self.state.volume_state_notify.notify_one(); @@ -1622,7 +1622,7 @@ impl VolumeServer for VolumeGrpcService { if let Err(e) = store.configure_volume(vid, rp) { let mut error = format!("volume configure {}: {}", vid, e); // Error recovery: try to re-mount anyway - if let Err(mount_err) = store.mount_volume_by_id(vid) { + if let Err(mount_err) = store.mount_volume_by_id(vid, None) { error += &format!(". Also failed to restore mount: {}", mount_err); } return Ok(Response::new(volume_server_pb::VolumeConfigureResponse { @@ -1631,7 +1631,7 @@ impl VolumeServer for VolumeGrpcService { } // Re-mount the volume - if let Err(e) = store.mount_volume_by_id(vid) { + if let Err(e) = store.mount_volume_by_id(vid, None) { return Ok(Response::new(volume_server_pb::VolumeConfigureResponse { error: format!("volume configure mount {}: {}", vid, e), })); @@ -3937,9 +3937,11 @@ impl VolumeServer for VolumeGrpcService { { let mut store = self.state.store.write().unwrap(); - store.mount_volume_by_id(vid).map_err(|e| { - Status::internal(format!("mount staged volume {}: {}", req.volume_id, e)) - })?; + store + .mount_volume_by_id(vid, Some(req.collection.as_str())) + .map_err(|e| { + Status::internal(format!("mount staged volume {}: {}", req.volume_id, e)) + })?; } self.state.volume_state_notify.notify_one(); tracing::info!( diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index 8086e7c1b..df692283d 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -496,10 +496,37 @@ impl Store { /// Mount a volume by id only (Go's MountVolume behavior). /// Scans all locations for a matching .dat file and loads with its collection prefix. - pub fn mount_volume_by_id(&mut self, vid: VolumeId) -> Result<(), VolumeError> { + /// When a collection hint is given, the expected _.vif/.idx + /// path is probed directly before falling back to the directory scan. + pub fn mount_volume_by_id( + &mut self, + vid: VolumeId, + collection: Option<&str>, + ) -> Result<(), VolumeError> { if self.find_volume(vid).is_some() { return Err(VolumeError::AlreadyExists); } + if let Some(collection) = collection { + for loc in &mut self.locations { + let base = + crate::storage::volume::volume_file_name(&loc.directory, collection, vid); + for ext in [".vif", ".idx"] { + if let Ok(meta) = std::fs::metadata(format!("{}{}", base, ext)) { + if !meta.is_dir() { + return loc.create_volume( + vid, + collection, + self.needle_map_kind, + None, + None, + 0, + Version::current(), + ); + } + } + } + } + } if let Some((loc_idx, _base_path, collection)) = self.find_volume_file_base(vid) { let loc = &mut self.locations[loc_idx]; return loc.create_volume( diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 4962a4e9f..de1bb235b 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -236,6 +236,7 @@ message VolumeIncrementalCopyResponse { message VolumeMountRequest { uint32 volume_id = 1; + optional string collection = 2; } message VolumeMountResponse { } diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index 24ad97778..30f0e4258 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.6 -// protoc v7.35.0 +// protoc v6.33.4 // source: volume_server.proto package volume_server_pb @@ -1216,6 +1216,7 @@ func (x *VolumeIncrementalCopyResponse) GetFileContent() []byte { type VolumeMountRequest struct { state protoimpl.MessageState `protogen:"open.v1"` VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"` + Collection *string `protobuf:"bytes,2,opt,name=collection,proto3,oneof" json:"collection,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1257,6 +1258,13 @@ func (x *VolumeMountRequest) GetVolumeId() uint32 { return 0 } +func (x *VolumeMountRequest) GetCollection() string { + if x != nil && x.Collection != nil { + return *x.Collection + } + return "" +} + type VolumeMountResponse struct { state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields @@ -7287,9 +7295,13 @@ const file_volume_server_proto_rawDesc = "" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x19\n" + "\bsince_ns\x18\x02 \x01(\x04R\asinceNs\"B\n" + "\x1dVolumeIncrementalCopyResponse\x12!\n" + - "\ffile_content\x18\x01 \x01(\fR\vfileContent\"1\n" + + "\ffile_content\x18\x01 \x01(\fR\vfileContent\"e\n" + "\x12VolumeMountRequest\x12\x1b\n" + - "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\"\x15\n" + + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12#\n" + + "\n" + + "collection\x18\x02 \x01(\tH\x00R\n" + + "collection\x88\x01\x01B\r\n" + + "\v_collection\"\x15\n" + "\x13VolumeMountResponse\"3\n" + "\x14VolumeUnmountRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\"\x17\n" + @@ -8113,6 +8125,7 @@ func file_volume_server_proto_init() { if File_volume_server_proto != nil { return } + file_volume_server_proto_msgTypes[21].OneofWrappers = []any{} file_volume_server_proto_msgTypes[45].OneofWrappers = []any{ (*ReceiveFileRequest_Info)(nil), (*ReceiveFileRequest_FileContent)(nil), diff --git a/weed/pb/volume_server_pb/volume_server_grpc.pb.go b/weed/pb/volume_server_pb/volume_server_grpc.pb.go index eb659eab0..72d98971a 100644 --- a/weed/pb/volume_server_pb/volume_server_grpc.pb.go +++ b/weed/pb/volume_server_pb/volume_server_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.2 -// - protoc v7.35.0 +// - protoc v6.33.4 // source: volume_server.proto package volume_server_pb diff --git a/weed/server/volume_grpc_admin.go b/weed/server/volume_grpc_admin.go index d12092af6..f5d8c71c7 100644 --- a/weed/server/volume_grpc_admin.go +++ b/weed/server/volume_grpc_admin.go @@ -131,7 +131,7 @@ func (vs *VolumeServer) VolumeMount(ctx context.Context, req *volume_server_pb.V return resp, err } - err := vs.store.MountVolume(needle.VolumeId(req.VolumeId)) + err := vs.store.MountVolume(needle.VolumeId(req.VolumeId), req.Collection) if err != nil { glog.Errorf("volume mount %v: %v", req, err) @@ -253,7 +253,7 @@ func (vs *VolumeServer) VolumeConfigure(ctx context.Context, req *volume_server_ glog.Errorf("volume configure %v: %v", req, err) resp.Error = fmt.Sprintf("volume configure %v: %v", req, err) // Try to re-mount to restore the volume state - if mountErr := vs.store.MountVolume(needle.VolumeId(req.VolumeId)); mountErr != nil { + if mountErr := vs.store.MountVolume(needle.VolumeId(req.VolumeId), nil); mountErr != nil { glog.Errorf("volume configure failed to restore mount %v: %v", req, mountErr) resp.Error += fmt.Sprintf(". Also failed to restore mount: %v", mountErr) } @@ -261,7 +261,7 @@ func (vs *VolumeServer) VolumeConfigure(ctx context.Context, req *volume_server_ } // mount - if err := vs.store.MountVolume(needle.VolumeId(req.VolumeId)); err != nil { + if err := vs.store.MountVolume(needle.VolumeId(req.VolumeId), nil); err != nil { glog.Errorf("volume configure mount %v: %v", req, err) resp.Error = fmt.Sprintf("volume configure mount %v: %v", req, err) return resp, nil diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index dcb039621..33ecf72c5 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -251,7 +251,7 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre // Load and validate the volume before announcing it to the master. A failed // validation is unloaded by the store without ever making the replica // routable. - err = vs.store.MountVolumeWithValidator(needle.VolumeId(req.VolumeId), func(targetVolume *storage.Volume) error { + err = vs.store.MountVolumeWithValidator(needle.VolumeId(req.VolumeId), &req.Collection, func(targetVolume *storage.Volume) error { if !shouldValidateCopyCounts { return nil } diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 1251045c4..283f52c9f 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -1206,7 +1206,7 @@ func (vs *VolumeServer) adoptStagedVolume(req *volume_server_pb.VolumeEcShardsTo } os.Remove(noteFile) - if err := vs.store.MountVolume(vid); err != nil { + if err := vs.store.MountVolume(vid, &req.Collection); err != nil { return nil, fmt.Errorf("mount staged volume %d: %w", req.VolumeId, err) } glog.V(0).Infof("VolumeEcShardsToVolume: adopted decoded volume %d from staging (%s)", req.VolumeId, base) diff --git a/weed/server/volume_grpc_tier_download_test.go b/weed/server/volume_grpc_tier_download_test.go index 378ca13f1..fdca2a8aa 100644 --- a/weed/server/volume_grpc_tier_download_test.go +++ b/weed/server/volume_grpc_tier_download_test.go @@ -297,7 +297,7 @@ func TestTierMoveDatFromRemote_KeepRemote_LeavesReplicaLocal(t *testing.T) { if err := store.UnmountVolume(vid); err != nil { t.Fatalf("unmount after download: %v", err) } - if err := store.MountVolume(vid); err != nil { + if err := store.MountVolume(vid, &req.Collection); err != nil { t.Fatalf("remount after download: %v", err) } v2 := store.GetVolume(vid) diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index ed17faa98..102689500 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -511,7 +511,15 @@ func (l *DiskLocation) deleteVolumeById(vid needle.VolumeId, onlyEmpty bool, kee return } -func (l *DiskLocation) LoadVolume(diskId uint32, vid needle.VolumeId, needleMapKind NeedleMapKind) bool { +func (l *DiskLocation) LoadVolume(diskId uint32, vid needle.VolumeId, needleMapKind NeedleMapKind, collection *string) bool { + if collection != nil { + for _, ext := range []string{".vif", ".idx"} { + filename := VolumeFileName(l.Directory, *collection, int(vid)) + ext + if fi, err := os.Stat(filename); err == nil && !fi.IsDir() { + return l.loadExistingVolume(fi.Name(), needleMapKind, false, 0, diskId) + } + } + } if fileInfo, found := l.LocateVolume(vid); found { return l.loadExistingVolume(fileInfo.Name(), needleMapKind, false, 0, diskId) } diff --git a/weed/storage/store.go b/weed/storage/store.go index 3b483d802..7d891fedc 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -966,20 +966,20 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error { return persistErr } -func (s *Store) MountVolume(i needle.VolumeId) error { - return s.mountVolume(i, nil) +func (s *Store) MountVolume(i needle.VolumeId, collection *string) error { + return s.mountVolume(i, collection, nil) } // MountVolumeWithValidator loads a volume, validates it before announcing it // to the master, and unloads it when validation fails. This keeps an invalid // newly copied replica out of the master's routable volume set. -func (s *Store) MountVolumeWithValidator(i needle.VolumeId, validator func(*Volume) error) error { - return s.mountVolume(i, validator) +func (s *Store) MountVolumeWithValidator(i needle.VolumeId, collection *string, validator func(*Volume) error) error { + return s.mountVolume(i, collection, validator) } -func (s *Store) mountVolume(i needle.VolumeId, validator func(*Volume) error) error { +func (s *Store) mountVolume(i needle.VolumeId, collection *string, validator func(*Volume) error) error { for diskId, location := range s.Locations { - if found := location.LoadVolume(uint32(diskId), i, s.NeedleMapKind); found == true { + if found := location.LoadVolume(uint32(diskId), i, s.NeedleMapKind, collection); found == true { glog.V(0).Infof("mount volume %d", i) v := s.findVolume(i) v.diskId = uint32(diskId) // Set disk ID when mounting diff --git a/weed/storage/store_consolidate_index_test.go b/weed/storage/store_consolidate_index_test.go index 282854e22..296a837cf 100644 --- a/weed/storage/store_consolidate_index_test.go +++ b/weed/storage/store_consolidate_index_test.go @@ -86,7 +86,7 @@ func TestConsolidateVolumeIndexMovesIdxToIdxDir(t *testing.T) { require.FileExists(t, dataIdx, "precondition: index co-located with the data") store := newIdxSplitStore(t, dataDir, idxDir) - require.NoError(t, store.MountVolume(vid)) + require.NoError(t, store.MountVolume(vid, nil)) // Write a needle so the relocate has real index state to preserve. mounted := store.findVolume(vid) @@ -122,7 +122,7 @@ func TestConsolidateVolumeIndexNoopWithoutIdxDir(t *testing.T) { v.Close() store := newIdxSplitStore(t, dataDir, dataDir) // idx dir == data dir - require.NoError(t, store.MountVolume(vid)) + require.NoError(t, store.MountVolume(vid, nil)) require.NoError(t, store.ConsolidateVolumeIndex(vid)) require.FileExists(t, filepath.Join(dataDir, "8.idx"), "index stays put without -dir.idx") diff --git a/weed/storage/store_mark_readonly_can_delete_test.go b/weed/storage/store_mark_readonly_can_delete_test.go index 907c35dcb..00376df8c 100644 --- a/weed/storage/store_mark_readonly_can_delete_test.go +++ b/weed/storage/store_mark_readonly_can_delete_test.go @@ -144,7 +144,7 @@ func TestMountVolumeAnnouncesReadOnlyState(t *testing.T) { require.NoError(t, store.MarkVolumeReadonly(vid, tc.canDelete, true)) require.NoError(t, store.UnmountVolume(vid)) <-store.DeletedVolumesChan - require.NoError(t, store.MountVolume(vid)) + require.NoError(t, store.MountVolume(vid, nil)) message := <-store.NewVolumesChan require.True(t, message.ReadOnly) diff --git a/weed/storage/store_mount_validation_test.go b/weed/storage/store_mount_validation_test.go index d2de1369b..cf6b16eb0 100644 --- a/weed/storage/store_mount_validation_test.go +++ b/weed/storage/store_mount_validation_test.go @@ -29,7 +29,7 @@ func TestMountVolumeWithValidatorAnnouncesOnlyAfterValidation(t *testing.T) { t.Cleanup(store.Close) validationErr := errors.New("copy counts differ") - err = store.MountVolumeWithValidator(vid, func(*Volume) error { + err = store.MountVolumeWithValidator(vid, nil, func(*Volume) error { select { case <-store.NewVolumesChan: t.Fatal("volume was announced before validation completed")