diff --git a/adapter/admin_grpc.go b/adapter/admin_grpc.go index a8cb10a8a..1e147330b 100644 --- a/adapter/admin_grpc.go +++ b/adapter/admin_grpc.go @@ -60,8 +60,9 @@ func (n NodeIdentity) toProto() *pb.NodeIdentity { // GetClusterOverview and GetRaftGroups; remaining RPCs return Unimplemented so // the generated client can still compile against older nodes during rollout. type AdminServer struct { - self NodeIdentity - members []NodeIdentity + self NodeIdentity + members []NodeIdentity + capabilities map[string]bool groupsMu sync.RWMutex groups map[uint64]AdminGroup @@ -89,10 +90,11 @@ type AdminServer struct { func NewAdminServer(self NodeIdentity, members []NodeIdentity) *AdminServer { cloned := append([]NodeIdentity(nil), members...) return &AdminServer{ - self: self, - members: cloned, - groups: make(map[uint64]AdminGroup), - now: time.Now, + self: self, + members: cloned, + capabilities: make(map[string]bool), + groups: make(map[uint64]AdminGroup), + now: time.Now, } } @@ -129,6 +131,20 @@ func (s *AdminServer) RegisterSampler(sampler KeyVizSampler) { s.groupsMu.Unlock() } +// SetCapability exposes a local binary or runtime capability in +// GetClusterOverview. Disabled capabilities are kept in the map so fan-out +// callers can distinguish "known false" from an older server that lacks the +// field or key entirely. +func (s *AdminServer) SetCapability(name string, enabled bool) { + name = strings.TrimSpace(name) + if s == nil || name == "" { + return + } + s.groupsMu.Lock() + s.capabilities[name] = enabled + s.groupsMu.Unlock() +} + // GetClusterOverview returns the local node identity, the current member // list, and per-group leader identity collected from the engines registered // via RegisterGroup. The member list is the union of (a) the bootstrap seed @@ -145,9 +161,23 @@ func (s *AdminServer) GetClusterOverview( Self: s.self.toProto(), Members: members, GroupLeaders: leaders, + Capabilities: s.snapshotCapabilities(), }, nil } +func (s *AdminServer) snapshotCapabilities() map[string]bool { + s.groupsMu.RLock() + defer s.groupsMu.RUnlock() + if len(s.capabilities) == 0 { + return nil + } + out := make(map[string]bool, len(s.capabilities)) + for name, enabled := range s.capabilities { + out[name] = enabled + } + return out +} + // snapshotMembers unions the seed members with the live Configuration of each // registered group, preferring the live address when the same NodeID appears // in both sources. A stale bootstrap entry cannot outvote a readdressed node: diff --git a/adapter/admin_grpc_test.go b/adapter/admin_grpc_test.go index 5724165ec..75fe0b9a2 100644 --- a/adapter/admin_grpc_test.go +++ b/adapter/admin_grpc_test.go @@ -71,6 +71,24 @@ func TestGetClusterOverviewReturnsSelfAndLeaders(t *testing.T) { } } +func TestGetClusterOverviewReturnsCapabilities(t *testing.T) { + t.Parallel() + srv := NewAdminServer(NodeIdentity{NodeID: "node-a"}, nil) + srv.SetCapability(S3BlobOffloadCapabilityName, false) + srv.SetCapability("feature-test", true) + + resp, err := srv.GetClusterOverview(context.Background(), &pb.GetClusterOverviewRequest{}) + if err != nil { + t.Fatalf("GetClusterOverview: %v", err) + } + if got, ok := resp.Capabilities[S3BlobOffloadCapabilityName]; !ok || got { + t.Fatalf("s3 blob offload capability = (%v, %v), want (false, true)", got, ok) + } + if got, ok := resp.Capabilities["feature-test"]; !ok || !got { + t.Fatalf("feature-test capability = (%v, %v), want (true, true)", got, ok) + } +} + func TestGetRaftGroupsExposesCommitApplied(t *testing.T) { t.Parallel() srv := NewAdminServer(NodeIdentity{NodeID: "n1"}, nil) diff --git a/adapter/s3.go b/adapter/s3.go index e267d7ccf..6c4874157 100644 --- a/adapter/s3.go +++ b/adapter/s3.go @@ -111,6 +111,9 @@ type S3Server struct { cleanupSem chan struct{} putAdmission *s3PutAdmission putAdmissionObserver S3PutAdmissionObserver + blobOffloadEnabled bool + blobOffloadChecker S3BlobOffloadCapabilityChecker + blobOffloadObserver S3BlobOffloadObserver } type s3BucketMeta struct { @@ -336,14 +339,15 @@ type s3ListPartEntry struct { func NewS3Server(listen net.Listener, s3Addr string, st store.MVCCStore, coordinate kv.Coordinator, leaderS3 map[string]string, opts ...S3ServerOption) *S3Server { s := &S3Server{ - listen: listen, - s3Addr: s3Addr, - region: s3DefaultRegion, - store: st, - coordinator: kv.WithKeyVizLabel(coordinate, keyviz.LabelS3), - leaderS3: cloneLeaderAddrMap(leaderS3), - cleanupSem: make(chan struct{}, s3ManifestCleanupWorkers), - putAdmission: newS3PutAdmissionFromEnv(), + listen: listen, + s3Addr: s3Addr, + region: s3DefaultRegion, + store: st, + coordinator: kv.WithKeyVizLabel(coordinate, keyviz.LabelS3), + leaderS3: cloneLeaderAddrMap(leaderS3), + cleanupSem: make(chan struct{}, s3ManifestCleanupWorkers), + putAdmission: newS3PutAdmissionFromEnv(), + blobOffloadEnabled: newS3BlobOffloadEnabledFromEnv(), } for _, opt := range opts { if opt != nil { @@ -877,6 +881,7 @@ func (s *S3Server) putObject(w http.ResponseWriter, r *http.Request, bucket stri if !s.admitS3PutRequest(w, r, bucket, objectKey, s3MaxObjectSizeBytes, "object exceeds maximum allowed size") { return } + s.observeS3BlobOffloadDecision(r.Context()) upload, uploadBodyErr, uploadErr := s.uploadS3ObjectData( r.Context(), r, streamBody, state, bucket, objectKey, expectedPayloadSHA, ) @@ -1228,6 +1233,7 @@ func (s *S3Server) uploadPart(w http.ResponseWriter, r *http.Request, bucket str if !s.admitS3PutRequest(w, r, bucket, objectKey, s3MaxPartSizeBytes, "part exceeds maximum allowed size") { return } + s.observeS3BlobOffloadDecision(r.Context()) upload, previous, uploadBodyErr, uploadErr := s.storeS3UploadPart( r.Context(), r, streamBody, state, bucket, objectKey, uploadID, admissionProtocol, ) diff --git a/adapter/s3_admin.go b/adapter/s3_admin.go index a9eb8915a..0d8c58d28 100644 --- a/adapter/s3_admin.go +++ b/adapter/s3_admin.go @@ -496,6 +496,7 @@ func bucketDeleteSafetyNetElems(bucket string, generation uint64) []*kv.Elem[kv. {Op: kv.DelPrefix, Key: s3keys.UploadMetaPrefixForBucket(bucket, generation)}, {Op: kv.DelPrefix, Key: s3keys.UploadPartPrefixForBucket(bucket, generation)}, {Op: kv.DelPrefix, Key: s3keys.BlobPrefixForBucket(bucket, generation)}, + {Op: kv.DelPrefix, Key: s3keys.ChunkRefPrefixForBucket(bucket, generation)}, {Op: kv.DelPrefix, Key: s3keys.GCUploadPrefixForBucket(bucket, generation)}, {Op: kv.DelPrefix, Key: s3keys.RoutePrefixForBucket(bucket, generation)}, } diff --git a/adapter/s3_admin_test.go b/adapter/s3_admin_test.go index 3a83677f2..5caf92e26 100644 --- a/adapter/s3_admin_test.go +++ b/adapter/s3_admin_test.go @@ -386,7 +386,7 @@ func TestS3Server_AdminDeleteBucket_SweepsOrphansAcrossAllPerBucketPrefixes(t *t gen := summary.Generation require.NotZero(t, gen) - // Plant orphan keys across the 5 non-manifest per-bucket + // Plant orphan keys across the 6 non-manifest per-bucket // prefixes. Each entry is what a concurrent PutObject (or its // in-flight multipart upload state) would leave behind if it // committed in the AdminDeleteBucket race window. The values @@ -401,6 +401,7 @@ func TestS3Server_AdminDeleteBucket_SweepsOrphansAcrossAllPerBucketPrefixes(t *t {Op: kv.Put, Key: s3keys.UploadMetaKey(bucket, gen, objectName, uploadID), Value: []byte("orphan-upload-meta")}, {Op: kv.Put, Key: s3keys.UploadPartKey(bucket, gen, objectName, uploadID, 1), Value: []byte("orphan-part")}, {Op: kv.Put, Key: s3keys.BlobKey(bucket, gen, objectName, uploadID, 1, 0), Value: []byte("orphan-chunk")}, + {Op: kv.Put, Key: s3keys.ChunkRefKey(bucket, gen, objectName, uploadID, 1, 0), Value: []byte("orphan-chunk-ref")}, {Op: kv.Put, Key: s3keys.GCUploadKey(bucket, gen, objectName, uploadID), Value: []byte("orphan-gc")}, {Op: kv.Put, Key: s3keys.RouteKey(bucket, gen, objectName), Value: []byte("orphan-route")}, } @@ -432,6 +433,7 @@ func TestS3Server_AdminDeleteBucket_SweepsOrphansAcrossAllPerBucketPrefixes(t *t {"upload_meta", s3keys.UploadMetaPrefixForBucket(bucket, gen)}, {"upload_part", s3keys.UploadPartPrefixForBucket(bucket, gen)}, {"blob", s3keys.BlobPrefixForBucket(bucket, gen)}, + {"chunk_ref", s3keys.ChunkRefPrefixForBucket(bucket, gen)}, {"gc_upload", s3keys.GCUploadPrefixForBucket(bucket, gen)}, {"route", s3keys.RoutePrefixForBucket(bucket, gen)}, } diff --git a/adapter/s3_blob_offload.go b/adapter/s3_blob_offload.go new file mode 100644 index 000000000..fb98dadd5 --- /dev/null +++ b/adapter/s3_blob_offload.go @@ -0,0 +1,118 @@ +package adapter + +import ( + "context" + "log/slog" + "os" + "strconv" + "strings" +) + +const ( + // S3BlobOffloadCapabilityName is the Admin GetClusterOverview capability + // key used by mixed-version PUT admission before writing chunkref metadata. + S3BlobOffloadCapabilityName = "feature_s3_blob_offload" + + s3BlobOffloadEnvVar = "ELASTICKV_S3_BLOB_OFFLOAD" + + s3BlobOffloadModeLegacy = "legacy" + s3BlobOffloadModeOffload = "offload" + + s3BlobOffloadReasonFlagDisabled = "flag_disabled" + s3BlobOffloadReasonCapabilityMissing = "capability_missing" + s3BlobOffloadReasonDataPathDisabled = "data_path_disabled" + s3BlobOffloadReasonEnabled = "enabled" +) + +// S3BlobOffloadCapabilityChecker is the cluster-wide mixed-version guard for +// admitting an S3 PUT into the offloaded chunkref/chunkblob keyspace. +type S3BlobOffloadCapabilityChecker interface { + AllPeersSupportS3BlobOffload(ctx context.Context) bool +} + +// S3BlobOffloadObserver records offload rollout and chunkblob durability +// outcomes. The data path is still fail-closed, so only decision counters are +// emitted today; the remaining counters are wired for the durable replication +// PR that will turn the offload path on. +type S3BlobOffloadObserver interface { + ObserveS3BlobOffloadDecision(mode, reason string) + ObserveS3ChunkBlobReplicationDegraded() + ObserveS3ChunkBlobSHAMismatch() + ObserveS3ChunkBlobUnrecoverable() +} + +type s3BlobOffloadDecision struct { + mode string + reason string +} + +// S3BlobOffloadLocalCapability reports whether this binary can safely serve the +// full offloaded S3 chunkref/chunkblob data path. The scaffolding PR keeps this +// false so future leaders do not mistake these nodes for offload readers. +func S3BlobOffloadLocalCapability() bool { + return false +} + +func WithS3BlobOffloadEnabled(enabled bool) S3ServerOption { + return func(server *S3Server) { + if server == nil { + return + } + server.blobOffloadEnabled = enabled + } +} + +func WithS3BlobOffloadCapabilityChecker(checker S3BlobOffloadCapabilityChecker) S3ServerOption { + return func(server *S3Server) { + if server == nil { + return + } + server.blobOffloadChecker = checker + } +} + +func WithS3BlobOffloadObserver(observer S3BlobOffloadObserver) S3ServerOption { + return func(server *S3Server) { + if server == nil { + return + } + server.blobOffloadObserver = observer + } +} + +func newS3BlobOffloadEnabledFromEnv() bool { + raw, ok := os.LookupEnv(s3BlobOffloadEnvVar) + if !ok { + return false + } + raw = strings.TrimSpace(raw) + if raw == "" { + return false + } + enabled, err := strconv.ParseBool(raw) + if err != nil { + slog.Warn("invalid S3 blob offload boolean env; using default", "name", s3BlobOffloadEnvVar, "value", raw, "default", false) + return false + } + return enabled +} + +func (s *S3Server) s3BlobOffloadDecision(ctx context.Context) s3BlobOffloadDecision { + if s == nil || !s.blobOffloadEnabled { + return s3BlobOffloadDecision{mode: s3BlobOffloadModeLegacy, reason: s3BlobOffloadReasonFlagDisabled} + } + if s.blobOffloadChecker == nil || !s.blobOffloadChecker.AllPeersSupportS3BlobOffload(ctx) { + return s3BlobOffloadDecision{mode: s3BlobOffloadModeLegacy, reason: s3BlobOffloadReasonCapabilityMissing} + } + if !S3BlobOffloadLocalCapability() { + return s3BlobOffloadDecision{mode: s3BlobOffloadModeLegacy, reason: s3BlobOffloadReasonDataPathDisabled} + } + return s3BlobOffloadDecision{mode: s3BlobOffloadModeOffload, reason: s3BlobOffloadReasonEnabled} +} + +func (s *S3Server) observeS3BlobOffloadDecision(ctx context.Context) { + decision := s.s3BlobOffloadDecision(ctx) + if s != nil && s.blobOffloadObserver != nil { + s.blobOffloadObserver.ObserveS3BlobOffloadDecision(decision.mode, decision.reason) + } +} diff --git a/adapter/s3_blob_offload_test.go b/adapter/s3_blob_offload_test.go new file mode 100644 index 000000000..3d3a8fb48 --- /dev/null +++ b/adapter/s3_blob_offload_test.go @@ -0,0 +1,126 @@ +package adapter + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/bootjp/elastickv/internal/s3keys" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" +) + +func TestS3BlobOffloadEnvDefaultsDisabled(t *testing.T) { + t.Setenv(s3BlobOffloadEnvVar, "") + + st := store.NewMVCCStore() + server := NewS3Server(nil, "", st, newLocalAdapterCoordinator(st), nil) + + require.False(t, server.blobOffloadEnabled) +} + +func TestS3BlobOffloadEnvCanEnableGate(t *testing.T) { + t.Setenv(s3BlobOffloadEnvVar, "true") + + st := store.NewMVCCStore() + server := NewS3Server(nil, "", st, newLocalAdapterCoordinator(st), nil) + + require.True(t, server.blobOffloadEnabled) +} + +func TestS3BlobOffloadDecisionFailsClosed(t *testing.T) { + t.Parallel() + + st := store.NewMVCCStore() + server := NewS3Server( + nil, + "", + st, + newLocalAdapterCoordinator(st), + nil, + WithS3BlobOffloadEnabled(true), + ) + + decision := server.s3BlobOffloadDecision(context.Background()) + require.Equal(t, s3BlobOffloadModeLegacy, decision.mode) + require.Equal(t, s3BlobOffloadReasonCapabilityMissing, decision.reason) +} + +func TestS3BlobOffloadDecisionRefusesUntilDataPathIsEnabled(t *testing.T) { + t.Parallel() + + st := store.NewMVCCStore() + server := NewS3Server( + nil, + "", + st, + newLocalAdapterCoordinator(st), + nil, + WithS3BlobOffloadEnabled(true), + WithS3BlobOffloadCapabilityChecker(alwaysS3BlobOffloadCapable{}), + ) + + decision := server.s3BlobOffloadDecision(context.Background()) + require.Equal(t, s3BlobOffloadModeLegacy, decision.mode) + require.Equal(t, s3BlobOffloadReasonDataPathDisabled, decision.reason) +} + +func TestS3Server_PutObjectBlobOffloadFlagFallsBackToLegacyWithoutCapability(t *testing.T) { + t.Parallel() + + st := store.NewMVCCStore() + coord := newLocalAdapterCoordinator(st) + observer := &recordingS3BlobOffloadObserver{} + server := NewS3Server( + nil, + "", + st, + coord, + nil, + WithS3BlobOffloadEnabled(true), + WithS3BlobOffloadObserver(observer), + ) + + rec := httptest.NewRecorder() + server.handle(rec, newS3TestRequest(http.MethodPut, "/bucket-a", nil)) + require.Equal(t, http.StatusOK, rec.Code) + + rec = httptest.NewRecorder() + server.handle(rec, newS3TestRequest(http.MethodPut, "/bucket-a/object.txt", strings.NewReader("payload"))) + require.Equal(t, http.StatusOK, rec.Code) + + readTS := server.readTS() + legacy, err := st.ScanAt(context.Background(), []byte(s3keys.BlobPrefix), prefixScanEnd([]byte(s3keys.BlobPrefix)), 10, readTS) + require.NoError(t, err) + require.NotEmpty(t, legacy) + + chunkRefs, err := st.ScanAt(context.Background(), []byte(s3keys.ChunkRefPrefix), prefixScanEnd([]byte(s3keys.ChunkRefPrefix)), 10, readTS) + require.NoError(t, err) + require.Empty(t, chunkRefs) + + require.Equal(t, []s3BlobOffloadDecision{ + {mode: s3BlobOffloadModeLegacy, reason: s3BlobOffloadReasonCapabilityMissing}, + }, observer.decisions) +} + +type alwaysS3BlobOffloadCapable struct{} + +func (alwaysS3BlobOffloadCapable) AllPeersSupportS3BlobOffload(context.Context) bool { + return true +} + +type recordingS3BlobOffloadObserver struct { + decisions []s3BlobOffloadDecision +} + +func (o *recordingS3BlobOffloadObserver) ObserveS3BlobOffloadDecision(mode, reason string) { + o.decisions = append(o.decisions, s3BlobOffloadDecision{mode: mode, reason: reason}) +} + +func (o *recordingS3BlobOffloadObserver) ObserveS3ChunkBlobReplicationDegraded() {} + +func (o *recordingS3BlobOffloadObserver) ObserveS3ChunkBlobSHAMismatch() {} + +func (o *recordingS3BlobOffloadObserver) ObserveS3ChunkBlobUnrecoverable() {} diff --git a/internal/s3keys/chunkref.go b/internal/s3keys/chunkref.go new file mode 100644 index 000000000..b14d6baf9 --- /dev/null +++ b/internal/s3keys/chunkref.go @@ -0,0 +1,151 @@ +package s3keys + +import ( + "bytes" + "encoding/binary" + "encoding/hex" + "errors" +) + +const ( + chunkBlobSHA256Bytes = 32 + chunkBlobSHA256HexBytes = chunkBlobSHA256Bytes * 2 + chunkRefValueVersionV1 = byte(1) + chunkRefValueFixedBytes = 1 + chunkBlobSHA256Bytes + u64Bytes + 2 + chunkRefSourcePeerLenSize = 2 + maxChunkRefSourcePeerLen = int(^uint16(0)) +) + +// ErrChunkRefSourcePeerTooLong is returned when a ChunkRefValue source peer +// cannot be encoded into the fixed-width value header. +var ErrChunkRefSourcePeerTooLong = errors.New("s3 chunkref source peer is too long") + +// ChunkRefValue is the stored value for a !s3|chunkref| entry. It points from +// an object part chunk to the content-addressed !s3|chunkblob| payload. +type ChunkRefValue struct { + ContentSHA256 [chunkBlobSHA256Bytes]byte + Size uint64 + SourcePeer string +} + +// ParseChunkRefKey decodes a !s3|chunkref| key into its object-part location. +func ParseChunkRefKey(key []byte) (bucket string, generation uint64, object string, uploadID string, partNo uint64, chunkNo uint64, ok bool) { + if !bytes.HasPrefix(key, chunkRefPrefixBytes) { + return "", 0, "", "", 0, 0, false + } + parts, ok := parseObjectChunkKey(key, len(chunkRefPrefixBytes)) + if !ok { + return "", 0, "", "", 0, 0, false + } + return parts.bucket, parts.generation, parts.object, parts.uploadID, parts.partNo, parts.chunkNo, true +} + +// ChunkBlobKey builds a content-addressed blob key from a SHA-256 digest. +func ChunkBlobKey(contentSHA256 [chunkBlobSHA256Bytes]byte) []byte { + out := make([]byte, 0, len(ChunkBlobPrefix)+chunkBlobSHA256HexBytes) + out = append(out, chunkBlobPrefixBytes...) + out = hex.AppendEncode(out, contentSHA256[:]) + return out +} + +// ParseChunkBlobKey decodes a !s3|chunkblob| key into its SHA-256 digest. +func ParseChunkBlobKey(key []byte) ([chunkBlobSHA256Bytes]byte, bool) { + var sha [chunkBlobSHA256Bytes]byte + if !bytes.HasPrefix(key, chunkBlobPrefixBytes) { + return sha, false + } + encoded := key[len(chunkBlobPrefixBytes):] + if len(encoded) != chunkBlobSHA256HexBytes { + return sha, false + } + if _, err := hex.Decode(sha[:], encoded); err != nil { + return sha, false + } + return sha, true +} + +// EncodeChunkRefValue encodes a chunk reference value using a versioned binary +// shape so future offload metadata can fail closed on unknown versions. +func EncodeChunkRefValue(ref ChunkRefValue) ([]byte, error) { + sourcePeerLen, err := chunkRefSourcePeerLen(len(ref.SourcePeer)) + if err != nil { + return nil, err + } + out := make([]byte, 0, chunkRefValueFixedBytes+len(ref.SourcePeer)) + out = append(out, chunkRefValueVersionV1) + out = append(out, ref.ContentSHA256[:]...) + out = appendU64(out, ref.Size) + var sourceLen [chunkRefSourcePeerLenSize]byte + binary.BigEndian.PutUint16(sourceLen[:], sourcePeerLen) + out = append(out, sourceLen[:]...) + out = append(out, ref.SourcePeer...) + return out, nil +} + +func chunkRefSourcePeerLen(n int) (uint16, error) { + if n < 0 || n > maxChunkRefSourcePeerLen { + return 0, ErrChunkRefSourcePeerTooLong + } + return uint16(n), nil +} + +// DecodeChunkRefValue decodes a chunk reference value. It returns ok=false for +// unknown versions, truncated values, or trailing-length mismatches. +func DecodeChunkRefValue(value []byte) (ChunkRefValue, bool) { + var ref ChunkRefValue + if len(value) < chunkRefValueFixedBytes || value[0] != chunkRefValueVersionV1 { + return ref, false + } + copy(ref.ContentSHA256[:], value[1:1+chunkBlobSHA256Bytes]) + sizeOffset := 1 + chunkBlobSHA256Bytes + var ok bool + ref.Size, sizeOffset, ok = readU64(value, sizeOffset) + if !ok || len(value)-sizeOffset < chunkRefSourcePeerLenSize { + return ChunkRefValue{}, false + } + sourceLen := int(binary.BigEndian.Uint16(value[sizeOffset : sizeOffset+chunkRefSourcePeerLenSize])) + sourceOffset := sizeOffset + chunkRefSourcePeerLenSize + if len(value)-sourceOffset != sourceLen { + return ChunkRefValue{}, false + } + ref.SourcePeer = string(value[sourceOffset:]) + return ref, true +} + +type parsedObjectChunkKey struct { + bucket string + generation uint64 + object string + uploadID string + partNo uint64 + chunkNo uint64 +} + +func parseObjectChunkKey(key []byte, offset int) (parsedObjectChunkKey, bool) { + var p parsedObjectChunkKey + bucketRaw, next, ok := decodeSegment(key, offset) + if !ok { + return p, false + } + if p.generation, next, ok = readU64(key, next); !ok { + return p, false + } + objectRaw, next, ok := decodeSegment(key, next) + if !ok { + return p, false + } + uploadIDRaw, next, ok := decodeSegment(key, next) + if !ok { + return p, false + } + if p.partNo, next, ok = readU64(key, next); !ok { + return p, false + } + if p.chunkNo, next, ok = readU64(key, next); !ok || next != len(key) { + return p, false + } + p.bucket = string(bucketRaw) + p.object = string(objectRaw) + p.uploadID = string(uploadIDRaw) + return p, true +} diff --git a/internal/s3keys/keys.go b/internal/s3keys/keys.go index 682105219..651f5834d 100644 --- a/internal/s3keys/keys.go +++ b/internal/s3keys/keys.go @@ -12,6 +12,8 @@ const ( UploadMetaPrefix = "!s3|upload|meta|" UploadPartPrefix = "!s3|upload|part|" BlobPrefix = "!s3|blob|" + ChunkRefPrefix = "!s3|chunkref|" + ChunkBlobPrefix = "!s3|chunkblob|" GCUploadPrefix = "!s3|gc|upload|" RoutePrefix = "!s3route|" @@ -33,6 +35,8 @@ var ( uploadMetaPrefixBytes = []byte(UploadMetaPrefix) uploadPartPrefixBytes = []byte(UploadPartPrefix) blobPrefixBytes = []byte(BlobPrefix) + chunkRefPrefixBytes = []byte(ChunkRefPrefix) + chunkBlobPrefixBytes = []byte(ChunkBlobPrefix) gcUploadPrefixBytes = []byte(GCUploadPrefix) routePrefixBytes = []byte(RoutePrefix) ) @@ -96,6 +100,10 @@ func BlobKey(bucket string, generation uint64, object string, uploadID string, p return buildObjectKey(blobPrefixBytes, bucket, generation, object, uploadID, partNo, chunkNo) } +func ChunkRefKey(bucket string, generation uint64, object string, uploadID string, partNo uint64, chunkNo uint64) []byte { + return buildObjectKey(chunkRefPrefixBytes, bucket, generation, object, uploadID, partNo, chunkNo) +} + // VersionedBlobKey returns the blob key for a specific part attempt identified by // partVersion (typically the part's commit timestamp). When partVersion is 0 the // result is identical to BlobKey, preserving backward compatibility with data @@ -152,6 +160,18 @@ func BlobPrefixForUpload(bucket string, generation uint64, object string, upload return out } +// ChunkRefPrefixForUpload returns the key prefix that covers all offload +// chunk references for a specific multipart upload. +func ChunkRefPrefixForUpload(bucket string, generation uint64, object string, uploadID string) []byte { + out := make([]byte, 0, len(ChunkRefPrefix)+len(bucket)+len(object)+len(uploadID)+u64Bytes+buildObjectExtraBytes) + out = append(out, chunkRefPrefixBytes...) + out = append(out, EncodeSegment([]byte(bucket))...) + out = appendU64(out, generation) + out = append(out, EncodeSegment([]byte(object))...) + out = append(out, EncodeSegment([]byte(uploadID))...) + return out +} + // ParseUploadPartKey extracts bucket, generation, object, uploadID, and partNo from a part descriptor key. func ParseUploadPartKey(key []byte) (bucket string, generation uint64, object string, uploadID string, partNo uint64, ok bool) { if !bytes.HasPrefix(key, uploadPartPrefixBytes) { @@ -194,8 +214,8 @@ func ObjectManifestPrefixForBucket(bucket string, generation uint64) []byte { } // UploadMetaPrefixForBucket / UploadPartPrefixForBucket / -// BlobPrefixForBucket / GCUploadPrefixForBucket / RoutePrefixForBucket -// each isolate to a single bucket+generation tuple. Used by +// BlobPrefixForBucket / ChunkRefPrefixForBucket / GCUploadPrefixForBucket / +// RoutePrefixForBucket each isolate to a single bucket+generation tuple. Used by // AdminDeleteBucket's DEL_PREFIX safety net (design doc // 2026_04_28_implemented_admin_delete_bucket_safety_net.md): the bucket- // must-be-empty empty-probe is racy against concurrent PutObject, so @@ -219,6 +239,10 @@ func BlobPrefixForBucket(bucket string, generation uint64) []byte { return bucketScopedPrefix(blobPrefixBytes, bucket, generation) } +func ChunkRefPrefixForBucket(bucket string, generation uint64) []byte { + return bucketScopedPrefix(chunkRefPrefixBytes, bucket, generation) +} + func GCUploadPrefixForBucket(bucket string, generation uint64) []byte { return bucketScopedPrefix(gcUploadPrefixBytes, bucket, generation) } @@ -437,6 +461,8 @@ func objectScopedPrefix(key []byte) []byte { return uploadPartPrefixBytes case bytes.HasPrefix(key, blobPrefixBytes): return blobPrefixBytes + case bytes.HasPrefix(key, chunkRefPrefixBytes): + return chunkRefPrefixBytes case bytes.HasPrefix(key, gcUploadPrefixBytes): return gcUploadPrefixBytes default: diff --git a/internal/s3keys/keys_test.go b/internal/s3keys/keys_test.go index e5e5fca5a..56664138c 100644 --- a/internal/s3keys/keys_test.go +++ b/internal/s3keys/keys_test.go @@ -2,6 +2,7 @@ package s3keys import ( "bytes" + "crypto/sha256" "testing" "github.com/stretchr/testify/require" @@ -275,11 +276,79 @@ func TestBlobPrefixForUpload_IsPrefixOfBlobKeys(t *testing.T) { require.False(t, bytes.HasPrefix(otherKey, prefix)) } +func TestChunkRefKey_RoundTripAndRouteKey(t *testing.T) { + t.Parallel() + + bucket := string([]byte{'b', 0x00, 'k'}) + object := string([]byte{'o', 0x00, '/', 'x'}) + uploadID := string([]byte{'u', 0x00, '1'}) + + key := ChunkRefKey(bucket, 11, object, uploadID, 7, 3) + gotBucket, gotGen, gotObject, gotUpload, gotPart, gotChunk, ok := ParseChunkRefKey(key) + require.True(t, ok) + require.Equal(t, bucket, gotBucket) + require.Equal(t, uint64(11), gotGen) + require.Equal(t, object, gotObject) + require.Equal(t, uploadID, gotUpload) + require.Equal(t, uint64(7), gotPart) + require.Equal(t, uint64(3), gotChunk) + require.Equal(t, RouteKey(bucket, 11, object), ExtractRouteKey(key)) +} + +func TestChunkBlobKey_RoundTrip(t *testing.T) { + t.Parallel() + + sum := sha256.Sum256([]byte("chunk data")) + key := ChunkBlobKey(sum) + + got, ok := ParseChunkBlobKey(key) + require.True(t, ok) + require.Equal(t, sum, got) + require.False(t, bytes.HasPrefix(key, []byte(BlobPrefix))) +} + +func TestChunkRefValue_RoundTrip(t *testing.T) { + t.Parallel() + + sum := sha256.Sum256([]byte("chunk data")) + value, err := EncodeChunkRefValue(ChunkRefValue{ + ContentSHA256: sum, + Size: 1234, + SourcePeer: "node-a", + }) + require.NoError(t, err) + + got, ok := DecodeChunkRefValue(value) + require.True(t, ok) + require.Equal(t, sum, got.ContentSHA256) + require.Equal(t, uint64(1234), got.Size) + require.Equal(t, "node-a", got.SourcePeer) +} + +func TestChunkRefValue_RejectsMalformedValues(t *testing.T) { + t.Parallel() + + sum := sha256.Sum256([]byte("chunk data")) + value, err := EncodeChunkRefValue(ChunkRefValue{ContentSHA256: sum, Size: 1, SourcePeer: "n1"}) + require.NoError(t, err) + + for _, malformed := range [][]byte{ + nil, + value[:1], + append([]byte{0xff}, value[1:]...), + append([]byte{}, value[:len(value)-1]...), + append(append([]byte{}, value...), 'x'), + } { + _, ok := DecodeChunkRefValue(malformed) + require.False(t, ok, "value %x should be rejected", malformed) + } +} + // TestPerBucketPrefixes_IsolateByBucketAndGeneration covers the // new *PrefixForBucket helpers used by AdminDeleteBucket's // DEL_PREFIX safety net (design doc // 2026_04_28_implemented_admin_delete_bucket_safety_net.md). For each -// of the six per-bucket key families, the test pins three +// of the seven per-bucket key families, the test pins three // invariants: // // 1. Every key constructed under (bucket, gen) starts with the @@ -341,6 +410,13 @@ func TestPerBucketPrefixes_IsolateByBucketAndGeneration(t *testing.T) { keyOther: BlobKey(other, gen, object, uploadID, 1, 0), keyOtherGen: BlobKey(bucket, otherGen, object, uploadID, 1, 0), }, + { + name: "chunk_ref", + prefix: ChunkRefPrefixForBucket(bucket, gen), + key: ChunkRefKey(bucket, gen, object, uploadID, 1, 0), + keyOther: ChunkRefKey(other, gen, object, uploadID, 1, 0), + keyOtherGen: ChunkRefKey(bucket, otherGen, object, uploadID, 1, 0), + }, { name: "gc_upload", prefix: GCUploadPrefixForBucket(bucket, gen), diff --git a/main.go b/main.go index 530dbf4b7..c7a1bae3a 100644 --- a/main.go +++ b/main.go @@ -1874,6 +1874,7 @@ func configureAdminService( token = loaded } srv := adapter.NewAdminServer(self, members) + srv.SetCapability(adapter.S3BlobOffloadCapabilityName, adapter.S3BlobOffloadLocalCapability()) unary, stream := adapter.AdminTokenAuth(token) var icept adminGRPCInterceptors if unary != nil { @@ -2569,6 +2570,7 @@ func (r *runtimeServerRunner) prepareAdminForwardServers() error { r.s3Address, r.shardStore, r.coordinate, r.leaderS3, r.s3Region, r.s3CredsFile, r.s3PathStyleOnly, r.readTracker, r.metricsRegistry.S3PutAdmissionObserver(), + r.metricsRegistry.S3BlobOffloadObserver(), ) if err != nil { return err diff --git a/main_s3.go b/main_s3.go index fac553d55..f23b58a3a 100644 --- a/main_s3.go +++ b/main_s3.go @@ -33,10 +33,12 @@ func startS3Server( pathStyleOnly bool, readTracker *kv.ActiveTimestampTracker, putAdmissionObserver adapter.S3PutAdmissionObserver, + blobOffloadObserver adapter.S3BlobOffloadObserver, ) (*adapter.S3Server, error) { s3Server, _, err := prepareS3Server( ctx, lc, s3Addr, shardStore, coordinate, leaderS3, region, credentialsFile, pathStyleOnly, readTracker, putAdmissionObserver, + blobOffloadObserver, ) if err != nil { return nil, err @@ -57,10 +59,11 @@ func prepareS3Server( pathStyleOnly bool, readTracker *kv.ActiveTimestampTracker, putAdmissionObserver adapter.S3PutAdmissionObserver, + blobOffloadObserver adapter.S3BlobOffloadObserver, ) (*adapter.S3Server, net.Listener, error) { s3Server, err := newS3Server( s3Addr, shardStore, coordinate, leaderS3, region, credentialsFile, - pathStyleOnly, readTracker, putAdmissionObserver, + pathStyleOnly, readTracker, putAdmissionObserver, blobOffloadObserver, ) if err != nil { return nil, nil, err @@ -82,6 +85,7 @@ func newS3Server( pathStyleOnly bool, readTracker *kv.ActiveTimestampTracker, putAdmissionObserver adapter.S3PutAdmissionObserver, + blobOffloadObserver adapter.S3BlobOffloadObserver, ) (*adapter.S3Server, error) { s3Addr = strings.TrimSpace(s3Addr) if s3Addr == "" { @@ -108,6 +112,7 @@ func newS3Server( adapter.WithS3StaticCredentials(staticCreds), adapter.WithS3ActiveTimestampTracker(readTracker), adapter.WithS3PutAdmissionObserver(putAdmissionObserver), + adapter.WithS3BlobOffloadObserver(blobOffloadObserver), ) return s3Server, nil } diff --git a/main_s3_test.go b/main_s3_test.go index 03e422bf6..162091112 100644 --- a/main_s3_test.go +++ b/main_s3_test.go @@ -13,14 +13,14 @@ import ( func TestStartS3ServerRejectsVirtualHostedStyleConfig(t *testing.T) { eg, ctx := errgroup.WithContext(context.Background()) - srv, err := startS3Server(ctx, &net.ListenConfig{}, eg, "localhost:9000", nil, nil, nil, "us-east-1", "", false, nil, nil) + srv, err := startS3Server(ctx, &net.ListenConfig{}, eg, "localhost:9000", nil, nil, nil, "us-east-1", "", false, nil, nil, nil) require.ErrorContains(t, err, "virtual-hosted style S3 requests are not implemented") require.Nil(t, srv) } func TestStartS3ServerAllowsEmptyAddress(t *testing.T) { eg, ctx := errgroup.WithContext(context.Background()) - srv, err := startS3Server(ctx, &net.ListenConfig{}, eg, "", nil, nil, nil, "us-east-1", "", false, nil, nil) + srv, err := startS3Server(ctx, &net.ListenConfig{}, eg, "", nil, nil, nil, "us-east-1", "", false, nil, nil, nil) require.NoError(t, err) require.Nil(t, srv) } diff --git a/monitoring/registry.go b/monitoring/registry.go index ac934ecb7..abfb4bad4 100644 --- a/monitoring/registry.go +++ b/monitoring/registry.go @@ -230,6 +230,17 @@ func (r *Registry) S3PutAdmissionObserver() S3PutAdmissionObserver { return r.s3 } +// S3BlobOffloadObserver returns the S3 blob-offload metrics observer backed by +// this registry. The offload data path is rollout-gated in the adapter; this +// observer records both fallback decisions and future chunkblob durability +// outcomes. +func (r *Registry) S3BlobOffloadObserver() S3BlobOffloadObserver { + if r == nil || r.s3 == nil { + return nil + } + return r.s3 +} + // WriteConflictCollector returns a collector that polls each MVCC // store's per-(kind, key_prefix) OCC conflict counters and mirrors // them into the elastickv_store_write_conflict_total Prometheus diff --git a/monitoring/s3.go b/monitoring/s3.go index 363f24634..cdc74def7 100644 --- a/monitoring/s3.go +++ b/monitoring/s3.go @@ -14,6 +14,16 @@ const ( s3PutAdmissionProtocolFixed = "fixed-length" s3PutAdmissionProtocolChunked = "chunked" s3PutAdmissionProtocolUnknown = "unknown" + + s3BlobOffloadModeLegacy = "legacy" + s3BlobOffloadModeOffload = "offload" + s3BlobOffloadModeUnknown = "unknown" + + s3BlobOffloadReasonFlagDisabled = "flag_disabled" + s3BlobOffloadReasonCapabilityMissing = "capability_missing" + s3BlobOffloadReasonDataPathDisabled = "data_path_disabled" + s3BlobOffloadReasonEnabled = "enabled" + s3BlobOffloadReasonUnknown = "unknown" ) type S3PutAdmissionObserver interface { @@ -22,10 +32,21 @@ type S3PutAdmissionObserver interface { ObserveS3PutAdmissionWait(stage, protocol string, duration time.Duration) } +type S3BlobOffloadObserver interface { + ObserveS3BlobOffloadDecision(mode, reason string) + ObserveS3ChunkBlobReplicationDegraded() + ObserveS3ChunkBlobSHAMismatch() + ObserveS3ChunkBlobUnrecoverable() +} + type S3Metrics struct { - putAdmissionInflightBytes prometheus.Gauge - putAdmissionRejections *prometheus.CounterVec - putAdmissionWait *prometheus.HistogramVec + putAdmissionInflightBytes prometheus.Gauge + putAdmissionRejections *prometheus.CounterVec + putAdmissionWait *prometheus.HistogramVec + blobOffloadWriteDecisions *prometheus.CounterVec + chunkBlobReplicationDegraded prometheus.Counter + chunkBlobSHAMismatches prometheus.Counter + chunkBlobUnrecoverableReads prometheus.Counter } func newS3Metrics(registerer prometheus.Registerer) *S3Metrics { @@ -51,11 +72,40 @@ func newS3Metrics(registerer prometheus.Registerer) *S3Metrics { }, []string{"stage", "protocol"}, ), + blobOffloadWriteDecisions: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "elastickv_s3_blob_offload_write_decisions_total", + Help: "Total S3 write-path blob offload decisions by selected mode and fallback reason.", + }, + []string{"mode", "reason"}, + ), + chunkBlobReplicationDegraded: prometheus.NewCounter( + prometheus.CounterOpts{ + Name: "elastickv_s3_chunkblob_replication_degraded_total", + Help: "Total S3 chunkblob writes that could not reach the configured replication target but remained recoverable.", + }, + ), + chunkBlobSHAMismatches: prometheus.NewCounter( + prometheus.CounterOpts{ + Name: "elastickv_s3_chunkblob_sha_mismatch_total", + Help: "Total S3 chunkblob reads or writes rejected because the stored payload did not match its content SHA-256.", + }, + ), + chunkBlobUnrecoverableReads: prometheus.NewCounter( + prometheus.CounterOpts{ + Name: "elastickv_s3_chunkblob_unrecoverable_total", + Help: "Total S3 chunkblob reads that could not recover the referenced content from any available peer.", + }, + ), } registerer.MustRegister( m.putAdmissionInflightBytes, m.putAdmissionRejections, m.putAdmissionWait, + m.blobOffloadWriteDecisions, + m.chunkBlobReplicationDegraded, + m.chunkBlobSHAMismatches, + m.chunkBlobUnrecoverableReads, ) return m } @@ -90,6 +140,37 @@ func (m *S3Metrics) ObserveS3PutAdmissionWait(stage, protocol string, duration t ).Observe(duration.Seconds()) } +func (m *S3Metrics) ObserveS3BlobOffloadDecision(mode, reason string) { + if m == nil { + return + } + m.blobOffloadWriteDecisions.WithLabelValues( + normalizeS3BlobOffloadMode(mode), + normalizeS3BlobOffloadReason(reason), + ).Inc() +} + +func (m *S3Metrics) ObserveS3ChunkBlobReplicationDegraded() { + if m == nil { + return + } + m.chunkBlobReplicationDegraded.Inc() +} + +func (m *S3Metrics) ObserveS3ChunkBlobSHAMismatch() { + if m == nil { + return + } + m.chunkBlobSHAMismatches.Inc() +} + +func (m *S3Metrics) ObserveS3ChunkBlobUnrecoverable() { + if m == nil { + return + } + m.chunkBlobUnrecoverableReads.Inc() +} + func normalizeS3PutAdmissionStage(stage string) string { switch stage { case s3PutAdmissionStagePrereserve, s3PutAdmissionStagePerBatch: @@ -99,6 +180,27 @@ func normalizeS3PutAdmissionStage(stage string) string { } } +func normalizeS3BlobOffloadMode(mode string) string { + switch mode { + case s3BlobOffloadModeLegacy, s3BlobOffloadModeOffload: + return mode + default: + return s3BlobOffloadModeUnknown + } +} + +func normalizeS3BlobOffloadReason(reason string) string { + switch reason { + case s3BlobOffloadReasonFlagDisabled, + s3BlobOffloadReasonCapabilityMissing, + s3BlobOffloadReasonDataPathDisabled, + s3BlobOffloadReasonEnabled: + return reason + default: + return s3BlobOffloadReasonUnknown + } +} + func normalizeS3PutAdmissionProtocol(protocol string) string { switch protocol { case s3PutAdmissionProtocolFixed, s3PutAdmissionProtocolChunked: diff --git a/monitoring/s3_test.go b/monitoring/s3_test.go index 8029a07b1..010ca560c 100644 --- a/monitoring/s3_test.go +++ b/monitoring/s3_test.go @@ -39,11 +39,49 @@ elastickv_s3_put_admission_rejections_total{protocol="fixed-length",stage="prere require.Equal(t, 1, testutil.CollectAndCount(metrics.putAdmissionWait)) } +func TestS3BlobOffloadMetricsObserve(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + metrics := newS3Metrics(reg) + + metrics.ObserveS3BlobOffloadDecision(s3BlobOffloadModeLegacy, s3BlobOffloadReasonCapabilityMissing) + metrics.ObserveS3BlobOffloadDecision("bogus", "bogus") + metrics.ObserveS3ChunkBlobReplicationDegraded() + metrics.ObserveS3ChunkBlobSHAMismatch() + metrics.ObserveS3ChunkBlobUnrecoverable() + + err := testutil.GatherAndCompare( + reg, + strings.NewReader(` +# HELP elastickv_s3_blob_offload_write_decisions_total Total S3 write-path blob offload decisions by selected mode and fallback reason. +# TYPE elastickv_s3_blob_offload_write_decisions_total counter +elastickv_s3_blob_offload_write_decisions_total{mode="legacy",reason="capability_missing"} 1 +elastickv_s3_blob_offload_write_decisions_total{mode="unknown",reason="unknown"} 1 +# HELP elastickv_s3_chunkblob_replication_degraded_total Total S3 chunkblob writes that could not reach the configured replication target but remained recoverable. +# TYPE elastickv_s3_chunkblob_replication_degraded_total counter +elastickv_s3_chunkblob_replication_degraded_total 1 +# HELP elastickv_s3_chunkblob_sha_mismatch_total Total S3 chunkblob reads or writes rejected because the stored payload did not match its content SHA-256. +# TYPE elastickv_s3_chunkblob_sha_mismatch_total counter +elastickv_s3_chunkblob_sha_mismatch_total 1 +# HELP elastickv_s3_chunkblob_unrecoverable_total Total S3 chunkblob reads that could not recover the referenced content from any available peer. +# TYPE elastickv_s3_chunkblob_unrecoverable_total counter +elastickv_s3_chunkblob_unrecoverable_total 1 +`), + "elastickv_s3_blob_offload_write_decisions_total", + "elastickv_s3_chunkblob_replication_degraded_total", + "elastickv_s3_chunkblob_sha_mismatch_total", + "elastickv_s3_chunkblob_unrecoverable_total", + ) + require.NoError(t, err) +} + func TestRegistryReturnsS3PutAdmissionObserver(t *testing.T) { t.Parallel() registry := NewRegistry("n1", "127.0.0.1:0") require.NotNil(t, registry.S3PutAdmissionObserver()) + require.NotNil(t, registry.S3BlobOffloadObserver()) } func TestRegistryS3PutAdmissionObserverNilWhenMetricsMissing(t *testing.T) { @@ -51,4 +89,5 @@ func TestRegistryS3PutAdmissionObserverNilWhenMetricsMissing(t *testing.T) { registry := &Registry{} require.Nil(t, registry.S3PutAdmissionObserver()) + require.Nil(t, registry.S3BlobOffloadObserver()) } diff --git a/proto/admin.pb.go b/proto/admin.pb.go index ace6b3462..e6b087967 100644 --- a/proto/admin.pb.go +++ b/proto/admin.pb.go @@ -282,6 +282,7 @@ type GetClusterOverviewResponse struct { Members []*NodeIdentity `protobuf:"bytes,2,rep,name=members,proto3" json:"members,omitempty"` GroupLeaders []*GroupLeader `protobuf:"bytes,3,rep,name=group_leaders,json=groupLeaders,proto3" json:"group_leaders,omitempty"` AggregateQps uint64 `protobuf:"varint,4,opt,name=aggregate_qps,json=aggregateQps,proto3" json:"aggregate_qps,omitempty"` + Capabilities map[string]bool `protobuf:"bytes,5,rep,name=capabilities,proto3" json:"capabilities,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -344,6 +345,13 @@ func (x *GetClusterOverviewResponse) GetAggregateQps() uint64 { return 0 } +func (x *GetClusterOverviewResponse) GetCapabilities() map[string]bool { + if x != nil { + return x.Capabilities + } + return nil +} + type RaftGroupState struct { state protoimpl.MessageState `protogen:"open.v1"` RaftGroupId uint64 `protobuf:"varint,1,opt,name=raft_group_id,json=raftGroupId,proto3" json:"raft_group_id,omitempty"` @@ -1366,12 +1374,16 @@ const file_admin_proto_rawDesc = "" + "\x0eleader_node_id\x18\x02 \x01(\tR\fleaderNodeId\x12\x1f\n" + "\vleader_term\x18\x03 \x01(\x04R\n" + "leaderTerm\"\x1b\n" + - "\x19GetClusterOverviewRequest\"\xc0\x01\n" + + "\x19GetClusterOverviewRequest\"\xd4\x02\n" + "\x1aGetClusterOverviewResponse\x12!\n" + "\x04self\x18\x01 \x01(\v2\r.NodeIdentityR\x04self\x12'\n" + "\amembers\x18\x02 \x03(\v2\r.NodeIdentityR\amembers\x121\n" + "\rgroup_leaders\x18\x03 \x03(\v2\f.GroupLeaderR\fgroupLeaders\x12#\n" + - "\raggregate_qps\x18\x04 \x01(\x04R\faggregateQps\"\xf4\x01\n" + + "\raggregate_qps\x18\x04 \x01(\x04R\faggregateQps\x12Q\n" + + "\fcapabilities\x18\x05 \x03(\v2-.GetClusterOverviewResponse.CapabilitiesEntryR\fcapabilities\x1a?\n" + + "\x11CapabilitiesEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + + "\x05value\x18\x02 \x01(\bR\x05value:\x028\x01\"\xf4\x01\n" + "\x0eRaftGroupState\x12\"\n" + "\rraft_group_id\x18\x01 \x01(\x04R\vraftGroupId\x12$\n" + "\x0eleader_node_id\x18\x02 \x01(\tR\fleaderNodeId\x12\x1f\n" + @@ -1485,7 +1497,7 @@ func file_admin_proto_rawDescGZIP() []byte { } var file_admin_proto_enumTypes = make([]protoimpl.EnumInfo, 2) -var file_admin_proto_msgTypes = make([]protoimpl.MessageInfo, 19) +var file_admin_proto_msgTypes = make([]protoimpl.MessageInfo, 20) var file_admin_proto_goTypes = []any{ (KeyVizSeries)(0), // 0: KeyVizSeries (SampleRole)(0), // 1: SampleRole @@ -1508,39 +1520,41 @@ var file_admin_proto_goTypes = []any{ (*StreamEventsEvent)(nil), // 18: StreamEventsEvent (*RouteTransition)(nil), // 19: RouteTransition (*KeyVizColumn)(nil), // 20: KeyVizColumn + nil, // 21: GetClusterOverviewResponse.CapabilitiesEntry } var file_admin_proto_depIdxs = []int32{ 2, // 0: GetClusterOverviewResponse.self:type_name -> NodeIdentity 2, // 1: GetClusterOverviewResponse.members:type_name -> NodeIdentity 3, // 2: GetClusterOverviewResponse.group_leaders:type_name -> GroupLeader - 6, // 3: GetRaftGroupsResponse.groups:type_name -> RaftGroupState - 9, // 4: GetAdapterSummaryResponse.summaries:type_name -> AdapterSummary - 1, // 5: KeyVizRow.sample_roles:type_name -> SampleRole - 0, // 6: GetKeyVizMatrixRequest.series:type_name -> KeyVizSeries - 12, // 7: GetKeyVizMatrixResponse.rows:type_name -> KeyVizRow - 12, // 8: GetRouteDetailResponse.row:type_name -> KeyVizRow - 9, // 9: GetRouteDetailResponse.per_adapter:type_name -> AdapterSummary - 19, // 10: StreamEventsEvent.route_transition:type_name -> RouteTransition - 20, // 11: StreamEventsEvent.keyviz_column:type_name -> KeyVizColumn - 0, // 12: KeyVizColumn.series:type_name -> KeyVizSeries - 12, // 13: KeyVizColumn.rows:type_name -> KeyVizRow - 4, // 14: Admin.GetClusterOverview:input_type -> GetClusterOverviewRequest - 7, // 15: Admin.GetRaftGroups:input_type -> GetRaftGroupsRequest - 10, // 16: Admin.GetAdapterSummary:input_type -> GetAdapterSummaryRequest - 13, // 17: Admin.GetKeyVizMatrix:input_type -> GetKeyVizMatrixRequest - 15, // 18: Admin.GetRouteDetail:input_type -> GetRouteDetailRequest - 17, // 19: Admin.StreamEvents:input_type -> StreamEventsRequest - 5, // 20: Admin.GetClusterOverview:output_type -> GetClusterOverviewResponse - 8, // 21: Admin.GetRaftGroups:output_type -> GetRaftGroupsResponse - 11, // 22: Admin.GetAdapterSummary:output_type -> GetAdapterSummaryResponse - 14, // 23: Admin.GetKeyVizMatrix:output_type -> GetKeyVizMatrixResponse - 16, // 24: Admin.GetRouteDetail:output_type -> GetRouteDetailResponse - 18, // 25: Admin.StreamEvents:output_type -> StreamEventsEvent - 20, // [20:26] is the sub-list for method output_type - 14, // [14:20] is the sub-list for method input_type - 14, // [14:14] is the sub-list for extension type_name - 14, // [14:14] is the sub-list for extension extendee - 0, // [0:14] is the sub-list for field type_name + 21, // 3: GetClusterOverviewResponse.capabilities:type_name -> GetClusterOverviewResponse.CapabilitiesEntry + 6, // 4: GetRaftGroupsResponse.groups:type_name -> RaftGroupState + 9, // 5: GetAdapterSummaryResponse.summaries:type_name -> AdapterSummary + 1, // 6: KeyVizRow.sample_roles:type_name -> SampleRole + 0, // 7: GetKeyVizMatrixRequest.series:type_name -> KeyVizSeries + 12, // 8: GetKeyVizMatrixResponse.rows:type_name -> KeyVizRow + 12, // 9: GetRouteDetailResponse.row:type_name -> KeyVizRow + 9, // 10: GetRouteDetailResponse.per_adapter:type_name -> AdapterSummary + 19, // 11: StreamEventsEvent.route_transition:type_name -> RouteTransition + 20, // 12: StreamEventsEvent.keyviz_column:type_name -> KeyVizColumn + 0, // 13: KeyVizColumn.series:type_name -> KeyVizSeries + 12, // 14: KeyVizColumn.rows:type_name -> KeyVizRow + 4, // 15: Admin.GetClusterOverview:input_type -> GetClusterOverviewRequest + 7, // 16: Admin.GetRaftGroups:input_type -> GetRaftGroupsRequest + 10, // 17: Admin.GetAdapterSummary:input_type -> GetAdapterSummaryRequest + 13, // 18: Admin.GetKeyVizMatrix:input_type -> GetKeyVizMatrixRequest + 15, // 19: Admin.GetRouteDetail:input_type -> GetRouteDetailRequest + 17, // 20: Admin.StreamEvents:input_type -> StreamEventsRequest + 5, // 21: Admin.GetClusterOverview:output_type -> GetClusterOverviewResponse + 8, // 22: Admin.GetRaftGroups:output_type -> GetRaftGroupsResponse + 11, // 23: Admin.GetAdapterSummary:output_type -> GetAdapterSummaryResponse + 14, // 24: Admin.GetKeyVizMatrix:output_type -> GetKeyVizMatrixResponse + 16, // 25: Admin.GetRouteDetail:output_type -> GetRouteDetailResponse + 18, // 26: Admin.StreamEvents:output_type -> StreamEventsEvent + 21, // [21:27] is the sub-list for method output_type + 15, // [15:21] is the sub-list for method input_type + 15, // [15:15] is the sub-list for extension type_name + 15, // [15:15] is the sub-list for extension extendee + 0, // [0:15] is the sub-list for field type_name } func init() { file_admin_proto_init() } @@ -1558,7 +1572,7 @@ func file_admin_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_admin_proto_rawDesc), len(file_admin_proto_rawDesc)), NumEnums: 2, - NumMessages: 19, + NumMessages: 20, NumExtensions: 0, NumServices: 1, }, diff --git a/proto/admin.proto b/proto/admin.proto index aae54e45a..96d111020 100644 --- a/proto/admin.proto +++ b/proto/admin.proto @@ -33,6 +33,7 @@ message GetClusterOverviewResponse { repeated NodeIdentity members = 2; repeated GroupLeader group_leaders = 3; uint64 aggregate_qps = 4; + map capabilities = 5; } message RaftGroupState {