gnodet-bot commented on code in PR #29379: URL: https://github.com/apache/camel/pull/29379#discussion_r4502476052
########## core/camel-console/src/main/java/org/apache/camel/impl/console/TimelineDevConsole.java: ########## @@ -0,0 +1,395 @@ +/* + * 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.impl.console; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; +import java.util.function.Predicate; + +import org.apache.camel.Exchange; +import org.apache.camel.ExchangeExtension; +import org.apache.camel.ExchangePropertyKey; +import org.apache.camel.NonManagedService; +import org.apache.camel.Route; +import org.apache.camel.api.management.ManagedCamelContext; +import org.apache.camel.api.management.mbean.ManagedRouteMBean; +import org.apache.camel.spi.CamelEvent; +import org.apache.camel.spi.Configurer; +import org.apache.camel.spi.Metadata; +import org.apache.camel.spi.annotations.DevConsole; +import org.apache.camel.support.EventNotifierSupport; +import org.apache.camel.support.ExchangeHelper; +import org.apache.camel.support.PatternHelper; +import org.apache.camel.support.console.AbstractDevConsole; +import org.apache.camel.util.json.JsonRecordSupport; + +/** + * The timeline of the integration: per second, for the last minutes, what each route did, the errors by exception, and + * a few exchanges of each second with the calls they made. It records from the start of the console, so the history is + * there when a tool first looks. + */ +@DevConsole(name = "timeline", displayName = "Camel Timeline", + description = "Per second, what each route did, the errors, and the failed and slowest exchanges") +@Configurer(extended = true) +public class TimelineDevConsole extends AbstractDevConsole { + + /** The exchange property camel-chaos marks a faulted exchange with, with the id of the rule. */ + static final String CHAOS_RULE = "CamelChaosRule"; + + public record Send( + @Metadata(description = "The endpoint URI") String endpointUri, + @Metadata(description = "Elapsed time of the call in milliseconds") long elapsed) { + } + + public record RouteSeries( + @Metadata(description = "The route ID") String routeId, + @Metadata(description = "The from endpoint URI (only present when known)") String from, + @Metadata(description = "Exchanges completed per second, not counting the recovered and handled ones") List<Long> completed, + @Metadata(description = "Exchanges per second that completed after a redelivery in this route") List<Long> recovered, + @Metadata(description = "Exchanges per second whose failure the error handler handled") List<Long> handled, + @Metadata(description = "Exchanges failed per second") List<Long> failed, + @Metadata(description = "Exchanges in flight at the end of each second") List<Long> inflight, + @Metadata(description = "Mean processing time per second in milliseconds") List<Long> meanMillis, + @Metadata(description = "p95 processing time per second in milliseconds, of the exchanges that came into this route") List<Long> p95Millis, + @Metadata(description = "Max processing time per second in milliseconds, of the exchanges that came into this route") List<Long> maxMillis, + @Metadata(description = "Redelivery attempts per second in this route") List<Long> redeliveryAttempts) { + } + + public record ErrorSeries( + @Metadata(description = "The route ID where the exchange failed") String routeId, + @Metadata(description = "The exception class name") String exception, + @Metadata(description = "The HTTP status code of the exception (only present when known)") Integer statusCode, + @Metadata(description = "The camel-chaos rule that made the error (only present for a chaos fault)") String chaosRule, + @Metadata(description = "Errors per second") List<Long> counts) { + } + + public record TimelineExchange( + @Metadata(description = "The exchange ID") String exchangeId, + @Metadata(description = "The route ID the exchange came in") String routeId, + @Metadata(description = "The from endpoint URI (only present when known)") String fromEndpointUri, + @Metadata(description = "Epoch time in milliseconds the exchange was created") long timestamp, + @Metadata(description = "Elapsed time in milliseconds") long elapsed, + @Metadata(description = "Whether the exchange failed") boolean failed, + @Metadata(description = "Whether the error handler handled a failure of the exchange") boolean handled, + @Metadata(description = "Whether the exchange was redelivered") boolean redelivered, + @Metadata(description = "The exception message (only present when the exchange failed, or when it recovered: the failure it was redelivered for)") String exception, + @Metadata(description = "The camel-chaos rule that faulted the exchange (only present when one did)") String chaosRule, + @Metadata(description = "The remote endpoints the exchange was sent to (only present when any)") List<Send> endpointSends) { + } + + public record Response( + @Metadata(description = "Epoch second of the last second (the oldest is end - seconds + 1)") long end, + @Metadata(description = "The number of seconds of each series, oldest first") int seconds, + @Metadata(description = "Exchanges done per second, in the routes they came in") List<Long> exchangesDone, + @Metadata(description = "The routes") List<RouteSeries> routes, + @Metadata(description = "The errors by route and exception (only present when there are any)") List<ErrorSeries> errors, + @Metadata(description = "Per second the failed, handled and recovered exchanges (up to 10) and the 3 slowest, oldest first (only present when asked for)") List<TimelineExchange> exchanges) { + } + + @Metadata(label = "query", description = "The number of seconds, up to the recorded seconds (the default)", + javaType = "java.lang.Integer") + public static final String SECONDS = "seconds"; + + @Metadata(label = "query", description = "Filter by route id (pattern)", javaType = "java.lang.String") + public static final String ROUTE_ID = "routeId"; + + @Metadata(label = "query", description = "Whether to include the kept exchanges", javaType = "java.lang.Boolean", + defaultValue = "true") + public static final String EXCHANGES = "exchanges"; + + @Metadata(defaultValue = "300", description = "The number of seconds to record (between 60 and 3600)") + private int seconds = 300; + + private volatile TimelineRecorder recorder; + private final TimelineEventNotifier listener = new TimelineEventNotifier(); + private ScheduledExecutorService executor; + + public TimelineDevConsole() { + super("camel", "timeline", "Camel Timeline", + "Per second, what each route did, the errors, and the failed and slowest exchanges"); + } + + public int getSeconds() { + return seconds; + } + + public void setSeconds(int seconds) { + if (seconds < 60 || seconds > 3600) { + throw new IllegalArgumentException("The seconds must be between 60 and 3600, was: " + seconds); + } + if (recorder != null && recorder.getSeconds() != seconds) { + // Camel Main configures the console after the registry has started it: start the recording again + recorder = new TimelineRecorder(seconds); + } + this.seconds = seconds; + } + + TimelineRecorder getRecorder() { + return recorder; + } + + @Override + protected void doInit() throws Exception { + recorder = new TimelineRecorder(seconds); + } + + @Override + protected void doStart() throws Exception { + getCamelContext().getManagementStrategy().addEventNotifier(listener); + executor = getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this, "Timeline"); + // the counters as they are now, so the first second counts from here + sample(); + long now = System.currentTimeMillis(); + // just after each whole second + executor.scheduleAtFixedRate(this::sample, 1000 - now % 1000 + 5, 1000, TimeUnit.MILLISECONDS); + } + + @Override + protected void doStop() throws Exception { + getCamelContext().getManagementStrategy().removeEventNotifier(listener); + if (executor != null) { + getCamelContext().getExecutorServiceManager().shutdown(executor); + executor = null; + } + } + + void sample() { + recorder.sample(System.currentTimeMillis(), counters()); + } + + private Map<String, TimelineRecorder.Counters> counters() { + Map<String, TimelineRecorder.Counters> answer = new HashMap<>(); + ManagedCamelContext mcc = getCamelContext().getCamelContextExtension().getContextPlugin(ManagedCamelContext.class); + if (mcc != null) { + for (Route route : getCamelContext().getRoutes()) { + ManagedRouteMBean mr = mcc.getManagedRoute(route.getRouteId()); + if (mr != null) { + answer.put(route.getRouteId(), new TimelineRecorder.Counters( + mr.getExchangesCompleted(), mr.getExchangesFailed(), mr.getFailuresHandled(), + mr.getExchangesInflight(), mr.getTotalProcessingTime())); + } + } + } + return answer; + } + + @Override + protected String doCallText(Map<String, Object> options) { + Response r = response(options, false); + StringBuilder sb = new StringBuilder(); + sb.append(String.format("Last %d seconds:%n", r.seconds())); + for (RouteSeries rs : r.routes()) { + sb.append(String.format(" %s: %d completed, %d recovered, %d handled, %d failed, %d redelivery attempts%n", + rs.routeId(), sum(rs.completed()), sum(rs.recovered()), sum(rs.handled()), sum(rs.failed()), + sum(rs.redeliveryAttempts()))); + } + if (r.errors() != null) { + sb.append("Errors:\n"); + for (ErrorSeries es : r.errors()) { + sb.append(String.format(" %s in %s: %d%s%n", es.exception(), es.routeId(), sum(es.counts()), + es.chaosRule() != null ? " (chaos rule " + es.chaosRule() + ")" : "")); + } + } + return sb.toString(); + } + + @Override + protected Map<String, Object> doCallJson(Map<String, Object> options) { + return JsonRecordSupport.toJsonObject(response(options, optionBoolean(options, EXCHANGES, true))); + } + + private Response response(Map<String, Object> options, boolean exchanges) { + TimelineRecorder rec = recorder; + int count = Math.min(optionInt(options, SECONDS, rec.getSeconds()), rec.getSeconds()); + count = Math.max(1, count); + String filter = optionString(options, ROUTE_ID); + Predicate<String> accept = id -> filter == null || PatternHelper.matchPattern(id, filter); + long end = rec.endSecond(System.currentTimeMillis()); + List<TimelineRecorder.Second> secs; + List<RouteSeries> routes = new ArrayList<>(); + List<ErrorSeries> errors = new ArrayList<>(); + List<TimelineExchange> kept = new ArrayList<>(); + List<Long> done = new ArrayList<>(count); + synchronized (rec) { + secs = rec.seconds(end, count); + for (TimelineRecorder.Second s : secs) { + done.add(s != null ? s.done : 0L); + } + for (String routeId : TimelineRecorder.routes(secs, accept)) { + Route route = getCamelContext().getRoute(routeId); Review Comment: ⚠️ `getCamelContext().getRoute(routeId)` acquires `routesLock` (via `AbstractCamelContext.getRoutes()`) while holding `synchronized(rec)`. This blocks all concurrent exchange events from updating the recorder for the full duration of the route URI lookup loop. No ABBA deadlock currently (the event path in `done()` releases `routesLock` before entering `synchronized(rec)`), but the recorder is effectively paused for the duration of a response query. Move the route URI lookup outside the `synchronized(rec)` block: snapshot the route IDs and series data inside the lock, then resolve consumer URIs outside it. -- 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]
