gnodet-bot commented on code in PR #26548: URL: https://github.com/apache/camel/pull/26548#discussion_r4036401120
########## dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TokenRateWindow.java: ########## @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.dsl.jbang.core.commands.tui; + +import java.util.ArrayDeque; +import java.util.Deque; + +/** + * Turns a monotonically increasing token counter, sampled at irregular intervals, into a tokens-per-second rate over a + * sliding window. The window smooths the bursts a speculative decoder produces (several tokens land at once, then + * nothing for a step) while still dropping to zero within one window of the counter standing still. + * <p/> + * The counter resets when a new request starts (llama-server counts decoded tokens per request), so a sample below the + * previous one starts a fresh baseline instead of producing a negative rate. + */ +final class TokenRateWindow { + + private final long windowMillis; + private final Deque<long[]> samples = new ArrayDeque<>(); + + TokenRateWindow(long windowMillis) { + this.windowMillis = Math.max(1, windowMillis); + } + + /** Records the counter value observed at {@code nowMillis}. */ + synchronized void sample(long nowMillis, long counter) { + long[] last = samples.peekLast(); + if (last != null && counter < last[1]) { + samples.clear(); + } + samples.addLast(new long[] { nowMillis, counter }); + // keep exactly one sample at or before the window start as the baseline + long cutoff = nowMillis - windowMillis; + while (samples.size() > 2) { + long[] first = samples.peekFirst(); + long[] second = samples.stream().skip(1).findFirst().orElse(first); Review Comment: 📝 **Nit: stream allocation for second-element lookup in a hot `synchronized` path** `samples.stream().skip(1).findFirst()` allocates a `Stream`, a `Spliterator`, and (on some JVMs) wraps the result in an `Optional` — all inside a `synchronized` block that fires every 1 second from `updateSlot()`. With `size() > 2` as the loop guard the deque is always small (typically 2–3 entries), but the allocation is avoidable: ```suggestion Iterator<long[]> it = samples.iterator(); it.next(); // skip first long[] second = it.hasNext() ? it.next() : first; ``` Same semantics, no heap allocation, no stream overhead. ########## dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/OllamaMonitor.java: ########## @@ -0,0 +1,1250 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.dsl.jbang.core.commands.tui; + +import java.io.InputStream; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Deque; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.camel.component.ai.observability.GenAiAttributes; +import org.apache.camel.dsl.jbang.core.commands.LlmClient; +import org.apache.camel.util.json.JsonArray; +import org.apache.camel.util.json.JsonObject; +import org.apache.camel.util.json.Jsoner; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Collects what the Ollama tab shows: the Ollama server and its loaded models over the REST API, the live state of the + * llama-server runner Ollama spawns on this machine, the load on the host, and a log of requests with the timings + * Ollama reports per request. + * <p/> + * {@link #poll()} is called from the TUI's background refresh while the tab is showing and throttles each source on its + * own interval. Requests arrive from two sides: the TUI's own AI panel calls {@link #recordRequest} with the usage of + * each Ollama reply, and calls made by Camel routes come in through {@link #ingestSpans} from the GenAI observability + * spans of the selected integration. The runner and host probes only run when the Ollama host is this machine; a remote + * or containerised Ollama still gets the API and per-request data. + * <p/> + * All I/O happens outside the lock; the {@code update*} methods apply results under it and are also what the tests + * feed. + */ +final class OllamaMonitor { + + private static final Logger LOG = LoggerFactory.getLogger(OllamaMonitor.class); + + static final int MAX_REQUESTS = 200; + static final int HISTORY_POINTS = 120; + + private static final long DETECT_INTERVAL_MS = 10_000; + private static final long VERSION_INTERVAL_MS = 30_000; + private static final long RECONNECT_INTERVAL_MS = 5_000; + private static final long PS_INTERVAL_MS = 1_000; + private static final long TAGS_INTERVAL_MS = 15_000; + private static final long RUNNER_SCAN_INTERVAL_MS = 5_000; + private static final long HOST_INTERVAL_MS = 1_000; + private static final long RATE_WINDOW_MS = 1_500; + private static final int MAX_SEEN_SPANS = 4_000; + private static final long SPAN_INTERVAL_MS = 5_000; + + // ---- data ---- + + record ServerInfo(String baseUrl, String version, boolean local) { + } + + record ModelShape(String architecture, int layers, int experts, int expertsUsed, long maxContext, + int embeddingLength, long parameters, List<String> capabilities) { + } + + record LoadedModel(String name, String family, String parameterSize, String quantization, long sizeBytes, + long sizeVram, long contextLength, Instant expiresAt, ModelShape shape) { + + LoadedModel withShape(ModelShape newShape) { + return new LoadedModel( + name, family, parameterSize, quantization, sizeBytes, sizeVram, contextLength, + expiresAt, newShape); + } + + /** Share of the model held in GPU memory; 100 means fully offloaded. */ + int gpuPercent() { + if (sizeBytes <= 0) { + return 0; + } + return (int) Math.min(100, sizeVram * 100 / sizeBytes); + } + } + + /** Folded state of the runner's slots; {@code decoded} counts tokens of the current or last request. */ + record SlotState(boolean processing, long promptTokens, long promptProcessed, long cacheTokens, long decoded, + long contextSize, String speculative, int slots, Instant sampledAt) { + + long contextUsed() { + return Math.max(promptTokens, cacheTokens) + decoded; + } + + int cacheHitPercent() { + if (promptTokens <= 0) { + return 0; + } + return (int) Math.min(100, cacheTokens * 100 / promptTokens); + } + } + + record RunnerInfo(long pid, int port, long contextSize, int parallel, String modelPath, String executable) { + } + + record GpuStats(String name, int utilizationPercent, long memoryUsedBytes, long memoryTotalBytes, int count) { + } + + record ProcessStats(long pid, String label, double cpuPercent, long rssBytes) { + } + + record HostStats(GpuStats gpu, ProcessStats server, ProcessStats runner, Instant sampledAt) { + } + + enum RequestSource { + TUI, + ROUTE + } + + /** + * One request as Ollama reported it (TUI) or as the GenAI span of a route recorded it (no phase timings). + * {@code contextSize} is the context window of the runner that served it, 0 when unknown. + */ + record RequestEntry(Instant timestamp, RequestSource source, String routeId, String model, int inputTokens, + int outputTokens, int cachedTokens, long prefillMs, long decodeMs, long loadMs, long totalMs, + String doneReason, long contextSize, int question, String questionText) { + + RequestEntry(Instant timestamp, RequestSource source, String routeId, String model, int inputTokens, + int outputTokens, int cachedTokens, long prefillMs, long decodeMs, long loadMs, long totalMs, + String doneReason, long contextSize) { + this(timestamp, source, routeId, model, inputTokens, outputTokens, cachedTokens, prefillMs, decodeMs, + loadMs, totalMs, doneReason, contextSize, 0, null); + } + + /** Requests of the same AI panel question (its tool-call steps) share this key; a route call stands alone. */ + String groupKey() { + if (source == RequestSource.ROUTE) { + return "route:" + timestamp.toEpochMilli() + ":" + routeId; + } + return "q" + question + ":" + (questionText != null ? questionText : ""); + } + + /** + * Everything the model had in front of it. Ollama's {@code prompt_eval_count} is the whole prompt; the cached + * count is the part of it served from the KV cache, not an addition. + */ + long promptTokens() { + return inputTokens; + } + + /** Prompt tokens the runner actually had to evaluate this time. */ + long evaluatedTokens() { + return Math.max(0, (long) inputTokens - cachedTokens); + } + + /** Share of the prompt Ollama served from its cache. */ + int cacheHitPercent() { + if (inputTokens <= 0) { + return 0; + } + return (int) Math.min(100, (long) cachedTokens * 100 / inputTokens); + } + + /** Share of the context window the prompt filled, or -1 when the window is unknown. */ + int contextPercent() { + if (contextSize <= 0) { + return -1; + } + return (int) Math.min(100, promptTokens() * 100 / contextSize); + } + + /** Prefill speed over the tokens that were not served from cache. */ + double prefillTokensPerSecond() { + return prefillMs > 0 && evaluatedTokens() > 0 ? evaluatedTokens() * 1000.0 / prefillMs : 0; + } + + double decodeTokensPerSecond() { + return decodeMs > 0 ? outputTokens * 1000.0 / decodeMs : 0; + } + + /** Time to first token: loading the model plus processing the prompt. Zero when timings are unknown. */ + long ttftMs() { + return hasTimings() ? loadMs + prefillMs : 0; + } + + boolean hasTimings() { + return prefillMs > 0 || decodeMs > 0; + } + + /** A load of a second or more means the model was not in memory when the request arrived. */ + boolean coldStart() { + return loadMs >= 1000; + } + } + + /** + * One AI panel question with all the requests it took (the model's tool calls each cost a request), oldest step + * first, or a single route call. Aggregates are what the user experienced for the whole question. + */ + record QuestionGroup(String key, RequestSource source, int question, String questionText, String routeId, + List<RequestEntry> steps) { + + RequestEntry first() { + return steps.get(0); + } + + RequestEntry last() { + return steps.get(steps.size() - 1); + } + + /** The largest prompt of the question: how far the context was pushed. */ + long promptTokens() { + long max = 0; + for (RequestEntry e : steps) { + max = Math.max(max, e.promptTokens()); + } + return max; + } + + long outputTokens() { + long sum = 0; + for (RequestEntry e : steps) { + sum += e.outputTokens(); + } + return sum; + } + + int cacheHitPercent() { + long prompt = 0; + long cached = 0; + for (RequestEntry e : steps) { + prompt += e.inputTokens(); + cached += e.cachedTokens(); + } + return prompt > 0 ? (int) Math.min(100, cached * 100 / prompt) : 0; + } + + int contextPercent() { + int max = -1; + for (RequestEntry e : steps) { + max = Math.max(max, e.contextPercent()); + } Review Comment: 📝 **Nit: `HttpClient` is never closed** `HttpClient` implements `AutoCloseable` since Java 21 (which is the minimum for `camel-jbang`). The instance is held for the lifetime of the TUI session and its daemon threads die with the JVM, so this is not a production leak — but the resource-management contract of the type says it should be closed. Having `OllamaMonitor` implement `Closeable` and adding a `close()` method that calls `http.close()` would let `CamelMonitor` close it on TUI shutdown, and makes the intent explicit: ```java void close() { http.close(); } ``` Call site in `CamelMonitor.doCall()` already holds a reference; wiring it into the TUI shutdown path is straightforward. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
