This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26678-chaos-live-state in repository https://gitbox.apache.org/repos/asf/camel.git
commit bd0d293897f483e3207cc89cc1289b3d2a975876 Author: Claus Ibsen <[email protected]> AuthorDate: Fri Nov 6 11:33:37 2026 +0100 CAMEL-26678: camel-chaos - live state of the rules Each rule reports its observed rate next to its own, the next match of an every:N rule, when it last fired and its faults per second over the last minute, and the effect of its faults on the exchanges (completed, recovered by redelivery, handled by the error handler, failed, in flight). The report keeps the last 20 faults, and a faulted exchange gets the exchange property CamelChaosRule. A lifecycle rule counts the exchanges sent to its route while the route is down. Shown by the dev console, camel get chaos, the MCP tools. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/catalog/dev-consoles/chaos.json | 50 +++- .../org/apache/camel/catalog/docs/chaos.adoc | 44 +++- .../org/apache/camel/dev-console/chaos.json | 50 +++- components/camel-chaos/src/main/docs/chaos.adoc | 44 +++- .../camel/component/chaos/ChaosDevConsole.java | 11 +- .../camel/component/chaos/ChaosEventNotifier.java | 11 +- .../camel/component/chaos/ChaosExchangeState.java | 100 ++++++++ .../apache/camel/component/chaos/ChaosFaults.java | 19 +- .../camel/component/chaos/ChaosLifecycle.java | 14 ++ .../apache/camel/component/chaos/ChaosManager.java | 24 +- .../camel/component/chaos/ChaosPollStrategy.java | 12 +- .../camel/component/chaos/ChaosRecentFaults.java | 60 +++++ .../apache/camel/component/chaos/ChaosReport.java | 204 +++++++++++++++- .../component/chaos/ChaosRoutePolicyFactory.java | 2 +- .../apache/camel/component/chaos/ChaosRule.java | 141 ++++++++++- .../camel/component/chaos/ChaosSendHook.java | 24 ++ .../camel/component/chaos/ChaosStepHook.java | 2 +- .../apache/camel/component/chaos/ChaosTargets.java | 2 +- .../camel/component/chaos/ChaosLifecycleTest.java | 34 ++- .../camel/component/chaos/ChaosLiveStateTest.java | 271 +++++++++++++++++++++ .../camel/component/chaos/ChaosPollFaultsTest.java | 8 + .../camel/component/chaos/ChaosRuleTest.java | 20 ++ .../modules/ROOT/pages/camel-jbang-mcp.adoc | 10 +- .../jbang-commands/camel-jbang-get-chaos.adoc | 2 +- .../META-INF/camel-jbang-commands-metadata.json | 2 +- .../dsl/jbang/core/commands/ai/ToolRegistry.java | 12 +- .../dsl/jbang/core/commands/process/ListChaos.java | 85 ++++++- .../jbang/core/commands/process/ListChaosTest.java | 78 ++++++ .../src/test/resources/golden/cli-help.txt | 10 +- .../dsl/jbang/core/commands/mcp/RuntimeTools.java | 12 +- 30 files changed, 1304 insertions(+), 54 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/chaos.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/chaos.json index ba6762479eed..faa3a563ff97 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/chaos.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/chaos.json @@ -66,10 +66,55 @@ "type": "integer", "description": "The faults that did nothing, as the rule fired at a message or an endpoint it cannot change" }, + "skippedReasons": { + "type": "object", + "additionalProperties": { + "type": "integer" + }, + "description": "Why the rule did nothing when it was skipped: the count per reason (only when it was skipped)" + }, + "lostAtDlq": { + "type": "integer", + "description": "The messages lost because the rule failed or dropped the send of the dead letter channel, which then handled them (only for a rule with at=dlq)" + }, "endsIn": { "type": "integer", "description": "The milliseconds until an active rule ends" }, + "rate": { + "type": "number", + "description": "The share of the matches the rule fires at: rate=, 1\/N for rate=every:N, else 1 (not for a lifecycle or stall rule)" + }, + "observedRate": { + "type": "number", + "description": "The share of the matches the rule fired at so far, to compare with rate (once it matched; not for a lifecycle or stall rule)" + }, + "nextFireIn": { + "type": "integer", + "description": "The matches until a rule with rate=every:N fires again (only for such a rule while it is active)" + }, + "lastFiredAgo": { + "type": "integer", + "description": "The milliseconds since the rule last fired (only when it fired)" + }, + "faultsLastMinute": { + "type": "integer", + "description": "The faults the rule fired in the last minute" + }, + "faultsPerSecond": { + "type": "array", + "items": { + "type": "integer" + }, + "description": "The faults the rule fired per second over the last minute, 60 counts from the oldest second to the current one (only when it fired in the last minute)" + }, + "effect": { + "type": "object", + "additionalProperties": { + "type": "integer" + }, + "description": "The effect of the faults on the exchanges they hit (for a lifecycle or stall rule: the exchanges sent to its route while it was down), each counted once when it is done: faulted, completed, recovered (by a redelivery of the error handler), handled (by the error handler, such as the dead letter channel), failed and inFlight (only when the rule faulted an exchange)" + }, "expectation": { "type": "object", "additionalProperties": true, @@ -84,7 +129,8 @@ "required": [ "version", "matches", - "skipped" + "skipped", + "faultsLastMinute" ] }, "description": "The current rules (not the removed or replaced ones; only when chaos is on)" @@ -92,7 +138,7 @@ "report": { "type": "object", "additionalProperties": true, - "description": "The report: verdict, final, seed, seedGiven, lost, reasons, all rules with the removed and replaced ones, and the counts per route (only when chaos is on)" + "description": "The report: verdict, final, seed, seedGiven, lost, reasons, all rules with the removed and replaced ones, the counts per route, and the recent faults (the last 20, the latest first: timestamp, ago, rule, version, fault, hook, route, endpoint or step, exchangeId) (only when chaos is on)" } }, "required": [ diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc index 6f83ea053c0f..c32902822915 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc @@ -126,7 +126,9 @@ integration; `chaos-clear` without one acts on every running integration that ha `chaos-clear` removes stay in the report; with `--reset` the report starts again, for a new experiment: no rules, route counts or verdict, as for an application that just started (with the same seed, and the ids of new rules going on from the last one). An id that does not exist is "not found" (exit code 3). `camel get chaos` lists the current rules with their id, version, state, -matches and faults, and `--report` shows the report (also as JSON with `--json`). Their `NOTE` column says what a +matches and faults, the observed rate (`RATE`), when the rule last fired (`LAST`) and the effect of its faults +(`EFFECT`), and `--report` shows the report, with the recent faults (also as JSON with `--json`); see +<<_the_live_state_of_the_rules,The live state of the rules>>. Their `NOTE` column says what a lifecycle rule does to its route (`route poller down, back in 12s`) and why a rule did nothing when it was skipped. Without a rule to show, `camel get chaos` says so on stderr (no rules, chaos off, or no integration with camel-chaos). @@ -505,6 +507,46 @@ give their reason only. In JSON an expectation has its `kind`, `op`, `count` and the `value` counted so far (and `completed`, the same value, for an expectation of completed exchanges). +=== The live state of the rules + +While the rules run, the report tells how each one is doing, and what its faults do: + +* the observed rate next to the rate of the rule: the share of the matches it fired at (`rate 4.8% of 5%`, and +`rate 20% of 1 in 5` for `rate=every:5`), and for `rate=every:N` the matches until it fires again; +* when it last fired, and its faults per second over the last minute; +* the effect of its faults on the exchanges they hit, each exchange counted once when it is done: `completed` (without a +redelivery after the fault, such as after a `delay`), `recovered` by a redelivery of the error handler, `handled` by +the error handler (such as the dead letter channel), `failed`, and the others still in flight. Only an exchange with a +unit of work counts (one that a route processes), as that tells when it is done. For a lifecycle rule (or `stall`), the +exchanges sent to the endpoint of its route while the route is down count, such as `direct:inventory` or +`seda:shipping` (not an HTTP call to a `platform-http` route, which is another endpoint): a 3 second stop of a route +called with `direct:inventory?block=false` that sends 5 orders to the dead letter channel shows as +`5 handled by the error handler`, as the redelivery is too short to ride it out; +* the last 20 faults of all rules, the latest first: when, the rule, the fault, the hook, the route, the endpoint +(without its query) or the step, and the exchange id. + +An exchange that a rule faulted (or that was sent to the route of a lifecycle rule while it was down) gets the exchange +property `CamelChaosRule` with the id of the rule (of the last one, when more rules did), so the tracer, the message history and a route of the dead letter channel can tell why +it is there: + +[source,text] +---- + r1 active: endpoint=kafka:orders* error rate=0.05 until=10m + matches 1250, fired 61 (send 61), skipped 0, ends in 7m12s + rate 4.9% of 5%, last fired 2s ago, 6 faults in the last minute + effect on 58 exchanges: 55 recovered by redelivery, 2 handled by the error handler, 1 in flight + recent faults: + 2s ago r1 error ChaosException at send kafka://orders, route orders, exchange 4F2A... +---- + +In JSON a rule has `rate` (the one of the rule: 1 without `rate=`) and `observedRate` (once it matched), `nextFireIn` +(the matches until a `rate=every:N` rule fires again), `lastFiredAgo` (milliseconds), `faultsLastMinute`, +`faultsPerSecond` (60 counts, from the oldest second to the current one, when it fired in the last minute) and `effect` +(`faulted`, `completed`, `recovered`, `handled`, `failed` and `inFlight`, once it faulted an exchange); a lifecycle rule +has no rate. The report has `recentFaults`, each with `timestamp`, `ago`, `rule`, `version`, `fault` (such as +`error status 503` or `delay 2s`), `hook` (none for a lifecycle rule), `route`, `endpoint` or `step`, and `exchangeId` +(none at a poll). `ChaosManager.reset()` (`chaos-clear --reset`) clears the recent faults too. + == Time and the seed The `delay` fault waits through `TimeSource` of camel-util, and the end of a rule, the time of an expectation and the diff --git a/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/dev-console/chaos.json b/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/dev-console/chaos.json index ba6762479eed..faa3a563ff97 100644 --- a/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/dev-console/chaos.json +++ b/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/dev-console/chaos.json @@ -66,10 +66,55 @@ "type": "integer", "description": "The faults that did nothing, as the rule fired at a message or an endpoint it cannot change" }, + "skippedReasons": { + "type": "object", + "additionalProperties": { + "type": "integer" + }, + "description": "Why the rule did nothing when it was skipped: the count per reason (only when it was skipped)" + }, + "lostAtDlq": { + "type": "integer", + "description": "The messages lost because the rule failed or dropped the send of the dead letter channel, which then handled them (only for a rule with at=dlq)" + }, "endsIn": { "type": "integer", "description": "The milliseconds until an active rule ends" }, + "rate": { + "type": "number", + "description": "The share of the matches the rule fires at: rate=, 1\/N for rate=every:N, else 1 (not for a lifecycle or stall rule)" + }, + "observedRate": { + "type": "number", + "description": "The share of the matches the rule fired at so far, to compare with rate (once it matched; not for a lifecycle or stall rule)" + }, + "nextFireIn": { + "type": "integer", + "description": "The matches until a rule with rate=every:N fires again (only for such a rule while it is active)" + }, + "lastFiredAgo": { + "type": "integer", + "description": "The milliseconds since the rule last fired (only when it fired)" + }, + "faultsLastMinute": { + "type": "integer", + "description": "The faults the rule fired in the last minute" + }, + "faultsPerSecond": { + "type": "array", + "items": { + "type": "integer" + }, + "description": "The faults the rule fired per second over the last minute, 60 counts from the oldest second to the current one (only when it fired in the last minute)" + }, + "effect": { + "type": "object", + "additionalProperties": { + "type": "integer" + }, + "description": "The effect of the faults on the exchanges they hit (for a lifecycle or stall rule: the exchanges sent to its route while it was down), each counted once when it is done: faulted, completed, recovered (by a redelivery of the error handler), handled (by the error handler, such as the dead letter channel), failed and inFlight (only when the rule faulted an exchange)" + }, "expectation": { "type": "object", "additionalProperties": true, @@ -84,7 +129,8 @@ "required": [ "version", "matches", - "skipped" + "skipped", + "faultsLastMinute" ] }, "description": "The current rules (not the removed or replaced ones; only when chaos is on)" @@ -92,7 +138,7 @@ "report": { "type": "object", "additionalProperties": true, - "description": "The report: verdict, final, seed, seedGiven, lost, reasons, all rules with the removed and replaced ones, and the counts per route (only when chaos is on)" + "description": "The report: verdict, final, seed, seedGiven, lost, reasons, all rules with the removed and replaced ones, the counts per route, and the recent faults (the last 20, the latest first: timestamp, ago, rule, version, fault, hook, route, endpoint or step, exchangeId) (only when chaos is on)" } }, "required": [ diff --git a/components/camel-chaos/src/main/docs/chaos.adoc b/components/camel-chaos/src/main/docs/chaos.adoc index 6f83ea053c0f..c32902822915 100644 --- a/components/camel-chaos/src/main/docs/chaos.adoc +++ b/components/camel-chaos/src/main/docs/chaos.adoc @@ -126,7 +126,9 @@ integration; `chaos-clear` without one acts on every running integration that ha `chaos-clear` removes stay in the report; with `--reset` the report starts again, for a new experiment: no rules, route counts or verdict, as for an application that just started (with the same seed, and the ids of new rules going on from the last one). An id that does not exist is "not found" (exit code 3). `camel get chaos` lists the current rules with their id, version, state, -matches and faults, and `--report` shows the report (also as JSON with `--json`). Their `NOTE` column says what a +matches and faults, the observed rate (`RATE`), when the rule last fired (`LAST`) and the effect of its faults +(`EFFECT`), and `--report` shows the report, with the recent faults (also as JSON with `--json`); see +<<_the_live_state_of_the_rules,The live state of the rules>>. Their `NOTE` column says what a lifecycle rule does to its route (`route poller down, back in 12s`) and why a rule did nothing when it was skipped. Without a rule to show, `camel get chaos` says so on stderr (no rules, chaos off, or no integration with camel-chaos). @@ -505,6 +507,46 @@ give their reason only. In JSON an expectation has its `kind`, `op`, `count` and the `value` counted so far (and `completed`, the same value, for an expectation of completed exchanges). +=== The live state of the rules + +While the rules run, the report tells how each one is doing, and what its faults do: + +* the observed rate next to the rate of the rule: the share of the matches it fired at (`rate 4.8% of 5%`, and +`rate 20% of 1 in 5` for `rate=every:5`), and for `rate=every:N` the matches until it fires again; +* when it last fired, and its faults per second over the last minute; +* the effect of its faults on the exchanges they hit, each exchange counted once when it is done: `completed` (without a +redelivery after the fault, such as after a `delay`), `recovered` by a redelivery of the error handler, `handled` by +the error handler (such as the dead letter channel), `failed`, and the others still in flight. Only an exchange with a +unit of work counts (one that a route processes), as that tells when it is done. For a lifecycle rule (or `stall`), the +exchanges sent to the endpoint of its route while the route is down count, such as `direct:inventory` or +`seda:shipping` (not an HTTP call to a `platform-http` route, which is another endpoint): a 3 second stop of a route +called with `direct:inventory?block=false` that sends 5 orders to the dead letter channel shows as +`5 handled by the error handler`, as the redelivery is too short to ride it out; +* the last 20 faults of all rules, the latest first: when, the rule, the fault, the hook, the route, the endpoint +(without its query) or the step, and the exchange id. + +An exchange that a rule faulted (or that was sent to the route of a lifecycle rule while it was down) gets the exchange +property `CamelChaosRule` with the id of the rule (of the last one, when more rules did), so the tracer, the message history and a route of the dead letter channel can tell why +it is there: + +[source,text] +---- + r1 active: endpoint=kafka:orders* error rate=0.05 until=10m + matches 1250, fired 61 (send 61), skipped 0, ends in 7m12s + rate 4.9% of 5%, last fired 2s ago, 6 faults in the last minute + effect on 58 exchanges: 55 recovered by redelivery, 2 handled by the error handler, 1 in flight + recent faults: + 2s ago r1 error ChaosException at send kafka://orders, route orders, exchange 4F2A... +---- + +In JSON a rule has `rate` (the one of the rule: 1 without `rate=`) and `observedRate` (once it matched), `nextFireIn` +(the matches until a `rate=every:N` rule fires again), `lastFiredAgo` (milliseconds), `faultsLastMinute`, +`faultsPerSecond` (60 counts, from the oldest second to the current one, when it fired in the last minute) and `effect` +(`faulted`, `completed`, `recovered`, `handled`, `failed` and `inFlight`, once it faulted an exchange); a lifecycle rule +has no rate. The report has `recentFaults`, each with `timestamp`, `ago`, `rule`, `version`, `fault` (such as +`error status 503` or `delay 2s`), `hook` (none for a lifecycle rule), `route`, `endpoint` or `step`, and `exchangeId` +(none at a poll). `ChaosManager.reset()` (`chaos-clear --reset`) clears the recent faults too. + == Time and the seed The `delay` fault waits through `TimeSource` of camel-util, and the end of a rule, the time of an expectation and the diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosDevConsole.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosDevConsole.java index 1dd4aa4d646a..aea432ec43df 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosDevConsole.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosDevConsole.java @@ -46,7 +46,16 @@ public class ChaosDevConsole extends AbstractDevConsole { @Metadata(description = "The sends, polls, exchanges or steps the target of the rule matched") long matches, @Metadata(description = "The faults the rule fired, per hook") Map<String, Long> fired, @Metadata(description = "The faults that did nothing, as the rule fired at a message or an endpoint it cannot change") long skipped, + @Metadata(description = "Why the rule did nothing when it was skipped: the count per reason (only when it was skipped)") Map<String, Long> skippedReasons, + @Metadata(description = "The messages lost because the rule failed or dropped the send of the dead letter channel, which then handled them (only for a rule with at=dlq)") Long lostAtDlq, @Metadata(description = "The milliseconds until an active rule ends") Long endsIn, + @Metadata(description = "The share of the matches the rule fires at: rate=, 1/N for rate=every:N, else 1 (not for a lifecycle or stall rule)") Double rate, + @Metadata(description = "The share of the matches the rule fired at so far, to compare with rate (once it matched; not for a lifecycle or stall rule)") Double observedRate, + @Metadata(description = "The matches until a rule with rate=every:N fires again (only for such a rule while it is active)") Long nextFireIn, + @Metadata(description = "The milliseconds since the rule last fired (only when it fired)") Long lastFiredAgo, + @Metadata(description = "The faults the rule fired in the last minute") long faultsLastMinute, + @Metadata(description = "The faults the rule fired per second over the last minute, 60 counts from the oldest second to the current one (only when it fired in the last minute)") List<Long> faultsPerSecond, + @Metadata(description = "The effect of the faults on the exchanges they hit (for a lifecycle or stall rule: the exchanges sent to its route while it was down), each counted once when it is done: faulted, completed, recovered (by a redelivery of the error handler), handled (by the error handler, such as the dead letter channel), failed and inFlight (only when the rule faulted an exchange)") Map<String, Long> effect, @Metadata(description = "The expectation of the rule (expect= and within=) and its outcome, when it has one") Map<String, Object> expectation, @Metadata(description = "What a lifecycle or stall rule did to its route: action, route, duration, state (pending, acting, down, comingBack, back, skipped, left, failed or cancelled), downAt, backAt, backIn, endedEarly and reason (only for such a rule)") Map<String, Object> lifecycle) { } @@ -55,7 +64,7 @@ public class ChaosDevConsole extends AbstractDevConsole { @Metadata(description = "Whether chaos is switched on: with profile dev or camel.chaos.enabled=true, never with profile prod") boolean enabled, @Metadata(description = "Why chaos is off (only when it is off)") String reason, @Metadata(description = "The current rules (not the removed or replaced ones; only when chaos is on)") List<RuleEntry> rules, - @Metadata(description = "The report: verdict, final, seed, seedGiven, lost, reasons, all rules with the removed and replaced ones, and the counts per route (only when chaos is on)") Map<String, Object> report) { + @Metadata(description = "The report: verdict, final, seed, seedGiven, lost, reasons, all rules with the removed and replaced ones, the counts per route, and the recent faults (the last 20, the latest first: timestamp, ago, rule, version, fault, hook, route, endpoint or step, exchangeId) (only when chaos is on)") Map<String, Object> report) { } public ChaosDevConsole() { diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosEventNotifier.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosEventNotifier.java index 20146e55679a..a027aedf5926 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosEventNotifier.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosEventNotifier.java @@ -18,9 +18,7 @@ package org.apache.camel.component.chaos; import org.apache.camel.Exchange; import org.apache.camel.ExchangeExtension; -import org.apache.camel.Route; import org.apache.camel.spi.CamelEvent; -import org.apache.camel.spi.UnitOfWork; import org.apache.camel.support.EventNotifierSupport; /** @@ -57,9 +55,12 @@ final class ChaosEventNotifier extends EventNotifierSupport { } if (event instanceof CamelEvent.ExchangeRedeliveryEvent redelivery && manager.isCounting()) { Exchange exchange = redelivery.getExchange(); - UnitOfWork uow = exchange.getUnitOfWork(); - Route route = uow != null ? uow.getRoute() : null; - String routeId = route != null ? route.getRouteId() : exchange.getFromRouteId(); + ChaosExchangeState state = ChaosExchangeState.peek(exchange); + if (state != null) { + // a rule that faulted the exchange before is recovered when the exchange then completes + state.redelivered(); + } + String routeId = ChaosExchangeState.routeId(exchange); if (routeId != null) { manager.redelivered(routeId); } diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosExchangeState.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosExchangeState.java index a11e49945e5b..7263c40adb99 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosExchangeState.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosExchangeState.java @@ -16,12 +16,17 @@ */ package org.apache.camel.component.chaos; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; import org.apache.camel.Exchange; import org.apache.camel.ExchangeExtension; +import org.apache.camel.Route; import org.apache.camel.SafeCopyProperty; +import org.apache.camel.spi.UnitOfWork; +import org.apache.camel.support.SynchronizationAdapter; /** * What chaos keeps of one exchange: how each route it went through is done with it, so a route counts the exchange once @@ -66,6 +71,8 @@ final class ChaosExchangeState implements SafeCopyProperty { } private final Map<String, RouteState> routes = new HashMap<>(4); + // the rules that faulted the exchange, counted in their effect when it is done (null until a rule faults it) + private Faults faults; // the dead letter channel is delivering the exchange boolean deadLetter; // the rule that failed or dropped that delivery, which loses the message when the dead letter channel then handles @@ -102,6 +109,99 @@ final class ChaosExchangeState implements SafeCopyProperty { return routes.computeIfAbsent(routeId, k -> new RouteState()); } + /** + * The route the exchange is in: the route of its unit of work, else the route that created it. + */ + static String routeId(Exchange exchange) { + UnitOfWork uow = exchange.getUnitOfWork(); + Route route = uow != null ? uow.getRoute() : null; + return route != null ? route.getRouteId() : exchange.getFromRouteId(); + } + + /** + * A rule faulted the exchange: the exchange property {@link ChaosManager#CHAOS_RULE} names the rule, and the effect + * of the rule counts the exchange once, when it is done. An exchange without a unit of work (such as one sent with + * a producer template to an endpoint, not to a route) is not counted in the effect, as nothing tells when it is + * done. + */ + static void faulted(ChaosRule rule, Exchange exchange) { + exchange.setProperty(ChaosManager.CHAOS_RULE, rule.id); + if (exchange.getUnitOfWork() == null) { + return; + } + ChaosExchangeState state = of(exchange); + if (state.faults == null) { + state.faults = new Faults(); + exchange.getExchangeExtension().addOnCompletion(state.faults); + } + state.faults.add(rule); + } + + /** The error handler redelivers the exchange: a rule that faulted it before is recovered when it then completes. */ + void redelivered() { + if (faults != null) { + faults.redelivered(); + } + } + + /** The rules that faulted an exchange, and whether the error handler redelivered it after each one. */ + private static final class Faults extends SynchronizationAdapter { + + private final List<ChaosRule> rules = new ArrayList<>(2); + private final List<Boolean> redelivered = new ArrayList<>(2); + + void add(ChaosRule rule) { + int i = rules.indexOf(rule); + if (i < 0) { + rules.add(rule); + redelivered.add(Boolean.FALSE); + rule.effect.faulted(); + } else { + // faulted again (such as at a redelivery): the redelivery that follows counts from now + redelivered.set(i, Boolean.FALSE); + } + } + + void redelivered() { + for (int i = 0; i < redelivered.size(); i++) { + redelivered.set(i, Boolean.TRUE); + } + } + + @Override + public boolean allowHandover() { + // the effect is how this exchange was done: not a copy that a seda endpoint (such as the one of a dead + // letter channel) hands the completions over to, which its route completes + return false; + } + + @Override + public void onDone(Exchange exchange) { + // an exchange marked rollback only has no exception, but its message is not processed either + boolean failed = exchange.isFailed() || exchange.isRollbackOnly() || exchange.isRollbackOnlyLast(); + ExchangeExtension ee = exchange.getExchangeExtension(); + boolean handled = !failed && ee.isErrorHandlerHandledSet() && ee.isErrorHandlerHandled(); + for (int i = 0; i < rules.size(); i++) { + ChaosRule.Effect.Outcome outcome; + if (failed) { + outcome = ChaosRule.Effect.Outcome.FAILED; + } else if (handled) { + outcome = ChaosRule.Effect.Outcome.HANDLED; + } else if (redelivered.get(i)) { + outcome = ChaosRule.Effect.Outcome.RECOVERED; + } else { + outcome = ChaosRule.Effect.Outcome.COMPLETED; + } + rules.get(i).effect.done(outcome); + } + } + + @Override + public String toString() { + return "ChaosFaults" + rules; + } + } + @Override public SafeCopyProperty safeCopy() { // a copy is another exchange: it is counted, and goes through the dead letter channel, on its own diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosFaults.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosFaults.java index 9a78d96b0a8b..fdae40f9edc0 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosFaults.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosFaults.java @@ -90,19 +90,19 @@ final class ChaosFaults { return httpFailure(rule, exchange, endpoint, hook); } exchange.setException(newException(rule)); - rule.fired(hook); + rule.fired(hook, exchange, null, endpoint.getEndpointUri()); return SendHook.PROCEED; } case DELAY -> { - rule.fired(hook); + rule.fired(hook, exchange, null, endpoint.getEndpointUri()); return rule.duration; } case DROP -> { - rule.fired(hook); + rule.fired(hook, exchange, null, endpoint.getEndpointUri()); return SendHook.SKIP; } case CORRUPT -> { - corrupt(rule, exchange, hook); + corrupt(rule, exchange, hook, null, endpoint.getEndpointUri()); return SendHook.PROCEED; } default -> throw new IllegalStateException("Unknown fault " + rule.fault); @@ -127,7 +127,7 @@ final class ChaosFaults { if (Boolean.FALSE.equals(toBoolean(option(endpoint, "throwExceptionOnFailure"))) || isOkStatus(rule.status, option(endpoint, "okStatusCodeRange"))) { response(message, rule, !undertow); - rule.fired(hook); + rule.fired(hook, exchange, null, endpoint.getEndpointUri()); return SendHook.SKIP; } Throwable failure; @@ -153,7 +153,7 @@ final class ChaosFaults { response(message, rule, true); } exchange.setException(failure); - rule.fired(hook); + rule.fired(hook, exchange, null, endpoint.getEndpointUri()); return SendHook.PROCEED; } @@ -346,8 +346,11 @@ final class ChaosFaults { /** * Changes the message of the exchange as the mode of the rule says, and counts the rule as fired at the hook, or as * skipped when the message has nothing the mode can change. + * + * @param routeId the route at its entry, else null for the route of the exchange + * @param where the endpoint URI of a send, else null */ - static void corrupt(ChaosRule rule, Exchange exchange, Hook hook) { + static void corrupt(ChaosRule rule, Exchange exchange, Hook hook, String routeId, String where) { // why the message was left as it is, or null when it was changed String skip; try { @@ -361,7 +364,7 @@ final class ChaosFaults { skip = "the body could not be read (" + e.getClass().getSimpleName() + ")"; } if (skip == null) { - rule.fired(hook); + rule.fired(hook, exchange, routeId, where); } else { rule.skipped(skip); } diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosLifecycle.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosLifecycle.java index c60b0a200735..0a4ae3c059e2 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosLifecycle.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosLifecycle.java @@ -317,7 +317,9 @@ final class ChaosLifecycle { } fault.state = FaultState.DOWN; fault.downAt = now; + rule.firedLifecycle(routeId, now); fault.route = route; + fault.downUri = route != null ? route.getEndpoint().getEndpointBaseUri() : null; fault.definition = definition; fault.comebackAs = comebackAs; fault.statusAfter = comebackAs == Action.REMOVE ? null : controller.getRouteStatus(routeId); @@ -444,11 +446,22 @@ final class ChaosLifecycle { private ServiceStatus statusAfter; // how the route comes back: the action of the rule, or how a half failed action left the route private Action comebackAs; + // the endpoint URI of the route, without its query, while it is down (read by the send hook without the lock), + // else null + private volatile String downUri; Fault(ChaosRule rule) { this.rule = rule; } + /** + * The endpoint URI of the route, without its query, while the fault has it down, so the exchanges sent to it + * then count in the effect of the rule; else null. + */ + String downEndpointUri() { + return downUri; + } + /** Whether the fault brings its route back now (the rule was removed, or the comeback runs). */ synchronized boolean isEnding() { return endRequested || endedEarly || state == FaultState.COMING_BACK; @@ -556,6 +569,7 @@ final class ChaosLifecycle { reason = why; task = null; route = null; + downUri = null; definition = null; if (owner != null) { owner.track(this, false); diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosManager.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosManager.java index ba1daa0e85ef..d23f2dc1658b 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosManager.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosManager.java @@ -54,6 +54,12 @@ import org.slf4j.LoggerFactory; */ public final class ChaosManager { + /** + * The exchange property with the id of the rule that faulted the exchange (the last one, when more did), such as + * {@code r1}, so the tracer, the message history and a route of the dead letter channel can tell. + */ + public static final String CHAOS_RULE = "CamelChaosRule"; + /** * The most rules at a time. When there are this many, adding a rule first moves the rules that ended (and whose * expectation is decided) into the report history; remove some to add more when that does not make room. @@ -87,6 +93,8 @@ public final class ChaosManager { private volatile boolean counting; private volatile boolean used; private final ConcurrentMap<String, RouteCounts> routes = new ConcurrentHashMap<>(); + // the last faults of the rules, for the chaos screen + private final ChaosRecentFaults recentFaults = new ChaosRecentFaults(); // the route lifecycle faults (stall and lifecycle), and whether the routes of the context are created (guarded by // lock), so the route of such a rule can be checked private final ChaosLifecycle lifecycle; @@ -176,6 +184,7 @@ public final class ChaosManager { List<ChaosRule> added = new ArrayList<>(); for (ChaosRule.Builder definition : definitions) { ChaosRule rule = new ChaosRule(definition, "r" + nextId++, seed, now, maxDuration); + rule.recentFaults = recentFaults; // a rule added before the context started starts its clock when it has started rule.clockStarted = camelContext.isStarted(); rules.add(rule); @@ -215,6 +224,7 @@ public final class ChaosManager { checkRoute(definition); long now = TimeSource.currentTimeMillis(); ChaosRule changed = new ChaosRule(definition, id, rules.get(index).version + 1, seed, now, maxDuration); + changed.recentFaults = recentFaults; changed.clockStarted = camelContext.isStarted(); ChaosRule old = rules.set(index, changed); old.removed(State.REPLACED, now); @@ -320,8 +330,9 @@ public final class ChaosManager { } /** - * Removes all rules and starts a new report: the removed rules, the counts per route and the verdict are gone, as - * for an application that just started (with the same seed). The ids of new rules go on from the last one. + * Removes all rules and starts a new report: the removed rules, the counts per route, the recent faults and the + * verdict are gone, as for an application that just started (with the same seed). The ids of new rules go on from + * the last one. * * @return the number of rules removed */ @@ -334,13 +345,15 @@ public final class ChaosManager { evictedPending = 0; evictedLostAtDlq = 0; routes.clear(); + recentFaults.clear(); LOG.info("Chaos report reset"); return count; } } /** - * The report: the rules and what they did, the counts per route, the expectations and the verdict. + * The report: the rules and what they did, the counts per route, the expectations, the recent faults and the + * verdict. */ public ChaosReport report() { List<ChaosRule> current; @@ -360,7 +373,7 @@ public final class ChaosManager { Map<String, RouteCounts> counts = new TreeMap<>(routes); return new ChaosReport( seed, seedGiven, TimeSource.currentTimeMillis(), current, removed, counts, unmet, lost, pending, - lostAtDlq); + lostAtDlq, recentFaults.latestFirst()); } // -- for the hooks, the event notifier and the plugin @@ -698,6 +711,8 @@ public final class ChaosManager { final ChaosRule[] poll; final ChaosRule[] entry; final ChaosRule[] step; + // the lifecycle rules (stall and lifecycle), for the exchanges sent to their route while it is down + final ChaosRule[] lifecycle; final ChaosRule.Expectation[] expectations; final ChaosRule[] expectationRules; @@ -708,6 +723,7 @@ public final class ChaosManager { poll = live.stream().filter(r -> r.hooks.contains(Hook.POLL)).toArray(ChaosRule[]::new); entry = live.stream().filter(r -> r.hooks.contains(Hook.ENTRY)).toArray(ChaosRule[]::new); step = live.stream().filter(r -> r.hooks.contains(Hook.STEP)).toArray(ChaosRule[]::new); + lifecycle = rules.stream().filter(r -> r.lifecycle != null).toArray(ChaosRule[]::new); // all of them, for the WARN about a route that does not exist when the context starts expectationRules = rules.stream().filter(r -> r.expectation != null).toArray(ChaosRule[]::new); expectations = Arrays.stream(expectationRules).map(r -> r.expectation) diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosPollStrategy.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosPollStrategy.java index 44fbbe4891d5..5c48418e2e6a 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosPollStrategy.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosPollStrategy.java @@ -21,6 +21,7 @@ import org.apache.camel.Endpoint; import org.apache.camel.RuntimeCamelException; import org.apache.camel.component.chaos.ChaosRule.Hook; import org.apache.camel.spi.PollingConsumerPollStrategy; +import org.apache.camel.spi.RouteIdAware; import org.apache.camel.support.DefaultPollingConsumerPollStrategy; import org.apache.camel.support.service.ServiceSupport; import org.apache.camel.util.TimeSource; @@ -69,6 +70,11 @@ final class ChaosPollStrategy implements PollingConsumerPollStrategy { return beginDelegate(consumer, endpoint); } + // the route of the consumer, for the recent faults + private static String routeId(Consumer consumer) { + return consumer instanceof RouteIdAware aware ? aware.getRouteId() : null; + } + // the strategy of the consumer begins the poll; when it skips it, the poll run is done private boolean beginDelegate(Consumer consumer, Endpoint endpoint) { boolean begun = delegate.begin(consumer, endpoint); @@ -81,12 +87,12 @@ final class ChaosPollStrategy implements PollingConsumerPollStrategy { private boolean fault(ChaosRule rule, Consumer consumer, Endpoint endpoint) { switch (rule.fault) { case DROP -> { - rule.fired(Hook.POLL); + rule.fired(Hook.POLL, null, routeId(consumer), endpoint.getEndpointUri()); endRun(); return false; } case DELAY -> { - rule.fired(Hook.POLL); + rule.fired(Hook.POLL, null, routeId(consumer), endpoint.getEndpointUri()); try { TimeSource.sleep(rule.duration); } catch (InterruptedException e) { @@ -109,7 +115,7 @@ final class ChaosPollStrategy implements PollingConsumerPollStrategy { rule.skipped("the poll strategy of the consumer skipped the poll"); return false; } - rule.fired(Hook.POLL); + rule.fired(Hook.POLL, null, routeId(consumer), endpoint.getEndpointUri()); Throwable failure = ChaosFaults.newException(rule); // the strategy declares no checked exception: the consumer gets it wrapped, as its cause. An Error is // wrapped too, as the consumer handles only an Exception (it would stop polling for good otherwise) diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRecentFaults.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRecentFaults.java new file mode 100644 index 000000000000..ebd6111c2816 --- /dev/null +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRecentFaults.java @@ -0,0 +1,60 @@ +/* + * 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.component.chaos; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; + +import org.apache.camel.component.chaos.ChaosRule.Hook; + +/** + * The last faults the rules fired, for the chaos screen: a small ring, the oldest fault drops out when it is full. + */ +final class ChaosRecentFaults { + + // the most faults it keeps + static final int SIZE = 20; + + private final Deque<Entry> entries = new ArrayDeque<>(SIZE); + + /** + * A fault: when, the rule, the hook (null for a lifecycle rule), the route, the endpoint URI of a send or poll or + * the id of a step (null at the entry of a route), and the exchange (null at a poll and for a lifecycle rule). + */ + record Entry(long time, ChaosRule rule, Hook hook, String routeId, String where, String exchangeId) { + } + + synchronized void add(long time, ChaosRule rule, Hook hook, String routeId, String where, String exchangeId) { + if (entries.size() == SIZE) { + entries.removeFirst(); + } + entries.addLast(new Entry(time, rule, hook, routeId, where, exchangeId)); + } + + /** The faults, the latest first. */ + synchronized List<Entry> latestFirst() { + List<Entry> answer = new ArrayList<>(entries.size()); + entries.descendingIterator().forEachRemaining(answer::add); + return answer; + } + + synchronized void clear() { + entries.clear(); + } +} diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosReport.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosReport.java index 6cc1e99a1813..eba7b7b23f7f 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosReport.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosReport.java @@ -66,12 +66,17 @@ public final class ChaosReport { private final List<String> reasons = new ArrayList<>(); private final long lost; private final boolean pending; + private final long now; + // the last faults, the latest first + private final List<ChaosRecentFaults.Entry> recentFaults; ChaosReport(long seed, boolean seedGiven, long now, List<ChaosRule> current, List<ChaosRule> removed, Map<String, RouteCounts> counts, int evictedUnmet, long evictedLost, int evictedPending, - long evictedLostAtDlq) { + long evictedLostAtDlq, List<ChaosRecentFaults.Entry> recentFaults) { this.seed = seed; this.seedGiven = seedGiven; + this.now = now; + this.recentFaults = List.copyOf(recentFaults); long lostCount = 0; boolean anyPending = false; for (ChaosRule rule : current) { @@ -210,6 +215,13 @@ public final class ChaosReport { if (r.state == State.ACTIVE && r.lifecycle == null) { sb.append(", ends in ").append(ChaosRuleParser.printTime(r.endsIn)); } + String live = liveText(r); + if (!live.isEmpty()) { + sb.append("\n ").append(live); + } + if (r.faulted > 0) { + sb.append("\n ").append(effectText(r)); + } if (r.lifecycle != null) { sb.append("\n ").append(lifecycleText(r)); } @@ -233,6 +245,12 @@ public final class ChaosReport { .append(r.failed).append(" (kept by the consumer ").append(r.failedKept).append("), handled ") .append(r.handled).append(", redeliveries ").append(r.redeliveries); } + if (!recentFaults.isEmpty()) { + sb.append("\n recent faults:"); + for (ChaosRecentFaults.Entry f : recentFaults) { + sb.append("\n ").append(recentFaultText(f)); + } + } sb.append("\n lost: ").append(lost); for (String reason : reasons) { sb.append("\n FAIL: ").append(reason); @@ -240,6 +258,89 @@ public final class ChaosReport { return sb.toString(); } + /** + * The live state of a rule, for a person: its observed rate next to the one of the rule, the next match an + * {@code every:N} rule fires at, when it last fired and its faults in the last minute. + */ + private static String liveText(RuleEntry r) { + StringJoiner sj = new StringJoiner(", "); + if (r.observedRate >= 0 && r.rule.targetRate() < 1.0d) { + sj.add("rate " + percent(r.observedRate) + " of " + + (r.rule.every > 0 ? "1 in " + r.rule.every : percent(r.rule.targetRate()))); + } + if (r.nextFireIn > 0) { + sj.add("fires again in " + r.nextFireIn + (r.nextFireIn == 1 ? " match" : " matches")); + } + if (r.lastFiredAgo >= 0) { + sj.add("last fired " + ChaosRuleParser.printTime(r.lastFiredAgo) + " ago"); + sj.add(r.faultsLastMinute + (r.faultsLastMinute == 1 ? " fault" : " faults") + " in the last minute"); + } + return sj.toString(); + } + + /** What the faults of a rule did to the exchanges, for a person; the counts that are 0 are left out. */ + private static String effectText(RuleEntry r) { + StringJoiner sj = new StringJoiner( + ", ", "effect on " + r.faulted + (r.faulted == 1 ? " exchange: " : " exchanges: "), + ""); + if (r.completed > 0) { + sj.add(r.completed + " completed"); + } + if (r.recovered > 0) { + sj.add(r.recovered + " recovered by redelivery"); + } + if (r.handled > 0) { + sj.add(r.handled + " handled by the error handler"); + } + if (r.failed > 0) { + sj.add(r.failed + " failed"); + } + if (r.inFlight > 0) { + sj.add(r.inFlight + " in flight"); + } + return sj.toString(); + } + + /** + * A recent fault, for a person, such as {@code 3s ago r1 error ChaosException at send mock://stock, route orders}. + */ + private String recentFaultText(ChaosRecentFaults.Entry f) { + StringBuilder sb = new StringBuilder(); + sb.append(ChaosRuleParser.printTime(Math.max(0, now - f.time()))).append(" ago ").append(f.rule().id); + if (f.rule().version > 1) { + sb.append(" v").append(f.rule().version); + } + sb.append(' ').append(f.rule().faultText()); + if (f.hook() == null) { + sb.append(" of route ").append(f.routeId()); + return sb.toString(); + } + if (f.hook() == Hook.ENTRY) { + sb.append(" at the entry of route ").append(f.routeId()); + } else { + sb.append(" at ").append(f.hook().key).append(' ').append(where(f)); + if (f.routeId() != null) { + sb.append(", route ").append(f.routeId()); + } + } + if (f.exchangeId() != null) { + sb.append(", exchange ").append(f.exchangeId()); + } + return sb.toString(); + } + + // the endpoint of a fault without its query, which can hold secrets, or the id of its step + private static String where(ChaosRecentFaults.Entry f) { + return f.hook() == Hook.STEP ? f.where() : ChaosTargets.withoutQuery(f.where()); + } + + // a share as a percent, such as 5% or 5.1% + private static String percent(double share) { + double p = share * 100; + double tenths = Math.round(p * 10) / 10.0d; + return (tenths == Math.rint(tenths) ? Long.toString((long) tenths) : Double.toString(tenths)) + "%"; + } + /** * The report as JSON, for a tool. */ @@ -264,9 +365,37 @@ public final class ChaosReport { routeArray.add(jo); } root.put("routes", routeArray); + root.put("recentFaults", recentFaultsJson()); return root; } + /** The recent faults as JSON, the latest first, as in {@link #toJson()}. */ + JsonArray recentFaultsJson() { + JsonArray answer = new JsonArray(); + for (ChaosRecentFaults.Entry f : recentFaults) { + JsonObject jo = new JsonObject(); + jo.put("timestamp", f.time()); + jo.put("ago", Math.max(0, now - f.time())); + jo.put("rule", f.rule().id); + jo.put("version", f.rule().version); + jo.put("fault", f.rule().faultText()); + if (f.hook() != null) { + jo.put("hook", f.hook().key); + } + if (f.routeId() != null) { + jo.put("route", f.routeId()); + } + if (f.where() != null) { + jo.put(f.hook() == Hook.STEP ? "step" : "endpoint", where(f)); + } + if (f.exchangeId() != null) { + jo.put("exchangeId", f.exchangeId()); + } + answer.add(jo); + } + return answer; + } + /** * The rules of the report as JSON, as in {@link #toJson()}: all of them, or only the current ones (not removed, * replaced or moved to the history), for the chaos dev console. @@ -310,6 +439,7 @@ public final class ChaosReport { if (r.state == State.ACTIVE && r.lifecycle == null) { jo.put("endsIn", r.endsIn); } + liveJson(r, jo); if (r.outcome != null) { JsonObject e = new JsonObject(); e.put("route", r.expectRoute); @@ -339,6 +469,45 @@ public final class ChaosReport { return toText(); } + /** + * The live state of a rule, for a tool: the rate of the rule and the observed one (the share of the matches it + * fired at), the next match an {@code every:N} rule fires at, when it last fired, its faults per second over the + * last minute, and the effect of its faults on the exchanges. + */ + private static void liveJson(RuleEntry r, JsonObject jo) { + if (r.lifecycle == null) { + jo.put("rate", r.rule.targetRate()); + if (r.observedRate >= 0) { + jo.put("observedRate", Math.round(r.observedRate * 10000) / 10000.0d); + } + } + if (r.nextFireIn > 0) { + jo.put("nextFireIn", r.nextFireIn); + } + if (r.lastFiredAgo >= 0) { + jo.put("lastFiredAgo", r.lastFiredAgo); + } + jo.put("faultsLastMinute", r.faultsLastMinute); + if (r.faultsLastMinute > 0) { + // the oldest second first; the last one is the current second + JsonArray perSecond = new JsonArray(); + for (long n : r.faultsPerSecond) { + perSecond.add(n); + } + jo.put("faultsPerSecond", perSecond); + } + if (r.faulted > 0) { + JsonObject e = new JsonObject(); + e.put("faulted", r.faulted); + e.put("completed", r.completed); + e.put("recovered", r.recovered); + e.put("handled", r.handled); + e.put("failed", r.failed); + e.put("inFlight", r.inFlight); + jo.put("effect", e); + } + } + /** * What a lifecycle rule did to its route, for a person; the times are after the rule was added, such as * {@code suspended route orders at +0ms, resumed at +10m0s}. @@ -419,6 +588,20 @@ public final class ChaosReport { final Map<String, Long> skipReasons; final long lostAtDlq; final long endsIn; + // the live state: the share of the matches the rule fired at (-1 without a match), the matches until an + // every:N rule fires again (0 when it does not), how long ago it last fired (-1 when it did not), its faults + // per second over the last minute, and the effect of its faults on the exchanges + final double observedRate; + final long nextFireIn; + final long lastFiredAgo; + final long[] faultsPerSecond; + final long faultsLastMinute; + final long faulted; + final long completed; + final long recovered; + final long handled; + final long failed; + final long inFlight; final String expectRoute; final ChaosRule.Expectation.Kind expectKind; final ChaosRule.Expectation.Op expectOp; @@ -459,6 +642,25 @@ public final class ChaosReport { skipReasons = rule.skipReasons(); lostAtDlq = rule.lostAtDlqCount(); endsIn = Math.max(0, rule.endsAt - now); + // a lifecycle rule acts once at its route: it has no rate + observedRate = matches > 0 && rule.lifecycle == null ? Math.min(1.0d, (double) (total + skipped) / matches) : -1; + nextFireIn = rule.every > 0 && state == State.ACTIVE ? rule.every - matches % rule.every : 0; + long last = rule.lastFiredAt(); + lastFiredAgo = last >= 0 ? Math.max(0, now - last) : -1; + faultsPerSecond = rule.faultsPerSecond(now); + long minute = 0; + for (long n : faultsPerSecond) { + minute += n; + } + faultsLastMinute = minute; + ChaosRule.Effect effect = rule.effect; + // the done counts first, so an exchange that is done while the report is made is not in flight twice + completed = effect.count(ChaosRule.Effect.Outcome.COMPLETED); + recovered = effect.count(ChaosRule.Effect.Outcome.RECOVERED); + handled = effect.count(ChaosRule.Effect.Outcome.HANDLED); + failed = effect.count(ChaosRule.Effect.Outcome.FAILED); + faulted = Math.max(effect.faultedCount(), completed + recovered + handled + failed); + inFlight = faulted - completed - recovered - handled - failed; within = rule.within; ChaosRule.Expectation expectation = rule.expectation; if (expectation != null) { diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRoutePolicyFactory.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRoutePolicyFactory.java index 0863d0be70e5..9100a0d0f095 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRoutePolicyFactory.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRoutePolicyFactory.java @@ -66,7 +66,7 @@ final class ChaosRoutePolicyFactory implements RoutePolicyFactory { long now = TimeSource.currentTimeMillis(); for (ChaosRule rule : rules) { if (rule.live(now) && rule.matchesId(route.getRouteId()) && rule.fires(now)) { - ChaosFaults.corrupt(rule, exchange, Hook.ENTRY); + ChaosFaults.corrupt(rule, exchange, Hook.ENTRY, route.getRouteId(), null); return; } } diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRule.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRule.java index a79ac472d861..fc60691c08a6 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRule.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRule.java @@ -25,6 +25,7 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.LongAdder; +import org.apache.camel.Exchange; import org.apache.camel.util.TimeSource; /** @@ -128,6 +129,9 @@ final class ChaosRule { // the golden ratio increment of SplitMix64 private static final long GOLDEN_GAMMA = 0x9e3779b97f4a7c15L; + // the seconds of the faults per second the rule keeps: the last minute + static final int MINUTE = 60; + // the definition (immutable) final String id; // 1 when the rule is added, one more each time it is changed (the report has an entry per version of an id) @@ -182,6 +186,14 @@ final class ChaosRule { boolean clockStarted; // what a lifecycle rule did to its route, else null final ChaosLifecycle.Fault lifecycle; + // the live state, for the chaos screen: when the rule last fired, its faults per second over the last minute (a + // ring of seconds, guarded by itself), the effect of its faults on the exchanges, and the recent faults of the + // manager (null in the tests of a rule alone; set by the manager before the rule is published to the hooks) + private volatile long lastFiredAt = -1; + private final long[] faultsPerSecond = new long[MINUTE]; + private final long[] secondOf = new long[MINUTE]; + final Effect effect = new Effect(); + ChaosRecentFaults recentFaults; ChaosRule(Builder b, String id, long seed, long addedAt, long maxDuration) { this(b, id, 1, seed, addedAt, maxDuration); @@ -260,9 +272,94 @@ final class ChaosRule { return now < endsAt && (untilCount == 0 || matches.get() < untilCount); } - /** Counts that the rule fired at the hook. */ + /** Counts that the rule fired at the hook, where there is no exchange (a poll). */ void fired(Hook hook) { + fired(hook, null, null, null); + } + + /** + * Counts that the rule fired at the hook, with the live state of the rule: the time, the faults per second, the + * recent faults, and the exchange it faulted (marked with {@link ChaosManager#CHAOS_RULE}, and counted in the + * effect of the rule). + * + * @param exchange the exchange, or null at a poll + * @param routeId the route, or null to take the route of the exchange + * @param where the endpoint URI of a send or poll, or the id of a step, or null at the entry of a route + */ + void fired(Hook hook, Exchange exchange, String routeId, String where) { fired[hook.ordinal()].increment(); + long now = TimeSource.currentTimeMillis(); + countFault(now); + if (exchange != null) { + ChaosExchangeState.faulted(this, exchange); + } + ChaosRecentFaults recent = recentFaults; + if (recent != null) { + recent.add(now, this, hook, routeId != null || exchange == null ? routeId : ChaosExchangeState.routeId(exchange), + where, exchange != null ? exchange.getExchangeId() : null); + } + } + + /** A lifecycle rule (or stall) took its route down: counted in its live state, as the faults at a hook are. */ + void firedLifecycle(String routeId, long now) { + countFault(now); + ChaosRecentFaults recent = recentFaults; + if (recent != null) { + recent.add(now, this, null, routeId, null, null); + } + } + + private void countFault(long now) { + lastFiredAt = now; + long second = now / 1000; + int i = (int) (second % MINUTE); + synchronized (faultsPerSecond) { + if (secondOf[i] != second) { + secondOf[i] = second; + faultsPerSecond[i] = 0; + } + faultsPerSecond[i]++; + } + } + + /** When the rule last fired, or -1 when it has not. */ + long lastFiredAt() { + return lastFiredAt; + } + + /** The faults per second over the last minute, the oldest second first: the last one is the current second. */ + long[] faultsPerSecond(long now) { + long second = now / 1000; + long[] answer = new long[MINUTE]; + synchronized (faultsPerSecond) { + for (int k = 0; k < MINUTE; k++) { + long s = second - (MINUTE - 1) + k; + int i = (int) Math.floorMod(s, MINUTE); + answer[k] = secondOf[i] == s ? faultsPerSecond[i] : 0; + } + } + return answer; + } + + /** + * The share of the matches the rule decides to fire at: the probability of {@code rate=}, one in N for + * {@code rate=every:N}, else all of them. + */ + double targetRate() { + return every > 0 ? 1.0d / every : probability; + } + + /** What the rule does when it fires, in a few words, such as {@code error status 503} or {@code delay 2s}. */ + String faultText() { + return switch (fault) { + case ERROR -> "error " + (status > 0 + ? "status " + status + : exceptionType != null ? exceptionType.getSimpleName() : ChaosException.class.getSimpleName()); + case DELAY -> "delay " + ChaosRuleParser.printTime(duration); + case DROP -> "drop"; + case CORRUPT -> "corrupt " + mode.key; + case LIFECYCLE -> (stall ? "stall" : action.key) + " " + ChaosRuleParser.printTime(duration); + }; } /** @@ -569,6 +666,48 @@ final class ChaosRule { } } + /** + * The effect of the faults of a rule on the exchanges it faulted, each counted once when it is done: completed + * (without a redelivery after the fault, such as after a delay), recovered by a redelivery of the error handler, + * handled by the error handler (such as the dead letter channel), or failed. The others are still in flight. Only + * the exchanges with a unit of work count, as it tells when the exchange is done. + */ + static final class Effect { + + /** How an exchange was done, after the rule faulted it. */ + enum Outcome { + COMPLETED, + RECOVERED, + HANDLED, + FAILED + } + + private final LongAdder faulted = new LongAdder(); + private final LongAdder[] done = new LongAdder[Outcome.values().length]; + + Effect() { + for (int i = 0; i < done.length; i++) { + done[i] = new LongAdder(); + } + } + + void faulted() { + faulted.increment(); + } + + void done(Outcome outcome) { + done[outcome.ordinal()].increment(); + } + + long faultedCount() { + return faulted.sum(); + } + + long count(Outcome outcome) { + return done[outcome.ordinal()].sum(); + } + } + /** The definition of a rule, as the parser builds it. */ static final class Builder { String text; diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosSendHook.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosSendHook.java index 9487bbadbb4a..f67d97c3358c 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosSendHook.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosSendHook.java @@ -40,6 +40,9 @@ final class ChaosSendHook implements SendHook { @Override public long beforeSend(Exchange exchange, Endpoint endpoint) { ChaosManager.RuleSet ruleSet = manager.ruleSet(); + if (ruleSet.lifecycle.length > 0) { + sentToDownRoute(ruleSet.lifecycle, exchange, endpoint); + } if (ruleSet.send.length == 0 && ruleSet.dlq.length == 0) { return PROCEED; } @@ -64,6 +67,27 @@ final class ChaosSendHook implements SendHook { return PROCEED; } + /** + * An exchange sent to the endpoint of a route that a lifecycle rule has down is marked with the rule, and counts in + * the effect of the rule (such as one that goes to the dead letter channel as the route is stopped). It is not a + * fault of its own: the rule fired when it took the route down. + */ + private static void sentToDownRoute(ChaosRule[] rules, Exchange exchange, Endpoint endpoint) { + String uri = null; + for (ChaosRule rule : rules) { + String down = rule.lifecycle.downEndpointUri(); + if (down != null) { + if (uri == null) { + // with or without the options of the endpoint (direct:shipping?block=false) + uri = endpoint.getEndpointBaseUri(); + } + if (down.equals(uri)) { + ChaosExchangeState.faulted(rule, exchange); + } + } + } + } + @Override public void afterSend(Exchange exchange, Endpoint endpoint) { // the faults are raised before the send diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosStepHook.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosStepHook.java index a575fb80511b..4bfd6b73464f 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosStepHook.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosStepHook.java @@ -45,7 +45,7 @@ final class ChaosStepHook implements StepHook { long now = TimeSource.currentTimeMillis(); for (ChaosRule rule : rules) { if (rule.live(now) && rule.matchesId(nodeId) && rule.fires(now)) { - rule.fired(Hook.STEP); + rule.fired(Hook.STEP, exchange, null, nodeId); if (rule.fault == Fault.DELAY) { return rule.duration; } diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosTargets.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosTargets.java index 8dead19bb74a..458e0f7dc89b 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosTargets.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosTargets.java @@ -120,7 +120,7 @@ final class ChaosTargets { // the endpoints are named as Camel shows them elsewhere (a pattern matches them with or without the // after the // scheme), and without their query, which can hold secrets - private static String withoutQuery(String uri) { + static String withoutQuery(String uri) { int query = uri.indexOf('?'); return URISupport.sanitizeUri(query > 0 ? uri.substring(0, query) : uri); } diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosLifecycleTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosLifecycleTest.java index 5afc42768f0c..3f7b5ef8985e 100644 --- a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosLifecycleTest.java +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosLifecycleTest.java @@ -22,6 +22,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.camel.Exchange; import org.apache.camel.Route; import org.apache.camel.ServiceStatus; import org.apache.camel.builder.RouteBuilder; @@ -54,6 +55,10 @@ class ChaosLifecycleTest extends ChaosTestSupport { public void configure() { from("direct:a").routeId("a").to("mock:a"); from("direct:b").routeId("b").to("mock:b"); + // a route that sends to route a, and to its dead letter channel when it cannot + from("direct:orders").routeId("orders") + .errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2).redeliveryDelay(0)) + .to("direct:a?block=false"); routeTemplate("tpl").templateParameter("name").from("direct:{{name}}").to("mock:tpl"); templatedRoute("tpl").routeId("t").parameter("name", "t"); // an exchange in flight that the test lets finish: a stop of the route waits for it @@ -158,7 +163,34 @@ class ChaosLifecycleTest extends ChaosTestSupport { await().atMost(10, TimeUnit.SECONDS).until(() -> status("a") == ServiceStatus.Stopped); awaitLifecycle(id, "back"); assertThat(status("a")).isEqualTo(ServiceStatus.Started); - assertThat(chaos().report().toText()).contains("stopped route a at +").contains(", started at +"); + assertThat(chaos().report().toText()).contains("stopped route a at +").contains(", started at +") + .contains(" ago " + id + " stop 2s of route a"); + assertThat(ruleReport(id).containsKey("lastFiredAgo")).isTrue(); + } + + @Test + void theExchangesSentToTheRouteWhileItIsDownAreTheEffectOfTheRule() throws Exception { + MockEndpoint dead = getMockEndpoint("mock:dead"); + dead.expectedMessageCount(2); + String id = chaos().addRule("route=a lifecycle action=stop duration=10m"); + awaitLifecycle(id, "down"); + template.sendBody("direct:orders", "x"); + template.sendBody("direct:orders", "y"); + + dead.assertIsSatisfied(); + // the redelivery was too short for the stop, so they went to the dead letter channel, which tells why + assertThat(dead.getExchanges()).allSatisfy(e -> assertThat(e.getProperty(ChaosManager.CHAOS_RULE)).isEqualTo(id)); + JsonObject effect = (JsonObject) ruleReport(id).get("effect"); + assertThat(effect.getLong("faulted")).isEqualTo(2); + assertThat(effect.getLong("handled")).isEqualTo(2); + assertThat(chaos().report().toText()).contains("effect on 2 exchanges: 2 handled by the error handler"); + + // once the route is back, the exchanges sent to it are not the effect of the rule + chaos().removeRule(id); + awaitLifecycle(id, "back"); + Exchange after = template.send("direct:orders", e -> e.getMessage().setBody("z")); + assertThat(after.getProperty(ChaosManager.CHAOS_RULE)).isNull(); + assertThat(((JsonObject) ruleReport(id).get("effect")).getLong("faulted")).isEqualTo(2); } @Test diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosLiveStateTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosLiveStateTest.java new file mode 100644 index 000000000000..b828338dc85c --- /dev/null +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosLiveStateTest.java @@ -0,0 +1,271 @@ +/* + * 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.component.chaos; + +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.util.json.JsonArray; +import org.apache.camel.util.json.JsonObject; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Isolated; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * The live state of the rules, for the chaos screen: the observed rate next to the rate of the rule, when it last fired + * and its faults per second, the recent faults, the exchange marked with the rule, and the effect of the faults on the + * exchanges. The time runs on a virtual clock (it is static, so this class runs alone). + */ +@Isolated +class ChaosLiveStateTest extends ChaosTestSupport { + + private final CountDownLatch release = new CountDownLatch(1); + private FakeClock clock; + + @Override + protected void setupResources() throws Exception { + super.setupResources(); + clock = FakeClock.install(); + } + + @Override + protected void cleanupResources() throws Exception { + release.countDown(); + clock.close(); + super.cleanupResources(); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:orders").routeId("orders") + .errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2).redeliveryDelay(0)) + .to("mock:stock").id("callStock"); + from("direct:plain").routeId("plain").errorHandler(noErrorHandler()).to("mock:stock"); + // a dead letter channel to a seda queue, which hands the completions over to its copy + from("direct:queued").routeId("queued") + .errorHandler(deadLetterChannel("seda:dlq").maximumRedeliveries(1).redeliveryDelay(0)) + .to("mock:stock"); + from("seda:dlq?pollTimeout=10").routeId("dlq").to("mock:dlq"); + from("direct:fast").routeId("fast").to("mock:fast?retainFirst=5"); + from("direct:slow").routeId("slow").to("mock:slow").process(e -> release.await(10, TimeUnit.SECONDS)); + } + }; + } + + private void send(String uri, int n) { + for (int i = 0; i < n; i++) { + template.send(uri, e -> e.getMessage().setBody("order")); + } + } + + private JsonObject effect(String id) { + return (JsonObject) ruleReport(id).get("effect"); + } + + private JsonArray recentFaults() { + return (JsonArray) chaos().report().toJson().get("recentFaults"); + } + + @Test + void theObservedRateIsNextToTheRateOfTheRule() { + String id = chaos().addRule("endpoint=mock:fast* drop rate=every:4"); + send("direct:fast", 10); + + JsonObject rule = ruleReport(id); + assertThat(rule.getDouble("rate")).isEqualTo(0.25d); + assertThat(rule.getDouble("observedRate")).isEqualTo(0.2d); + assertThat(rule.getLong("nextFireIn")).isEqualTo(2); + assertThat(rule.getLong("lastFiredAgo")).isZero(); + assertThat(rule.getLong("faultsLastMinute")).isEqualTo(2); + List<Object> perSecond = rule.getCollection("faultsPerSecond"); + assertThat(perSecond).hasSize(60).last().isEqualTo(2L); + + clock.advance(3_000); + rule = ruleReport(id); + assertThat(rule.getLong("lastFiredAgo")).isEqualTo(3_000); + // the second of the faults moved 3 seconds back in the minute + List<Object> moved = rule.getCollection("faultsPerSecond"); + assertThat(moved).element(56).isEqualTo(2L); + assertThat(chaos().report().toText()) + .contains("rate 20% of 1 in 4, fires again in 2 matches, last fired 3s ago, 2 faults in the last minute"); + + // a minute later the faults are out of the last minute + clock.advance(60_000); + rule = ruleReport(id); + assertThat(rule.getLong("faultsLastMinute")).isZero(); + assertThat(rule.containsKey("faultsPerSecond")).isFalse(); + assertThat(rule.getLong("lastFiredAgo")).isEqualTo(63_000); + } + + @Test + void aRuleWithoutRateFiresAtAllItsMatches() { + String id = chaos().addRule("endpoint=mock:fast* drop until=3"); + assertThat(ruleReport(id).getDouble("rate")).isEqualTo(1.0d); + // no match yet: no observed rate, and no fault + assertThat(ruleReport(id).containsKey("observedRate")).isFalse(); + assertThat(ruleReport(id).containsKey("lastFiredAgo")).isFalse(); + assertThat(ruleReport(id).containsKey("nextFireIn")).isFalse(); + + send("direct:fast", 5); + assertThat(ruleReport(id).getDouble("observedRate")).isEqualTo(1.0d); + // the rate is not news for such a rule + assertThat(chaos().report().toText()).doesNotContain("rate 100%") + .contains("last fired 0ms ago, 3 faults in the last minute"); + } + + @Test + void aRedeliveryThatSucceedsRecoversTheExchange() { + String id = chaos().addRule("endpoint=mock:stock error until=1"); + send("direct:orders", 1); + + JsonObject effect = effect(id); + assertThat(effect.getLong("faulted")).isEqualTo(1); + assertThat(effect.getLong("recovered")).isEqualTo(1); + assertThat(effect.getLong("handled")).isZero(); + assertThat(effect.getLong("inFlight")).isZero(); + assertThat(chaos().report().toText()).contains("effect on 1 exchange: 1 recovered by redelivery"); + } + + @Test + void theDeadLetterChannelHandlesTheFaultedExchangesAndSeesTheRule() throws Exception { + MockEndpoint dead = getMockEndpoint("mock:dead"); + dead.expectedMessageCount(2); + String id = chaos().addRule("endpoint=mock:stock error"); + send("direct:orders", 2); + // an exchange of a route without an error handler fails + Exchange failed = template.send("direct:plain", e -> e.getMessage().setBody("order")); + + dead.assertIsSatisfied(); + // the exchange tells which rule faulted it, so a dead letter route (or the tracer) can tell + assertThat(dead.getExchanges()).allSatisfy(e -> assertThat(e.getProperty(ChaosManager.CHAOS_RULE)).isEqualTo(id)); + assertThat(failed.getProperty(ChaosManager.CHAOS_RULE)).isEqualTo(id); + + JsonObject effect = effect(id); + // faulted three times each (and two redeliveries), but counted once per exchange + assertThat(effect.getLong("faulted")).isEqualTo(3); + assertThat(effect.getLong("handled")).isEqualTo(2); + assertThat(effect.getLong("failed")).isEqualTo(1); + assertThat(effect.getLong("recovered")).isZero(); + assertThat(chaos().report().toText()) + .contains("effect on 3 exchanges: 2 handled by the error handler, 1 failed"); + } + + @Test + void theEffectIsTheOneOfTheExchangeNotOfTheCopyOfASedaDeadLetterChannel() throws Exception { + MockEndpoint dlq = getMockEndpoint("mock:dlq"); + dlq.expectedMessageCount(2); + String id = chaos().addRule("endpoint=mock:stock error"); + send("direct:queued", 2); + + dlq.assertIsSatisfied(); + // the copy in the queue keeps the property + assertThat(dlq.getExchanges()).allSatisfy(e -> assertThat(e.getProperty(ChaosManager.CHAOS_RULE)).isEqualTo(id)); + JsonObject effect = effect(id); + assertThat(effect.getLong("handled")).isEqualTo(2); + assertThat(effect.getLong("recovered")).isZero(); + assertThat(effect.getLong("completed")).isZero(); + } + + @Test + void aDelayedExchangeCompletes() { + String id = chaos().addRule("endpoint=mock:fast* delay duration=1ms"); + send("direct:fast", 2); + + assertThat(effect(id).getLong("completed")).isEqualTo(2); + assertThat(getMockEndpoint("mock:fast").getReceivedExchanges()) + .allSatisfy(e -> assertThat(e.getProperty(ChaosManager.CHAOS_RULE)).isEqualTo(id)); + } + + @Test + void anExchangeThatIsNotDoneIsInFlight() { + String id = chaos().addRule("endpoint=mock:slow delay duration=1ms"); + CompletableFuture<Exchange> future = template.asyncSend("direct:slow", e -> e.getMessage().setBody("order")); + + await().atMost(10, TimeUnit.SECONDS).until(() -> effect(id) != null && effect(id).getLong("inFlight") == 1); + assertThat(effect(id).getLong("completed")).isZero(); + assertThat(chaos().report().toText()).contains("effect on 1 exchange: 1 in flight"); + + release.countDown(); + future.join(); + await().atMost(10, TimeUnit.SECONDS).until(() -> effect(id).getLong("completed") == 1); + assertThat(effect(id).getLong("inFlight")).isZero(); + } + + @Test + void theRecentFaultsAreTheLatestFirst() { + String id = chaos().addRule("endpoint=mock:fast* drop"); + Exchange last = null; + for (int i = 0; i < 25; i++) { + last = template.send("direct:fast", e -> e.getMessage().setBody("order")); + } + + JsonArray recent = recentFaults(); + assertThat(recent).hasSize(ChaosRecentFaults.SIZE); + JsonObject latest = (JsonObject) recent.get(0); + assertThat(latest.getString("rule")).isEqualTo(id); + assertThat(latest.getInteger("version")).isEqualTo(1); + assertThat(latest.getString("fault")).isEqualTo("drop"); + assertThat(latest.getString("hook")).isEqualTo("send"); + assertThat(latest.getString("route")).isEqualTo("fast"); + // without the query of the endpoint, which can hold secrets + assertThat(latest.getString("endpoint")).isEqualTo("mock://fast"); + assertThat(latest.getString("exchangeId")).isEqualTo(last.getExchangeId()); + assertThat(latest.getLong("timestamp")).isEqualTo(clock.currentTimeMillis()); + assertThat(latest.getLong("ago")).isZero(); + assertThat(chaos().report().toText()).contains("recent faults:\n 0ms ago " + id + + " drop at send mock://fast, route fast, exchange " + + last.getExchangeId()); + + // a reset starts a new report, without the faults + chaos().reset(); + assertThat(recentFaults()).isEmpty(); + assertThat(chaos().report().toText()).doesNotContain("recent faults"); + } + + @Test + void aFaultAtAStepOrAtTheEntryOfARoute() { + String step = chaos().addRule("step=callStock error until=1"); + Exchange exchange = template.send("direct:orders", e -> e.getMessage().setBody("order")); + + JsonObject fault = (JsonObject) recentFaults().get(0); + assertThat(fault.getString("rule")).isEqualTo(step); + assertThat(fault.getString("hook")).isEqualTo("step"); + assertThat(fault.getString("step")).isEqualTo("callStock"); + assertThat(fault.getString("route")).isEqualTo("orders"); + assertThat(fault.getString("fault")).isEqualTo("error ChaosException"); + assertThat(fault.containsKey("endpoint")).isFalse(); + assertThat(fault.getString("exchangeId")).isEqualTo(exchange.getExchangeId()); + + String entry = chaos().addRule("route=fast corrupt mode=headers"); + template.send("direct:fast", e -> e.getMessage().setHeader("country", "DK")); + fault = (JsonObject) recentFaults().get(0); + assertThat(fault.getString("rule")).isEqualTo(entry); + assertThat(fault.getString("hook")).isEqualTo("entry"); + assertThat(fault.getString("route")).isEqualTo("fast"); + assertThat(chaos().report().toText()).contains(entry + " corrupt headers at the entry of route fast, exchange "); + } +} diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosPollFaultsTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosPollFaultsTest.java index e129790b2bdb..56f16bd84e83 100644 --- a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosPollFaultsTest.java +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosPollFaultsTest.java @@ -32,6 +32,8 @@ import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.chaos.TestPollComponent.TestPollConsumer; import org.apache.camel.spi.PollingConsumerPollStrategy; import org.apache.camel.support.ScheduledPollConsumer; +import org.apache.camel.util.json.JsonArray; +import org.apache.camel.util.json.JsonObject; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -94,6 +96,12 @@ class ChaosPollFaultsTest extends ChaosTestSupport { int polls = a.polls.get(); int begins = strategy.begins.get(); await().atMost(10, TimeUnit.SECONDS).until(() -> fired(id, "poll") >= 6); + // a fault at a poll has no exchange: the route and the endpoint, without its query + JsonObject fault = (JsonObject) ((JsonArray) chaos().report().toJson().get("recentFaults")).get(0); + assertThat(fault.getString("hook")).isEqualTo("poll"); + assertThat(fault.getString("route")).isEqualTo("a"); + assertThat(fault.getString("endpoint")).isEqualTo("testpoll://a"); + assertThat(fault.containsKey("exchangeId")).isFalse(); // a poll that began before the rule was added may still finish assertThat(a.polls.get()).isLessThanOrEqualTo(polls + 1); assertThat(strategy.begins.get()).isLessThanOrEqualTo(begins + 1); diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleTest.java index 20a1ab0e7305..223b6035579e 100644 --- a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleTest.java +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleTest.java @@ -184,4 +184,24 @@ class ChaosRuleTest { assertThat(rule("endpoint=x error until=1m expect=o:1 within=2m", "r1", 1).expectation.deadline()) .isEqualTo(T0 + 120_000L); } + + @Test + void theFaultInAFewWords() { + assertThat(rule("endpoint=x error", "r1", 1).faultText()).isEqualTo("error ChaosException"); + assertThat(rule("endpoint=x error exception=java.io.IOException", "r1", 1).faultText()) + .isEqualTo("error IOException"); + assertThat(rule("endpoint=http:x* error status=503", "r1", 1).faultText()).isEqualTo("error status 503"); + assertThat(rule("endpoint=x delay duration=2s", "r1", 1).faultText()).isEqualTo("delay 2s"); + assertThat(rule("endpoint=x drop", "r1", 1).faultText()).isEqualTo("drop"); + assertThat(rule("route=orders corrupt mode=headers", "r1", 1).faultText()).isEqualTo("corrupt headers"); + assertThat(rule("route=orders stall duration=20s", "r1", 1).faultText()).isEqualTo("stall 20s"); + assertThat(rule("route=orders lifecycle action=stop duration=10s", "r1", 1).faultText()).isEqualTo("stop 10s"); + } + + @Test + void theFaultsPerSecondOverTheLastMinute() { + ChaosRule r = rule("endpoint=x error", "r1", 1); + assertThat(r.faultsPerSecond(T0)).hasSize(ChaosRule.MINUTE).containsOnly(0L); + assertThat(r.lastFiredAt()).isEqualTo(-1); + } } diff --git a/docs/user-manual/modules/ROOT/pages/camel-jbang-mcp.adoc b/docs/user-manual/modules/ROOT/pages/camel-jbang-mcp.adoc index de06851ac263..161caeec6b06 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-jbang-mcp.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-jbang-mcp.adoc @@ -1069,13 +1069,15 @@ that changes rules is refused. The list and the report read the chaos console fr | `camel_runtime_chaos_list` | `true` | `false` -| The current rules: id, version, the rule, state, where it fires, matches, faults fired, time left and the - expectation; or why chaos is off. Read from the status, which the application writes every second. +| The current rules: id, version, the rule, state, where it fires, matches, faults fired, time left, the + expectation and the live state (observed rate next to the rate of the rule, when it last fired, faults per second over + the last minute, and the effect of its faults on the exchanges); or why chaos is off. Read from the status, which the + application writes every second. | `camel_runtime_chaos_report` | `true` | `false` -| The report: the verdict and whether it is final, the seed, every version of every rule, the expectations and the - counts per route. +| The report: the verdict and whether it is final, the seed, every version of every rule with its live state, the + expectations, the counts per route and the last 20 faults. | `camel_runtime_chaos_add` | `false` | `true` diff --git a/docs/user-manual/modules/ROOT/pages/jbang-commands/camel-jbang-get-chaos.adoc b/docs/user-manual/modules/ROOT/pages/jbang-commands/camel-jbang-get-chaos.adoc index d1dcda915185..57ba4786873e 100644 --- a/docs/user-manual/modules/ROOT/pages/jbang-commands/camel-jbang-get-chaos.adoc +++ b/docs/user-manual/modules/ROOT/pages/jbang-commands/camel-jbang-get-chaos.adoc @@ -21,7 +21,7 @@ camel get chaos [options] | Option | Description | Default | Type | `--duration` | How long to keep watching or following, in millis or as a duration such as 30s (default: until Enter is pressed) | | long | `--json` | Output in JSON Format | | boolean -| `--report` | Show the report: verdict, seed, all rules, the counts per route | | boolean +| `--report` | Show the report: verdict, seed, all rules, the counts per route, the recent faults | | boolean | `--sort` | Sort by pid, name or age | pid | String | `--watch` | Execute periodically and showing output fullscreen | | boolean | `-h,--help` | Display the help and sub-commands | | boolean diff --git a/dsl/camel-jbang/camel-jbang-core/src/generated/resources/META-INF/camel-jbang-commands-metadata.json b/dsl/camel-jbang/camel-jbang-core/src/generated/resources/META-INF/camel-jbang-commands-metadata.json index 88361384e3ea..3889a13323bf 100644 --- a/dsl/camel-jbang/camel-jbang-core/src/generated/resources/META-INF/camel-jbang-commands-metadata.json +++ b/dsl/camel-jbang/camel-jbang-core/src/generated/resources/META-INF/camel-jbang-commands-metadata.json @@ -15,7 +15,7 @@ { "name": "eval", "fullName": "eval", "description": "Evaluate Camel expressions and scripts", "sourceClass": "org.apache.camel.dsl.jbang.core.commands.EvalCommand", "options": [ { "names": "-h,--help", "description": "Display the help and sub-commands", "javaType": "boolean", "type": "boolean" } ], "subcommands": [ { "name": "expression", "fullName": "eval expression", "description": "Evaluates Camel expression", "sourceClass": "org.apache.camel.dsl.jbang.core.commands.action.EvalEx [...] { "name": "explain", "fullName": "explain", "description": "Explain what a Camel route does using AI\/LLM", "sourceClass": "org.apache.camel.dsl.jbang.core.commands.Explain", "options": [ { "names": "--api-key", "description": "API key for authentication. Also reads ANTHROPIC_API_KEY, OPENAI_API_KEY, WATSONX_APIKEY, or LLM_API_KEY env vars", "javaType": "java.lang.String", "type": "string" }, { "names": "--api-type", "description": "API type: 'ollama', 'openai' (OpenAI-compatible), ' [...] { "name": "export", "fullName": "export", "description": "Export to other runtimes (Camel Main, Spring Boot, or Quarkus)", "sourceClass": "org.apache.camel.dsl.jbang.core.commands.Export", "options": [ { "names": "--build-property", "description": "Maven build properties, ex. --build-property=prop1=foo", "javaType": "java.util.List", "type": "array" }, { "names": "--camel-spring-boot-version", "description": "Camel version to use with Spring Boot", "javaType": "java.lang.String", "ty [...] - { "name": "get", "fullName": "get", "description": "Get status of Camel integrations", "sourceClass": "org.apache.camel.dsl.jbang.core.commands.process.CamelStatus", "options": [ { "names": "--watch", "description": "Execute periodically and showing output fullscreen", "javaType": "boolean", "type": "boolean" }, { "names": "-h,--help", "description": "Display the help and sub-commands", "javaType": "boolean", "type": "boolean" } ], "subcommands": [ { "name": "activity", "fullName": " [...] + { "name": "get", "fullName": "get", "description": "Get status of Camel integrations", "sourceClass": "org.apache.camel.dsl.jbang.core.commands.process.CamelStatus", "options": [ { "names": "--watch", "description": "Execute periodically and showing output fullscreen", "javaType": "boolean", "type": "boolean" }, { "names": "-h,--help", "description": "Display the help and sub-commands", "javaType": "boolean", "type": "boolean" } ], "subcommands": [ { "name": "activity", "fullName": " [...] { "name": "harden", "fullName": "harden", "description": "Suggest security hardening for Camel routes using AI\/LLM", "sourceClass": "org.apache.camel.dsl.jbang.core.commands.Harden", "options": [ { "names": "--api-key", "description": "API key for authentication. Also reads OPENAI_API_KEY or LLM_API_KEY env vars", "javaType": "java.lang.String", "type": "string" }, { "names": "--api-type", "description": "API type: 'ollama' or 'openai' (OpenAI-compatible)", "defaultValue": "ollama", [...] { "name": "hawtio", "fullName": "hawtio", "description": "Launch Hawtio web console", "sourceClass": "org.apache.camel.dsl.jbang.core.commands.process.Hawtio", "options": [ { "names": "--host", "description": "Hostname to bind the Hawtio web console to", "defaultValue": "127.0.0.1", "javaType": "java.lang.String", "type": "string" }, { "names": "--open-url,--openUrl", "description": "To automatic open Hawtio web console in the web browser (without an interactive terminal the default [...] { "name": "infra", "fullName": "infra", "description": "List and Run external services for testing and prototyping", "sourceClass": "org.apache.camel.dsl.jbang.core.commands.infra.InfraCommand", "options": [ { "names": "--json", "description": "Output in JSON Format", "javaType": "boolean", "type": "boolean" }, { "names": "-h,--help", "description": "Display the help and sub-commands", "javaType": "boolean", "type": "boolean" } ], "subcommands": [ { "name": "get", "fullName": "infra [...] diff --git a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistry.java b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistry.java index 541cc2a1c106..fc5a50f7a967 100644 --- a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistry.java +++ b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistry.java @@ -712,8 +712,11 @@ public final class ToolRegistry { private static void registerChaosTools() { register(tool("chaos_list", "List the chaos rules of the integration (camel-chaos, development only): id, version, the rule, state, " - + "where it fires, matches, faults fired, time left and the expectation; or why " - + "chaos is off.") + + "where it fires, matches, faults fired, time left, the expectation, and the live " + + "state: observed rate next to the rate of the rule, when it last fired, faults " + + "per second over the last minute, and the effect of its faults (exchanges " + + "completed, recovered by redelivery, handled by the error handler, failed, in " + + "flight); or why chaos is off.") .param("name", "string", CHAOS_NAME, false) .executor((ctx, args) -> { selectNamedProcess(ctx, args); @@ -722,8 +725,9 @@ public final class ToolRegistry { register(tool("chaos_report", "Get the chaos report of the integration (camel-chaos): the verdict (PASS or FAIL) and whether it is " - + "final, the seed, every version of every rule with what it fired, the " - + "expectations and the counts per route; or why chaos is off.") + + "final, the seed, every version of every rule with what it fired and its live " + + "state, the expectations, the counts per route and the last 20 faults (rule, " + + "fault, route, endpoint or step, exchange id); or why chaos is off.") .param("name", "string", CHAOS_NAME, false) .executor((ctx, args) -> { selectNamedProcess(ctx, args); diff --git a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/process/ListChaos.java b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/process/ListChaos.java index 5d01776fc7c0..df4306985589 100644 --- a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/process/ListChaos.java +++ b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/process/ListChaos.java @@ -41,8 +41,9 @@ import picocli.CommandLine.Command; showDefaultValues = true, footer = { "%nLists the current rules of the integrations that have camel-chaos switched on: id, version, state,", - "matches and the faults that fired. With --report it shows the report instead: the verdict, the seed,", - "all rules (also the removed and replaced versions), the counts per route and the reasons of a FAIL.", + "matches, the faults that fired, the observed rate, when a rule last fired and the effect of its faults", + "on the exchanges. With --report it shows the report instead: the verdict, the seed, all rules (also", + "the removed and replaced versions), the counts per route, the recent faults and the reasons of a FAIL.", "%nExamples:", " camel get chaos", " camel get chaos myApp --report", @@ -51,11 +52,16 @@ public class ListChaos extends ProcessWatchCommand { private static final RowSort<Row> ORDER = RowSort.<Row> integration(r -> r.pid, r -> r.name, r -> r.uptime); + // the keys of a rule with its live state, besides its counts, that --json copies when the rule has them + private static final List<String> LIVE_KEYS = List.of( + "lostAtDlq", "rate", "observedRate", "nextFireIn", "lastFiredAgo", "faultsLastMinute", "faultsPerSecond", + "effect"); + @CommandLine.Parameters(description = "Name or pid of running Camel integration", arity = "0..1") String name = "*"; @CommandLine.Option(names = { "--report" }, - description = "Show the report: verdict, seed, all rules, the counts per route") + description = "Show the report: verdict, seed, all rules, the counts per route, the recent faults") boolean report; @CommandLine.Option(names = { "--sort" }, completionCandidates = PidNameAgeCompletionCandidates.class, @@ -133,6 +139,12 @@ public class ListChaos extends ProcessWatchCommand { if (r.rule.get("skippedReasons") != null) { jo.put("skippedReasons", r.rule.get("skippedReasons")); } + // the live state of the rule + for (String key : LIVE_KEYS) { + if (r.rule.get(key) != null) { + jo.put(key, r.rule.get(key)); + } + } return jo; }).toList())); return; @@ -142,6 +154,8 @@ public class ListChaos extends ProcessWatchCommand { return; } boolean notes = rows.stream().anyMatch(r -> !note(r.rule).isEmpty()); + boolean rates = rows.stream().anyMatch(r -> !rate(r.rule).isEmpty()); + boolean effects = rows.stream().anyMatch(r -> !effect(r.rule).isEmpty()); printer().println(AsciiTable.getTable(AsciiTable.NO_BORDERS, rows, List.of( new Column().header("PID").headerAlign(HorizontalAlign.CENTER).with(r -> r.integration.pid), new Column().header("NAME").dataAlign(HorizontalAlign.LEFT).maxWidth(30, OverflowBehaviour.ELLIPSIS_RIGHT) @@ -156,8 +170,14 @@ public class ListChaos extends ProcessWatchCommand { .with(r -> Long.toString(fired(r.rule))), new Column().header("SKIPPED").headerAlign(HorizontalAlign.RIGHT).dataAlign(HorizontalAlign.RIGHT) .with(r -> String.valueOf(r.rule.get("skipped"))), + new Column().header("RATE").visible(rates).headerAlign(HorizontalAlign.RIGHT) + .dataAlign(HorizontalAlign.RIGHT).with(r -> rate(r.rule)), + new Column().header("LAST").headerAlign(HorizontalAlign.RIGHT).dataAlign(HorizontalAlign.RIGHT) + .with(r -> lastFired(r.rule)), new Column().header("ENDS IN").headerAlign(HorizontalAlign.RIGHT).dataAlign(HorizontalAlign.RIGHT) .with(r -> endsIn(r.rule)), + new Column().header("EFFECT").visible(effects).dataAlign(HorizontalAlign.LEFT) + .maxWidth(40, OverflowBehaviour.NEWLINE).with(r -> effect(r.rule)), new Column().header("NOTE").visible(notes).dataAlign(HorizontalAlign.LEFT) .maxWidth(40, OverflowBehaviour.NEWLINE).with(r -> note(r.rule)), new Column().header("RULE").dataAlign(HorizontalAlign.LEFT).maxWidth(80, OverflowBehaviour.NEWLINE) @@ -217,6 +237,7 @@ public class ListChaos extends ProcessWatchCommand { if (rules != null && !rules.isEmpty()) { List<JsonObject> list = rules.stream().map(JsonObject.class::cast).toList(); boolean notes = list.stream().anyMatch(r -> !note(r).isEmpty()); + boolean effects = list.stream().anyMatch(r -> !effect(r).isEmpty()); printer().println(AsciiTable.getTable(AsciiTable.NO_BORDERS, list, List.of( new Column().header("ID").dataAlign(HorizontalAlign.LEFT).with(r -> r.getString("id")), new Column().header("VER").headerAlign(HorizontalAlign.RIGHT).dataAlign(HorizontalAlign.RIGHT) @@ -229,6 +250,8 @@ public class ListChaos extends ProcessWatchCommand { new Column().header("SKIPPED").headerAlign(HorizontalAlign.RIGHT).dataAlign(HorizontalAlign.RIGHT) .with(r -> String.valueOf(r.get("skipped"))), new Column().header("EXPECT").dataAlign(HorizontalAlign.LEFT).with(ListChaos::expectation), + new Column().header("EFFECT").visible(effects).dataAlign(HorizontalAlign.LEFT) + .maxWidth(40, OverflowBehaviour.NEWLINE).with(ListChaos::effect), new Column().header("NOTE").visible(notes).dataAlign(HorizontalAlign.LEFT) .maxWidth(40, OverflowBehaviour.NEWLINE).with(ListChaos::note), new Column().header("RULE").dataAlign(HorizontalAlign.LEFT).maxWidth(80, OverflowBehaviour.NEWLINE) @@ -251,6 +274,25 @@ public class ListChaos extends ProcessWatchCommand { .dataAlign(HorizontalAlign.RIGHT) .with(r -> String.valueOf(r.get("redeliveries")))))); } + JsonArray recent = report.getCollection("recentFaults"); + if (recent != null && !recent.isEmpty()) { + List<JsonObject> list = recent.stream().map(JsonObject.class::cast).toList(); + printer().println(" recent faults:"); + printer().println(AsciiTable.getTable(AsciiTable.NO_BORDERS, list, List.of( + new Column().header("AGO").headerAlign(HorizontalAlign.RIGHT).dataAlign(HorizontalAlign.RIGHT) + .with(r -> r.get("ago") instanceof Number n ? TimeUtils.printDuration(n.longValue()) : ""), + new Column().header("RULE").dataAlign(HorizontalAlign.LEFT).with(r -> r.getString("rule")), + new Column().header("FAULT").dataAlign(HorizontalAlign.LEFT).with(r -> r.getString("fault")), + new Column().header("HOOK").dataAlign(HorizontalAlign.LEFT) + .with(r -> r.getStringOrDefault("hook", "")), + new Column().header("ROUTE").dataAlign(HorizontalAlign.LEFT) + .with(r -> r.getStringOrDefault("route", "")), + new Column().header("AT").dataAlign(HorizontalAlign.LEFT) + .maxWidth(50, OverflowBehaviour.ELLIPSIS_RIGHT) + .with(r -> r.getStringOrDefault("endpoint", r.getStringOrDefault("step", ""))), + new Column().header("EXCHANGE").dataAlign(HorizontalAlign.LEFT) + .with(r -> r.getStringOrDefault("exchangeId", ""))))); + } } /** @@ -315,6 +357,43 @@ public class ListChaos extends ProcessWatchCommand { return total; } + /** The observed rate of a rule next to its own, such as {@code 4.8% of 5%}, for a rule with a rate below 100%. */ + static String rate(JsonObject rule) { + if (rule.get("rate") instanceof Number rate && rate.doubleValue() < 1.0d + && rule.get("observedRate") instanceof Number observed) { + return percent(observed.doubleValue()) + " of " + percent(rate.doubleValue()); + } + return ""; + } + + private static String percent(double share) { + double tenths = Math.round(share * 1000) / 10.0d; + return (tenths == Math.rint(tenths) ? Long.toString((long) tenths) : Double.toString(tenths)) + "%"; + } + + private static String lastFired(JsonObject rule) { + return rule.get("lastFiredAgo") instanceof Number n ? TimeUtils.printDuration(n.longValue()) + " ago" : ""; + } + + /** + * What the faults of a rule did to the exchanges, such as {@code 10 recovered, 2 handled, 1 in flight}; the counts + * that are 0 are left out. + */ + static String effect(JsonObject rule) { + if (!(rule.get("effect") instanceof JsonObject e)) { + return ""; + } + StringJoiner sj = new StringJoiner(", "); + for (String[] key : new String[][] { + { "completed", "completed" }, { "recovered", "recovered" }, { "handled", "handled" }, + { "failed", "failed" }, { "inFlight", "in flight" } }) { + if (e.get(key[0]) instanceof Number n && n.longValue() > 0) { + sj.add(n + " " + key[1]); + } + } + return sj.toString(); + } + private static String endsIn(JsonObject rule) { return rule.get("endsIn") instanceof Number n ? TimeUtils.printDuration(n.longValue()) : ""; } diff --git a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/process/ListChaosTest.java b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/process/ListChaosTest.java index 98866d1252d8..9dc0c46f7728 100644 --- a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/process/ListChaosTest.java +++ b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/process/ListChaosTest.java @@ -103,6 +103,55 @@ class ListChaosTest extends ProcessCommandTestSupport { assertThat(copied.getLong("downAt")).isEqualTo(4); } + @Test + void theLiveStateOfTheRules() throws Exception { + writeStatus(withLiveState(chaosOn())); + + assertThat(run()).isZero(); + + String out = printer.getStdout(); + assertThat(out).contains("RATE").contains("LAST").contains("EFFECT"); + assertThat(lineOf(out, "r1")).contains("40% of 50%").contains("3s ago") + .contains("2 recovered, 1 handled, 1 in flight"); + } + + @Test + void theLiveStateAsJson() throws Exception { + writeStatus(withLiveState(chaosOn())); + + assertThat(run("--json")).isZero(); + + JsonObject row = (JsonObject) ((JsonArray) Jsoner.deserialize(printer.getStdout())).get(0); + assertThat(row.getDouble("rate")).isEqualTo(0.5d); + assertThat(row.getDouble("observedRate")).isEqualTo(0.4d); + assertThat(row.getLong("lastFiredAgo")).isEqualTo(3000); + assertThat(row.getLong("faultsLastMinute")).isEqualTo(5); + assertThat(((JsonObject) row.get("effect")).getLong("inFlight")).isEqualTo(1); + } + + @Test + void theRecentFaultsInTheReport() throws Exception { + writeStatus(withLiveState(chaosOn())); + + assertThat(run("--report")).isZero(); + + String out = printer.getStdout(); + assertThat(out).contains("recent faults:").contains("FAULT").contains("EXCHANGE"); + assertThat(lineOf(out, "ABC-1")).contains("2s").contains("r1").contains("error ChaosException").contains("send") + .contains("orders").contains("kafka://orders"); + assertThat(lineOf(out, "r1 ")).contains("2 recovered, 1 handled, 1 in flight"); + } + + @Test + void withoutLiveStateTheColumnsAreLeftOut() throws Exception { + writeStatus(chaosOn()); + + assertThat(run()).isZero(); + + // a rule of an older camel-chaos has no live state + assertThat(printer.getStdout()).doesNotContain("RATE").doesNotContain("EFFECT"); + } + @Test void report() throws Exception { writeStatus(chaosOn()); @@ -313,6 +362,35 @@ class ListChaosTest extends ProcessCommandTestSupport { return chaos; } + /** The live state of rule r1, as camel-chaos adds it to its rules and its report. */ + private static JsonObject withLiveState(JsonObject chaos) { + JsonObject rule = (JsonObject) ((JsonArray) chaos.get("rules")).get(0); + rule.put("rate", 0.5d); + rule.put("observedRate", 0.4d); + rule.put("lastFiredAgo", 3000L); + rule.put("faultsLastMinute", 5L); + JsonObject effect = new JsonObject(); + effect.put("faulted", 4L); + effect.put("completed", 0L); + effect.put("recovered", 2L); + effect.put("handled", 1L); + effect.put("failed", 0L); + effect.put("inFlight", 1L); + rule.put("effect", effect); + JsonObject fault = new JsonObject(); + fault.put("timestamp", 1_000_000L); + fault.put("ago", 2000L); + fault.put("rule", "r1"); + fault.put("version", 2); + fault.put("fault", "error ChaosException"); + fault.put("hook", "send"); + fault.put("route", "orders"); + fault.put("endpoint", "kafka://orders"); + fault.put("exchangeId", "ABC-1"); + ((JsonObject) chaos.get("report")).put("recentFaults", new JsonArray(List.of(fault))); + return chaos; + } + private static JsonObject rule(int version, String state, String text) { JsonObject rule = new JsonObject(); rule.put("id", "r1"); diff --git a/dsl/camel-jbang/camel-jbang-core/src/test/resources/golden/cli-help.txt b/dsl/camel-jbang/camel-jbang-core/src/test/resources/golden/cli-help.txt index 99ea7b26cafa..1b415d7067fd 100644 --- a/dsl/camel-jbang/camel-jbang-core/src/test/resources/golden/cli-help.txt +++ b/dsl/camel-jbang/camel-jbang-core/src/test/resources/golden/cli-help.txt @@ -3441,15 +3441,17 @@ Get the chaos rules of running integrations, or their report (camel-chaos) duration such as 30s (default: until Enter is pressed) Default: 0 --report Show the report: verdict, seed, all rules, the counts per - route + route, the recent faults --sort=<sort> Sort by pid, name or age Default: pid Lists the current rules of the integrations that have camel-chaos switched on: id, version, state, -matches and the faults that fired. With --report it shows the report instead: -the verdict, the seed, -all rules (also the removed and replaced versions), the counts per route and +matches, the faults that fired, the observed rate, when a rule last fired and +the effect of its faults +on the exchanges. With --report it shows the report instead: the verdict, the +seed, all rules (also +the removed and replaced versions), the counts per route, the recent faults and the reasons of a FAIL. Examples: diff --git a/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java b/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java index e5b3c737a0a2..71267c7bdc74 100644 --- a/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java +++ b/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java @@ -1072,8 +1072,11 @@ public class RuntimeTools { @Tool(annotations = @Tool.Annotations(readOnlyHint = true, destructiveHint = false, openWorldHint = false), description = """ List the chaos rules of the Camel application (camel-chaos, development only): id, version, the \ - rule, state (active or ended), where it fires, matches, faults fired, time left and the expectation; \ - or why chaos is off. The rules come from the status of the application, written every second.""") + rule, state (active or ended), where it fires, matches, faults fired, time left, the expectation, \ + and the live state: observed rate next to the rate of the rule, when it last fired, faults per \ + second over the last minute, and the effect of its faults (exchanges completed, recovered by \ + redelivery, handled by the error handler, failed, in flight); or why chaos is off. The rules come \ + from the status of the application, written every second.""") public JsonObject camel_runtime_chaos_list( @ToolArg(description = NAME_OR_PID_DESC, required = false) String nameOrPid) { return delegateToRegistry("chaos_list", nameOrPid, Map.of()); @@ -1083,8 +1086,9 @@ public class RuntimeTools { @Tool(annotations = @Tool.Annotations(readOnlyHint = true, destructiveHint = false, openWorldHint = false), description = """ Get the chaos report of the Camel application: the verdict (PASS or FAIL) and whether it is final, \ - the seed, every version of every rule with what it fired, the expectations (a pending one is not \ - decided yet) and the counts per route; or why chaos is off.""") + the seed, every version of every rule with what it fired and its live state, the expectations (a \ + pending one is not decided yet), the counts per route and the last 20 faults (rule, fault, route, \ + endpoint or step, exchange id); or why chaos is off.""") public JsonObject camel_runtime_chaos_report( @ToolArg(description = NAME_OR_PID_DESC, required = false) String nameOrPid) { return delegateToRegistry("chaos_report", nameOrPid, Map.of());
