From 8112f2733a427c3aa1f5b2728cd871614bdec8a1 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 3 Sep 2026 14:52:05 -0700 Subject: [PATCH] filer: batch exact lookup RPC, authoritative volume lookup, VolumeDelete status codes (#11122) * storage: make DeleteVolume errors inspectable with errors.Is An absent volume wraps ErrVolumeNotFound and an only-empty refusal now wraps ErrVolumeNotEmpty with %w instead of %v, so callers no longer have to match on the message. Claude-Session: https://claude.ai/code/session_01T4MEV3ETqFFKN46Uu2ZrUm * volume server: return NotFound and FailedPrecondition from VolumeDelete An absent volume maps to codes.NotFound and a non-empty volume under only_empty to codes.FailedPrecondition, so a caller retiring a volume can treat NotFound as already done. The store message is kept in the status description because the EC empty-replica sweep still matches on it. Claude-Session: https://claude.ai/code/session_01T4MEV3ETqFFKN46Uu2ZrUm * wdclient: add LookupVolumeIdsAuthoritative Bypasses the vid map and asks the provider directly, for callers where a stale positive location is unsafe. Claude-Session: https://claude.ai/code/session_01T4MEV3ETqFFKN46Uu2ZrUm * filer: add LookupDirectoryEntries batch lookup RPC Up to 4096 exact-path lookups in one call, resolved concurrently with results in request order, plus one deduplicated location lookup for every volume the returned entries reference and per-fid read tokens when the filer signs reads. unavailable_volume_is_miss lets cache-style callers take an entry whose volume has no live location as a miss, resolved against the master rather than the filer's location cache. Claude-Session: https://claude.ai/code/session_01T4MEV3ETqFFKN46Uu2ZrUm * filer: test that an expired file entry is deleted on read Claude-Session: https://claude.ai/code/session_01T4MEV3ETqFFKN46Uu2ZrUm * filer: test that AssignVolume and CreateEntry resolve the same TTL rule Claude-Session: https://claude.ai/code/session_01T4MEV3ETqFFKN46Uu2ZrUm * master: refuse partial lookups while warming up LookupVolume returned Unavailable during warm-up only when every requested volume was missing. A batch mixing a reported volume with one whose server has not reconnected yet came back as a partial answer with a per-volume not-found, which a caller treating the master as authoritative reads as gone. Any not-found during warm-up is now Unavailable, which callers already retry. Claude-Session: https://claude.ai/code/session_01T4MEV3ETqFFKN46Uu2ZrUm * filer: build batch test requests instead of copying a proto message Copying a generated message copies its internal mutex, which go vet's copylocks check rejects. Claude-Session: https://claude.ai/code/session_01T4MEV3ETqFFKN46Uu2ZrUm * filer: match ErrNotFound with errors.Is and state the miss rule's contract A wrapped not-found from the store would otherwise be reported as an error rather than a miss. The comments now say why a nil location map is the only sign of an unanswered lookup: the provider returns nil when it got no answer and a populated map, with unserved volumes reported as errors, when the master did answer. Claude-Session: https://claude.ai/code/session_01T4MEV3ETqFFKN46Uu2ZrUm * volume server: map absent and non-empty VolumeDelete errors in the Rust server Matches the Go server: an absent volume is NotFound and an only_empty refusal is FailedPrecondition instead of Internal, with the messages the EC empty-replica sweep matches on. Claude-Session: https://claude.ai/code/session_01T4MEV3ETqFFKN46Uu2ZrUm * filer: test that a malformed entry keeps its error outside cache mode Same test file as the enterprise tree, so the next sync sees one version. Claude-Session: https://claude.ai/code/session_01T4MEV3ETqFFKN46Uu2ZrUm --- other/java/client/src/main/proto/filer.proto | 41 + seaweed-volume/src/server/grpc_server.rs | 44 +- weed/filer/filer_prefix_object_test.go | 29 + weed/pb/filer.proto | 40 + weed/pb/filer_pb/filer.pb.go | 445 +++++++-- weed/pb/filer_pb/filer_grpc.pb.go | 40 +- weed/pb/filer_pb/filer_vtproto.pb.go | 917 ++++++++++++++++++ weed/server/filer_grpc_server_batch_lookup.go | 247 +++++ .../filer_grpc_server_batch_lookup_test.go | 418 ++++++++ .../filer_server_storage_rule_ttl_test.go | 46 + weed/server/master_grpc_server_volume.go | 9 +- .../master_grpc_server_volume_warmup_test.go | 63 ++ weed/server/volume_grpc_admin.go | 15 + weed/server/volume_grpc_admin_delete_test.go | 37 + weed/storage/store.go | 4 +- weed/storage/store_delete_volume_test.go | 35 + weed/wdclient/vidmap_authoritative_test.go | 49 + weed/wdclient/vidmap_client.go | 9 + 18 files changed, 2378 insertions(+), 110 deletions(-) create mode 100644 weed/server/filer_grpc_server_batch_lookup.go create mode 100644 weed/server/filer_grpc_server_batch_lookup_test.go create mode 100644 weed/server/master_grpc_server_volume_warmup_test.go create mode 100644 weed/server/volume_grpc_admin_delete_test.go create mode 100644 weed/storage/store_delete_volume_test.go create mode 100644 weed/wdclient/vidmap_authoritative_test.go diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index 324cbf5bb..1515c169a 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -13,6 +13,9 @@ service SeaweedFiler { rpc LookupDirectoryEntry (LookupDirectoryEntryRequest) returns (LookupDirectoryEntryResponse) { } + rpc LookupDirectoryEntries (LookupDirectoryEntriesRequest) returns (LookupDirectoryEntriesResponse) { + } + rpc ListEntries (ListEntriesRequest) returns (stream ListEntriesResponse) { } @@ -581,6 +584,7 @@ message Location { string public_url = 2; uint32 grpc_port = 3; string data_center = 4; + bool data_in_remote = 5; } message LookupVolumeResponse { map locations_map = 1; @@ -912,3 +916,40 @@ message MountInfo { int64 last_seen_ns = 3; string data_center = 4; } + +// LookupDirectoryEntriesRequest batches independent exact-path lookups into +// one bounded RPC. Results preserve this order. Empty batches and batches over +// 4096 requests are rejected. Declared last to keep generated message indices +// stable. +message LookupDirectoryEntriesRequest { + repeated LookupDirectoryEntryRequest requests = 1; + // Cache callers may treat an Entry whose Volume the master no longer + // reports as a miss. A lookup the filer could not complete still marks the + // Entry with an error. Ordinary filesystem callers keep the fail-closed + // error in both cases. + bool unavailable_volume_is_miss = 2; +} + +message LookupDirectoryEntryResult { + // found reports metadata presence. An error makes this item unusable even + // when entry is present, for example when a chunk volume cannot be located. + bool found = 1; + Entry entry = 2; + string error = 3; + // Per-entry read fence. Every filer event at or below this position is + // reflected in found/entry/error for this item. + int64 log_ts_ns = 4; + int32 log_signature = 5; +} + +message LookupDirectoryEntriesResponse { + repeated LookupDirectoryEntryResult results = 1; + // One deduplicated location set for every volume referenced by returned + // Entry.chunks. A missing/unavailable volume has an empty Locations value + // and marks each affected result with an error. + map locations_map = 2; + // Short-lived, exact-FID read capabilities. Direct data-plane clients must + // present the matching token to a Volume instead of treating a shared + // service credential as authority to read arbitrary needles. + map read_auth = 3; +} diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 673ead5f2..f6f546ff4 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -1039,7 +1039,15 @@ impl VolumeServer for VolumeGrpcService { } store .delete_volume(vid, req.only_empty, req.keep_remote_data) - .map_err(|e| Status::internal(e.to_string()))?; + .map_err(|e| match e { + crate::storage::volume::VolumeError::NotFound => { + Status::not_found(format!("not found volume id {}", vid)) + } + crate::storage::volume::VolumeError::NotEmpty => { + Status::failed_precondition("volume not empty") + } + other => Status::internal(other.to_string()), + })?; self.state.volume_state_notify.notify_one(); Ok(Response::new(volume_server_pb::VolumeDeleteResponse {})) } @@ -6423,4 +6431,38 @@ mod tests { assert!(resp.details.is_empty(), "{:?}", resp.details); assert_eq!(resp.total_files, 1); } + + #[tokio::test] + async fn volume_delete_reports_absent_volume_as_not_found() { + let (service, _tmp) = make_service_with_seed_masters(&[]); + let err = service + .volume_delete(Request::new(volume_server_pb::VolumeDeleteRequest { + volume_id: 4242, + only_empty: false, + keep_remote_data: false, + })) + .await + .expect_err("deleting an absent volume must fail"); + assert_eq!(err.code(), tonic::Code::NotFound, "{err:?}"); + assert!(err.message().contains("not found"), "{err:?}"); + } + + #[tokio::test] + async fn volume_delete_only_empty_refuses_a_volume_with_data() { + let (service, _tmp) = make_local_service_with_volume("", None); + let err = service + .volume_delete(Request::new(volume_server_pb::VolumeDeleteRequest { + volume_id: 1, + only_empty: true, + keep_remote_data: false, + })) + .await + .expect_err("only_empty must refuse a volume holding a needle"); + assert_eq!(err.code(), tonic::Code::FailedPrecondition, "{err:?}"); + assert!(err.message().contains("volume not empty"), "{err:?}"); + assert!( + service.state.store.read().unwrap().find_volume(VolumeId(1)).is_some(), + "refused delete must leave the volume mounted" + ); + } } diff --git a/weed/filer/filer_prefix_object_test.go b/weed/filer/filer_prefix_object_test.go index 9d1e488c2..b18b9d6ba 100644 --- a/weed/filer/filer_prefix_object_test.go +++ b/weed/filer/filer_prefix_object_test.go @@ -75,3 +75,32 @@ func TestExpiredDirectoryIsNotDeletedOnRead(t *testing.T) { require.NoError(t, err) require.NotNil(t, stillThere) } + +// TestExpiredFileIsDeletedOnRead pins the native cache-TTL path: regular +// entries become invisible and their metadata row is removed without a +// cache-specific transaction or reverse Volume index. +func TestExpiredFileIsDeletedOnRead(t *testing.T) { + f, store := newTestFilerWithStubStore() + ctx := context.Background() + + filePath := util.FullPath("/buckets/lmcache/expired") + expired := time.Now().Add(-2 * time.Hour) + require.NoError(t, store.InsertEntry(ctx, &Entry{ + FullPath: filePath, + Attr: Attr{ + Mode: 0o644, + Crtime: expired, + Mtime: expired, + TtlSec: 60, + }, + Chunks: []*filer_pb.FileChunk{{FileId: "7,01", Size: 4}}, + })) + + found, err := f.FindEntry(ctx, filePath) + require.ErrorIs(t, err, filer_pb.ErrNotFound) + require.Nil(t, found) + + _, err = store.FindEntry(ctx, filePath) + require.ErrorIs(t, err, filer_pb.ErrNotFound, + "native TTL lookup should remove the expired metadata row") +} diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index fd440bbae..1515c169a 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -13,6 +13,9 @@ service SeaweedFiler { rpc LookupDirectoryEntry (LookupDirectoryEntryRequest) returns (LookupDirectoryEntryResponse) { } + rpc LookupDirectoryEntries (LookupDirectoryEntriesRequest) returns (LookupDirectoryEntriesResponse) { + } + rpc ListEntries (ListEntriesRequest) returns (stream ListEntriesResponse) { } @@ -913,3 +916,40 @@ message MountInfo { int64 last_seen_ns = 3; string data_center = 4; } + +// LookupDirectoryEntriesRequest batches independent exact-path lookups into +// one bounded RPC. Results preserve this order. Empty batches and batches over +// 4096 requests are rejected. Declared last to keep generated message indices +// stable. +message LookupDirectoryEntriesRequest { + repeated LookupDirectoryEntryRequest requests = 1; + // Cache callers may treat an Entry whose Volume the master no longer + // reports as a miss. A lookup the filer could not complete still marks the + // Entry with an error. Ordinary filesystem callers keep the fail-closed + // error in both cases. + bool unavailable_volume_is_miss = 2; +} + +message LookupDirectoryEntryResult { + // found reports metadata presence. An error makes this item unusable even + // when entry is present, for example when a chunk volume cannot be located. + bool found = 1; + Entry entry = 2; + string error = 3; + // Per-entry read fence. Every filer event at or below this position is + // reflected in found/entry/error for this item. + int64 log_ts_ns = 4; + int32 log_signature = 5; +} + +message LookupDirectoryEntriesResponse { + repeated LookupDirectoryEntryResult results = 1; + // One deduplicated location set for every volume referenced by returned + // Entry.chunks. A missing/unavailable volume has an empty Locations value + // and marks each affected result with an error. + map locations_map = 2; + // Short-lived, exact-FID read capabilities. Direct data-plane clients must + // present the matching token to a Volume instead of treating a shared + // service credential as authority to read arbitrary needles. + map read_auth = 3; +} diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 46bacb313..ab71038df 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -6630,6 +6630,212 @@ func (x *MountInfo) GetDataCenter() string { return "" } +// LookupDirectoryEntriesRequest batches independent exact-path lookups into +// one bounded RPC. Results preserve this order. Empty batches and batches over +// 4096 requests are rejected. Declared last to keep generated message indices +// stable. +type LookupDirectoryEntriesRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Requests []*LookupDirectoryEntryRequest `protobuf:"bytes,1,rep,name=requests,proto3" json:"requests,omitempty"` + // Cache callers may treat an Entry whose Volume the master no longer + // reports as a miss. A lookup the filer could not complete still marks the + // Entry with an error. Ordinary filesystem callers keep the fail-closed + // error in both cases. + UnavailableVolumeIsMiss bool `protobuf:"varint,2,opt,name=unavailable_volume_is_miss,json=unavailableVolumeIsMiss,proto3" json:"unavailable_volume_is_miss,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LookupDirectoryEntriesRequest) Reset() { + *x = LookupDirectoryEntriesRequest{} + mi := &file_filer_proto_msgTypes[91] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LookupDirectoryEntriesRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LookupDirectoryEntriesRequest) ProtoMessage() {} + +func (x *LookupDirectoryEntriesRequest) ProtoReflect() protoreflect.Message { + mi := &file_filer_proto_msgTypes[91] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use LookupDirectoryEntriesRequest.ProtoReflect.Descriptor instead. +func (*LookupDirectoryEntriesRequest) Descriptor() ([]byte, []int) { + return file_filer_proto_rawDescGZIP(), []int{91} +} + +func (x *LookupDirectoryEntriesRequest) GetRequests() []*LookupDirectoryEntryRequest { + if x != nil { + return x.Requests + } + return nil +} + +func (x *LookupDirectoryEntriesRequest) GetUnavailableVolumeIsMiss() bool { + if x != nil { + return x.UnavailableVolumeIsMiss + } + return false +} + +type LookupDirectoryEntryResult struct { + state protoimpl.MessageState `protogen:"open.v1"` + // found reports metadata presence. An error makes this item unusable even + // when entry is present, for example when a chunk volume cannot be located. + Found bool `protobuf:"varint,1,opt,name=found,proto3" json:"found,omitempty"` + Entry *Entry `protobuf:"bytes,2,opt,name=entry,proto3" json:"entry,omitempty"` + Error string `protobuf:"bytes,3,opt,name=error,proto3" json:"error,omitempty"` + // Per-entry read fence. Every filer event at or below this position is + // reflected in found/entry/error for this item. + LogTsNs int64 `protobuf:"varint,4,opt,name=log_ts_ns,json=logTsNs,proto3" json:"log_ts_ns,omitempty"` + LogSignature int32 `protobuf:"varint,5,opt,name=log_signature,json=logSignature,proto3" json:"log_signature,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LookupDirectoryEntryResult) Reset() { + *x = LookupDirectoryEntryResult{} + mi := &file_filer_proto_msgTypes[92] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LookupDirectoryEntryResult) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LookupDirectoryEntryResult) ProtoMessage() {} + +func (x *LookupDirectoryEntryResult) ProtoReflect() protoreflect.Message { + mi := &file_filer_proto_msgTypes[92] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use LookupDirectoryEntryResult.ProtoReflect.Descriptor instead. +func (*LookupDirectoryEntryResult) Descriptor() ([]byte, []int) { + return file_filer_proto_rawDescGZIP(), []int{92} +} + +func (x *LookupDirectoryEntryResult) GetFound() bool { + if x != nil { + return x.Found + } + return false +} + +func (x *LookupDirectoryEntryResult) GetEntry() *Entry { + if x != nil { + return x.Entry + } + return nil +} + +func (x *LookupDirectoryEntryResult) GetError() string { + if x != nil { + return x.Error + } + return "" +} + +func (x *LookupDirectoryEntryResult) GetLogTsNs() int64 { + if x != nil { + return x.LogTsNs + } + return 0 +} + +func (x *LookupDirectoryEntryResult) GetLogSignature() int32 { + if x != nil { + return x.LogSignature + } + return 0 +} + +type LookupDirectoryEntriesResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + Results []*LookupDirectoryEntryResult `protobuf:"bytes,1,rep,name=results,proto3" json:"results,omitempty"` + // One deduplicated location set for every volume referenced by returned + // Entry.chunks. A missing/unavailable volume has an empty Locations value + // and marks each affected result with an error. + LocationsMap map[string]*Locations `protobuf:"bytes,2,rep,name=locations_map,json=locationsMap,proto3" json:"locations_map,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // Short-lived, exact-FID read capabilities. Direct data-plane clients must + // present the matching token to a Volume instead of treating a shared + // service credential as authority to read arbitrary needles. + ReadAuth map[string]string `protobuf:"bytes,3,rep,name=read_auth,json=readAuth,proto3" json:"read_auth,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LookupDirectoryEntriesResponse) Reset() { + *x = LookupDirectoryEntriesResponse{} + mi := &file_filer_proto_msgTypes[93] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LookupDirectoryEntriesResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LookupDirectoryEntriesResponse) ProtoMessage() {} + +func (x *LookupDirectoryEntriesResponse) ProtoReflect() protoreflect.Message { + mi := &file_filer_proto_msgTypes[93] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use LookupDirectoryEntriesResponse.ProtoReflect.Descriptor instead. +func (*LookupDirectoryEntriesResponse) Descriptor() ([]byte, []int) { + return file_filer_proto_rawDescGZIP(), []int{93} +} + +func (x *LookupDirectoryEntriesResponse) GetResults() []*LookupDirectoryEntryResult { + if x != nil { + return x.Results + } + return nil +} + +func (x *LookupDirectoryEntriesResponse) GetLocationsMap() map[string]*Locations { + if x != nil { + return x.LocationsMap + } + return nil +} + +func (x *LookupDirectoryEntriesResponse) GetReadAuth() map[string]string { + if x != nil { + return x.ReadAuth + } + return nil +} + // Clause is one primitive comparison. IF_ETAG_MATCH holds when the current // entry's ETag equals any value in etags; IF_ETAG_NOT_MATCH holds when it // equals none. allow_weak permits weak-comparison (ignoring the W/ prefix). @@ -6664,7 +6870,7 @@ type WriteCondition_Clause struct { func (x *WriteCondition_Clause) Reset() { *x = WriteCondition_Clause{} - mi := &file_filer_proto_msgTypes[92] + mi := &file_filer_proto_msgTypes[95] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -6676,7 +6882,7 @@ func (x *WriteCondition_Clause) String() string { func (*WriteCondition_Clause) ProtoMessage() {} func (x *WriteCondition_Clause) ProtoReflect() protoreflect.Message { - mi := &file_filer_proto_msgTypes[92] + mi := &file_filer_proto_msgTypes[95] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -6767,7 +6973,7 @@ type LocateBrokerResponse_Resource struct { func (x *LocateBrokerResponse_Resource) Reset() { *x = LocateBrokerResponse_Resource{} - mi := &file_filer_proto_msgTypes[97] + mi := &file_filer_proto_msgTypes[100] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -6779,7 +6985,7 @@ func (x *LocateBrokerResponse_Resource) String() string { func (*LocateBrokerResponse_Resource) ProtoMessage() {} func (x *LocateBrokerResponse_Resource) ProtoReflect() protoreflect.Message { - mi := &file_filer_proto_msgTypes[97] + mi := &file_filer_proto_msgTypes[100] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -6834,7 +7040,7 @@ type FilerConf_PathConf struct { func (x *FilerConf_PathConf) Reset() { *x = FilerConf_PathConf{} - mi := &file_filer_proto_msgTypes[98] + mi := &file_filer_proto_msgTypes[101] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -6846,7 +7052,7 @@ func (x *FilerConf_PathConf) String() string { func (*FilerConf_PathConf) ProtoMessage() {} func (x *FilerConf_PathConf) ProtoReflect() protoreflect.Message { - mi := &file_filer_proto_msgTypes[98] + mi := &file_filer_proto_msgTypes[101] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -7585,7 +7791,26 @@ const file_filer_proto_rawDesc = "" + "\flast_seen_ns\x18\x03 \x01(\x03R\n" + "lastSeenNs\x12\x1f\n" + "\vdata_center\x18\x04 \x01(\tR\n" + - "dataCenter*7\n" + + "dataCenter\"\x9f\x01\n" + + "\x1dLookupDirectoryEntriesRequest\x12A\n" + + "\brequests\x18\x01 \x03(\v2%.filer_pb.LookupDirectoryEntryRequestR\brequests\x12;\n" + + "\x1aunavailable_volume_is_miss\x18\x02 \x01(\bR\x17unavailableVolumeIsMiss\"\xb0\x01\n" + + "\x1aLookupDirectoryEntryResult\x12\x14\n" + + "\x05found\x18\x01 \x01(\bR\x05found\x12%\n" + + "\x05entry\x18\x02 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x12\x14\n" + + "\x05error\x18\x03 \x01(\tR\x05error\x12\x1a\n" + + "\tlog_ts_ns\x18\x04 \x01(\x03R\alogTsNs\x12#\n" + + "\rlog_signature\x18\x05 \x01(\x05R\flogSignature\"\xa9\x03\n" + + "\x1eLookupDirectoryEntriesResponse\x12>\n" + + "\aresults\x18\x01 \x03(\v2$.filer_pb.LookupDirectoryEntryResultR\aresults\x12_\n" + + "\rlocations_map\x18\x02 \x03(\v2:.filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntryR\flocationsMap\x12S\n" + + "\tread_auth\x18\x03 \x03(\v26.filer_pb.LookupDirectoryEntriesResponse.ReadAuthEntryR\breadAuth\x1aT\n" + + "\x11LocationsMapEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12)\n" + + "\x05value\x18\x02 \x01(\v2\x13.filer_pb.LocationsR\x05value:\x028\x01\x1a;\n" + + "\rReadAuthEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01*7\n" + "\aSSEType\x12\b\n" + "\x04NONE\x10\x00\x12\t\n" + "\x05SSE_C\x10\x01\x12\v\n" + @@ -7610,9 +7835,10 @@ const file_filer_proto_rawDesc = "" + "\x13RELEASE_POSIX_OWNER\x10\x03\x12\x17\n" + "\x13RELEASE_FLOCK_OWNER\x10\x04\x12\x0e\n" + "\n" + - "KEEP_ALIVE\x10\x052\xae\x17\n" + + "KEEP_ALIVE\x10\x052\x9d\x18\n" + "\fSeaweedFiler\x12g\n" + - "\x14LookupDirectoryEntry\x12%.filer_pb.LookupDirectoryEntryRequest\x1a&.filer_pb.LookupDirectoryEntryResponse\"\x00\x12N\n" + + "\x14LookupDirectoryEntry\x12%.filer_pb.LookupDirectoryEntryRequest\x1a&.filer_pb.LookupDirectoryEntryResponse\"\x00\x12m\n" + + "\x16LookupDirectoryEntries\x12'.filer_pb.LookupDirectoryEntriesRequest\x1a(.filer_pb.LookupDirectoryEntriesResponse\"\x00\x12N\n" + "\vListEntries\x12\x1c.filer_pb.ListEntriesRequest\x1a\x1d.filer_pb.ListEntriesResponse\"\x000\x01\x12L\n" + "\vCreateEntry\x12\x1c.filer_pb.CreateEntryRequest\x1a\x1d.filer_pb.CreateEntryResponse\"\x00\x12L\n" + "\vUpdateEntry\x12\x1c.filer_pb.UpdateEntryRequest\x1a\x1d.filer_pb.UpdateEntryResponse\"\x00\x12X\n" + @@ -7663,7 +7889,7 @@ func file_filer_proto_rawDescGZIP() []byte { } var file_filer_proto_enumTypes = make([]protoimpl.EnumInfo, 5) -var file_filer_proto_msgTypes = make([]protoimpl.MessageInfo, 99) +var file_filer_proto_msgTypes = make([]protoimpl.MessageInfo, 104) var file_filer_proto_goTypes = []any{ (SSEType)(0), // 0: filer_pb.SSEType (FilerError)(0), // 1: filer_pb.FilerError @@ -7761,21 +7987,26 @@ var file_filer_proto_goTypes = []any{ (*MountListRequest)(nil), // 93: filer_pb.MountListRequest (*MountListResponse)(nil), // 94: filer_pb.MountListResponse (*MountInfo)(nil), // 95: filer_pb.MountInfo - nil, // 96: filer_pb.Entry.ExtendedEntry - (*WriteCondition_Clause)(nil), // 97: filer_pb.WriteCondition.Clause - nil, // 98: filer_pb.ObjectMutation.SetExtendedEntry - nil, // 99: filer_pb.Recompute.CopyExtendedEntry - nil, // 100: filer_pb.UpdateEntryRequest.ExpectedExtendedEntry - nil, // 101: filer_pb.LookupVolumeResponse.LocationsMapEntry - (*LocateBrokerResponse_Resource)(nil), // 102: filer_pb.LocateBrokerResponse.Resource - (*FilerConf_PathConf)(nil), // 103: filer_pb.FilerConf.PathConf + (*LookupDirectoryEntriesRequest)(nil), // 96: filer_pb.LookupDirectoryEntriesRequest + (*LookupDirectoryEntryResult)(nil), // 97: filer_pb.LookupDirectoryEntryResult + (*LookupDirectoryEntriesResponse)(nil), // 98: filer_pb.LookupDirectoryEntriesResponse + nil, // 99: filer_pb.Entry.ExtendedEntry + (*WriteCondition_Clause)(nil), // 100: filer_pb.WriteCondition.Clause + nil, // 101: filer_pb.ObjectMutation.SetExtendedEntry + nil, // 102: filer_pb.Recompute.CopyExtendedEntry + nil, // 103: filer_pb.UpdateEntryRequest.ExpectedExtendedEntry + nil, // 104: filer_pb.LookupVolumeResponse.LocationsMapEntry + (*LocateBrokerResponse_Resource)(nil), // 105: filer_pb.LocateBrokerResponse.Resource + (*FilerConf_PathConf)(nil), // 106: filer_pb.FilerConf.PathConf + nil, // 107: filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry + nil, // 108: filer_pb.LookupDirectoryEntriesResponse.ReadAuthEntry } var file_filer_proto_depIdxs = []int32{ 10, // 0: filer_pb.LookupDirectoryEntryResponse.entry:type_name -> filer_pb.Entry 10, // 1: filer_pb.ListEntriesResponse.entry:type_name -> filer_pb.Entry 13, // 2: filer_pb.Entry.chunks:type_name -> filer_pb.FileChunk 16, // 3: filer_pb.Entry.attributes:type_name -> filer_pb.FuseAttributes - 96, // 4: filer_pb.Entry.extended:type_name -> filer_pb.Entry.ExtendedEntry + 99, // 4: filer_pb.Entry.extended:type_name -> filer_pb.Entry.ExtendedEntry 9, // 5: filer_pb.Entry.remote_entry:type_name -> filer_pb.RemoteEntry 10, // 6: filer_pb.FullEntry.entry:type_name -> filer_pb.Entry 10, // 7: filer_pb.EventNotification.old_entry:type_name -> filer_pb.Entry @@ -7786,12 +8017,12 @@ var file_filer_proto_depIdxs = []int32{ 13, // 12: filer_pb.FileChunkManifest.chunks:type_name -> filer_pb.FileChunk 10, // 13: filer_pb.CreateEntryRequest.entry:type_name -> filer_pb.Entry 18, // 14: filer_pb.CreateEntryRequest.condition:type_name -> filer_pb.WriteCondition - 97, // 15: filer_pb.WriteCondition.clauses:type_name -> filer_pb.WriteCondition.Clause + 100, // 15: filer_pb.WriteCondition.clauses:type_name -> filer_pb.WriteCondition.Clause 4, // 16: filer_pb.ObjectMutation.type:type_name -> filer_pb.ObjectMutation.Type 10, // 17: filer_pb.ObjectMutation.entry:type_name -> filer_pb.Entry - 98, // 18: filer_pb.ObjectMutation.set_extended:type_name -> filer_pb.ObjectMutation.SetExtendedEntry + 101, // 18: filer_pb.ObjectMutation.set_extended:type_name -> filer_pb.ObjectMutation.SetExtendedEntry 20, // 19: filer_pb.ObjectMutation.recompute:type_name -> filer_pb.Recompute - 99, // 20: filer_pb.Recompute.copy_extended:type_name -> filer_pb.Recompute.CopyExtendedEntry + 102, // 20: filer_pb.Recompute.copy_extended:type_name -> filer_pb.Recompute.CopyExtendedEntry 18, // 21: filer_pb.ObjectTransactionRequest.condition:type_name -> filer_pb.WriteCondition 19, // 22: filer_pb.ObjectTransactionRequest.mutations:type_name -> filer_pb.ObjectMutation 1, // 23: filer_pb.ObjectTransactionResponse.error_code:type_name -> filer_pb.FilerError @@ -7804,7 +8035,7 @@ var file_filer_proto_depIdxs = []int32{ 59, // 30: filer_pb.CreateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse 1, // 31: filer_pb.CreateEntryResponse.error_code:type_name -> filer_pb.FilerError 10, // 32: filer_pb.UpdateEntryRequest.entry:type_name -> filer_pb.Entry - 100, // 33: filer_pb.UpdateEntryRequest.expected_extended:type_name -> filer_pb.UpdateEntryRequest.ExpectedExtendedEntry + 103, // 33: filer_pb.UpdateEntryRequest.expected_extended:type_name -> filer_pb.UpdateEntryRequest.ExpectedExtendedEntry 18, // 34: filer_pb.UpdateEntryRequest.condition:type_name -> filer_pb.WriteCondition 59, // 35: filer_pb.UpdateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse 13, // 36: filer_pb.AppendToEntryRequest.chunks:type_name -> filer_pb.FileChunk @@ -7813,7 +8044,7 @@ var file_filer_proto_depIdxs = []int32{ 45, // 39: filer_pb.AssignVolumeResponse.location:type_name -> filer_pb.Location 45, // 40: filer_pb.AssignVolumeResponse.replicas:type_name -> filer_pb.Location 45, // 41: filer_pb.Locations.locations:type_name -> filer_pb.Location - 101, // 42: filer_pb.LookupVolumeResponse.locations_map:type_name -> filer_pb.LookupVolumeResponse.LocationsMapEntry + 104, // 42: filer_pb.LookupVolumeResponse.locations_map:type_name -> filer_pb.LookupVolumeResponse.LocationsMapEntry 47, // 43: filer_pb.CollectionListResponse.collections:type_name -> filer_pb.Collection 12, // 44: filer_pb.SubscribeMetadataResponse.event_notification:type_name -> filer_pb.EventNotification 59, // 45: filer_pb.SubscribeMetadataResponse.events:type_name -> filer_pb.SubscribeMetadataResponse @@ -7821,8 +8052,8 @@ var file_filer_proto_depIdxs = []int32{ 62, // 47: filer_pb.ListMetadataSubscribersResponse.subscribers:type_name -> filer_pb.MetadataSubscriber 13, // 48: filer_pb.LogFileChunkRef.chunks:type_name -> filer_pb.FileChunk 10, // 49: filer_pb.TraverseBfsMetadataResponse.entry:type_name -> filer_pb.Entry - 102, // 50: filer_pb.LocateBrokerResponse.resources:type_name -> filer_pb.LocateBrokerResponse.Resource - 103, // 51: filer_pb.FilerConf.locations:type_name -> filer_pb.FilerConf.PathConf + 105, // 50: filer_pb.LocateBrokerResponse.resources:type_name -> filer_pb.LocateBrokerResponse.Resource + 106, // 51: filer_pb.FilerConf.locations:type_name -> filer_pb.FilerConf.PathConf 10, // 52: filer_pb.CacheRemoteObjectToLocalClusterResponse.entry:type_name -> filer_pb.Entry 59, // 53: filer_pb.CacheRemoteObjectToLocalClusterResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse 84, // 54: filer_pb.TransferLocksRequest.locks:type_name -> filer_pb.Lock @@ -7835,81 +8066,89 @@ var file_filer_proto_depIdxs = []int32{ 36, // 61: filer_pb.StreamMutateEntryResponse.delete_response:type_name -> filer_pb.DeleteEntryResponse 40, // 62: filer_pb.StreamMutateEntryResponse.rename_response:type_name -> filer_pb.StreamRenameEntryResponse 95, // 63: filer_pb.MountListResponse.mounts:type_name -> filer_pb.MountInfo - 3, // 64: filer_pb.WriteCondition.Clause.kind:type_name -> filer_pb.WriteCondition.Kind - 44, // 65: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations - 5, // 66: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest - 7, // 67: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest - 17, // 68: filer_pb.SeaweedFiler.CreateEntry:input_type -> filer_pb.CreateEntryRequest - 29, // 69: filer_pb.SeaweedFiler.UpdateEntry:input_type -> filer_pb.UpdateEntryRequest - 31, // 70: filer_pb.SeaweedFiler.TouchAccessTime:input_type -> filer_pb.TouchAccessTimeRequest - 33, // 71: filer_pb.SeaweedFiler.AppendToEntry:input_type -> filer_pb.AppendToEntryRequest - 35, // 72: filer_pb.SeaweedFiler.DeleteEntry:input_type -> filer_pb.DeleteEntryRequest - 21, // 73: filer_pb.SeaweedFiler.ObjectTransaction:input_type -> filer_pb.ObjectTransactionRequest - 26, // 74: filer_pb.SeaweedFiler.ObjectTransactionBatch:input_type -> filer_pb.ObjectTransactionBatchRequest - 24, // 75: filer_pb.SeaweedFiler.PosixLock:input_type -> filer_pb.PosixLockRequest - 37, // 76: filer_pb.SeaweedFiler.AtomicRenameEntry:input_type -> filer_pb.AtomicRenameEntryRequest - 39, // 77: filer_pb.SeaweedFiler.StreamRenameEntry:input_type -> filer_pb.StreamRenameEntryRequest - 89, // 78: filer_pb.SeaweedFiler.StreamMutateEntry:input_type -> filer_pb.StreamMutateEntryRequest - 41, // 79: filer_pb.SeaweedFiler.AssignVolume:input_type -> filer_pb.AssignVolumeRequest - 43, // 80: filer_pb.SeaweedFiler.LookupVolume:input_type -> filer_pb.LookupVolumeRequest - 48, // 81: filer_pb.SeaweedFiler.CollectionList:input_type -> filer_pb.CollectionListRequest - 50, // 82: filer_pb.SeaweedFiler.DeleteCollection:input_type -> filer_pb.DeleteCollectionRequest - 52, // 83: filer_pb.SeaweedFiler.Statistics:input_type -> filer_pb.StatisticsRequest - 54, // 84: filer_pb.SeaweedFiler.Ping:input_type -> filer_pb.PingRequest - 56, // 85: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest - 64, // 86: filer_pb.SeaweedFiler.TraverseBfsMetadata:input_type -> filer_pb.TraverseBfsMetadataRequest - 58, // 87: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest - 58, // 88: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest - 60, // 89: filer_pb.SeaweedFiler.ListMetadataSubscribers:input_type -> filer_pb.ListMetadataSubscribersRequest - 71, // 90: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest - 73, // 91: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest - 76, // 92: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest - 78, // 93: filer_pb.SeaweedFiler.DistributedLock:input_type -> filer_pb.LockRequest - 80, // 94: filer_pb.SeaweedFiler.DistributedUnlock:input_type -> filer_pb.UnlockRequest - 82, // 95: filer_pb.SeaweedFiler.FindLockOwner:input_type -> filer_pb.FindLockOwnerRequest - 85, // 96: filer_pb.SeaweedFiler.TransferLocks:input_type -> filer_pb.TransferLocksRequest - 87, // 97: filer_pb.SeaweedFiler.ReplicateLock:input_type -> filer_pb.ReplicateLockRequest - 91, // 98: filer_pb.SeaweedFiler.MountRegister:input_type -> filer_pb.MountRegisterRequest - 93, // 99: filer_pb.SeaweedFiler.MountList:input_type -> filer_pb.MountListRequest - 6, // 100: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse - 8, // 101: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse - 28, // 102: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse - 30, // 103: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse - 32, // 104: filer_pb.SeaweedFiler.TouchAccessTime:output_type -> filer_pb.TouchAccessTimeResponse - 34, // 105: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse - 36, // 106: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse - 22, // 107: filer_pb.SeaweedFiler.ObjectTransaction:output_type -> filer_pb.ObjectTransactionResponse - 27, // 108: filer_pb.SeaweedFiler.ObjectTransactionBatch:output_type -> filer_pb.ObjectTransactionBatchResponse - 25, // 109: filer_pb.SeaweedFiler.PosixLock:output_type -> filer_pb.PosixLockResponse - 38, // 110: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse - 40, // 111: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse - 90, // 112: filer_pb.SeaweedFiler.StreamMutateEntry:output_type -> filer_pb.StreamMutateEntryResponse - 42, // 113: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse - 46, // 114: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse - 49, // 115: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse - 51, // 116: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse - 53, // 117: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse - 55, // 118: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse - 57, // 119: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse - 65, // 120: filer_pb.SeaweedFiler.TraverseBfsMetadata:output_type -> filer_pb.TraverseBfsMetadataResponse - 59, // 121: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse - 59, // 122: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse - 61, // 123: filer_pb.SeaweedFiler.ListMetadataSubscribers:output_type -> filer_pb.ListMetadataSubscribersResponse - 72, // 124: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse - 74, // 125: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse - 77, // 126: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse - 79, // 127: filer_pb.SeaweedFiler.DistributedLock:output_type -> filer_pb.LockResponse - 81, // 128: filer_pb.SeaweedFiler.DistributedUnlock:output_type -> filer_pb.UnlockResponse - 83, // 129: filer_pb.SeaweedFiler.FindLockOwner:output_type -> filer_pb.FindLockOwnerResponse - 86, // 130: filer_pb.SeaweedFiler.TransferLocks:output_type -> filer_pb.TransferLocksResponse - 88, // 131: filer_pb.SeaweedFiler.ReplicateLock:output_type -> filer_pb.ReplicateLockResponse - 92, // 132: filer_pb.SeaweedFiler.MountRegister:output_type -> filer_pb.MountRegisterResponse - 94, // 133: filer_pb.SeaweedFiler.MountList:output_type -> filer_pb.MountListResponse - 100, // [100:134] is the sub-list for method output_type - 66, // [66:100] is the sub-list for method input_type - 66, // [66:66] is the sub-list for extension type_name - 66, // [66:66] is the sub-list for extension extendee - 0, // [0:66] is the sub-list for field type_name + 5, // 64: filer_pb.LookupDirectoryEntriesRequest.requests:type_name -> filer_pb.LookupDirectoryEntryRequest + 10, // 65: filer_pb.LookupDirectoryEntryResult.entry:type_name -> filer_pb.Entry + 97, // 66: filer_pb.LookupDirectoryEntriesResponse.results:type_name -> filer_pb.LookupDirectoryEntryResult + 107, // 67: filer_pb.LookupDirectoryEntriesResponse.locations_map:type_name -> filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry + 108, // 68: filer_pb.LookupDirectoryEntriesResponse.read_auth:type_name -> filer_pb.LookupDirectoryEntriesResponse.ReadAuthEntry + 3, // 69: filer_pb.WriteCondition.Clause.kind:type_name -> filer_pb.WriteCondition.Kind + 44, // 70: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations + 44, // 71: filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations + 5, // 72: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest + 96, // 73: filer_pb.SeaweedFiler.LookupDirectoryEntries:input_type -> filer_pb.LookupDirectoryEntriesRequest + 7, // 74: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest + 17, // 75: filer_pb.SeaweedFiler.CreateEntry:input_type -> filer_pb.CreateEntryRequest + 29, // 76: filer_pb.SeaweedFiler.UpdateEntry:input_type -> filer_pb.UpdateEntryRequest + 31, // 77: filer_pb.SeaweedFiler.TouchAccessTime:input_type -> filer_pb.TouchAccessTimeRequest + 33, // 78: filer_pb.SeaweedFiler.AppendToEntry:input_type -> filer_pb.AppendToEntryRequest + 35, // 79: filer_pb.SeaweedFiler.DeleteEntry:input_type -> filer_pb.DeleteEntryRequest + 21, // 80: filer_pb.SeaweedFiler.ObjectTransaction:input_type -> filer_pb.ObjectTransactionRequest + 26, // 81: filer_pb.SeaweedFiler.ObjectTransactionBatch:input_type -> filer_pb.ObjectTransactionBatchRequest + 24, // 82: filer_pb.SeaweedFiler.PosixLock:input_type -> filer_pb.PosixLockRequest + 37, // 83: filer_pb.SeaweedFiler.AtomicRenameEntry:input_type -> filer_pb.AtomicRenameEntryRequest + 39, // 84: filer_pb.SeaweedFiler.StreamRenameEntry:input_type -> filer_pb.StreamRenameEntryRequest + 89, // 85: filer_pb.SeaweedFiler.StreamMutateEntry:input_type -> filer_pb.StreamMutateEntryRequest + 41, // 86: filer_pb.SeaweedFiler.AssignVolume:input_type -> filer_pb.AssignVolumeRequest + 43, // 87: filer_pb.SeaweedFiler.LookupVolume:input_type -> filer_pb.LookupVolumeRequest + 48, // 88: filer_pb.SeaweedFiler.CollectionList:input_type -> filer_pb.CollectionListRequest + 50, // 89: filer_pb.SeaweedFiler.DeleteCollection:input_type -> filer_pb.DeleteCollectionRequest + 52, // 90: filer_pb.SeaweedFiler.Statistics:input_type -> filer_pb.StatisticsRequest + 54, // 91: filer_pb.SeaweedFiler.Ping:input_type -> filer_pb.PingRequest + 56, // 92: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest + 64, // 93: filer_pb.SeaweedFiler.TraverseBfsMetadata:input_type -> filer_pb.TraverseBfsMetadataRequest + 58, // 94: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest + 58, // 95: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest + 60, // 96: filer_pb.SeaweedFiler.ListMetadataSubscribers:input_type -> filer_pb.ListMetadataSubscribersRequest + 71, // 97: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest + 73, // 98: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest + 76, // 99: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest + 78, // 100: filer_pb.SeaweedFiler.DistributedLock:input_type -> filer_pb.LockRequest + 80, // 101: filer_pb.SeaweedFiler.DistributedUnlock:input_type -> filer_pb.UnlockRequest + 82, // 102: filer_pb.SeaweedFiler.FindLockOwner:input_type -> filer_pb.FindLockOwnerRequest + 85, // 103: filer_pb.SeaweedFiler.TransferLocks:input_type -> filer_pb.TransferLocksRequest + 87, // 104: filer_pb.SeaweedFiler.ReplicateLock:input_type -> filer_pb.ReplicateLockRequest + 91, // 105: filer_pb.SeaweedFiler.MountRegister:input_type -> filer_pb.MountRegisterRequest + 93, // 106: filer_pb.SeaweedFiler.MountList:input_type -> filer_pb.MountListRequest + 6, // 107: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse + 98, // 108: filer_pb.SeaweedFiler.LookupDirectoryEntries:output_type -> filer_pb.LookupDirectoryEntriesResponse + 8, // 109: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse + 28, // 110: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse + 30, // 111: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse + 32, // 112: filer_pb.SeaweedFiler.TouchAccessTime:output_type -> filer_pb.TouchAccessTimeResponse + 34, // 113: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse + 36, // 114: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse + 22, // 115: filer_pb.SeaweedFiler.ObjectTransaction:output_type -> filer_pb.ObjectTransactionResponse + 27, // 116: filer_pb.SeaweedFiler.ObjectTransactionBatch:output_type -> filer_pb.ObjectTransactionBatchResponse + 25, // 117: filer_pb.SeaweedFiler.PosixLock:output_type -> filer_pb.PosixLockResponse + 38, // 118: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse + 40, // 119: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse + 90, // 120: filer_pb.SeaweedFiler.StreamMutateEntry:output_type -> filer_pb.StreamMutateEntryResponse + 42, // 121: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse + 46, // 122: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse + 49, // 123: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse + 51, // 124: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse + 53, // 125: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse + 55, // 126: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse + 57, // 127: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse + 65, // 128: filer_pb.SeaweedFiler.TraverseBfsMetadata:output_type -> filer_pb.TraverseBfsMetadataResponse + 59, // 129: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse + 59, // 130: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse + 61, // 131: filer_pb.SeaweedFiler.ListMetadataSubscribers:output_type -> filer_pb.ListMetadataSubscribersResponse + 72, // 132: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse + 74, // 133: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse + 77, // 134: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse + 79, // 135: filer_pb.SeaweedFiler.DistributedLock:output_type -> filer_pb.LockResponse + 81, // 136: filer_pb.SeaweedFiler.DistributedUnlock:output_type -> filer_pb.UnlockResponse + 83, // 137: filer_pb.SeaweedFiler.FindLockOwner:output_type -> filer_pb.FindLockOwnerResponse + 86, // 138: filer_pb.SeaweedFiler.TransferLocks:output_type -> filer_pb.TransferLocksResponse + 88, // 139: filer_pb.SeaweedFiler.ReplicateLock:output_type -> filer_pb.ReplicateLockResponse + 92, // 140: filer_pb.SeaweedFiler.MountRegister:output_type -> filer_pb.MountRegisterResponse + 94, // 141: filer_pb.SeaweedFiler.MountList:output_type -> filer_pb.MountListResponse + 107, // [107:142] is the sub-list for method output_type + 72, // [72:107] is the sub-list for method input_type + 72, // [72:72] is the sub-list for extension type_name + 72, // [72:72] is the sub-list for extension extendee + 0, // [0:72] is the sub-list for field type_name } func init() { file_filer_proto_init() } @@ -7930,14 +8169,14 @@ func file_filer_proto_init() { (*StreamMutateEntryResponse_DeleteResponse)(nil), (*StreamMutateEntryResponse_RenameResponse)(nil), } - file_filer_proto_msgTypes[98].OneofWrappers = []any{} + file_filer_proto_msgTypes[101].OneofWrappers = []any{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_filer_proto_rawDesc), len(file_filer_proto_rawDesc)), NumEnums: 5, - NumMessages: 99, + NumMessages: 104, NumExtensions: 0, NumServices: 1, }, diff --git a/weed/pb/filer_pb/filer_grpc.pb.go b/weed/pb/filer_pb/filer_grpc.pb.go index 2bbe9e23c..6a2b494a3 100644 --- a/weed/pb/filer_pb/filer_grpc.pb.go +++ b/weed/pb/filer_pb/filer_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.1 +// - protoc v7.35.0 // source: filer.proto package filer_pb @@ -20,6 +20,7 @@ const _ = grpc.SupportPackageIsVersion9 const ( SeaweedFiler_LookupDirectoryEntry_FullMethodName = "/filer_pb.SeaweedFiler/LookupDirectoryEntry" + SeaweedFiler_LookupDirectoryEntries_FullMethodName = "/filer_pb.SeaweedFiler/LookupDirectoryEntries" SeaweedFiler_ListEntries_FullMethodName = "/filer_pb.SeaweedFiler/ListEntries" SeaweedFiler_CreateEntry_FullMethodName = "/filer_pb.SeaweedFiler/CreateEntry" SeaweedFiler_UpdateEntry_FullMethodName = "/filer_pb.SeaweedFiler/UpdateEntry" @@ -60,6 +61,7 @@ const ( // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. type SeaweedFilerClient interface { LookupDirectoryEntry(ctx context.Context, in *LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*LookupDirectoryEntryResponse, error) + LookupDirectoryEntries(ctx context.Context, in *LookupDirectoryEntriesRequest, opts ...grpc.CallOption) (*LookupDirectoryEntriesResponse, error) ListEntries(ctx context.Context, in *ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ListEntriesResponse], error) CreateEntry(ctx context.Context, in *CreateEntryRequest, opts ...grpc.CallOption) (*CreateEntryResponse, error) UpdateEntry(ctx context.Context, in *UpdateEntryRequest, opts ...grpc.CallOption) (*UpdateEntryResponse, error) @@ -118,6 +120,16 @@ func (c *seaweedFilerClient) LookupDirectoryEntry(ctx context.Context, in *Looku return out, nil } +func (c *seaweedFilerClient) LookupDirectoryEntries(ctx context.Context, in *LookupDirectoryEntriesRequest, opts ...grpc.CallOption) (*LookupDirectoryEntriesResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(LookupDirectoryEntriesResponse) + err := c.cc.Invoke(ctx, SeaweedFiler_LookupDirectoryEntries_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + func (c *seaweedFilerClient) ListEntries(ctx context.Context, in *ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ListEntriesResponse], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) stream, err := c.cc.NewStream(ctx, &SeaweedFiler_ServiceDesc.Streams[0], SeaweedFiler_ListEntries_FullMethodName, cOpts...) @@ -501,6 +513,7 @@ func (c *seaweedFilerClient) MountList(ctx context.Context, in *MountListRequest // for forward compatibility. type SeaweedFilerServer interface { LookupDirectoryEntry(context.Context, *LookupDirectoryEntryRequest) (*LookupDirectoryEntryResponse, error) + LookupDirectoryEntries(context.Context, *LookupDirectoryEntriesRequest) (*LookupDirectoryEntriesResponse, error) ListEntries(*ListEntriesRequest, grpc.ServerStreamingServer[ListEntriesResponse]) error CreateEntry(context.Context, *CreateEntryRequest) (*CreateEntryResponse, error) UpdateEntry(context.Context, *UpdateEntryRequest) (*UpdateEntryResponse, error) @@ -552,6 +565,9 @@ type UnimplementedSeaweedFilerServer struct{} func (UnimplementedSeaweedFilerServer) LookupDirectoryEntry(context.Context, *LookupDirectoryEntryRequest) (*LookupDirectoryEntryResponse, error) { return nil, status.Error(codes.Unimplemented, "method LookupDirectoryEntry not implemented") } +func (UnimplementedSeaweedFilerServer) LookupDirectoryEntries(context.Context, *LookupDirectoryEntriesRequest) (*LookupDirectoryEntriesResponse, error) { + return nil, status.Error(codes.Unimplemented, "method LookupDirectoryEntries not implemented") +} func (UnimplementedSeaweedFilerServer) ListEntries(*ListEntriesRequest, grpc.ServerStreamingServer[ListEntriesResponse]) error { return status.Error(codes.Unimplemented, "method ListEntries not implemented") } @@ -690,6 +706,24 @@ func _SeaweedFiler_LookupDirectoryEntry_Handler(srv interface{}, ctx context.Con return interceptor(ctx, in, info, handler) } +func _SeaweedFiler_LookupDirectoryEntries_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(LookupDirectoryEntriesRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(SeaweedFilerServer).LookupDirectoryEntries(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: SeaweedFiler_LookupDirectoryEntries_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(SeaweedFilerServer).LookupDirectoryEntries(ctx, req.(*LookupDirectoryEntriesRequest)) + } + return interceptor(ctx, in, info, handler) +} + func _SeaweedFiler_ListEntries_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(ListEntriesRequest) if err := stream.RecvMsg(m); err != nil { @@ -1249,6 +1283,10 @@ var SeaweedFiler_ServiceDesc = grpc.ServiceDesc{ MethodName: "LookupDirectoryEntry", Handler: _SeaweedFiler_LookupDirectoryEntry_Handler, }, + { + MethodName: "LookupDirectoryEntries", + Handler: _SeaweedFiler_LookupDirectoryEntries_Handler, + }, { MethodName: "CreateEntry", Handler: _SeaweedFiler_CreateEntry_Handler, diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 0b7b4d153..17e958767 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -6265,6 +6265,217 @@ func (m *MountInfo) MarshalToSizedBufferVT(dAtA []byte) (int, error) { return len(dAtA) - i, nil } +func (m *LookupDirectoryEntriesRequest) MarshalVT() (dAtA []byte, err error) { + if m == nil { + return nil, nil + } + size := m.SizeVT() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBufferVT(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *LookupDirectoryEntriesRequest) MarshalToVT(dAtA []byte) (int, error) { + size := m.SizeVT() + return m.MarshalToSizedBufferVT(dAtA[:size]) +} + +func (m *LookupDirectoryEntriesRequest) MarshalToSizedBufferVT(dAtA []byte) (int, error) { + if m == nil { + return 0, nil + } + i := len(dAtA) + _ = i + var l int + _ = l + if m.unknownFields != nil { + i -= len(m.unknownFields) + copy(dAtA[i:], m.unknownFields) + } + if m.UnavailableVolumeIsMiss { + i-- + if m.UnavailableVolumeIsMiss { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x10 + } + if len(m.Requests) > 0 { + for iNdEx := len(m.Requests) - 1; iNdEx >= 0; iNdEx-- { + size, err := m.Requests[iNdEx].MarshalToSizedBufferVT(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = protohelpers.EncodeVarint(dAtA, i, uint64(size)) + i-- + dAtA[i] = 0xa + } + } + return len(dAtA) - i, nil +} + +func (m *LookupDirectoryEntryResult) MarshalVT() (dAtA []byte, err error) { + if m == nil { + return nil, nil + } + size := m.SizeVT() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBufferVT(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *LookupDirectoryEntryResult) MarshalToVT(dAtA []byte) (int, error) { + size := m.SizeVT() + return m.MarshalToSizedBufferVT(dAtA[:size]) +} + +func (m *LookupDirectoryEntryResult) MarshalToSizedBufferVT(dAtA []byte) (int, error) { + if m == nil { + return 0, nil + } + i := len(dAtA) + _ = i + var l int + _ = l + if m.unknownFields != nil { + i -= len(m.unknownFields) + copy(dAtA[i:], m.unknownFields) + } + if m.LogSignature != 0 { + i = protohelpers.EncodeVarint(dAtA, i, uint64(m.LogSignature)) + i-- + dAtA[i] = 0x28 + } + if m.LogTsNs != 0 { + i = protohelpers.EncodeVarint(dAtA, i, uint64(m.LogTsNs)) + i-- + dAtA[i] = 0x20 + } + if len(m.Error) > 0 { + i -= len(m.Error) + copy(dAtA[i:], m.Error) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.Error))) + i-- + dAtA[i] = 0x1a + } + if m.Entry != nil { + size, err := m.Entry.MarshalToSizedBufferVT(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = protohelpers.EncodeVarint(dAtA, i, uint64(size)) + i-- + dAtA[i] = 0x12 + } + if m.Found { + i-- + if m.Found { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x8 + } + return len(dAtA) - i, nil +} + +func (m *LookupDirectoryEntriesResponse) MarshalVT() (dAtA []byte, err error) { + if m == nil { + return nil, nil + } + size := m.SizeVT() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBufferVT(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *LookupDirectoryEntriesResponse) MarshalToVT(dAtA []byte) (int, error) { + size := m.SizeVT() + return m.MarshalToSizedBufferVT(dAtA[:size]) +} + +func (m *LookupDirectoryEntriesResponse) MarshalToSizedBufferVT(dAtA []byte) (int, error) { + if m == nil { + return 0, nil + } + i := len(dAtA) + _ = i + var l int + _ = l + if m.unknownFields != nil { + i -= len(m.unknownFields) + copy(dAtA[i:], m.unknownFields) + } + if len(m.ReadAuth) > 0 { + for k := range m.ReadAuth { + v := m.ReadAuth[k] + baseI := i + i -= len(v) + copy(dAtA[i:], v) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(v))) + i-- + dAtA[i] = 0x12 + i -= len(k) + copy(dAtA[i:], k) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(k))) + i-- + dAtA[i] = 0xa + i = protohelpers.EncodeVarint(dAtA, i, uint64(baseI-i)) + i-- + dAtA[i] = 0x1a + } + } + if len(m.LocationsMap) > 0 { + for k := range m.LocationsMap { + v := m.LocationsMap[k] + baseI := i + size, err := v.MarshalToSizedBufferVT(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = protohelpers.EncodeVarint(dAtA, i, uint64(size)) + i-- + dAtA[i] = 0x12 + i -= len(k) + copy(dAtA[i:], k) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(k))) + i-- + dAtA[i] = 0xa + i = protohelpers.EncodeVarint(dAtA, i, uint64(baseI-i)) + i-- + dAtA[i] = 0x12 + } + } + if len(m.Results) > 0 { + for iNdEx := len(m.Results) - 1; iNdEx >= 0; iNdEx-- { + size, err := m.Results[iNdEx].MarshalToSizedBufferVT(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = protohelpers.EncodeVarint(dAtA, i, uint64(size)) + i-- + dAtA[i] = 0xa + } + } + return len(dAtA) - i, nil +} + func (m *LookupDirectoryEntryRequest) SizeVT() (n int) { if m == nil { return 0 @@ -8777,6 +8988,89 @@ func (m *MountInfo) SizeVT() (n int) { return n } +func (m *LookupDirectoryEntriesRequest) SizeVT() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if len(m.Requests) > 0 { + for _, e := range m.Requests { + l = e.SizeVT() + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } + } + if m.UnavailableVolumeIsMiss { + n += 2 + } + n += len(m.unknownFields) + return n +} + +func (m *LookupDirectoryEntryResult) SizeVT() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if m.Found { + n += 2 + } + if m.Entry != nil { + l = m.Entry.SizeVT() + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } + l = len(m.Error) + if l > 0 { + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } + if m.LogTsNs != 0 { + n += 1 + protohelpers.SizeOfVarint(uint64(m.LogTsNs)) + } + if m.LogSignature != 0 { + n += 1 + protohelpers.SizeOfVarint(uint64(m.LogSignature)) + } + n += len(m.unknownFields) + return n +} + +func (m *LookupDirectoryEntriesResponse) SizeVT() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if len(m.Results) > 0 { + for _, e := range m.Results { + l = e.SizeVT() + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } + } + if len(m.LocationsMap) > 0 { + for k, v := range m.LocationsMap { + _ = k + _ = v + l = 0 + if v != nil { + l = v.SizeVT() + } + l += 1 + protohelpers.SizeOfVarint(uint64(l)) + mapEntrySize := 1 + len(k) + protohelpers.SizeOfVarint(uint64(len(k))) + l + n += mapEntrySize + 1 + protohelpers.SizeOfVarint(uint64(mapEntrySize)) + } + } + if len(m.ReadAuth) > 0 { + for k, v := range m.ReadAuth { + _ = k + _ = v + mapEntrySize := 1 + len(k) + protohelpers.SizeOfVarint(uint64(len(k))) + 1 + len(v) + protohelpers.SizeOfVarint(uint64(len(v))) + n += mapEntrySize + 1 + protohelpers.SizeOfVarint(uint64(mapEntrySize)) + } + } + n += len(m.unknownFields) + return n +} + func (m *LookupDirectoryEntryRequest) UnmarshalVT(dAtA []byte) error { l := len(dAtA) iNdEx := 0 @@ -24867,3 +25161,626 @@ func (m *MountInfo) UnmarshalVT(dAtA []byte) error { } return nil } +func (m *LookupDirectoryEntriesRequest) UnmarshalVT(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: LookupDirectoryEntriesRequest: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: LookupDirectoryEntriesRequest: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Requests", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Requests = append(m.Requests, &LookupDirectoryEntryRequest{}) + if err := m.Requests[len(m.Requests)-1].UnmarshalVT(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + case 2: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field UnavailableVolumeIsMiss", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.UnavailableVolumeIsMiss = bool(v != 0) + default: + iNdEx = preIndex + skippy, err := protohelpers.Skip(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return protohelpers.ErrInvalidLength + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.unknownFields = append(m.unknownFields, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *LookupDirectoryEntryResult) UnmarshalVT(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: LookupDirectoryEntryResult: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: LookupDirectoryEntryResult: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Found", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.Found = bool(v != 0) + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Entry", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Entry == nil { + m.Entry = &Entry{} + } + if err := m.Entry.UnmarshalVT(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + case 3: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Error", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Error = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 4: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field LogTsNs", wireType) + } + m.LogTsNs = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.LogTsNs |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } + case 5: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field LogSignature", wireType) + } + m.LogSignature = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.LogSignature |= int32(b&0x7F) << shift + if b < 0x80 { + break + } + } + default: + iNdEx = preIndex + skippy, err := protohelpers.Skip(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return protohelpers.ErrInvalidLength + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.unknownFields = append(m.unknownFields, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *LookupDirectoryEntriesResponse) UnmarshalVT(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: LookupDirectoryEntriesResponse: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: LookupDirectoryEntriesResponse: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Results", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Results = append(m.Results, &LookupDirectoryEntryResult{}) + if err := m.Results[len(m.Results)-1].UnmarshalVT(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field LocationsMap", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.LocationsMap == nil { + m.LocationsMap = make(map[string]*Locations) + } + var mapkey string + var mapvalue *Locations + for iNdEx < postIndex { + entryPreIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + if fieldNum == 1 { + var stringLenmapkey uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLenmapkey |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLenmapkey := int(stringLenmapkey) + if intStringLenmapkey < 0 { + return protohelpers.ErrInvalidLength + } + postStringIndexmapkey := iNdEx + intStringLenmapkey + if postStringIndexmapkey < 0 { + return protohelpers.ErrInvalidLength + } + if postStringIndexmapkey > l { + return io.ErrUnexpectedEOF + } + mapkey = string(dAtA[iNdEx:postStringIndexmapkey]) + iNdEx = postStringIndexmapkey + } else if fieldNum == 2 { + var mapmsglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + mapmsglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if mapmsglen < 0 { + return protohelpers.ErrInvalidLength + } + postmsgIndex := iNdEx + mapmsglen + if postmsgIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postmsgIndex > l { + return io.ErrUnexpectedEOF + } + mapvalue = &Locations{} + if err := mapvalue.UnmarshalVT(dAtA[iNdEx:postmsgIndex]); err != nil { + return err + } + iNdEx = postmsgIndex + } else { + iNdEx = entryPreIndex + skippy, err := protohelpers.Skip(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return protohelpers.ErrInvalidLength + } + if (iNdEx + skippy) > postIndex { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + m.LocationsMap[mapkey] = mapvalue + iNdEx = postIndex + case 3: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ReadAuth", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.ReadAuth == nil { + m.ReadAuth = make(map[string]string) + } + var mapkey string + var mapvalue string + for iNdEx < postIndex { + entryPreIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + if fieldNum == 1 { + var stringLenmapkey uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLenmapkey |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLenmapkey := int(stringLenmapkey) + if intStringLenmapkey < 0 { + return protohelpers.ErrInvalidLength + } + postStringIndexmapkey := iNdEx + intStringLenmapkey + if postStringIndexmapkey < 0 { + return protohelpers.ErrInvalidLength + } + if postStringIndexmapkey > l { + return io.ErrUnexpectedEOF + } + mapkey = string(dAtA[iNdEx:postStringIndexmapkey]) + iNdEx = postStringIndexmapkey + } else if fieldNum == 2 { + var stringLenmapvalue uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLenmapvalue |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLenmapvalue := int(stringLenmapvalue) + if intStringLenmapvalue < 0 { + return protohelpers.ErrInvalidLength + } + postStringIndexmapvalue := iNdEx + intStringLenmapvalue + if postStringIndexmapvalue < 0 { + return protohelpers.ErrInvalidLength + } + if postStringIndexmapvalue > l { + return io.ErrUnexpectedEOF + } + mapvalue = string(dAtA[iNdEx:postStringIndexmapvalue]) + iNdEx = postStringIndexmapvalue + } else { + iNdEx = entryPreIndex + skippy, err := protohelpers.Skip(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return protohelpers.ErrInvalidLength + } + if (iNdEx + skippy) > postIndex { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + m.ReadAuth[mapkey] = mapvalue + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := protohelpers.Skip(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return protohelpers.ErrInvalidLength + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.unknownFields = append(m.unknownFields, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} diff --git a/weed/server/filer_grpc_server_batch_lookup.go b/weed/server/filer_grpc_server_batch_lookup.go new file mode 100644 index 000000000..959073970 --- /dev/null +++ b/weed/server/filer_grpc_server_batch_lookup.go @@ -0,0 +1,247 @@ +package weed_server + +import ( + "context" + "errors" + "fmt" + "sort" + "strconv" + "strings" + "sync" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/wdclient" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +const ( + maxFilerBatchLookupRequests = 4096 + maxFilerBatchLookupWorkers = 32 +) + +type filerBatchEntryFinder func(context.Context, util.FullPath) (*filer.Entry, int64, error) +type filerBatchVolumeLookup func(context.Context, []string) (map[string][]wdclient.Location, error) + +// LookupDirectoryEntries performs exact entry lookups concurrently while +// preserving request order, then resolves every referenced volume in one +// deduplicated master lookup. +func (fs *FilerServer) LookupDirectoryEntries(ctx context.Context, req *filer_pb.LookupDirectoryEntriesRequest) (*filer_pb.LookupDirectoryEntriesResponse, error) { + lookupVolumes := selectFilerBatchVolumeLookup(req, + fs.filer.MasterClient.LookupVolumeIdsWithFallback, + fs.filer.MasterClient.LookupVolumeIdsAuthoritative) + response, err := lookupDirectoryEntries(ctx, req, fs.filer.Signature, fs.fencedFindEntry, lookupVolumes) + if err != nil { + return nil, err + } + if fs.volumeGuard != nil { + populateBatchReadAuth(response, fs.maybeGetVolumeReadJwtAuthorizationToken) + } + return response, nil +} + +func selectFilerBatchVolumeLookup(req *filer_pb.LookupDirectoryEntriesRequest, cached, authoritative filerBatchVolumeLookup) filerBatchVolumeLookup { + if req != nil && req.GetUnavailableVolumeIsMiss() { + // DeletedVids is an asynchronous cache invalidation hint and its delivery + // is deliberately best-effort. Cache reads must therefore confirm the + // unique Volume IDs with Master before treating a location as live. + return authoritative + } + return cached +} + +func populateBatchReadAuth(response *filer_pb.LookupDirectoryEntriesResponse, mint func(string) string) { + response.ReadAuth = make(map[string]string) + for _, result := range response.Results { + if result == nil || result.Entry == nil { + continue + } + for _, chunk := range result.Entry.Chunks { + fid := chunk.GetFileIdString() + if fid == "" { + continue + } + if token := mint(fid); token != "" { + response.ReadAuth[fid] = token + } + } + } +} + +func lookupDirectoryEntries( + ctx context.Context, + req *filer_pb.LookupDirectoryEntriesRequest, + logSignature int32, + findEntry filerBatchEntryFinder, + lookupVolumes filerBatchVolumeLookup, +) (*filer_pb.LookupDirectoryEntriesResponse, error) { + if err := validateFilerBatchLookupRequest(req); err != nil { + return nil, err + } + + results := make([]*filer_pb.LookupDirectoryEntryResult, len(req.Requests)) + resultVolumeIDs := make([]map[string]struct{}, len(req.Requests)) + jobs := make(chan int) + workerCount := min(len(req.Requests), maxFilerBatchLookupWorkers) + + var workers sync.WaitGroup + workers.Add(workerCount) + for range workerCount { + go func() { + defer workers.Done() + for index := range jobs { + request := req.Requests[index] + entry, logTsNs, err := findEntry(ctx, util.JoinPath(request.Directory, request.Name)) + result := &filer_pb.LookupDirectoryEntryResult{ + LogTsNs: logTsNs, + LogSignature: logSignature, + } + switch { + case errors.Is(err, filer_pb.ErrNotFound): + results[index] = result + case err != nil: + result.Error = err.Error() + results[index] = result + case entry == nil: + result.Error = "entry lookup returned no entry" + results[index] = result + default: + pbEntry := entry.ToProtoEntry() + volumeIDs, volumeErr := filerBatchEntryVolumeIDs(pbEntry) + result.Found = true + result.Entry = pbEntry + if volumeErr != nil { + result.Error = volumeErr.Error() + } + results[index] = result + resultVolumeIDs[index] = volumeIDs + } + } + }() + } + + for index := range req.Requests { + select { + case jobs <- index: + case <-ctx.Done(): + close(jobs) + workers.Wait() + return nil, status.FromContextError(ctx.Err()).Err() + } + } + close(jobs) + workers.Wait() + if err := ctx.Err(); err != nil { + return nil, status.FromContextError(err).Err() + } + + allVolumeIDs := make(map[string]struct{}) + for _, volumeIDs := range resultVolumeIDs { + for volumeID := range volumeIDs { + allVolumeIDs[volumeID] = struct{}{} + } + } + volumeIDs := make([]string, 0, len(allVolumeIDs)) + for volumeID := range allVolumeIDs { + volumeIDs = append(volumeIDs, volumeID) + } + sort.Strings(volumeIDs) + + response := &filer_pb.LookupDirectoryEntriesResponse{ + Results: results, + LocationsMap: make(map[string]*filer_pb.Locations, len(volumeIDs)), + } + if len(volumeIDs) == 0 { + return response, nil + } + + locationsByVolume, lookupErr := lookupVolumes(ctx, volumeIDs) + // The provider returns a nil map when it got no answer, and a populated map + // with the volumes the master does not serve reported as errors when it did, + // so a nil map is the only sign of an unanswered lookup. Only a volume the + // master itself left out may become a miss; anything else stays an error. + missIsAuthoritative := req.UnavailableVolumeIsMiss && locationsByVolume != nil + for _, volumeID := range volumeIDs { + locations := locationsByVolume[volumeID] + if len(locations) != 0 { + response.LocationsMap[volumeID] = &filer_pb.Locations{ + Locations: wdclientLocationsToPb(locations), + } + continue + } + if !missIsAuthoritative { + response.LocationsMap[volumeID] = &filer_pb.Locations{} + } + for index, entryVolumeIDs := range resultVolumeIDs { + if _, affected := entryVolumeIDs[volumeID]; !affected { + continue + } + if missIsAuthoritative && results[index].Error == "" { + results[index].Found = false + results[index].Entry = nil + resultVolumeIDs[index] = nil + continue + } + message := fmt.Sprintf("volume %s has no locations", volumeID) + if lookupErr != nil { + message += ": " + lookupErr.Error() + } + results[index].Error = appendFilerBatchLookupError(results[index].Error, message) + } + } + + return response, nil +} + +func validateFilerBatchLookupRequest(req *filer_pb.LookupDirectoryEntriesRequest) error { + if req == nil || len(req.Requests) == 0 { + return status.Error(codes.InvalidArgument, "batch lookup requires at least one request") + } + if len(req.Requests) > maxFilerBatchLookupRequests { + return status.Errorf(codes.ResourceExhausted, "batch lookup has %d requests; maximum is %d", len(req.Requests), maxFilerBatchLookupRequests) + } + for index, request := range req.Requests { + if request == nil { + return status.Errorf(codes.InvalidArgument, "batch lookup request %d is missing", index) + } + if request.Directory == "" || !strings.HasPrefix(request.Directory, "/") || strings.ContainsRune(request.Directory, '\x00') { + return status.Errorf(codes.InvalidArgument, "batch lookup request %d has invalid directory", index) + } + if request.Name == "" || request.Name == "." || request.Name == ".." || strings.ContainsAny(request.Name, "/\x00") { + return status.Errorf(codes.InvalidArgument, "batch lookup request %d has invalid name", index) + } + } + return nil +} + +func filerBatchEntryVolumeIDs(entry *filer_pb.Entry) (map[string]struct{}, error) { + volumeIDs := make(map[string]struct{}) + for index, chunk := range entry.GetChunks() { + if chunk == nil { + return volumeIDs, fmt.Errorf("entry chunk %d is missing", index) + } + if chunk.Fid != nil { + volumeIDs[strconv.FormatUint(uint64(chunk.Fid.VolumeId), 10)] = struct{}{} + continue + } + if chunk.FileId == "" { + return volumeIDs, fmt.Errorf("entry chunk %d has no file id", index) + } + fid, err := needle.ParseFileIdFromString(chunk.FileId) + if err != nil { + return volumeIDs, fmt.Errorf("entry chunk %d has invalid file id: %w", index, err) + } + volumeIDs[strconv.FormatUint(uint64(fid.VolumeId), 10)] = struct{}{} + } + return volumeIDs, nil +} + +func appendFilerBatchLookupError(existing, next string) string { + if existing == "" { + return next + } + return existing + "; " + next +} diff --git a/weed/server/filer_grpc_server_batch_lookup_test.go b/weed/server/filer_grpc_server_batch_lookup_test.go new file mode 100644 index 000000000..7bc099dfb --- /dev/null +++ b/weed/server/filer_grpc_server_batch_lookup_test.go @@ -0,0 +1,418 @@ +package weed_server + +import ( + "context" + "errors" + "fmt" + "reflect" + "sort" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/wdclient" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func TestLookupDirectoryEntriesRejectsInvalidRequests(t *testing.T) { + valid := &filer_pb.LookupDirectoryEntryRequest{Directory: "/batch", Name: "key"} + tests := []struct { + name string + req *filer_pb.LookupDirectoryEntriesRequest + wantCode codes.Code + }{ + {name: "nil request", wantCode: codes.InvalidArgument}, + {name: "empty batch", req: &filer_pb.LookupDirectoryEntriesRequest{}, wantCode: codes.InvalidArgument}, + {name: "oversized", req: repeatedBatchLookupRequest(maxFilerBatchLookupRequests+1, valid), wantCode: codes.ResourceExhausted}, + {name: "nil item", req: &filer_pb.LookupDirectoryEntriesRequest{Requests: []*filer_pb.LookupDirectoryEntryRequest{nil}}, wantCode: codes.InvalidArgument}, + {name: "relative directory", req: &filer_pb.LookupDirectoryEntriesRequest{Requests: []*filer_pb.LookupDirectoryEntryRequest{{Directory: "batch", Name: "key"}}}, wantCode: codes.InvalidArgument}, + {name: "empty name", req: &filer_pb.LookupDirectoryEntriesRequest{Requests: []*filer_pb.LookupDirectoryEntryRequest{{Directory: "/batch"}}}, wantCode: codes.InvalidArgument}, + {name: "nested name", req: &filer_pb.LookupDirectoryEntriesRequest{Requests: []*filer_pb.LookupDirectoryEntryRequest{{Directory: "/batch", Name: "a/b"}}}, wantCode: codes.InvalidArgument}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + called := false + _, err := lookupDirectoryEntries(context.Background(), tt.req, 17, + func(context.Context, util.FullPath) (*filer.Entry, int64, error) { + called = true + return nil, 0, nil + }, + func(context.Context, []string) (map[string][]wdclient.Location, error) { + called = true + return nil, nil + }, + ) + if status.Code(err) != tt.wantCode { + t.Fatalf("error code = %v, want %v: %v", status.Code(err), tt.wantCode, err) + } + if called { + t.Fatal("invalid request reached lookup implementation") + } + }) + } +} + +func TestLookupDirectoryEntriesAcceptsMaximumBatch(t *testing.T) { + req := repeatedBatchLookupRequest(maxFilerBatchLookupRequests, &filer_pb.LookupDirectoryEntryRequest{ + Directory: "/batch", + Name: "key", + }) + response, err := lookupDirectoryEntries(context.Background(), req, 13, + func(context.Context, util.FullPath) (*filer.Entry, int64, error) { + return nil, 99, filer_pb.ErrNotFound + }, + func(context.Context, []string) (map[string][]wdclient.Location, error) { + t.Fatal("volume lookup called for an all-miss batch") + return nil, nil + }, + ) + if err != nil { + t.Fatalf("maximum-sized batch failed: %v", err) + } + if got := len(response.Results); got != maxFilerBatchLookupRequests { + t.Fatalf("result count = %d, want %d", got, maxFilerBatchLookupRequests) + } + for _, index := range []int{0, maxFilerBatchLookupRequests - 1} { + if got := response.Results[index]; got.LogTsNs != 99 || got.LogSignature != 13 { + t.Fatalf("miss fence[%d] = (%d,%d), want (99,13)", index, got.LogTsNs, got.LogSignature) + } + } +} + +func TestLookupDirectoryEntriesUsesFilerSignatureForFoundAndMiss(t *testing.T) { + store := newRenameTestStore() + if err := store.InsertEntry(context.Background(), newFileEntry("/batch/found", 1)); err != nil { + t.Fatalf("seed entry: %v", err) + } + testFiler := newRenameTestFiler(t, store) + testFiler.Signature = 73 + server := &FilerServer{ + filer: testFiler, + entryLockTable: util.NewLockTable[util.FullPath](), + } + + response, err := server.LookupDirectoryEntries(context.Background(), &filer_pb.LookupDirectoryEntriesRequest{ + Requests: []*filer_pb.LookupDirectoryEntryRequest{ + {Directory: "/batch", Name: "found"}, + {Directory: "/batch", Name: "missing"}, + }, + }) + if err != nil { + t.Fatalf("batch lookup: %v", err) + } + if !response.Results[0].Found || response.Results[1].Found { + t.Fatalf("unexpected found states: %v, %v", response.Results[0].Found, response.Results[1].Found) + } + for index, result := range response.Results { + if result.LogSignature != 73 || result.LogTsNs == 0 { + t.Fatalf("result[%d] fence = (%d,%d), want nonzero timestamp and signature 73", index, result.LogTsNs, result.LogSignature) + } + } +} + +func TestPopulateBatchReadAuthMintsExactFidCapabilities(t *testing.T) { + response := &filer_pb.LookupDirectoryEntriesResponse{Results: []*filer_pb.LookupDirectoryEntryResult{ + {Found: true, Entry: &filer_pb.Entry{Chunks: []*filer_pb.FileChunk{ + {FileId: "7,abc"}, {FileId: "8,def"}, {FileId: "7,abc"}, + }}}, + {Found: false}, + }} + populateBatchReadAuth(response, func(fid string) string { return "jwt:" + fid }) + if !reflect.DeepEqual(response.ReadAuth, map[string]string{ + "7,abc": "jwt:7,abc", + "8,def": "jwt:8,def", + }) { + t.Fatalf("read auth = %#v", response.ReadAuth) + } +} + +func TestPopulateBatchReadAuthDoesNotInventCapabilitiesWhenSigningIsDisabled(t *testing.T) { + response := &filer_pb.LookupDirectoryEntriesResponse{Results: []*filer_pb.LookupDirectoryEntryResult{ + {Found: true, Entry: &filer_pb.Entry{Chunks: []*filer_pb.FileChunk{{FileId: "7,abc"}}}}, + }} + populateBatchReadAuth(response, func(string) string { return "" }) + if len(response.ReadAuth) != 0 { + t.Fatalf("read auth must be empty without a signing key: %#v", response.ReadAuth) + } +} + +func TestLookupDirectoryEntriesPreservesOrderBoundsParallelismAndDeduplicatesVolumes(t *testing.T) { + const requestCount = 96 + requests := make([]*filer_pb.LookupDirectoryEntryRequest, requestCount) + for index := range requests { + requests[index] = &filer_pb.LookupDirectoryEntryRequest{ + Directory: "/batch", + Name: fmt.Sprintf("item-%03d", index), + } + } + + var active atomic.Int32 + var activeMax atomic.Int32 + findEntry := func(_ context.Context, path util.FullPath) (*filer.Entry, int64, error) { + index, err := strconv.Atoi(strings.TrimPrefix(string(path), "/batch/item-")) + if err != nil { + return nil, 0, err + } + current := active.Add(1) + defer active.Add(-1) + for { + previous := activeMax.Load() + if current <= previous || activeMax.CompareAndSwap(previous, current) { + break + } + } + // Reverse completion order within each wave so append-based implementations fail. + time.Sleep(time.Duration(requestCount-index) * 50 * time.Microsecond) + if index == 7 { + return nil, int64(1000 + index), errors.New("lookup failed") + } + if index == 11 { + return nil, int64(1000 + index), filer_pb.ErrNotFound + } + volumeID := uint32(index%3 + 1) + chunk := &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: volumeID, FileKey: uint64(index + 1), Cookie: 17}} + if index == 5 { + chunk = &filer_pb.FileChunk{FileId: "2,0294cbb9892b"} + } + return &filer.Entry{FullPath: path, Chunks: []*filer_pb.FileChunk{chunk, chunk}}, int64(1000 + index), nil + } + + volumeLookupCalls := 0 + var lookedUpVolumeIDs []string + lookupVolumes := func(_ context.Context, volumeIDs []string) (map[string][]wdclient.Location, error) { + volumeLookupCalls++ + lookedUpVolumeIDs = append([]string(nil), volumeIDs...) + return map[string][]wdclient.Location{ + "1": {{Url: "volume-1:8080"}}, + "2": {{Url: "volume-2:8080"}}, + "3": {{Url: "volume-3:8080"}}, + }, nil + } + + const logSignature = 0x1234 + response, err := lookupDirectoryEntries(context.Background(), &filer_pb.LookupDirectoryEntriesRequest{Requests: requests}, logSignature, findEntry, lookupVolumes) + if err != nil { + t.Fatalf("batch lookup: %v", err) + } + if got := activeMax.Load(); got <= 1 || got > maxFilerBatchLookupWorkers { + t.Fatalf("parallel workers peaked at %d, want 2..%d", got, maxFilerBatchLookupWorkers) + } + if volumeLookupCalls != 1 { + t.Fatalf("volume lookup calls = %d, want 1", volumeLookupCalls) + } + if want := []string{"1", "2", "3"}; !reflect.DeepEqual(lookedUpVolumeIDs, want) { + t.Fatalf("volume IDs = %v, want %v", lookedUpVolumeIDs, want) + } + if got := len(response.LocationsMap); got != 3 { + t.Fatalf("location map size = %d, want 3", got) + } + + for index, result := range response.Results { + if result.LogTsNs != int64(1000+index) || result.LogSignature != logSignature { + t.Fatalf("result[%d] fence = (%d,%d), want (%d,%d)", index, result.LogTsNs, result.LogSignature, 1000+index, logSignature) + } + switch index { + case 7: + if result.Found || result.Entry != nil || result.Error != "lookup failed" { + t.Fatalf("error result[%d] = %+v", index, result) + } + case 11: + if result.Found || result.Entry != nil || result.Error != "" { + t.Fatalf("not-found result[%d] = %+v", index, result) + } + default: + if !result.Found || result.Entry == nil || result.Error != "" { + t.Fatalf("found result[%d] = %+v", index, result) + } + if want := fmt.Sprintf("item-%03d", index); result.Entry.Name != want { + t.Fatalf("result[%d] name = %q, want %q", index, result.Entry.Name, want) + } + } + } +} + +func TestLookupDirectoryEntriesMarksEntriesWithUnavailableVolumes(t *testing.T) { + response, err := lookupDirectoryEntries(context.Background(), &filer_pb.LookupDirectoryEntriesRequest{ + Requests: []*filer_pb.LookupDirectoryEntryRequest{{Directory: "/batch", Name: "key"}}, + }, 42, func(_ context.Context, path util.FullPath) (*filer.Entry, int64, error) { + return &filer.Entry{ + FullPath: path, + Chunks: []*filer_pb.FileChunk{{ + Fid: &filer_pb.FileId{VolumeId: 9, FileKey: 1, Cookie: 2}, + }}, + }, 123, nil + }, func(_ context.Context, volumeIDs []string) (map[string][]wdclient.Location, error) { + if !sort.StringsAreSorted(volumeIDs) { + t.Fatalf("volume IDs are not sorted: %v", volumeIDs) + } + return nil, errors.New("master unavailable") + }) + if err != nil { + t.Fatalf("batch lookup: %v", err) + } + result := response.Results[0] + if result.LogTsNs != 123 || result.LogSignature != 42 { + t.Fatalf("result fence = (%d,%d), want (123,42)", result.LogTsNs, result.LogSignature) + } + if !result.Found || result.Entry == nil { + t.Fatalf("metadata result lost when volume lookup failed: %+v", result) + } + if !strings.Contains(result.Error, "volume 9 has no locations") || !strings.Contains(result.Error, "master unavailable") { + t.Fatalf("result error = %q", result.Error) + } + locations, ok := response.LocationsMap["9"] + if !ok || locations == nil || len(locations.Locations) != 0 { + t.Fatalf("missing volume map entry = %#v, present=%v", locations, ok) + } +} + +func TestLookupDirectoryEntriesKeepsMalformedEntryError(t *testing.T) { + response, err := lookupDirectoryEntries(context.Background(), &filer_pb.LookupDirectoryEntriesRequest{ + Requests: []*filer_pb.LookupDirectoryEntryRequest{{Directory: "/batch", Name: "key"}}, + }, 42, func(_ context.Context, path util.FullPath) (*filer.Entry, int64, error) { + return &filer.Entry{ + FullPath: path, + Chunks: []*filer_pb.FileChunk{ + {Fid: &filer_pb.FileId{VolumeId: 9, FileKey: 1, Cookie: 2}}, + {FileId: "not-a-file-id"}, + }, + }, 123, nil + }, func(context.Context, []string) (map[string][]wdclient.Location, error) { + return map[string][]wdclient.Location{}, nil + }) + if err != nil { + t.Fatalf("batch lookup: %v", err) + } + result := response.Results[0] + if !result.Found || result.Entry == nil || !strings.Contains(result.Error, "invalid file id") { + t.Fatalf("malformed entry was reported as a clean miss: %+v", result) + } +} + +func cacheVolumeEntry(_ context.Context, path util.FullPath) (*filer.Entry, int64, error) { + return &filer.Entry{ + FullPath: path, + Chunks: []*filer_pb.FileChunk{{ + Fid: &filer_pb.FileId{VolumeId: 9, FileKey: 1, Cookie: 2}, + }}, + }, 123, nil +} + +func TestLookupDirectoryEntriesCanTreatRetiredCacheVolumeAsMiss(t *testing.T) { + response, err := lookupDirectoryEntries(context.Background(), &filer_pb.LookupDirectoryEntriesRequest{ + Requests: []*filer_pb.LookupDirectoryEntryRequest{{Directory: "/batch", Name: "key"}}, + UnavailableVolumeIsMiss: true, + }, 42, cacheVolumeEntry, func(_ context.Context, volumeIDs []string) (map[string][]wdclient.Location, error) { + // The master answered and left volume 9 out. + return map[string][]wdclient.Location{}, errors.New("volume 9: volume id 9 not found") + }) + if err != nil { + t.Fatalf("batch lookup: %v", err) + } + result := response.Results[0] + if result.Found || result.Entry != nil || result.Error != "" { + t.Fatalf("retired cache volume did not become a clean miss: %+v", result) + } + if result.LogTsNs != 123 || result.LogSignature != 42 { + t.Fatalf("miss fence = (%d,%d), want (123,42)", result.LogTsNs, result.LogSignature) + } + if _, exists := response.LocationsMap["9"]; exists { + t.Fatalf("cache miss returned an unusable location: %#v", response.LocationsMap["9"]) + } +} + +func TestLookupDirectoryEntriesKeepsCacheEntryWhenMasterIsUnreachable(t *testing.T) { + response, err := lookupDirectoryEntries(context.Background(), &filer_pb.LookupDirectoryEntriesRequest{ + Requests: []*filer_pb.LookupDirectoryEntryRequest{{Directory: "/batch", Name: "key"}}, + UnavailableVolumeIsMiss: true, + }, 42, cacheVolumeEntry, func(context.Context, []string) (map[string][]wdclient.Location, error) { + return nil, errors.New("master unavailable") + }) + if err != nil { + t.Fatalf("batch lookup: %v", err) + } + result := response.Results[0] + if !result.Found || result.Entry == nil { + t.Fatalf("unanswered lookup was reported as a miss: %+v", result) + } + if !strings.Contains(result.Error, "volume 9 has no locations") || !strings.Contains(result.Error, "master unavailable") { + t.Fatalf("result error = %q", result.Error) + } + locations, ok := response.LocationsMap["9"] + if !ok || locations == nil || len(locations.Locations) != 0 { + t.Fatalf("unresolved volume map entry = %#v, present=%v", locations, ok) + } +} + +func TestLookupDirectoryEntriesKeepsMalformedEntryErrorInCacheMode(t *testing.T) { + response, err := lookupDirectoryEntries(context.Background(), &filer_pb.LookupDirectoryEntriesRequest{ + Requests: []*filer_pb.LookupDirectoryEntryRequest{{Directory: "/batch", Name: "key"}}, + UnavailableVolumeIsMiss: true, + }, 42, func(_ context.Context, path util.FullPath) (*filer.Entry, int64, error) { + return &filer.Entry{ + FullPath: path, + Chunks: []*filer_pb.FileChunk{ + {Fid: &filer_pb.FileId{VolumeId: 9, FileKey: 1, Cookie: 2}}, + {FileId: "not-a-file-id"}, + }, + }, 123, nil + }, func(context.Context, []string) (map[string][]wdclient.Location, error) { + return map[string][]wdclient.Location{}, nil + }) + if err != nil { + t.Fatalf("batch lookup: %v", err) + } + result := response.Results[0] + if !result.Found || result.Entry == nil || !strings.Contains(result.Error, "invalid file id") { + t.Fatalf("malformed entry was reported as a clean miss: %+v", result) + } +} + +func TestCacheBatchLookupSelectsAuthoritativeVolumeLocations(t *testing.T) { + cachedCalls, authoritativeCalls := 0, 0 + cached := func(context.Context, []string) (map[string][]wdclient.Location, error) { + cachedCalls++ + return map[string][]wdclient.Location{"9": {{Url: "retired-volume:8080"}}}, nil + } + authoritative := func(context.Context, []string) (map[string][]wdclient.Location, error) { + authoritativeCalls++ + return nil, nil + } + + lookup := selectFilerBatchVolumeLookup(&filer_pb.LookupDirectoryEntriesRequest{ + UnavailableVolumeIsMiss: true, + }, cached, authoritative) + locations, err := lookup(context.Background(), []string{"9"}) + if err != nil { + t.Fatal(err) + } + if cachedCalls != 0 || authoritativeCalls != 1 || len(locations) != 0 { + t.Fatalf("cache lookup used stale path: cached=%d authoritative=%d locations=%v", cachedCalls, authoritativeCalls, locations) + } + + lookup = selectFilerBatchVolumeLookup(&filer_pb.LookupDirectoryEntriesRequest{}, cached, authoritative) + if _, err := lookup(context.Background(), []string{"9"}); err != nil { + t.Fatal(err) + } + if cachedCalls != 1 || authoritativeCalls != 1 { + t.Fatalf("ordinary lookup did not keep cached path: cached=%d authoritative=%d", cachedCalls, authoritativeCalls) + } +} + +func repeatedBatchLookupRequest(count int, request *filer_pb.LookupDirectoryEntryRequest) *filer_pb.LookupDirectoryEntriesRequest { + requests := make([]*filer_pb.LookupDirectoryEntryRequest, count) + for index := range requests { + requests[index] = &filer_pb.LookupDirectoryEntryRequest{ + Directory: request.Directory, + Name: request.Name, + } + } + return &filer_pb.LookupDirectoryEntriesRequest{Requests: requests} +} diff --git a/weed/server/filer_server_storage_rule_ttl_test.go b/weed/server/filer_server_storage_rule_ttl_test.go index 33639f156..1d1942d85 100644 --- a/weed/server/filer_server_storage_rule_ttl_test.go +++ b/weed/server/filer_server_storage_rule_ttl_test.go @@ -27,6 +27,52 @@ func addTtlRule(t *testing.T, f *filer.Filer) { } } +// The native LMCache path allocates chunks and publishes the Filer Entry in +// separate RPCs. Both RPCs must resolve the same storage rule, or metadata can +// expire while its Volume remains permanent (or the reverse). +func TestNativeCacheWriteUsesOneTtlForAllocationAndEntry(t *testing.T) { + store := newRenameTestStore() + store.entries[ttlRulePrefix] = newDirectoryEntry(ttlRulePrefix, 10) + server := &FilerServer{ + filer: newRenameTestFiler(t, store), + option: &FilerOption{}, + entryLockTable: util.NewLockTable[util.FullPath](), + } + addTtlRule(t, server.filer) + + allocation, err := server.resolveAssignStorageOption(context.Background(), &filer_pb.AssignVolumeRequest{ + Path: ttlRulePrefix + "cache-key", + Collection: "lmcache", + }) + if err != nil { + t.Fatalf("resolveAssignStorageOption: %v", err) + } + if allocation.Collection != "lmcache" { + t.Fatalf("allocation collection = %q, want lmcache", allocation.Collection) + } + if allocation.TtlSeconds != 180 { + t.Fatalf("allocation TTL = %d, want 180", allocation.TtlSeconds) + } + + if _, err := server.CreateEntry(context.Background(), &filer_pb.CreateEntryRequest{ + Directory: "/buckets/ttl", + Entry: &filer_pb.Entry{ + Name: "cache-key", + Attributes: &filer_pb.FuseAttributes{FileMode: 0644}, + }, + }); err != nil { + t.Fatalf("CreateEntry: %v", err) + } + + entry, err := store.FindEntry(context.Background(), ttlRulePrefix+"cache-key") + if err != nil { + t.Fatalf("FindEntry: %v", err) + } + if entry.TtlSec != allocation.TtlSeconds { + t.Fatalf("entry TTL = %d, allocation TTL = %d", entry.TtlSec, allocation.TtlSeconds) + } +} + // An object written through ObjectTransaction (the routed S3 write path) must // pick up the path's TTL rule, the same as one written through CreateEntry. func TestObjectTransactionPutAppliesRuleTtl(t *testing.T) { diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index 386b4dae4..07fda3b0f 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/weed/server/master_grpc_server_volume.go @@ -195,9 +195,12 @@ func (ms *MasterServer) LookupVolume(ctx context.Context, req *master_pb.LookupV } } - // Only return Unavailable during warmup when every requested ID was a transient not-found - if len(req.VolumeOrFileIds) > 0 && notFoundCount == len(req.VolumeOrFileIds) && ms.Topo.IsLeader() && ms.Topo.IsWarmingUp() { - glog.V(0).Infof("lookup volume warming up: topology is still loading (%d not found)", notFoundCount) + // While warming up, a not-found may only mean the volume server has not + // reported yet, so no part of the answer can be treated as authoritative. + // Callers retry Unavailable; a partial answer would let them take a + // missing volume as gone. + if notFoundCount > 0 && ms.Topo.IsLeader() && ms.Topo.IsWarmingUp() { + glog.V(0).Infof("lookup volume warming up: topology is still loading (%d of %d not found)", notFoundCount, len(req.VolumeOrFileIds)) return nil, status.Errorf(codes.Unavailable, "master is warming up, topology is still loading") } diff --git a/weed/server/master_grpc_server_volume_warmup_test.go b/weed/server/master_grpc_server_volume_warmup_test.go new file mode 100644 index 000000000..b62d4a32d --- /dev/null +++ b/weed/server/master_grpc_server_volume_warmup_test.go @@ -0,0 +1,63 @@ +package weed_server + +import ( + "context" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// A leader that already knew volumes at the leader change is warming up until +// every volume server has had time to report again. +func newWarmingUpMaster(t *testing.T) *MasterServer { + t.Helper() + ms := newLeaderMaster() + node := ms.Topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, 18080, "127.0.0.1", "node1", map[string]uint32{"": 10}) + ms.Topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{{Id: 7, Size: 100}}, node) + ms.Topo.SetLastLeaderChangeTime(time.Now()) + if !ms.Topo.IsWarmingUp() { + t.Fatal("precondition: master is not warming up") + } + return ms +} + +func TestLookupVolumeDuringWarmupRefusesPartialNotFound(t *testing.T) { + ms := newWarmingUpMaster(t) + _, err := ms.LookupVolume(context.Background(), &master_pb.LookupVolumeRequest{VolumeOrFileIds: []string{"7", "8"}}) + if status.Code(err) != codes.Unavailable { + t.Fatalf("partial not-found during warmup = %v, want Unavailable", err) + } +} + +func TestLookupVolumeDuringWarmupAnswersFullyKnownBatch(t *testing.T) { + ms := newWarmingUpMaster(t) + resp, err := ms.LookupVolume(context.Background(), &master_pb.LookupVolumeRequest{VolumeOrFileIds: []string{"7"}}) + if err != nil { + t.Fatalf("known volume during warmup: %v", err) + } + if len(resp.VolumeIdLocations) != 1 || len(resp.VolumeIdLocations[0].Locations) != 1 || resp.VolumeIdLocations[0].Error != "" { + t.Fatalf("known volume answer = %+v", resp.VolumeIdLocations) + } +} + +func TestLookupVolumeAfterWarmupReportsNotFoundPerVolume(t *testing.T) { + ms := newWarmingUpMaster(t) + ms.Topo.SetLastLeaderChangeTime(time.Now().Add(-time.Hour)) + resp, err := ms.LookupVolume(context.Background(), &master_pb.LookupVolumeRequest{VolumeOrFileIds: []string{"7", "8"}}) + if err != nil { + t.Fatalf("lookup after warmup: %v", err) + } + if len(resp.VolumeIdLocations) != 2 { + t.Fatalf("answers = %+v, want two", resp.VolumeIdLocations) + } + if len(resp.VolumeIdLocations[0].Locations) != 1 || resp.VolumeIdLocations[0].Error != "" { + t.Fatalf("known volume answer = %+v", resp.VolumeIdLocations[0]) + } + if len(resp.VolumeIdLocations[1].Locations) != 0 || resp.VolumeIdLocations[1].Error == "" { + t.Fatalf("absent volume answer = %+v", resp.VolumeIdLocations[1]) + } +} diff --git a/weed/server/volume_grpc_admin.go b/weed/server/volume_grpc_admin.go index f4d330508..2ee9a4f9a 100644 --- a/weed/server/volume_grpc_admin.go +++ b/weed/server/volume_grpc_admin.go @@ -2,6 +2,7 @@ package weed_server import ( "context" + "errors" "fmt" "net" "path/filepath" @@ -200,6 +201,7 @@ func (vs *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb. if err != nil { glog.Errorf("volume delete %v: %v", req, err) + return resp, volumeDeleteStatusError(err) } else { // V(0) so destructive RPCs are always traceable. glog.Infof("volume delete %v", req) @@ -209,6 +211,19 @@ func (vs *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb. } +// volumeDeleteStatusError keeps the store's message so callers matching on +// "not found" or "volume not empty" keep working, and adds the status code so +// new callers do not have to. +func volumeDeleteStatusError(err error) error { + if errors.Is(err, storage.ErrVolumeNotFound) { + return status.Error(codes.NotFound, err.Error()) + } + if errors.Is(err, storage.ErrVolumeNotEmpty) { + return status.Error(codes.FailedPrecondition, err.Error()) + } + return err +} + func (vs *VolumeServer) VolumeConfigure(ctx context.Context, req *volume_server_pb.VolumeConfigureRequest) (*volume_server_pb.VolumeConfigureResponse, error) { resp := &volume_server_pb.VolumeConfigureResponse{} diff --git a/weed/server/volume_grpc_admin_delete_test.go b/weed/server/volume_grpc_admin_delete_test.go new file mode 100644 index 000000000..721340f13 --- /dev/null +++ b/weed/server/volume_grpc_admin_delete_test.go @@ -0,0 +1,37 @@ +package weed_server + +import ( + "context" + "errors" + "fmt" + "testing" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestVolumeDeleteStatusErrorDistinguishesAbsentFromTransportFailure(t *testing.T) { + notFound := volumeDeleteStatusError(fmt.Errorf("delete volume 17 not found on disk: %w", storage.ErrVolumeNotFound)) + assert.Equal(t, codes.NotFound, status.Code(notFound)) + assert.Contains(t, notFound.Error(), "not found", "the store message must survive for callers that match on it") + + transport := errors.New("connection reset") + require.ErrorIs(t, volumeDeleteStatusError(transport), transport) + assert.NotEqual(t, codes.NotFound, status.Code(transport)) + + notEmpty := volumeDeleteStatusError(storage.ErrVolumeNotEmpty) + assert.Equal(t, codes.FailedPrecondition, status.Code(notEmpty)) + assert.Contains(t, notEmpty.Error(), "volume not empty") +} + +func TestVolumeDeleteMapsAbsentStoreVolumeToNotFound(t *testing.T) { + vs := &VolumeServer{store: newTraversalTestStore(t.TempDir())} + + _, err := vs.VolumeDelete(context.Background(), &volume_server_pb.VolumeDeleteRequest{VolumeId: 17}) + assert.Equal(t, codes.NotFound, status.Code(err), err) +} diff --git a/weed/storage/store.go b/weed/storage/store.go index 24a09cf11..cb46a7d2a 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -1018,7 +1018,7 @@ func (s *Store) DeleteVolume(i needle.VolumeId, onlyEmpty bool, keepRemoteData b } else if err == ErrVolumeNotEmpty { // onlyEmpty: a non-empty copy aborts the delete rather than leaving a // partial result across disks. - return fmt.Errorf("DeleteVolume %d: %v", i, err) + return fmt.Errorf("DeleteVolume %d: %w", i, err) } else { // A real failure on one disk must not be masked by another copy's // success: a stale copy left on the failing disk would re-register. @@ -1030,7 +1030,7 @@ func (s *Store) DeleteVolume(i needle.VolumeId, onlyEmpty bool, keepRemoteData b return fmt.Errorf("DeleteVolume %d failed on some disks: %w", i, errors.Join(errs...)) } if !deletedAny { - return fmt.Errorf("delete volume %d not found on disk", i) + return fmt.Errorf("delete volume %d not found on disk: %w", i, ErrVolumeNotFound) } return nil } diff --git a/weed/storage/store_delete_volume_test.go b/weed/storage/store_delete_volume_test.go new file mode 100644 index 000000000..634c4b6e7 --- /dev/null +++ b/weed/storage/store_delete_volume_test.go @@ -0,0 +1,35 @@ +package storage + +import ( + "errors" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" +) + +func TestDeleteVolumeErrorsAreInspectable(t *testing.T) { + store := newTestStore(t, 1) + mountCollectionVolume(t, store.Locations[0], 5, "") + n := &needle.Needle{Id: types.Uint64ToNeedleId(1), Data: []byte("keep")} + if _, err := store.WriteVolumeNeedle(5, n, false, false); err != nil { + t.Fatal(err) + } + + err := store.DeleteVolume(5, true, false) + if !errors.Is(err, ErrVolumeNotEmpty) { + t.Fatalf("only-empty delete of a non-empty volume = %v, want ErrVolumeNotEmpty", err) + } + if _, found := store.Locations[0].FindVolume(5); !found { + t.Fatal("refused delete removed the volume") + } + + err = store.DeleteVolume(99, false, false) + if !errors.Is(err, ErrVolumeNotFound) { + t.Fatalf("delete of an absent volume = %v, want ErrVolumeNotFound", err) + } + + if err := store.DeleteVolume(5, false, false); err != nil { + t.Fatalf("forced delete: %v", err) + } +} diff --git a/weed/wdclient/vidmap_authoritative_test.go b/weed/wdclient/vidmap_authoritative_test.go new file mode 100644 index 000000000..499865b32 --- /dev/null +++ b/weed/wdclient/vidmap_authoritative_test.go @@ -0,0 +1,49 @@ +package wdclient + +import ( + "context" + "reflect" + "testing" +) + +type authoritativeLookupProvider struct { + calls int + result map[string][]Location + err error +} + +func (p *authoritativeLookupProvider) LookupVolumeIds(_ context.Context, volumeIDs []string) (map[string][]Location, error) { + p.calls++ + return p.result, p.err +} + +func TestAuthoritativeVolumeLookupDoesNotTrustCachedLocations(t *testing.T) { + provider := &authoritativeLookupProvider{result: map[string][]Location{}} + client := newVidMapClient(provider, "", DefaultVidMapCacheSize) + client.addLocation(17, Location{Url: "retired-volume:8080"}) + + locations, err := client.LookupVolumeIdsAuthoritative(context.Background(), []string{"17"}) + if err != nil { + t.Fatalf("authoritative lookup: %v", err) + } + if provider.calls != 1 { + t.Fatalf("provider calls = %d, want 1", provider.calls) + } + if len(locations) != 0 { + t.Fatalf("authoritative lookup returned stale cache: %#v", locations) + } +} + +func TestAuthoritativeVolumeLookupReturnsProviderResult(t *testing.T) { + want := map[string][]Location{"17": {{Url: "current-volume:8080"}}} + provider := &authoritativeLookupProvider{result: want} + client := newVidMapClient(provider, "", DefaultVidMapCacheSize) + + got, err := client.LookupVolumeIdsAuthoritative(context.Background(), []string{"17"}) + if err != nil { + t.Fatalf("authoritative lookup: %v", err) + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("locations = %#v, want %#v", got, want) + } +} diff --git a/weed/wdclient/vidmap_client.go b/weed/wdclient/vidmap_client.go index 631adf5d4..8ca713edc 100644 --- a/weed/wdclient/vidmap_client.go +++ b/weed/wdclient/vidmap_client.go @@ -247,6 +247,15 @@ func (vc *vidMapClient) LookupVolumeIdsWithFallback(ctx context.Context, volumeI return result, errors.Join(lookupErrors...) } +// LookupVolumeIdsAuthoritative bypasses the asynchronously maintained vid map. +// Use it when a stale positive result is unsafe, such as before serving cache +// payload from a Volume that the master may have retired. The provider still +// batches all requested IDs into one lookup, returns a nil map when it got no +// answer, and reports the volumes the master does not serve as errors. +func (vc *vidMapClient) LookupVolumeIdsAuthoritative(ctx context.Context, volumeIds []string) (map[string][]Location, error) { + return vc.provider.LookupVolumeIds(ctx, volumeIds) +} + // Public methods for external access // // The vidMap itself is never replaced, so these all read the one map under its