mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-15 00:09:00 +01:00
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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chris LuandClaude Opus 5.5 authored and GitHub committed
1 parent
c819e3fbea
commit
858f8cbad8
8 files changed
+650
-118
No files matched your search
@@ -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
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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...))
|
||||
}
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user