This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26718-chaos-in-route in repository https://gitbox.apache.org/repos/asf/camel.git
commit 5651a3a5ac3f8ad5e853ea9c0e9e2003bce639cc Author: Claus Ibsen <[email protected]> AuthorDate: Sat Nov 7 20:19:25 2026 +0100 CAMEL-26718: camel-chaos - in= narrows a rule to the sends and steps of a route, and a route can fail or be slow at its entry Step 3 of the stage 2 order decided on CAMEL-26607. in= (S2-2): endpoint=http:* in=orders error status=503, step=* in=orders delay duration=500ms. It is the route of the unit of work, not the routes it calls; a pattern takes more routes; with at=poll it is the route whose consumer polls. It is checked before the rate, so rate and an until count count only the matches in those routes. Not with route=, stall or lifecycle. Adding a rule warns when no route matches in=, or when the target matches nothing inside those routes, and the hooks that can fire look at those routes only. The rule in plain words says it, the chaos screen shows it with the target, and the chaos model has the option (pick: routes). Errors and delays at the route entry (S2-1): route=orders error rate=0.05, route=orders delay duration=2s. The route policy decides once per exchange as it enters the route, and the step hook fails or delays the exchange at the first step of the route, inside the error handler of the route, so onException, redelivery and the dead letter channel apply, and a redelivery of that step runs the step itself. No drop at the entry. New recipes: the route fails, the route is slow, the sends of the route fail. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../org/apache/camel/catalog/chaos-model.json | 163 ++++++++++++++++- .../org/apache/camel/catalog/docs/chaos.adoc | 55 +++++- .../org/apache/camel/chaos/chaos-model.json | 163 ++++++++++++++++- components/camel-chaos/src/main/docs/chaos.adoc | 55 +++++- .../camel/component/chaos/ChaosExchangeState.java | 4 + .../apache/camel/component/chaos/ChaosManager.java | 4 + .../apache/camel/component/chaos/ChaosModel.java | 74 +++++++- .../camel/component/chaos/ChaosPollStrategy.java | 3 +- .../component/chaos/ChaosRoutePolicyFactory.java | 19 +- .../apache/camel/component/chaos/ChaosRule.java | 29 ++- .../camel/component/chaos/ChaosRuleParser.java | 36 +++- .../camel/component/chaos/ChaosSendHook.java | 4 +- .../camel/component/chaos/ChaosStepHook.java | 46 ++++- .../apache/camel/component/chaos/ChaosTargets.java | 116 +++++++++++- .../camel/component/chaos/ChaosInRouteTest.java | 194 +++++++++++++++++++++ .../camel/component/chaos/ChaosModelTest.java | 7 +- .../component/chaos/ChaosRouteEntryErrorTest.java | 188 ++++++++++++++++++++ .../component/chaos/ChaosRuleDescribeTest.java | 6 + .../camel/component/chaos/ChaosRuleParserTest.java | 44 ++++- design/chaos.adoc | 8 +- .../core/commands/action/CamelChaosAddAction.java | 2 + .../src/test/resources/golden/cli-help.txt | 3 + .../dsl/jbang/core/commands/tui/ChaosInfo.java | 22 ++- .../dsl/jbang/core/commands/tui/ChaosTab.java | 5 +- .../src/main/resources/tui/help/chaos.md | 2 +- .../core/commands/tui/ChaosTabRenderTest.java | 11 ++ 26 files changed, 1187 insertions(+), 76 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/chaos-model.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/chaos-model.json index 34095d99a9b8..6c3a7a52f5e2 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/chaos-model.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/chaos-model.json @@ -5,7 +5,7 @@ "targets": [ { "name": "route", - "description": "A route id, with * for any characters; one route id (no pattern) for stall and lifecycle", + "description": "A route id, with * for any characters; one route id (no pattern) for stall and lifecycle. Its exchanges are faulted as they enter: error and delay at the first step of the route, where its error handler applies; to reach the sends and steps of a route, use in= with another target", "pattern": true, "pick": "routes" }, @@ -89,11 +89,13 @@ }, { "name": "entry", - "description": "Each exchange that enters a matching route, before the first step", + "description": "Each exchange that enters a matching route, once: corrupt changes it before the first step, error and delay act at the first step, inside the error handler of the route", "targets": [ "route" ], "faults": [ + "error", + "delay", "corrupt" ], "onlyWithAt": false @@ -118,6 +120,7 @@ "targets": [ "component", "endpoint", + "route", "step" ], "hooks": [ @@ -125,6 +128,7 @@ "reply", "dlq", "poll", + "entry", "step" ], "parameters": [ @@ -188,6 +192,7 @@ "at", "expect", "from", + "in", "rate", "until", "within" @@ -199,12 +204,14 @@ "targets": [ "component", "endpoint", + "route", "step" ], "hooks": [ "send", "dlq", "poll", + "entry", "step" ], "parameters": [ @@ -221,6 +228,7 @@ "at", "expect", "from", + "in", "rate", "until", "within" @@ -245,6 +253,7 @@ "at", "expect", "from", + "in", "rate", "until", "within" @@ -291,6 +300,7 @@ "at", "expect", "from", + "in", "rate", "until", "within" @@ -313,6 +323,7 @@ "at", "expect", "from", + "in", "rate", "until", "within" @@ -442,6 +453,15 @@ "description": "When the rule starts after it is added, such as after the warm-up (+30s or 30s)", "example": "+30s" }, + { + "name": "in", + "type": "string", + "required": false, + "description": "Narrows the sends, replies, polls and steps of the target to a route id, with * for any characters: the route of the exchange, not the routes it calls; with at=poll the route whose consumer polls. Not with route=, stall or lifecycle", + "example": "orders", + "pattern": true, + "pick": "routes" + }, { "name": "rate", "type": "rate", @@ -1214,6 +1234,145 @@ } ] }, + { + "id": "route-error", + "group": "Consumers and routes", + "title": "The route fails", + "description": "An exchange fails at the first step of the route, and the error handling of the route applies (onException, redelivery, the dead letter channel). When the route does not handle it, the consumer sees the failure: a JMS rollback, the file left in place, an HTTP 500", + "uncovers": [ + "A route without error handling, whose consumer then gets the failure: is that what the sender expects (a rollback, a retry, an HTTP 500)?", + "onException and redelivery that do not cover the route", + "Callers of the route that do not handle its failure" + ], + "avoid": [ + "To fail one call of the route: use the sends of the route (in=) or a step instead" + ], + "watch": [ + "The effect: recovered, handled and failed", + "What the consumer does with a failed exchange, such as a message going back to the broker" + ], + "rule": "route=${route} error rate=${rate} until=${until}", + "placeholders": [ + { + "name": "route", + "type": "route", + "description": "The route", + "example": "orders", + "pick": { + "from": "routes" + } + }, + { + "name": "rate", + "type": "rate", + "description": "How often: a probability, every:N or burst:<time>\/<period>", + "default": "0.05", + "example": "0.05" + }, + { + "name": "until", + "type": "countOrDuration", + "description": "When the rule ends: a number of matches or a time", + "default": "10m", + "example": "10m" + } + ] + }, + { + "id": "route-slow", + "group": "Consumers and routes", + "title": "The route is slow", + "description": "Each exchange waits before the first step of the route. It waits on the thread of the consumer, so the consumer takes fewer messages", + "uncovers": [ + "A consumer that falls behind, and the backlog in front of it", + "Timeouts of the callers of the route", + "How the route behaves with fewer messages per second" + ], + "avoid": [ + "On a route others call and wait for, unless their timeouts are the test" + ], + "watch": [ + "The mean and the slowest time of the route", + "The queue or the lag in front of the route" + ], + "rule": "route=${route} delay duration=${duration} rate=${rate} until=${until}", + "placeholders": [ + { + "name": "route", + "type": "route", + "description": "The route", + "example": "orders", + "pick": { + "from": "routes" + } + }, + { + "name": "duration", + "type": "duration", + "description": "The time, or a range of times", + "default": "500ms..2s", + "example": "500ms..2s" + }, + { + "name": "rate", + "type": "rate", + "description": "How often: a probability, every:N or burst:<time>\/<period>", + "default": "0.25", + "example": "0.25" + }, + { + "name": "until", + "type": "countOrDuration", + "description": "When the rule ends: a number of matches or a time", + "default": "10m", + "example": "10m" + } + ] + }, + { + "id": "route-sends-error", + "group": "Consumers and routes", + "title": "The sends of the route fail", + "description": "Each send of the route (to, enrich, wireTap...) fails with a ChaosException at the rate, whatever it sends to. The sends of the routes it calls are not in it", + "uncovers": [ + "Which calls of the route have no error handling of their own", + "A route that goes on after a failed send, with half of its work done", + "The error handler of the route against each kind of send" + ], + "avoid": [ + "When one call is the test: pick its endpoint instead of *" + ], + "watch": [ + "The effect: recovered, handled and failed", + "The sends of the route that failed, on the Routes tab" + ], + "rule": "endpoint=* in=${route} error at=send rate=${rate} until=${until}", + "placeholders": [ + { + "name": "route", + "type": "route", + "description": "The route", + "example": "orders", + "pick": { + "from": "routes" + } + }, + { + "name": "rate", + "type": "rate", + "description": "How often: a probability, every:N or burst:<time>\/<period>", + "default": "0.1", + "example": "0.1" + }, + { + "name": "until", + "type": "countOrDuration", + "description": "When the rule ends: a number of matches or a time", + "default": "10m", + "example": "10m" + } + ] + }, { "id": "dlq-unreachable", "group": "Error handling", 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 88733c7ea84d..1b51bbfc206e 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 @@ -171,8 +171,8 @@ only the list and the report. == Rules -A rule is one line: the target, the fault and its parameters as `name=value`, then `at`, `rate`, `from` and `until`, -and optionally an expectation. A value with spaces or a `;` is quoted with `"` or `'` (`\"`, `\'` and `\\` inside quotes +A rule is one line: the target, the fault and its parameters as `name=value`, then `at`, `in`, `rate`, `from` and +`until`, and optionally an expectation. A value with spaces or a `;` is quoted with `"` or `'` (`\"`, `\'` and `\\` inside quotes are a quote and a backslash). [source] @@ -180,6 +180,8 @@ are a quote and a backslash). target: route=<id> | endpoint=<uri> | component=<name> | step=<id> (* matches any characters) fault: error | delay | drop | corrupt | duplicate | stall | lifecycle, with its parameters at: send | reply | dlq | poll | entry | step (optional) +in: in=<route id> narrows the sends, polls and steps of endpoint=, component= and step= to a route + (* matches any characters; optional) rate: a probability (rate=0.3), every Nth match (rate=every:5), or bursts (rate=burst:5s/30s: every match during the first 5s of each 30s); without it every match fires from: when the rule starts after it is added (from=+30s), such as after the warm-up; without it at once @@ -205,6 +207,26 @@ the endpoint URI `netty-http://http://inventory/stock`, and a camel-netty-http c matched with `netty-http` in front too. A `route=` pattern is matched against the route id, and a `step=` pattern against the id of the step (`to1`, or the id set with `id("enrich")`). +=== The sends and steps of a route + +`in=` narrows the target of a rule to the routes whose id it matches, for the sends, steps and polls that you know by +their route rather than by an endpoint URI or a step id (which are often generated, such as `to3`): + +[source] +---- +endpoint=http:* in=orders error status=503 # the HTTP calls of route orders +step=* in=orders delay duration=500ms # every step of route orders +component=kafka in=order* drop # the Kafka sends of the routes whose id starts with order +endpoint=* in=orders error at=send rate=0.1 # 10% of all the sends of route orders +---- + +The route is the route of the exchange: a send in a route that `orders` calls with `direct:` is in that other route, +not in `orders`; take more routes with a pattern (`in=order*`). With `at=poll` it is the route whose consumer polls. +`in=` is checked before the rate, so `rate` and an `until` count count only the matches in those routes. A `route=` +rule is about its route already, and `stall` and `lifecycle` act on the route of their `route=`, so they take no `in=`. +When no route matches `in=`, or the target matches no send, poll or step inside those routes, adding the rule warns, +with what is there. + A rule gets an id when it is added (`r1`, `r2`, ...), which change and remove take. A change replaces the rule with a new one with the same id and the next version (a rule is version 1 when it is added), and the replaced one stays in the report: so the report has an entry per version of the id, and a tool tells them apart by the `version` of the entry @@ -239,7 +261,7 @@ target offers, is rejected when it is added, with an error that names the hooks |`route=` |The messages that enter the route (`entry`), and the route itself. -|`entry`: `corrupt`. The route itself: `stall`, `lifecycle` (they take no `at`). +|`entry`: `error`, `delay`, `corrupt`. The route itself: `stall`, `lifecycle` (they take no `at`). |`step=` |The matching steps (`step`). @@ -253,9 +275,10 @@ the hooks where a rule can fire in the integration now (`hooks`): of the hooks o JMS), `send` (and `reply`) only when something sends to a matching endpoint (a `to`, a `toD`, `wireTap`, `enrich`, `recipientList`, `routingSlip` or `dynamicRouter` whose URI may match, or an endpoint a template sent to), and any hook where the rule fired already. The rule still fires at all its hooks, so a send that comes later (such as from a -`ProducerTemplate` to the endpoint of a route) is faulted too, and its `fired` counts have every hook of the rule. `route=orders error` is rejected: the -entry of a route offers only `corrupt`, and `route=` does not reach the sends and steps of the route (they are -targeted with `endpoint=`, `component=` or `step=`). A rule that matches nothing, such as a poll rule for a consumer +`ProducerTemplate` to the endpoint of a route) is faulted too, and its `fired` counts have every hook of the rule. +`route=orders error rate=0.05` fails 5% of the exchanges that enter the route, each once (see the `entry` hook below); +the sends and steps of the route are targeted with `endpoint=`, `component=` or `step=`, narrowed to the route with +`in=orders`. A rule that matches nothing, such as a poll rule for a consumer that does not poll on a schedule (Kafka, JMS), fires 0 times, as the report shows. When several rules match the same send, poll, message or step, they are asked in the order they were added, and the @@ -392,9 +415,17 @@ stopped or suspended meanwhile. The consumers that `pollEnrich` and a `ConsumerT route, and chaos does not reach their polls. |`entry` -|Each exchange that enters a matching route (a route policy, `onExchangeBegin`), before the first step. There is no -`drop` at the entry: the exchange already belongs to its consumer, and dropping it would make the consumer commit a -message the route never processed. +|Each exchange that enters a matching route (a route policy, `onExchangeBegin`), once, also when it enters the route +again (a loop, a redelivery). A `corrupt` changes the message before the first step. An `error` or a `delay` fires +there, and acts at the first step of the route, through the step hook: inside the error handler of the route, so +`onException`, redelivery and the dead letter channel of the route apply, as for a real failure inside it, and a +redelivery of that step runs the step itself. When the route does not handle the error, its consumer gets the failed +exchange, as it would for any failure of the route: a JMS message is rolled back, a file stays where it was, an HTTP +client gets a 500. A `delay` waits on the thread of the consumer (for most consumers), so the consumer takes fewer +messages. This is the honest form of "fail 5% of the order route": `step=* rate=0.05` would compound the rate over +every step. There is no `drop` at the entry: the exchange already belongs to its consumer, and dropping it would make +the consumer commit a message the route never processed. A fault of the consumer itself, before there is an exchange, +is a poll fault (`at=poll`) for a scheduled poll consumer, or a route that is down (`stall`, `lifecycle`). |`step` |Each step of a route, through the step hook of camel-core, right before the processor of the step is called, once for @@ -609,6 +640,12 @@ seconds |`step=enrich* error exception=java.sql.SQLException expect=orders:handled<=0` |fail every run of step enrich* with SQLException; expects no exchanges handled by the error handler in route orders +|`endpoint=http:stock* in=orders error status=503 rate=0.5` +|fail half of the sends to http:stock* in route orders with HTTP status 503 + +|`route=orders error rate=0.05` +|fail 5% of the messages entering route orders + |`route=poller lifecycle action=remove duration=30s repeat=3 every=1m` |remove route poller for 30 seconds, then add it back, 3 times, every 1 minute |=== diff --git a/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/chaos/chaos-model.json b/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/chaos/chaos-model.json index 34095d99a9b8..6c3a7a52f5e2 100644 --- a/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/chaos/chaos-model.json +++ b/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/chaos/chaos-model.json @@ -5,7 +5,7 @@ "targets": [ { "name": "route", - "description": "A route id, with * for any characters; one route id (no pattern) for stall and lifecycle", + "description": "A route id, with * for any characters; one route id (no pattern) for stall and lifecycle. Its exchanges are faulted as they enter: error and delay at the first step of the route, where its error handler applies; to reach the sends and steps of a route, use in= with another target", "pattern": true, "pick": "routes" }, @@ -89,11 +89,13 @@ }, { "name": "entry", - "description": "Each exchange that enters a matching route, before the first step", + "description": "Each exchange that enters a matching route, once: corrupt changes it before the first step, error and delay act at the first step, inside the error handler of the route", "targets": [ "route" ], "faults": [ + "error", + "delay", "corrupt" ], "onlyWithAt": false @@ -118,6 +120,7 @@ "targets": [ "component", "endpoint", + "route", "step" ], "hooks": [ @@ -125,6 +128,7 @@ "reply", "dlq", "poll", + "entry", "step" ], "parameters": [ @@ -188,6 +192,7 @@ "at", "expect", "from", + "in", "rate", "until", "within" @@ -199,12 +204,14 @@ "targets": [ "component", "endpoint", + "route", "step" ], "hooks": [ "send", "dlq", "poll", + "entry", "step" ], "parameters": [ @@ -221,6 +228,7 @@ "at", "expect", "from", + "in", "rate", "until", "within" @@ -245,6 +253,7 @@ "at", "expect", "from", + "in", "rate", "until", "within" @@ -291,6 +300,7 @@ "at", "expect", "from", + "in", "rate", "until", "within" @@ -313,6 +323,7 @@ "at", "expect", "from", + "in", "rate", "until", "within" @@ -442,6 +453,15 @@ "description": "When the rule starts after it is added, such as after the warm-up (+30s or 30s)", "example": "+30s" }, + { + "name": "in", + "type": "string", + "required": false, + "description": "Narrows the sends, replies, polls and steps of the target to a route id, with * for any characters: the route of the exchange, not the routes it calls; with at=poll the route whose consumer polls. Not with route=, stall or lifecycle", + "example": "orders", + "pattern": true, + "pick": "routes" + }, { "name": "rate", "type": "rate", @@ -1214,6 +1234,145 @@ } ] }, + { + "id": "route-error", + "group": "Consumers and routes", + "title": "The route fails", + "description": "An exchange fails at the first step of the route, and the error handling of the route applies (onException, redelivery, the dead letter channel). When the route does not handle it, the consumer sees the failure: a JMS rollback, the file left in place, an HTTP 500", + "uncovers": [ + "A route without error handling, whose consumer then gets the failure: is that what the sender expects (a rollback, a retry, an HTTP 500)?", + "onException and redelivery that do not cover the route", + "Callers of the route that do not handle its failure" + ], + "avoid": [ + "To fail one call of the route: use the sends of the route (in=) or a step instead" + ], + "watch": [ + "The effect: recovered, handled and failed", + "What the consumer does with a failed exchange, such as a message going back to the broker" + ], + "rule": "route=${route} error rate=${rate} until=${until}", + "placeholders": [ + { + "name": "route", + "type": "route", + "description": "The route", + "example": "orders", + "pick": { + "from": "routes" + } + }, + { + "name": "rate", + "type": "rate", + "description": "How often: a probability, every:N or burst:<time>\/<period>", + "default": "0.05", + "example": "0.05" + }, + { + "name": "until", + "type": "countOrDuration", + "description": "When the rule ends: a number of matches or a time", + "default": "10m", + "example": "10m" + } + ] + }, + { + "id": "route-slow", + "group": "Consumers and routes", + "title": "The route is slow", + "description": "Each exchange waits before the first step of the route. It waits on the thread of the consumer, so the consumer takes fewer messages", + "uncovers": [ + "A consumer that falls behind, and the backlog in front of it", + "Timeouts of the callers of the route", + "How the route behaves with fewer messages per second" + ], + "avoid": [ + "On a route others call and wait for, unless their timeouts are the test" + ], + "watch": [ + "The mean and the slowest time of the route", + "The queue or the lag in front of the route" + ], + "rule": "route=${route} delay duration=${duration} rate=${rate} until=${until}", + "placeholders": [ + { + "name": "route", + "type": "route", + "description": "The route", + "example": "orders", + "pick": { + "from": "routes" + } + }, + { + "name": "duration", + "type": "duration", + "description": "The time, or a range of times", + "default": "500ms..2s", + "example": "500ms..2s" + }, + { + "name": "rate", + "type": "rate", + "description": "How often: a probability, every:N or burst:<time>\/<period>", + "default": "0.25", + "example": "0.25" + }, + { + "name": "until", + "type": "countOrDuration", + "description": "When the rule ends: a number of matches or a time", + "default": "10m", + "example": "10m" + } + ] + }, + { + "id": "route-sends-error", + "group": "Consumers and routes", + "title": "The sends of the route fail", + "description": "Each send of the route (to, enrich, wireTap...) fails with a ChaosException at the rate, whatever it sends to. The sends of the routes it calls are not in it", + "uncovers": [ + "Which calls of the route have no error handling of their own", + "A route that goes on after a failed send, with half of its work done", + "The error handler of the route against each kind of send" + ], + "avoid": [ + "When one call is the test: pick its endpoint instead of *" + ], + "watch": [ + "The effect: recovered, handled and failed", + "The sends of the route that failed, on the Routes tab" + ], + "rule": "endpoint=* in=${route} error at=send rate=${rate} until=${until}", + "placeholders": [ + { + "name": "route", + "type": "route", + "description": "The route", + "example": "orders", + "pick": { + "from": "routes" + } + }, + { + "name": "rate", + "type": "rate", + "description": "How often: a probability, every:N or burst:<time>\/<period>", + "default": "0.1", + "example": "0.1" + }, + { + "name": "until", + "type": "countOrDuration", + "description": "When the rule ends: a number of matches or a time", + "default": "10m", + "example": "10m" + } + ] + }, { "id": "dlq-unreachable", "group": "Error handling", diff --git a/components/camel-chaos/src/main/docs/chaos.adoc b/components/camel-chaos/src/main/docs/chaos.adoc index 88733c7ea84d..1b51bbfc206e 100644 --- a/components/camel-chaos/src/main/docs/chaos.adoc +++ b/components/camel-chaos/src/main/docs/chaos.adoc @@ -171,8 +171,8 @@ only the list and the report. == Rules -A rule is one line: the target, the fault and its parameters as `name=value`, then `at`, `rate`, `from` and `until`, -and optionally an expectation. A value with spaces or a `;` is quoted with `"` or `'` (`\"`, `\'` and `\\` inside quotes +A rule is one line: the target, the fault and its parameters as `name=value`, then `at`, `in`, `rate`, `from` and +`until`, and optionally an expectation. A value with spaces or a `;` is quoted with `"` or `'` (`\"`, `\'` and `\\` inside quotes are a quote and a backslash). [source] @@ -180,6 +180,8 @@ are a quote and a backslash). target: route=<id> | endpoint=<uri> | component=<name> | step=<id> (* matches any characters) fault: error | delay | drop | corrupt | duplicate | stall | lifecycle, with its parameters at: send | reply | dlq | poll | entry | step (optional) +in: in=<route id> narrows the sends, polls and steps of endpoint=, component= and step= to a route + (* matches any characters; optional) rate: a probability (rate=0.3), every Nth match (rate=every:5), or bursts (rate=burst:5s/30s: every match during the first 5s of each 30s); without it every match fires from: when the rule starts after it is added (from=+30s), such as after the warm-up; without it at once @@ -205,6 +207,26 @@ the endpoint URI `netty-http://http://inventory/stock`, and a camel-netty-http c matched with `netty-http` in front too. A `route=` pattern is matched against the route id, and a `step=` pattern against the id of the step (`to1`, or the id set with `id("enrich")`). +=== The sends and steps of a route + +`in=` narrows the target of a rule to the routes whose id it matches, for the sends, steps and polls that you know by +their route rather than by an endpoint URI or a step id (which are often generated, such as `to3`): + +[source] +---- +endpoint=http:* in=orders error status=503 # the HTTP calls of route orders +step=* in=orders delay duration=500ms # every step of route orders +component=kafka in=order* drop # the Kafka sends of the routes whose id starts with order +endpoint=* in=orders error at=send rate=0.1 # 10% of all the sends of route orders +---- + +The route is the route of the exchange: a send in a route that `orders` calls with `direct:` is in that other route, +not in `orders`; take more routes with a pattern (`in=order*`). With `at=poll` it is the route whose consumer polls. +`in=` is checked before the rate, so `rate` and an `until` count count only the matches in those routes. A `route=` +rule is about its route already, and `stall` and `lifecycle` act on the route of their `route=`, so they take no `in=`. +When no route matches `in=`, or the target matches no send, poll or step inside those routes, adding the rule warns, +with what is there. + A rule gets an id when it is added (`r1`, `r2`, ...), which change and remove take. A change replaces the rule with a new one with the same id and the next version (a rule is version 1 when it is added), and the replaced one stays in the report: so the report has an entry per version of the id, and a tool tells them apart by the `version` of the entry @@ -239,7 +261,7 @@ target offers, is rejected when it is added, with an error that names the hooks |`route=` |The messages that enter the route (`entry`), and the route itself. -|`entry`: `corrupt`. The route itself: `stall`, `lifecycle` (they take no `at`). +|`entry`: `error`, `delay`, `corrupt`. The route itself: `stall`, `lifecycle` (they take no `at`). |`step=` |The matching steps (`step`). @@ -253,9 +275,10 @@ the hooks where a rule can fire in the integration now (`hooks`): of the hooks o JMS), `send` (and `reply`) only when something sends to a matching endpoint (a `to`, a `toD`, `wireTap`, `enrich`, `recipientList`, `routingSlip` or `dynamicRouter` whose URI may match, or an endpoint a template sent to), and any hook where the rule fired already. The rule still fires at all its hooks, so a send that comes later (such as from a -`ProducerTemplate` to the endpoint of a route) is faulted too, and its `fired` counts have every hook of the rule. `route=orders error` is rejected: the -entry of a route offers only `corrupt`, and `route=` does not reach the sends and steps of the route (they are -targeted with `endpoint=`, `component=` or `step=`). A rule that matches nothing, such as a poll rule for a consumer +`ProducerTemplate` to the endpoint of a route) is faulted too, and its `fired` counts have every hook of the rule. +`route=orders error rate=0.05` fails 5% of the exchanges that enter the route, each once (see the `entry` hook below); +the sends and steps of the route are targeted with `endpoint=`, `component=` or `step=`, narrowed to the route with +`in=orders`. A rule that matches nothing, such as a poll rule for a consumer that does not poll on a schedule (Kafka, JMS), fires 0 times, as the report shows. When several rules match the same send, poll, message or step, they are asked in the order they were added, and the @@ -392,9 +415,17 @@ stopped or suspended meanwhile. The consumers that `pollEnrich` and a `ConsumerT route, and chaos does not reach their polls. |`entry` -|Each exchange that enters a matching route (a route policy, `onExchangeBegin`), before the first step. There is no -`drop` at the entry: the exchange already belongs to its consumer, and dropping it would make the consumer commit a -message the route never processed. +|Each exchange that enters a matching route (a route policy, `onExchangeBegin`), once, also when it enters the route +again (a loop, a redelivery). A `corrupt` changes the message before the first step. An `error` or a `delay` fires +there, and acts at the first step of the route, through the step hook: inside the error handler of the route, so +`onException`, redelivery and the dead letter channel of the route apply, as for a real failure inside it, and a +redelivery of that step runs the step itself. When the route does not handle the error, its consumer gets the failed +exchange, as it would for any failure of the route: a JMS message is rolled back, a file stays where it was, an HTTP +client gets a 500. A `delay` waits on the thread of the consumer (for most consumers), so the consumer takes fewer +messages. This is the honest form of "fail 5% of the order route": `step=* rate=0.05` would compound the rate over +every step. There is no `drop` at the entry: the exchange already belongs to its consumer, and dropping it would make +the consumer commit a message the route never processed. A fault of the consumer itself, before there is an exchange, +is a poll fault (`at=poll`) for a scheduled poll consumer, or a route that is down (`stall`, `lifecycle`). |`step` |Each step of a route, through the step hook of camel-core, right before the processor of the step is called, once for @@ -609,6 +640,12 @@ seconds |`step=enrich* error exception=java.sql.SQLException expect=orders:handled<=0` |fail every run of step enrich* with SQLException; expects no exchanges handled by the error handler in route orders +|`endpoint=http:stock* in=orders error status=503 rate=0.5` +|fail half of the sends to http:stock* in route orders with HTTP status 503 + +|`route=orders error rate=0.05` +|fail 5% of the messages entering route orders + |`route=poller lifecycle action=remove duration=30s repeat=3 every=1m` |remove route poller for 30 seconds, then add it back, 3 times, every 1 minute |=== 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 6cd53cc10e0d..1dff5138ff72 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 @@ -79,6 +79,10 @@ final class ChaosExchangeState implements SafeCopyProperty { boolean deadLetter; // chaos skipped the send the exchange is in (drop), so afterSend asks no rule with at=reply boolean skippedSend; + // a rule with route= that fired as the exchange entered its route (error or delay), which acts at the next step: the + // first step of the route, inside its error handler; and that route + ChaosRule entryFault; + String entryRoute; // the rule that failed or dropped that delivery, which loses the message when the dead letter channel then handles // it ChaosRule deadLetterFault; 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 e9ea87854167..b96a11f80a0a 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 @@ -795,6 +795,9 @@ public final class ChaosManager { final ChaosRule[] poll; final ChaosRule[] entry; final ChaosRule[] step; + // some rules with route= fail or delay the exchanges at the first step of their route, so the step hook looks + // for them + final boolean entryAtStep; // the lifecycle rules (stall and lifecycle), for the exchanges sent to their route while it is down final ChaosRule[] lifecycle; final ChaosRule.Expectation[] expectations; @@ -809,6 +812,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); + entryAtStep = Arrays.stream(entry).anyMatch(r -> r.fault != ChaosRule.Fault.CORRUPT); 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); diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosModel.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosModel.java index 958a65f6dabe..51260d1fad67 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosModel.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosModel.java @@ -98,7 +98,9 @@ final class ChaosModel { } case ROUTE -> { jo.put("description", "A route id, with * for any characters; one route id (no pattern) for" - + " stall and lifecycle"); + + " stall and lifecycle. Its exchanges are faulted as they enter: error and" + + " delay at the first step of the route, where its error handler applies;" + + " to reach the sends and steps of a route, use in= with another target"); jo.put("pattern", true); jo.put("pick", "routes"); } @@ -136,7 +138,8 @@ final class ChaosModel { + " with at=reply)"; case DLQ -> "The send of the dead letter channel to a matching endpoint (only with at=dlq)"; case POLL -> "The polls of the scheduled poll consumers of the routes"; - case ENTRY -> "Each exchange that enters a matching route, before the first step"; + case ENTRY -> "Each exchange that enters a matching route, once: corrupt changes it before the first" + + " step, error and delay act at the first step, inside the error handler of the route"; case STEP -> "Each matching step of a route, right before its processor is called"; }); jo.put("targets", new JsonArray(targets.getOrDefault(hook, List.of()))); @@ -297,6 +300,17 @@ final class ChaosModel { p.put("enum", hooks); yield p; } + case "in" -> { + JsonObject p = param("in", "string", false, + "Narrows the sends, replies, polls and steps of the target to a route id, with * for any" + + " characters: the route of the exchange, not the routes" + + " it calls; with at=poll the route whose consumer polls." + + " Not with route=, stall or lifecycle", + "orders"); + p.put("pattern", true); + p.put("pick", "routes"); + yield p; + } case "rate" -> { JsonObject p = param("rate", "rate", false, "How often the rule fires: a probability above 0 and at most 1, every Nth match, or bursts" @@ -481,6 +495,30 @@ final class ChaosModel { + "one route is reloaded or redeployed", "route=${route} lifecycle action=remove duration=${duration}", List.of(route(), new Placeholder("duration", "duration", "How long it is gone", "30s", "30s", null)))); + answer.add(new Recipe( + "route-error", "Consumers and routes", "The route fails", + "An exchange fails at the first step of the route, and the error handling of the " + + "route applies (onException, redelivery, the dead letter " + + "channel). When the route does not handle it, the consumer " + + "sees the failure: a JMS rollback, the file left in place, " + + "an HTTP 500", + "route=${route} error rate=${rate} until=${until}", + List.of(route(), rate("0.05"), until("10m")))); + answer.add(new Recipe( + "route-slow", "Consumers and routes", "The route is slow", + "Each exchange waits before the first step of the route. It waits on the thread " + + "of the consumer, so the consumer takes fewer messages", + "route=${route} delay duration=${duration} rate=${rate} until=${until}", + List.of(route(), new Placeholder( + "duration", "duration", "The time, or a range of times", "500ms..2s", + "500ms..2s", null), rate("0.25"), until("10m")))); + answer.add(new Recipe( + "route-sends-error", "Consumers and routes", "The sends of the route fail", + "Each send of the route (to, enrich, wireTap...) fails with a ChaosException at " + + "the rate, whatever it sends to. The sends of the routes it " + + "calls are not in it", + "endpoint=* in=${route} error at=send rate=${rate} until=${until}", + List.of(route(), rate("0.1"), until("10m")))); // error handling answer.add(new Recipe( "dlq-unreachable", "Error handling", "The dead letter channel is unreachable", @@ -663,6 +701,38 @@ final class ChaosModel { List.of( "The state of the route", "The callers, and the queue in front of the route"))); + answer.put("route-error", new Guide( + List.of( + "A route without error handling, whose consumer then gets the failure: is that " + + "what the sender expects (a rollback, a retry, an HTTP 500)?", + "onException and redelivery that do not cover the route", + "Callers of the route that do not handle its failure"), + List.of( + "To fail one call of the route: use the sends of the route (in=) or a step instead"), + List.of( + "The effect: recovered, handled and failed", + "What the consumer does with a failed exchange, such as a message going back to " + + "the broker"))); + answer.put("route-slow", new Guide( + List.of( + "A consumer that falls behind, and the backlog in front of it", + "Timeouts of the callers of the route", + "How the route behaves with fewer messages per second"), + List.of( + "On a route others call and wait for, unless their timeouts are the test"), + List.of( + "The mean and the slowest time of the route", + "The queue or the lag in front of the route"))); + answer.put("route-sends-error", new Guide( + List.of( + "Which calls of the route have no error handling of their own", + "A route that goes on after a failed send, with half of its work done", + "The error handler of the route against each kind of send"), + List.of( + "When one call is the test: pick its endpoint instead of *"), + List.of( + "The effect: recovered, handled and failed", + "The sends of the route that failed, on the Routes tab"))); answer.put("flap", new Guide( List.of( "Messages lost or delivered twice across a restart", 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 7c0d9f888752..7fb0e80ab2dd 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 @@ -62,7 +62,8 @@ final class ChaosPollStrategy implements PollingConsumerPollStrategy { long now = TimeSource.currentTimeMillis(); String uri = ChaosPatterns.endpointUri(endpoint); for (ChaosRule rule : rules) { - if (rule.live(now) && rule.matchesUri(uri) && rule.fires(now)) { + // in= is the route that consumes + if (rule.live(now) && rule.matchesUri(uri) && rule.matchesIn(routeId(consumer)) && rule.fires(now)) { return fault(rule, consumer, endpoint); } } 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 9100a0d0f095..c0dd6625bffb 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 @@ -44,10 +44,11 @@ final class ChaosRoutePolicyFactory implements RoutePolicyFactory { } /** - * Changes the message of an exchange as it enters the route ({@code onExchangeBegin}); the first rule that fires - * decides the fault. Counts the exchange when the route is done with it ({@code onExchangeDone}), while there are - * rules. camel-core calls both each time the exchange goes through the route, but an exchange meets the entry rules - * of a route, and counts in it, once (see {@link ChaosExchangeState}). + * Changes the message of an exchange as it enters the route ({@code onExchangeBegin}), or marks it for an error or + * a delay at the first step of the route, where the error handler of the route applies (see {@link ChaosStepHook}); + * the first rule that fires decides the fault. Counts the exchange when the route is done with it + * ({@code onExchangeDone}), while there are rules. camel-core calls both each time the exchange goes through the + * route, but an exchange meets the entry rules of a route, and counts in it, once (see {@link ChaosExchangeState}). */ static final class ChaosRoutePolicy extends RoutePolicySupport { @@ -66,7 +67,15 @@ 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, route.getRouteId(), null); + if (rule.fault == ChaosRule.Fault.CORRUPT) { + ChaosFaults.corrupt(rule, exchange, Hook.ENTRY, route.getRouteId(), null); + } else { + // the step hook fails or delays the exchange at the first step of the route + rule.fired(Hook.ENTRY, exchange, route.getRouteId(), null); + ChaosExchangeState state = ChaosExchangeState.of(exchange); + state.entryFault = rule; + state.entryRoute = route.getRouteId(); + } 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 f19b4b62355a..ef6bf49151b9 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 @@ -149,6 +149,8 @@ final class ChaosRule { final String text; final Target target; final String pattern; + // the routes the rule is narrowed to (in=, a route id or a pattern), else null + final String in; // the endpoint= pattern without the // after its scheme, made once (the hooks match it on every send and poll) private final String uriPattern; final Set<Hook> hooks; @@ -231,6 +233,7 @@ final class ChaosRule { this.text = b.text; this.target = b.target; this.pattern = b.pattern; + this.in = b.in; this.uriPattern = target == Target.ENDPOINT ? ChaosPatterns.withoutSlashes(pattern) : null; this.hooks = Set.copyOf(b.hooks); this.fault = b.fault; @@ -526,11 +529,19 @@ final class ChaosRule { private String noun(Set<Hook> at, boolean plural) { return switch (target) { case ROUTE -> (plural ? "messages" : "message") + " entering " + routeWords(); - case STEP -> (plural ? "runs" : "run") + " of step " + pattern; - case ENDPOINT, COMPONENT -> endpointNoun(at, plural); + case STEP -> (plural ? "runs" : "run") + " of step " + pattern + inWords(); + case ENDPOINT, COMPONENT -> endpointNoun(at, plural) + inWords(); }; } + // the routes of in=, after what the rule matches + private String inWords() { + if (in == null) { + return ""; + } + return (in.indexOf('*') >= 0 ? " in routes " : " in route ") + in; + } + private String endpointNoun(Set<Hook> at, boolean plural) { String what = target == Target.COMPONENT ? "the " + pattern + " endpoints" : pattern; List<String> kinds = new ArrayList<>(); @@ -727,6 +738,19 @@ final class ChaosRule { return ChaosPatterns.matchesUriPattern(uriPattern, uri); } + /** + * Whether the exchange is in a route of {@code in=}: the route of its unit of work, which is the route a + * {@code direct} call is in, not the route that called it. True for a rule without {@code in=}. + */ + boolean matchesIn(Exchange exchange) { + return in == null || matchesIn(ChaosExchangeState.routeId(exchange)); + } + + /** Whether the route is one of {@code in=} (true for a rule without it). */ + boolean matchesIn(String routeId) { + return in == null || routeId != null && ChaosPatterns.matches(in, routeId); + } + /** Whether the rule targets this route id or step id (for {@code route=} and {@code step=}). */ boolean matchesId(String id) { return id != null && ChaosPatterns.matches(pattern, id); @@ -979,6 +1003,7 @@ final class ChaosRule { String text; Target target; String pattern; + String in; Hook at; Set<Hook> hooks; Fault fault; diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRuleParser.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRuleParser.java index 16ab1146432e..eb695dd9e157 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRuleParser.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRuleParser.java @@ -57,7 +57,7 @@ final class ChaosRuleParser { // the options of a rule besides the parameters of its fault: all of them for the faults at a hook, and those of a // stall or lifecycle rule, which acts on its route at no hook - static final Set<String> GENERAL_KEYS = Set.of("at", "rate", "until", "from", "expect", "within"); + static final Set<String> GENERAL_KEYS = Set.of("at", "in", "rate", "until", "from", "expect", "within"); static final Set<String> LIFECYCLE_KEYS = Set.of("from", "expect", "within"); // the prefix of the response headers of error status=: header.<name>=<value> static final String HEADER_PREFIX = "header."; @@ -232,6 +232,7 @@ final class ChaosRuleParser { + " to match the endpoint"); } from(b, params.get("from"), maxDuration); + in(b, params.get("in")); if (b.fault == Fault.LIFECYCLE) { // stall and lifecycle act on the route: no hook, rate or until lifecycle(b, params, maxDuration); @@ -773,9 +774,10 @@ final class ChaosRuleParser { /** What to use instead, when a target does not offer a fault at all. */ private static String hint(Target target, Fault fault) { - if (target == Target.ROUTE) { - return "; route= does not reach the sends and steps of the route: target them with endpoint=," - + " component= or step="; + if (target == Target.ROUTE && (fault == Fault.DROP || fault == Fault.DUPLICATE)) { + return "; " + fault.key + " is a fault of the sends" + (fault == Fault.DROP ? " and polls" : "") + + ": target the sends of the route with endpoint=* in=<route>, or one of them with endpoint=<uri>" + + " in=<route>"; } if (target == Target.STEP && fault == Fault.DROP) { return "; drop is a fault of the sends and polls: target them with endpoint= or component="; @@ -812,7 +814,8 @@ final class ChaosRuleParser { answer.put(Hook.DLQ, EnumSet.of(Fault.ERROR, Fault.DELAY, Fault.DROP, Fault.CORRUPT)); answer.put(Hook.POLL, EnumSet.of(Fault.ERROR, Fault.DELAY, Fault.DROP)); } - case ROUTE -> answer.put(Hook.ENTRY, EnumSet.of(Fault.CORRUPT)); + // error and delay act at the first step of the route, inside its error handler + case ROUTE -> answer.put(Hook.ENTRY, EnumSet.of(Fault.ERROR, Fault.DELAY, Fault.CORRUPT)); case STEP -> answer.put(Hook.STEP, EnumSet.of(Fault.ERROR, Fault.DELAY)); default -> throw new IllegalStateException("Unknown target " + target); } @@ -983,6 +986,29 @@ final class ChaosRuleParser { } } + /** + * The routes a rule is narrowed to ({@code in=orders}, or a pattern such as {@code in=order*}): the sends, replies, + * steps and polls of its target in those routes only. A route is the target of {@code route=} already, and + * {@code stall} and {@code lifecycle} act on the route of their {@code route=}. + */ + private static void in(ChaosRule.Builder b, String in) { + if (in == null) { + return; + } + if (b.fault == Fault.LIFECYCLE) { + String word = b.stall ? "stall" : "lifecycle"; + throw new IllegalArgumentException( + "in= narrows the sends and steps of a rule to a route, but the fault " + word + + " acts on the route of its route=<id>"); + } + if (b.target == Target.ROUTE) { + throw new IllegalArgumentException( + "in= narrows the sends and steps of endpoint=, component= and step= to a route, but route=" + + b.pattern + " is a route already"); + } + b.in = in; + } + /** * The start of a rule after it is added ({@code from=+30s}, or {@code from=30s}): it fires, or acts on its route, * only from then on, such as after the warm-up of the application. 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 74e9c11abacf..5a84ca2ee0ee 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 @@ -69,7 +69,7 @@ final class ChaosSendHook implements SendHook, Ordered { long now = TimeSource.currentTimeMillis(); String uri = ChaosPatterns.endpointUri(endpoint); for (ChaosRule rule : rules) { - if (rule.live(now) && rule.matchesUri(uri) && rule.fires(now)) { + if (rule.live(now) && rule.matchesUri(uri) && rule.matchesIn(exchange) && rule.fires(now)) { if (deadLetter && (rule.fault == ChaosRule.Fault.ERROR || rule.fault == ChaosRule.Fault.DROP)) { // the message does not reach the dead letter channel: lost, when the channel then handles it state.deadLetterFault = rule; @@ -118,7 +118,7 @@ final class ChaosSendHook implements SendHook, Ordered { long now = TimeSource.currentTimeMillis(); String uri = ChaosPatterns.endpointUri(endpoint); for (ChaosRule rule : rules) { - if (rule.live(now) && rule.matchesUri(uri) && rule.fires(now)) { + if (rule.live(now) && rule.matchesUri(uri) && rule.matchesIn(exchange) && rule.fires(now)) { ChaosFaults.atReply(rule, exchange, endpoint); return; } 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 2cf2f5216335..abb9c4e1b3b0 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 @@ -39,26 +39,54 @@ final class ChaosStepHook implements StepHook, Ordered { @Override public long beforeStep(Exchange exchange, String nodeId) { - ChaosRule[] rules = manager.ruleSet().step; + ChaosManager.RuleSet ruleSet = manager.ruleSet(); + if (ruleSet.entryAtStep) { + ChaosRule entry = entryFault(exchange); + if (entry != null) { + // fired as the exchange entered the route: it counted there + return fault(entry, exchange); + } + } + ChaosRule[] rules = ruleSet.step; if (rules.length == 0 || nodeId == null) { return PROCEED; } long now = TimeSource.currentTimeMillis(); for (ChaosRule rule : rules) { - if (rule.live(now) && rule.matchesId(nodeId) && rule.fires(now)) { + if (rule.live(now) && rule.matchesId(nodeId) && rule.matchesIn(exchange) && rule.fires(now)) { rule.fired(Hook.STEP, exchange, null, nodeId); - if (rule.fault == Fault.DELAY) { - return rule.delayFor(); - } - if (ChaosFaults.waitBeforeError(rule, exchange)) { - exchange.setException(ChaosFaults.newException(rule)); - } - return PROCEED; + return fault(rule, exchange); } } return PROCEED; } + /** + * The rule with route= that fired as the exchange entered the route it is in now, which acts at this step, the + * first step of the route, once: a redelivery of the step by the error handler runs the step itself. + */ + private static ChaosRule entryFault(Exchange exchange) { + ChaosExchangeState state = ChaosExchangeState.peek(exchange); + if (state == null || state.entryFault == null) { + return null; + } + ChaosRule rule = state.entryFault; + String route = state.entryRoute; + state.entryFault = null; + state.entryRoute = null; + return route.equals(ChaosExchangeState.routeId(exchange)) ? rule : null; + } + + private static long fault(ChaosRule rule, Exchange exchange) { + if (rule.fault == Fault.DELAY) { + return rule.delayFor(); + } + if (ChaosFaults.waitBeforeError(rule, exchange)) { + exchange.setException(ChaosFaults.newException(rule)); + } + return PROCEED; + } + /** * The lowest order: the hook runs innermost, after the other hooks, right next to the processor of the step. */ 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 50e9e0b1c710..936fa7417670 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 @@ -21,7 +21,9 @@ import java.util.Collection; import java.util.Collections; import java.util.EnumSet; import java.util.IdentityHashMap; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.StringJoiner; import java.util.TreeSet; @@ -67,11 +69,70 @@ final class ChaosTargets { if (rule.lifecycle != null) { return null; } - return switch (rule.target) { + String answer = switch (rule.target) { case ENDPOINT, COMPONENT -> checkEndpoints(camelContext, rule); case ROUTE -> checkRoutes(camelContext, rule); case STEP -> checkSteps(camelContext, rule); }; + return answer == null && rule.in != null ? checkIn(camelContext, rule) : answer; + } + + /** + * Why a rule with {@code in=} finds nothing to fire at in its routes: no route matches {@code in=}, or its target + * matches nothing inside those routes (a send, a step, or the poll of the consumer of the route). A send inside a + * route that one of them calls ({@code direct:}) is in that other route. + */ + private static String checkIn(CamelContext camelContext, ChaosRule rule) { + List<Route> routes = new ArrayList<>(); + TreeSet<String> ids = new TreeSet<>(); + for (Route route : camelContext.getRoutes()) { + ids.add(route.getRouteId()); + if (rule.matchesIn(route.getRouteId())) { + routes.add(route); + } + } + if (routes.isEmpty()) { + return "in=" + rule.in + " matches no route of the context now, so the rule fires 0 times; the routes are: " + + some(ids); + } + String in = (routes.size() == 1 ? "route " : "routes ") + rule.in + " (in=" + rule.in + ")"; + if (rule.target == ChaosRule.Target.STEP) { + TreeSet<String> steps = new TreeSet<>(); + for (Route route : routes) { + for (Processor processor : route.filter("*")) { + String id = processor instanceof IdAware idAware ? idAware.getId() : null; + if (id != null) { + if (rule.matchesId(id)) { + return null; + } + steps.add(id); + } + } + } + return "step=" + rule.pattern + " matches no step of " + in + ", so the rule fires 0 times; the steps there" + + " are: " + some(steps); + } + if (rule.hooks.contains(Hook.DLQ)) { + // the dead letter channel is the error handler's, not a send of the route + return null; + } + Reach reach; + try { + reach = reach(camelContext, routes); + } catch (RuntimeException e) { + // the routes changed while they were read (such as a reload) + return null; + } + if (!hooksThatCanFire(reach, rule).isEmpty()) { + return null; + } + TreeSet<String> uris = new TreeSet<>(); + reach.routes().values().forEach(r -> r.sentTo().forEach(uri -> uris.add(withoutQuery(uri)))); + return rule.target.key + "=" + rule.pattern + " matches no " + (rule.hooks.contains(Hook.POLL) + ? "send or poll" + : "send") + + " of " + in + ", so the rule fires 0 times (in= is the route of the exchange, not the routes" + + " it calls); the sends there are: " + some(uris); } private static String checkEndpoints(CamelContext camelContext, ChaosRule rule) { @@ -118,15 +179,34 @@ final class ChaosTargets { || rule.lifecycle != null || reach == null) { return rule.hooks; } + // with in=, what the routes of in= reach, else what all of them do (and the sends of a ProducerTemplate) + List<String> polled = reach.polled; + List<String> sentTo = reach.sentTo; + List<String> dynamic = reach.dynamic; + if (rule.in != null) { + polled = new ArrayList<>(); + sentTo = new ArrayList<>(); + dynamic = new ArrayList<>(); + for (Map.Entry<String, RouteReach> entry : reach.routes.entrySet()) { + if (rule.matchesIn(entry.getKey())) { + RouteReach r = entry.getValue(); + if (r.polled() != null) { + polled.add(r.polled()); + } + sentTo.addAll(r.sentTo()); + dynamic.addAll(r.dynamic()); + } + } + } Set<Hook> answer = EnumSet.noneOf(Hook.class); Boolean sent = null; for (Hook hook : rule.hooks) { boolean can = switch (hook) { - case POLL -> reach.polled.stream().anyMatch(rule::matchesUri); + case POLL -> polled.stream().anyMatch(rule::matchesUri); case SEND, REPLY -> { if (sent == null) { - sent = reach.sentTo.stream().anyMatch(rule::matchesUri) - || reach.dynamic.stream().anyMatch(uri -> mayMatch(rule, uri)); + sent = sentTo.stream().anyMatch(rule::matchesUri) + || dynamic.stream().anyMatch(uri -> mayMatch(rule, uri)); } yield sent; } @@ -163,23 +243,31 @@ final class ChaosTargets { Set<Endpoint> sentTo = Collections.newSetFromMap(new IdentityHashMap<>()); List<String> polled = new ArrayList<>(); List<String> dynamic = new ArrayList<>(); + Map<String, RouteReach> byRoute = new LinkedHashMap<>(); for (Route route : routes) { Endpoint from = route.getEndpoint(); + String polledHere = null; if (from != null) { consumed.add(from); if (route.getConsumer() instanceof ScheduledPollConsumer) { - polled.add(ChaosPatterns.endpointUri(from)); + polledHere = ChaosPatterns.endpointUri(from); + polled.add(polledHere); } } + List<String> sentHere = new ArrayList<>(); for (Processor processor : route.filter("*")) { // a to, whose endpoint is known when the route starts (also one a route consumes, such as direct) if (processor instanceof EndpointAware aware && aware.getEndpoint() != null) { sentTo.add(aware.getEndpoint()); + sentHere.add(ChaosPatterns.endpointUri(aware.getEndpoint())); } } + List<String> dynamicHere = new ArrayList<>(); if (route.getRoute() instanceof RouteDefinition definition) { - dynamicSends(definition.getOutputs(), dynamic); + dynamicSends(definition.getOutputs(), dynamicHere); + dynamic.addAll(dynamicHere); } + byRoute.put(route.getRouteId(), new RouteReach(sentHere, dynamicHere, polledHere)); } List<String> uris = new ArrayList<>(); for (Endpoint endpoint : camelContext.getEndpoints()) { @@ -188,7 +276,7 @@ final class ChaosTargets { uris.add(ChaosPatterns.endpointUri(endpoint)); } } - return new Reach(uris, dynamic, polled); + return new Reach(uris, dynamic, polled, byRoute); } private static void dynamicSends(List<ProcessorDefinition<?>> outputs, List<String> uris) { @@ -263,8 +351,18 @@ final class ChaosTargets { return answer; } - /** What the routes of a context reach, see {@link #reach(CamelContext)}. */ - record Reach(List<String> sentTo, List<String> dynamic, List<String> polled) { + /** + * What the routes of a context reach, see {@link #reach(CamelContext)}, and what each route does, by its id (for + * the rules with {@code in=}). + */ + record Reach(List<String> sentTo, List<String> dynamic, List<String> polled, Map<String, RouteReach> routes) { + } + + /** + * What one route reaches: the URIs of the endpoints its steps send to, of the sends known only when they send, and + * of the endpoint it consumes with a scheduled poll (null when its consumer does not poll on a schedule). + */ + record RouteReach(List<String> sentTo, List<String> dynamic, String polled) { } private static String checkRoutes(CamelContext camelContext, ChaosRule rule) { diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosInRouteTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosInRouteTest.java new file mode 100644 index 000000000000..8de7c79a75f7 --- /dev/null +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosInRouteTest.java @@ -0,0 +1,194 @@ +/* + * 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.TimeUnit; + +import org.apache.camel.BindToRegistry; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.chaos.TestPollComponent.TestPollConsumer; +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 static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * {@code in=} narrows the target of a rule to a route: the sends, steps and polls of the target in the routes of + * {@code in=} only, which is the route of the exchange (not the routes it calls), and for a poll the route that + * consumes. + */ +class ChaosInRouteTest extends ChaosTestSupport { + + @BindToRegistry("testpoll") + private final TestPollComponent testPoll = new TestPollComponent(); + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + // both routes send to mock:shared; orders calls stock, which sends to it too + from("direct:orders").routeId("orders") + .setHeader("seen", constant("orders")).id("mark") + .to("direct:stock").id("callStock") + .to("mock:shared").id("ordersShared"); + from("direct:stock").routeId("stock") + .to("mock:shared").id("stockShared"); + from("direct:billing").routeId("orderBilling") + .to("mock:shared").id("billingShared"); + from("direct:other").routeId("other") + .to("mock:shared").id("otherShared"); + from("testpoll:a?delay=5&initialDelay=0").routeId("pollA").to("mock:a"); + from("testpoll:b?delay=5&initialDelay=0").routeId("pollB").to("mock:b"); + } + }; + } + + @Test + void onlyTheSendsOfTheRoute() throws Exception { + String id = chaos().addRule("endpoint=mock:shared in=other error"); + getMockEndpoint("mock:shared").expectedMessageCount(3); + + Exchange other = template.send("direct:other", e -> e.getMessage().setBody("x")); + // orders sends to mock:shared, and so does stock, which it calls + Exchange orders = template.send("direct:orders", e -> e.getMessage().setBody("x")); + template.sendBody("direct:billing", "x"); + + assertThat(other.getException()).isInstanceOf(ChaosException.class); + assertThat(orders.getException()).isNull(); + MockEndpoint.assertIsSatisfied(context); + assertThat(fired(id, "send")).isEqualTo(1); + } + + @Test + void theRouteOfTheExchangeNotTheRoutesItCalls() throws Exception { + String id = chaos().addRule("endpoint=mock:shared in=orders error"); + getMockEndpoint("mock:shared").expectedMessageCount(1); + + Exchange orders = template.send("direct:orders", e -> e.getMessage().setBody("x")); + + // the send of stock went through, the send of orders after it failed + assertThat(orders.getException()).isInstanceOf(ChaosException.class); + MockEndpoint.assertIsSatisfied(context); + assertThat(fired(id, "send")).isEqualTo(1); + JsonArray recent = (JsonArray) chaos().report().toJson().get("recentFaults"); + assertThat(((JsonObject) recent.get(0)).getString("route")).isEqualTo("orders"); + } + + @Test + void aPatternTakesMoreRoutes() throws Exception { + String id = chaos().addRule("endpoint=mock:shared in=order* error"); + getMockEndpoint("mock:shared").expectedMessageCount(2); + + assertThat(template.send("direct:orders", e -> e.getMessage().setBody("x")).getException()) + .isInstanceOf(ChaosException.class); + assertThat(template.send("direct:billing", e -> e.getMessage().setBody("x")).getException()) + .isInstanceOf(ChaosException.class); + template.sendBody("direct:other", "x"); + + MockEndpoint.assertIsSatisfied(context); + assertThat(fired(id, "send")).isEqualTo(2); + } + + @Test + void theRateCountsOnlyTheMatchesInTheRoute() { + String id = chaos().addRule("endpoint=mock:shared in=other error rate=every:2"); + + for (int i = 0; i < 3; i++) { + template.sendBody("direct:billing", "x"); + } + assertThat(template.send("direct:other", e -> e.getMessage().setBody("x")).getException()).isNull(); + assertThat(template.send("direct:other", e -> e.getMessage().setBody("x")).getException()) + .isInstanceOf(ChaosException.class); + + assertThat(ruleReport(id).getLong("matches")).isEqualTo(2); + } + + @Test + void theStepsOfTheRoute() throws Exception { + String id = chaos().addRule("step=* in=stock error"); + getMockEndpoint("mock:shared").expectedMessageCount(0); + + Exchange orders = template.send("direct:orders", e -> e.getMessage().setBody("x")); + + // the steps of orders ran up to the call of stock, whose first step failed + assertThat(orders.getException()).isInstanceOf(ChaosException.class); + assertThat(orders.getMessage().getHeader("seen")).isEqualTo("orders"); + MockEndpoint.assertIsSatisfied(context); + assertThat(fired(id, "step")).isEqualTo(1); + JsonArray recent = (JsonArray) chaos().report().toJson().get("recentFaults"); + assertThat(((JsonObject) recent.get(0)).getString("step")).isEqualTo("stockShared"); + } + + @Test + void thePollsOfTheRouteThatConsumes() { + TestPollConsumer b = (TestPollConsumer) context.getRoute("pollB").getConsumer(); + String id = chaos().addRule("component=testpoll drop at=poll in=pollA"); + + await().atMost(10, TimeUnit.SECONDS).until(() -> fired(id, "poll") >= 3); + int polls = b.polls.get(); + await().atMost(10, TimeUnit.SECONDS).until(() -> b.polls.get() >= polls + 3); + JsonArray recent = (JsonArray) chaos().report().toJson().get("recentFaults"); + // b polls on, and every fault is a poll of a + assertThat(recent).allSatisfy(f -> assertThat(((JsonObject) f).getString("route")).isEqualTo("pollA")); + } + + @Test + void aWarningWhenNoRouteMatches() { + String id = chaos().addRule("endpoint=mock:shared in=ordres error"); + assertThat(chaos().checkRule(id)).singleElement().asString() + .startsWith("in=ordres matches no route of the context now, so the rule fires 0 times; the routes are:" + + " orderBilling, orders, other, pollA, pollB"); + } + + @Test + void aWarningWhenTheTargetIsNotInTheRoute() { + // the endpoint is there, but orders does not send to it + String id = chaos().addRule("endpoint=mock:a in=orders error"); + assertThat(chaos().checkRule(id)).singleElement().asString() + .isEqualTo("endpoint=mock:a matches no send or poll of route orders (in=orders), so the rule fires 0 times" + + " (in= is the route of the exchange, not the routes it calls); the sends there are:" + + " direct://stock, mock://shared"); + String step = chaos().addRule("step=billing* in=orders delay duration=1ms"); + assertThat(chaos().checkRule(step)).singleElement().asString() + .isEqualTo("step=billing* matches no step of route orders (in=orders), so the rule fires 0 times; the" + + " steps there are: callStock, mark, ordersShared"); + // the steps of stock are not in orders + assertThat(chaos().checkRule(chaos().addRule("step=stockShared in=orders error"))).hasSize(1); + assertThat(chaos().checkRule(chaos().addRule("endpoint=mock:shared in=order* error"))).isEmpty(); + assertThat(chaos().checkRule(chaos().addRule("step=mark in=orders error"))).isEmpty(); + } + + @Test + void theHooksThatCanFireAreThoseOfTheRoute() { + assertThat(hooks(chaos().addRule("component=testpoll error in=pollA"))).containsExactly("poll"); + // orders consumes direct, which does not poll, and sends to no testpoll endpoint + assertThat(hooks(chaos().addRule("component=testpoll error in=orders"))).isEmpty(); + assertThat(hooks(chaos().addRule("endpoint=mock:shared delay duration=1ms in=stock"))).containsExactly("send"); + assertThat(ruleReport("r3").getString("description")) + .isEqualTo("delay every send to mock:shared in route stock by 1 millisecond"); + } + + private List<Object> hooks(String id) { + return (JsonArray) ruleReport(id).get("hooks"); + } +} diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosModelTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosModelTest.java index 0c2d80da559f..205c803c6d4e 100644 --- a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosModelTest.java +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosModelTest.java @@ -94,12 +94,15 @@ class ChaosModelTest { assertThat(paramNames(byName("faults", "error"))) .containsExactly("after", "body", "exception", "header.<name>", "message", "status"); assertThat(strings(byName("faults", "lifecycle").get("options"))).containsExactly("expect", "from", "within"); - assertThat(strings(byName("faults", "error").get("hooks"))).containsExactly("send", "reply", "dlq", "poll", "step"); + assertThat(strings(byName("faults", "error").get("hooks"))).containsExactly("send", "reply", "dlq", "poll", "entry", + "step"); // duplicate is a fault of the sends only assertThat(strings(byName("faults", "duplicate").get("hooks"))).containsExactly("send"); assertThat(strings(byName("faults", "duplicate").get("targets"))).containsExactly("component", "endpoint"); assertThat(paramNames(byName("faults", "duplicate"))).isEmpty(); - assertThat(names("options")).containsExactly("at", "expect", "from", "rate", "until", "within"); + assertThat(names("options")).containsExactly("at", "expect", "from", "in", "rate", "until", "within"); + // the TUI form picks the route of in= + assertThat(byName("options", "in").getString("pick")).isEqualTo("routes"); } @Test diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRouteEntryErrorTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRouteEntryErrorTest.java new file mode 100644 index 000000000000..4b589e1d622e --- /dev/null +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRouteEntryErrorTest.java @@ -0,0 +1,188 @@ +/* + * 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.net.ConnectException; + +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 static org.assertj.core.api.Assertions.assertThat; + +/** + * {@code route=X error} and {@code route=X delay}: the rule fires once per exchange as it enters the route, and the + * exchange fails or waits at the first step of the route, inside the error handler of the route. + */ +class ChaosRouteEntryErrorTest extends ChaosTestSupport { + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:plain").routeId("plain") + .setHeader("seen", constant("yes")).id("mark") + .to("mock:plain").id("send"); + + from("direct:redeliver").routeId("redeliver") + .errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2).redeliveryDelay(0)) + .to("mock:flaky").id("flaky"); + + from("direct:dlc").routeId("dlc") + .errorHandler(deadLetterChannel("mock:dead").redeliveryDelay(0)) + .to("mock:never").id("never"); + + from("direct:onexception").routeId("onexception") + .onException(ConnectException.class).handled(true).to("mock:handled").end() + .to("mock:io").id("io"); + + from("direct:caller").routeId("caller") + .to("direct:plain").id("callPlain") + .to("mock:caller").id("afterCall"); + } + }; + } + + @Test + void theExchangeFailsAtTheFirstStep() throws Exception { + String id = chaos().addRule("route=plain error"); + getMockEndpoint("mock:plain").expectedMessageCount(0); + + Exchange out = template.send("direct:plain", e -> e.getMessage().setBody("x")); + + // without error handling the caller gets the failure, and no step ran + assertThat(out.getException()).isInstanceOf(ChaosException.class); + assertThat(out.getMessage().getHeader("seen")).isNull(); + MockEndpoint.assertIsSatisfied(context); + assertThat(fired(id, "entry")).isEqualTo(1); + assertThat(routeReport("plain").getLong("failed")).isEqualTo(1); + JsonObject fault = (JsonObject) ((JsonArray) chaos().report().toJson().get("recentFaults")).get(0); + assertThat(fault.getString("hook")).isEqualTo("entry"); + assertThat(fault.getString("route")).isEqualTo("plain"); + } + + @Test + void theRedeliveryOfTheFirstStepRecovers() throws Exception { + String id = chaos().addRule("route=redeliver error"); + getMockEndpoint("mock:flaky").expectedMessageCount(2); + getMockEndpoint("mock:dead").expectedMessageCount(0); + + template.sendBody("direct:redeliver", "x"); + template.sendBody("direct:redeliver", "y"); + + MockEndpoint.assertIsSatisfied(context); + // once per exchange, as it entered the route: the redelivery runs the step itself + assertThat(fired(id, "entry")).isEqualTo(2); + assertThat(ruleReport(id).getLong("matches")).isEqualTo(2); + assertThat(routeReport("redeliver").getLong("redeliveries")).isEqualTo(2); + assertThat(routeReport("redeliver").getLong("completed")).isEqualTo(2); + JsonObject effect = (JsonObject) ruleReport(id).get("effect"); + assertThat(effect.getLong("recovered")).isEqualTo(2); + } + + @Test + void theDeadLetterChannelOfTheRouteHandlesIt() throws Exception { + String id = chaos().addRule("route=dlc error"); + getMockEndpoint("mock:never").expectedMessageCount(0); + getMockEndpoint("mock:dead").expectedMessageCount(1); + + Exchange out = template.send("direct:dlc", e -> e.getMessage().setBody("x")); + + // no redeliveries, so the dead letter channel takes it + assertThat(out.getException()).isNull(); + MockEndpoint.assertIsSatisfied(context); + assertThat(fired(id, "entry")).isEqualTo(1); + assertThat(routeReport("dlc").getLong("handled")).isEqualTo(1); + } + + @Test + void onExceptionOfTheRouteHandlesTheTypedFault() throws Exception { + chaos().addRule("route=onexception error exception=" + ConnectException.class.getName() + " message=refused"); + getMockEndpoint("mock:io").expectedMessageCount(0); + getMockEndpoint("mock:handled").expectedMessageCount(1); + + Exchange out = template.send("direct:onexception", e -> e.getMessage().setBody("x")); + + assertThat(out.getException()).isNull(); + MockEndpoint.assertIsSatisfied(context); + Exception caught = getMockEndpoint("mock:handled").getReceivedExchanges().get(0) + .getProperty(Exchange.EXCEPTION_CAUGHT, Exception.class); + assertThat(caught).isExactlyInstanceOf(ConnectException.class).hasMessage("refused"); + } + + @Test + void aRouteThatIsCalledFailsAtItsOwnFirstStep() throws Exception { + String id = chaos().addRule("route=plain error"); + getMockEndpoint("mock:caller").expectedMessageCount(0); + getMockEndpoint("mock:plain").expectedMessageCount(0); + + Exchange out = template.send("direct:caller", e -> e.getMessage().setBody("x")); + + assertThat(out.getException()).isInstanceOf(ChaosException.class); + MockEndpoint.assertIsSatisfied(context); + assertThat(fired(id, "entry")).isEqualTo(1); + JsonObject fault = (JsonObject) ((JsonArray) chaos().report().toJson().get("recentFaults")).get(0); + assertThat(fault.getString("route")).isEqualTo("plain"); + } + + @Test + void theRateDecidesPerExchange() { + String id = chaos().addRule("route=plain error rate=every:2"); + + assertThat(template.send("direct:plain", e -> e.getMessage().setBody("1")).getException()).isNull(); + assertThat(template.send("direct:plain", e -> e.getMessage().setBody("2")).getException()) + .isInstanceOf(ChaosException.class); + assertThat(template.send("direct:plain", e -> e.getMessage().setBody("3")).getException()).isNull(); + + assertThat(ruleReport(id).getLong("matches")).isEqualTo(3); + assertThat(fired(id, "entry")).isEqualTo(1); + } + + @Test + void aDelayWaitsAtTheFirstStepThenTheRouteGoesOn() throws Exception { + String id = chaos().addRule("route=plain delay duration=50ms"); + getMockEndpoint("mock:plain").expectedMessageCount(1); + + long start = System.nanoTime(); + Exchange out = template.send("direct:plain", e -> e.getMessage().setBody("x")); + long millis = (System.nanoTime() - start) / 1_000_000; + + assertThat(out.getException()).isNull(); + assertThat(millis).isGreaterThanOrEqualTo(50); + MockEndpoint.assertIsSatisfied(context); + assertThat(fired(id, "entry")).isEqualTo(1); + assertThat(ruleReport(id).getString("description")).isEqualTo("delay every message entering route plain by 50" + + " milliseconds"); + } + + @Test + void corruptAtTheEntryStillChangesTheMessageBeforeTheFirstStep() throws Exception { + String error = chaos().addRule("route=redeliver error until=1"); + String corrupt = chaos().addRule("route=plain corrupt mode=empty"); + getMockEndpoint("mock:plain").expectedBodiesReceived(""); + + template.sendBody("direct:plain", "x"); + + MockEndpoint.assertIsSatisfied(context); + assertThat(fired(corrupt, "entry")).isEqualTo(1); + assertThat(fired(error, "entry")).isZero(); + } +} diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleDescribeTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleDescribeTest.java index ab638f3c66a7..6848fb942193 100644 --- a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleDescribeTest.java +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleDescribeTest.java @@ -72,6 +72,12 @@ class ChaosRuleDescribeTest { endpoint=x error at=send expect=orders:failed=3 | fail every send to x; expects exactly 3 exchanges failed in route orders endpoint=x error at=send expect=orders:redeliveries>=2 | fail every send to x; expects at least 2 redeliveries in route orders endpoint=x error at=send expect=orders:completed<=1 | fail every send to x; expects at most 1 exchange completed in route orders + endpoint=http:stock* in=orders error status=503 rate=0.5 | fail half of the sends to http:stock* in route orders with HTTP status 503 + step=* in=order* delay duration=500ms | delay every run of step * in routes order* by 500 milliseconds + component=file drop at=poll in=inbox | drop every poll of the file endpoints in route inbox + route=orders error rate=0.05 | fail 5% of the messages entering route orders + route=orders error exception=java.net.ConnectException | fail every message entering route orders with ConnectException + route=order* delay duration=2s until=10 | delay every message entering routes order* by 2 seconds, for the next 10 messages route=poller stall duration=20s | suspend route poller for 20 seconds route=poller lifecycle action=stop duration=1m30s | stop route poller for 1 minute 30 seconds route=poller lifecycle action=remove duration=30s repeat=3 every=1m from=+10s | remove route poller for 30 seconds, then add it back, 3 times, every 1 minute, starting after 10 seconds diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleParserTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleParserTest.java index 81f7efcbca7f..cf6503e4ad92 100644 --- a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleParserTest.java +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleParserTest.java @@ -131,6 +131,30 @@ class ChaosRuleParserTest { assertThat(parse("component=kafka duplicate at=send until=10").hooks).containsExactly(Hook.SEND); } + @Test + void inNarrowsTheTargetToTheRoutes() { + ChaosRule.Builder send = parse("endpoint=http:* in=orders error status=503"); + assertThat(send.in).isEqualTo("orders"); + assertThat(send.hooks).containsExactlyInAnyOrder(Hook.SEND); + ChaosRule.Builder step = parse("step=* in=order* delay duration=500ms"); + assertThat(step.in).isEqualTo("order*"); + assertThat(step.hooks).containsExactly(Hook.STEP); + assertThat(parse("component=file in=inbox drop at=poll").in).isEqualTo("inbox"); + assertThat(parse("endpoint=x error").in).isNull(); + } + + @Test + void errorAndDelayAtTheEntryOfARoute() { + ChaosRule.Builder error = parse("route=orders error exception=java.net.ConnectException rate=0.05"); + assertThat(error.fault).isEqualTo(Fault.ERROR); + assertThat(error.hooks).containsExactly(Hook.ENTRY); + assertThat(error.exceptionType).isEqualTo(ConnectException.class); + ChaosRule.Builder delay = parse("route=order* delay duration=2s"); + assertThat(delay.hooks).containsExactly(Hook.ENTRY); + assertThat(delay.duration).isEqualTo(2000L); + assertThat(parse("route=orders error after=1s at=entry").after).isEqualTo(1000L); + } + @Test void lifecycle() { ChaosRule.Builder stall = parse("route=poller stall duration=20s"); @@ -296,11 +320,18 @@ class ChaosRuleParserTest { "endpoint=x error when=x|when= (a condition on the message) is not supported yet", "endpoint=x error action=stop|action= is a parameter of the fault lifecycle (route=<id> lifecycle action=", "component=kafka error status=503|(http, https, netty-http, undertow, vertx-http), not for kafka", - "route=orders error|route= does not reach the sends and steps of the route: target them with endpoint=," - + " component= or step=", + "route=orders drop|drop is a fault of the sends and polls: target the sends of the route with endpoint=*" + + " in=<route>, or one of them with endpoint=<uri> in=<route>", "step=s1 drop|drop is a fault of the sends and polls: target them with endpoint= or component=", "step=s1 duplicate|duplicate is a fault of the sends: target them with endpoint= or component=", - "route=orders duplicate|route= does not reach the sends and steps of the route", + "route=orders duplicate|duplicate is a fault of the sends: target the sends of the route with endpoint=*", + "route=orders error status=503|status= is the failure of a send to an HTTP component, so it fires at the" + + " sends of an endpoint= or component= target (not route=)", + "route=orders error in=orders|in= narrows the sends and steps of endpoint=, component= and step= to a route," + + " but route=orders is a route already", + "route=orders stall duration=1m in=orders|in= narrows the sends and steps of a rule to a route, but the" + + " fault stall acts on the route of its route=<id>", + "route=orders lifecycle action=stop duration=1m in=x|but the fault lifecycle acts on the route of its route=<id>", "endpoint=x duplicate at=poll|the fault duplicate is not offered at=poll", "endpoint=x duplicate at=reply|the fault duplicate is not offered at=reply", "endpoint=x duplicate at=dlq|the fault duplicate is not offered at=dlq", @@ -338,8 +369,7 @@ class ChaosRuleParserTest { + " pattern", "route=orders stall error duration=1m|has one fault", "rate=1 duration=1m stall|needs a target", - "route=orders error|the fault error is not offered at any hook of the target route=", - "route=orders delay duration=1s|the hooks of route= are: at=entry offers corrupt", + "route=orders duplicate|the hooks of route= are: at=entry offers error, delay, corrupt", "route=orders drop|is not offered", "step=s1 drop|is not offered", "step=s1 corrupt mode=headers|is not offered", @@ -508,8 +538,8 @@ class ChaosRuleParserTest { @Test void theReasonNamesTheRule() { - assertThatThrownBy(() -> parse("route=orders error")) - .hasMessageStartingWith("Invalid chaos rule [route=orders error]: "); + assertThatThrownBy(() -> parse("route=orders drop")) + .hasMessageStartingWith("Invalid chaos rule [route=orders drop]: "); } @Test diff --git a/design/chaos.adoc b/design/chaos.adoc index ee9368b2c6ac..e43ee7cb79d4 100644 --- a/design/chaos.adoc +++ b/design/chaos.adoc @@ -228,9 +228,11 @@ poll rule for a consumer that does not poll on a schedule (Kafka, JMS), fires 0 - `endpoint=file:inbox* drop` (or `error`) fires at both hooks: a poll of the `file:inbox` consumer is skipped, and a send to `file:inbox` reports success without writing. `at=poll` or `at=send` keeps one. With `corrupt` the rule fires at the sends only. -- `route=orders error` and `route=orders delay` are rejected: the entry of a route offers only `corrupt`. `route=` - does not reach the sends and steps of the route; they are targeted with `endpoint=`, `component=` or `step=`. One - kind of target per hook keeps a rule, and its counts in the report, in one kind of place. +- `route=orders error` and `route=orders delay` fire once per exchange as it enters the route, and act at the first + step of the route, inside its error handler (stage 2, CAMEL-26718; stage 1 offered only `corrupt` at the entry). + `route=` does not reach the sends and steps of the route; they are targeted with `endpoint=`, `component=` or + `step=`, narrowed to the route with `in=orders`. One kind of target per hook keeps a rule, and its counts in the + report, in one kind of place. - `component=kafka` matches the sends to `kafka:` endpoints and the polls of its consumers (none: they do not poll on a schedule), not the entry of a route that starts from Kafka, which is `route=`. diff --git a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelChaosAddAction.java b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelChaosAddAction.java index f8a32c862cc1..7ecc712d4b36 100644 --- a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelChaosAddAction.java +++ b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelChaosAddAction.java @@ -42,6 +42,8 @@ import picocli.CommandLine; " camel cmd chaos-add --rule='endpoint=kafka:orders* delay duration=2s rate=0.2 until=5m'", " camel cmd chaos-add myApp --rule='endpoint=file:inbox* drop at=poll rate=0.5 until=2m'", " camel cmd chaos-add myApp --rule='step=enrich* error exception=java.sql.SQLException rate=0.1'", + " camel cmd chaos-add myApp --rule='endpoint=http:* in=orders error status=503 rate=0.1'", + " camel cmd chaos-add myApp --rule='route=orders error rate=0.05'", " camel cmd chaos-add myApp --rule='endpoint=seda:dlq error at=dlq' --rule='endpoint=http:stock* error'" }) public class CamelChaosAddAction extends CamelChaosAction { 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 791a7ba53358..b080c71eb061 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 @@ -672,6 +672,9 @@ until=5m' until=2m' camel cmd chaos-add myApp --rule='step=enrich* error exception=java.sql. SQLException rate=0.1' + camel cmd chaos-add myApp --rule='endpoint=http:* in=orders error status=503 +rate=0.1' + camel cmd chaos-add myApp --rule='route=orders error rate=0.05' camel cmd chaos-add myApp --rule='endpoint=seda:dlq error at=dlq' --rule='endpoint=http:stock* error' diff --git a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosInfo.java b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosInfo.java index 668d5517f146..6fb2106f54f1 100644 --- a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosInfo.java +++ b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosInfo.java @@ -92,8 +92,21 @@ final class ChaosInfo { long lifecycleBackIn = -1; String lifecycleReason; - /** The target of the rule as it is written, such as {@code endpoint=kafka:orders*}. */ + /** + * The target of the rule as it is written, such as {@code endpoint=kafka:orders*}, with the route it is + * narrowed to ({@code endpoint=http:* in=orders}). + */ String target() { + String target = targetToken(); + for (String token : tokens()) { + if (token.startsWith("in=")) { + return target.isEmpty() ? token : target + " " + token; + } + } + return target; + } + + private String targetToken() { for (String token : tokens()) { if (token.startsWith("endpoint=") || token.startsWith("component=") || token.startsWith("route=") || token.startsWith("step=")) { @@ -105,14 +118,15 @@ final class ChaosInfo { /** * The fault of the rule with its parameters, as it is written, without the target and the options of the rule - * (at, rate, until, from, expect, within), such as {@code error status=503}. + * (at, in, rate, until, from, expect, within), such as {@code error status=503}. */ String fault() { StringBuilder sb = new StringBuilder(); + String target = targetToken(); for (String token : tokens()) { int eq = token.indexOf('='); String key = eq > 0 ? token.substring(0, eq) : null; - if (token.equals(target()) || key != null && OPTIONS.contains(key)) { + if (token.equals(target) || key != null && OPTIONS.contains(key)) { continue; } if (!sb.isEmpty()) { @@ -228,7 +242,7 @@ final class ChaosInfo { } // the options of a rule besides its target and its fault - private static final List<String> OPTIONS = List.of("at", "rate", "until", "from", "expect", "within"); + private static final List<String> OPTIONS = List.of("at", "in", "rate", "until", "from", "expect", "within"); /** The counts of a route in the report. */ record RouteCounts(String route, long completed, long failed, long failedKept, long handled, long redeliveries) { diff --git a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosTab.java b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosTab.java index 8cb3a5de72ac..1949295904ee 100644 --- a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosTab.java +++ b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosTab.java @@ -318,10 +318,11 @@ class ChaosTab extends AbstractTableTab { /** * The target of a rule for the table: the URI pattern of an {@code endpoint=} target (the most common one), else - * the kind and the pattern, such as {@code route poller}. + * the kind and the pattern, such as {@code route poller}, and the route it is narrowed to + * ({@code http:* in orders}). */ static String target(ChaosInfo.Rule r) { - String t = r.target(); + String t = r.target().replace(" in=", " in "); if (t.startsWith("endpoint=")) { return t.substring(9); } diff --git a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/resources/tui/help/chaos.md b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/resources/tui/help/chaos.md index f35c7eb31129..c321456cf809 100644 --- a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/resources/tui/help/chaos.md +++ b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/resources/tui/help/chaos.md @@ -30,7 +30,7 @@ minute, and the seed of the random draws. - **ID** — The id of the rule (r1, r2, ...), with its version when it was changed - **STATE** — `active`, `waiting` (it starts later, `from=`) or `ended` -- **TARGET** — What the rule matches: `endpoint=`, `component=`, `route=` or `step=` +- **TARGET** — What the rule matches: `endpoint=`, `component=`, `route=` or `step=`, with `in=` when it is narrowed to the sends and steps of a route - **FAULT** — What it does, in a few words: `error status 503`, `error ConnectException`, `delay 500ms..1s`, `stop 5s ×3 every 20s` (the rule in full is the title of the panel under the table) - **RATE** — The share of the matches it fired at, next to its own: `24% of 25%`, `20% of 1 in 5`, `16% of 5s in 30s`; for a flapping route the round - **FIRED** — How many faults it fired diff --git a/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosTabRenderTest.java b/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosTabRenderTest.java index 9d74e9be995a..a8bf18aa3734 100644 --- a/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosTabRenderTest.java +++ b/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/ChaosTabRenderTest.java @@ -296,5 +296,16 @@ class ChaosTabRenderTest { assertThat(all.target()).isEqualTo("step=a"); assertThat(ChaosTab.target(all)).isEqualTo("step a"); assertThat(all.fault()).isEqualTo("delay duration='1s'"); + + // narrowed to the sends and steps of a route (in=): with the target, not with the fault + ChaosInfo.Rule in = new ChaosInfo.Rule(); + in.text = "endpoint=http:* in=orders error status=503 rate=0.1"; + assertThat(in.target()).isEqualTo("endpoint=http:* in=orders"); + assertThat(ChaosTab.target(in)).isEqualTo("http:* in orders"); + assertThat(in.fault()).isEqualTo("error status=503"); + ChaosInfo.Rule steps = new ChaosInfo.Rule(); + steps.text = "step=* in=order* delay duration=1s"; + assertThat(ChaosTab.target(steps)).isEqualTo("step * in order*"); + assertThat(steps.fault()).isEqualTo("delay duration=1s"); } }
