This is an automated email from the ASF dual-hosted git repository. luigidemasi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit 65db362c960bc449746c9a7a66ba73aa66296c35 Author: Luigi De Masi <[email protected]> AuthorDate: Fri Nov 13 09:17:08 2026 +0100 CAMEL-25648: Export semantic assessments with Micrometer Add an optional listener using the application registry for bounded typed result counters, observed duration histograms and expert scalar gauges. Document registry ownership, collection loss and Prometheus queries. Runtime-specific registration and native validation remain follow-up work. Co-authored-by: OpenAI Codex <[email protected]> Signed-off-by: Luigi De Masi <[email protected]> --- .../camel/catalog/docs/semantic-language.adoc | 130 ++++++- components/camel-ai/camel-semantic/pom.xml | 10 + .../src/main/docs/semantic-language.adoc | 130 ++++++- .../micrometer/MicrometerSemanticMetrics.java | 316 +++++++++++++++++ .../micrometer/MicrometerSemanticMetricsTest.java | 388 +++++++++++++++++++++ 5 files changed, 970 insertions(+), 4 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/semantic-language.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/semantic-language.adoc index 93fb065dafbd..29a0a531b1ab 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/semantic-language.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/semantic-language.adoc @@ -1507,8 +1507,9 @@ camel semantic monitoring my-app --source=departmentMetrics The Semantic *Monitoring* sibling, `tui_get_monitoring`, `semantic-monitoring` console and CLI read this same current-value cache. The TUI offers expert/definition grouping, filters and an inspector for counts, durations, distributions, meaning, observation times and collection health. It has no historical-period, reference or -comparison controls. Pausing affects the display only. A shared Micrometer exporter for Camel Main, -Spring Boot and Quarkus is tracked separately by CAMEL-25648; it is not implemented by this local viewer. +comparison controls. Pausing affects the display only. The optional Micrometer listener below exports +measurements independently of this viewer. Runtime-specific automatic registration and scrape endpoint +integration are separate from the shared listener. When JMX is enabled, the context registers one read-only aggregate facade and one per registered metric source under `type=semantic`, with logical names `all` and `Source(<registry-name>)`. @@ -1516,3 +1517,128 @@ The object-name pattern is `<domain>:context="<management-name>",type=semantic,n Each exposes only a JSON `Snapshot` attribute backed by the same cache. There are no per-event MBeans, inference/backend reads, reset, reference-update or detector controls in JMX. Facades are removed at context shutdown. Monitoring also works with JMX disabled and without a connected TUI. + +=== Optional Micrometer instrumentation + +`MicrometerSemanticMetrics` consumes typed assessment completion events directly. Add `io.micrometer:micrometer-core` +to your application, or use the registry dependency supplied by your runtime. Micrometer is an optional dependency +of `camel-semantic`; applications that do not register this listener do not need it. Prometheus is not a runtime +dependency of the listener. + +Register one listener before starting the context, passing the application's existing `MeterRegistry`: + +[source,java] +---- +import org.apache.camel.semantic.micrometer.MicrometerSemanticMetrics; + +// registry is the application's existing Micrometer MeterRegistry. +// Optional, increasing score bucket bounds may include negative values. +context.getManagementStrategy().addEventNotifier( + new MicrometerSemanticMetrics(registry, -1, 0, 0.5, 1, 2, 5)); +---- + +The listener does not create or close a registry or install an HTTP server. The application/runtime owns +registry configuration, HTTP exposure, authentication and the registry lifecycle. Give contexts sharing a registry +distinct names. Registering more than one listener in a context fails at startup to prevent double counting. +Stop removes meters created by the listener; restarting starts new counters. Existing application meters remain. + +Each completed assessment member is counted once, including individual batch members. Starts, request outcomes +and locally aggregated `typed_results` snapshots are not counted again. Provider attempts/retries are not inferred +from semantic completions. Audit capture, input capture, JMX, the local monitoring cache and UI polling are not +required. Collection failures are isolated from the assessment and reported by the collection health meters. + +[cols="2,3",options="header"] +|=== +|Micrometer name |Meaning +|`camel.semantic.evaluations` |Counter by execution `status`: `success`, `failed`, or `not_executed`. +|`camel.semantic.duration` |Timer recording each available execution duration, including failures. Skipped or +missing durations are not zero-duration observations. Percentile histograms use the original durations. +|`camel.semantic.result.availability` |Successful assessments by `availability`: `available` or `unavailable`. +|`camel.semantic.results` |Counter by returned `answer`. Boolean values are final answers, without another +probability threshold. Choice counts the selected answer, not probability entries. Classification counts each +selected label once per assessment, so membership percentages need not sum to 100%. +|`camel.semantic.classification.empty` |Successful classification results with no selected labels. +|`camel.semantic.score.observations` |Count of available finite numeric score observations. +|`camel.semantic.score.sum`, `.minimum`, `.maximum` |Gauges preserving signed, fractional scores since collector +start. The sum can decrease; do not treat it as a monotonic counter. +|`camel.semantic.score.bucket` |Cumulative counters with inclusive `le` bounds supplied to the constructor and a +`+Inf` bucket. Bounds describe numeric values, not severity or policy decisions. +|`camel.semantic.collection.loss` |Counter by fixed `reason`: `population_limit`, `answer_limit`, `invalid_event`, +or `unsupported_measurement`. Loss is excluded from the corresponding result distribution. +|`camel.semantic.collection.errors` |Number of metric collection failures since collector start. +|=== + +Assessment meters have `camel.context`, `expert`, `definition`, `operation`, `origin` and `result.type` tags. +Missing identity is the empty string; direct calls remain distinguishable by origin and the absence of a definition. +The listener retains at most 100 distinct combinations and 64 answer labels of at most 256 characters per combination. +Score histograms accept at most 64 finite, strictly increasing bounds; an empty array keeps count, sum, minimum, +maximum and the overflow bucket. Configure identical bounds on instances you aggregate together. +Micrometer's timer histogram configuration can be adjusted with the application's usual meter filters. + +Prompts, input, interpretation, exception messages and per-execution identifiers are never metric labels. +Choose stable expert/definition names and result labels. Bounds limit retained series, but collection loss still +means that a distribution may cover only part of the evaluated population. Collection error counts do not prove +that every event was delivered: the shared AI notification contract remains synchronous and best effort. + +==== Expert-published scalar gauges + +The same listener accepts `SemanticMonitoringEvent` snapshots with `kind=gauge`, a single finite numeric +`values.value`, and an explicit unit of `1`, `ratio`, `bytes`, `seconds`, `milliseconds`, `nanoseconds`, `count`, +`requests` or `assessments`. If supplied, `availability` must be `available`. It exports +`camel.semantic.expert.value` with `source`, `measurement` (the snapshot ID), `unit` and the supplied +`expert`, `definition`, `operation`, `provider`, `taskId`, `taskVersion`, `revision` dimensions. Missing dimensions +are empty strings. Other dimension keys are rejected, including correlation and message identifiers. + +Values replace earlier values for that identity; repeated publication never increments a counter. At most 100 +such identities are retained, independently of assessment populations. The producer owns meaning and update cadence. +The last value remains until the next publication or collector shutdown; scraping does not refresh it. A publication +with an unavailable or invalid value marks an existing gauge as `NaN`, rather than continuing to expose its old value. +Only published snapshots are consumed: the listener does not call `SemanticMetricSource.snapshots()` or perform I/O +while an assessment completes. Unsupported kinds, units, shapes or dimensions increment `unsupported_measurement`; +aggregated counts are not guessed to be deltas, and snapshot means are never converted to latency observations. + +==== Prometheus queries and lifecycle + +Expose the supplied registry through the runtime's normal metrics endpoint. Prometheus scrapes that endpoint and +owns time-series retention; it does not depend on the Camel TUI being connected. Select a scrape interval that fits +your operational needs (for example 15 seconds) and use a rate window containing several scrapes. +The shared listener does not provide automatic Main, Spring Boot or Quarkus registration, a Prometheus query client, +historical TUI views or dashboards. Runtime-specific examples and Quarkus native validation are follow-up work. + +For a Prometheus registry, dots become underscores and counters gain `_total`. These example backend queries aggregate +across instances while retaining the selected expert or definition grouping: + +[source,promql] +---- +# Executed assessment rate per expert; skipped members are excluded. +sum by (expert) (rate(camel_semantic_evaluations_total{status=~"success|failed"}[5m])) + +# Failed execution rate per expert and definition. +sum by (expert, definition) (rate(camel_semantic_evaluations_total{status="failed"}[5m])) + +# Returned-answer distribution per expert and definition. +sum by (expert, definition, result_type, answer) (rate(camel_semantic_results_total[5m])) + +# 95th percentile from actual duration buckets, aggregated across instances. +histogram_quantile(0.95, + sum by (expert, le) (rate(camel_semantic_duration_seconds_bucket[5m]))) + +# Available score distribution using application-declared bounds. +sum by (expert, definition, le) (rate(camel_semantic_score_bucket_total[5m])) + +# Counter resets observed on individual application series. +resets(camel_semantic_evaluations_total[1h]) + +# Collection exclusions over the observation window. +sum by (reason) (increase(camel_semantic_collection_loss_total[5m])) +---- + +Apply `rate` or `increase` before summing across instances so resets are handled per series. A restart resets these +counters; Prometheus history remains in Prometheus. The first scrape after startup cannot reconstruct evaluations +that happened before a counter's first observed sample. Backend labels such as `job` and `instance` distinguish +application instances; they are supplied by the scrape configuration, not unique evaluation IDs. + +Exemplars depend on the runtime's tracing context, registry and exposition format. This listener neither fabricates +trace context nor promises an exemplar for every assessment, particularly for detached or asynchronous completions. +Historical investigations may query independently retained audit records for a matching `[since, until)` interval, +definition/operation and typed result. Audit retention and metric observations can cover different populations. diff --git a/components/camel-ai/camel-semantic/pom.xml b/components/camel-ai/camel-semantic/pom.xml index b3571086b700..508b8c70d295 100644 --- a/components/camel-ai/camel-semantic/pom.xml +++ b/components/camel-ai/camel-semantic/pom.xml @@ -29,6 +29,16 @@ <name>Camel :: AI :: Semantic Evaluation</name> <description>Invoke named evaluations of message content to produce boolean decisions, categories, scores and label sets through provider adapters</description> <dependencies> + <dependency> + <groupId>io.micrometer</groupId> + <artifactId>micrometer-core</artifactId> + <optional>true</optional> + </dependency> + <dependency> + <groupId>io.micrometer</groupId> + <artifactId>micrometer-registry-prometheus</artifactId> + <scope>test</scope> + </dependency> <dependency> <groupId>org.apache.camel</groupId> <artifactId>camel-support</artifactId> diff --git a/components/camel-ai/camel-semantic/src/main/docs/semantic-language.adoc b/components/camel-ai/camel-semantic/src/main/docs/semantic-language.adoc index 93fb065dafbd..29a0a531b1ab 100644 --- a/components/camel-ai/camel-semantic/src/main/docs/semantic-language.adoc +++ b/components/camel-ai/camel-semantic/src/main/docs/semantic-language.adoc @@ -1507,8 +1507,9 @@ camel semantic monitoring my-app --source=departmentMetrics The Semantic *Monitoring* sibling, `tui_get_monitoring`, `semantic-monitoring` console and CLI read this same current-value cache. The TUI offers expert/definition grouping, filters and an inspector for counts, durations, distributions, meaning, observation times and collection health. It has no historical-period, reference or -comparison controls. Pausing affects the display only. A shared Micrometer exporter for Camel Main, -Spring Boot and Quarkus is tracked separately by CAMEL-25648; it is not implemented by this local viewer. +comparison controls. Pausing affects the display only. The optional Micrometer listener below exports +measurements independently of this viewer. Runtime-specific automatic registration and scrape endpoint +integration are separate from the shared listener. When JMX is enabled, the context registers one read-only aggregate facade and one per registered metric source under `type=semantic`, with logical names `all` and `Source(<registry-name>)`. @@ -1516,3 +1517,128 @@ The object-name pattern is `<domain>:context="<management-name>",type=semantic,n Each exposes only a JSON `Snapshot` attribute backed by the same cache. There are no per-event MBeans, inference/backend reads, reset, reference-update or detector controls in JMX. Facades are removed at context shutdown. Monitoring also works with JMX disabled and without a connected TUI. + +=== Optional Micrometer instrumentation + +`MicrometerSemanticMetrics` consumes typed assessment completion events directly. Add `io.micrometer:micrometer-core` +to your application, or use the registry dependency supplied by your runtime. Micrometer is an optional dependency +of `camel-semantic`; applications that do not register this listener do not need it. Prometheus is not a runtime +dependency of the listener. + +Register one listener before starting the context, passing the application's existing `MeterRegistry`: + +[source,java] +---- +import org.apache.camel.semantic.micrometer.MicrometerSemanticMetrics; + +// registry is the application's existing Micrometer MeterRegistry. +// Optional, increasing score bucket bounds may include negative values. +context.getManagementStrategy().addEventNotifier( + new MicrometerSemanticMetrics(registry, -1, 0, 0.5, 1, 2, 5)); +---- + +The listener does not create or close a registry or install an HTTP server. The application/runtime owns +registry configuration, HTTP exposure, authentication and the registry lifecycle. Give contexts sharing a registry +distinct names. Registering more than one listener in a context fails at startup to prevent double counting. +Stop removes meters created by the listener; restarting starts new counters. Existing application meters remain. + +Each completed assessment member is counted once, including individual batch members. Starts, request outcomes +and locally aggregated `typed_results` snapshots are not counted again. Provider attempts/retries are not inferred +from semantic completions. Audit capture, input capture, JMX, the local monitoring cache and UI polling are not +required. Collection failures are isolated from the assessment and reported by the collection health meters. + +[cols="2,3",options="header"] +|=== +|Micrometer name |Meaning +|`camel.semantic.evaluations` |Counter by execution `status`: `success`, `failed`, or `not_executed`. +|`camel.semantic.duration` |Timer recording each available execution duration, including failures. Skipped or +missing durations are not zero-duration observations. Percentile histograms use the original durations. +|`camel.semantic.result.availability` |Successful assessments by `availability`: `available` or `unavailable`. +|`camel.semantic.results` |Counter by returned `answer`. Boolean values are final answers, without another +probability threshold. Choice counts the selected answer, not probability entries. Classification counts each +selected label once per assessment, so membership percentages need not sum to 100%. +|`camel.semantic.classification.empty` |Successful classification results with no selected labels. +|`camel.semantic.score.observations` |Count of available finite numeric score observations. +|`camel.semantic.score.sum`, `.minimum`, `.maximum` |Gauges preserving signed, fractional scores since collector +start. The sum can decrease; do not treat it as a monotonic counter. +|`camel.semantic.score.bucket` |Cumulative counters with inclusive `le` bounds supplied to the constructor and a +`+Inf` bucket. Bounds describe numeric values, not severity or policy decisions. +|`camel.semantic.collection.loss` |Counter by fixed `reason`: `population_limit`, `answer_limit`, `invalid_event`, +or `unsupported_measurement`. Loss is excluded from the corresponding result distribution. +|`camel.semantic.collection.errors` |Number of metric collection failures since collector start. +|=== + +Assessment meters have `camel.context`, `expert`, `definition`, `operation`, `origin` and `result.type` tags. +Missing identity is the empty string; direct calls remain distinguishable by origin and the absence of a definition. +The listener retains at most 100 distinct combinations and 64 answer labels of at most 256 characters per combination. +Score histograms accept at most 64 finite, strictly increasing bounds; an empty array keeps count, sum, minimum, +maximum and the overflow bucket. Configure identical bounds on instances you aggregate together. +Micrometer's timer histogram configuration can be adjusted with the application's usual meter filters. + +Prompts, input, interpretation, exception messages and per-execution identifiers are never metric labels. +Choose stable expert/definition names and result labels. Bounds limit retained series, but collection loss still +means that a distribution may cover only part of the evaluated population. Collection error counts do not prove +that every event was delivered: the shared AI notification contract remains synchronous and best effort. + +==== Expert-published scalar gauges + +The same listener accepts `SemanticMonitoringEvent` snapshots with `kind=gauge`, a single finite numeric +`values.value`, and an explicit unit of `1`, `ratio`, `bytes`, `seconds`, `milliseconds`, `nanoseconds`, `count`, +`requests` or `assessments`. If supplied, `availability` must be `available`. It exports +`camel.semantic.expert.value` with `source`, `measurement` (the snapshot ID), `unit` and the supplied +`expert`, `definition`, `operation`, `provider`, `taskId`, `taskVersion`, `revision` dimensions. Missing dimensions +are empty strings. Other dimension keys are rejected, including correlation and message identifiers. + +Values replace earlier values for that identity; repeated publication never increments a counter. At most 100 +such identities are retained, independently of assessment populations. The producer owns meaning and update cadence. +The last value remains until the next publication or collector shutdown; scraping does not refresh it. A publication +with an unavailable or invalid value marks an existing gauge as `NaN`, rather than continuing to expose its old value. +Only published snapshots are consumed: the listener does not call `SemanticMetricSource.snapshots()` or perform I/O +while an assessment completes. Unsupported kinds, units, shapes or dimensions increment `unsupported_measurement`; +aggregated counts are not guessed to be deltas, and snapshot means are never converted to latency observations. + +==== Prometheus queries and lifecycle + +Expose the supplied registry through the runtime's normal metrics endpoint. Prometheus scrapes that endpoint and +owns time-series retention; it does not depend on the Camel TUI being connected. Select a scrape interval that fits +your operational needs (for example 15 seconds) and use a rate window containing several scrapes. +The shared listener does not provide automatic Main, Spring Boot or Quarkus registration, a Prometheus query client, +historical TUI views or dashboards. Runtime-specific examples and Quarkus native validation are follow-up work. + +For a Prometheus registry, dots become underscores and counters gain `_total`. These example backend queries aggregate +across instances while retaining the selected expert or definition grouping: + +[source,promql] +---- +# Executed assessment rate per expert; skipped members are excluded. +sum by (expert) (rate(camel_semantic_evaluations_total{status=~"success|failed"}[5m])) + +# Failed execution rate per expert and definition. +sum by (expert, definition) (rate(camel_semantic_evaluations_total{status="failed"}[5m])) + +# Returned-answer distribution per expert and definition. +sum by (expert, definition, result_type, answer) (rate(camel_semantic_results_total[5m])) + +# 95th percentile from actual duration buckets, aggregated across instances. +histogram_quantile(0.95, + sum by (expert, le) (rate(camel_semantic_duration_seconds_bucket[5m]))) + +# Available score distribution using application-declared bounds. +sum by (expert, definition, le) (rate(camel_semantic_score_bucket_total[5m])) + +# Counter resets observed on individual application series. +resets(camel_semantic_evaluations_total[1h]) + +# Collection exclusions over the observation window. +sum by (reason) (increase(camel_semantic_collection_loss_total[5m])) +---- + +Apply `rate` or `increase` before summing across instances so resets are handled per series. A restart resets these +counters; Prometheus history remains in Prometheus. The first scrape after startup cannot reconstruct evaluations +that happened before a counter's first observed sample. Backend labels such as `job` and `instance` distinguish +application instances; they are supplied by the scrape configuration, not unique evaluation IDs. + +Exemplars depend on the runtime's tracing context, registry and exposition format. This listener neither fabricates +trace context nor promises an exemplar for every assessment, particularly for detached or asynchronous completions. +Historical investigations may query independently retained audit records for a matching `[since, until)` interval, +definition/operation and typed result. Audit retention and metric observations can cover different populations. diff --git a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/micrometer/MicrometerSemanticMetrics.java b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/micrometer/MicrometerSemanticMetrics.java new file mode 100644 index 000000000000..39348968f8b6 --- /dev/null +++ b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/micrometer/MicrometerSemanticMetrics.java @@ -0,0 +1,316 @@ +/* + * 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.semantic.micrometer; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.Gauge; +import io.micrometer.core.instrument.Meter; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Tags; +import io.micrometer.core.instrument.Timer; +import org.apache.camel.semantic.SemanticEvent; +import org.apache.camel.semantic.SemanticExpert.ResultType; +import org.apache.camel.semantic.SemanticMonitoringEvent; +import org.apache.camel.spi.CamelEvent; +import org.apache.camel.support.EventNotifierSupport; + +/** + * Optional Micrometer instrumentation of completed semantic assessment members. Register one notifier with the Camel + * management strategy, supplying the application's registry. The registry remains application-owned; this notifier + * removes its meters on stop. It never reads message input, audit history or a remote monitoring backend. + * + * @since 4.24 + */ +public final class MicrometerSemanticMetrics extends EventNotifierSupport { + private static final int MAX_POPULATIONS = 100; + private static final int MAX_ANSWERS = 64; + private static final Set<String> STATUSES = Set.of("success", "failed", "not_executed"); + private static final Set<String> UNITS = Set.of("1", "ratio", "bytes", "seconds", "milliseconds", "nanoseconds", + "count", "requests", "assessments"); + private static final Set<String> DIMENSIONS = Set.of("expert", "definition", "operation", "provider", + "taskId", "taskVersion", "revision"); + private final MeterRegistry registry; + private final double[] scoreBounds; + private final Map<Tags, Population> populations = new HashMap<>(); + private final Map<Tags, AtomicLong> measurements = new HashMap<>(); + private final Set<Meter> owned = new HashSet<>(); + private final AtomicLong errors = new AtomicLong(); + private Set<Meter> existing = Set.of(); + private Tags contextTags; + private Counter populationLoss; + private Counter answerLoss; + private Counter invalidEvents; + private Counter unsupportedMeasurements; + + /** + * @param registry application-owned registry, shared with the runtime's usual metrics endpoint + * @param scoreBounds finite, increasing, inclusive score histogram bounds; negative bounds are supported. An empty + * array exports count, sum, minimum and maximum without finite histogram buckets. + */ + public MicrometerSemanticMetrics(MeterRegistry registry, double... scoreBounds) { + setupIgnore(true); + this.registry = Objects.requireNonNull(registry, "registry"); + this.scoreBounds = Objects.requireNonNull(scoreBounds, "scoreBounds").clone(); + if (scoreBounds.length > 64) { + throw new IllegalArgumentException("At most 64 score bounds are supported"); + } + for (int i = 0; i < scoreBounds.length; i++) { + if (!Double.isFinite(scoreBounds[i]) || i > 0 && scoreBounds[i] <= scoreBounds[i - 1]) { + throw new IllegalArgumentException("Score bounds must be finite and increasing"); + } + } + } + + @Override + public boolean isEnabled(CamelEvent event) { + return event instanceof SemanticEvent semantic && semantic.getKind() == SemanticEvent.Kind.EVALUATION_COMPLETED + || event instanceof SemanticMonitoringEvent; + } + + @Override + public synchronized void notify(CamelEvent event) { + if (!isStarted() || !isEnabled(event)) { + return; + } + try { + if (event instanceof SemanticEvent semantic) { + record(semantic.getData()); + } else { + measurement(((SemanticMonitoringEvent) event).getSnapshot().toMap()); + } + } catch (Exception | AssertionError | LinkageError failure) { + // Instrumentation must not change the assessment or expose payloads through exception logging. + errors.incrementAndGet(); + } + } + + private void measurement(Map<String, Object> data) { + if ("typed_results".equals(data.get("kind"))) { + return; // Assessment events are the only ingestion path for result statistics. + } + Map<?, ?> dimensions = data.get("dimensions") instanceof Map<?, ?> map ? map : Map.of(); + if (!"gauge".equals(data.get("kind")) || !UNITS.contains(text(data, "unit")) + || !DIMENSIONS.containsAll(dimensions.keySet())) { + unsupportedMeasurements.increment(); + return; + } + Tags tags = contextTags.and("source", text(data, "source"), "measurement", text(data, "id"), + "unit", text(data, "unit")); + // Every series has the same tag keys, including absent dimensions, as required by Prometheus. + for (String dimension : DIMENSIONS) { + tags = tags.and(dimension, text(dimensions, dimension)); + } + AtomicLong current = measurements.get(tags); + if (!(data.get("values") instanceof Map<?, ?> values) || values.size() != 1 + || !(values.get("value") instanceof Number number) || !Double.isFinite(number.doubleValue()) + || data.containsKey("availability") && !"available".equals(data.get("availability"))) { + if (current != null) { + current.set(Double.doubleToRawLongBits(Double.NaN)); + } + unsupportedMeasurements.increment(); + return; + } + if (current == null) { + if (measurements.size() >= MAX_POPULATIONS) { + populationLoss.increment(); + return; + } + current = new AtomicLong(); + own(Gauge.builder("camel.semantic.expert.value", current, v -> Double.longBitsToDouble(v.get())) + .description("Expert-published current scalar; unit and measurement identify its meaning") + .tags(tags).register(registry)); + measurements.put(tags, current); + } + current.set(Double.doubleToRawLongBits(number.doubleValue())); + } + + private void record(Map<String, Object> data) { + String status = text(data, "status"); + if (!STATUSES.contains(status)) { + invalidEvents.increment(); + return; + } + String type = data.get("semantics") instanceof Map<?, ?> semantics ? text(semantics, "resultType") : ""; + if (!Set.of("", "BOOLEAN", "CHOICE", "SCORE", "CLASSIFICATION").contains(type)) { + invalidEvents.increment(); + return; + } + Tags tags = contextTags.and("expert", text(data, "expert"), "definition", text(data, "definition"), + "operation", text(data, "operation"), "origin", text(data, "origin"), "result.type", type); + Population population = populations.get(tags); + if (population == null) { + if (populations.size() >= MAX_POPULATIONS) { + populationLoss.increment(); + return; + } + population = new Population(tags); + populations.put(tags, population); + } + population.outcomes.computeIfAbsent(status, + value -> counter("evaluations", tags.and("status", value))).increment(); + if (!"not_executed".equals(status) && data.get("durationNanos") instanceof Number duration + && duration.longValue() >= 0) { + population.durations.computeIfAbsent(status, value -> own(Timer.builder("camel.semantic.duration") + .description("Observed semantic assessment duration, including failed executions") + .tags(tags.and("status", value)).publishPercentileHistogram().register(registry))) + .record(duration.longValue(), TimeUnit.NANOSECONDS); + } + if (!"success".equals(status)) { + return; + } + Object value = data.get("result") instanceof Map<?, ?> result ? result.get("value") : null; + boolean available = switch (type) { + case "BOOLEAN" -> value instanceof Boolean; + case "CHOICE" -> value instanceof String; + case "SCORE" -> value instanceof Number number && Double.isFinite(number.doubleValue()); + case "CLASSIFICATION" -> value instanceof List<?> list && list.stream().allMatch(String.class::isInstance); + default -> false; + }; + population.availability.computeIfAbsent(available ? "available" : "unavailable", + state -> counter("result.availability", tags.and("availability", state))).increment(); + if (!available) { + return; + } + switch (ResultType.valueOf(type)) { + case BOOLEAN, CHOICE -> population.answer(value.toString()); + case CLASSIFICATION -> { + var labels = (List<?>) value; + if (labels.isEmpty()) { + if (population.empty == null) { + population.empty = counter("classification.empty", tags); + } + population.empty.increment(); + } + labels.stream().distinct().map(String.class::cast).forEach(population::answer); + } + case SCORE -> population.score(((Number) value).doubleValue()); + } + } + + private static String text(Map<?, ?> data, String key) { + return data.get(key) instanceof String value ? value : ""; + } + + private Counter counter(String name, Tags tags) { + return own(Counter.builder("camel.semantic." + name).tags(tags).register(registry)); + } + + private <T extends Meter> T own(T meter) { + if (!existing.contains(meter)) { + owned.add(meter); + } + return meter; + } + + @Override + protected synchronized void doStart() throws Exception { + super.doStart(); + long collectors = getCamelContext().getManagementStrategy().getEventNotifiers().stream() + .filter(MicrometerSemanticMetrics.class::isInstance).count(); + if (collectors > 1) { + throw new IllegalStateException("Only one Semantic Micrometer collector may be registered per CamelContext"); + } + existing = new HashSet<>(registry.getMeters()); + contextTags = Tags.of("camel.context", getCamelContext().getName()); + errors.set(0); + own(Gauge.builder("camel.semantic.collection.errors", errors, AtomicLong::doubleValue) + .description("Metric collection failures since this collector started").tags(contextTags).register(registry)); + populationLoss = counter("collection.loss", contextTags.and("reason", "population_limit")); + answerLoss = counter("collection.loss", contextTags.and("reason", "answer_limit")); + invalidEvents = counter("collection.loss", contextTags.and("reason", "invalid_event")); + unsupportedMeasurements = counter("collection.loss", contextTags.and("reason", "unsupported_measurement")); + } + + @Override + protected synchronized void doStop() throws Exception { + for (Meter meter : owned) { + registry.remove(meter); + } + owned.clear(); + existing = Set.of(); + populations.clear(); + measurements.clear(); + super.doStop(); + } + + private final class Population { + private final Tags tags; + private final Map<String, Counter> outcomes = new HashMap<>(); + private final Map<String, Timer> durations = new HashMap<>(); + private final Map<String, Counter> availability = new HashMap<>(); + private final Map<String, Counter> answers = new HashMap<>(); + private final List<Counter> buckets = new ArrayList<>(); + private Counter empty; + private Counter scores; + private volatile double sum; + private volatile double minimum = Double.NaN; + private volatile double maximum = Double.NaN; + + private Population(Tags tags) { + this.tags = tags; + } + + private void answer(String answer) { + Counter counter = answers.get(answer); + if (counter == null) { + if (answers.size() >= MAX_ANSWERS || answer.length() > 256) { + answerLoss.increment(); + return; + } + counter = counter("results", tags.and("answer", answer)); + answers.put(answer, counter); + } + counter.increment(); + } + + private void score(double value) { + if (scores == null) { + Counter observations = counter("score.observations", tags); + own(Gauge.builder("camel.semantic.score.sum", this, p -> p.sum).tags(tags).register(registry)); + own(Gauge.builder("camel.semantic.score.minimum", this, p -> p.minimum).tags(tags).register(registry)); + own(Gauge.builder("camel.semantic.score.maximum", this, p -> p.maximum).tags(tags).register(registry)); + List<Counter> registeredBuckets = new ArrayList<>(); + for (double bound : scoreBounds) { + registeredBuckets.add(counter("score.bucket", tags.and("le", Double.toString(bound)))); + } + registeredBuckets.add(counter("score.bucket", tags.and("le", "+Inf"))); + buckets.addAll(registeredBuckets); + scores = observations; + } + scores.increment(); + sum += value; + minimum = Double.isNaN(minimum) ? value : Math.min(minimum, value); + maximum = Double.isNaN(maximum) ? value : Math.max(maximum, value); + for (int i = 0; i < scoreBounds.length; i++) { + if (value <= scoreBounds[i]) { + buckets.get(i).increment(); + } + } + buckets.get(scoreBounds.length).increment(); + } + } +} diff --git a/components/camel-ai/camel-semantic/src/test/java/org/apache/camel/semantic/micrometer/MicrometerSemanticMetricsTest.java b/components/camel-ai/camel-semantic/src/test/java/org/apache/camel/semantic/micrometer/MicrometerSemanticMetricsTest.java new file mode 100644 index 000000000000..3acca7ddb481 --- /dev/null +++ b/components/camel-ai/camel-semantic/src/test/java/org/apache/camel/semantic/micrometer/MicrometerSemanticMetricsTest.java @@ -0,0 +1,388 @@ +/* + * 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.semantic.micrometer; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import io.micrometer.core.instrument.Meter; +import io.micrometer.core.instrument.config.MeterFilter; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.micrometer.prometheusmetrics.PrometheusConfig; +import io.micrometer.prometheusmetrics.PrometheusMeterRegistry; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.language.semantic.SemanticLanguage; +import org.apache.camel.semantic.SemanticAdapter; +import org.apache.camel.semantic.SemanticAuditQuery; +import org.apache.camel.semantic.SemanticCapabilities; +import org.apache.camel.semantic.SemanticEvaluation; +import org.apache.camel.semantic.SemanticEvaluations; +import org.apache.camel.semantic.SemanticEvent; +import org.apache.camel.semantic.SemanticExpert; +import org.apache.camel.semantic.SemanticExpert.InputType; +import org.apache.camel.semantic.SemanticExpert.ResultType; +import org.apache.camel.semantic.SemanticMonitoringEvent; +import org.apache.camel.semantic.SemanticMonitoringSnapshot; +import org.apache.camel.semantic.SemanticOperation; +import org.apache.camel.semantic.SemanticResult; +import org.apache.camel.semantic.SemanticResultMetrics; +import org.apache.camel.semantic.internal.SemanticAuditService; +import org.apache.camel.support.DefaultExchange; +import org.apache.camel.support.EventHelper; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.within; + +class MicrometerSemanticMetricsTest { + @Test + void semanticEvaluationDoesNotRequireMicrometerOnTheApplicationClasspath(@TempDir Path temporary) throws Exception { + List<String> classpath = new ArrayList<>(); + for (String entry : System.getProperty("surefire.test.class.path", System.getProperty("java.class.path")) + .split(File.pathSeparator)) { + if (!Path.of(entry).getFileName().toString().startsWith("micrometer-")) { + classpath.add(entry); + } + } + Path output = temporary.resolve("without-micrometer.log"); + var process = new ProcessBuilder( + Path.of(System.getProperty("java.home"), "bin", "java").toString(), + "-cp", String.join(File.pathSeparator, classpath), WithoutMicrometer.class.getName()) + .redirectErrorStream(true).redirectOutput(output.toFile()).start(); + try { + assertThat(process.waitFor(30, TimeUnit.SECONDS)).as("isolated semantic evaluation completed").isTrue(); + assertThat(process.exitValue()).as(Files.readString(output)).isZero(); + } finally { + process.destroyForcibly(); + } + } + + @Test + void rejectsInvalidHistogramConfiguration() { + var registry = new SimpleMeterRegistry(); + try { + for (double[] bounds : List.of(new double[] { 2, 1 }, new double[] { 1, 1 }, + new double[] { Double.NaN }, new double[] { Double.POSITIVE_INFINITY }, new double[65])) { + assertThatThrownBy(() -> new MicrometerSemanticMetrics(registry, bounds)) + .isInstanceOf(IllegalArgumentException.class); + } + } finally { + registry.close(); + } + } + + @Test + void realDirectAndBatchEvaluationsWorkWithoutAuditAndAlongsideLocalMonitoring() throws Exception { + var registry = new SimpleMeterRegistry(); + try (AutoCloseable closeRegistry = registry::close; var context = new DefaultCamelContext()) { + context.disableJMX(); + context.getManagementStrategy().addEventNotifier(new MicrometerSemanticMetrics(registry)); + context.getRegistry().bind("expert", new BooleanExpert()); + context.getRegistry().bind("local", new SemanticResultMetrics( + "local", "Boolean answers", Map.of(), ResultType.BOOLEAN, event -> true, 2)); + var evaluation = new SemanticEvaluation("containsText", "expert", null, Map.of()); + SemanticEvaluations.get(context).replace("test", Map.of("first", evaluation, "second", evaluation)); + context.start(); + var language = (SemanticLanguage) context.resolveLanguage("semantic"); + assertThat(language.evaluate(evaluation, "private-input").getValue()).isEqualTo(true); + var exchange = new DefaultExchange(context); + exchange.getMessage().setBody("batch input"); + assertThat(language.createExpression("refs:first,second").evaluate(exchange, Map.class)).hasSize(2); + assertThat(registry.find("camel.semantic.evaluations").counters().stream().mapToDouble(c -> c.count()).sum()) + .isEqualTo(3); + assertThat(registry.get("camel.semantic.evaluations").tag("definition", "first").counter().count()).isEqualTo(1); + assertThat(registry.get("camel.semantic.evaluations").tag("definition", "second").counter().count()).isEqualTo(1); + assertThat(registry.get("camel.semantic.evaluations").tag("definition", "").tag("origin", "direct") + .counter().count()).isEqualTo(1); + assertThat(SemanticAuditService.get(context).getReader().query(new SemanticAuditQuery(Map.of(), null, null, 10)) + .getRecords()).isEmpty(); + } + } + + @Test + void suppliedScalarGaugeReplacesValuesAndRejectsUnsafeDimensionsAndAmbiguousAggregates() throws Exception { + var registry = new SimpleMeterRegistry(); + try (AutoCloseable closeRegistry = registry::close; var context = new DefaultCamelContext()) { + context.getManagementStrategy().addEventNotifier(new MicrometerSemanticMetrics(registry)); + context.start(); + for (int value : new int[] { 3, 5 }) { + var snapshot = new SemanticMonitoringSnapshot( + Map.of( + "id", "queue", "source", "expert", "name", "Queue", "meaning", "Pending requests", + "kind", "gauge", "unit", "requests", "values", Map.of("value", value))); + EventHelper.notifyAiEvent(context, () -> new SemanticMonitoringEvent(snapshot)); + } + assertThat(registry.get("camel.semantic.expert.value").gauge().value()).isEqualTo(5); + var unsafe = new SemanticMonitoringSnapshot( + Map.of( + "id", "queue", "source", "expert", "name", "Queue", "meaning", "Pending requests", + "kind", "gauge", "unit", "requests", "values", Map.of("value", 8), + "dimensions", Map.of("exchangeId", "private-id"))); + EventHelper.notifyAiEvent(context, () -> new SemanticMonitoringEvent(unsafe)); + assertThat(registry.get("camel.semantic.collection.loss").tag("reason", "unsupported_measurement") + .counter().count()).isEqualTo(1); + assertThat(registry.find("camel.semantic.expert.value").gauges()).hasSize(1); + assertThat(registry.getMeters()).allSatisfy(meter -> assertThat(meter.getId().getTags().toString()) + .doesNotContain("private-id", "exchangeId")); + var unavailable = new SemanticMonitoringSnapshot( + Map.of( + "id", "queue", "source", "expert", "name", "Queue", "meaning", "Pending requests", + "kind", "gauge", "unit", "requests", "availability", "unavailable")); + EventHelper.notifyAiEvent(context, () -> new SemanticMonitoringEvent(unavailable)); + assertThat(registry.get("camel.semantic.expert.value").gauge().value()).isNaN(); + } + } + + @Test + void scoreCollectionRecoversAfterAnInitialMeterRegistrationFailure() throws Exception { + var registry = new SimpleMeterRegistry(); + try (AutoCloseable closeRegistry = registry::close; var context = new DefaultCamelContext()) { + var rejectOnce = new AtomicBoolean(true); + registry.config().meterFilter(new MeterFilter() { + @Override + public Meter.Id map(Meter.Id id) { + if (id.getName().equals("camel.semantic.score.minimum") && rejectOnce.getAndSet(false)) { + throw new IllegalStateException("Temporary registration failure"); + } + return id; + } + }); + context.getManagementStrategy().addEventNotifier(new MicrometerSemanticMetrics(registry, 0, 1, 2)); + context.start(); + send(context, event(ResultType.SCORE, Map.of("value", 0.4), "success", 1)); + send(context, event(ResultType.SCORE, Map.of("value", 1.6), "success", 1)); + assertThat(registry.get("camel.semantic.collection.errors").gauge().value()).isEqualTo(1); + assertThat(registry.get("camel.semantic.score.observations").counter().count()).isEqualTo(1); + assertThat(registry.get("camel.semantic.score.sum").gauge().value()).isEqualTo(1.6); + assertThat(registry.get("camel.semantic.score.bucket").tag("le", "+Inf").counter().count()).isEqualTo(1); + } + } + + @Test + void countsOnlyCompletionsAndUsesReturnedBooleanWithoutAnotherThreshold() throws Exception { + var registry = new SimpleMeterRegistry(); + try (AutoCloseable closeRegistry = registry::close; var context = new DefaultCamelContext()) { + var metrics = new MicrometerSemanticMetrics(registry); + context.getManagementStrategy().addEventNotifier(metrics); + context.disableJMX(); + context.start(); + var completed = event(ResultType.BOOLEAN, Map.of("value", false, "probability", 0.99), "success", 1000000); + send(context, new SemanticEvent(SemanticEvent.Kind.EVALUATION_STARTED, completed.getData())); + send(context, completed); + send(context, event(ResultType.BOOLEAN, Map.of("value", true), "success", 3000000)); + send(context, event(ResultType.BOOLEAN, Map.of(), "failed", 5000000)); + send(context, event(ResultType.BOOLEAN, Map.of(), "success", -1)); + send(context, event(ResultType.BOOLEAN, Map.of(), "not_executed", 0)); + EventHelper.notifyAiEvent(context, () -> new SemanticMonitoringEvent( + new SemanticMonitoringSnapshot( + Map.of("id", "ignored", "source", "expert", "name", "Results", "meaning", "Counts", + "kind", "typed_results", "unit", "assessments", "values", Map.of("completed", 100))))); + assertThat(registry.get("camel.semantic.evaluations").tag("status", "success").counter().count()).isEqualTo(3); + assertThat(registry.get("camel.semantic.evaluations").tag("status", "failed").counter().count()).isEqualTo(1); + assertThat(registry.get("camel.semantic.results").tag("answer", "false").counter().count()).isEqualTo(1); + assertThat(registry.get("camel.semantic.results").tag("answer", "true").counter().count()).isEqualTo(1); + assertThat(registry.get("camel.semantic.result.availability").tag("availability", "unavailable") + .counter().count()).isEqualTo(1); + var timers = registry.find("camel.semantic.duration").timers(); + assertThat(timers.stream().mapToLong(timer -> timer.count()).sum()).isEqualTo(3); + assertThat(timers.stream().mapToDouble(timer -> timer.totalTime(TimeUnit.MILLISECONDS)).sum()).isEqualTo(9); + assertThat(context.getCamelContextExtension().isEventNotificationApplicable()).isFalse(); + } + } + + @Test + void choicesCountSelectionsAndClassificationCountsMembershipAndEmptySets() throws Exception { + var registry = new SimpleMeterRegistry(); + try (AutoCloseable closeRegistry = registry::close; var context = new DefaultCamelContext()) { + context.getManagementStrategy().addEventNotifier(new MicrometerSemanticMetrics(registry)); + context.start(); + send(context, event(ResultType.CHOICE, + Map.of("value", "billing", "probabilities", Map.of("billing", 0.1, "other", 0.9)), "success", 1)); + send(context, event(ResultType.CLASSIFICATION, Map.of("value", List.of("pii", "secrets")), "success", 1)); + send(context, event(ResultType.CLASSIFICATION, Map.of("value", List.of()), "success", 1)); + assertThat(registry.get("camel.semantic.results").tag("answer", "billing").counter().count()).isEqualTo(1); + assertThat(registry.find("camel.semantic.results").tag("answer", "other").counter()).isNull(); + assertThat(registry.get("camel.semantic.results").tag("answer", "pii").counter().count()).isEqualTo(1); + assertThat(registry.get("camel.semantic.results").tag("answer", "secrets").counter().count()).isEqualTo(1); + assertThat(registry.get("camel.semantic.classification.empty").counter().count()).isEqualTo(1); + } + } + + @Test + void signedFractionalScoresAndObservedDurationsSurvivePrometheusExport() throws Exception { + var registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); + try (AutoCloseable closeRegistry = registry::close; var context = new DefaultCamelContext()) { + context.getManagementStrategy().addEventNotifier(new MicrometerSemanticMetrics(registry, -1, 0, 1, 2)); + context.start(); + send(context, event(ResultType.SCORE, Map.of("value", -0.4), "success", 1000000)); + send(context, event(ResultType.SCORE, Map.of("value", 1.6), "success", 9000000)); + assertThat(registry.get("camel.semantic.score.sum").gauge().value()).isCloseTo(1.2, within(1e-12)); + assertThat(registry.get("camel.semantic.score.minimum").gauge().value()).isEqualTo(-0.4); + assertThat(registry.get("camel.semantic.score.maximum").gauge().value()).isEqualTo(1.6); + assertThat(registry.get("camel.semantic.score.observations").counter().count()).isEqualTo(2); + assertThat(registry.get("camel.semantic.score.bucket").tag("le", "0.0").counter().count()).isEqualTo(1); + assertThat(registry.get("camel.semantic.score.bucket").tag("le", "+Inf").counter().count()).isEqualTo(2); + String scrape = registry.scrape(); + assertThat(scrape).contains("camel_semantic_duration_seconds_bucket", "camel_semantic_duration_seconds_count", + "camel_semantic_duration_seconds_sum", "camel_semantic_score_observations_total") + .doesNotContain("private-input", "eventId", "breadcrumbId", "invocationId"); + assertThat(registry.get("camel.semantic.duration").timer().totalTime(TimeUnit.MILLISECONDS)).isEqualTo(10); + } + } + + @Test + void cardinalityIsBoundedAndLossIsVisible() throws Exception { + var registry = new SimpleMeterRegistry(); + try (AutoCloseable closeRegistry = registry::close; var context = new DefaultCamelContext()) { + context.getManagementStrategy().addEventNotifier(new MicrometerSemanticMetrics(registry)); + context.start(); + for (int i = 0; i < 70; i++) { + send(context, event(ResultType.CHOICE, Map.of("value", "answer-" + i), "success", 1)); + } + assertThat(registry.find("camel.semantic.results").counters()).hasSize(64); + assertThat(registry.get("camel.semantic.collection.loss").tag("reason", "answer_limit").counter().count()) + .isEqualTo(6); + for (int i = 0; i < 110; i++) { + var data = new LinkedHashMap<>(event(ResultType.BOOLEAN, Map.of("value", true), "success", 1).getData()); + data.put("definition", "definition-" + i); + send(context, new SemanticEvent(SemanticEvent.Kind.EVALUATION_COMPLETED, data)); + } + assertThat(registry.get("camel.semantic.collection.loss").tag("reason", "population_limit").counter().count()) + .isEqualTo(11); + assertThat(registry.find("camel.semantic.evaluations").counters()).hasSize(100); + } + } + + @Test + void restartRemovesOnlyOwnedMetersAndNeverClosesTheApplicationRegistry() throws Exception { + var registry = new SimpleMeterRegistry(); + try (AutoCloseable closeRegistry = registry::close; var context = new DefaultCamelContext()) { + var unrelated = registry.counter("application.requests"); + unrelated.increment(); + var metrics = new MicrometerSemanticMetrics(registry); + context.getManagementStrategy().addEventNotifier(metrics); + context.start(); + send(context, event(ResultType.BOOLEAN, Map.of("value", true), "success", 1)); + context.stop(); + assertThat(registry.isClosed()).isFalse(); + assertThat(registry.getMeters()).containsExactly(unrelated); + context.start(); + send(context, event(ResultType.BOOLEAN, Map.of("value", true), "success", 1)); + assertThat(registry.get("camel.semantic.evaluations").counter().count()).isEqualTo(1); + } + } + + @Test + void registryFailureIsVisibleAndDoesNotEscapeTheCallback() throws Exception { + var registry = new SimpleMeterRegistry(); + try (AutoCloseable closeRegistry = registry::close; var context = new DefaultCamelContext()) { + registry.config().meterFilter(new MeterFilter() { + @Override + public Meter.Id map(Meter.Id id) { + if (id.getName().equals("camel.semantic.duration")) { + throw new IllegalStateException("private-provider-error"); + } + return id; + } + }); + var metrics = new MicrometerSemanticMetrics(registry); + context.getManagementStrategy().addEventNotifier(metrics); + context.start(); + metrics.notify(event(ResultType.BOOLEAN, Map.of("value", true), "success", 1)); + assertThat(registry.get("camel.semantic.collection.errors").gauge().value()).isEqualTo(1); + } + } + + @Test + void duplicateCollectorsAreRejectedBeforeTheyCanDoubleCount() throws Exception { + var registry = new SimpleMeterRegistry(); + try (AutoCloseable closeRegistry = registry::close; var context = new DefaultCamelContext()) { + context.getManagementStrategy().addEventNotifier(new MicrometerSemanticMetrics(registry)); + context.getManagementStrategy().addEventNotifier(new MicrometerSemanticMetrics(registry)); + assertThatThrownBy(context::start).hasMessageContaining("Semantic Micrometer"); + } + } + + private static void send(DefaultCamelContext context, SemanticEvent event) { + EventHelper.notifyAiEvent(context, () -> event); + } + + private static SemanticEvent event(ResultType type, Map<String, Object> result, String status, long nanos) { + Map<String, Object> data = new LinkedHashMap<>( + Map.of( + "schemaVersion", 1, "eventId", UUID.randomUUID().toString(), "timestamp", "2026-11-12T12:00:00Z", + "category", "evaluation", "status", status, "expert", "test-expert", "origin", "direct", + "semantics", Map.of("resultType", type.name()))); + if (!result.isEmpty()) { + data.put("result", result); + } + if (nanos >= 0) { + data.put("durationNanos", nanos); + } + return new SemanticEvent(SemanticEvent.Kind.EVALUATION_COMPLETED, data); + } + + @SemanticExpert(name = "text", description = "Checks for nonempty text", provider = "local", artifactId = "test", + operations = @SemanticOperation(name = "containsText", description = "Has text", + inputTypes = InputType.TEXT, inputRequirements = "Text", + resultType = ResultType.BOOLEAN, resultMeaning = "Contains text")) + static class BooleanExpert implements SemanticAdapter { + @Override + public void validate(SemanticEvaluation evaluation) { + SemanticCapabilities.from(getClass()).validate(evaluation); + } + + @Override + public SemanticResult evaluate(SemanticEvaluation evaluation, Object state) { + return new SemanticResult(!state.toString().isBlank(), null, Map.of(), null, Map.of()); + } + } + + public static class WithoutMicrometer { + public static void main(String[] args) throws Exception { + try { + Class.forName("io.micrometer.core.instrument.MeterRegistry"); + throw new IllegalStateException("Micrometer unexpectedly present"); + } catch (ClassNotFoundException expected) { + // This process deliberately omits every Micrometer artifact. + } + try (var context = new DefaultCamelContext()) { + context.disableJMX(); + context.getRegistry().bind("expert", new BooleanExpert()); + context.start(); + var language = (SemanticLanguage) context.resolveLanguage("semantic"); + Object result = language.evaluate(new SemanticEvaluation("containsText", "expert", null, Map.of()), "hello") + .getValue(); + if (!Boolean.TRUE.equals(result)) { + throw new IllegalStateException("Semantic evaluation failed without Micrometer"); + } + } catch (Exception failure) { + throw new IllegalStateException(failure); + } + } + } +}
