mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-20 15:04:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Mount req with collection (#11249)

Browse Source 
* volume mount req support specify collection

* rust mirror change
This commit is contained in:
Feng Shao
2026-09-09 12:55:47 -07:00
committed by GitHub
parent c968084b34
commit 13bf056a15
15 changed files with 80 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/proto/volume_server.proto
+1
View File
@@ -236,6 +236,7 @@ message VolumeIncrementalCopyResponse {
message VolumeMountRequest {
uint32 volume_id = 1;
optional string collection = 2;
}
message VolumeMountResponse {
}
seaweed-volume/src/server/grpc_server.rs
+8 -6
View File
@@ -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!(
seaweed-volume/src/storage/store.rs
+28 -1
View File
@@ -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 <collection>_<vid>.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(
weed/pb/volume_server.proto
+1
View File
@@ -236,6 +236,7 @@ message VolumeIncrementalCopyResponse {
message VolumeMountRequest {
uint32 volume_id = 1;
optional string collection = 2;
}
message VolumeMountResponse {
}
weed/pb/volume_server_pb/volume_server.pb.go
+16 -3
View File
@@ -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),
weed/pb/volume_server_pb/volume_server_grpc.pb.go
+1 -1
View File
@@ -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
weed/server/volume_grpc_admin.go
+3 -3
View File
@@ -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
weed/server/volume_grpc_copy.go
+1 -1
View File
@@ -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
}
weed/server/volume_grpc_erasure_coding.go
+1 -1
View File
@@ -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)
weed/server/volume_grpc_tier_download_test.go
+1 -1
View File
@@ -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)
weed/storage/disk_location.go
+9 -1
View File
@@ -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)
}
weed/storage/store.go
+6 -6
View File
@@ -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
weed/storage/store_consolidate_index_test.go
+2 -2
View File
@@ -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")
weed/storage/store_mark_readonly_can_delete_test.go
+1 -1
View File
@@ -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)
weed/storage/store_mount_validation_test.go
+1 -1
View File
@@ -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")