diff --git a/adapter/admin_grpc.go b/adapter/admin_grpc.go index a8cb10a8a..eb1185103 100644 --- a/adapter/admin_grpc.go +++ b/adapter/admin_grpc.go @@ -8,6 +8,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/bootjp/elastickv/internal/raftengine" @@ -41,6 +42,7 @@ type KeyVizSampler interface { type AdminGroup interface { Status() raftengine.Status Configuration(ctx context.Context) (raftengine.Configuration, error) + SnapshotEvery() uint64 } // NodeIdentity is the value form of the protobuf NodeIdentity message used for @@ -55,6 +57,54 @@ func (n NodeIdentity) toProto() *pb.NodeIdentity { return &pb.NodeIdentity{NodeId: n.NodeID, GrpcAddress: n.GRPCAddress} } +// LeaderVersionProbe fetches the Admin service version for a peer address. +// Implementations must honor ctx for the 500ms async GetRaftGroups probe +// budget and any auth metadata copied from the inbound Admin request. +type LeaderVersionProbe func(ctx context.Context, grpcAddress string) (string, error) + +// AdminOption adjusts optional AdminServer behavior without changing existing +// test construction call sites. +type AdminOption func(*AdminServer) + +func WithAdminNodeVersion(version string) AdminOption { + return func(s *AdminServer) { + s.nodeVersion = version + } +} + +func WithAdminLeaderVersionProbe(probe LeaderVersionProbe) AdminOption { + return func(s *AdminServer) { + s.leaderVersionProbe = probe + } +} + +func WithAdminLeaderVersionProbeTimeout(timeout time.Duration) AdminOption { + return func(s *AdminServer) { + if timeout > 0 { + s.leaderVersionProbeTimeout = timeout + } + } +} + +func WithAdminLeaderVersionCacheTTL(ttl time.Duration) AdminOption { + return func(s *AdminServer) { + if ttl > 0 { + s.leaderVersionCacheTTL = ttl + } + } +} + +type versionCacheEntry struct { + version string + fetchedAt time.Time + probeID uint64 +} + +const ( + defaultAdminLeaderVersionProbeTimeout = 500 * time.Millisecond + defaultAdminLeaderVersionCacheTTL = 10 * time.Second +) + // AdminServer implements the node-side Admin gRPC service described in // docs/admin_ui_key_visualizer_design.md §4 (Layer A). Phase 0 only implements // GetClusterOverview and GetRaftGroups; remaining RPCs return Unimplemented so @@ -78,6 +128,13 @@ type AdminServer struct { // pairs atomically with concurrent RPC reads. sampler KeyVizSampler + nodeVersion string + leaderVersionProbe LeaderVersionProbe + leaderVersionProbeTimeout time.Duration + leaderVersionCacheTTL time.Duration + leaderVersionProbeSeq atomic.Uint64 + versionCache sync.Map + pb.UnimplementedAdminServer } @@ -86,14 +143,22 @@ type AdminServer struct { // snapshot shipped to the admin binary; callers that already have a membership // source may pass nil and let the admin binary's fan-out layer discover peers // by other means. -func NewAdminServer(self NodeIdentity, members []NodeIdentity) *AdminServer { +func NewAdminServer(self NodeIdentity, members []NodeIdentity, opts ...AdminOption) *AdminServer { cloned := append([]NodeIdentity(nil), members...) - return &AdminServer{ - self: self, - members: cloned, - groups: make(map[uint64]AdminGroup), - now: time.Now, + srv := &AdminServer{ + self: self, + members: cloned, + groups: make(map[uint64]AdminGroup), + now: time.Now, + leaderVersionProbeTimeout: defaultAdminLeaderVersionProbeTimeout, + leaderVersionCacheTTL: defaultAdminLeaderVersionCacheTTL, + } + for _, opt := range opts { + if opt != nil { + opt(srv) + } } + return srv } // SetClock overrides the clock used by GetRaftGroups, letting tests inject a @@ -380,7 +445,7 @@ func mergeSeedMembers(seeds []NodeIdentity, selfID string, live *liveMembers) { // GetRaftGroups returns per-group state snapshots. Phase 0 wires commit/applied // indices only; per-follower contact and term history land in later phases. func (s *AdminServer) GetRaftGroups( - _ context.Context, + ctx context.Context, _ *pb.GetRaftGroupsRequest, ) (*pb.GetRaftGroupsResponse, error) { s.groupsMu.RLock() @@ -410,11 +475,178 @@ func (s *AdminServer) GetRaftGroups( CommitIndex: st.CommitIndex, AppliedIndex: st.AppliedIndex, LastContactUnixMs: lastContactUnixMs, + LeaderNodeVersion: s.leaderNodeVersion(ctx, st.Leader, now, s.leaderProbeAddressesLocked(st.Leader)), }) } return &pb.GetRaftGroupsResponse{Groups: out}, nil } +func (s *AdminServer) GetNodeVersion( + context.Context, + *pb.GetNodeVersionRequest, +) (*pb.GetNodeVersionResponse, error) { + return &pb.GetNodeVersionResponse{NodeVersion: s.nodeVersion}, nil +} + +func (s *AdminServer) leaderNodeVersion(ctx context.Context, leader raftengine.LeaderInfo, now time.Time, addresses []string) string { + if version, ok := s.localLeaderVersion(leader); ok { + return version + } + key := leaderVersionCacheKey(leader) + if version, ok := s.cachedLeaderVersion(key, now); ok { + return version + } + if s.leaderVersionProbe == nil || len(addresses) == 0 { + return "" + } + version, reservation, reserved := s.reserveLeaderVersionProbe(key, now) + if !reserved { + return version + } + s.probeLeaderVersionAsync(ctx, key, addresses, reservation) + return "" +} + +func (s *AdminServer) leaderProbeAddressesLocked(leader raftengine.LeaderInfo) []string { + seen := map[string]struct{}{} + addresses := make([]string, 0, 1) + add := func(address string) { + address = strings.TrimSpace(address) + if address == "" { + return + } + if _, ok := seen[address]; ok { + return + } + seen[address] = struct{}{} + addresses = append(addresses, address) + } + add(leader.Address) + if leader.ID == "" { + return addresses + } + for _, id := range sortedGroupIDs(s.groups) { + group := s.groups[id] + if group == nil { + continue + } + st := group.Status() + if st.Leader.ID == leader.ID { + add(st.Leader.Address) + } + } + return addresses +} + +func (s *AdminServer) localLeaderVersion(leader raftengine.LeaderInfo) (string, bool) { + if leader.ID == "" && leader.Address == "" { + return "", true + } + if leader.ID == s.self.NodeID || (leader.Address != "" && leader.Address == s.self.GRPCAddress) { + return s.nodeVersion, true + } + return "", false +} + +func leaderVersionCacheKey(leader raftengine.LeaderInfo) string { + if leader.ID != "" && leader.Address != "" { + return leader.ID + "\x00" + leader.Address + } + if leader.ID != "" { + return leader.ID + } + return leader.Address +} + +func (s *AdminServer) cachedLeaderVersion(key string, now time.Time) (string, bool) { + if key == "" || s.leaderVersionCacheTTL <= 0 { + return "", false + } + actual, ok := s.versionCache.Load(key) + if !ok { + return "", false + } + entry, ok := actual.(versionCacheEntry) + if !ok { + s.versionCache.Delete(key) + return "", false + } + if now.Sub(entry.fetchedAt) > s.leaderVersionCacheTTL { + s.versionCache.CompareAndDelete(key, entry) + return "", false + } + return entry.version, true +} + +func (s *AdminServer) reserveLeaderVersionProbe(key string, now time.Time) (string, versionCacheEntry, bool) { + reservation := versionCacheEntry{ + fetchedAt: now, + probeID: s.leaderVersionProbeSeq.Add(1), + } + for { + actual, loaded := s.versionCache.LoadOrStore(key, reservation) + if !loaded { + return "", reservation, true + } + entry, ok := actual.(versionCacheEntry) + switch { + case !ok: + s.versionCache.Store(key, reservation) + return "", reservation, true + case now.Sub(entry.fetchedAt) <= s.leaderVersionCacheTTL: + return entry.version, versionCacheEntry{}, false + default: + if s.versionCache.CompareAndSwap(key, entry, reservation) { + return "", reservation, true + } + } + } +} + +func (s *AdminServer) probeLeaderVersionAsync(ctx context.Context, key string, addresses []string, reservation versionCacheEntry) { + probe := s.leaderVersionProbe + timeout := s.leaderVersionProbeTimeout + now := s.now + candidates := append([]string(nil), addresses...) + attemptTimeout := leaderVersionProbeAttemptTimeout(timeout, len(candidates)) + var md metadata.MD + if incoming, ok := metadata.FromIncomingContext(ctx); ok { + md = incoming.Copy() + } + go func() { + probeCtx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + version := "" + for _, address := range candidates { + attemptCtx, attemptCancel := context.WithTimeout(probeCtx, attemptTimeout) + if md != nil { + attemptCtx = metadata.NewOutgoingContext(attemptCtx, md) + } + probed, err := probe(attemptCtx, address) + attemptCancel() + if err == nil { + version = probed + break + } + if probeCtx.Err() != nil { + break + } + } + s.versionCache.CompareAndSwap(key, reservation, versionCacheEntry{version: version, fetchedAt: now()}) + }() +} + +func leaderVersionProbeAttemptTimeout(total time.Duration, candidates int) time.Duration { + if total <= 0 || candidates <= 1 { + return total + } + perCandidate := total / time.Duration(candidates) + if perCandidate <= 0 { + return total + } + return perCandidate +} + func (s *AdminServer) snapshotLeaders() []*pb.GroupLeader { s.groupsMu.RLock() defer s.groupsMu.RUnlock() diff --git a/adapter/admin_grpc_test.go b/adapter/admin_grpc_test.go index 5724165ec..6357cb9f3 100644 --- a/adapter/admin_grpc_test.go +++ b/adapter/admin_grpc_test.go @@ -8,6 +8,7 @@ import ( "github.com/bootjp/elastickv/internal/raftengine" pb "github.com/bootjp/elastickv/proto" + "github.com/stretchr/testify/require" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" @@ -15,17 +16,19 @@ import ( ) type fakeGroup struct { - leaderID string - term uint64 - commit uint64 - applied uint64 - servers []raftengine.Server - cfgErr error + leaderID string + leaderAddr string + term uint64 + commit uint64 + applied uint64 + servers []raftengine.Server + cfgErr error + snapshotEach uint64 } func (f fakeGroup) Status() raftengine.Status { return raftengine.Status{ - Leader: raftengine.LeaderInfo{ID: f.leaderID}, + Leader: raftengine.LeaderInfo{ID: f.leaderID, Address: f.leaderAddr}, Term: f.term, CommitIndex: f.commit, AppliedIndex: f.applied, @@ -39,6 +42,13 @@ func (f fakeGroup) Configuration(context.Context) (raftengine.Configuration, err return raftengine.Configuration{Servers: append([]raftengine.Server(nil), f.servers...)}, nil } +func (f fakeGroup) SnapshotEvery() uint64 { + if f.snapshotEach == 0 { + return 10_000 + } + return f.snapshotEach +} + func TestGetClusterOverviewReturnsSelfAndLeaders(t *testing.T) { t.Parallel() srv := NewAdminServer( @@ -71,6 +81,23 @@ func TestGetClusterOverviewReturnsSelfAndLeaders(t *testing.T) { } } +func TestGetNodeVersionReturnsConfiguredVersion(t *testing.T) { + t.Parallel() + srv := NewAdminServer( + NodeIdentity{NodeID: "n1"}, + nil, + WithAdminNodeVersion("v1.2.3"), + ) + + resp, err := srv.GetNodeVersion(context.Background(), &pb.GetNodeVersionRequest{}) + if err != nil { + t.Fatalf("GetNodeVersion: %v", err) + } + if resp.NodeVersion != "v1.2.3" { + t.Fatalf("NodeVersion = %q, want v1.2.3", resp.NodeVersion) + } +} + func TestGetRaftGroupsExposesCommitApplied(t *testing.T) { t.Parallel() srv := NewAdminServer(NodeIdentity{NodeID: "n1"}, nil) @@ -99,6 +126,358 @@ func TestGetRaftGroupsExposesCommitApplied(t *testing.T) { } } +func TestGetRaftGroupsReportsLocalLeaderVersion(t *testing.T) { + t.Parallel() + srv := NewAdminServer( + NodeIdentity{NodeID: "n1", GRPCAddress: "10.0.0.11:50051"}, + nil, + WithAdminNodeVersion("v-local"), + ) + srv.RegisterGroup(1, fakeGroup{leaderID: "n1", leaderAddr: "10.0.0.11:50051"}) + + resp, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + if err != nil { + t.Fatalf("GetRaftGroups: %v", err) + } + if got := resp.Groups[0].LeaderNodeVersion; got != "v-local" { + t.Fatalf("LeaderNodeVersion = %q, want v-local", got) + } +} + +func TestGetRaftGroupsLeaderVersionAsync(t *testing.T) { + t.Parallel() + probeStarted := make(chan struct{}, 1) + releaseProbe := make(chan struct{}) + var calls int + var callsMu sync.Mutex + srv := NewAdminServer( + NodeIdentity{NodeID: "n1", GRPCAddress: "10.0.0.11:50051"}, + nil, + WithAdminLeaderVersionProbeTimeout(2*time.Second), + WithAdminLeaderVersionCacheTTL(time.Second), + WithAdminLeaderVersionProbe(func(ctx context.Context, addr string) (string, error) { + if addr != "10.0.0.12:50051" { + t.Errorf("probe addr = %q, want 10.0.0.12:50051", addr) + } + callsMu.Lock() + calls++ + callsMu.Unlock() + probeStarted <- struct{}{} + select { + case <-releaseProbe: + return "v-remote", nil + case <-ctx.Done(): + return "", ctx.Err() + } + }), + ) + srv.RegisterGroup(1, fakeGroup{leaderID: "n2", leaderAddr: "10.0.0.12:50051"}) + + resp, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + if err != nil { + t.Fatalf("GetRaftGroups first: %v", err) + } + requireLeaderVersion(t, resp, "", "first") + select { + case <-probeStarted: + case <-time.After(time.Second): + t.Fatal("leader version probe was not started") + } + second, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + if err != nil { + t.Fatalf("GetRaftGroups second: %v", err) + } + requireLeaderVersion(t, second, "", "second") + close(releaseProbe) + waitForLeaderVersion(t, srv, "v-remote") + callsMu.Lock() + gotCalls := calls + callsMu.Unlock() + if gotCalls != 1 { + t.Fatalf("probe calls = %d, want 1 while cache is fresh", gotCalls) + } +} + +func TestGetRaftGroupsLeaderVersionProbeUsesServerClockForCacheStamp(t *testing.T) { + t.Parallel() + base := time.Unix(1_700_000_000, 0) + var clockMu sync.Mutex + now := base + setNow := func(next time.Time) { + clockMu.Lock() + now = next + clockMu.Unlock() + } + var callsMu sync.Mutex + calls := 0 + srv := NewAdminServer( + NodeIdentity{NodeID: "n1", GRPCAddress: "10.0.0.11:50051"}, + nil, + WithAdminLeaderVersionProbeTimeout(time.Second), + WithAdminLeaderVersionCacheTTL(time.Second), + WithAdminLeaderVersionProbe(func(context.Context, string) (string, error) { + callsMu.Lock() + defer callsMu.Unlock() + calls++ + if calls == 1 { + return "v1", nil + } + return "v2", nil + }), + ) + srv.SetClock(func() time.Time { + clockMu.Lock() + defer clockMu.Unlock() + return now + }) + srv.RegisterGroup(1, fakeGroup{leaderID: "n2", leaderAddr: "10.0.0.12:50051"}) + + _, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + if err != nil { + t.Fatalf("GetRaftGroups first: %v", err) + } + waitForLeaderVersion(t, srv, "v1") + + setNow(base.Add(2 * time.Second)) + resp, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + if err != nil { + t.Fatalf("GetRaftGroups after TTL: %v", err) + } + requireLeaderVersion(t, resp, "", "expired") + waitForLeaderVersion(t, srv, "v2") +} + +func TestGetRaftGroupsLeaderVersionStaleProbeCannotOverwriteFreshCache(t *testing.T) { + t.Parallel() + base := time.Unix(1_700_000_000, 0) + var clockMu sync.Mutex + now := base + setNow := func(next time.Time) { + clockMu.Lock() + now = next + clockMu.Unlock() + } + probeCalls := make(chan chan leaderVersionProbeResult, 2) + srv := NewAdminServer( + NodeIdentity{NodeID: "n1", GRPCAddress: "10.0.0.11:50051"}, + nil, + WithAdminLeaderVersionProbeTimeout(5*time.Second), + WithAdminLeaderVersionCacheTTL(time.Second), + WithAdminLeaderVersionProbe(blockingLeaderVersionProbe(probeCalls)), + ) + srv.SetClock(func() time.Time { + clockMu.Lock() + defer clockMu.Unlock() + return now + }) + srv.RegisterGroup(1, fakeGroup{leaderID: "n2", leaderAddr: "10.0.0.12:50051"}) + + _, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + require.NoError(t, err) + firstProbe := receiveLeaderVersionProbeCall(t, probeCalls, "first") + + setNow(base.Add(2 * time.Second)) + resp, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + require.NoError(t, err) + requireLeaderVersion(t, resp, "", "second") + secondProbe := receiveLeaderVersionProbeCall(t, probeCalls, "second") + secondProbe <- leaderVersionProbeResult{version: "v-fresh"} + waitForLeaderVersion(t, srv, "v-fresh") + + firstProbe <- leaderVersionProbeResult{version: "v-stale"} + requireLeaderVersionNever(t, srv, "v-stale", 100*time.Millisecond) + resp, err = srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + require.NoError(t, err) + requireLeaderVersion(t, resp, "v-fresh", "after stale completion") +} + +type leaderVersionProbeResult struct { + version string + err error +} + +func blockingLeaderVersionProbe(calls chan<- chan leaderVersionProbeResult) LeaderVersionProbe { + return func(ctx context.Context, _ string) (string, error) { + resultCh := make(chan leaderVersionProbeResult) + select { + case calls <- resultCh: + case <-ctx.Done(): + return "", ctx.Err() + } + select { + case result := <-resultCh: + return result.version, result.err + case <-ctx.Done(): + return "", ctx.Err() + } + } +} + +func requireLeaderVersionNever(t *testing.T, srv *AdminServer, unwanted string, waitFor time.Duration) { + t.Helper() + require.Never(t, func() bool { + resp, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + return err == nil && len(resp.Groups) == 1 && resp.Groups[0].LeaderNodeVersion == unwanted + }, waitFor, 10*time.Millisecond) +} + +func receiveLeaderVersionProbeCall( + t *testing.T, + calls <-chan chan leaderVersionProbeResult, + label string, +) chan leaderVersionProbeResult { + t.Helper() + select { + case resultCh := <-calls: + return resultCh + case <-time.After(time.Second): + t.Fatalf("%s leader version probe was not started", label) + return nil + } +} + +func TestGetRaftGroupsLeaderVersionTriesAlternateAddressForSameLeaderID(t *testing.T) { + t.Parallel() + var callsMu sync.Mutex + callsByAddr := map[string]int{} + srv := NewAdminServer( + NodeIdentity{NodeID: "n1", GRPCAddress: "10.0.0.11:50051"}, + nil, + WithAdminLeaderVersionProbeTimeout(time.Second), + WithAdminLeaderVersionCacheTTL(time.Second), + WithAdminLeaderVersionProbe(func(_ context.Context, addr string) (string, error) { + callsMu.Lock() + callsByAddr[addr]++ + callsMu.Unlock() + if addr == "10.0.0.12:50051" { + return "", status.Error(codes.Unavailable, "stale listener") + } + return "v-good", nil + }), + ) + srv.RegisterGroup(1, fakeGroup{leaderID: "n2", leaderAddr: "10.0.0.12:50051"}) + srv.RegisterGroup(2, fakeGroup{leaderID: "n2", leaderAddr: "10.0.0.22:50051"}) + + _, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + if err != nil { + t.Fatalf("GetRaftGroups first: %v", err) + } + require.Eventually(t, func() bool { + resp, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + if err != nil || len(resp.Groups) != 2 { + return false + } + return resp.Groups[0].LeaderNodeVersion == "v-good" && resp.Groups[1].LeaderNodeVersion == "v-good" + }, time.Second, 10*time.Millisecond) + callsMu.Lock() + defer callsMu.Unlock() + if callsByAddr["10.0.0.12:50051"] == 0 || callsByAddr["10.0.0.22:50051"] == 0 { + t.Fatalf("probe calls by address = %v, want both stale and alternate addresses", callsByAddr) + } +} + +func TestGetRaftGroupsLeaderVersionBoundsEachCandidateProbe(t *testing.T) { + t.Parallel() + var callsMu sync.Mutex + callsByAddr := map[string]int{} + srv := NewAdminServer( + NodeIdentity{NodeID: "n1", GRPCAddress: "10.0.0.11:50051"}, + nil, + WithAdminLeaderVersionProbeTimeout(200*time.Millisecond), + WithAdminLeaderVersionCacheTTL(time.Second), + WithAdminLeaderVersionProbe(func(ctx context.Context, addr string) (string, error) { + callsMu.Lock() + callsByAddr[addr]++ + callsMu.Unlock() + if addr == "10.0.0.12:50051" { + <-ctx.Done() + return "", ctx.Err() + } + return "v-good", nil + }), + ) + srv.RegisterGroup(1, fakeGroup{leaderID: "n2", leaderAddr: "10.0.0.12:50051"}) + srv.RegisterGroup(2, fakeGroup{leaderID: "n2", leaderAddr: "10.0.0.22:50051"}) + + _, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + if err != nil { + t.Fatalf("GetRaftGroups first: %v", err) + } + require.Eventually(t, func() bool { + resp, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + if err != nil || len(resp.Groups) != 2 { + return false + } + return resp.Groups[0].LeaderNodeVersion == "v-good" && resp.Groups[1].LeaderNodeVersion == "v-good" + }, time.Second, 10*time.Millisecond) + callsMu.Lock() + defer callsMu.Unlock() + if callsByAddr["10.0.0.12:50051"] == 0 || callsByAddr["10.0.0.22:50051"] == 0 { + t.Fatalf("probe calls by address = %v, want timeout candidate and alternate address", callsByAddr) + } +} + +func requireLeaderVersion(t *testing.T, resp *pb.GetRaftGroupsResponse, want, label string) { + t.Helper() + if len(resp.Groups) != 1 { + t.Fatalf("%s groups = %d, want 1", label, len(resp.Groups)) + } + if got := resp.Groups[0].LeaderNodeVersion; got != want { + t.Fatalf("%s LeaderNodeVersion = %q, want %q", label, got, want) + } +} + +func waitForLeaderVersion(t *testing.T, srv *AdminServer, want string) { + t.Helper() + deadline := time.After(time.Second) + for { + resp, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + if err != nil { + t.Fatalf("GetRaftGroups while waiting for version: %v", err) + } + if len(resp.Groups) == 1 && resp.Groups[0].LeaderNodeVersion == want { + return + } + select { + case <-deadline: + t.Fatalf("leader version cache was not populated with %q", want) + case <-time.After(10 * time.Millisecond): + } + } +} + +func TestGetRaftGroupsLeaderVersionProbePropagatesAuthMetadata(t *testing.T) { + t.Parallel() + gotAuth := make(chan []string, 1) + srv := NewAdminServer( + NodeIdentity{NodeID: "n1", GRPCAddress: "10.0.0.11:50051"}, + nil, + WithAdminLeaderVersionProbeTimeout(time.Second), + WithAdminLeaderVersionProbe(func(ctx context.Context, _ string) (string, error) { + md, _ := metadata.FromOutgoingContext(ctx) + gotAuth <- md.Get("authorization") + return "v-remote", nil + }), + ) + srv.RegisterGroup(1, fakeGroup{leaderID: "n2", leaderAddr: "10.0.0.12:50051"}) + + ctx := metadata.NewIncomingContext( + context.Background(), + metadata.Pairs("authorization", "Bearer admin-token"), + ) + if _, err := srv.GetRaftGroups(ctx, &pb.GetRaftGroupsRequest{}); err != nil { + t.Fatalf("GetRaftGroups: %v", err) + } + select { + case got := <-gotAuth: + if len(got) != 1 || got[0] != "Bearer admin-token" { + t.Fatalf("authorization metadata = %v, want bearer token", got) + } + case <-time.After(time.Second): + t.Fatal("leader version probe did not run") + } +} + // TestGetClusterOverviewUnionsSeedsAndLiveConfig asserts that // GetClusterOverview picks up a node that was added to a Raft group after the // admin server was constructed (scale-out). Without Configuration polling, @@ -327,6 +706,7 @@ func (t testProbeGroup) Configuration(context.Context) (raftengine.Configuration } return raftengine.Configuration{}, nil } +func (t testProbeGroup) SnapshotEvery() uint64 { return 10_000 } // TestGetClusterOverviewKeepsSeedOnPartialConfigFailure asserts that a // bootstrap seed is NOT pruned when one group's Configuration succeeds and @@ -470,6 +850,8 @@ func (f fakeGroupWithContact) Configuration(context.Context) (raftengine.Configu return raftengine.Configuration{}, nil } +func (f fakeGroupWithContact) SnapshotEvery() uint64 { return 10_000 } + // TestGroupOrderingIsStable locks in deterministic ascending-by-RaftGroupId // ordering so admin UIs and diff-based tests do not see rows jump around. func TestGroupOrderingIsStable(t *testing.T) { diff --git a/internal/raftadmin/server_test.go b/internal/raftadmin/server_test.go index 4ec02aba6..f99bba9a7 100644 --- a/internal/raftadmin/server_test.go +++ b/internal/raftadmin/server_test.go @@ -108,6 +108,8 @@ func (f *fakeEngine) Configuration(context.Context) (raftengine.Configuration, e return f.config, nil } +func (f *fakeEngine) SnapshotEvery() uint64 { return 10_000 } + func (f *fakeEngine) CheckServing(context.Context) error { f.mu.Lock() defer f.mu.Unlock() @@ -390,6 +392,8 @@ func (s stateOnlyEngine) Configuration(context.Context) (raftengine.Configuratio return raftengine.Configuration{}, nil } +func (s stateOnlyEngine) SnapshotEvery() uint64 { return 10_000 } + func TestCurrentHealthStatusFallsBackToLocalState(t *testing.T) { t.Parallel() diff --git a/internal/raftengine/engine.go b/internal/raftengine/engine.go index 75ceacc8c..4b67eb458 100644 --- a/internal/raftengine/engine.go +++ b/internal/raftengine/engine.go @@ -235,6 +235,10 @@ type ConfigReader interface { Configuration(ctx context.Context) (Configuration, error) } +type SnapshotReader interface { + SnapshotEvery() uint64 +} + type HealthReader interface { CheckServing(ctx context.Context) error } @@ -303,5 +307,6 @@ type Engine interface { LeaderView StatusReader ConfigReader + SnapshotReader io.Closer } diff --git a/internal/raftengine/etcd/engine.go b/internal/raftengine/etcd/engine.go index df65236b7..46b3ad936 100644 --- a/internal/raftengine/etcd/engine.go +++ b/internal/raftengine/etcd/engine.go @@ -4326,6 +4326,12 @@ func (e *Engine) snapshotThreshold() uint64 { return e.snapshotEvery } +// SnapshotEvery returns the configured FSM-snapshot trigger threshold for +// admin/control-plane code that needs to reason about snapshot headroom. +func (e *Engine) SnapshotEvery() uint64 { + return e.snapshotThreshold() +} + // snapshotEveryFromEnv returns the FSM-snapshot trigger threshold (in applied // raft entries past the last snapshot). Operators can override via // ELASTICKV_RAFT_SNAPSHOT_COUNT; invalid or missing values fall back to diff --git a/internal/raftengine/etcd/wal_purge_test.go b/internal/raftengine/etcd/wal_purge_test.go index 2e2bf8d97..dcb46d685 100644 --- a/internal/raftengine/etcd/wal_purge_test.go +++ b/internal/raftengine/etcd/wal_purge_test.go @@ -213,3 +213,8 @@ func TestSnapshotEveryFromEnv_FallsBackOnInvalid(t *testing.T) { t.Setenv(snapshotEveryEnvVar, "not-a-number") require.Equal(t, uint64(defaultSnapshotEvery), snapshotEveryFromEnv()) } + +func TestEngineSnapshotEveryReadsConfiguredThreshold(t *testing.T) { + require.Equal(t, uint64(defaultSnapshotEvery), (&Engine{}).SnapshotEvery()) + require.Equal(t, uint64(5000), (&Engine{snapshotEvery: 5000}).SnapshotEvery()) +} diff --git a/kv/coordinator_retry_test.go b/kv/coordinator_retry_test.go index 56f0ef361..222d68390 100644 --- a/kv/coordinator_retry_test.go +++ b/kv/coordinator_retry_test.go @@ -40,7 +40,8 @@ func (stubLeaderEngine) Status() raftengine.Status { func (stubLeaderEngine) Configuration(context.Context) (raftengine.Configuration, error) { return raftengine.Configuration{}, nil } -func (stubLeaderEngine) Close() error { return nil } +func (stubLeaderEngine) SnapshotEvery() uint64 { return 10_000 } +func (stubLeaderEngine) Close() error { return nil } // scriptedTransactional returns the error registered in errs for the // 0-indexed call that triggered Commit; calls without a registered diff --git a/kv/leader_proxy_test.go b/kv/leader_proxy_test.go index ea02d6c2a..7f0d17320 100644 --- a/kv/leader_proxy_test.go +++ b/kv/leader_proxy_test.go @@ -63,7 +63,8 @@ func (s *stubFollowerEngine) Status() raftengine.Status { func (s *stubFollowerEngine) Configuration(context.Context) (raftengine.Configuration, error) { return raftengine.Configuration{}, nil } -func (s *stubFollowerEngine) Close() error { return nil } +func (s *stubFollowerEngine) SnapshotEvery() uint64 { return 10_000 } +func (s *stubFollowerEngine) Close() error { return nil } func TestLeaderProxy_CommitLocalWhenLeader(t *testing.T) { t.Parallel() @@ -186,7 +187,8 @@ func (e *togglingFollowerEngine) Status() raftengine.Status { func (e *togglingFollowerEngine) Configuration(context.Context) (raftengine.Configuration, error) { return raftengine.Configuration{}, nil } -func (e *togglingFollowerEngine) Close() error { return nil } +func (e *togglingFollowerEngine) SnapshotEvery() uint64 { return 10_000 } +func (e *togglingFollowerEngine) Close() error { return nil } func TestLeaderProxy_ForwardsAfterLeaderPublishes(t *testing.T) { t.Parallel() diff --git a/kv/lease_read_test.go b/kv/lease_read_test.go index d62c103d2..05fe41ac2 100644 --- a/kv/lease_read_test.go +++ b/kv/lease_read_test.go @@ -62,6 +62,7 @@ func (e *fakeLeaseEngine) Status() raftengine.Status { func (e *fakeLeaseEngine) Configuration(context.Context) (raftengine.Configuration, error) { return raftengine.Configuration{}, nil } +func (e *fakeLeaseEngine) SnapshotEvery() uint64 { return 10_000 } func (e *fakeLeaseEngine) Propose(context.Context, []byte) (*raftengine.ProposalResult, error) { e.proposeCalls.Add(1) if e.proposeHook != nil { @@ -163,6 +164,7 @@ func (e *nonLeaseEngine) Status() raftengine.Status { func (e *nonLeaseEngine) Configuration(context.Context) (raftengine.Configuration, error) { return raftengine.Configuration{}, nil } +func (e *nonLeaseEngine) SnapshotEvery() uint64 { return 10_000 } func (e *nonLeaseEngine) Propose(context.Context, []byte) (*raftengine.ProposalResult, error) { return &raftengine.ProposalResult{}, nil } diff --git a/kv/sharded_coordinator_txn_test.go b/kv/sharded_coordinator_txn_test.go index 964f2713f..4c49c9ac8 100644 --- a/kv/sharded_coordinator_txn_test.go +++ b/kv/sharded_coordinator_txn_test.go @@ -412,7 +412,8 @@ func (noopEngine) Status() raftengine.Status { return r func (noopEngine) Configuration(_ context.Context) (raftengine.Configuration, error) { return raftengine.Configuration{}, nil } -func (noopEngine) Close() error { return nil } +func (noopEngine) SnapshotEvery() uint64 { return 10_000 } +func (noopEngine) Close() error { return nil } func TestValidateReadOnlyShards_DetectsConflictOnReadOnlyShard(t *testing.T) { t.Parallel() diff --git a/main.go b/main.go index f8f8dbd17..19fe831a9 100644 --- a/main.go +++ b/main.go @@ -1441,43 +1441,27 @@ type serversInput struct { // per-group Raft listeners needed for quorum traffic, waits for any requested // startup rotation, then opens the public service listeners. func startServersAfterStartupRotation(waitRotateOnStartup startupRotationWaiter, in serversInput) error { - adminServer, adminGRPCOpts, err := setupAdminService(*raftId, *myAddr, in.runtimes, in.bootstrapServers, in.keyvizSampler) - if err != nil { - return err - } - // roleStore + connCache are gated on *adminEnabled. With admin - // disabled, building either is wasted work AND a security - // regression risk: a non-empty -adminFullAccessKeys flag would - // otherwise still flip forwardDeps.readyForRegistration() to - // true, registering the leader-side gRPC AdminForward service - // and re-exposing the table-write surface a follower-direct - // admin call could reach (P1/Major review on #648). - // The HTTP admin listener already short-circuits in - // prepareAdminFromFlags when *adminEnabled is false; the gRPC path - // must do the same. var ( roleStore admin.RoleStore connCache *kv.GRPCConnCache ) + adminGRPCEnabled := *adminTokenFile != "" || *adminInsecureNoAuth + // roleStore is gated on *adminEnabled. With admin HTTP disabled, building + // it is wasted work AND a security regression risk: a non-empty + // -adminFullAccessKeys flag would otherwise still flip + // forwardDeps.readyForRegistration() to true, registering the leader-side + // gRPC AdminForward service and re-exposing the table-write surface a + // follower-direct admin call could reach (P1/Major review on #648). + // + // connCache is also used by the gRPC Admin GetNodeVersion probe, so create + // it when either admin HTTP or gRPC Admin is enabled. + connCache = prepareAdminConnCache(in.ctx, in.eg, *adminEnabled || adminGRPCEnabled) + adminServer, adminGRPCOpts, err := setupAdminService(*raftId, *myAddr, in.runtimes, in.bootstrapServers, in.keyvizSampler, connCache) + if err != nil { + return err + } if *adminEnabled { roleStore = roleStoreFromFlags(parseCSV(*adminFullAccessKeys), parseCSV(*adminReadOnlyAccessKeys)) - // connCache is shared between the follower-side LeaderForwarder - // (built inside prepareAdminFromFlags) and any future bridge that - // dials the leader's gRPC ports. Keeping a single instance per - // process means the two paths re-use TLS / HTTP/2 connections - // rather than each maintaining a parallel pool. The shutdown - // goroutine drains the cache on context cancellation so the - // accumulated HTTP/2 connections are not leaked when the - // process exits gracefully (Claude review on #648). - connCache = &kv.GRPCConnCache{} - cache := connCache - in.eg.Go(func() error { - <-in.ctx.Done() - if err := cache.Close(); err != nil { - return errors.Wrap(err, "close admin gRPC connection cache") - } - return nil - }) } publicKVGate := &startupPublicKVGate{} installHLCLeaseRenewalBlocker(in.coordinate, waitRotateOnStartup.BlockMutators) @@ -1628,6 +1612,7 @@ func setupAdminService( runtimes []*raftGroupRuntime, bootstrapServers []raftengine.Server, keyvizSampler *keyviz.MemSampler, + connCache *kv.GRPCConnCache, ) (*adapter.AdminServer, adminGRPCInterceptors, error) { members := adminMembersFromBootstrap(nodeID, bootstrapServers) // In multi-group mode the process does not listen on *myAddr — each group @@ -1641,6 +1626,7 @@ func setupAdminService( *adminInsecureNoAuth, adapter.NodeIdentity{NodeID: nodeID, GRPCAddress: selfAddr}, members, + connCache, ) if err != nil { return nil, adminGRPCInterceptors{}, err @@ -1862,6 +1848,7 @@ func configureAdminService( insecureNoAuth bool, self adapter.NodeIdentity, members []adapter.NodeIdentity, + connCache *kv.GRPCConnCache, ) (*adapter.AdminServer, adminGRPCInterceptors, error) { if tokenPath == "" && !insecureNoAuth { return nil, adminGRPCInterceptors{}, nil @@ -1877,7 +1864,11 @@ func configureAdminService( } token = loaded } - srv := adapter.NewAdminServer(self, members) + opts := []adapter.AdminOption{adapter.WithAdminNodeVersion(buildVersion())} + if probe := adminLeaderVersionProbe(connCache); probe != nil { + opts = append(opts, adapter.WithAdminLeaderVersionProbe(probe)) + } + srv := adapter.NewAdminServer(self, members, opts...) unary, stream := adapter.AdminTokenAuth(token) var icept adminGRPCInterceptors if unary != nil { @@ -1889,6 +1880,23 @@ func configureAdminService( return srv, icept, nil } +func adminLeaderVersionProbe(connCache *kv.GRPCConnCache) adapter.LeaderVersionProbe { + if connCache == nil { + return nil + } + return func(ctx context.Context, address string) (string, error) { + conn, err := connCache.ConnFor(address) + if err != nil { + return "", errors.Wrap(err, "admin leader version probe: dial peer") + } + resp, err := pb.NewAdminClient(conn).GetNodeVersion(ctx, &pb.GetNodeVersionRequest{}) + if err != nil { + return "", errors.Wrap(err, "admin leader version probe: get node version") + } + return resp.GetNodeVersion(), nil + } +} + // loadAdminTokenFile materialises --adminTokenFile with a strict upper bound // so a misconfigured path (for example a log file) cannot force an arbitrary // allocation before the bearer-token check. Delegates to the shared helper in @@ -2567,6 +2575,21 @@ func (r *runtimeServerRunner) startRaftTransport() error { return nil } +func prepareAdminConnCache(ctx context.Context, eg *errgroup.Group, enabled bool) *kv.GRPCConnCache { + if !enabled { + return nil + } + connCache := &kv.GRPCConnCache{} + eg.Go(func() error { + <-ctx.Done() + if err := connCache.Close(); err != nil { + return errors.Wrap(err, "close admin gRPC connection cache") + } + return nil + }) + return connCache +} + func (r *runtimeServerRunner) prepareAdminForwardServers() error { r.dynamoServer = newDynamoDBServer(r.shardStore, r.coordinate, r.leaderDynamo, r.metricsRegistry, r.readTracker) s3Server, err := newS3Server( diff --git a/main_admin.go b/main_admin.go index 6fdce1785..6083ccfa4 100644 --- a/main_admin.go +++ b/main_admin.go @@ -8,6 +8,7 @@ import ( "net" "net/http" "os" + "runtime/debug" "strings" "time" @@ -33,16 +34,43 @@ const ( adminWriteTimeout = 10 * time.Second adminIdleTimeout = 30 * time.Second adminShutdownTimeout = 5 * time.Second +) + +// adminBuildVersion is surfaced by both GET /admin/api/v1/cluster and the +// Admin gRPC GetNodeVersion RPC. Release builds may stamp it with: +// +// -ldflags "-X main.adminBuildVersion=" +var adminBuildVersion = "dev" - // adminBuildVersion is surfaced in GET /admin/api/v1/cluster. Until - // we wire real ldflags-injected build info, a placeholder is fine. - adminBuildVersion = "dev" +const ( + buildInfoVCSRevisionKey = "vcs.revision" + buildInfoVCSModifiedKey = "vcs.modified" ) -// buildVersion returns the elastickv binary version for admin purposes. -// It is intentionally a function, not a constant, so build tooling can -// link-replace it via -ldflags in the future. -func buildVersion() string { return adminBuildVersion } +func buildVersion() string { + if version := strings.TrimSpace(adminBuildVersion); version != "" && version != "dev" { + return version + } + if info, ok := debug.ReadBuildInfo(); ok { + revision := "" + modified := false + for _, setting := range info.Settings { + switch setting.Key { + case buildInfoVCSRevisionKey: + revision = strings.TrimSpace(setting.Value) + case buildInfoVCSModifiedKey: + modified = setting.Value == "true" + } + } + if revision != "" { + if modified { + return revision + "-modified" + } + return revision + } + } + return "dev" +} // adminListenerConfig is the subset of startup inputs that goes into the // admin listener. Collecting them in a struct keeps the main.go call site diff --git a/main_admin_test.go b/main_admin_test.go index f12d7314d..98d52f857 100644 --- a/main_admin_test.go +++ b/main_admin_test.go @@ -26,13 +26,15 @@ import ( "github.com/bootjp/elastickv/internal/admin" "github.com/bootjp/elastickv/internal/raftengine" "github.com/bootjp/elastickv/kv" + pb "github.com/bootjp/elastickv/proto" "github.com/stretchr/testify/require" "golang.org/x/sync/errgroup" + "google.golang.org/grpc" ) func TestConfigureAdminServiceDisabledByDefault(t *testing.T) { t.Parallel() - srv, icept, err := configureAdminService("", false, adapter.NodeIdentity{NodeID: "n1"}, nil) + srv, icept, err := configureAdminService("", false, adapter.NodeIdentity{NodeID: "n1"}, nil, nil) if err != nil { t.Fatalf("disabled-by-default should not error: %v", err) } @@ -48,7 +50,7 @@ func TestConfigureAdminServiceRejectsMutualExclusion(t *testing.T) { if err := os.WriteFile(tokPath, []byte("x"), 0o600); err != nil { t.Fatal(err) } - if _, _, err := configureAdminService(tokPath, true, adapter.NodeIdentity{}, nil); err == nil { + if _, _, err := configureAdminService(tokPath, true, adapter.NodeIdentity{}, nil, nil); err == nil { t.Fatal("expected mutual-exclusion error") } } @@ -60,7 +62,7 @@ func TestConfigureAdminServiceTokenFile(t *testing.T) { if err := os.WriteFile(tokPath, []byte("hunter2\n"), 0o600); err != nil { t.Fatal(err) } - srv, icept, err := configureAdminService(tokPath, false, adapter.NodeIdentity{NodeID: "n1"}, nil) + srv, icept, err := configureAdminService(tokPath, false, adapter.NodeIdentity{NodeID: "n1"}, nil, nil) if err != nil { t.Fatalf("configureAdminService: %v", err) } @@ -75,7 +77,7 @@ func TestConfigureAdminServiceTokenFile(t *testing.T) { func TestConfigureAdminServiceInsecureNoAuth(t *testing.T) { t.Parallel() - srv, icept, err := configureAdminService("", true, adapter.NodeIdentity{NodeID: "n1"}, nil) + srv, icept, err := configureAdminService("", true, adapter.NodeIdentity{NodeID: "n1"}, nil, nil) if err != nil { t.Fatalf("insecure mode should succeed: %v", err) } @@ -87,6 +89,90 @@ func TestConfigureAdminServiceInsecureNoAuth(t *testing.T) { } } +func TestConfigureAdminServiceWiresLeaderVersionProbe(t *testing.T) { + t.Parallel() + var lc net.ListenConfig + lis, err := lc.Listen(context.Background(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + remote := adapter.NewAdminServer( + adapter.NodeIdentity{NodeID: "n2", GRPCAddress: lis.Addr().String()}, + nil, + adapter.WithAdminNodeVersion("v-remote"), + ) + gs := grpc.NewServer() + pb.RegisterAdminServer(gs, remote) + serveErr := make(chan error, 1) + go func() { + serveErr <- gs.Serve(lis) + }() + t.Cleanup(func() { + gs.Stop() + _ = lis.Close() + <-serveErr + }) + + var cache kv.GRPCConnCache + t.Cleanup(func() { require.NoError(t, cache.Close()) }) + srv, _, err := configureAdminService( + "", + true, + adapter.NodeIdentity{NodeID: "n1", GRPCAddress: "127.0.0.1:50051"}, + nil, + &cache, + ) + require.NoError(t, err) + require.NotNil(t, srv) + srv.RegisterGroup(1, adminVersionProbeGroup{leader: raftengine.LeaderInfo{ + ID: "n2", + Address: lis.Addr().String(), + }}) + + resp, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + require.NoError(t, err) + require.Len(t, resp.Groups, 1) + require.Empty(t, resp.Groups[0].LeaderNodeVersion) + require.Eventually(t, func() bool { + resp, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{}) + return err == nil && len(resp.Groups) == 1 && resp.Groups[0].LeaderNodeVersion == "v-remote" + }, time.Second, 10*time.Millisecond) +} + +func TestConfigureAdminServiceUsesStampedBuildVersion(t *testing.T) { + old := adminBuildVersion + adminBuildVersion = "v-admin-test" + t.Cleanup(func() { adminBuildVersion = old }) + + if got := buildVersion(); got != "v-admin-test" { + t.Fatalf("buildVersion = %q, want v-admin-test", got) + } + srv, _, err := configureAdminService( + "", + true, + adapter.NodeIdentity{NodeID: "n1", GRPCAddress: "127.0.0.1:50051"}, + nil, + nil, + ) + require.NoError(t, err) + require.NotNil(t, srv) + resp, err := srv.GetNodeVersion(context.Background(), &pb.GetNodeVersionRequest{}) + require.NoError(t, err) + require.Equal(t, "v-admin-test", resp.GetNodeVersion()) +} + +type adminVersionProbeGroup struct { + leader raftengine.LeaderInfo +} + +func (g adminVersionProbeGroup) Status() raftengine.Status { + return raftengine.Status{Leader: g.leader} +} + +func (adminVersionProbeGroup) Configuration(context.Context) (raftengine.Configuration, error) { + return raftengine.Configuration{}, nil +} + +func (adminVersionProbeGroup) SnapshotEvery() uint64 { return 0 } + func TestAdminMembersFromBootstrapExcludesSelf(t *testing.T) { t.Parallel() servers := []raftengine.Server{ diff --git a/main_sqs_leadership_refusal_test.go b/main_sqs_leadership_refusal_test.go index c7889c0a1..7f7731423 100644 --- a/main_sqs_leadership_refusal_test.go +++ b/main_sqs_leadership_refusal_test.go @@ -33,7 +33,8 @@ func (f *fakeLeadershipController) Status() raftengine.Status { return raftengin func (f *fakeLeadershipController) Configuration(_ context.Context) (raftengine.Configuration, error) { return raftengine.Configuration{}, nil } -func (f *fakeLeadershipController) Close() error { return nil } +func (f *fakeLeadershipController) SnapshotEvery() uint64 { return 10_000 } +func (f *fakeLeadershipController) Close() error { return nil } // fakeLeadershipController is a sqsLeadershipController test // double. It records TransferLeadership invocations and exposes diff --git a/proto/admin.pb.go b/proto/admin.pb.go index ace6b3462..186312aec 100644 --- a/proto/admin.pb.go +++ b/proto/admin.pb.go @@ -356,6 +356,9 @@ type RaftGroupState struct { // engine is a follower that has never heard from a leader); UIs should // render that case as "unknown" rather than "contacted at epoch". LastContactUnixMs int64 `protobuf:"varint,6,opt,name=last_contact_unix_ms,json=lastContactUnixMs,proto3" json:"last_contact_unix_ms,omitempty"` + // leader_node_version is populated asynchronously from the leader's Admin + // GetNodeVersion RPC. Empty means unknown/unreachable/cache miss. + LeaderNodeVersion string `protobuf:"bytes,7,opt,name=leader_node_version,json=leaderNodeVersion,proto3" json:"leader_node_version,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -432,6 +435,13 @@ func (x *RaftGroupState) GetLastContactUnixMs() int64 { return 0 } +func (x *RaftGroupState) GetLeaderNodeVersion() string { + if x != nil { + return x.LeaderNodeVersion + } + return "" +} + type GetRaftGroupsRequest struct { state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields @@ -1107,6 +1117,634 @@ func (x *GetRouteDetailResponse) GetPerAdapter() []*AdapterSummary { return nil } +type BackupShardApplied 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"` + AppliedIndex uint64 `protobuf:"varint,2,opt,name=applied_index,json=appliedIndex,proto3" json:"applied_index,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *BackupShardApplied) Reset() { + *x = BackupShardApplied{} + mi := &file_admin_proto_msgTypes[15] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *BackupShardApplied) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*BackupShardApplied) ProtoMessage() {} + +func (x *BackupShardApplied) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[15] + 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 BackupShardApplied.ProtoReflect.Descriptor instead. +func (*BackupShardApplied) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{15} +} + +func (x *BackupShardApplied) GetRaftGroupId() uint64 { + if x != nil { + return x.RaftGroupId + } + return 0 +} + +func (x *BackupShardApplied) GetAppliedIndex() uint64 { + if x != nil { + return x.AppliedIndex + } + return 0 +} + +type BackupExpectedKeys struct { + state protoimpl.MessageState `protogen:"open.v1"` + Adapter string `protobuf:"bytes,1,opt,name=adapter,proto3" json:"adapter,omitempty"` + Scope string `protobuf:"bytes,2,opt,name=scope,proto3" json:"scope,omitempty"` + KeyCount uint64 `protobuf:"varint,3,opt,name=key_count,json=keyCount,proto3" json:"key_count,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *BackupExpectedKeys) Reset() { + *x = BackupExpectedKeys{} + mi := &file_admin_proto_msgTypes[16] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *BackupExpectedKeys) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*BackupExpectedKeys) ProtoMessage() {} + +func (x *BackupExpectedKeys) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[16] + 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 BackupExpectedKeys.ProtoReflect.Descriptor instead. +func (*BackupExpectedKeys) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{16} +} + +func (x *BackupExpectedKeys) GetAdapter() string { + if x != nil { + return x.Adapter + } + return "" +} + +func (x *BackupExpectedKeys) GetScope() string { + if x != nil { + return x.Scope + } + return "" +} + +func (x *BackupExpectedKeys) GetKeyCount() uint64 { + if x != nil { + return x.KeyCount + } + return 0 +} + +type BeginBackupRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + TtlMs uint64 `protobuf:"varint,1,opt,name=ttl_ms,json=ttlMs,proto3" json:"ttl_ms,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *BeginBackupRequest) Reset() { + *x = BeginBackupRequest{} + mi := &file_admin_proto_msgTypes[17] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *BeginBackupRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*BeginBackupRequest) ProtoMessage() {} + +func (x *BeginBackupRequest) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[17] + 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 BeginBackupRequest.ProtoReflect.Descriptor instead. +func (*BeginBackupRequest) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{17} +} + +func (x *BeginBackupRequest) GetTtlMs() uint64 { + if x != nil { + return x.TtlMs + } + return 0 +} + +type BeginBackupResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + ReadTs uint64 `protobuf:"varint,1,opt,name=read_ts,json=readTs,proto3" json:"read_ts,omitempty"` + PinToken []byte `protobuf:"bytes,2,opt,name=pin_token,json=pinToken,proto3" json:"pin_token,omitempty"` + TtlMsEffective uint64 `protobuf:"varint,3,opt,name=ttl_ms_effective,json=ttlMsEffective,proto3" json:"ttl_ms_effective,omitempty"` + Shards []*BackupShardApplied `protobuf:"bytes,4,rep,name=shards,proto3" json:"shards,omitempty"` + ExpectedKeys []*BackupExpectedKeys `protobuf:"bytes,5,rep,name=expected_keys,json=expectedKeys,proto3" json:"expected_keys,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *BeginBackupResponse) Reset() { + *x = BeginBackupResponse{} + mi := &file_admin_proto_msgTypes[18] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *BeginBackupResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*BeginBackupResponse) ProtoMessage() {} + +func (x *BeginBackupResponse) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[18] + 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 BeginBackupResponse.ProtoReflect.Descriptor instead. +func (*BeginBackupResponse) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{18} +} + +func (x *BeginBackupResponse) GetReadTs() uint64 { + if x != nil { + return x.ReadTs + } + return 0 +} + +func (x *BeginBackupResponse) GetPinToken() []byte { + if x != nil { + return x.PinToken + } + return nil +} + +func (x *BeginBackupResponse) GetTtlMsEffective() uint64 { + if x != nil { + return x.TtlMsEffective + } + return 0 +} + +func (x *BeginBackupResponse) GetShards() []*BackupShardApplied { + if x != nil { + return x.Shards + } + return nil +} + +func (x *BeginBackupResponse) GetExpectedKeys() []*BackupExpectedKeys { + if x != nil { + return x.ExpectedKeys + } + return nil +} + +type RenewBackupRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + PinToken []byte `protobuf:"bytes,1,opt,name=pin_token,json=pinToken,proto3" json:"pin_token,omitempty"` + TtlMs uint64 `protobuf:"varint,2,opt,name=ttl_ms,json=ttlMs,proto3" json:"ttl_ms,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *RenewBackupRequest) Reset() { + *x = RenewBackupRequest{} + mi := &file_admin_proto_msgTypes[19] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *RenewBackupRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RenewBackupRequest) ProtoMessage() {} + +func (x *RenewBackupRequest) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[19] + 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 RenewBackupRequest.ProtoReflect.Descriptor instead. +func (*RenewBackupRequest) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{19} +} + +func (x *RenewBackupRequest) GetPinToken() []byte { + if x != nil { + return x.PinToken + } + return nil +} + +func (x *RenewBackupRequest) GetTtlMs() uint64 { + if x != nil { + return x.TtlMs + } + return 0 +} + +type RenewBackupResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + TtlMsEffective uint64 `protobuf:"varint,1,opt,name=ttl_ms_effective,json=ttlMsEffective,proto3" json:"ttl_ms_effective,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *RenewBackupResponse) Reset() { + *x = RenewBackupResponse{} + mi := &file_admin_proto_msgTypes[20] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *RenewBackupResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RenewBackupResponse) ProtoMessage() {} + +func (x *RenewBackupResponse) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[20] + 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 RenewBackupResponse.ProtoReflect.Descriptor instead. +func (*RenewBackupResponse) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{20} +} + +func (x *RenewBackupResponse) GetTtlMsEffective() uint64 { + if x != nil { + return x.TtlMsEffective + } + return 0 +} + +type EndBackupRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + PinToken []byte `protobuf:"bytes,1,opt,name=pin_token,json=pinToken,proto3" json:"pin_token,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *EndBackupRequest) Reset() { + *x = EndBackupRequest{} + mi := &file_admin_proto_msgTypes[21] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *EndBackupRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*EndBackupRequest) ProtoMessage() {} + +func (x *EndBackupRequest) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[21] + 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 EndBackupRequest.ProtoReflect.Descriptor instead. +func (*EndBackupRequest) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{21} +} + +func (x *EndBackupRequest) GetPinToken() []byte { + if x != nil { + return x.PinToken + } + return nil +} + +type EndBackupResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *EndBackupResponse) Reset() { + *x = EndBackupResponse{} + mi := &file_admin_proto_msgTypes[22] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *EndBackupResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*EndBackupResponse) ProtoMessage() {} + +func (x *EndBackupResponse) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[22] + 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 EndBackupResponse.ProtoReflect.Descriptor instead. +func (*EndBackupResponse) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{22} +} + +type ListAdaptersAndScopesRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + PinToken []byte `protobuf:"bytes,1,opt,name=pin_token,json=pinToken,proto3" json:"pin_token,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ListAdaptersAndScopesRequest) Reset() { + *x = ListAdaptersAndScopesRequest{} + mi := &file_admin_proto_msgTypes[23] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ListAdaptersAndScopesRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ListAdaptersAndScopesRequest) ProtoMessage() {} + +func (x *ListAdaptersAndScopesRequest) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[23] + 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 ListAdaptersAndScopesRequest.ProtoReflect.Descriptor instead. +func (*ListAdaptersAndScopesRequest) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{23} +} + +func (x *ListAdaptersAndScopesRequest) GetPinToken() []byte { + if x != nil { + return x.PinToken + } + return nil +} + +type BackupScope struct { + state protoimpl.MessageState `protogen:"open.v1"` + Adapter string `protobuf:"bytes,1,opt,name=adapter,proto3" json:"adapter,omitempty"` + Scope string `protobuf:"bytes,2,opt,name=scope,proto3" json:"scope,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *BackupScope) Reset() { + *x = BackupScope{} + mi := &file_admin_proto_msgTypes[24] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *BackupScope) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*BackupScope) ProtoMessage() {} + +func (x *BackupScope) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[24] + 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 BackupScope.ProtoReflect.Descriptor instead. +func (*BackupScope) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{24} +} + +func (x *BackupScope) GetAdapter() string { + if x != nil { + return x.Adapter + } + return "" +} + +func (x *BackupScope) GetScope() string { + if x != nil { + return x.Scope + } + return "" +} + +type ListAdaptersAndScopesResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + Scopes []*BackupScope `protobuf:"bytes,1,rep,name=scopes,proto3" json:"scopes,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ListAdaptersAndScopesResponse) Reset() { + *x = ListAdaptersAndScopesResponse{} + mi := &file_admin_proto_msgTypes[25] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ListAdaptersAndScopesResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ListAdaptersAndScopesResponse) ProtoMessage() {} + +func (x *ListAdaptersAndScopesResponse) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[25] + 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 ListAdaptersAndScopesResponse.ProtoReflect.Descriptor instead. +func (*ListAdaptersAndScopesResponse) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{25} +} + +func (x *ListAdaptersAndScopesResponse) GetScopes() []*BackupScope { + if x != nil { + return x.Scopes + } + return nil +} + +type GetNodeVersionRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetNodeVersionRequest) Reset() { + *x = GetNodeVersionRequest{} + mi := &file_admin_proto_msgTypes[26] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetNodeVersionRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetNodeVersionRequest) ProtoMessage() {} + +func (x *GetNodeVersionRequest) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[26] + 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 GetNodeVersionRequest.ProtoReflect.Descriptor instead. +func (*GetNodeVersionRequest) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{26} +} + +type GetNodeVersionResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + NodeVersion string `protobuf:"bytes,1,opt,name=node_version,json=nodeVersion,proto3" json:"node_version,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetNodeVersionResponse) Reset() { + *x = GetNodeVersionResponse{} + mi := &file_admin_proto_msgTypes[27] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetNodeVersionResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetNodeVersionResponse) ProtoMessage() {} + +func (x *GetNodeVersionResponse) ProtoReflect() protoreflect.Message { + mi := &file_admin_proto_msgTypes[27] + 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 GetNodeVersionResponse.ProtoReflect.Descriptor instead. +func (*GetNodeVersionResponse) Descriptor() ([]byte, []int) { + return file_admin_proto_rawDescGZIP(), []int{27} +} + +func (x *GetNodeVersionResponse) GetNodeVersion() string { + if x != nil { + return x.NodeVersion + } + return "" +} + type StreamEventsRequest struct { state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields @@ -1115,7 +1753,7 @@ type StreamEventsRequest struct { func (x *StreamEventsRequest) Reset() { *x = StreamEventsRequest{} - mi := &file_admin_proto_msgTypes[15] + mi := &file_admin_proto_msgTypes[28] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1127,7 +1765,7 @@ func (x *StreamEventsRequest) String() string { func (*StreamEventsRequest) ProtoMessage() {} func (x *StreamEventsRequest) ProtoReflect() protoreflect.Message { - mi := &file_admin_proto_msgTypes[15] + mi := &file_admin_proto_msgTypes[28] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1140,7 +1778,7 @@ func (x *StreamEventsRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use StreamEventsRequest.ProtoReflect.Descriptor instead. func (*StreamEventsRequest) Descriptor() ([]byte, []int) { - return file_admin_proto_rawDescGZIP(), []int{15} + return file_admin_proto_rawDescGZIP(), []int{28} } type StreamEventsEvent struct { @@ -1156,7 +1794,7 @@ type StreamEventsEvent struct { func (x *StreamEventsEvent) Reset() { *x = StreamEventsEvent{} - mi := &file_admin_proto_msgTypes[16] + mi := &file_admin_proto_msgTypes[29] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1168,7 +1806,7 @@ func (x *StreamEventsEvent) String() string { func (*StreamEventsEvent) ProtoMessage() {} func (x *StreamEventsEvent) ProtoReflect() protoreflect.Message { - mi := &file_admin_proto_msgTypes[16] + mi := &file_admin_proto_msgTypes[29] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1181,7 +1819,7 @@ func (x *StreamEventsEvent) ProtoReflect() protoreflect.Message { // Deprecated: Use StreamEventsEvent.ProtoReflect.Descriptor instead. func (*StreamEventsEvent) Descriptor() ([]byte, []int) { - return file_admin_proto_rawDescGZIP(), []int{16} + return file_admin_proto_rawDescGZIP(), []int{29} } func (x *StreamEventsEvent) GetEvent() isStreamEventsEvent_Event { @@ -1237,7 +1875,7 @@ type RouteTransition struct { func (x *RouteTransition) Reset() { *x = RouteTransition{} - mi := &file_admin_proto_msgTypes[17] + mi := &file_admin_proto_msgTypes[30] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1249,7 +1887,7 @@ func (x *RouteTransition) String() string { func (*RouteTransition) ProtoMessage() {} func (x *RouteTransition) ProtoReflect() protoreflect.Message { - mi := &file_admin_proto_msgTypes[17] + mi := &file_admin_proto_msgTypes[30] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1262,7 +1900,7 @@ func (x *RouteTransition) ProtoReflect() protoreflect.Message { // Deprecated: Use RouteTransition.ProtoReflect.Descriptor instead. func (*RouteTransition) Descriptor() ([]byte, []int) { - return file_admin_proto_rawDescGZIP(), []int{17} + return file_admin_proto_rawDescGZIP(), []int{30} } func (x *RouteTransition) GetParentRouteId() uint64 { @@ -1304,7 +1942,7 @@ type KeyVizColumn struct { func (x *KeyVizColumn) Reset() { *x = KeyVizColumn{} - mi := &file_admin_proto_msgTypes[18] + mi := &file_admin_proto_msgTypes[31] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1316,7 +1954,7 @@ func (x *KeyVizColumn) String() string { func (*KeyVizColumn) ProtoMessage() {} func (x *KeyVizColumn) ProtoReflect() protoreflect.Message { - mi := &file_admin_proto_msgTypes[18] + mi := &file_admin_proto_msgTypes[31] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1329,7 +1967,7 @@ func (x *KeyVizColumn) ProtoReflect() protoreflect.Message { // Deprecated: Use KeyVizColumn.ProtoReflect.Descriptor instead. func (*KeyVizColumn) Descriptor() ([]byte, []int) { - return file_admin_proto_rawDescGZIP(), []int{18} + return file_admin_proto_rawDescGZIP(), []int{31} } func (x *KeyVizColumn) GetColumnUnixMs() int64 { @@ -1371,7 +2009,7 @@ const file_admin_proto_rawDesc = "" + "\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\"\xa4\x02\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" + @@ -1379,7 +2017,8 @@ const file_admin_proto_rawDesc = "" + "leaderTerm\x12!\n" + "\fcommit_index\x18\x04 \x01(\x04R\vcommitIndex\x12#\n" + "\rapplied_index\x18\x05 \x01(\x04R\fappliedIndex\x12/\n" + - "\x14last_contact_unix_ms\x18\x06 \x01(\x03R\x11lastContactUnixMs\"\x16\n" + + "\x14last_contact_unix_ms\x18\x06 \x01(\x03R\x11lastContactUnixMs\x12.\n" + + "\x13leader_node_version\x18\a \x01(\tR\x11leaderNodeVersion\"\x16\n" + "\x14GetRaftGroupsRequest\"@\n" + "\x15GetRaftGroupsResponse\x12'\n" + "\x06groups\x18\x01 \x03(\v2\x0f.RaftGroupStateR\x06groups\"\xfe\x01\n" + @@ -1435,7 +2074,40 @@ const file_admin_proto_rawDesc = "" + "\x03row\x18\x01 \x01(\v2\n" + ".KeyVizRowR\x03row\x120\n" + "\vper_adapter\x18\x02 \x03(\v2\x0f.AdapterSummaryR\n" + - "perAdapter\"\x15\n" + + "perAdapter\"]\n" + + "\x12BackupShardApplied\x12\"\n" + + "\rraft_group_id\x18\x01 \x01(\x04R\vraftGroupId\x12#\n" + + "\rapplied_index\x18\x02 \x01(\x04R\fappliedIndex\"a\n" + + "\x12BackupExpectedKeys\x12\x18\n" + + "\aadapter\x18\x01 \x01(\tR\aadapter\x12\x14\n" + + "\x05scope\x18\x02 \x01(\tR\x05scope\x12\x1b\n" + + "\tkey_count\x18\x03 \x01(\x04R\bkeyCount\"+\n" + + "\x12BeginBackupRequest\x12\x15\n" + + "\x06ttl_ms\x18\x01 \x01(\x04R\x05ttlMs\"\xdc\x01\n" + + "\x13BeginBackupResponse\x12\x17\n" + + "\aread_ts\x18\x01 \x01(\x04R\x06readTs\x12\x1b\n" + + "\tpin_token\x18\x02 \x01(\fR\bpinToken\x12(\n" + + "\x10ttl_ms_effective\x18\x03 \x01(\x04R\x0ettlMsEffective\x12+\n" + + "\x06shards\x18\x04 \x03(\v2\x13.BackupShardAppliedR\x06shards\x128\n" + + "\rexpected_keys\x18\x05 \x03(\v2\x13.BackupExpectedKeysR\fexpectedKeys\"H\n" + + "\x12RenewBackupRequest\x12\x1b\n" + + "\tpin_token\x18\x01 \x01(\fR\bpinToken\x12\x15\n" + + "\x06ttl_ms\x18\x02 \x01(\x04R\x05ttlMs\"?\n" + + "\x13RenewBackupResponse\x12(\n" + + "\x10ttl_ms_effective\x18\x01 \x01(\x04R\x0ettlMsEffective\"/\n" + + "\x10EndBackupRequest\x12\x1b\n" + + "\tpin_token\x18\x01 \x01(\fR\bpinToken\"\x13\n" + + "\x11EndBackupResponse\";\n" + + "\x1cListAdaptersAndScopesRequest\x12\x1b\n" + + "\tpin_token\x18\x01 \x01(\fR\bpinToken\"=\n" + + "\vBackupScope\x12\x18\n" + + "\aadapter\x18\x01 \x01(\tR\aadapter\x12\x14\n" + + "\x05scope\x18\x02 \x01(\tR\x05scope\"E\n" + + "\x1dListAdaptersAndScopesResponse\x12$\n" + + "\x06scopes\x18\x01 \x03(\v2\f.BackupScopeR\x06scopes\"\x17\n" + + "\x15GetNodeVersionRequest\";\n" + + "\x16GetNodeVersionResponse\x12!\n" + + "\fnode_version\x18\x01 \x01(\tR\vnodeVersion\"\x15\n" + "\x13StreamEventsRequest\"\x91\x01\n" + "\x11StreamEventsEvent\x12=\n" + "\x10route_transition\x18\x01 \x01(\v2\x10.RouteTransitionH\x00R\x0frouteTransition\x124\n" + @@ -1463,13 +2135,18 @@ const file_admin_proto_rawDesc = "" + "\x17SAMPLE_ROLE_UNSPECIFIED\x10\x00\x12\x1c\n" + "\x18SAMPLE_ROLE_LEADER_WRITE\x10\x01\x12\x1b\n" + "\x17SAMPLE_ROLE_LEADER_READ\x10\x02\x12\x1d\n" + - "\x19SAMPLE_ROLE_FOLLOWER_READ\x10\x032\xb3\x03\n" + + "\x19SAMPLE_ROLE_FOLLOWER_READ\x10\x032\x80\x06\n" + "\x05Admin\x12O\n" + "\x12GetClusterOverview\x12\x1a.GetClusterOverviewRequest\x1a\x1b.GetClusterOverviewResponse\"\x00\x12@\n" + "\rGetRaftGroups\x12\x15.GetRaftGroupsRequest\x1a\x16.GetRaftGroupsResponse\"\x00\x12L\n" + "\x11GetAdapterSummary\x12\x19.GetAdapterSummaryRequest\x1a\x1a.GetAdapterSummaryResponse\"\x00\x12F\n" + "\x0fGetKeyVizMatrix\x12\x17.GetKeyVizMatrixRequest\x1a\x18.GetKeyVizMatrixResponse\"\x00\x12C\n" + - "\x0eGetRouteDetail\x12\x16.GetRouteDetailRequest\x1a\x17.GetRouteDetailResponse\"\x00\x12<\n" + + "\x0eGetRouteDetail\x12\x16.GetRouteDetailRequest\x1a\x17.GetRouteDetailResponse\"\x00\x12:\n" + + "\vBeginBackup\x12\x13.BeginBackupRequest\x1a\x14.BeginBackupResponse\"\x00\x12:\n" + + "\vRenewBackup\x12\x13.RenewBackupRequest\x1a\x14.RenewBackupResponse\"\x00\x124\n" + + "\tEndBackup\x12\x11.EndBackupRequest\x1a\x12.EndBackupResponse\"\x00\x12X\n" + + "\x15ListAdaptersAndScopes\x12\x1d.ListAdaptersAndScopesRequest\x1a\x1e.ListAdaptersAndScopesResponse\"\x00\x12C\n" + + "\x0eGetNodeVersion\x12\x16.GetNodeVersionRequest\x1a\x17.GetNodeVersionResponse\"\x00\x12<\n" + "\fStreamEvents\x12\x14.StreamEventsRequest\x1a\x12.StreamEventsEvent\"\x000\x01B#Z!github.com/bootjp/elastickv/protob\x06proto3" var ( @@ -1485,29 +2162,42 @@ 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, 32) var file_admin_proto_goTypes = []any{ - (KeyVizSeries)(0), // 0: KeyVizSeries - (SampleRole)(0), // 1: SampleRole - (*NodeIdentity)(nil), // 2: NodeIdentity - (*GroupLeader)(nil), // 3: GroupLeader - (*GetClusterOverviewRequest)(nil), // 4: GetClusterOverviewRequest - (*GetClusterOverviewResponse)(nil), // 5: GetClusterOverviewResponse - (*RaftGroupState)(nil), // 6: RaftGroupState - (*GetRaftGroupsRequest)(nil), // 7: GetRaftGroupsRequest - (*GetRaftGroupsResponse)(nil), // 8: GetRaftGroupsResponse - (*AdapterSummary)(nil), // 9: AdapterSummary - (*GetAdapterSummaryRequest)(nil), // 10: GetAdapterSummaryRequest - (*GetAdapterSummaryResponse)(nil), // 11: GetAdapterSummaryResponse - (*KeyVizRow)(nil), // 12: KeyVizRow - (*GetKeyVizMatrixRequest)(nil), // 13: GetKeyVizMatrixRequest - (*GetKeyVizMatrixResponse)(nil), // 14: GetKeyVizMatrixResponse - (*GetRouteDetailRequest)(nil), // 15: GetRouteDetailRequest - (*GetRouteDetailResponse)(nil), // 16: GetRouteDetailResponse - (*StreamEventsRequest)(nil), // 17: StreamEventsRequest - (*StreamEventsEvent)(nil), // 18: StreamEventsEvent - (*RouteTransition)(nil), // 19: RouteTransition - (*KeyVizColumn)(nil), // 20: KeyVizColumn + (KeyVizSeries)(0), // 0: KeyVizSeries + (SampleRole)(0), // 1: SampleRole + (*NodeIdentity)(nil), // 2: NodeIdentity + (*GroupLeader)(nil), // 3: GroupLeader + (*GetClusterOverviewRequest)(nil), // 4: GetClusterOverviewRequest + (*GetClusterOverviewResponse)(nil), // 5: GetClusterOverviewResponse + (*RaftGroupState)(nil), // 6: RaftGroupState + (*GetRaftGroupsRequest)(nil), // 7: GetRaftGroupsRequest + (*GetRaftGroupsResponse)(nil), // 8: GetRaftGroupsResponse + (*AdapterSummary)(nil), // 9: AdapterSummary + (*GetAdapterSummaryRequest)(nil), // 10: GetAdapterSummaryRequest + (*GetAdapterSummaryResponse)(nil), // 11: GetAdapterSummaryResponse + (*KeyVizRow)(nil), // 12: KeyVizRow + (*GetKeyVizMatrixRequest)(nil), // 13: GetKeyVizMatrixRequest + (*GetKeyVizMatrixResponse)(nil), // 14: GetKeyVizMatrixResponse + (*GetRouteDetailRequest)(nil), // 15: GetRouteDetailRequest + (*GetRouteDetailResponse)(nil), // 16: GetRouteDetailResponse + (*BackupShardApplied)(nil), // 17: BackupShardApplied + (*BackupExpectedKeys)(nil), // 18: BackupExpectedKeys + (*BeginBackupRequest)(nil), // 19: BeginBackupRequest + (*BeginBackupResponse)(nil), // 20: BeginBackupResponse + (*RenewBackupRequest)(nil), // 21: RenewBackupRequest + (*RenewBackupResponse)(nil), // 22: RenewBackupResponse + (*EndBackupRequest)(nil), // 23: EndBackupRequest + (*EndBackupResponse)(nil), // 24: EndBackupResponse + (*ListAdaptersAndScopesRequest)(nil), // 25: ListAdaptersAndScopesRequest + (*BackupScope)(nil), // 26: BackupScope + (*ListAdaptersAndScopesResponse)(nil), // 27: ListAdaptersAndScopesResponse + (*GetNodeVersionRequest)(nil), // 28: GetNodeVersionRequest + (*GetNodeVersionResponse)(nil), // 29: GetNodeVersionResponse + (*StreamEventsRequest)(nil), // 30: StreamEventsRequest + (*StreamEventsEvent)(nil), // 31: StreamEventsEvent + (*RouteTransition)(nil), // 32: RouteTransition + (*KeyVizColumn)(nil), // 33: KeyVizColumn } var file_admin_proto_depIdxs = []int32{ 2, // 0: GetClusterOverviewResponse.self:type_name -> NodeIdentity @@ -1520,27 +2210,40 @@ var file_admin_proto_depIdxs = []int32{ 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 + 17, // 10: BeginBackupResponse.shards:type_name -> BackupShardApplied + 18, // 11: BeginBackupResponse.expected_keys:type_name -> BackupExpectedKeys + 26, // 12: ListAdaptersAndScopesResponse.scopes:type_name -> BackupScope + 32, // 13: StreamEventsEvent.route_transition:type_name -> RouteTransition + 33, // 14: StreamEventsEvent.keyviz_column:type_name -> KeyVizColumn + 0, // 15: KeyVizColumn.series:type_name -> KeyVizSeries + 12, // 16: KeyVizColumn.rows:type_name -> KeyVizRow + 4, // 17: Admin.GetClusterOverview:input_type -> GetClusterOverviewRequest + 7, // 18: Admin.GetRaftGroups:input_type -> GetRaftGroupsRequest + 10, // 19: Admin.GetAdapterSummary:input_type -> GetAdapterSummaryRequest + 13, // 20: Admin.GetKeyVizMatrix:input_type -> GetKeyVizMatrixRequest + 15, // 21: Admin.GetRouteDetail:input_type -> GetRouteDetailRequest + 19, // 22: Admin.BeginBackup:input_type -> BeginBackupRequest + 21, // 23: Admin.RenewBackup:input_type -> RenewBackupRequest + 23, // 24: Admin.EndBackup:input_type -> EndBackupRequest + 25, // 25: Admin.ListAdaptersAndScopes:input_type -> ListAdaptersAndScopesRequest + 28, // 26: Admin.GetNodeVersion:input_type -> GetNodeVersionRequest + 30, // 27: Admin.StreamEvents:input_type -> StreamEventsRequest + 5, // 28: Admin.GetClusterOverview:output_type -> GetClusterOverviewResponse + 8, // 29: Admin.GetRaftGroups:output_type -> GetRaftGroupsResponse + 11, // 30: Admin.GetAdapterSummary:output_type -> GetAdapterSummaryResponse + 14, // 31: Admin.GetKeyVizMatrix:output_type -> GetKeyVizMatrixResponse + 16, // 32: Admin.GetRouteDetail:output_type -> GetRouteDetailResponse + 20, // 33: Admin.BeginBackup:output_type -> BeginBackupResponse + 22, // 34: Admin.RenewBackup:output_type -> RenewBackupResponse + 24, // 35: Admin.EndBackup:output_type -> EndBackupResponse + 27, // 36: Admin.ListAdaptersAndScopes:output_type -> ListAdaptersAndScopesResponse + 29, // 37: Admin.GetNodeVersion:output_type -> GetNodeVersionResponse + 31, // 38: Admin.StreamEvents:output_type -> StreamEventsEvent + 28, // [28:39] is the sub-list for method output_type + 17, // [17:28] is the sub-list for method input_type + 17, // [17:17] is the sub-list for extension type_name + 17, // [17:17] is the sub-list for extension extendee + 0, // [0:17] is the sub-list for field type_name } func init() { file_admin_proto_init() } @@ -1548,7 +2251,7 @@ func file_admin_proto_init() { if File_admin_proto != nil { return } - file_admin_proto_msgTypes[16].OneofWrappers = []any{ + file_admin_proto_msgTypes[29].OneofWrappers = []any{ (*StreamEventsEvent_RouteTransition)(nil), (*StreamEventsEvent_KeyvizColumn)(nil), } @@ -1558,7 +2261,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: 32, NumExtensions: 0, NumServices: 1, }, diff --git a/proto/admin.proto b/proto/admin.proto index aae54e45a..f6ae04836 100644 --- a/proto/admin.proto +++ b/proto/admin.proto @@ -12,6 +12,11 @@ service Admin { rpc GetAdapterSummary (GetAdapterSummaryRequest) returns (GetAdapterSummaryResponse) {} rpc GetKeyVizMatrix (GetKeyVizMatrixRequest) returns (GetKeyVizMatrixResponse) {} rpc GetRouteDetail (GetRouteDetailRequest) returns (GetRouteDetailResponse) {} + rpc BeginBackup (BeginBackupRequest) returns (BeginBackupResponse) {} + rpc RenewBackup (RenewBackupRequest) returns (RenewBackupResponse) {} + rpc EndBackup (EndBackupRequest) returns (EndBackupResponse) {} + rpc ListAdaptersAndScopes (ListAdaptersAndScopesRequest) returns (ListAdaptersAndScopesResponse) {} + rpc GetNodeVersion (GetNodeVersionRequest) returns (GetNodeVersionResponse) {} rpc StreamEvents (StreamEventsRequest) returns (stream StreamEventsEvent) {} } @@ -46,6 +51,9 @@ message RaftGroupState { // engine is a follower that has never heard from a leader); UIs should // render that case as "unknown" rather than "contacted at epoch". int64 last_contact_unix_ms = 6; + // leader_node_version is populated asynchronously from the leader's Admin + // GetNodeVersion RPC. Empty means unknown/unreachable/cache miss. + string leader_node_version = 7; } message GetRaftGroupsRequest {} @@ -154,6 +162,63 @@ message GetRouteDetailResponse { repeated AdapterSummary per_adapter = 2; } +message BackupShardApplied { + uint64 raft_group_id = 1; + uint64 applied_index = 2; +} + +message BackupExpectedKeys { + string adapter = 1; + string scope = 2; + uint64 key_count = 3; +} + +message BeginBackupRequest { + uint64 ttl_ms = 1; +} + +message BeginBackupResponse { + uint64 read_ts = 1; + bytes pin_token = 2; + uint64 ttl_ms_effective = 3; + repeated BackupShardApplied shards = 4; + repeated BackupExpectedKeys expected_keys = 5; +} + +message RenewBackupRequest { + bytes pin_token = 1; + uint64 ttl_ms = 2; +} + +message RenewBackupResponse { + uint64 ttl_ms_effective = 1; +} + +message EndBackupRequest { + bytes pin_token = 1; +} + +message EndBackupResponse {} + +message ListAdaptersAndScopesRequest { + bytes pin_token = 1; +} + +message BackupScope { + string adapter = 1; + string scope = 2; +} + +message ListAdaptersAndScopesResponse { + repeated BackupScope scopes = 1; +} + +message GetNodeVersionRequest {} + +message GetNodeVersionResponse { + string node_version = 1; +} + message StreamEventsRequest {} message StreamEventsEvent { diff --git a/proto/admin_grpc.pb.go b/proto/admin_grpc.pb.go index 021b1e834..06829837d 100644 --- a/proto/admin_grpc.pb.go +++ b/proto/admin_grpc.pb.go @@ -19,12 +19,17 @@ import ( const _ = grpc.SupportPackageIsVersion9 const ( - Admin_GetClusterOverview_FullMethodName = "/Admin/GetClusterOverview" - Admin_GetRaftGroups_FullMethodName = "/Admin/GetRaftGroups" - Admin_GetAdapterSummary_FullMethodName = "/Admin/GetAdapterSummary" - Admin_GetKeyVizMatrix_FullMethodName = "/Admin/GetKeyVizMatrix" - Admin_GetRouteDetail_FullMethodName = "/Admin/GetRouteDetail" - Admin_StreamEvents_FullMethodName = "/Admin/StreamEvents" + Admin_GetClusterOverview_FullMethodName = "/Admin/GetClusterOverview" + Admin_GetRaftGroups_FullMethodName = "/Admin/GetRaftGroups" + Admin_GetAdapterSummary_FullMethodName = "/Admin/GetAdapterSummary" + Admin_GetKeyVizMatrix_FullMethodName = "/Admin/GetKeyVizMatrix" + Admin_GetRouteDetail_FullMethodName = "/Admin/GetRouteDetail" + Admin_BeginBackup_FullMethodName = "/Admin/BeginBackup" + Admin_RenewBackup_FullMethodName = "/Admin/RenewBackup" + Admin_EndBackup_FullMethodName = "/Admin/EndBackup" + Admin_ListAdaptersAndScopes_FullMethodName = "/Admin/ListAdaptersAndScopes" + Admin_GetNodeVersion_FullMethodName = "/Admin/GetNodeVersion" + Admin_StreamEvents_FullMethodName = "/Admin/StreamEvents" ) // AdminClient is the client API for Admin service. @@ -41,6 +46,11 @@ type AdminClient interface { GetAdapterSummary(ctx context.Context, in *GetAdapterSummaryRequest, opts ...grpc.CallOption) (*GetAdapterSummaryResponse, error) GetKeyVizMatrix(ctx context.Context, in *GetKeyVizMatrixRequest, opts ...grpc.CallOption) (*GetKeyVizMatrixResponse, error) GetRouteDetail(ctx context.Context, in *GetRouteDetailRequest, opts ...grpc.CallOption) (*GetRouteDetailResponse, error) + BeginBackup(ctx context.Context, in *BeginBackupRequest, opts ...grpc.CallOption) (*BeginBackupResponse, error) + RenewBackup(ctx context.Context, in *RenewBackupRequest, opts ...grpc.CallOption) (*RenewBackupResponse, error) + EndBackup(ctx context.Context, in *EndBackupRequest, opts ...grpc.CallOption) (*EndBackupResponse, error) + ListAdaptersAndScopes(ctx context.Context, in *ListAdaptersAndScopesRequest, opts ...grpc.CallOption) (*ListAdaptersAndScopesResponse, error) + GetNodeVersion(ctx context.Context, in *GetNodeVersionRequest, opts ...grpc.CallOption) (*GetNodeVersionResponse, error) StreamEvents(ctx context.Context, in *StreamEventsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[StreamEventsEvent], error) } @@ -102,6 +112,56 @@ func (c *adminClient) GetRouteDetail(ctx context.Context, in *GetRouteDetailRequ return out, nil } +func (c *adminClient) BeginBackup(ctx context.Context, in *BeginBackupRequest, opts ...grpc.CallOption) (*BeginBackupResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(BeginBackupResponse) + err := c.cc.Invoke(ctx, Admin_BeginBackup_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *adminClient) RenewBackup(ctx context.Context, in *RenewBackupRequest, opts ...grpc.CallOption) (*RenewBackupResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(RenewBackupResponse) + err := c.cc.Invoke(ctx, Admin_RenewBackup_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *adminClient) EndBackup(ctx context.Context, in *EndBackupRequest, opts ...grpc.CallOption) (*EndBackupResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(EndBackupResponse) + err := c.cc.Invoke(ctx, Admin_EndBackup_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *adminClient) ListAdaptersAndScopes(ctx context.Context, in *ListAdaptersAndScopesRequest, opts ...grpc.CallOption) (*ListAdaptersAndScopesResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(ListAdaptersAndScopesResponse) + err := c.cc.Invoke(ctx, Admin_ListAdaptersAndScopes_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *adminClient) GetNodeVersion(ctx context.Context, in *GetNodeVersionRequest, opts ...grpc.CallOption) (*GetNodeVersionResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(GetNodeVersionResponse) + err := c.cc.Invoke(ctx, Admin_GetNodeVersion_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + func (c *adminClient) StreamEvents(ctx context.Context, in *StreamEventsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[StreamEventsEvent], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) stream, err := c.cc.NewStream(ctx, &Admin_ServiceDesc.Streams[0], Admin_StreamEvents_FullMethodName, cOpts...) @@ -135,6 +195,11 @@ type AdminServer interface { GetAdapterSummary(context.Context, *GetAdapterSummaryRequest) (*GetAdapterSummaryResponse, error) GetKeyVizMatrix(context.Context, *GetKeyVizMatrixRequest) (*GetKeyVizMatrixResponse, error) GetRouteDetail(context.Context, *GetRouteDetailRequest) (*GetRouteDetailResponse, error) + BeginBackup(context.Context, *BeginBackupRequest) (*BeginBackupResponse, error) + RenewBackup(context.Context, *RenewBackupRequest) (*RenewBackupResponse, error) + EndBackup(context.Context, *EndBackupRequest) (*EndBackupResponse, error) + ListAdaptersAndScopes(context.Context, *ListAdaptersAndScopesRequest) (*ListAdaptersAndScopesResponse, error) + GetNodeVersion(context.Context, *GetNodeVersionRequest) (*GetNodeVersionResponse, error) StreamEvents(*StreamEventsRequest, grpc.ServerStreamingServer[StreamEventsEvent]) error mustEmbedUnimplementedAdminServer() } @@ -161,6 +226,21 @@ func (UnimplementedAdminServer) GetKeyVizMatrix(context.Context, *GetKeyVizMatri func (UnimplementedAdminServer) GetRouteDetail(context.Context, *GetRouteDetailRequest) (*GetRouteDetailResponse, error) { return nil, status.Error(codes.Unimplemented, "method GetRouteDetail not implemented") } +func (UnimplementedAdminServer) BeginBackup(context.Context, *BeginBackupRequest) (*BeginBackupResponse, error) { + return nil, status.Error(codes.Unimplemented, "method BeginBackup not implemented") +} +func (UnimplementedAdminServer) RenewBackup(context.Context, *RenewBackupRequest) (*RenewBackupResponse, error) { + return nil, status.Error(codes.Unimplemented, "method RenewBackup not implemented") +} +func (UnimplementedAdminServer) EndBackup(context.Context, *EndBackupRequest) (*EndBackupResponse, error) { + return nil, status.Error(codes.Unimplemented, "method EndBackup not implemented") +} +func (UnimplementedAdminServer) ListAdaptersAndScopes(context.Context, *ListAdaptersAndScopesRequest) (*ListAdaptersAndScopesResponse, error) { + return nil, status.Error(codes.Unimplemented, "method ListAdaptersAndScopes not implemented") +} +func (UnimplementedAdminServer) GetNodeVersion(context.Context, *GetNodeVersionRequest) (*GetNodeVersionResponse, error) { + return nil, status.Error(codes.Unimplemented, "method GetNodeVersion not implemented") +} func (UnimplementedAdminServer) StreamEvents(*StreamEventsRequest, grpc.ServerStreamingServer[StreamEventsEvent]) error { return status.Error(codes.Unimplemented, "method StreamEvents not implemented") } @@ -275,6 +355,96 @@ func _Admin_GetRouteDetail_Handler(srv interface{}, ctx context.Context, dec fun return interceptor(ctx, in, info, handler) } +func _Admin_BeginBackup_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(BeginBackupRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AdminServer).BeginBackup(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Admin_BeginBackup_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(AdminServer).BeginBackup(ctx, req.(*BeginBackupRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Admin_RenewBackup_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(RenewBackupRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AdminServer).RenewBackup(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Admin_RenewBackup_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(AdminServer).RenewBackup(ctx, req.(*RenewBackupRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Admin_EndBackup_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(EndBackupRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AdminServer).EndBackup(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Admin_EndBackup_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(AdminServer).EndBackup(ctx, req.(*EndBackupRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Admin_ListAdaptersAndScopes_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(ListAdaptersAndScopesRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AdminServer).ListAdaptersAndScopes(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Admin_ListAdaptersAndScopes_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(AdminServer).ListAdaptersAndScopes(ctx, req.(*ListAdaptersAndScopesRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Admin_GetNodeVersion_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(GetNodeVersionRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AdminServer).GetNodeVersion(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Admin_GetNodeVersion_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(AdminServer).GetNodeVersion(ctx, req.(*GetNodeVersionRequest)) + } + return interceptor(ctx, in, info, handler) +} + func _Admin_StreamEvents_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(StreamEventsRequest) if err := stream.RecvMsg(m); err != nil { @@ -313,6 +483,26 @@ var Admin_ServiceDesc = grpc.ServiceDesc{ MethodName: "GetRouteDetail", Handler: _Admin_GetRouteDetail_Handler, }, + { + MethodName: "BeginBackup", + Handler: _Admin_BeginBackup_Handler, + }, + { + MethodName: "RenewBackup", + Handler: _Admin_RenewBackup_Handler, + }, + { + MethodName: "EndBackup", + Handler: _Admin_EndBackup_Handler, + }, + { + MethodName: "ListAdaptersAndScopes", + Handler: _Admin_ListAdaptersAndScopes_Handler, + }, + { + MethodName: "GetNodeVersion", + Handler: _Admin_GetNodeVersion_Handler, + }, }, Streams: []grpc.StreamDesc{ {