From aaf896d8def69206cc015f5246cef609acb75055 Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 14 Jun 2026 18:26:09 +0400 Subject: [PATCH] feat(server): llama-server health probe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../kotlin/com/correx/apps/server/Main.kt | 13 ++ .../server/health/LlamaServerHealthProbe.kt | 159 ++++++++++++++ .../health/LlamaServerHealthProbeTest.kt | 202 ++++++++++++++++++ .../com/correx/core/config/CorrexConfig.kt | 7 + 4 files changed, 381 insertions(+) create mode 100644 apps/server/src/main/kotlin/com/correx/apps/server/health/LlamaServerHealthProbe.kt create mode 100644 apps/server/src/test/kotlin/com/correx/apps/server/health/LlamaServerHealthProbeTest.kt diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index 159ac1da..d23fab0c 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -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, diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/health/LlamaServerHealthProbe.kt b/apps/server/src/main/kotlin/com/correx/apps/server/health/LlamaServerHealthProbe.kt new file mode 100644 index 00000000..b75c950d --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/health/LlamaServerHealthProbe.kt @@ -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 = 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 + } +} diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/health/LlamaServerHealthProbeTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/health/LlamaServerHealthProbeTest.kt new file mode 100644 index 00000000..32fdc020 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/health/LlamaServerHealthProbeTest.kt @@ -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 = mutableListOf()) : EventStore { + override suspend fun append(event: NewEvent): StoredEvent = error("unused") + override suspend fun appendAll(events: List): List = error("unused") + override fun read(sessionId: SessionId): List = events.filter { it.metadata.sessionId == sessionId } + override fun readFrom(sessionId: SessionId, fromSequence: Long): List = + 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 = emptyFlow() + override fun subscribeAll(): Flow = emptyFlow() + override suspend fun lastGlobalSequence(): Long = events.lastOrNull()?.sequence ?: 0L + override fun allEvents(): Sequence = events.asSequence() + override fun allSessionIds(): Set = 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) + } +} diff --git a/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt b/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt index 73e36a9c..e7533e9b 100644 --- a/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt +++ b/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt @@ -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, ) /**