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()) + } +}