feat(server): llama-server health probe
Second HealthProbe (after disk-watermark): liveness via GET /health on the configured llama base URL + tokens/sec degradation watch derived from recorded InferenceCompletedEvents (windowed average vs configurable threshold, opt-in via llamaTpsWarnBelow). Plugs into the existing HealthMonitor edge-detection loop; no new events. Observability spec §4.
This commit is contained in:
@@ -125,6 +125,8 @@ fun main() {
|
||||
// it stays null on the static path. System RAM (overlaid below) is owner-agnostic, so the gauge
|
||||
// renders on both paths regardless of whether correx manages the model process.
|
||||
var managedPidSupplier: () -> Long? = { null }
|
||||
// Base URL used by the llama-server health probe; derived from whichever path is active.
|
||||
var llamaBaseUrl: String = "http://127.0.0.1:10000"
|
||||
|
||||
if (correxConfig.models.isNotEmpty()) {
|
||||
val settings = correxConfig.modelsSettings
|
||||
@@ -151,6 +153,7 @@ fun main() {
|
||||
inferenceRouter = managedRouter
|
||||
modelSwapper = managedRouter
|
||||
managedPidSupplier = { modelManager.currentPid() }
|
||||
llamaBaseUrl = "http://${settings.host}:${settings.port}"
|
||||
// Shutdown hook to kill the managed llama-server
|
||||
Runtime.getRuntime().addShutdownHook(Thread {
|
||||
log.info("Shutdown: unloading managed model '{}'", targetModelConfig.id)
|
||||
@@ -165,6 +168,9 @@ fun main() {
|
||||
firstProvider = staticProviders.first()
|
||||
infraRegistry = InfrastructureModule.createProviderRegistry(staticProviders)
|
||||
inferenceRouter = DefaultInferenceRouter(infraRegistry, FirstAvailableRoutingStrategy())
|
||||
llamaBaseUrl = correxConfig.providers.firstOrNull()?.url
|
||||
?: System.getenv("CORREX_LLAMA_URL")
|
||||
?: "http://127.0.0.1:10000"
|
||||
}
|
||||
|
||||
// Build the resource gauge for whichever GPU vendor is present (pid-less GPU/VRAM reads work on
|
||||
@@ -436,6 +442,13 @@ fun main() {
|
||||
warnBytes = hc.diskWarnBytes,
|
||||
warnGrowthBytesPerMin = hc.diskWarnGrowthBytesPerMin,
|
||||
),
|
||||
com.correx.apps.server.health.LlamaServerHealthProbe(
|
||||
llamaBaseUrl = llamaBaseUrl,
|
||||
httpClient = io.ktor.client.HttpClient(io.ktor.client.engine.cio.CIO),
|
||||
eventStore = eventStore,
|
||||
livenessTimeoutMs = hc.llamaLivenessTimeoutMs,
|
||||
tpsWarnBelow = hc.llamaTpsWarnBelow,
|
||||
),
|
||||
),
|
||||
intervalMs = hc.intervalMs,
|
||||
initialStatuses = seeded,
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
package com.correx.apps.server.health
|
||||
|
||||
import com.correx.core.events.events.HealthSubject
|
||||
import com.correx.core.events.events.InferenceCompletedEvent
|
||||
import com.correx.core.events.stores.EventStore
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.request.get
|
||||
import io.ktor.http.isSuccess
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.withContext
|
||||
import kotlinx.coroutines.withTimeout
|
||||
|
||||
private const val MS_PER_SECOND = 1000.0
|
||||
private const val TPS_WINDOW = 10
|
||||
|
||||
/**
|
||||
* Health probe for the local llama-server (observability-spec §4: LLAMA_SERVER subject).
|
||||
*
|
||||
* **Liveness** — every tick issues a bounded `GET [llamaBaseUrl]/health` via [httpClient].
|
||||
* A 200-family response is live; any non-2xx, connection error, or [livenessTimeoutMs]
|
||||
* expiry is DEGRADED. The [httpClient] is injected so tests can supply a [MockEngine].
|
||||
*
|
||||
* **Degradation (tokens/sec)** — on each tick the probe scans new [InferenceCompletedEvent]s
|
||||
* from the event store (events appended since the previous tick, tracked by
|
||||
* [lastScannedGlobalSequence]), accumulates the per-call tokens/sec into a sliding window
|
||||
* of size [TPS_WINDOW], and reports DEGRADED when the window average falls below
|
||||
* [tpsWarnBelow]. A threshold of 0.0 disables the degradation dimension; a window that
|
||||
* has not yet collected any samples reports HEALTHY (no baseline).
|
||||
*
|
||||
* The probe only observes and returns a [HealthObservation]; it never emits events.
|
||||
* [HealthMonitor] owns edge detection and event emission (Invariants #8/#9).
|
||||
*/
|
||||
class LlamaServerHealthProbe(
|
||||
private val llamaBaseUrl: String,
|
||||
private val httpClient: HttpClient,
|
||||
private val eventStore: EventStore,
|
||||
private val livenessTimeoutMs: Long = 5_000,
|
||||
private val tpsWarnBelow: Double = 0.0,
|
||||
) : HealthProbe {
|
||||
|
||||
override val subject: HealthSubject = HealthSubject.LLAMA_SERVER
|
||||
|
||||
/**
|
||||
* Monotonic global sequence up to which we have already scanned for inference events.
|
||||
* Assumes single-coroutine sequential invocation: [HealthMonitor] calls probes sequentially
|
||||
* within one coroutine tick, so no concurrent mutation of this field or [tpsWindow] occurs.
|
||||
*/
|
||||
private var lastScannedGlobalSequence: Long = -1L
|
||||
|
||||
/**
|
||||
* Sliding window of recent per-call tokens/sec values (newest at the end).
|
||||
* See [lastScannedGlobalSequence] for the single-coroutine assumption.
|
||||
*/
|
||||
private val tpsWindow: ArrayDeque<Double> = ArrayDeque(TPS_WINDOW)
|
||||
|
||||
override suspend fun observe(): HealthObservation {
|
||||
val livenessObservation = checkLiveness()
|
||||
if (livenessObservation.status == HealthStatus.DEGRADED) return livenessObservation
|
||||
return checkThroughput() ?: livenessObservation
|
||||
}
|
||||
|
||||
private suspend fun checkLiveness(): HealthObservation {
|
||||
val result = runCatching {
|
||||
withTimeout(livenessTimeoutMs) {
|
||||
httpClient.get("$llamaBaseUrl/health")
|
||||
}
|
||||
}
|
||||
val response = result.getOrNull()
|
||||
return when {
|
||||
result.isFailure -> degraded(
|
||||
metric = "liveness",
|
||||
value = 0L,
|
||||
threshold = 1L,
|
||||
detail = "GET $llamaBaseUrl/health failed: ${result.exceptionOrNull()?.message}",
|
||||
)
|
||||
response != null && !response.status.isSuccess() -> degraded(
|
||||
metric = "liveness",
|
||||
value = response.status.value.toLong(),
|
||||
threshold = 200L,
|
||||
detail = "GET $llamaBaseUrl/health returned HTTP ${response.status.value}",
|
||||
)
|
||||
else -> healthy(
|
||||
metric = "liveness",
|
||||
value = 1L,
|
||||
threshold = 1L,
|
||||
detail = "llama-server live at $llamaBaseUrl",
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun checkThroughput(): HealthObservation? {
|
||||
if (tpsWarnBelow <= 0.0) {
|
||||
// Dimension disabled: still advance the cursor so a later live-enable doesn't rescan
|
||||
// the entire history from sequence 0.
|
||||
lastScannedGlobalSequence = withContext(Dispatchers.IO) { eventStore.lastGlobalSequence() }
|
||||
return null
|
||||
}
|
||||
withContext(Dispatchers.IO) { scanNewInferenceEvents() }
|
||||
if (tpsWindow.isEmpty()) return null
|
||||
val avgTps = tpsWindow.average()
|
||||
val avgTpsLong = (avgTps * TPS_SCALE).toLong()
|
||||
val thresholdLong = (tpsWarnBelow * TPS_SCALE).toLong()
|
||||
return if (avgTps < tpsWarnBelow) {
|
||||
degraded(
|
||||
metric = "tokens_per_second",
|
||||
value = avgTpsLong,
|
||||
threshold = thresholdLong,
|
||||
detail = "avg tokens/sec %.2f below warn threshold %.2f (window=%d)".format(
|
||||
avgTps, tpsWarnBelow, tpsWindow.size,
|
||||
),
|
||||
)
|
||||
} else {
|
||||
healthy(
|
||||
metric = "tokens_per_second",
|
||||
value = avgTpsLong,
|
||||
threshold = thresholdLong,
|
||||
detail = "avg tokens/sec %.2f (window=%d)".format(avgTps, tpsWindow.size),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Scans [EventStore.allEvents] for [InferenceCompletedEvent]s with a global sequence
|
||||
* number beyond [lastScannedGlobalSequence], appending each call's tokens/sec to
|
||||
* [tpsWindow] (capped at [TPS_WINDOW]). Updates [lastScannedGlobalSequence] so the
|
||||
* next tick only scans new events — O(new events), not O(all events).
|
||||
*/
|
||||
private fun scanNewInferenceEvents() {
|
||||
var maxSeen = lastScannedGlobalSequence
|
||||
eventStore.allEvents()
|
||||
.filter { it.sequence > lastScannedGlobalSequence }
|
||||
.forEach { event ->
|
||||
if (event.sequence > maxSeen) maxSeen = event.sequence
|
||||
val payload = event.payload as? InferenceCompletedEvent ?: return@forEach
|
||||
// Skip events with no completion tokens (e.g. stop-token-only responses) to
|
||||
// avoid dragging the windowed average to zero on a healthy model.
|
||||
if (payload.tokensUsed.completionTokens <= 0) return@forEach
|
||||
val tps = if (payload.latencyMs > 0L) {
|
||||
payload.tokensUsed.completionTokens.toDouble() / (payload.latencyMs / MS_PER_SECOND)
|
||||
} else {
|
||||
0.0
|
||||
}
|
||||
if (tpsWindow.size >= TPS_WINDOW) tpsWindow.removeFirst()
|
||||
tpsWindow.addLast(tps)
|
||||
}
|
||||
lastScannedGlobalSequence = maxSeen
|
||||
}
|
||||
|
||||
private fun degraded(metric: String, value: Long, threshold: Long, detail: String) =
|
||||
HealthObservation(subject, HealthStatus.DEGRADED, metric, value, threshold, detail)
|
||||
|
||||
private fun healthy(metric: String, value: Long, threshold: Long, detail: String) =
|
||||
HealthObservation(subject, HealthStatus.HEALTHY, metric, value, threshold, detail)
|
||||
|
||||
companion object {
|
||||
/** Scale factor: store tps×100 as Long to fit [HealthObservation.observedValue]. */
|
||||
private const val TPS_SCALE = 100L
|
||||
}
|
||||
}
|
||||
+202
@@ -0,0 +1,202 @@
|
||||
package com.correx.apps.server.health
|
||||
|
||||
import com.correx.core.events.events.EventMetadata
|
||||
import com.correx.core.events.events.InferenceCompletedEvent
|
||||
import com.correx.core.events.events.NewEvent
|
||||
import com.correx.core.events.events.StoredEvent
|
||||
import com.correx.core.events.stores.EventStore
|
||||
import com.correx.core.events.types.EventId
|
||||
import com.correx.core.events.types.SessionId
|
||||
import com.correx.core.inference.TokenUsage
|
||||
import com.correx.core.utils.TypeId
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.engine.mock.MockEngine
|
||||
import io.ktor.client.engine.mock.respond
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.emptyFlow
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.datetime.Instant
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Assertions.assertTrue
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
class LlamaServerHealthProbeTest {
|
||||
|
||||
private val sessionId: SessionId = TypeId("s-1")
|
||||
private val stageId = TypeId("stage-1")
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
private fun httpClient(status: HttpStatusCode): HttpClient =
|
||||
HttpClient(MockEngine { respond("", status) })
|
||||
|
||||
private fun httpClientThrowing(): HttpClient =
|
||||
HttpClient(MockEngine { error("connection refused") })
|
||||
|
||||
private fun inference(seq: Long, completionTokens: Int, latencyMs: Long): StoredEvent =
|
||||
StoredEvent(
|
||||
metadata = EventMetadata(
|
||||
eventId = EventId("e$seq"),
|
||||
sessionId = sessionId,
|
||||
timestamp = Instant.parse("2026-01-01T00:00:00Z"),
|
||||
schemaVersion = 1,
|
||||
causationId = null,
|
||||
correlationId = null,
|
||||
),
|
||||
sequence = seq,
|
||||
sessionSequence = seq,
|
||||
payload = InferenceCompletedEvent(
|
||||
requestId = TypeId("req-$seq"),
|
||||
sessionId = sessionId,
|
||||
stageId = stageId,
|
||||
providerId = TypeId("lfm"),
|
||||
tokensUsed = TokenUsage(promptTokens = 10, completionTokens = completionTokens),
|
||||
latencyMs = latencyMs,
|
||||
responseArtifactId = TypeId("art-$seq"),
|
||||
),
|
||||
)
|
||||
|
||||
/** Minimal EventStore backed by a mutable list; allEvents() returns all stored events. */
|
||||
private class ListEventStore(private val events: MutableList<StoredEvent> = mutableListOf()) : EventStore {
|
||||
override suspend fun append(event: NewEvent): StoredEvent = error("unused")
|
||||
override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> = error("unused")
|
||||
override fun read(sessionId: SessionId): List<StoredEvent> = events.filter { it.metadata.sessionId == sessionId }
|
||||
override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> =
|
||||
events.filter { it.metadata.sessionId == sessionId && it.sequence >= fromSequence }
|
||||
override fun lastSequence(sessionId: SessionId): Long? =
|
||||
events.filter { it.metadata.sessionId == sessionId }.lastOrNull()?.sequence
|
||||
override fun subscribe(sessionId: SessionId): Flow<StoredEvent> = emptyFlow()
|
||||
override fun subscribeAll(): Flow<StoredEvent> = emptyFlow()
|
||||
override suspend fun lastGlobalSequence(): Long = events.lastOrNull()?.sequence ?: 0L
|
||||
override fun allEvents(): Sequence<StoredEvent> = events.asSequence()
|
||||
override fun allSessionIds(): Set<SessionId> = events.map { it.metadata.sessionId }.toSet()
|
||||
|
||||
fun add(event: StoredEvent) { events.add(event) }
|
||||
}
|
||||
|
||||
private fun probe(
|
||||
status: HttpStatusCode = HttpStatusCode.OK,
|
||||
store: ListEventStore = ListEventStore(),
|
||||
tpsWarnBelow: Double = 0.0,
|
||||
) = LlamaServerHealthProbe(
|
||||
llamaBaseUrl = "http://localhost:11434",
|
||||
httpClient = httpClient(status),
|
||||
eventStore = store,
|
||||
livenessTimeoutMs = 5_000,
|
||||
tpsWarnBelow = tpsWarnBelow,
|
||||
)
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Liveness tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
fun `healthy when llama-server returns 200`(): Unit = runBlocking {
|
||||
val obs = probe(status = HttpStatusCode.OK).observe()
|
||||
assertEquals(HealthStatus.HEALTHY, obs.status)
|
||||
assertEquals("liveness", obs.metric)
|
||||
assertEquals(1L, obs.observedValue)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `degraded when llama-server returns non-200`(): Unit = runBlocking {
|
||||
val obs = probe(status = HttpStatusCode.ServiceUnavailable).observe()
|
||||
assertEquals(HealthStatus.DEGRADED, obs.status)
|
||||
assertEquals("liveness", obs.metric)
|
||||
assertTrue(obs.detail.contains("503"), "expected 503 in detail: ${obs.detail}")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `degraded on connection error`(): Unit = runBlocking {
|
||||
val p = LlamaServerHealthProbe(
|
||||
llamaBaseUrl = "http://localhost:11434",
|
||||
httpClient = httpClientThrowing(),
|
||||
eventStore = ListEventStore(),
|
||||
livenessTimeoutMs = 5_000,
|
||||
tpsWarnBelow = 0.0,
|
||||
)
|
||||
val obs = p.observe()
|
||||
assertEquals(HealthStatus.DEGRADED, obs.status)
|
||||
assertEquals("liveness", obs.metric)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Tokens/sec degradation tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
fun `healthy when tps above threshold`(): Unit = runBlocking {
|
||||
// 100 tokens in 1000 ms = 100 tps, threshold is 50 tps
|
||||
val store = ListEventStore()
|
||||
store.add(inference(seq = 1, completionTokens = 100, latencyMs = 1_000))
|
||||
val obs = probe(store = store, tpsWarnBelow = 50.0).observe()
|
||||
assertEquals(HealthStatus.HEALTHY, obs.status)
|
||||
assertEquals("tokens_per_second", obs.metric)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `degraded when tps below threshold`(): Unit = runBlocking {
|
||||
// 10 tokens in 1000 ms = 10 tps, threshold is 50 tps
|
||||
val store = ListEventStore()
|
||||
store.add(inference(seq = 1, completionTokens = 10, latencyMs = 1_000))
|
||||
val obs = probe(store = store, tpsWarnBelow = 50.0).observe()
|
||||
assertEquals(HealthStatus.DEGRADED, obs.status)
|
||||
assertEquals("tokens_per_second", obs.metric)
|
||||
assertTrue(obs.detail.contains("below"), "detail should mention 'below': ${obs.detail}")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `healthy when no inference events and tpsWarnBelow configured`(): Unit = runBlocking {
|
||||
// No baseline — can't degrade without samples
|
||||
val obs = probe(tpsWarnBelow = 50.0).observe()
|
||||
assertEquals(HealthStatus.HEALTHY, obs.status)
|
||||
// No inference events means the liveness path dominates
|
||||
assertEquals("liveness", obs.metric)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `tpsWarnBelow zero disables degradation check`(): Unit = runBlocking {
|
||||
val store = ListEventStore()
|
||||
store.add(inference(seq = 1, completionTokens = 1, latencyMs = 100_000)) // very slow
|
||||
val obs = probe(store = store, tpsWarnBelow = 0.0).observe()
|
||||
// Degradation disabled; liveness is healthy -> overall HEALTHY
|
||||
assertEquals(HealthStatus.HEALTHY, obs.status)
|
||||
assertEquals("liveness", obs.metric)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `only scans new events on subsequent ticks`(): Unit = runBlocking {
|
||||
val store = ListEventStore()
|
||||
store.add(inference(seq = 1, completionTokens = 100, latencyMs = 1_000)) // 100 tps
|
||||
|
||||
val p = LlamaServerHealthProbe(
|
||||
llamaBaseUrl = "http://localhost:11434",
|
||||
httpClient = httpClient(HttpStatusCode.OK),
|
||||
eventStore = store,
|
||||
livenessTimeoutMs = 5_000,
|
||||
tpsWarnBelow = 50.0,
|
||||
)
|
||||
|
||||
// First tick: sees inference at seq=1, 100 tps -> HEALTHY
|
||||
assertEquals(HealthStatus.HEALTHY, p.observe().status)
|
||||
|
||||
// Add a very slow inference at seq=2
|
||||
store.add(inference(seq = 2, completionTokens = 1, latencyMs = 10_000)) // 0.1 tps
|
||||
|
||||
// Second tick: sees only seq=2 (new), window = [100, 0.1], avg ~50.05 -> still HEALTHY
|
||||
val obs2 = p.observe()
|
||||
// avg(100, 0.1) = 50.05 which is above 50 threshold -> HEALTHY
|
||||
assertEquals(HealthStatus.HEALTHY, obs2.status)
|
||||
|
||||
// Add another very slow inference
|
||||
store.add(inference(seq = 3, completionTokens = 1, latencyMs = 10_000)) // 0.1 tps
|
||||
|
||||
// Third tick: window = [100, 0.1, 0.1], avg ~33.4 -> DEGRADED
|
||||
val obs3 = p.observe()
|
||||
assertEquals(HealthStatus.DEGRADED, obs3.status)
|
||||
assertEquals("tokens_per_second", obs3.metric)
|
||||
}
|
||||
}
|
||||
@@ -24,6 +24,11 @@ data class CorrexConfig(
|
||||
* the on-disk footprint every [intervalMs] and records a degraded/restored event on the edge.
|
||||
* [diskWarnBytes] is the absolute event-log + CAS size that trips a warning; [diskWarnGrowthBytesPerMin]
|
||||
* is the growth rate that warns log compaction is becoming non-theoretical. Defaults suit a homelab.
|
||||
*
|
||||
* [llamaLivenessTimeoutMs] bounds the GET /health ping to the llama-server; a connection error or
|
||||
* timeout within this window is reported as DEGRADED (liveness). [llamaTpsWarnBelow] is the
|
||||
* tokens/sec floor below which the probe reports degradation (KV-cache pressure, thermal throttle);
|
||||
* 0.0 disables the degradation dimension (no baseline to compare against on a fresh install).
|
||||
*/
|
||||
@Serializable
|
||||
data class HealthConfig(
|
||||
@@ -31,6 +36,8 @@ data class HealthConfig(
|
||||
val intervalMs: Long = 30_000,
|
||||
val diskWarnBytes: Long = 5_000_000_000,
|
||||
val diskWarnGrowthBytesPerMin: Long = 50_000_000,
|
||||
val llamaLivenessTimeoutMs: Long = 5_000,
|
||||
val llamaTpsWarnBelow: Double = 0.0,
|
||||
)
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user