diff --git a/internal/controllers/raft_status_controller.go b/internal/controllers/raft_status_controller.go new file mode 100644 index 0000000..a3752c6 --- /dev/null +++ b/internal/controllers/raft_status_controller.go @@ -0,0 +1,69 @@ +package controllers + +import ( + "net/http" + + "github.com/gin-gonic/gin" +) + +// RaftGroupStatus is the local mirror of the pkg/app.RaftGroupStatus +// interface, kept here to avoid importing pkg/app from internal/. +// Both types satisfy each other structurally because Go interfaces +// are nominal-by-method. +type RaftGroupStatus interface { + IsLeader() bool + SelfID() string + BindAddr() string + LeaderInfo() (id, addr string) +} + +// raftStatusController exposes the local node's per-shard raft state. +// In raft mode there is one RaftGroupStatus per Pebble shard; without +// raft the controller returns enabled=false and an empty groups list. +type raftStatusController struct { + groups []RaftGroupStatus +} + +// NewRaftStatusController returns a controller that surfaces the given +// per-shard raft groups. Pass nil/empty for non-raft deployments. +func NewRaftStatusController(groups []RaftGroupStatus) *raftStatusController { + return &raftStatusController{groups: groups} +} + +type raftGroupView struct { + ShardIdx int `json:"shardIdx"` + IsLeader bool `json:"isLeader"` + SelfID string `json:"selfId"` + SelfAddr string `json:"selfAddr"` + LeaderID string `json:"leaderId"` + LeaderAddr string `json:"leaderAddr"` + HasLeader bool `json:"hasLeader"` +} + +type raftStatusResponse struct { + Enabled bool `json:"enabled"` + NumGroups int `json:"numGroups"` + Groups []raftGroupView `json:"groups,omitempty"` +} + +// Handle answers GET /v1/codeq/raft/status. Returns 200 always — the +// payload is the source of truth (enabled=false ⇒ no raft on this node). +func (h *raftStatusController) Handle(c *gin.Context) { + resp := raftStatusResponse{ + Enabled: len(h.groups) > 0, + NumGroups: len(h.groups), + } + for i, g := range h.groups { + leaderID, leaderAddr := g.LeaderInfo() + resp.Groups = append(resp.Groups, raftGroupView{ + ShardIdx: i, + IsLeader: g.IsLeader(), + SelfID: g.SelfID(), + SelfAddr: g.BindAddr(), + LeaderID: leaderID, + LeaderAddr: leaderAddr, + HasLeader: leaderID != "", + }) + } + c.JSON(http.StatusOK, resp) +} diff --git a/internal/raft/db.go b/internal/raft/db.go index f337035..a37044b 100644 --- a/internal/raft/db.go +++ b/internal/raft/db.go @@ -331,6 +331,24 @@ func (d *DB) IsLeader() bool { // becomes the leader, false when it loses leadership. func (d *DB) LeaderObservation() <-chan bool { return d.leaderCh } +// LeaderInfo returns the current leader's id and bind address according +// to local state. Both are empty if no leader is known (election in +// progress). This is the local view — under network partition it can +// disagree with other nodes for a brief window. +func (d *DB) LeaderInfo() (id, addr string) { + if d.raft == nil { + return "", "" + } + rawAddr, rawID := d.raft.LeaderWithID() + return string(rawID), string(rawAddr) +} + +// SelfID returns the configured raft ServerID for this node. +func (d *DB) SelfID() string { return d.cfg.SelfID } + +// BindAddr returns the raft bind address configured for this node. +func (d *DB) BindAddr() string { return d.cfg.BindAddr } + // WaitLeader blocks until this node IS the leader, or ctx is done. // Useful in tests that bootstrap and then need to write. func (d *DB) WaitLeader(ctx context.Context) error { diff --git a/pkg/app/application.go b/pkg/app/application.go index 818bfcf..4b6e0ea 100644 --- a/pkg/app/application.go +++ b/pkg/app/application.go @@ -24,6 +24,17 @@ import ( "github.com/go-redis/redis/v8" ) +// RaftGroupStatus is the small slice of raft state the status endpoint +// + observability surfaces need. *internal/raft.DB satisfies it. +// Defined here (not in the raft package) so the public Application +// type avoids importing internal/raft transitively. +type RaftGroupStatus interface { + IsLeader() bool + SelfID() string + BindAddr() string + LeaderInfo() (id, addr string) +} + type Application struct { Config *config.Config Engine *gin.Engine @@ -35,7 +46,10 @@ type Application struct { ProducerValidator auth.Validator WorkerValidator auth.Validator RateLimiter ratelimit.Limiter - TracingShutdown func(context.Context) error + // RaftGroups, when non-nil, is the per-shard raft state in raft + // mode. Index = shardIdx. Empty when raft is disabled. + RaftGroups []RaftGroupStatus + TracingShutdown func(context.Context) error } // ApplicationOption configures the Application diff --git a/pkg/app/application_pebble.go b/pkg/app/application_pebble.go index 5d2b0cd..ee0542d 100644 --- a/pkg/app/application_pebble.go +++ b/pkg/app/application_pebble.go @@ -435,6 +435,14 @@ func newPebbleApplication( TZ: loc, RateLimiter: limiter, } + if cfg.Raft.Enabled { + app.RaftGroups = make([]RaftGroupStatus, 0, len(raftNodes)) + for _, r := range raftNodes { + if r != nil { + app.RaftGroups = append(app.RaftGroups, r) + } + } + } cleanupStartupFailure := func() { bgCancel() diff --git a/pkg/app/raft_status_endpoint_test.go b/pkg/app/raft_status_endpoint_test.go new file mode 100644 index 0000000..4482e50 --- /dev/null +++ b/pkg/app/raft_status_endpoint_test.go @@ -0,0 +1,236 @@ +package app + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + _ "github.com/osvaldoandrade/codeq/pkg/auth/static" + "github.com/osvaldoandrade/codeq/pkg/config" +) + +// raftStatusResp mirrors the controller's JSON shape so the test can +// decode it without importing the internal controllers package. +type raftStatusResp struct { + Enabled bool `json:"enabled"` + NumGroups int `json:"numGroups"` + Groups []struct { + ShardIdx int `json:"shardIdx"` + IsLeader bool `json:"isLeader"` + SelfID string `json:"selfId"` + SelfAddr string `json:"selfAddr"` + LeaderID string `json:"leaderId"` + LeaderAddr string `json:"leaderAddr"` + HasLeader bool `json:"hasLeader"` + } `json:"groups"` +} + +func TestRaftStatusEndpoint_RaftDisabled(t *testing.T) { + pcfg, _ := json.Marshal(map[string]any{"path": t.TempDir() + "/pebble"}) + cfg := &config.Config{ + Port: 0, + Timezone: "UTC", + LogLevel: "error", + LogFormat: "json", + Env: "dev", + DefaultLeaseSeconds: 60, + RequeueInspectLimit: 50, + LocalArtifactsDir: t.TempDir(), + MaxAttemptsDefault: 5, + BackoffPolicy: "fixed", + BackoffBaseSeconds: 1, + BackoffMaxSeconds: 3, + WebhookHmacSecret: "secret", + WorkerAudience: "codeq-worker", + SubscriptionMinIntervalSeconds: 5, + SubscriptionCleanupIntervalSeconds: 60, + ResultWebhookMaxAttempts: 1, + ResultWebhookBaseBackoffSeconds: 1, + ResultWebhookMaxBackoffSeconds: 2, + ProducerAuthProvider: "static", + ProducerAuthConfig: json.RawMessage(`{"token":"dev-token","subject":"producer-dev","email":"dev@codeq.local","raw":{"role":"ADMIN","tenantId":"dev-tenant"}}`), + WorkerAuthProvider: "static", + WorkerAuthConfig: json.RawMessage(`{"token":"dev-token","subject":"worker-dev","scopes":["codeq:claim","codeq:heartbeat","codeq:abandon","codeq:nack","codeq:result","codeq:subscribe"],"eventTypes":["*"],"raw":{"tenantId":"dev-tenant"}}`), + PersistenceProvider: "pebble", + PersistenceConfig: pcfg, + RedisAddr: "127.0.0.1:0", + } + if err := cfg.Validate(); err != nil { + t.Fatalf("validate: %v", err) + } + app, err := NewApplication(cfg) + if err != nil { + t.Fatalf("NewApplication: %v", err) + } + defer func() { + if app.TracingShutdown != nil { + _ = app.TracingShutdown(context.Background()) + } + }() + SetupMappings(app) + srv := httptest.NewServer(app.Engine) + defer srv.Close() + + resp := getStatus(t, srv.URL) + if resp.Enabled { + t.Errorf("raft disabled: want Enabled=false, got true") + } + if resp.NumGroups != 0 { + t.Errorf("raft disabled: want NumGroups=0, got %d", resp.NumGroups) + } +} + +func TestRaftStatusEndpoint_SingleShard_Leader(t *testing.T) { + d := openSingleNodeRaftApp(t) + defer d.cleanup() + + // Wait for the lone node to elect. + deadline := time.Now().Add(3 * time.Second) + var resp raftStatusResp + for time.Now().Before(deadline) { + resp = getStatus(t, d.server.URL) + if len(resp.Groups) == 1 && resp.Groups[0].IsLeader { + break + } + time.Sleep(30 * time.Millisecond) + } + if !resp.Enabled { + t.Fatalf("want Enabled=true, got false: %+v", resp) + } + if resp.NumGroups != 1 { + t.Errorf("want NumGroups=1, got %d", resp.NumGroups) + } + if len(resp.Groups) != 1 { + t.Fatalf("want 1 group, got %d", len(resp.Groups)) + } + g := resp.Groups[0] + if !g.IsLeader { + t.Errorf("single-node should be leader, got isLeader=false") + } + if g.SelfID != "node-1" { + t.Errorf("SelfID: want node-1, got %q", g.SelfID) + } + if !g.HasLeader || g.LeaderID != "node-1" { + t.Errorf("LeaderID: want node-1 + HasLeader=true, got %q + %v", g.LeaderID, g.HasLeader) + } +} + +type singleNodeRaftApp struct { + app *Application + server *httptest.Server + cleanup func() +} + +func openSingleNodeRaftApp(t *testing.T) singleNodeRaftApp { + t.Helper() + port := pickContiguousFreePorts(t, 1) + pcfg, _ := json.Marshal(map[string]any{"path": t.TempDir() + "/pebble"}) + cfg := &config.Config{ + Port: 0, + Timezone: "UTC", + LogLevel: "error", + LogFormat: "json", + Env: "dev", + DefaultLeaseSeconds: 60, + RequeueInspectLimit: 50, + LocalArtifactsDir: t.TempDir(), + MaxAttemptsDefault: 5, + BackoffPolicy: "fixed", + BackoffBaseSeconds: 1, + BackoffMaxSeconds: 3, + WebhookHmacSecret: "secret", + WorkerAudience: "codeq-worker", + SubscriptionMinIntervalSeconds: 5, + SubscriptionCleanupIntervalSeconds: 60, + ResultWebhookMaxAttempts: 1, + ResultWebhookBaseBackoffSeconds: 1, + ResultWebhookMaxBackoffSeconds: 2, + ProducerAuthProvider: "static", + ProducerAuthConfig: json.RawMessage(`{"token":"dev-token","subject":"producer-dev","email":"dev@codeq.local","raw":{"role":"ADMIN","tenantId":"dev-tenant"}}`), + WorkerAuthProvider: "static", + WorkerAuthConfig: json.RawMessage(`{"token":"dev-token","subject":"worker-dev","scopes":["codeq:claim","codeq:heartbeat","codeq:abandon","codeq:nack","codeq:result","codeq:subscribe"],"eventTypes":["*"],"raw":{"tenantId":"dev-tenant"}}`), + PersistenceProvider: "pebble", + PersistenceConfig: pcfg, + RedisAddr: "127.0.0.1:0", + Raft: config.RaftConfig{ + Enabled: true, + SelfID: "node-1", + BindAddr: "127.0.0.1:" + portString(port), + Bootstrap: true, + HeartbeatMS: 50, + ElectionMS: 50, + LeaderLeaseMS: 50, + CommitMS: 10, + ApplyTimeoutSeconds: 2, + }, + } + if err := cfg.Validate(); err != nil { + t.Fatalf("validate: %v", err) + } + app, err := NewApplication(cfg) + if err != nil { + t.Fatalf("NewApplication: %v", err) + } + SetupMappings(app) + srv := httptest.NewServer(app.Engine) + return singleNodeRaftApp{ + app: app, + server: srv, + cleanup: func() { + srv.Close() + if app.TracingShutdown != nil { + _ = app.TracingShutdown(context.Background()) + } + }, + } +} + +func getStatus(t *testing.T, baseURL string) raftStatusResp { + t.Helper() + req, _ := http.NewRequest(http.MethodGet, baseURL+"/v1/codeq/raft/status", nil) + req.Header.Set("Authorization", "Bearer dev-token") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("status request: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status: want 200, got %d", resp.StatusCode) + } + var out raftStatusResp + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + t.Fatalf("decode: %v", err) + } + return out +} + +func portString(p int) string { + // fmt.Sprintf("%d") with the strconv route avoids pulling in fmt + // for one call when the rest of the file uses %s formatting. + return intToString(p) +} + +func intToString(n int) string { + if n == 0 { + return "0" + } + negative := n < 0 + if negative { + n = -n + } + var buf [20]byte + i := len(buf) + for n > 0 { + i-- + buf[i] = byte('0' + n%10) + n /= 10 + } + if negative { + i-- + buf[i] = '-' + } + return string(buf[i:]) +} diff --git a/pkg/app/url_mappings.go b/pkg/app/url_mappings.go index d2bd40a..a8f8d92 100644 --- a/pkg/app/url_mappings.go +++ b/pkg/app/url_mappings.go @@ -32,6 +32,12 @@ func SetupMappings(app *Application) { anyAuth.GET("/tasks/:id", controllers.NewGetTaskController(app.Scheduler).Handle) anyAuth.GET("/tasks/:id/result", controllers.NewGetResultController(app.Results).Handle) + // Raft status — local-node view of per-shard leadership. + // Public (anyAuth) so ops tooling and Prometheus scrapers + // can poll without an admin token. The payload reveals no + // task data, only routing metadata (peer IDs + bind addrs). + anyAuth.GET("/raft/status", controllers.NewRaftStatusController(adaptRaftGroups(app.RaftGroups)).Handle) + admin := producer.Group("/admin", middleware.RequireAdmin()) admin.GET("/queues", controllers.NewQueuesAdminController(app.Scheduler).Handle) admin.GET("/queues/:command", controllers.NewQueueStatsController(app.Scheduler).Handle) @@ -40,3 +46,17 @@ func SetupMappings(app *Application) { admin.POST("/tasks/cleanup", middleware.RateLimitAdminCleanup(app.RateLimiter, app.Config), controllers.NewCleanupExpiredController(app.Scheduler).Handle) } } + +// adaptRaftGroups converts the public app.RaftGroupStatus slice to the +// controllers package's mirror interface. Both have the same method +// set so the conversion is a no-op wrapper. +func adaptRaftGroups(in []RaftGroupStatus) []controllers.RaftGroupStatus { + if len(in) == 0 { + return nil + } + out := make([]controllers.RaftGroupStatus, len(in)) + for i, g := range in { + out[i] = g + } + return out +}