From 858f8cbad8339a05439e27ede5c4d74796d1cb2b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 14 Nov 2026 10:53:07 -0800 Subject: [PATCH] s3 lifecycle: check internal S3 authentication before starting a pass (#12645) * s3 lifecycle: add a LifecycleReady RPC that only authorizes the caller LifecycleReady runs the same checkAdminAuth as LifecycleDelete and reads no bucket or object metadata, so a lifecycle worker can confirm that an S3 front accepts its internal credentials before it starts a pass. Co-Authored-By: Claude Opus 5.5 * s3 lifecycle: check S3 fronts with LifecycleReady before the pass Execute used to dial the first registered S3 front and start loading policies and walking buckets, so a missing or mismatched jwt.filer_signing.key surfaced only as repeated per-bucket delete failures. It now checks each front in order with LifecycleReady, using the credentials LifecycleDelete sends, under a 10s bound that also follows parent cancellation. The first front that passes serves the pass; fronts that fail are reported in the job's progress and the log. When none passes, the job fails before any listing with an error that names the front and says whether this worker has the key set, without key or token material. Fronts that predate LifecycleReady are checked with an empty LifecycleDelete, which they authorize and then reject before any lookup. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/pb/s3_lifecycle.proto | 7 + weed/pb/s3_lifecycle_pb/s3_lifecycle.pb.go | 293 +++++++++++------- .../s3_lifecycle_pb/s3_lifecycle_grpc.pb.go | 50 ++- weed/s3api/s3api_internal_lifecycle.go | 9 + weed/s3api/s3api_internal_lifecycle_test.go | 32 ++ weed/worker/tasks/s3_lifecycle/handler.go | 23 +- weed/worker/tasks/s3_lifecycle/readiness.go | 74 +++++ .../tasks/s3_lifecycle/readiness_test.go | 280 +++++++++++++++++ 8 files changed, 650 insertions(+), 118 deletions(-) create mode 100644 weed/worker/tasks/s3_lifecycle/readiness.go create mode 100644 weed/worker/tasks/s3_lifecycle/readiness_test.go diff --git a/weed/pb/s3_lifecycle.proto b/weed/pb/s3_lifecycle.proto index 77addaf1b5..68646af133 100644 --- a/weed/pb/s3_lifecycle.proto +++ b/weed/pb/s3_lifecycle.proto @@ -41,6 +41,9 @@ enum ActionKind { // to the appropriate internal helper. service SeaweedS3LifecycleInternal { rpc LifecycleDelete(LifecycleDeleteRequest) returns (LifecycleDeleteResponse); + // LifecycleReady runs LifecycleDelete's authorization and nothing else, + // so a worker can check a front before it starts a pass. + rpc LifecycleReady(LifecycleReadyRequest) returns (LifecycleReadyResponse); } // LifecycleDeleteOutcome captures every per-event verdict the worker needs. @@ -93,6 +96,10 @@ message LifecycleDeleteResponse { string reason = 2; } +message LifecycleReadyRequest {} + +message LifecycleReadyResponse {} + // MessagePosition mirrors weed/util/log_buffer.MessagePosition for durable // storage. ts_ns + offset uniquely identifies a position within a per-filer // log buffer; per-shard cursors (see ReaderState) use this shape to skip past diff --git a/weed/pb/s3_lifecycle_pb/s3_lifecycle.pb.go b/weed/pb/s3_lifecycle_pb/s3_lifecycle.pb.go index dddcaff019..7f7f707827 100644 --- a/weed/pb/s3_lifecycle_pb/s3_lifecycle.pb.go +++ b/weed/pb/s3_lifecycle_pb/s3_lifecycle.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.6 -// protoc v6.33.4 +// protoc v3.21.12 // source: s3_lifecycle.proto package s3_lifecycle_pb @@ -270,7 +270,7 @@ func (x LifecycleState_RuleMode) Number() protoreflect.EnumNumber { // Deprecated: Use LifecycleState_RuleMode.Descriptor instead. func (LifecycleState_RuleMode) EnumDescriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{5, 0} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{7, 0} } // DEGRADED_REASON_UNSPECIFIED replaces the old NONE sentinel: an unset @@ -334,7 +334,7 @@ func (x LifecycleState_DegradedReason) Number() protoreflect.EnumNumber { // Deprecated: Use LifecycleState_DegradedReason.Descriptor instead. func (LifecycleState_DegradedReason) EnumDescriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{5, 1} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{7, 1} } type LifecycleDeleteRequest struct { @@ -524,6 +524,78 @@ func (x *LifecycleDeleteResponse) GetReason() string { return "" } +type LifecycleReadyRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LifecycleReadyRequest) Reset() { + *x = LifecycleReadyRequest{} + mi := &file_s3_lifecycle_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LifecycleReadyRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LifecycleReadyRequest) ProtoMessage() {} + +func (x *LifecycleReadyRequest) ProtoReflect() protoreflect.Message { + mi := &file_s3_lifecycle_proto_msgTypes[2] + 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 LifecycleReadyRequest.ProtoReflect.Descriptor instead. +func (*LifecycleReadyRequest) Descriptor() ([]byte, []int) { + return file_s3_lifecycle_proto_rawDescGZIP(), []int{2} +} + +type LifecycleReadyResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LifecycleReadyResponse) Reset() { + *x = LifecycleReadyResponse{} + mi := &file_s3_lifecycle_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LifecycleReadyResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LifecycleReadyResponse) ProtoMessage() {} + +func (x *LifecycleReadyResponse) ProtoReflect() protoreflect.Message { + mi := &file_s3_lifecycle_proto_msgTypes[3] + 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 LifecycleReadyResponse.ProtoReflect.Descriptor instead. +func (*LifecycleReadyResponse) Descriptor() ([]byte, []int) { + return file_s3_lifecycle_proto_rawDescGZIP(), []int{3} +} + // MessagePosition mirrors weed/util/log_buffer.MessagePosition for durable // storage. ts_ns + offset uniquely identifies a position within a per-filer // log buffer; per-shard cursors (see ReaderState) use this shape to skip past @@ -538,7 +610,7 @@ type MessagePosition struct { func (x *MessagePosition) Reset() { *x = MessagePosition{} - mi := &file_s3_lifecycle_proto_msgTypes[2] + mi := &file_s3_lifecycle_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -550,7 +622,7 @@ func (x *MessagePosition) String() string { func (*MessagePosition) ProtoMessage() {} func (x *MessagePosition) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[2] + mi := &file_s3_lifecycle_proto_msgTypes[4] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -563,7 +635,7 @@ func (x *MessagePosition) ProtoReflect() protoreflect.Message { // Deprecated: Use MessagePosition.ProtoReflect.Descriptor instead. func (*MessagePosition) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{2} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{4} } func (x *MessagePosition) GetTsNs() int64 { @@ -593,7 +665,7 @@ type FilerShardCursor struct { func (x *FilerShardCursor) Reset() { *x = FilerShardCursor{} - mi := &file_s3_lifecycle_proto_msgTypes[3] + mi := &file_s3_lifecycle_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -605,7 +677,7 @@ func (x *FilerShardCursor) String() string { func (*FilerShardCursor) ProtoMessage() {} func (x *FilerShardCursor) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[3] + mi := &file_s3_lifecycle_proto_msgTypes[5] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -618,7 +690,7 @@ func (x *FilerShardCursor) ProtoReflect() protoreflect.Message { // Deprecated: Use FilerShardCursor.ProtoReflect.Descriptor instead. func (*FilerShardCursor) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{3} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{5} } func (x *FilerShardCursor) GetFilerId() string { @@ -650,7 +722,7 @@ type EntryIdentity struct { func (x *EntryIdentity) Reset() { *x = EntryIdentity{} - mi := &file_s3_lifecycle_proto_msgTypes[4] + mi := &file_s3_lifecycle_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -662,7 +734,7 @@ func (x *EntryIdentity) String() string { func (*EntryIdentity) ProtoMessage() {} func (x *EntryIdentity) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[4] + mi := &file_s3_lifecycle_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -675,7 +747,7 @@ func (x *EntryIdentity) ProtoReflect() protoreflect.Message { // Deprecated: Use EntryIdentity.ProtoReflect.Descriptor instead. func (*EntryIdentity) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{4} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{6} } func (x *EntryIdentity) GetMtimeNs() int64 { @@ -730,15 +802,15 @@ type LifecycleState struct { LastSafetyScanTsNs int64 `protobuf:"varint,10,opt,name=last_safety_scan_ts_ns,json=lastSafetyScanTsNs,proto3" json:"last_safety_scan_ts_ns,omitempty"` NextSafetyScanTsNs int64 `protobuf:"varint,11,opt,name=next_safety_scan_ts_ns,json=nextSafetyScanTsNs,proto3" json:"next_safety_scan_ts_ns,omitempty"` // Counters for observability; no behavior depends on these. - // - evaluated_total: live entries the worker considered for this action - // (filter-matched + due-checked); includes objects that ended up not - // yet eligible. - // - expired_total: successful DONE outcomes (object/version/marker/MPU - // removed) under this action. - // - metadata_only_total: subset of expired_total where the worker - // short-circuited via volume-TTL routing (chunks left to volume GC). - // - error_total: outcomes that did NOT advance the cursor under this - // action (RETRY_LATER, BLOCKED, FATAL_EVENT_ERROR). + // - evaluated_total: live entries the worker considered for this action + // (filter-matched + due-checked); includes objects that ended up not + // yet eligible. + // - expired_total: successful DONE outcomes (object/version/marker/MPU + // removed) under this action. + // - metadata_only_total: subset of expired_total where the worker + // short-circuited via volume-TTL routing (chunks left to volume GC). + // - error_total: outcomes that did NOT advance the cursor under this + // action (RETRY_LATER, BLOCKED, FATAL_EVENT_ERROR). EvaluatedTotal int64 `protobuf:"varint,12,opt,name=evaluated_total,json=evaluatedTotal,proto3" json:"evaluated_total,omitempty"` ExpiredTotal int64 `protobuf:"varint,13,opt,name=expired_total,json=expiredTotal,proto3" json:"expired_total,omitempty"` MetadataOnlyTotal int64 `protobuf:"varint,14,opt,name=metadata_only_total,json=metadataOnlyTotal,proto3" json:"metadata_only_total,omitempty"` @@ -749,7 +821,7 @@ type LifecycleState struct { func (x *LifecycleState) Reset() { *x = LifecycleState{} - mi := &file_s3_lifecycle_proto_msgTypes[5] + mi := &file_s3_lifecycle_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -761,7 +833,7 @@ func (x *LifecycleState) String() string { func (*LifecycleState) ProtoMessage() {} func (x *LifecycleState) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[5] + mi := &file_s3_lifecycle_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -774,7 +846,7 @@ func (x *LifecycleState) ProtoReflect() protoreflect.Message { // Deprecated: Use LifecycleState.ProtoReflect.Descriptor instead. func (*LifecycleState) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{5} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{7} } func (x *LifecycleState) GetRuleHash() []byte { @@ -897,7 +969,7 @@ type PendingItem struct { func (x *PendingItem) Reset() { *x = PendingItem{} - mi := &file_s3_lifecycle_proto_msgTypes[6] + mi := &file_s3_lifecycle_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -909,7 +981,7 @@ func (x *PendingItem) String() string { func (*PendingItem) ProtoMessage() {} func (x *PendingItem) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[6] + mi := &file_s3_lifecycle_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -922,7 +994,7 @@ func (x *PendingItem) ProtoReflect() protoreflect.Message { // Deprecated: Use PendingItem.ProtoReflect.Descriptor instead. func (*PendingItem) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{6} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{8} } func (x *PendingItem) GetObjectPath() string { @@ -967,7 +1039,7 @@ type BootstrapState struct { func (x *BootstrapState) Reset() { *x = BootstrapState{} - mi := &file_s3_lifecycle_proto_msgTypes[7] + mi := &file_s3_lifecycle_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -979,7 +1051,7 @@ func (x *BootstrapState) String() string { func (*BootstrapState) ProtoMessage() {} func (x *BootstrapState) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[7] + mi := &file_s3_lifecycle_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -992,7 +1064,7 @@ func (x *BootstrapState) ProtoReflect() protoreflect.Message { // Deprecated: Use BootstrapState.ProtoReflect.Descriptor instead. func (*BootstrapState) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{7} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{9} } func (x *BootstrapState) GetLastScannedPath() string { @@ -1038,7 +1110,7 @@ type ReaderState struct { func (x *ReaderState) Reset() { *x = ReaderState{} - mi := &file_s3_lifecycle_proto_msgTypes[8] + mi := &file_s3_lifecycle_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1050,7 +1122,7 @@ func (x *ReaderState) String() string { func (*ReaderState) ProtoMessage() {} func (x *ReaderState) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[8] + mi := &file_s3_lifecycle_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1063,7 +1135,7 @@ func (x *ReaderState) ProtoReflect() protoreflect.Message { // Deprecated: Use ReaderState.ProtoReflect.Descriptor instead. func (*ReaderState) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{8} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{10} } func (x *ReaderState) GetPrimaryFilerEndpoint() string { @@ -1110,7 +1182,7 @@ type FilerShardCursorList struct { func (x *FilerShardCursorList) Reset() { *x = FilerShardCursorList{} - mi := &file_s3_lifecycle_proto_msgTypes[9] + mi := &file_s3_lifecycle_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1122,7 +1194,7 @@ func (x *FilerShardCursorList) String() string { func (*FilerShardCursorList) ProtoMessage() {} func (x *FilerShardCursorList) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[9] + mi := &file_s3_lifecycle_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1135,7 +1207,7 @@ func (x *FilerShardCursorList) ProtoReflect() protoreflect.Message { // Deprecated: Use FilerShardCursorList.ProtoReflect.Descriptor instead. func (*FilerShardCursorList) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{9} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{11} } func (x *FilerShardCursorList) GetCursors() []*FilerShardCursor { @@ -1159,7 +1231,7 @@ type TailDrainedKey struct { func (x *TailDrainedKey) Reset() { *x = TailDrainedKey{} - mi := &file_s3_lifecycle_proto_msgTypes[10] + mi := &file_s3_lifecycle_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1171,7 +1243,7 @@ func (x *TailDrainedKey) String() string { func (*TailDrainedKey) ProtoMessage() {} func (x *TailDrainedKey) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[10] + mi := &file_s3_lifecycle_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1184,7 +1256,7 @@ func (x *TailDrainedKey) ProtoReflect() protoreflect.Message { // Deprecated: Use TailDrainedKey.ProtoReflect.Descriptor instead. func (*TailDrainedKey) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{10} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{12} } func (x *TailDrainedKey) GetStreamKind() StreamKind { @@ -1241,7 +1313,7 @@ type BlockerRecord struct { func (x *BlockerRecord) Reset() { *x = BlockerRecord{} - mi := &file_s3_lifecycle_proto_msgTypes[11] + mi := &file_s3_lifecycle_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1253,7 +1325,7 @@ func (x *BlockerRecord) String() string { func (*BlockerRecord) ProtoMessage() {} func (x *BlockerRecord) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[11] + mi := &file_s3_lifecycle_proto_msgTypes[13] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1266,7 +1338,7 @@ func (x *BlockerRecord) ProtoReflect() protoreflect.Message { // Deprecated: Use BlockerRecord.ProtoReflect.Descriptor instead. func (*BlockerRecord) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{11} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{13} } func (x *BlockerRecord) GetStreamKind() StreamKind { @@ -1384,7 +1456,7 @@ type StreamKey struct { func (x *StreamKey) Reset() { *x = StreamKey{} - mi := &file_s3_lifecycle_proto_msgTypes[12] + mi := &file_s3_lifecycle_proto_msgTypes[14] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1396,7 +1468,7 @@ func (x *StreamKey) String() string { func (*StreamKey) ProtoMessage() {} func (x *StreamKey) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[12] + mi := &file_s3_lifecycle_proto_msgTypes[14] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1409,7 +1481,7 @@ func (x *StreamKey) ProtoReflect() protoreflect.Message { // Deprecated: Use StreamKey.ProtoReflect.Descriptor instead. func (*StreamKey) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{12} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{14} } func (x *StreamKey) GetKey() isStreamKey_Key { @@ -1494,7 +1566,7 @@ type OriginalKey struct { func (x *OriginalKey) Reset() { *x = OriginalKey{} - mi := &file_s3_lifecycle_proto_msgTypes[13] + mi := &file_s3_lifecycle_proto_msgTypes[15] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1506,7 +1578,7 @@ func (x *OriginalKey) String() string { func (*OriginalKey) ProtoMessage() {} func (x *OriginalKey) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[13] + mi := &file_s3_lifecycle_proto_msgTypes[15] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1519,7 +1591,7 @@ func (x *OriginalKey) ProtoReflect() protoreflect.Message { // Deprecated: Use OriginalKey.ProtoReflect.Descriptor instead. func (*OriginalKey) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{13} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{15} } func (x *OriginalKey) GetShard() string { @@ -1553,7 +1625,7 @@ type PredicateKey struct { func (x *PredicateKey) Reset() { *x = PredicateKey{} - mi := &file_s3_lifecycle_proto_msgTypes[14] + mi := &file_s3_lifecycle_proto_msgTypes[16] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1565,7 +1637,7 @@ func (x *PredicateKey) String() string { func (*PredicateKey) ProtoMessage() {} func (x *PredicateKey) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[14] + mi := &file_s3_lifecycle_proto_msgTypes[16] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1578,7 +1650,7 @@ func (x *PredicateKey) ProtoReflect() protoreflect.Message { // Deprecated: Use PredicateKey.ProtoReflect.Descriptor instead. func (*PredicateKey) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{14} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{16} } func (x *PredicateKey) GetShard() string { @@ -1608,7 +1680,7 @@ type BootstrapKey struct { func (x *BootstrapKey) Reset() { *x = BootstrapKey{} - mi := &file_s3_lifecycle_proto_msgTypes[15] + mi := &file_s3_lifecycle_proto_msgTypes[17] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1620,7 +1692,7 @@ func (x *BootstrapKey) String() string { func (*BootstrapKey) ProtoMessage() {} func (x *BootstrapKey) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[15] + mi := &file_s3_lifecycle_proto_msgTypes[17] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1633,7 +1705,7 @@ func (x *BootstrapKey) ProtoReflect() protoreflect.Message { // Deprecated: Use BootstrapKey.ProtoReflect.Descriptor instead. func (*BootstrapKey) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{15} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{17} } func (x *BootstrapKey) GetBucket() string { @@ -1684,7 +1756,7 @@ type PendingKey struct { func (x *PendingKey) Reset() { *x = PendingKey{} - mi := &file_s3_lifecycle_proto_msgTypes[16] + mi := &file_s3_lifecycle_proto_msgTypes[18] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1696,7 +1768,7 @@ func (x *PendingKey) String() string { func (*PendingKey) ProtoMessage() {} func (x *PendingKey) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[16] + mi := &file_s3_lifecycle_proto_msgTypes[18] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1709,7 +1781,7 @@ func (x *PendingKey) ProtoReflect() protoreflect.Message { // Deprecated: Use PendingKey.ProtoReflect.Descriptor instead. func (*PendingKey) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{16} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{18} } func (x *PendingKey) GetBucket() string { @@ -1762,7 +1834,7 @@ type RetryBudgetEntry struct { func (x *RetryBudgetEntry) Reset() { *x = RetryBudgetEntry{} - mi := &file_s3_lifecycle_proto_msgTypes[17] + mi := &file_s3_lifecycle_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1774,7 +1846,7 @@ func (x *RetryBudgetEntry) String() string { func (*RetryBudgetEntry) ProtoMessage() {} func (x *RetryBudgetEntry) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[17] + mi := &file_s3_lifecycle_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1787,7 +1859,7 @@ func (x *RetryBudgetEntry) ProtoReflect() protoreflect.Message { // Deprecated: Use RetryBudgetEntry.ProtoReflect.Descriptor instead. func (*RetryBudgetEntry) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{17} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{19} } func (x *RetryBudgetEntry) GetKey() *StreamKey { @@ -1836,7 +1908,7 @@ type RetryTarget struct { func (x *RetryTarget) Reset() { *x = RetryTarget{} - mi := &file_s3_lifecycle_proto_msgTypes[18] + mi := &file_s3_lifecycle_proto_msgTypes[20] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1848,7 +1920,7 @@ func (x *RetryTarget) String() string { func (*RetryTarget) ProtoMessage() {} func (x *RetryTarget) ProtoReflect() protoreflect.Message { - mi := &file_s3_lifecycle_proto_msgTypes[18] + mi := &file_s3_lifecycle_proto_msgTypes[20] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1861,7 +1933,7 @@ func (x *RetryTarget) ProtoReflect() protoreflect.Message { // Deprecated: Use RetryTarget.ProtoReflect.Descriptor instead. func (*RetryTarget) Descriptor() ([]byte, []int) { - return file_s3_lifecycle_proto_rawDescGZIP(), []int{18} + return file_s3_lifecycle_proto_rawDescGZIP(), []int{20} } func (x *RetryTarget) GetKey() *StreamKey { @@ -1929,7 +2001,9 @@ const file_s3_lifecycle_proto_rawDesc = "" + "\x12engine_snapshot_id\x18\x15 \x01(\x04R\x10engineSnapshotId\"t\n" + "\x17LifecycleDeleteResponse\x12A\n" + "\aoutcome\x18\x01 \x01(\x0e2'.s3_lifecycle_pb.LifecycleDeleteOutcomeR\aoutcome\x12\x16\n" + - "\x06reason\x18\x02 \x01(\tR\x06reason\">\n" + + "\x06reason\x18\x02 \x01(\tR\x06reason\"\x17\n" + + "\x15LifecycleReadyRequest\"\x18\n" + + "\x16LifecycleReadyResponse\">\n" + "\x0fMessagePosition\x12\x13\n" + "\x05ts_ns\x18\x01 \x01(\x03R\x04tsNs\x12\x16\n" + "\x06offset\x18\x02 \x01(\x03R\x06offset\"k\n" + @@ -2093,9 +2167,10 @@ const file_s3_lifecycle_proto_rawDesc = "" + "\bORIGINAL\x10\x01\x12\r\n" + "\tPREDICATE\x10\x02\x12\r\n" + "\tBOOTSTRAP\x10\x03\x12\v\n" + - "\aPENDING\x10\x042\x82\x01\n" + + "\aPENDING\x10\x042\xe5\x01\n" + "\x1aSeaweedS3LifecycleInternal\x12d\n" + - "\x0fLifecycleDelete\x12'.s3_lifecycle_pb.LifecycleDeleteRequest\x1a(.s3_lifecycle_pb.LifecycleDeleteResponseB\\\n" + + "\x0fLifecycleDelete\x12'.s3_lifecycle_pb.LifecycleDeleteRequest\x1a(.s3_lifecycle_pb.LifecycleDeleteResponse\x12a\n" + + "\x0eLifecycleReady\x12&.s3_lifecycle_pb.LifecycleReadyRequest\x1a'.s3_lifecycle_pb.LifecycleReadyResponseB\\\n" + "\x10seaweedfs.clientB\x10S3LifecycleProtoZ6github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pbb\x06proto3" var ( @@ -2111,7 +2186,7 @@ func file_s3_lifecycle_proto_rawDescGZIP() []byte { } var file_s3_lifecycle_proto_enumTypes = make([]protoimpl.EnumInfo, 5) -var file_s3_lifecycle_proto_msgTypes = make([]protoimpl.MessageInfo, 20) +var file_s3_lifecycle_proto_msgTypes = make([]protoimpl.MessageInfo, 22) var file_s3_lifecycle_proto_goTypes = []any{ (ActionKind)(0), // 0: s3_lifecycle_pb.ActionKind (LifecycleDeleteOutcome)(0), // 1: s3_lifecycle_pb.LifecycleDeleteOutcome @@ -2120,58 +2195,62 @@ var file_s3_lifecycle_proto_goTypes = []any{ (LifecycleState_DegradedReason)(0), // 4: s3_lifecycle_pb.LifecycleState.DegradedReason (*LifecycleDeleteRequest)(nil), // 5: s3_lifecycle_pb.LifecycleDeleteRequest (*LifecycleDeleteResponse)(nil), // 6: s3_lifecycle_pb.LifecycleDeleteResponse - (*MessagePosition)(nil), // 7: s3_lifecycle_pb.MessagePosition - (*FilerShardCursor)(nil), // 8: s3_lifecycle_pb.FilerShardCursor - (*EntryIdentity)(nil), // 9: s3_lifecycle_pb.EntryIdentity - (*LifecycleState)(nil), // 10: s3_lifecycle_pb.LifecycleState - (*PendingItem)(nil), // 11: s3_lifecycle_pb.PendingItem - (*BootstrapState)(nil), // 12: s3_lifecycle_pb.BootstrapState - (*ReaderState)(nil), // 13: s3_lifecycle_pb.ReaderState - (*FilerShardCursorList)(nil), // 14: s3_lifecycle_pb.FilerShardCursorList - (*TailDrainedKey)(nil), // 15: s3_lifecycle_pb.TailDrainedKey - (*BlockerRecord)(nil), // 16: s3_lifecycle_pb.BlockerRecord - (*StreamKey)(nil), // 17: s3_lifecycle_pb.StreamKey - (*OriginalKey)(nil), // 18: s3_lifecycle_pb.OriginalKey - (*PredicateKey)(nil), // 19: s3_lifecycle_pb.PredicateKey - (*BootstrapKey)(nil), // 20: s3_lifecycle_pb.BootstrapKey - (*PendingKey)(nil), // 21: s3_lifecycle_pb.PendingKey - (*RetryBudgetEntry)(nil), // 22: s3_lifecycle_pb.RetryBudgetEntry - (*RetryTarget)(nil), // 23: s3_lifecycle_pb.RetryTarget - nil, // 24: s3_lifecycle_pb.ReaderState.LastProcessedOriginalEntry + (*LifecycleReadyRequest)(nil), // 7: s3_lifecycle_pb.LifecycleReadyRequest + (*LifecycleReadyResponse)(nil), // 8: s3_lifecycle_pb.LifecycleReadyResponse + (*MessagePosition)(nil), // 9: s3_lifecycle_pb.MessagePosition + (*FilerShardCursor)(nil), // 10: s3_lifecycle_pb.FilerShardCursor + (*EntryIdentity)(nil), // 11: s3_lifecycle_pb.EntryIdentity + (*LifecycleState)(nil), // 12: s3_lifecycle_pb.LifecycleState + (*PendingItem)(nil), // 13: s3_lifecycle_pb.PendingItem + (*BootstrapState)(nil), // 14: s3_lifecycle_pb.BootstrapState + (*ReaderState)(nil), // 15: s3_lifecycle_pb.ReaderState + (*FilerShardCursorList)(nil), // 16: s3_lifecycle_pb.FilerShardCursorList + (*TailDrainedKey)(nil), // 17: s3_lifecycle_pb.TailDrainedKey + (*BlockerRecord)(nil), // 18: s3_lifecycle_pb.BlockerRecord + (*StreamKey)(nil), // 19: s3_lifecycle_pb.StreamKey + (*OriginalKey)(nil), // 20: s3_lifecycle_pb.OriginalKey + (*PredicateKey)(nil), // 21: s3_lifecycle_pb.PredicateKey + (*BootstrapKey)(nil), // 22: s3_lifecycle_pb.BootstrapKey + (*PendingKey)(nil), // 23: s3_lifecycle_pb.PendingKey + (*RetryBudgetEntry)(nil), // 24: s3_lifecycle_pb.RetryBudgetEntry + (*RetryTarget)(nil), // 25: s3_lifecycle_pb.RetryTarget + nil, // 26: s3_lifecycle_pb.ReaderState.LastProcessedOriginalEntry } var file_s3_lifecycle_proto_depIdxs = []int32{ 0, // 0: s3_lifecycle_pb.LifecycleDeleteRequest.action_kind:type_name -> s3_lifecycle_pb.ActionKind - 9, // 1: s3_lifecycle_pb.LifecycleDeleteRequest.expected_identity:type_name -> s3_lifecycle_pb.EntryIdentity + 11, // 1: s3_lifecycle_pb.LifecycleDeleteRequest.expected_identity:type_name -> s3_lifecycle_pb.EntryIdentity 1, // 2: s3_lifecycle_pb.LifecycleDeleteResponse.outcome:type_name -> s3_lifecycle_pb.LifecycleDeleteOutcome - 7, // 3: s3_lifecycle_pb.FilerShardCursor.position:type_name -> s3_lifecycle_pb.MessagePosition + 9, // 3: s3_lifecycle_pb.FilerShardCursor.position:type_name -> s3_lifecycle_pb.MessagePosition 0, // 4: s3_lifecycle_pb.LifecycleState.action_kind:type_name -> s3_lifecycle_pb.ActionKind 3, // 5: s3_lifecycle_pb.LifecycleState.mode:type_name -> s3_lifecycle_pb.LifecycleState.RuleMode 4, // 6: s3_lifecycle_pb.LifecycleState.degraded_reason:type_name -> s3_lifecycle_pb.LifecycleState.DegradedReason - 9, // 7: s3_lifecycle_pb.PendingItem.expected_identity:type_name -> s3_lifecycle_pb.EntryIdentity - 24, // 8: s3_lifecycle_pb.ReaderState.last_processed_original:type_name -> s3_lifecycle_pb.ReaderState.LastProcessedOriginalEntry - 8, // 9: s3_lifecycle_pb.ReaderState.last_processed_predicate:type_name -> s3_lifecycle_pb.FilerShardCursor - 15, // 10: s3_lifecycle_pb.ReaderState.tail_drained_streams:type_name -> s3_lifecycle_pb.TailDrainedKey - 8, // 11: s3_lifecycle_pb.FilerShardCursorList.cursors:type_name -> s3_lifecycle_pb.FilerShardCursor + 11, // 7: s3_lifecycle_pb.PendingItem.expected_identity:type_name -> s3_lifecycle_pb.EntryIdentity + 26, // 8: s3_lifecycle_pb.ReaderState.last_processed_original:type_name -> s3_lifecycle_pb.ReaderState.LastProcessedOriginalEntry + 10, // 9: s3_lifecycle_pb.ReaderState.last_processed_predicate:type_name -> s3_lifecycle_pb.FilerShardCursor + 17, // 10: s3_lifecycle_pb.ReaderState.tail_drained_streams:type_name -> s3_lifecycle_pb.TailDrainedKey + 10, // 11: s3_lifecycle_pb.FilerShardCursorList.cursors:type_name -> s3_lifecycle_pb.FilerShardCursor 2, // 12: s3_lifecycle_pb.TailDrainedKey.stream_kind:type_name -> s3_lifecycle_pb.StreamKind 2, // 13: s3_lifecycle_pb.BlockerRecord.stream_kind:type_name -> s3_lifecycle_pb.StreamKind - 7, // 14: s3_lifecycle_pb.BlockerRecord.position:type_name -> s3_lifecycle_pb.MessagePosition + 9, // 14: s3_lifecycle_pb.BlockerRecord.position:type_name -> s3_lifecycle_pb.MessagePosition 0, // 15: s3_lifecycle_pb.BlockerRecord.action_kind:type_name -> s3_lifecycle_pb.ActionKind - 18, // 16: s3_lifecycle_pb.StreamKey.original:type_name -> s3_lifecycle_pb.OriginalKey - 19, // 17: s3_lifecycle_pb.StreamKey.predicate:type_name -> s3_lifecycle_pb.PredicateKey - 20, // 18: s3_lifecycle_pb.StreamKey.bootstrap:type_name -> s3_lifecycle_pb.BootstrapKey - 21, // 19: s3_lifecycle_pb.StreamKey.pending:type_name -> s3_lifecycle_pb.PendingKey - 7, // 20: s3_lifecycle_pb.OriginalKey.position:type_name -> s3_lifecycle_pb.MessagePosition - 7, // 21: s3_lifecycle_pb.PredicateKey.position:type_name -> s3_lifecycle_pb.MessagePosition + 20, // 16: s3_lifecycle_pb.StreamKey.original:type_name -> s3_lifecycle_pb.OriginalKey + 21, // 17: s3_lifecycle_pb.StreamKey.predicate:type_name -> s3_lifecycle_pb.PredicateKey + 22, // 18: s3_lifecycle_pb.StreamKey.bootstrap:type_name -> s3_lifecycle_pb.BootstrapKey + 23, // 19: s3_lifecycle_pb.StreamKey.pending:type_name -> s3_lifecycle_pb.PendingKey + 9, // 20: s3_lifecycle_pb.OriginalKey.position:type_name -> s3_lifecycle_pb.MessagePosition + 9, // 21: s3_lifecycle_pb.PredicateKey.position:type_name -> s3_lifecycle_pb.MessagePosition 0, // 22: s3_lifecycle_pb.BootstrapKey.action_kind:type_name -> s3_lifecycle_pb.ActionKind 0, // 23: s3_lifecycle_pb.PendingKey.action_kind:type_name -> s3_lifecycle_pb.ActionKind - 17, // 24: s3_lifecycle_pb.RetryBudgetEntry.key:type_name -> s3_lifecycle_pb.StreamKey - 17, // 25: s3_lifecycle_pb.RetryTarget.key:type_name -> s3_lifecycle_pb.StreamKey + 19, // 24: s3_lifecycle_pb.RetryBudgetEntry.key:type_name -> s3_lifecycle_pb.StreamKey + 19, // 25: s3_lifecycle_pb.RetryTarget.key:type_name -> s3_lifecycle_pb.StreamKey 0, // 26: s3_lifecycle_pb.RetryTarget.action_kind:type_name -> s3_lifecycle_pb.ActionKind - 14, // 27: s3_lifecycle_pb.ReaderState.LastProcessedOriginalEntry.value:type_name -> s3_lifecycle_pb.FilerShardCursorList + 16, // 27: s3_lifecycle_pb.ReaderState.LastProcessedOriginalEntry.value:type_name -> s3_lifecycle_pb.FilerShardCursorList 5, // 28: s3_lifecycle_pb.SeaweedS3LifecycleInternal.LifecycleDelete:input_type -> s3_lifecycle_pb.LifecycleDeleteRequest - 6, // 29: s3_lifecycle_pb.SeaweedS3LifecycleInternal.LifecycleDelete:output_type -> s3_lifecycle_pb.LifecycleDeleteResponse - 29, // [29:30] is the sub-list for method output_type - 28, // [28:29] is the sub-list for method input_type + 7, // 29: s3_lifecycle_pb.SeaweedS3LifecycleInternal.LifecycleReady:input_type -> s3_lifecycle_pb.LifecycleReadyRequest + 6, // 30: s3_lifecycle_pb.SeaweedS3LifecycleInternal.LifecycleDelete:output_type -> s3_lifecycle_pb.LifecycleDeleteResponse + 8, // 31: s3_lifecycle_pb.SeaweedS3LifecycleInternal.LifecycleReady:output_type -> s3_lifecycle_pb.LifecycleReadyResponse + 30, // [30:32] is the sub-list for method output_type + 28, // [28:30] is the sub-list for method input_type 28, // [28:28] is the sub-list for extension type_name 28, // [28:28] is the sub-list for extension extendee 0, // [0:28] is the sub-list for field type_name @@ -2182,7 +2261,7 @@ func file_s3_lifecycle_proto_init() { if File_s3_lifecycle_proto != nil { return } - file_s3_lifecycle_proto_msgTypes[12].OneofWrappers = []any{ + file_s3_lifecycle_proto_msgTypes[14].OneofWrappers = []any{ (*StreamKey_Original)(nil), (*StreamKey_Predicate)(nil), (*StreamKey_Bootstrap)(nil), @@ -2194,7 +2273,7 @@ func file_s3_lifecycle_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_s3_lifecycle_proto_rawDesc), len(file_s3_lifecycle_proto_rawDesc)), NumEnums: 5, - NumMessages: 20, + NumMessages: 22, NumExtensions: 0, NumServices: 1, }, diff --git a/weed/pb/s3_lifecycle_pb/s3_lifecycle_grpc.pb.go b/weed/pb/s3_lifecycle_pb/s3_lifecycle_grpc.pb.go index d773deb627..6bb8930e31 100644 --- a/weed/pb/s3_lifecycle_pb/s3_lifecycle_grpc.pb.go +++ b/weed/pb/s3_lifecycle_pb/s3_lifecycle_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.5.1 -// - protoc v6.33.4 +// - protoc-gen-go-grpc v1.6.2 +// - protoc v3.21.12 // source: s3_lifecycle.proto package s3_lifecycle_pb @@ -20,6 +20,7 @@ const _ = grpc.SupportPackageIsVersion9 const ( SeaweedS3LifecycleInternal_LifecycleDelete_FullMethodName = "/s3_lifecycle_pb.SeaweedS3LifecycleInternal/LifecycleDelete" + SeaweedS3LifecycleInternal_LifecycleReady_FullMethodName = "/s3_lifecycle_pb.SeaweedS3LifecycleInternal/LifecycleReady" ) // SeaweedS3LifecycleInternalClient is the client API for SeaweedS3LifecycleInternal service. @@ -34,6 +35,9 @@ const ( // to the appropriate internal helper. type SeaweedS3LifecycleInternalClient interface { LifecycleDelete(ctx context.Context, in *LifecycleDeleteRequest, opts ...grpc.CallOption) (*LifecycleDeleteResponse, error) + // LifecycleReady runs LifecycleDelete's authorization and nothing else, + // so a worker can check a front before it starts a pass. + LifecycleReady(ctx context.Context, in *LifecycleReadyRequest, opts ...grpc.CallOption) (*LifecycleReadyResponse, error) } type seaweedS3LifecycleInternalClient struct { @@ -54,6 +58,16 @@ func (c *seaweedS3LifecycleInternalClient) LifecycleDelete(ctx context.Context, return out, nil } +func (c *seaweedS3LifecycleInternalClient) LifecycleReady(ctx context.Context, in *LifecycleReadyRequest, opts ...grpc.CallOption) (*LifecycleReadyResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(LifecycleReadyResponse) + err := c.cc.Invoke(ctx, SeaweedS3LifecycleInternal_LifecycleReady_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // SeaweedS3LifecycleInternalServer is the server API for SeaweedS3LifecycleInternal service. // All implementations must embed UnimplementedSeaweedS3LifecycleInternalServer // for forward compatibility. @@ -66,6 +80,9 @@ func (c *seaweedS3LifecycleInternalClient) LifecycleDelete(ctx context.Context, // to the appropriate internal helper. type SeaweedS3LifecycleInternalServer interface { LifecycleDelete(context.Context, *LifecycleDeleteRequest) (*LifecycleDeleteResponse, error) + // LifecycleReady runs LifecycleDelete's authorization and nothing else, + // so a worker can check a front before it starts a pass. + LifecycleReady(context.Context, *LifecycleReadyRequest) (*LifecycleReadyResponse, error) mustEmbedUnimplementedSeaweedS3LifecycleInternalServer() } @@ -77,7 +94,10 @@ type SeaweedS3LifecycleInternalServer interface { type UnimplementedSeaweedS3LifecycleInternalServer struct{} func (UnimplementedSeaweedS3LifecycleInternalServer) LifecycleDelete(context.Context, *LifecycleDeleteRequest) (*LifecycleDeleteResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method LifecycleDelete not implemented") + return nil, status.Error(codes.Unimplemented, "method LifecycleDelete not implemented") +} +func (UnimplementedSeaweedS3LifecycleInternalServer) LifecycleReady(context.Context, *LifecycleReadyRequest) (*LifecycleReadyResponse, error) { + return nil, status.Error(codes.Unimplemented, "method LifecycleReady not implemented") } func (UnimplementedSeaweedS3LifecycleInternalServer) mustEmbedUnimplementedSeaweedS3LifecycleInternalServer() { } @@ -91,7 +111,7 @@ type UnsafeSeaweedS3LifecycleInternalServer interface { } func RegisterSeaweedS3LifecycleInternalServer(s grpc.ServiceRegistrar, srv SeaweedS3LifecycleInternalServer) { - // If the following call pancis, it indicates UnimplementedSeaweedS3LifecycleInternalServer was + // If the following call panics, it indicates UnimplementedSeaweedS3LifecycleInternalServer was // embedded by pointer and is nil. This will cause panics if an // unimplemented method is ever invoked, so we test this at initialization // time to prevent it from happening at runtime later due to I/O. @@ -119,6 +139,24 @@ func _SeaweedS3LifecycleInternal_LifecycleDelete_Handler(srv interface{}, ctx co return interceptor(ctx, in, info, handler) } +func _SeaweedS3LifecycleInternal_LifecycleReady_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(LifecycleReadyRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(SeaweedS3LifecycleInternalServer).LifecycleReady(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: SeaweedS3LifecycleInternal_LifecycleReady_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(SeaweedS3LifecycleInternalServer).LifecycleReady(ctx, req.(*LifecycleReadyRequest)) + } + return interceptor(ctx, in, info, handler) +} + // SeaweedS3LifecycleInternal_ServiceDesc is the grpc.ServiceDesc for SeaweedS3LifecycleInternal service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -130,6 +168,10 @@ var SeaweedS3LifecycleInternal_ServiceDesc = grpc.ServiceDesc{ MethodName: "LifecycleDelete", Handler: _SeaweedS3LifecycleInternal_LifecycleDelete_Handler, }, + { + MethodName: "LifecycleReady", + Handler: _SeaweedS3LifecycleInternal_LifecycleReady_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "s3_lifecycle.proto", diff --git a/weed/s3api/s3api_internal_lifecycle.go b/weed/s3api/s3api_internal_lifecycle.go index 720c626f7e..47c927d7d4 100644 --- a/weed/s3api/s3api_internal_lifecycle.go +++ b/weed/s3api/s3api_internal_lifecycle.go @@ -59,6 +59,15 @@ func (s3a *S3ApiServer) LifecycleDelete(ctx context.Context, req *s3_lifecycle_p return s3a.lifecycleDispatch(ctx, req, entry) } +// LifecycleReady authorizes the caller as LifecycleDelete does and touches +// no bucket or object metadata. +func (s3a *S3ApiServer) LifecycleReady(ctx context.Context, _ *s3_lifecycle_pb.LifecycleReadyRequest) (*s3_lifecycle_pb.LifecycleReadyResponse, error) { + if err := s3a.checkAdminAuth(ctx); err != nil { + return nil, err + } + return &s3_lifecycle_pb.LifecycleReadyResponse{}, nil +} + func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest, entry *filer_pb.Entry) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) { metadataOnly := entryUsesMetadataOnlyDelete(entry) switch req.ActionKind { diff --git a/weed/s3api/s3api_internal_lifecycle_test.go b/weed/s3api/s3api_internal_lifecycle_test.go index 0009dbe753..147be1eba4 100644 --- a/weed/s3api/s3api_internal_lifecycle_test.go +++ b/weed/s3api/s3api_internal_lifecycle_test.go @@ -384,6 +384,38 @@ func TestLifecycleDelete_NoSigningKey_LocalPeerOnly(t *testing.T) { } } +func TestLifecycleReady_AuthorizesLikeLifecycleDelete(t *testing.T) { + good := security.GenJwtForFilerAdmin(security.SigningKey(testS3IamCacheSigningKey), 60) + bad := security.GenJwtForFilerAdmin(security.SigningKey("a-different-key"), 60) + remoteCtx := peer.NewContext(context.Background(), + &peer.Peer{Addr: &net.TCPAddr{IP: net.ParseIP("203.0.113.7"), Port: 11111}}) + cases := []struct { + name string + signingKey string + ctx context.Context + want codes.Code + }{ + {"key, no token", testS3IamCacheSigningKey, context.Background(), codes.Unauthenticated}, + {"key, mis-signed token", testS3IamCacheSigningKey, s3IamCacheBearerCtx(string(bad)), codes.Unauthenticated}, + {"key, valid token", testS3IamCacheSigningKey, s3IamCacheBearerCtx(string(good)), codes.OK}, + {"no key, remote peer", "", remoteCtx, codes.Unauthenticated}, + {"no key, loopback peer", "", s3LocalPeerCtx(), codes.OK}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + s := newTestS3IamCacheServer(t, tc.signingKey) + _, err := s.LifecycleReady(tc.ctx, &s3_lifecycle_pb.LifecycleReadyRequest{}) + if got := status.Code(err); got != tc.want { + t.Fatalf("LifecycleReady: got code %v, want %v (err=%v)", got, tc.want, err) + } + _, deleteErr := s.LifecycleDelete(tc.ctx, &s3_lifecycle_pb.LifecycleDeleteRequest{}) + if got := status.Code(deleteErr); got != tc.want { + t.Fatalf("LifecycleDelete disagrees: got code %v, want %v (err=%v)", got, tc.want, deleteErr) + } + }) + } +} + func TestLifecycleDelete_CurrentExpirationRequiresLatestVersion(t *testing.T) { listed := &filer_pb.Entry{ Name: "v_old", diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 3bdf41f66b..48c84a704e 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -2,6 +2,7 @@ package s3_lifecycle import ( "context" + "errors" "fmt" "math" "strconv" @@ -249,13 +250,20 @@ func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequ return fmt.Errorf("buckets path: %w", err) } - dialCtx, dialCancel = context.WithTimeout(runCtx, 30*time.Second) - s3Conn, err := pb.GrpcDial(dialCtx, s3Endpoints[0], false, h.grpcDialOption) - dialCancel() - if err != nil { - return fmt.Errorf("dial s3 %s: %w", s3Endpoints[0], err) + s3Conn, s3Endpoint, skipped := h.selectS3Front(runCtx, s3Endpoints) + if s3Conn == nil { + return noReadyS3Error(skipped) } defer s3Conn.Close() + if len(skipped) > 0 { + skippedErr := errors.Join(skipped...) + glog.Warningf("s3 lifecycle: using s3 %s, skipped: %v", s3Endpoint, skippedErr) + _ = sender.SendProgress(&plugin_pb.JobProgressUpdate{ + JobId: request.Job.JobId, JobType: jobType, + State: plugin_pb.JobState_JOB_STATE_RUNNING, Stage: "starting", + Message: fmt.Sprintf("using s3 %s, skipped: %v", s3Endpoint, skippedErr), + }) + } rpc := s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(s3Conn) report, err := h.executeDailyReplay(runCtx, request, bucketsPath, filerClient, rpc, cfg, sender) @@ -453,8 +461,9 @@ func parsePositiveInt(s string) (int, bool) { } // clusterS3Endpoints returns the master-discovered S3 gRPC addresses for the -// cluster. The handler dials the first reachable one; the master refreshes -// the list on KeepConnected so a stale entry self-heals on the next run. +// cluster. The handler uses the first one that passes checkLifecycleReady; +// the master refreshes the list on KeepConnected so a stale entry self-heals +// on the next run. func clusterS3Endpoints(cc *plugin_pb.ClusterContext) []string { if cc == nil { return nil diff --git a/weed/worker/tasks/s3_lifecycle/readiness.go b/weed/worker/tasks/s3_lifecycle/readiness.go new file mode 100644 index 0000000000..bb828b8e83 --- /dev/null +++ b/weed/worker/tasks/s3_lifecycle/readiness.go @@ -0,0 +1,74 @@ +package s3_lifecycle + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/seaweedfs/seaweedfs/weed/credential" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +var lifecycleReadyTimeout = 10 * time.Second + +// checkLifecycleReady authenticates to one S3 front with the credentials +// LifecycleDelete uses, without touching bucket or object metadata. +func checkLifecycleReady(ctx context.Context, c s3_lifecycle_pb.SeaweedS3LifecycleInternalClient) error { + ctx, cancel := context.WithTimeout(ctx, lifecycleReadyTimeout) + defer cancel() + ctx, _ = credential.WithS3InternalAdminAuth(ctx) + _, err := c.LifecycleReady(ctx, &s3_lifecycle_pb.LifecycleReadyRequest{}) + if status.Code(err) == codes.Unimplemented { + // Older fronts authorize an empty delete, then reject it before any lookup. + _, err = c.LifecycleDelete(ctx, &s3_lifecycle_pb.LifecycleDeleteRequest{}) + } + return err +} + +// selectS3Front returns a connection to the first front that passes +// checkLifecycleReady, plus why each earlier front was skipped. A nil +// connection means none passed. +func (h *Handler) selectS3Front(ctx context.Context, endpoints []string) (*grpc.ClientConn, string, []error) { + var skipped []error + for _, addr := range endpoints { + if err := ctx.Err(); err != nil { + return nil, "", append(skipped, err) + } + dialCtx, dialCancel := context.WithTimeout(ctx, 30*time.Second) + conn, err := pb.GrpcDial(dialCtx, addr, false, h.grpcDialOption) + dialCancel() + if err != nil { + skipped = append(skipped, fmt.Errorf("dial s3 %s: %w", addr, err)) + continue + } + if err := checkLifecycleReady(ctx, s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(conn)); err != nil { + conn.Close() + skipped = append(skipped, readinessError(addr, err)) + continue + } + return conn, addr, skipped + } + return nil, "", skipped +} + +func readinessError(addr string, err error) error { + switch status.Code(err) { + case codes.Unauthenticated, codes.PermissionDenied: + workerKey := "is not set" + if util.GetViper().GetString("jwt.filer_signing.key") != "" { + workerKey = "is set" + } + return fmt.Errorf("s3 %s rejected lifecycle credentials: %w; this worker's jwt.filer_signing.key %s, and it must match the key in the s3 server's security.toml (an s3 server without the key accepts only local callers)", addr, err, workerKey) + } + return fmt.Errorf("s3 %s readiness check: %w", addr, err) +} + +func noReadyS3Error(skipped []error) error { + return fmt.Errorf("no s3 server is ready for lifecycle deletes: %w", errors.Join(skipped...)) +} diff --git a/weed/worker/tasks/s3_lifecycle/readiness_test.go b/weed/worker/tasks/s3_lifecycle/readiness_test.go new file mode 100644 index 0000000000..ddc5b35d62 --- /dev/null +++ b/weed/worker/tasks/s3_lifecycle/readiness_test.go @@ -0,0 +1,280 @@ +package s3_lifecycle + +import ( + "context" + "errors" + "net" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" +) + +const ( + frontKey = "readiness-test-front-key" + otherKey = "readiness-test-other-key" +) + +type testFront struct { + s3_lifecycle_pb.UnimplementedSeaweedS3LifecycleInternalServer + key string + legacy bool + block bool + readies atomic.Int32 + deletes atomic.Int32 +} + +func (f *testFront) LifecycleReady(ctx context.Context, _ *s3_lifecycle_pb.LifecycleReadyRequest) (*s3_lifecycle_pb.LifecycleReadyResponse, error) { + f.readies.Add(1) + if f.legacy { + return nil, status.Error(codes.Unimplemented, "method LifecycleReady not implemented") + } + if f.block { + <-ctx.Done() + return nil, status.FromContextError(ctx.Err()).Err() + } + if err := security.CheckGrpcAdminToken(ctx, security.SigningKey(f.key)); err != nil { + return nil, err + } + return &s3_lifecycle_pb.LifecycleReadyResponse{}, nil +} + +func (f *testFront) LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) { + f.deletes.Add(1) + if err := security.CheckGrpcAdminToken(ctx, security.SigningKey(f.key)); err != nil { + return nil, err + } + if req.Bucket == "" || req.ObjectPath == "" { + return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED}, nil + } + return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}, nil +} + +type countingFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + lists atomic.Int32 +} + +func (f *countingFiler) GetFilerConfiguration(context.Context, *filer_pb.GetFilerConfigurationRequest) (*filer_pb.GetFilerConfigurationResponse, error) { + return &filer_pb.GetFilerConfigurationResponse{DirBuckets: "/buckets"}, nil +} + +func (f *countingFiler) ListEntries(*filer_pb.ListEntriesRequest, grpc.ServerStreamingServer[filer_pb.ListEntriesResponse]) error { + f.lists.Add(1) + return nil +} + +func serveGrpc(t *testing.T, register func(*grpc.Server)) string { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + s := grpc.NewServer() + register(s) + go s.Serve(lis) + t.Cleanup(s.Stop) + return lis.Addr().String() +} + +func startFront(t *testing.T, f *testFront) string { + return serveGrpc(t, func(s *grpc.Server) { s3_lifecycle_pb.RegisterSeaweedS3LifecycleInternalServer(s, f) }) +} + +func setWorkerKey(t *testing.T, key string) { + t.Helper() + v := util.GetViper() + prev := v.GetString("jwt.filer_signing.key") + v.Set("jwt.filer_signing.key", key) + t.Cleanup(func() { v.Set("jwt.filer_signing.key", prev) }) +} + +func newTestHandler() *Handler { + return NewHandler(grpc.WithTransportCredentials(insecure.NewCredentials())) +} + +func assertNoSecrets(t *testing.T, msg string) { + t.Helper() + for _, secret := range []string{frontKey, otherKey, "Bearer", "eyJ"} { + if strings.Contains(msg, secret) { + t.Fatalf("error exposes %q: %s", secret, msg) + } + } +} + +func TestSelectS3Front_SharedKeyPassesWithoutDeletes(t *testing.T) { + setWorkerKey(t, frontKey) + front := &testFront{key: frontKey} + addr := startFront(t, front) + + conn, got, skipped := newTestHandler().selectS3Front(context.Background(), []string{addr}) + if conn == nil { + t.Fatalf("expected a ready front, skipped: %v", skipped) + } + conn.Close() + if got != addr || len(skipped) != 0 { + t.Fatalf("got front %q skipped %v, want %q and none", got, skipped, addr) + } + if front.deletes.Load() != 0 { + t.Fatalf("readiness check sent %d LifecycleDelete calls", front.deletes.Load()) + } +} + +func TestSelectS3Front_KeyProblemsNameTheConfig(t *testing.T) { + cases := []struct { + name, workerKey, wantState string + }{ + {"missing worker key", "", "is not set"}, + {"mismatched worker key", otherKey, "is set"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + setWorkerKey(t, tc.workerKey) + addr := startFront(t, &testFront{key: frontKey}) + + conn, _, skipped := newTestHandler().selectS3Front(context.Background(), []string{addr}) + if conn != nil { + conn.Close() + t.Fatal("front with a different key was selected") + } + if len(skipped) != 1 || status.Code(errors.Unwrap(skipped[0])) != codes.Unauthenticated { + t.Fatalf("want one Unauthenticated skip, got %v", skipped) + } + msg := noReadyS3Error(skipped).Error() + for _, want := range []string{addr, "jwt.filer_signing.key " + tc.wantState, "security.toml"} { + if !strings.Contains(msg, want) { + t.Fatalf("error %q does not mention %q", msg, want) + } + } + assertNoSecrets(t, msg) + }) + } +} + +func TestSelectS3Front_SkipsFrontWithOtherKey(t *testing.T) { + setWorkerKey(t, frontKey) + bad := &testFront{key: otherKey} + good := &testFront{key: frontKey} + badAddr, goodAddr := startFront(t, bad), startFront(t, good) + + conn, got, skipped := newTestHandler().selectS3Front(context.Background(), []string{badAddr, goodAddr}) + if conn == nil { + t.Fatalf("expected the second front, skipped: %v", skipped) + } + conn.Close() + if got != goodAddr { + t.Fatalf("selected %q, want %q", got, goodAddr) + } + if len(skipped) != 1 || !strings.Contains(skipped[0].Error(), badAddr) || !strings.Contains(skipped[0].Error(), "rejected lifecycle credentials") { + t.Fatalf("want the first front reported as a credential problem, got %v", skipped) + } + if bad.deletes.Load() != 0 || good.deletes.Load() != 0 { + t.Fatal("readiness check sent LifecycleDelete calls") + } +} + +func TestSelectS3Front_OlderFrontUsesEmptyDelete(t *testing.T) { + setWorkerKey(t, frontKey) + ok := &testFront{key: frontKey, legacy: true} + okAddr := startFront(t, ok) + conn, _, skipped := newTestHandler().selectS3Front(context.Background(), []string{okAddr}) + if conn == nil { + t.Fatalf("older front with the shared key was skipped: %v", skipped) + } + conn.Close() + if ok.deletes.Load() != 1 { + t.Fatalf("want one empty LifecycleDelete, got %d", ok.deletes.Load()) + } + + mismatched := &testFront{key: otherKey, legacy: true} + conn, _, skipped = newTestHandler().selectS3Front(context.Background(), []string{startFront(t, mismatched)}) + if conn != nil { + conn.Close() + t.Fatal("older front with another key was selected") + } + if len(skipped) != 1 || !strings.Contains(skipped[0].Error(), "rejected lifecycle credentials") { + t.Fatalf("want a credential problem, got %v", skipped) + } +} + +func TestSelectS3Front_TimeoutMovesOn(t *testing.T) { + setWorkerKey(t, frontKey) + prev := lifecycleReadyTimeout + lifecycleReadyTimeout = 200 * time.Millisecond + t.Cleanup(func() { lifecycleReadyTimeout = prev }) + stuckAddr := startFront(t, &testFront{key: frontKey, block: true}) + goodAddr := startFront(t, &testFront{key: frontKey}) + + start := time.Now() + conn, got, skipped := newTestHandler().selectS3Front(context.Background(), []string{stuckAddr, goodAddr}) + if conn == nil { + t.Fatalf("expected the second front, skipped: %v", skipped) + } + conn.Close() + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("readiness check took %v", elapsed) + } + if got != goodAddr || len(skipped) != 1 || status.Code(errors.Unwrap(skipped[0])) != codes.DeadlineExceeded { + t.Fatalf("got %q skipped %v, want %q after a DeadlineExceeded skip", got, skipped, goodAddr) + } +} + +func TestSelectS3Front_ParentCancelStops(t *testing.T) { + setWorkerKey(t, frontKey) + next := &testFront{key: frontKey} + stuckAddr := startFront(t, &testFront{key: frontKey, block: true}) + nextAddr := startFront(t, next) + + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(100*time.Millisecond, cancel) + start := time.Now() + conn, _, skipped := newTestHandler().selectS3Front(ctx, []string{stuckAddr, nextAddr}) + if conn != nil { + conn.Close() + t.Fatal("selected a front after the parent was cancelled") + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("cancelled selection took %v", elapsed) + } + if !errors.Is(noReadyS3Error(skipped), context.Canceled) { + t.Fatalf("want context.Canceled, got %v", skipped) + } + if next.readies.Load() != 0 { + t.Fatal("tried another front after the parent was cancelled") + } +} + +func TestExecute_RejectedCredentialsFailBeforeListing(t *testing.T) { + setWorkerKey(t, otherKey) + filer := &countingFiler{} + filerAddr := serveGrpc(t, func(s *grpc.Server) { filer_pb.RegisterSeaweedFilerServer(s, filer) }) + front := &testFront{key: frontKey} + frontAddr := startFront(t, front) + + err := newTestHandler().Execute(context.Background(), &plugin_pb.ExecuteJobRequest{ + Job: &plugin_pb.JobSpec{ + JobId: "j1", JobType: jobType, + Parameters: map[string]*plugin_pb.ConfigValue{ + "filer_grpc_address": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: filerAddr}}, + }, + }, + ClusterContext: &plugin_pb.ClusterContext{S3GrpcAddresses: []string{frontAddr}}, + }, &recordingExecSender{}) + if err == nil || !strings.Contains(err.Error(), "jwt.filer_signing.key") { + t.Fatalf("want a signing-key error, got %v", err) + } + assertNoSecrets(t, err.Error()) + if filer.lists.Load() != 0 || front.deletes.Load() != 0 { + t.Fatalf("pass started anyway: %d listings, %d deletes", filer.lists.Load(), front.deletes.Load()) + } +}