luigidemasi commented on code in PR #27791: URL: https://github.com/apache/camel/pull/27791#discussion_r4269041812
########## components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticMonitoringService.java: ########## @@ -0,0 +1,392 @@ +/* + * 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.internal; + +import java.time.Instant; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import javax.management.ObjectName; +import javax.management.StandardMBean; + +import org.apache.camel.CamelContext; +import org.apache.camel.CamelContextAware; +import org.apache.camel.builder.ThreadPoolBuilder; +import org.apache.camel.semantic.SemanticMetricSource; +import org.apache.camel.semantic.SemanticMonitoringEvent; +import org.apache.camel.semantic.SemanticMonitoringSnapshot; +import org.apache.camel.semantic.SemanticMonitoringStore; +import org.apache.camel.spi.CamelEvent; +import org.apache.camel.spi.EventNotifier; +import org.apache.camel.support.EventNotifierSupport; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy; +import org.apache.camel.util.json.JsonObject; + +/** Bounded read-only projection of supplied measurements, shared by all surfaces. */ +public final class SemanticMonitoringService extends EventNotifierSupport { + private static final int CAPACITY = 100; + private static final int MAX_SOURCES = 32; + private final Map<String, SemanticMetricSource> sources = new LinkedHashMap<>(); + private final Map<String, SemanticMonitoringStore> stores = new LinkedHashMap<>(); + private final Map<String, Cached> latest = new LinkedHashMap<>(); + private final Deque<Cached> history = new ArrayDeque<>(); + private final List<EventNotifier> ownedNotifiers = new ArrayList<>(); + private final List<ObjectName> mbeans = new ArrayList<>(); + private final AtomicLong sourceErrors = new AtomicLong(); + private final AtomicLong storeErrors = new AtomicLong(); + private final AtomicLong dropped = new AtomicLong(); + private final AtomicLong evicted = new AtomicLong(); + private final AtomicLong managementErrors = new AtomicLong(); + private volatile ExecutorService writer; + private ScheduledExecutorService sampler; + private volatile String lastSample; + private final Map<String, Map<String, Object>> sourceHealth = new ConcurrentHashMap<>(); + private final Map<String, Map<String, Object>> storeHealth = new ConcurrentHashMap<>(); + private int omittedSources; + private int omittedStores; + private volatile String recovery = "memory_only"; + + private record Cached(String owner, SemanticMonitoringSnapshot snapshot, boolean recovered) { + } + + private SemanticMonitoringService(CamelContext context) { Review Comment: Fixed in 8321d79cecc6 and verified at 55c709c9d4bd: the constructor now calls `setupIgnore(true)`. `SemanticMonitoringTest.semanticMonitoringDoesNotEnableExchangeEvents` checks that exchange notification remains inapplicable both with and without a metrics bean, while semantic AI events still arrive. _Codex (AI-generated) on behalf of @luigidemasi._ ########## components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticResultMetrics.java: ########## @@ -0,0 +1,259 @@ +/* + * 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; + +import java.math.BigDecimal; +import java.math.MathContext; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.function.Predicate; + +import org.apache.camel.semantic.SemanticExpert.ResultType; +import org.apache.camel.spi.CamelEvent; +import org.apache.camel.support.EventNotifierSupport; + +/** + * Bounded typed-result statistics for one producer-defined population. The owner explicitly supplies dimensions and + * selection; no question/model identity, Boolean threshold, label meaning or series version is inferred. Register the + * collector as an EventNotifier and optionally as a SemanticMetricSource for the shared management surfaces. + */ +public final class SemanticResultMetrics extends EventNotifierSupport implements SemanticMetricSource { + private final String source; + private final String meaning; + private final Map<String, String> dimensions; + private final ResultType type; + private final Predicate<SemanticEvent> selection; + private final int maxLabels; + private final double[] bounds; + private final long[] histogram; + private final long[] probabilityHistogram = new long[10]; + private final Map<String, Long> labels = new LinkedHashMap<>(); + private String epoch = UUID.randomUUID().toString(); + private Instant from = Instant.now(); + private Instant to = from; + private long completed; + private long answers; + private long failed; + private long excluded; + private long unavailable; + private long lostLabels; + private long empty; + private long probabilities; + private long timed; + private BigDecimal durationSum = BigDecimal.ZERO; + private Long durationMinimum; + private Long durationMaximum; + private BigDecimal sum = BigDecimal.ZERO; + private BigDecimal minimum; + private BigDecimal maximum; + + public SemanticResultMetrics(String source, String meaning, Map<String, String> dimensions, ResultType type, Review Comment: Fixed in 8321d79cecc6: `SemanticResultMetrics` now calls `setupIgnore(true)`. The metrics-enabled case of `semanticMonitoringDoesNotEnableExchangeEvents` covers this constructor and verifies that receiving semantic AI events does not enable exchange-event creation for the context. _Codex (AI-generated) on behalf of @luigidemasi._ ########## components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticResultMetrics.java: ########## @@ -0,0 +1,259 @@ +/* + * 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; + +import java.math.BigDecimal; +import java.math.MathContext; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.function.Predicate; + +import org.apache.camel.semantic.SemanticExpert.ResultType; +import org.apache.camel.spi.CamelEvent; +import org.apache.camel.support.EventNotifierSupport; + +/** + * Bounded typed-result statistics for one producer-defined population. The owner explicitly supplies dimensions and + * selection; no question/model identity, Boolean threshold, label meaning or series version is inferred. Register the + * collector as an EventNotifier and optionally as a SemanticMetricSource for the shared management surfaces. + */ +public final class SemanticResultMetrics extends EventNotifierSupport implements SemanticMetricSource { + private final String source; + private final String meaning; + private final Map<String, String> dimensions; + private final ResultType type; + private final Predicate<SemanticEvent> selection; + private final int maxLabels; + private final double[] bounds; + private final long[] histogram; + private final long[] probabilityHistogram = new long[10]; + private final Map<String, Long> labels = new LinkedHashMap<>(); + private String epoch = UUID.randomUUID().toString(); + private Instant from = Instant.now(); + private Instant to = from; + private long completed; + private long answers; + private long failed; + private long excluded; + private long unavailable; + private long lostLabels; + private long empty; + private long probabilities; + private long timed; + private BigDecimal durationSum = BigDecimal.ZERO; + private Long durationMinimum; + private Long durationMaximum; + private BigDecimal sum = BigDecimal.ZERO; + private BigDecimal minimum; + private BigDecimal maximum; + + public SemanticResultMetrics(String source, String meaning, Map<String, String> dimensions, ResultType type, + Predicate<SemanticEvent> selection, int maxLabels, double... scoreBounds) { + if (source == null || source.isBlank() || source.length() > 128 || meaning == null || meaning.isBlank() + || meaning.length() > 256 || dimensions.size() > 16 || maxLabels < 2 || maxLabels > 64 + || scoreBounds.length > 64 || type == null || selection == null) { + throw new IllegalArgumentException("Invalid typed metric declaration"); + } + this.source = source; + this.meaning = meaning; + this.dimensions = Map.copyOf(dimensions); + this.type = type; + this.selection = selection; + this.maxLabels = maxLabels; + this.bounds = scoreBounds.clone(); + for (int i = 0; i < bounds.length; i++) { + if (!Double.isFinite(bounds[i]) || i > 0 && bounds[i] <= bounds[i - 1]) { + throw new IllegalArgumentException("Histogram bounds must be finite and increasing"); + } + } + histogram = new long[bounds.length + 1]; + snapshots(); // Validate the supplied descriptions and dimensions before events arrive. + } + + @Override + public boolean isEnabled(CamelEvent event) { + return event instanceof SemanticEvent semantic && semantic.getKind() == SemanticEvent.Kind.EVALUATION_COMPLETED; + } + + @Override + public void notify(CamelEvent event) { + if (!(event instanceof SemanticEvent semantic) || !isEnabled(event) || !selection.test(semantic)) { + return; + } + collect(semantic.getData()); + } + + private synchronized void collect(Map<String, Object> data) { + completed++; + to = Instant.now(); + if (!"not_executed".equals(data.get("status")) && data.get("durationNanos") instanceof Number duration + && duration.longValue() >= 0) { + long nanos = duration.longValue(); + timed++; + durationSum = durationSum.add(BigDecimal.valueOf(nanos)); + durationMinimum = durationMinimum == null ? nanos : Math.min(durationMinimum, nanos); + durationMaximum = durationMaximum == null ? nanos : Math.max(durationMaximum, nanos); + } + if (!"success".equals(data.get("status"))) { + if ("not_executed".equals(data.get("status"))) { + excluded++; + } else { + failed++; + } + return; + } + if (!(data.get("semantics") instanceof Map<?, ?> semantics) || !type.name().equals(semantics.get("resultType")) + || !(data.get("result") instanceof Map<?, ?> result) || !result.containsKey("value")) { + unavailable++; + return; + } + Object value = result.get("value"); + switch (type) { + case BOOLEAN -> { + if (!(value instanceof Boolean)) { + unavailable++; + return; + } + label(value.toString()); + } + case CHOICE -> { + if (!(value instanceof String selected)) { + unavailable++; + return; + } + label(selected); + } + case CLASSIFICATION -> { + if (!(value instanceof List<?> selected) || selected.stream().anyMatch(item -> !(item instanceof String))) { + unavailable++; + return; + } + if (selected.isEmpty()) { + empty++; + } + selected.stream().distinct().map(String.class::cast).forEach(this::label); + } + case SCORE -> { + if (!(value instanceof Number number) || !Double.isFinite(number.doubleValue())) { + unavailable++; + return; + } + BigDecimal score = new BigDecimal(number.toString()); + sum = sum.add(score); + minimum = minimum == null ? score : minimum.min(score); + maximum = maximum == null ? score : maximum.max(score); + int bucket = 0; + while (bucket < bounds.length && number.doubleValue() > bounds[bucket]) { + bucket++; + } + histogram[bucket]++; + } + } + answers++; + if (result.get("probability") instanceof Number probability) { + double valueProbability = probability.doubleValue(); + if (Double.isFinite(valueProbability) && valueProbability >= 0 && valueProbability <= 1) { + probabilityHistogram[Math.min(9, (int) (valueProbability * 10))]++; + probabilities++; + } + } + } + + private void label(String value) { + if (labels.containsKey(value) || labels.size() < maxLabels && value.length() <= 256) { + labels.merge(value, 1L, Long::sum); + } else { + lostLabels++; + } + } + + @Override + public synchronized List<SemanticMonitoringSnapshot> snapshots() { + Map<String, Object> values = new LinkedHashMap<>(); + values.put("completed", completed); + values.put("answers", answers); + values.put("failed", failed); + values.put("notExecuted", excluded); + values.put("unavailable", unavailable); + values.put("labelMembershipLoss", lostLabels); + values.put("providerAttempts", "not_provided"); + values.put("durationCount", timed); + if (timed > 0) { + values.put("meanDurationNanos", durationSum.divide(BigDecimal.valueOf(timed), MathContext.DECIMAL64)); + values.put("minimumDurationNanos", durationMinimum); + values.put("maximumDurationNanos", durationMaximum); + } + if (type == ResultType.SCORE) { + values.put("sum", sum); + if (minimum != null) { + values.put("minimum", minimum); + values.put("maximum", maximum); + } + values.put("histogramUpperBounds", Arrays.stream(bounds).boxed().toList()); + values.put("histogramCounts", Arrays.stream(histogram).boxed().toList()); + values.put("histogramMeaning", "Upper bounds inclusive; final bucket is above the last bound"); + } else { + List<Map<String, Object>> distribution = new ArrayList<>(); + labels.forEach((label, count) -> distribution.add(Map.of("label", label, "count", count, + "denominator", answers, "proportion", answers == 0 ? 0D : count.doubleValue() / answers))); + values.put("distribution", distribution); + if (type == ResultType.CLASSIFICATION) { + values.put("emptySets", empty); + values.put("distributionMeaning", "Per-label membership per assessment; proportions need not sum to one"); + } + } + values.put("reportedProbabilityCount", probabilities); + values.put("reportedProbabilityHistogram", Arrays.stream(probabilityHistogram).boxed().toList()); + values.put("probabilityHistogramMeaning", + "Ten width-0.1 buckets; lower bounds inclusive; 1 belongs to the final bucket"); + Map<String, Object> snapshot = new LinkedHashMap<>(); + snapshot.put("id", source); + snapshot.put("source", source); + snapshot.put("name", type.name() + " result distribution"); + snapshot.put("meaning", meaning); + snapshot.put("kind", "typed_results"); + snapshot.put("unit", "assessments"); + snapshot.put("values", values); + snapshot.put("dimensions", dimensions); + snapshot.put("epoch", epoch); + snapshot.put("from", from.toString()); + snapshot.put("to", to.toString()); + snapshot.put("availability", "available"); + snapshot.put("denominator", answers); + snapshot.put("exclusions", excluded + failed + unavailable); + snapshot.put("loss", lostLabels); + return List.of(new SemanticMonitoringSnapshot(snapshot)); + } + + @Override + protected synchronized void doStart() { + epoch = UUID.randomUUID().toString(); + from = Instant.now(); + to = from; + completed = answers = failed = excluded = unavailable = lostLabels = empty = probabilities = 0; Review Comment: Fixed in 8321d79cecc6: `doStart()` resets `timed`, `durationSum`, `durationMinimum` and `durationMaximum` together with the counters and epoch. `SemanticResultMetricsTest.restartBeginsANewEpochWithoutPreviousDurations` covers stop/start and the first duration in the new epoch. _Codex (AI-generated) on behalf of @luigidemasi._ ########## components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticAuditService.java: ########## @@ -371,9 +394,26 @@ public Invocation begin( observerErrors.incrementAndGet(); } } + EventHelper.notifyAiEvent(context, () -> event); return invocation; } + private Set<String> supportedExtensions(Supplier<Set<String>> capabilities) { + try { + return Set.copyOf(capabilities.get()); + } catch (Exception | AssertionError | LinkageError failure) { + capabilityErrors.incrementAndGet(); + return Set.of(); + } + } + + private boolean hasEventConsumers() { Review Comment: Fixed in 5a81d372b3bf: the eligibility check now skips disabled notifiers and those with `ignoreAiEvents=true`, and isolates failures per notifier as `EventHelper.notifyAiEvent` does. `SemanticEventTest.disabledAuditSkipsOccurrenceConstructionWithoutAnEligibleConsumer` covers absent, disabled, ignored and broken listeners; `failedEligibilityDoesNotPreventAnotherConsumerOrEvaluation` covers delivery to a later eligible listener. _Codex (AI-generated) on behalf of @luigidemasi._ ########## components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticMonitoringService.java: ########## @@ -0,0 +1,392 @@ +/* + * 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.internal; + +import java.time.Instant; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import javax.management.ObjectName; +import javax.management.StandardMBean; + +import org.apache.camel.CamelContext; +import org.apache.camel.CamelContextAware; +import org.apache.camel.builder.ThreadPoolBuilder; +import org.apache.camel.semantic.SemanticMetricSource; +import org.apache.camel.semantic.SemanticMonitoringEvent; +import org.apache.camel.semantic.SemanticMonitoringSnapshot; +import org.apache.camel.semantic.SemanticMonitoringStore; +import org.apache.camel.spi.CamelEvent; +import org.apache.camel.spi.EventNotifier; +import org.apache.camel.support.EventNotifierSupport; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy; +import org.apache.camel.util.json.JsonObject; + +/** Bounded read-only projection of supplied measurements, shared by all surfaces. */ +public final class SemanticMonitoringService extends EventNotifierSupport { + private static final int CAPACITY = 100; + private static final int MAX_SOURCES = 32; + private final Map<String, SemanticMetricSource> sources = new LinkedHashMap<>(); + private final Map<String, SemanticMonitoringStore> stores = new LinkedHashMap<>(); + private final Map<String, Cached> latest = new LinkedHashMap<>(); + private final Deque<Cached> history = new ArrayDeque<>(); + private final List<EventNotifier> ownedNotifiers = new ArrayList<>(); + private final List<ObjectName> mbeans = new ArrayList<>(); + private final AtomicLong sourceErrors = new AtomicLong(); + private final AtomicLong storeErrors = new AtomicLong(); + private final AtomicLong dropped = new AtomicLong(); + private final AtomicLong evicted = new AtomicLong(); + private final AtomicLong managementErrors = new AtomicLong(); + private volatile ExecutorService writer; + private ScheduledExecutorService sampler; + private volatile String lastSample; + private final Map<String, Map<String, Object>> sourceHealth = new ConcurrentHashMap<>(); + private final Map<String, Map<String, Object>> storeHealth = new ConcurrentHashMap<>(); + private int omittedSources; + private int omittedStores; + private volatile String recovery = "memory_only"; + + private record Cached(String owner, SemanticMonitoringSnapshot snapshot, boolean recovered) { + } + + private SemanticMonitoringService(CamelContext context) { + setCamelContext(context); + } + + public static SemanticMonitoringService get(CamelContext context) { + synchronized (context) { + var extension = context.getCamelContextExtension(); + var service = extension.getContextPlugin(SemanticMonitoringService.class); + if (service == null) { + service = new SemanticMonitoringService(context); + extension.addContextPlugin(SemanticMonitoringService.class, service); + try { + context.addService(service, true, true); + } catch (Exception failure) { + throw new IllegalStateException("Cannot start semantic monitoring", failure); + } + } + return service; + } + } + + @Override + protected void doStart() throws Exception { + CamelContext context = getCamelContext(); + sources.clear(); + stores.clear(); + sourceHealth.clear(); + storeHealth.clear(); + var availableSources = context.getRegistry().findByTypeWithName(SemanticMetricSource.class); + var availableStores = context.getRegistry().findByTypeWithName(SemanticMonitoringStore.class); + omittedSources = Math.max(0, availableSources.size() - MAX_SOURCES); + omittedStores = Math.max(0, availableStores.size() - 8); + availableSources.entrySet().stream() + .sorted(Map.Entry.comparingByKey()).limit(MAX_SOURCES) + .forEach(entry -> sources.put(entry.getKey(), entry.getValue())); + availableStores.entrySet().stream() + .sorted(Map.Entry.comparingByKey()).limit(8).forEach(entry -> stores.put(entry.getKey(), entry.getValue())); + if (!stores.isEmpty()) { + writer = new ThreadPoolBuilder(context).poolSize(1).maxPoolSize(1).maxQueueSize(256) + .rejectedPolicy(ThreadPoolRejectedPolicy.Abort).build(this, "SemanticMonitoringStore"); + } + for (var source : sources.values()) { + if (source instanceof EventNotifier notifier + && !context.getManagementStrategy().getEventNotifiers().contains(notifier)) { + if (notifier instanceof CamelContextAware aware) { + aware.setCamelContext(context); + } + try { + ServiceHelper.startService(notifier); + context.getManagementStrategy().addEventNotifier(notifier); + ownedNotifiers.add(notifier); + } catch (Exception | AssertionError | LinkageError failure) { + sourceErrors.incrementAndGet(); + } + } + } + context.getManagementStrategy().addEventNotifier(this); + for (var entry : stores.entrySet()) { + try { + ServiceHelper.startService(entry.getValue()); + storeHealth.put(entry.getKey(), Map.of("state", "started", "persistent", entry.getValue().isPersistent())); + } catch (Exception | AssertionError | LinkageError failure) { + storeErrors.incrementAndGet(); + storeHealth.put(entry.getKey(), Map.of("state", "start_failed", "persistent", "unknown")); + } + } + recovery = stores.isEmpty() ? "memory_only" : "loading"; + if (!stores.isEmpty()) { + writer.execute(this::recover); + } + sample(); + if (!sources.isEmpty()) { + sampler = context.getExecutorServiceManager().newSingleThreadScheduledExecutor(this, "SemanticMetricSnapshots"); + sampler.scheduleWithFixedDelay(this::sample, 1, 1, TimeUnit.SECONDS); + } + context.addStartupListener((started, alreadyStarted) -> { + manage("all", null); + for (String source : sources.keySet()) { + manage("Source(" + source + ")", source); + } + }); + } + + private void manage(String name, String source) { + var agent = getCamelContext().getManagementStrategy().getManagementAgent(); + if (agent == null) { + return; + } + try { + ObjectName objectName = new ObjectName( + agent.getMBeanObjectDomainName() + ":context=" + + ObjectName.quote(getCamelContext().getManagementName()) + + ",type=semantic,name=" + ObjectName.quote(name)); + if (mbeans.contains(objectName)) { + return; + } + SnapshotMBean facade = () -> new JsonObject(snapshot(source, false)).toJson(); + agent.register(new StandardMBean(facade, SnapshotMBean.class), objectName); + mbeans.add(objectName); + } catch (Exception failure) { + managementErrors.incrementAndGet(); + } + } + + /** JMX deliberately exposes attributes only: no reset, detector, reference or lifecycle operations. */ + public interface SnapshotMBean { + String getSnapshot(); + } + + private void recover() { + boolean failed = false; + for (var entry : stores.entrySet()) { + try { + if ("start_failed".equals(storeHealth.get(entry.getKey()).get("state"))) { + failed = true; + continue; + } + List<SemanticMonitoringSnapshot> saved = entry.getValue().readRecent(CAPACITY); + if (saved.size() > CAPACITY) { + throw new IllegalArgumentException("Monitoring recovery limit"); + } + for (int i = saved.size() - 1; i >= 0; i--) { + accept(saved.get(i).getSource(), saved.get(i), false, false); + } + storeState(entry.getKey(), "recovered", "lastRead", Instant.now().toString()); + } catch (Exception | AssertionError | LinkageError failure) { + storeErrors.incrementAndGet(); + storeState(entry.getKey(), "read_failed", "lastFailure", Instant.now().toString()); + failed = true; + } + } + recovery = failed ? "partial_or_failed" : "bounded_history_loaded"; + } + + private void sample() { + for (var entry : sources.entrySet()) { + try { + List<SemanticMonitoringSnapshot> values = entry.getValue().snapshots(); + if (values.size() > 64) { + throw new IllegalArgumentException("Metric table limit"); + } + for (var snapshot : values) { + accept(entry.getKey(), snapshot, true, true); + } + lastSample = Instant.now().toString(); + sourceState(entry.getKey(), values.isEmpty() ? "no_snapshots" : "available", "lastSuccess", lastSample); + } catch (Exception | AssertionError | LinkageError failure) { + sourceErrors.incrementAndGet(); + sourceState(entry.getKey(), "sample_failed", "lastFailure", Instant.now().toString()); + } + } + } + + private void sourceState(String name, String state, String field, String time) { + updateHealth(sourceHealth, name, state, field, time); + } + + private void storeState(String name, String state, String field, String time) { + updateHealth(storeHealth, name, state, field, time); + } + + private static void updateHealth( + Map<String, Map<String, Object>> health, String name, String state, String field, String time) { + health.compute(name, (key, previous) -> { + Map<String, Object> value = new LinkedHashMap<>(previous == null ? Map.of() : previous); + value.put("state", state); + value.put(field, time); + return Map.copyOf(value); + }); + } + + @Override + public boolean isEnabled(CamelEvent event) { + return event instanceof SemanticMonitoringEvent; + } + + @Override + public void notify(CamelEvent event) { + if (event instanceof SemanticMonitoringEvent monitoring) { + accept(monitoring.getSnapshot().getSource(), monitoring.getSnapshot(), true, true); + } + } + + private void accept(String owner, SemanticMonitoringSnapshot snapshot, boolean persist, boolean replace) { + synchronized (this) { + String key = snapshot.getSource() + "\u0000" + snapshot.getId(); + Cached previous = latest.get(key); + if (previous != null && previous.snapshot.toMap().equals(snapshot.toMap())) { + return; + } + if (previous == null && latest.size() == CAPACITY) { + latest.remove(latest.keySet().iterator().next()); + evicted.incrementAndGet(); + } + Cached captured = new Cached(owner, snapshot, !replace); + if (previous == null || replace || previous.recovered && newer(snapshot, previous.snapshot)) { + latest.put(key, captured); + } + if (history.size() == CAPACITY) { + history.removeLast(); + evicted.incrementAndGet(); + } + history.addFirst(captured); Review Comment: Changed in 8321d79cecc6. Latest snapshots still refresh every second, but changed sampled history/store writes default to once per minute per series, with an immediate first sample. A `SemanticMonitoringConfiguration` bean controls `historyInterval` (1..86400 seconds) and can be updated at runtime. Explicit monitoring events bypass that interval; unchanged snapshots add no history. The shared 100-entry limit remains explicit, and responses report retained count and capture range. `SemanticMonitoringSamplingTest` covers the cadence, runtime updates and explicit-event behavior. _Codex (AI-generated) on behalf of @luigidemasi._ ########## components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticMonitoringService.java: ########## @@ -0,0 +1,392 @@ +/* + * 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.internal; + +import java.time.Instant; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import javax.management.ObjectName; +import javax.management.StandardMBean; + +import org.apache.camel.CamelContext; +import org.apache.camel.CamelContextAware; +import org.apache.camel.builder.ThreadPoolBuilder; +import org.apache.camel.semantic.SemanticMetricSource; +import org.apache.camel.semantic.SemanticMonitoringEvent; +import org.apache.camel.semantic.SemanticMonitoringSnapshot; +import org.apache.camel.semantic.SemanticMonitoringStore; +import org.apache.camel.spi.CamelEvent; +import org.apache.camel.spi.EventNotifier; +import org.apache.camel.support.EventNotifierSupport; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy; +import org.apache.camel.util.json.JsonObject; + +/** Bounded read-only projection of supplied measurements, shared by all surfaces. */ +public final class SemanticMonitoringService extends EventNotifierSupport { + private static final int CAPACITY = 100; + private static final int MAX_SOURCES = 32; + private final Map<String, SemanticMetricSource> sources = new LinkedHashMap<>(); + private final Map<String, SemanticMonitoringStore> stores = new LinkedHashMap<>(); + private final Map<String, Cached> latest = new LinkedHashMap<>(); + private final Deque<Cached> history = new ArrayDeque<>(); + private final List<EventNotifier> ownedNotifiers = new ArrayList<>(); + private final List<ObjectName> mbeans = new ArrayList<>(); + private final AtomicLong sourceErrors = new AtomicLong(); + private final AtomicLong storeErrors = new AtomicLong(); + private final AtomicLong dropped = new AtomicLong(); + private final AtomicLong evicted = new AtomicLong(); + private final AtomicLong managementErrors = new AtomicLong(); + private volatile ExecutorService writer; + private ScheduledExecutorService sampler; + private volatile String lastSample; + private final Map<String, Map<String, Object>> sourceHealth = new ConcurrentHashMap<>(); + private final Map<String, Map<String, Object>> storeHealth = new ConcurrentHashMap<>(); + private int omittedSources; + private int omittedStores; + private volatile String recovery = "memory_only"; + + private record Cached(String owner, SemanticMonitoringSnapshot snapshot, boolean recovered) { + } + + private SemanticMonitoringService(CamelContext context) { + setCamelContext(context); + } + + public static SemanticMonitoringService get(CamelContext context) { + synchronized (context) { Review Comment: Fixed in 8321d79cecc6: service creation now synchronizes on a private static `CREATION_LOCK`, matching the audit service pattern, rather than locking the public `CamelContext`. _Codex (AI-generated) on behalf of @luigidemasi._ ########## docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc: ########## @@ -13,6 +13,39 @@ See the xref:camel-upgrade-recipes-tool.adoc[documentation] page for details. == Upgrading Camel 4.23 to 4.24 +=== camel-semantic - explicit experts Review Comment: Thanks for confirming this in your follow-up. The explicit-expert change remains in this PR, with the preview removals documented in the upgrade guide. CAMEL-25618 now describes that scope explicitly, and the branch has been rebased and organized into four commits for review. _Codex (AI-generated) on behalf of @luigidemasi._ ########## docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc: ########## @@ -13,6 +13,39 @@ See the xref:camel-upgrade-recipes-tool.adoc[documentation] page for details. == Upgrading Camel 4.23 to 4.24 +=== camel-semantic - explicit experts + +Every evaluation in the preview semantic API now requires an explicit `expert` registry bean name. +Move block-level `expert` values onto each evaluation in YAML and XML; in Java, move `.expert(...)` +from `SemanticEvaluationsBuilder` to each `SemanticEvaluationBuilder`. + +The `camel.language.semantic.default-expert` and `camel.language.semantic.adapter` options, +`SemanticLanguage.setDefaultExpert` / `setAdapter`, automatic selection of a sole expert, and +`META-INF/services/org/apache/camel/semantic-adapter` discovery are removed. Register an expert +bean explicitly, including its initialization and destruction methods where needed, and reference +it from each evaluation. Expert references accept property placeholders, but do not fall back to +class loading. The default state selector is unchanged. + +The semantic metadata response no longer includes `defaultExpert`, `defaultError` or +`defaultErrorCode`. Operation inspection and completion require an explicit expert reference. + +=== camel-semantic - evaluation-only auditing + +The preview semantic audit API no longer records route actions. `SemanticAuditDecision`, +`SemanticAudit.get(context).decision(...)` and `SemanticObserver.decision(...)` have been removed. Review Comment: CAMEL-25618 now explicitly specifies outright removal of the preview route-action audit APIs, matching the implementation and upgrade guide. Its fix version is also set to 4.24.0. I rechecked both fields in Jira. _Codex (AI-generated) on behalf of @luigidemasi._ -- 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]
