davsclaus commented on code in PR #27791: URL: https://github.com/apache/camel/pull/27791#discussion_r4266955901
########## 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: CAMEL-25618 does not cover this section or the commit behind it (removing `default-expert`, `adapter`, sole-expert selection, `semantic-adapter` discovery and block-level `expert`). It is a separate user-facing breaking change, and it accounts for most of the PR's deletions. Please give it its own JIRA and PR. ########## 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: Same as `SemanticMonitoringService`: this notifier only wants `SemanticEvent`, but `EventNotifierSupport` defaults to receiving exchange events. That turns on exchange event creation for the whole context. Call `setupIgnore(true)` in the constructor. ########## 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: `EventNotifierSupport`'s constructor calls `setupIgnore(false)`, so this notifier subscribes to exchange events. `SemanticAuditService.doStart()` always creates this service, and `SemanticReloadPlugin` activates the audit service on every context start. `DefaultManagementStrategy.addEventNotifier` therefore sets `eventNotificationApplicable=true`. After that, every exchange in every route builds and dispatches Created / Sending / Sent / Completed / Step events that `isEnabled()` then rejects. This happens even with audit off and no sources. ```java private SemanticMonitoringService(CamelContext context) { setCamelContext(context); setupIgnore(true); // only SemanticMonitoringEvent (an AiEvent) is wanted } ``` AI events are not part of `setupIgnore`, so `notifyAiEvent` still delivers them. ########## 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: Nit: this synchronizes on the `CamelContext` object, while `SemanticAuditService.get` uses a private `CREATION_LOCK`. Please use the same pattern here, so the code does not lock a public object that other code may also synchronize on. ########## 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: This counts every started notifier except the monitoring service, including disabled ones and ones with `ignoreAiEvents=true` (for example `MicrometerExchangeEventNotifier`). With such a notifier present, the audit-disabled `noop` path is skipped. Every evaluation then builds and validates a `SemanticEvent` that nobody receives. Please match what `EventHelper.notifyAiEvent` filters on: ```java .anyMatch(n -> !(n instanceof SemanticMonitoringService) && !n.isDisabled() && !n.isIgnoreAiEvents()) ``` ########## 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: Question: `SemanticResultMetrics` produces a different snapshot after every evaluation (counts and `to` change). The 1-second sampler therefore adds a history entry, plus a store append, for each active source every second. `CAPACITY` is 100 and shared by all sources, so with a few busy sources `--history` covers only tens of seconds. Is that intended? An alternative: sampled snapshots update `latest` only, and history and stores keep `SemanticMonitoringEvent` outputs or use a coarser interval. ########## 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: The JIRA's migration plan says to *deprecate* `SemanticAuditDecision` and to remove it later through the normal compatibility process. This PR removes it, together with `SemanticAudit` and `SemanticObserver.decision`. For a Preview API, removal is acceptable to me, but please update the JIRA so the two agree. ########## 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: `timed`, `durationSum`, `durationMinimum` and `durationMaximum` are not reset. After a context stop/start, the new epoch starts its counts at zero but reports mean, min and max durations that include the previous epoch. ```java timed = 0; durationSum = BigDecimal.ZERO; durationMinimum = durationMaximum = null; ``` A restart test would catch this. -- 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]
