feat(raft): expose raft metrics on /metrics and /health
CI / test (push) Has been cancelled
CI / verify (push) Has been cancelled
Clients CI / build-server (push) Has been cancelled
Clients CI / test-python (push) Has been cancelled
Clients CI / test-javascript (push) Has been cancelled
Clients CI / test-nim (push) Has been cancelled
Clients CI / test-rust (push) Has been cancelled
CI / test (push) Has been cancelled
CI / verify (push) Has been cancelled
Clients CI / build-server (push) Has been cancelled
Clients CI / test-python (push) Has been cancelled
Clients CI / test-javascript (push) Has been cancelled
Clients CI / test-nim (push) Has been cancelled
Clients CI / test-rust (push) Has been cancelled
- RaftMetrics counters: elections, term changes, appends, commit waits/ms, timeouts, lost leadership, forwards, applies, compactions - Gauges via prometheusText: is_leader, term, log size, commit/applied, apply lag, snapshot index - Wire httpServer.raftNode; extend GET /metrics and GET /health
This commit is contained in:
@@ -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`.
|
**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
|
```nim
|
||||||
import barabadb/core/raft
|
import barabadb/core/raft
|
||||||
|
|
||||||
|
|||||||
@@ -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.
|
**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
|
```nim
|
||||||
import barabadb/core/raft
|
import barabadb/core/raft
|
||||||
|
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ import ../protocol/auth
|
|||||||
import ../protocol/ratelimit
|
import ../protocol/ratelimit
|
||||||
import ../core/registry
|
import ../core/registry
|
||||||
import ../core/backup
|
import ../core/backup
|
||||||
|
import ../core/raft
|
||||||
|
|
||||||
type
|
type
|
||||||
HttpServer* = ref object
|
HttpServer* = ref object
|
||||||
@@ -37,6 +38,8 @@ type
|
|||||||
authManager*: AuthManager
|
authManager*: AuthManager
|
||||||
rateLimiter*: RateLimiter
|
rateLimiter*: RateLimiter
|
||||||
ws*: WsServer
|
ws*: WsServer
|
||||||
|
## Optional live raft node for /metrics and /health (set from main).
|
||||||
|
raftNode*: RaftNode
|
||||||
|
|
||||||
Metrics* = ref object
|
Metrics* = ref object
|
||||||
queriesTotal*: int
|
queriesTotal*: int
|
||||||
@@ -254,21 +257,47 @@ proc queryHandler(server: HttpServer): RequestHandler =
|
|||||||
server.metrics.queryErrors += 1
|
server.metrics.queryErrors += 1
|
||||||
ctx.json(%*{"error": errMsg}, 400)
|
ctx.json(%*{"error": errMsg}, 400)
|
||||||
|
|
||||||
proc healthHandler(): RequestHandler =
|
proc healthHandler(server: HttpServer): RequestHandler =
|
||||||
return proc(request: Request) {.gcsafe.} =
|
return proc(request: Request) {.gcsafe.} =
|
||||||
let ctx = newContext(request)
|
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 =
|
proc metricsHandler(server: HttpServer): RequestHandler =
|
||||||
return proc(request: Request) {.gcsafe.} =
|
return proc(request: Request) {.gcsafe.} =
|
||||||
let ctx = newContext(request)
|
let ctx = newContext(request)
|
||||||
if not server.checkAuth(request, ctx):
|
if not server.checkAuth(request, ctx):
|
||||||
return
|
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_query_errors_total " & $server.metrics.queryErrors & "\n" &
|
||||||
"baradb_inserts_total " & $server.metrics.insertCount & "\n" &
|
"baradb_inserts_total " & $server.metrics.insertCount & "\n" &
|
||||||
"baradb_selects_total " & $server.metrics.selectCount & "\n" &
|
"baradb_selects_total " & $server.metrics.selectCount & "\n" &
|
||||||
"baradb_connections_active " & $server.metrics.activeConnections & "\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)
|
request.respond(200, @[("Content-Type", "text/plain; charset=utf-8")], prometheus)
|
||||||
|
|
||||||
proc authHandler(server: HttpServer): RequestHandler =
|
proc authHandler(server: HttpServer): RequestHandler =
|
||||||
@@ -886,7 +915,7 @@ proc run*(server: HttpServer, port: int = 9470) =
|
|||||||
router.get("/admin", server.adminHandler())
|
router.get("/admin", server.adminHandler())
|
||||||
router.get("/", server.adminHandler())
|
router.get("/", server.adminHandler())
|
||||||
router.post("/query", server.queryHandler())
|
router.post("/query", server.queryHandler())
|
||||||
router.get("/health", healthHandler())
|
router.get("/health", server.healthHandler())
|
||||||
router.get("/metrics", server.metricsHandler())
|
router.get("/metrics", server.metricsHandler())
|
||||||
router.post("/auth", server.authHandler())
|
router.post("/auth", server.authHandler())
|
||||||
router.post("/auth/scram/start", server.scramStartHandler())
|
router.post("/auth/scram/start", server.scramStartHandler())
|
||||||
|
|||||||
@@ -25,6 +25,21 @@ type
|
|||||||
command*: string
|
command*: string
|
||||||
data*: seq[byte]
|
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
|
RaftNode* = ref object
|
||||||
id*: string
|
id*: string
|
||||||
state*: RaftState
|
state*: RaftState
|
||||||
@@ -41,6 +56,7 @@ type
|
|||||||
lastSnapshotTerm*: uint64
|
lastSnapshotTerm*: uint64
|
||||||
## Trigger compaction when log.len exceeds this (0 = default 256).
|
## Trigger compaction when log.len exceeds this (0 = default 256).
|
||||||
logMaxEntries*: int
|
logMaxEntries*: int
|
||||||
|
metrics*: RaftMetrics
|
||||||
# State machine callback
|
# State machine callback
|
||||||
applyCommand*: proc(cmd: string, data: seq[byte]) {.gcsafe.}
|
applyCommand*: proc(cmd: string, data: seq[byte]) {.gcsafe.}
|
||||||
# Distributed transaction callbacks (for raft→disttxn integration)
|
# Distributed transaction callbacks (for raft→disttxn integration)
|
||||||
@@ -172,6 +188,7 @@ proc newRaftNode*(id: string, peers: seq[string], raftPort: int = 0,
|
|||||||
lastSnapshotIndex: 0,
|
lastSnapshotIndex: 0,
|
||||||
lastSnapshotTerm: 0,
|
lastSnapshotTerm: 0,
|
||||||
logMaxEntries: 256,
|
logMaxEntries: 256,
|
||||||
|
metrics: RaftMetrics(),
|
||||||
nextIndex: initTable[string, uint64](),
|
nextIndex: initTable[string, uint64](),
|
||||||
matchIndex: initTable[string, uint64](),
|
matchIndex: initTable[string, uint64](),
|
||||||
peers: peers,
|
peers: peers,
|
||||||
@@ -254,6 +271,8 @@ proc compactLog*(node: RaftNode) =
|
|||||||
node.lastApplied = node.lastSnapshotIndex
|
node.lastApplied = node.lastSnapshotIndex
|
||||||
if node.commitIndex < node.lastSnapshotIndex:
|
if node.commitIndex < node.lastSnapshotIndex:
|
||||||
node.commitIndex = node.lastSnapshotIndex
|
node.commitIndex = node.lastSnapshotIndex
|
||||||
|
if node.metrics != nil:
|
||||||
|
inc node.metrics.compactionsTotal
|
||||||
node.saveState()
|
node.saveState()
|
||||||
|
|
||||||
proc applyCommitted(node: RaftNode) =
|
proc applyCommitted(node: RaftNode) =
|
||||||
@@ -280,9 +299,13 @@ proc applyCommitted(node: RaftNode) =
|
|||||||
else:
|
else:
|
||||||
if node.applyCommand != nil:
|
if node.applyCommand != nil:
|
||||||
node.applyCommand(entry.command, entry.data)
|
node.applyCommand(entry.command, entry.data)
|
||||||
|
if node.metrics != nil:
|
||||||
|
inc node.metrics.appliesTotal
|
||||||
node.compactLog()
|
node.compactLog()
|
||||||
|
|
||||||
proc becomeFollower*(node: RaftNode, term: uint64) =
|
proc becomeFollower*(node: RaftNode, term: uint64) =
|
||||||
|
if term > node.currentTerm and node.metrics != nil:
|
||||||
|
inc node.metrics.termChangesTotal
|
||||||
node.state = rsFollower
|
node.state = rsFollower
|
||||||
node.currentTerm = term
|
node.currentTerm = term
|
||||||
node.votedFor = ""
|
node.votedFor = ""
|
||||||
@@ -294,6 +317,8 @@ proc becomeFollower*(node: RaftNode, term: uint64) =
|
|||||||
proc becomeCandidate*(node: RaftNode) =
|
proc becomeCandidate*(node: RaftNode) =
|
||||||
node.state = rsCandidate
|
node.state = rsCandidate
|
||||||
inc node.currentTerm
|
inc node.currentTerm
|
||||||
|
if node.metrics != nil:
|
||||||
|
inc node.metrics.termChangesTotal
|
||||||
node.votedFor = node.id
|
node.votedFor = node.id
|
||||||
node.votesReceived.clear()
|
node.votesReceived.clear()
|
||||||
node.votesReceived.incl(node.id)
|
node.votesReceived.incl(node.id)
|
||||||
@@ -302,6 +327,8 @@ proc becomeCandidate*(node: RaftNode) =
|
|||||||
proc becomeLeader*(node: RaftNode) =
|
proc becomeLeader*(node: RaftNode) =
|
||||||
node.state = rsLeader
|
node.state = rsLeader
|
||||||
node.leaderId = node.id
|
node.leaderId = node.id
|
||||||
|
if node.metrics != nil:
|
||||||
|
inc node.metrics.electionsTotal
|
||||||
info("Raft node " & node.id & " became leader for term " & $node.currentTerm)
|
info("Raft node " & node.id & " became leader for term " & $node.currentTerm)
|
||||||
for peer in node.peers:
|
for peer in node.peers:
|
||||||
node.nextIndex[peer] = node.lastLogIndex + 1
|
node.nextIndex[peer] = node.lastLogIndex + 1
|
||||||
@@ -437,6 +464,8 @@ proc appendLog*(node: RaftNode, command: string, data: seq[byte] = @[]): LogEntr
|
|||||||
data: data,
|
data: data,
|
||||||
)
|
)
|
||||||
node.log.add(result)
|
node.log.add(result)
|
||||||
|
if node.metrics != nil:
|
||||||
|
inc node.metrics.appendsTotal
|
||||||
node.saveState()
|
node.saveState()
|
||||||
|
|
||||||
proc handleVoteReply*(node: RaftNode, reply: RaftMessage) =
|
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 leaderId*(node: RaftNode): string = node.leaderId
|
||||||
proc logLen*(node: RaftNode): int = node.log.len
|
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
|
# Leader election timer loop
|
||||||
type
|
type
|
||||||
ElectionTimer* = ref object
|
ElectionTimer* = ref object
|
||||||
|
|||||||
@@ -292,11 +292,18 @@ proc forwardQueryToLeader*(host: string, port: int, query: string,
|
|||||||
try: sock.close() except CatchableError: discard
|
try: sock.close() except CatchableError: discard
|
||||||
|
|
||||||
proc waitRaftCommit(node: RaftNode, lastIdx: uint64, timeoutMs: int): Future[(bool, string)] {.async.} =
|
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:
|
while node.commitIndex < lastIdx and getMonoTime() < deadline:
|
||||||
await sleepAsync(10)
|
await sleepAsync(10)
|
||||||
|
let waitedMs = int64((getMonoTime() - start).inMilliseconds)
|
||||||
if node.commitIndex < lastIdx:
|
if node.commitIndex < lastIdx:
|
||||||
|
if node.metrics != nil:
|
||||||
|
inc node.metrics.commitTimeoutsTotal
|
||||||
return (false, "raft commit timeout")
|
return (false, "raft commit timeout")
|
||||||
|
if node.metrics != nil:
|
||||||
|
inc node.metrics.commitWaitsTotal
|
||||||
|
node.metrics.commitWaitMsTotal += waitedMs
|
||||||
return (true, "")
|
return (true, "")
|
||||||
|
|
||||||
proc appendWriteToRaft*(node: RaftNode, kvPairs: seq[(string, seq[byte])],
|
proc appendWriteToRaft*(node: RaftNode, kvPairs: seq[(string, seq[byte])],
|
||||||
@@ -316,6 +323,8 @@ proc appendWriteToRaft*(node: RaftNode, kvPairs: seq[(string, seq[byte])],
|
|||||||
else:
|
else:
|
||||||
node.appendLog("delete", cast[seq[byte]](key))
|
node.appendLog("delete", cast[seq[byte]](key))
|
||||||
if entry.index == 0:
|
if entry.index == 0:
|
||||||
|
if node.metrics != nil:
|
||||||
|
inc node.metrics.lostLeadershipTotal
|
||||||
return (false, "lost leadership during raft append")
|
return (false, "lost leadership during raft append")
|
||||||
lastIdx = entry.index
|
lastIdx = entry.index
|
||||||
return await waitRaftCommit(node, lastIdx, timeoutMs)
|
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).
|
## MUST be called outside the storage gate (same as appendWriteToRaft).
|
||||||
let entry = node.appendLog("ddl", cast[seq[byte]](sql))
|
let entry = node.appendLog("ddl", cast[seq[byte]](sql))
|
||||||
if entry.index == 0:
|
if entry.index == 0:
|
||||||
|
if node.metrics != nil:
|
||||||
|
inc node.metrics.lostLeadershipTotal
|
||||||
return (false, "lost leadership during raft append")
|
return (false, "lost leadership during raft append")
|
||||||
return await waitRaftCommit(node, entry.index, timeoutMs)
|
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)
|
return (false, QueryResult(), e.msg)
|
||||||
# Follower write/DDL: proxy to leader SQL port (outside the storage gate).
|
# Follower write/DDL: proxy to leader SQL port (outside the storage gate).
|
||||||
if needsForward:
|
if needsForward:
|
||||||
return await forwardQueryToLeader(forwardHost, forwardPort, query, params,
|
let (okF, qrF, errF) = await forwardQueryToLeader(forwardHost, forwardPort,
|
||||||
raftWriteTimeoutMs)
|
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).
|
# Raft log append + majority wait (outside the storage gate).
|
||||||
# DDL batches ship the original SQL once (re-executed on apply). Pure DML
|
# 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
|
# ships KV pairs. Mixed DDL+DML in one query uses the DDL path only so the
|
||||||
|
|||||||
+5
-1
@@ -319,7 +319,7 @@ proc main() =
|
|||||||
|
|
||||||
# Start HTTP server (blocking, multi-threaded via hunos) in background thread
|
# Start HTTP server (blocking, multi-threaded via hunos) in background thread
|
||||||
var httpServer = newHttpServerWithRegistry(config, registry)
|
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
|
# Start background compaction loop for all databases
|
||||||
let mcm = newMultiCompactionManager(registry)
|
let mcm = newMultiCompactionManager(registry)
|
||||||
@@ -346,6 +346,7 @@ proc main() =
|
|||||||
if config.raftLogMaxEntries > 0:
|
if config.raftLogMaxEntries > 0:
|
||||||
raftNode.logMaxEntries = config.raftLogMaxEntries
|
raftNode.logMaxEntries = config.raftLogMaxEntries
|
||||||
tcpServer.raftNode = raftNode # C3b: executeQuery rejects writes on followers
|
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
|
# Wire state machine: committed entries update LSM + secondary indexes
|
||||||
# (B-tree/FTS/HNSW/graphs) and re-execute schema DDL on every node.
|
# (B-tree/FTS/HNSW/graphs) and re-execute schema DDL on every node.
|
||||||
let defaultDbInfo = getDatabaseInfo(registry, "default")
|
let defaultDbInfo = getDatabaseInfo(registry, "default")
|
||||||
@@ -370,6 +371,9 @@ proc main() =
|
|||||||
raftNet = newRaftNetwork(raftNode)
|
raftNet = newRaftNetwork(raftNode)
|
||||||
asyncCheck raftNet.run()
|
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
|
# Start replication health check and reconnection loops
|
||||||
asyncCheck tcpServer.replicationManager.startHealthCheck(5000)
|
asyncCheck tcpServer.replicationManager.startHealthCheck(5000)
|
||||||
asyncCheck tcpServer.replicationManager.startReconnectionLoop(10000)
|
asyncCheck tcpServer.replicationManager.startReconnectionLoop(10000)
|
||||||
|
|||||||
@@ -2706,6 +2706,25 @@ suite "Raft SQL Write Path":
|
|||||||
check res.keyValuePairs.len == 1
|
check res.keyValuePairs.len == 1
|
||||||
check res.keyValuePairs[0][1].len == 0
|
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":
|
test "compactLog discards applied prefix and preserves lastSnapshot base":
|
||||||
var n = newRaftNode("n1", @[], raftPort = 29120)
|
var n = newRaftNode("n1", @[], raftPort = 29120)
|
||||||
n.logMaxEntries = 8
|
n.logMaxEntries = 8
|
||||||
|
|||||||
Reference in New Issue
Block a user