luigidemasi commented on code in PR #27791: URL: https://github.com/apache/camel/pull/27791#discussion_r4270312714
########## 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: Follow-up to my earlier answer: 673b410779fd supersedes the history-interval approach. Camel now keeps a bounded cache of current values only (up to 100 series), with registered sources sampled every second. There are no monitoring history entries, store appends, recovery or history-interval configuration. Successful complete samples retire missing series; failed or invalid samples retain the last valid values and report collection health. Metric history and retention are left to external monitoring. [CAMEL-25648](https://issues.apache.org/jira/browse/CAMEL-25648) tracks optional Micrometer instrumentation and Prometheus export, rather than an implementation of the removed snapshot-store SPI. `SemanticMonitoringSamplingTest` covers replacement, retired series, failure preservation, malformed samples and restart behavior. _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]
