Github-Mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-08 15:41:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

filer: batch exact lookup RPC, authoritative volume lookup, VolumeDelete status codes (#11122)

Browse Source 
* 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
This commit is contained in:
Chris Lu
2026-09-03 14:52:05 -07:00
committed by GitHub
parent eb9f54cebb
commit 8112f2733a
18 changed files with 2378 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/filer/filer_prefix_object_test.go
+29
View File
@@ -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")
}
weed/pb/filer.proto
+40
View File
@@ -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<string, Locations> 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<string, string> read_auth = 3;
}
weed/pb/filer_pb/filer.pb.go
+342 -103
View File
@@ -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,
},
weed/pb/filer_pb/filer_grpc.pb.go
+39 -1
View File
@@ -1,7 +1,7 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.6.2
// - protoc v7.35.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,
weed/pb/filer_pb/filer_vtproto.pb.go
+917
View File
@@ -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
}
weed/server/filer_grpc_server_batch_lookup.go
+247
View File
@@ -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
}
weed/server/filer_grpc_server_batch_lookup_test.go
+418
View File
@@ -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}
}
weed/server/filer_server_storage_rule_ttl_test.go
+46
View File
@@ -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) {
weed/server/master_grpc_server_volume.go
+6 -3
View File
@@ -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")
}
weed/server/master_grpc_server_volume_warmup_test.go
+63
View File
@@ -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])
}
}
weed/server/volume_grpc_admin.go
+15
View File
@@ -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{}
weed/server/volume_grpc_admin_delete_test.go
+37
View File
@@ -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)
}
weed/storage/store.go
+2 -2
View File
@@ -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
}
weed/storage/store_delete_volume_test.go
+35
View File
@@ -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)
}
}
weed/wdclient/vidmap_authoritative_test.go
+49
View File
@@ -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)
}
}
weed/wdclient/vidmap_client.go
+9
View File
@@ -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