diff --git a/docs/bg/distributed.md b/docs/bg/distributed.md index 2fe135f..0d025b1 100644 --- a/docs/bg/distributed.md +++ b/docs/bg/distributed.md @@ -23,6 +23,8 @@ Leader election и log репликация през TCP. Включване: **Log compaction (v1):** след apply node-ът може да изреже safe prefix, когато log-ът надхвърли `BARADB_RAFT_LOG_MAX_ENTRIES`. Leader не реже след matchIndex на peer (catch-up с AppendEntries). Snapshot metadata се пази в `raft_state.bin`. +**Metrics:** при включен raft `GET /metrics` (HTTP = `BARADB_PORT + 440`) дава Prometheus редове: `baradb_raft_is_leader`, `baradb_raft_term`, `baradb_raft_log_entries`, `baradb_raft_apply_lag`, `baradb_raft_commit_wait_ms_total`, `baradb_raft_elections_total`, `baradb_raft_forwards_total`, `baradb_raft_compactions_total`. `GET /health` включва обект `raft` (`role`, `term`, `leader_id`, …). + ```nim import barabadb/core/raft diff --git a/docs/en/distributed.md b/docs/en/distributed.md index 8d43fe8..5530467 100644 --- a/docs/en/distributed.md +++ b/docs/en/distributed.md @@ -23,6 +23,8 @@ When Raft is enabled, SQL DML (`INSERT`/`UPDATE`/`DELETE`/`MERGE` and transactio **Log compaction (v1):** after apply, each node may drop a fully-safe log prefix once `log.len` exceeds `BARADB_RAFT_LOG_MAX_ENTRIES`. The leader never discards past any peer's `matchIndex` (so lagging followers still catch up via AppendEntries). Snapshot metadata (`lastSnapshotIndex`/`Term`) is persisted in `raft_state.bin`; full InstallSnapshot state-machine payloads are not required while this safe-prefix policy holds. +**Metrics:** with raft enabled, `GET /metrics` (HTTP port = `BARADB_PORT + 440`) includes Prometheus lines such as `baradb_raft_is_leader`, `baradb_raft_term`, `baradb_raft_log_entries`, `baradb_raft_apply_lag`, `baradb_raft_commit_wait_ms_total`, `baradb_raft_elections_total`, `baradb_raft_forwards_total`, and `baradb_raft_compactions_total`. `GET /health` embeds a `raft` object (`role`, `term`, `leader_id`, `commit_index`, `apply_lag`, …). + ```nim import barabadb/core/raft diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim index 8ca90fe..f142ed5 100644 --- a/src/barabadb/core/httpserver.nim +++ b/src/barabadb/core/httpserver.nim @@ -24,6 +24,7 @@ import ../protocol/auth import ../protocol/ratelimit import ../core/registry import ../core/backup +import ../core/raft type HttpServer* = ref object @@ -37,6 +38,8 @@ type authManager*: AuthManager rateLimiter*: RateLimiter ws*: WsServer + ## Optional live raft node for /metrics and /health (set from main). + raftNode*: RaftNode Metrics* = ref object queriesTotal*: int @@ -254,21 +257,47 @@ proc queryHandler(server: HttpServer): RequestHandler = server.metrics.queryErrors += 1 ctx.json(%*{"error": errMsg}, 400) -proc healthHandler(): RequestHandler = +proc healthHandler(server: HttpServer): RequestHandler = return proc(request: Request) {.gcsafe.} = let ctx = newContext(request) - ctx.json(%*{"status": "ok", "version": "1.1.6"}) + var body = %*{ + "status": "ok", + "version": "1.1.6" + } + if server.raftNode != nil: + let n = server.raftNode + let role = case n.state + of rsLeader: "leader" + of rsCandidate: "candidate" + of rsFollower: "follower" + body["raft"] = %*{ + "enabled": true, + "node_id": n.id, + "role": role, + "term": n.currentTerm, + "leader_id": n.leaderId, + "commit_index": n.commitIndex, + "last_applied": n.lastApplied, + "apply_lag": n.applyLag, + "log_entries": n.log.len, + "snapshot_index": n.lastSnapshotIndex + } + else: + body["raft"] = %*{"enabled": false} + ctx.json(body) proc metricsHandler(server: HttpServer): RequestHandler = return proc(request: Request) {.gcsafe.} = let ctx = newContext(request) if not server.checkAuth(request, ctx): return - let prometheus = "baradb_queries_total " & $server.metrics.queriesTotal & "\n" & + var prometheus = "baradb_queries_total " & $server.metrics.queriesTotal & "\n" & "baradb_query_errors_total " & $server.metrics.queryErrors & "\n" & "baradb_inserts_total " & $server.metrics.insertCount & "\n" & "baradb_selects_total " & $server.metrics.selectCount & "\n" & "baradb_connections_active " & $server.metrics.activeConnections & "\n" + if server.raftNode != nil: + prometheus.add(server.raftNode.prometheusText()) request.respond(200, @[("Content-Type", "text/plain; charset=utf-8")], prometheus) proc authHandler(server: HttpServer): RequestHandler = @@ -886,7 +915,7 @@ proc run*(server: HttpServer, port: int = 9470) = router.get("/admin", server.adminHandler()) router.get("/", server.adminHandler()) router.post("/query", server.queryHandler()) - router.get("/health", healthHandler()) + router.get("/health", server.healthHandler()) router.get("/metrics", server.metricsHandler()) router.post("/auth", server.authHandler()) router.post("/auth/scram/start", server.scramStartHandler()) diff --git a/src/barabadb/core/raft.nim b/src/barabadb/core/raft.nim index 30a6c73..11e3f60 100644 --- a/src/barabadb/core/raft.nim +++ b/src/barabadb/core/raft.nim @@ -25,6 +25,21 @@ type command*: string data*: seq[byte] + ## Counters / gauges for Prometheus (/metrics). Updated on the raft/async + ## path; HTTP reads them without locks (best-effort consistency). + RaftMetrics* = ref object + electionsTotal*: int64 # times this node became leader + termChangesTotal*: int64 # currentTerm increases + appendsTotal*: int64 # appendLog successes + commitWaitsTotal*: int64 # successful wait-for-commit finishes + commitWaitMsTotal*: int64 # sum of wait durations (ms) + commitTimeoutsTotal*: int64 # raft commit timeout + lostLeadershipTotal*: int64 # append returned index 0 + forwardsTotal*: int64 # follower→leader SQL forwards + forwardErrorsTotal*: int64 # failed forwards + appliesTotal*: int64 # applyCommand invocations + compactionsTotal*: int64 # compactLog that actually dropped entries + RaftNode* = ref object id*: string state*: RaftState @@ -41,6 +56,7 @@ type lastSnapshotTerm*: uint64 ## Trigger compaction when log.len exceeds this (0 = default 256). logMaxEntries*: int + metrics*: RaftMetrics # State machine callback applyCommand*: proc(cmd: string, data: seq[byte]) {.gcsafe.} # Distributed transaction callbacks (for raft→disttxn integration) @@ -172,6 +188,7 @@ proc newRaftNode*(id: string, peers: seq[string], raftPort: int = 0, lastSnapshotIndex: 0, lastSnapshotTerm: 0, logMaxEntries: 256, + metrics: RaftMetrics(), nextIndex: initTable[string, uint64](), matchIndex: initTable[string, uint64](), peers: peers, @@ -254,6 +271,8 @@ proc compactLog*(node: RaftNode) = node.lastApplied = node.lastSnapshotIndex if node.commitIndex < node.lastSnapshotIndex: node.commitIndex = node.lastSnapshotIndex + if node.metrics != nil: + inc node.metrics.compactionsTotal node.saveState() proc applyCommitted(node: RaftNode) = @@ -280,9 +299,13 @@ proc applyCommitted(node: RaftNode) = else: if node.applyCommand != nil: node.applyCommand(entry.command, entry.data) + if node.metrics != nil: + inc node.metrics.appliesTotal node.compactLog() proc becomeFollower*(node: RaftNode, term: uint64) = + if term > node.currentTerm and node.metrics != nil: + inc node.metrics.termChangesTotal node.state = rsFollower node.currentTerm = term node.votedFor = "" @@ -294,6 +317,8 @@ proc becomeFollower*(node: RaftNode, term: uint64) = proc becomeCandidate*(node: RaftNode) = node.state = rsCandidate inc node.currentTerm + if node.metrics != nil: + inc node.metrics.termChangesTotal node.votedFor = node.id node.votesReceived.clear() node.votesReceived.incl(node.id) @@ -302,6 +327,8 @@ proc becomeCandidate*(node: RaftNode) = proc becomeLeader*(node: RaftNode) = node.state = rsLeader node.leaderId = node.id + if node.metrics != nil: + inc node.metrics.electionsTotal info("Raft node " & node.id & " became leader for term " & $node.currentTerm) for peer in node.peers: node.nextIndex[peer] = node.lastLogIndex + 1 @@ -437,6 +464,8 @@ proc appendLog*(node: RaftNode, command: string, data: seq[byte] = @[]): LogEntr data: data, ) node.log.add(result) + if node.metrics != nil: + inc node.metrics.appendsTotal node.saveState() proc handleVoteReply*(node: RaftNode, reply: RaftMessage) = @@ -506,6 +535,81 @@ proc isLeader*(node: RaftNode): bool = node.state == rsLeader proc leaderId*(node: RaftNode): string = node.leaderId proc logLen*(node: RaftNode): int = node.log.len +proc applyLag*(node: RaftNode): uint64 = + ## commitIndex - lastApplied (0 when caught up). + if node.commitIndex > node.lastApplied: + return node.commitIndex - node.lastApplied + return 0 + +proc prometheusText*(node: RaftNode): string = + ## Prometheus exposition lines for this raft node (gauges + counters). + let m = if node.metrics != nil: node.metrics else: RaftMetrics() + let isLead = if node.isLeader: 1 else: 0 + let role = case node.state + of rsLeader: "leader" + of rsCandidate: "candidate" + of rsFollower: "follower" + result = "" + result.add("# HELP baradb_raft_is_leader 1 if this node is the raft leader\n") + result.add("# TYPE baradb_raft_is_leader gauge\n") + result.add("baradb_raft_is_leader{node=\"" & node.id & "\",role=\"" & role & "\"} " & $isLead & "\n") + result.add("# HELP baradb_raft_term Current raft term\n") + result.add("# TYPE baradb_raft_term gauge\n") + result.add("baradb_raft_term{node=\"" & node.id & "\"} " & $node.currentTerm & "\n") + result.add("# HELP baradb_raft_log_entries In-memory raft log length\n") + result.add("# TYPE baradb_raft_log_entries gauge\n") + result.add("baradb_raft_log_entries{node=\"" & node.id & "\"} " & $node.log.len & "\n") + result.add("# HELP baradb_raft_commit_index Raft commit index\n") + result.add("# TYPE baradb_raft_commit_index gauge\n") + result.add("baradb_raft_commit_index{node=\"" & node.id & "\"} " & $node.commitIndex & "\n") + result.add("# HELP baradb_raft_last_applied Raft lastApplied index\n") + result.add("# TYPE baradb_raft_last_applied gauge\n") + result.add("baradb_raft_last_applied{node=\"" & node.id & "\"} " & $node.lastApplied & "\n") + result.add("# HELP baradb_raft_apply_lag commitIndex - lastApplied\n") + result.add("# TYPE baradb_raft_apply_lag gauge\n") + result.add("baradb_raft_apply_lag{node=\"" & node.id & "\"} " & $node.applyLag & "\n") + result.add("# HELP baradb_raft_snapshot_index lastSnapshotIndex (compacted base)\n") + result.add("# TYPE baradb_raft_snapshot_index gauge\n") + result.add("baradb_raft_snapshot_index{node=\"" & node.id & "\"} " & $node.lastSnapshotIndex & "\n") + result.add("# HELP baradb_raft_elections_total Times this node became leader\n") + result.add("# TYPE baradb_raft_elections_total counter\n") + result.add("baradb_raft_elections_total{node=\"" & node.id & "\"} " & $m.electionsTotal & "\n") + result.add("# HELP baradb_raft_term_changes_total Term increases observed\n") + result.add("# TYPE baradb_raft_term_changes_total counter\n") + result.add("baradb_raft_term_changes_total{node=\"" & node.id & "\"} " & $m.termChangesTotal & "\n") + result.add("# HELP baradb_raft_appends_total Log appends on this node\n") + result.add("# TYPE baradb_raft_appends_total counter\n") + result.add("baradb_raft_appends_total{node=\"" & node.id & "\"} " & $m.appendsTotal & "\n") + result.add("# HELP baradb_raft_commit_waits_total Successful wait-for-commit completions\n") + result.add("# TYPE baradb_raft_commit_waits_total counter\n") + result.add("baradb_raft_commit_waits_total{node=\"" & node.id & "\"} " & $m.commitWaitsTotal & "\n") + result.add("# HELP baradb_raft_commit_wait_ms_total Sum of commit-wait durations in ms\n") + result.add("# TYPE baradb_raft_commit_wait_ms_total counter\n") + result.add("baradb_raft_commit_wait_ms_total{node=\"" & node.id & "\"} " & $m.commitWaitMsTotal & "\n") + result.add("# HELP baradb_raft_commit_timeouts_total Raft commit wait timeouts\n") + result.add("# TYPE baradb_raft_commit_timeouts_total counter\n") + result.add("baradb_raft_commit_timeouts_total{node=\"" & node.id & "\"} " & $m.commitTimeoutsTotal & "\n") + result.add("# HELP baradb_raft_lost_leadership_total Appends rejected (not leader)\n") + result.add("# TYPE baradb_raft_lost_leadership_total counter\n") + result.add("baradb_raft_lost_leadership_total{node=\"" & node.id & "\"} " & $m.lostLeadershipTotal & "\n") + result.add("# HELP baradb_raft_forwards_total Follower SQL forwards to leader\n") + result.add("# TYPE baradb_raft_forwards_total counter\n") + result.add("baradb_raft_forwards_total{node=\"" & node.id & "\"} " & $m.forwardsTotal & "\n") + result.add("# HELP baradb_raft_forward_errors_total Failed leader forwards\n") + result.add("# TYPE baradb_raft_forward_errors_total counter\n") + result.add("baradb_raft_forward_errors_total{node=\"" & node.id & "\"} " & $m.forwardErrorsTotal & "\n") + result.add("# HELP baradb_raft_applies_total State-machine applyCommand calls\n") + result.add("# TYPE baradb_raft_applies_total counter\n") + result.add("baradb_raft_applies_total{node=\"" & node.id & "\"} " & $m.appliesTotal & "\n") + result.add("# HELP baradb_raft_compactions_total Log prefix compactions\n") + result.add("# TYPE baradb_raft_compactions_total counter\n") + result.add("baradb_raft_compactions_total{node=\"" & node.id & "\"} " & $m.compactionsTotal & "\n") + if m.commitWaitsTotal > 0: + let avg = m.commitWaitMsTotal div m.commitWaitsTotal + result.add("# HELP baradb_raft_commit_wait_ms_avg Average commit-wait latency (ms)\n") + result.add("# TYPE baradb_raft_commit_wait_ms_avg gauge\n") + result.add("baradb_raft_commit_wait_ms_avg{node=\"" & node.id & "\"} " & $avg & "\n") + # Leader election timer loop type ElectionTimer* = ref object diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index c23ec75..cc6f848 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -292,11 +292,18 @@ proc forwardQueryToLeader*(host: string, port: int, query: string, try: sock.close() except CatchableError: discard proc waitRaftCommit(node: RaftNode, lastIdx: uint64, timeoutMs: int): Future[(bool, string)] {.async.} = - let deadline = getMonoTime() + initDuration(milliseconds = timeoutMs) + let start = getMonoTime() + let deadline = start + initDuration(milliseconds = timeoutMs) while node.commitIndex < lastIdx and getMonoTime() < deadline: await sleepAsync(10) + let waitedMs = int64((getMonoTime() - start).inMilliseconds) if node.commitIndex < lastIdx: + if node.metrics != nil: + inc node.metrics.commitTimeoutsTotal return (false, "raft commit timeout") + if node.metrics != nil: + inc node.metrics.commitWaitsTotal + node.metrics.commitWaitMsTotal += waitedMs return (true, "") proc appendWriteToRaft*(node: RaftNode, kvPairs: seq[(string, seq[byte])], @@ -316,6 +323,8 @@ proc appendWriteToRaft*(node: RaftNode, kvPairs: seq[(string, seq[byte])], else: node.appendLog("delete", cast[seq[byte]](key)) if entry.index == 0: + if node.metrics != nil: + inc node.metrics.lostLeadershipTotal return (false, "lost leadership during raft append") lastIdx = entry.index return await waitRaftCommit(node, lastIdx, timeoutMs) @@ -327,6 +336,8 @@ proc appendDdlToRaft*(node: RaftNode, sql: string, ## MUST be called outside the storage gate (same as appendWriteToRaft). let entry = node.appendLog("ddl", cast[seq[byte]](sql)) if entry.index == 0: + if node.metrics != nil: + inc node.metrics.lostLeadershipTotal return (false, "lost leadership during raft append") return await waitRaftCommit(node, entry.index, timeoutMs) @@ -434,8 +445,12 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq return (false, QueryResult(), e.msg) # Follower write/DDL: proxy to leader SQL port (outside the storage gate). if needsForward: - return await forwardQueryToLeader(forwardHost, forwardPort, query, params, - raftWriteTimeoutMs) + let (okF, qrF, errF) = await forwardQueryToLeader(forwardHost, forwardPort, + query, params, raftWriteTimeoutMs) + if raftNode != nil and raftNode.metrics != nil: + if okF: inc raftNode.metrics.forwardsTotal + else: inc raftNode.metrics.forwardErrorsTotal + return (okF, qrF, errF) # Raft log append + majority wait (outside the storage gate). # DDL batches ship the original SQL once (re-executed on apply). Pure DML # ships KV pairs. Mixed DDL+DML in one query uses the DDL path only so the diff --git a/src/baradadb.nim b/src/baradadb.nim index c5534b0..0df918c 100644 --- a/src/baradadb.nim +++ b/src/baradadb.nim @@ -319,7 +319,7 @@ proc main() = # Start HTTP server (blocking, multi-threaded via hunos) in background thread var httpServer = newHttpServerWithRegistry(config, registry) - spawn httpServer.run(config.port + 440) # HTTP port = TCP port + 440 + # raftNode is attached below before runTcpServer; HTTP thread reads it later. # Start background compaction loop for all databases let mcm = newMultiCompactionManager(registry) @@ -346,6 +346,7 @@ proc main() = if config.raftLogMaxEntries > 0: raftNode.logMaxEntries = config.raftLogMaxEntries tcpServer.raftNode = raftNode # C3b: executeQuery rejects writes on followers + httpServer.raftNode = raftNode # /metrics + /health raft gauges # Wire state machine: committed entries update LSM + secondary indexes # (B-tree/FTS/HNSW/graphs) and re-execute schema DDL on every node. let defaultDbInfo = getDatabaseInfo(registry, "default") @@ -370,6 +371,9 @@ proc main() = raftNet = newRaftNetwork(raftNode) asyncCheck raftNet.run() + # HTTP (hunos) after raft wiring so /metrics can see raftNode + spawn httpServer.run(config.port + 440) # HTTP port = TCP port + 440 + # Start replication health check and reconnection loops asyncCheck tcpServer.replicationManager.startHealthCheck(5000) asyncCheck tcpServer.replicationManager.startReconnectionLoop(10000) diff --git a/tests/test_all.nim b/tests/test_all.nim index a694222..161eb90 100644 --- a/tests/test_all.nim +++ b/tests/test_all.nim @@ -2706,6 +2706,25 @@ suite "Raft SQL Write Path": check res.keyValuePairs.len == 1 check res.keyValuePairs[0][1].len == 0 + test "raft metrics track elections, appends, and prometheus text": + var n = newRaftNode("n1", @[], raftPort = 29130) + check n.metrics != nil + check n.metrics.electionsTotal == 0 + n.becomeLeader() + check n.metrics.electionsTotal == 1 + discard n.appendLog("put", cast[seq[byte]]("a\x00b")) + check n.metrics.appendsTotal == 1 + let text = n.prometheusText() + check "baradb_raft_is_leader" in text + check "baradb_raft_elections_total" in text + check "baradb_raft_appends_total" in text + check "baradb_raft_log_entries" in text + check "node=\"n1\"" in text + check n.applyLag == 0 + n.commitIndex = 5 + n.lastApplied = 2 + check n.applyLag == 3 + test "compactLog discards applied prefix and preserves lastSnapshot base": var n = newRaftNode("n1", @[], raftPort = 29120) n.logMaxEntries = 8