This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26817-suggest-expectation in repository https://gitbox.apache.org/repos/asf/camel.git
commit cf5493ae1f5d67e5e20d14541859a05677b08e4e Author: Claus Ibsen <[email protected]> AuthorDate: Tue Nov 10 21:09:56 2026 +0100 CAMEL-26817: camel-chaos - the files a file route took can be put back (replay), so an experiment is run again with the same messages Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../org/apache/camel/catalog/docs/chaos.adoc | 15 +- components/camel-chaos/src/main/docs/chaos.adoc | 15 +- .../apache/camel/component/chaos/ChaosAdvisor.java | 107 +++++++++++++- .../apache/camel/component/chaos/ChaosManager.java | 86 ++++++++++- .../camel/component/chaos/ChaosReplayTest.java | 157 +++++++++++++++++++++ .../modules/ROOT/pages/camel-jbang-managing.adoc | 5 + .../apache/camel/cli/connector/ChaosAction.java | 14 +- .../LocalCliConnectorChaosActionTest.java | 15 ++ .../dsl/jbang/core/common/ChaosExperimentRun.java | 51 ++++++- .../jbang/core/common/ChaosExperimentRunTest.java | 24 ++++ 10 files changed, 477 insertions(+), 12 deletions(-) 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 8c62dc8519ab..3cf17107f73d 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 @@ -245,6 +245,13 @@ system: the CLI says so, and `real` is `true` in the JSON. Run with `--stub=remo (`timeout`) |An HTTP call, in a route with a circuit breaker |The service is down for a while (`outage`): does the breaker open, and does the fallback answer? +|An HTTP call that is known only when it sends (`toD`) |By its address up to the first expression +(`endpoint=http://stock/items/*` for `+http://stock/items/${variable.sku}+`): the service answers 503 (or is down for a +while, with a circuit breaker), and the resource is not there (`status=404`). An address that starts with an +expression says nothing of where it goes, and is left out +|A call to another route (`direct`), in a route with a circuit breaker |That route stands for the service: it is down +for a while (`outage`), and half of the calls fail (`rate=0.5`), past the failure rate of the breaker. Does the breaker +open, and does the fallback answer? |A call over another protocol |The connection is refused, and the call is slow |An input that can deliver a message again (Kafka, JMS, AMQP, SQS and more, or a consumer that reads a failed message again, such as a file without `moveFailed`) |The call out is made twice (`duplicate`): is it safe to repeat? @@ -291,7 +298,13 @@ The rules leave the query of the endpoint out: its options do not tell which end `why`, `tried`, `routes` for a dead letter channel that several routes share, `feed`, and `rules` for two faults at once. A `feed` has its `kind` (one of `self`, `http`, `file` and `send`), a `hint`, `real`, and `fed` when the route gets messages without a feed: it makes its own (a timer), or a route that does sends to it, also with an HTTP call to -the server of the integration. Such a suggestion can run as an experiment without sending anything. `rules` has all the +the server of the integration. Such a suggestion can run as an experiment without sending anything. A route that starts +from `file:<folder>` and moved the files it took (to `.camel`, or the folder of its `move` option) has `replay`, that +folder, and `files`, how many there are: `ChaosManager.replayFiles(route)` (the chaos command `replay`) puts them back +into its folder under their own names, so the route takes the same files again. The route moves them over the files +there, so a replay is the same each time, as an experiment that is run again needs. A route with `noop=true` never takes +a file of the same name again, and one with `delete=true` keeps none: they have no `replay`. A route that such a file +route sends to has them too, with `replayRoute`, the file route to replay for it. `rules` has all the rules to add together. The chaos screen of the Camel TUI lists the suggestions (key `w`), where Enter types the rule in to change or add (the rules of two faults at once separated by `;`). diff --git a/components/camel-chaos/src/main/docs/chaos.adoc b/components/camel-chaos/src/main/docs/chaos.adoc index 8c62dc8519ab..3cf17107f73d 100644 --- a/components/camel-chaos/src/main/docs/chaos.adoc +++ b/components/camel-chaos/src/main/docs/chaos.adoc @@ -245,6 +245,13 @@ system: the CLI says so, and `real` is `true` in the JSON. Run with `--stub=remo (`timeout`) |An HTTP call, in a route with a circuit breaker |The service is down for a while (`outage`): does the breaker open, and does the fallback answer? +|An HTTP call that is known only when it sends (`toD`) |By its address up to the first expression +(`endpoint=http://stock/items/*` for `+http://stock/items/${variable.sku}+`): the service answers 503 (or is down for a +while, with a circuit breaker), and the resource is not there (`status=404`). An address that starts with an +expression says nothing of where it goes, and is left out +|A call to another route (`direct`), in a route with a circuit breaker |That route stands for the service: it is down +for a while (`outage`), and half of the calls fail (`rate=0.5`), past the failure rate of the breaker. Does the breaker +open, and does the fallback answer? |A call over another protocol |The connection is refused, and the call is slow |An input that can deliver a message again (Kafka, JMS, AMQP, SQS and more, or a consumer that reads a failed message again, such as a file without `moveFailed`) |The call out is made twice (`duplicate`): is it safe to repeat? @@ -291,7 +298,13 @@ The rules leave the query of the endpoint out: its options do not tell which end `why`, `tried`, `routes` for a dead letter channel that several routes share, `feed`, and `rules` for two faults at once. A `feed` has its `kind` (one of `self`, `http`, `file` and `send`), a `hint`, `real`, and `fed` when the route gets messages without a feed: it makes its own (a timer), or a route that does sends to it, also with an HTTP call to -the server of the integration. Such a suggestion can run as an experiment without sending anything. `rules` has all the +the server of the integration. Such a suggestion can run as an experiment without sending anything. A route that starts +from `file:<folder>` and moved the files it took (to `.camel`, or the folder of its `move` option) has `replay`, that +folder, and `files`, how many there are: `ChaosManager.replayFiles(route)` (the chaos command `replay`) puts them back +into its folder under their own names, so the route takes the same files again. The route moves them over the files +there, so a replay is the same each time, as an experiment that is run again needs. A route with `noop=true` never takes +a file of the same name again, and one with `delete=true` keeps none: they have no `replay`. A route that such a file +route sends to has them too, with `replayRoute`, the file route to replay for it. `rules` has all the rules to add together. The chaos screen of the Camel TUI lists the suggestions (key `w`), where Enter types the rule in to change or add (the rules of two faults at once separated by `;`). diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosAdvisor.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosAdvisor.java index a2554f06aebb..4f65002691ee 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosAdvisor.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosAdvisor.java @@ -16,8 +16,12 @@ */ package org.apache.camel.component.chaos; +import java.io.File; +import java.net.URLDecoder; +import java.nio.charset.StandardCharsets; import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.Arrays; import java.util.Comparator; import java.util.Deque; import java.util.HashMap; @@ -133,12 +137,19 @@ final class ChaosAdvisor { /** * How to feed a route with messages (CAMEL-26771), so a rule can be tried on it. * - * @param kind {@code self} (the route feeds itself), {@code http}, {@code file} or {@code send} - * @param hint how, such as the command to run - * @param real whether the messages go to the real system the route reads, rather than a stub of it - * ({@code camel run --stub=remote}) + * @param kind {@code self} (the route feeds itself), {@code http}, {@code file} or {@code send} + * @param hint how, such as the command to run + * @param real whether the messages go to the real system the route reads, rather than a stub of it + * ({@code camel run --stub=remote}) + * @param replay the folder the route moved the files it took to, when it holds some: the chaos command replay puts + * them back, so the route takes the same files again (CAMEL-26817); null otherwise + * @param files how many files the replay puts back */ - record Feed(String kind, String hint, boolean real) { + record Feed(String kind, String hint, boolean real, String replay, int files) { + + Feed(String kind, String hint, boolean real) { + this(kind, hint, real, null, 0); + } } /** @@ -288,6 +299,15 @@ final class ChaosAdvisor { } if (LOCAL_FILES.contains(scheme)) { String folder = ChaosPatterns.withoutSlashes(uri(from)).substring(scheme.length() + 1); + File done = doneFolder(from); + List<File> taken = done != null ? takenFiles(done) : List.of(); + if (!taken.isEmpty()) { + return new Feed( + "file", "the " + taken.size() + (taken.size() == 1 ? " file" : " files") + " the route took, in " + + done + ", put back into " + folder + ": an experiment with --feed=" + route.getRouteId() + + " --body=replay:", + false, done.getAbsolutePath(), taken.size()); + } return new Feed("file", "write a file into " + folder, false); } // the route id as the endpoint: the send console sends to the input of the route, with all its options @@ -298,6 +318,61 @@ final class ChaosAdvisor { from.isRemote()); } + /** + * The folder a file input moves the files it took to (CAMEL-26817): {@code .camel} in its folder, or the folder of + * its {@code move} option. Null when it leaves them (noop, which also never takes a file of the same name again), + * deletes them, or moves them to a name made by an expression other than the file name. + */ + static File doneFolder(Endpoint from) { + String uri = from.getEndpointUri(); + if (!"file".equals(ChaosPatterns.scheme(uri))) { + return null; + } + Map<String, Object> options; + try { + options = URISupport.parseQuery(URISupport.extractQuery(uri)); + } catch (Exception e) { + return null; + } + if ("true".equals(String.valueOf(options.get("noop"))) || "true".equals(String.valueOf(options.get("delete")))) { + return null; + } + String move = ".camel"; + if (options.get("move") != null) { + try { + // as the endpoint URI has it, %24%7B for ${; a + stays a + + move = URLDecoder.decode(options.get("move").toString().replace("+", "%2B"), StandardCharsets.UTF_8); + } catch (IllegalArgumentException e) { + return null; + } + } + for (String name : List.of("${file:name}", "${file:onlyname}")) { + if (move.endsWith(name)) { + move = move.substring(0, move.length() - name.length()); + } + } + if (move.isEmpty() || move.contains("${") || move.contains("$simple{")) { + return null; + } + File done = new File(move); + return done.isAbsolute() ? done : new File(inputFolder(from), move); + } + + /** The folder a file input takes its files from, as its URI names it. */ + static File inputFolder(Endpoint from) { + return new File(ChaosPatterns.withoutSlashes(uri(from)).substring("file:".length())); + } + + /** The files in the folder a file input moved them to, by name: not the hidden ones, which it does not take. */ + static List<File> takenFiles(File folder) { + File[] files = folder.listFiles(f -> f.isFile() && !f.getName().startsWith(".")); + if (files == null) { + return List.of(); + } + Arrays.sort(files, Comparator.comparing(File::getName)); + return List.of(files); + } + /** * The route of the integration that receives a send: the route that consumes the endpoint (direct, seda), or for an * HTTP call to this machine (http, netty-http, undertow, vertx-http) the HTTP server route of its path, on its port @@ -838,6 +913,28 @@ final class ChaosAdvisor { return reached != null ? reached.keySet() : null; } + /** + * Per route, the file route whose files, put back, reach it (CAMEL-26817): a file route that took files reaches + * itself, and the routes it sends to, also over HTTP to the integration's own server. A route reached through a + * send known only when it sends is left out. + */ + static Map<String, String> replayVia(CamelContext camelContext) { + Map<String, String> answer = new HashMap<>(); + for (Route route : camelContext.getRoutes()) { + Endpoint from = route.getEndpoint(); + File done = from != null ? doneFolder(from) : null; + if (done == null || takenFiles(done).isEmpty()) { + continue; + } + answer.putIfAbsent(route.getRouteId(), route.getRouteId()); + Map<String, Boolean> reached = reach(camelContext, List.of(route.getRouteId()), true); + if (reached != null) { + reached.keySet().forEach(r -> answer.putIfAbsent(r, route.getRouteId())); + } + } + return answer; + } + /** * Why an experiment says nothing because its feed does not reach the routes of its expectations (CAMEL-26803), or * null when it reaches one, or that is not known. A route of an expectation gets messages when the route the feed 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 e3d50ac64061..5c2b5ce7df33 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 @@ -16,6 +16,11 @@ */ package org.apache.camel.component.chaos; +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; @@ -79,6 +84,9 @@ public final class ChaosManager { // the most removed, replaced or ended rules the report keeps static final int MAX_HISTORY = 100; + // the most files a replay puts back at a time + static final int MAX_REPLAY = 1000; + private static final Logger LOG = LoggerFactory.getLogger(ChaosManager.class); private final CamelContext camelContext; @@ -532,7 +540,9 @@ public final class ChaosManager { * @return {@code suggestions}: a list of {@code route}, {@code kind}, {@code recipe}, {@code title}, {@code rule}, * {@code why}, {@code tried} when a rule of its kind faulted the route already, and {@code routes} when it * is for several routes (a dead letter channel they share), and the {@code feed} of the route, with - * {@code fed} when it gets messages without one + * {@code fed} when it gets messages without one, and {@code replay} (a folder), {@code files} and + * {@code replayRoute} when the files the route took, or a file route that sends to it took, can be put back + * ({@link #replayFiles(String)} of that route) */ public Map<String, Object> suggest() { JsonArray list = new JsonArray(); @@ -549,6 +559,8 @@ public final class ChaosManager { Map<String, ChaosAdvisor.Feed> feeds = new HashMap<>(); // the routes that get messages without a feed, so an experiment can run without one (null when not known) Set<String> fed = ChaosAdvisor.fedRoutes(camelContext); + // the routes that get messages when a file route takes the files it took again + Map<String, String> replayVia = ChaosAdvisor.replayVia(camelContext); for (ChaosAdvisor.Suggestion s : ChaosAdvisor.suggest(camelContext)) { JsonObject jo = new JsonObject(); // a dead letter channel that routes share is tried by a rule of its kind in any of them; two faults at once @@ -584,6 +596,21 @@ public final class ChaosManager { if ("self".equals(feed.kind()) || fed != null && fed.contains(s.route())) { f.put("fed", true); } + // the files the route took, or a file route that sends to it took, which the chaos command replay puts + // back (CAMEL-26817) + String via = feed.replay() != null ? s.route() : replayVia.get(s.route()); + ChaosAdvisor.Feed files = via != null + ? feeds.computeIfAbsent(via, k -> ChaosAdvisor.feed(camelContext.getRoute(k))) + : null; + if (files != null && files.replay() != null) { + if (!via.equals(s.route())) { + // the route gets its messages from the file route + f.put("hint", "route " + via + " sends to it: " + files.hint()); + } + f.put("replay", files.replay()); + f.put("files", files.files()); + f.put("replayRoute", via); + } jo.put("feed", f); } list.add(jo); @@ -593,6 +620,63 @@ public final class ChaosManager { return answer; } + /** + * Puts the files a file route took back into its folder, so it takes the same files again (CAMEL-26817): the files + * it moved to {@code .camel} (or the folder of its {@code move} option), under their own names. The route moves + * them there again, over the copies, so a replay is the same each time, as an experiment that is run again needs. + * Each is written under a hidden name first, which the route does not take, and then renamed. A file of that name + * that the route has not taken yet is left as it is. + * + * @param routeId the id of the route + * @return {@code route}, {@code folder} (where the files were), {@code files} (how many + * were put back) and {@code skipped} (those still waiting in the folder) + * @throws IllegalArgumentException when the route does not take files it moves, or has taken none + */ + public Map<String, Object> replayFiles(String routeId) { + Route route = routeId != null ? camelContext.getRoute(routeId) : null; + if (route == null) { + throw new IllegalArgumentException("There is no route " + routeId); + } + File done = route.getEndpoint() != null ? ChaosAdvisor.doneFolder(route.getEndpoint()) : null; + if (done == null) { + throw new IllegalArgumentException( + "Route " + routeId + " does not take files it moves to a folder (file: without noop or delete)"); + } + List<File> taken = ChaosAdvisor.takenFiles(done); + if (taken.isEmpty()) { + throw new IllegalArgumentException("Route " + routeId + " has taken no files yet: " + done + " is empty"); + } + if (taken.size() > MAX_REPLAY) { + taken = taken.subList(0, MAX_REPLAY); + } + File folder = ChaosAdvisor.inputFolder(route.getEndpoint()); + int files = 0; + int skipped = 0; + for (File file : taken) { + Path target = folder.toPath().resolve(file.getName()); + if (Files.exists(target)) { + skipped++; + continue; + } + Path hidden = folder.toPath().resolve(".chaos-" + file.getName()); + try { + Files.copy(file.toPath(), hidden, StandardCopyOption.REPLACE_EXISTING); + Files.move(hidden, target, StandardCopyOption.ATOMIC_MOVE); + files++; + } catch (IOException e) { + throw new IllegalStateException("Cannot put back " + file + ": " + e.getMessage(), e); + } + } + Map<String, Object> answer = new JsonObject(); + answer.put("route", routeId); + answer.put("folder", done.getAbsolutePath()); + answer.put("files", files); + if (skipped > 0) { + answer.put("skipped", skipped); + } + return answer; + } + /** * Removes a rule. It stays in the report, and its expectation, if not decided yet, is cancelled. * diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosReplayTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosReplayTest.java new file mode 100644 index 000000000000..1d0aa41ab001 --- /dev/null +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosReplayTest.java @@ -0,0 +1,157 @@ +/* + * 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.io.File; +import java.nio.file.Files; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.util.FileUtil; +import org.apache.camel.util.json.JsonArray; +import org.apache.camel.util.json.JsonObject; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.awaitility.Awaitility.await; + +/** + * CAMEL-26817: the files a file route took, which it moved to .camel, are put back, so the route takes the same files + * again and an experiment can be run again with the same messages. + */ +class ChaosReplayTest extends ChaosTestSupport { + + private static final File BASE = new File("target/chaos-replay"); + + @BeforeEach + void files() throws Exception { + FileUtil.removeDir(BASE); + new File(BASE, "orders").mkdirs(); + // written under a hidden name, which the routes do not take, and then renamed + write("inbox/a.txt", "A"); + write("inbox/b.txt", "B"); + write("kept/c.txt", "C"); + } + + private static void write(String name, String body) throws Exception { + File file = new File(BASE, name); + file.getParentFile().mkdirs(); + File hidden = new File(file.getParentFile(), "." + file.getName()); + Files.writeString(hidden.toPath(), body); + Files.move(hidden.toPath(), file.toPath()); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("file:target/chaos-replay/inbox?initialDelay=0&delay=50").routeId("inbox").to("mock:inbox") + .to("seda:ship"); + from("seda:ship").routeId("ship").to("file:target/chaos-replay/archive"); + from("file:target/chaos-replay/orders?move=done/${file:name}&initialDelay=0&delay=50").routeId("orders") + .to("mock:orders"); + from("file:target/chaos-replay/kept?noop=true&initialDelay=0&delay=50").routeId("kept").to("mock:kept"); + } + }; + } + + @Test + void theFilesTheRouteTookArePutBack() throws Exception { + MockEndpoint inbox = getMockEndpoint("mock:inbox"); + inbox.expectedBodiesReceivedInAnyOrder("A", "B"); + MockEndpoint.assertIsSatisfied(context); + File done = new File(BASE, "inbox/.camel"); + await().atMost(10, TimeUnit.SECONDS).until(() -> ChaosAdvisor.takenFiles(done).size() == 2); + + // what to try says how: the files it took, put back + ChaosAdvisor.Feed feed = ChaosAdvisor.feed(context.getRoute("inbox")); + assertThat(feed.replay()).isEqualTo(done.getAbsolutePath()); + assertThat(feed.files()).isEqualTo(2); + assertThat(feed.hint()).contains("the 2 files the route took") + .endsWith(": an experiment with --feed=inbox --body=replay:"); + + inbox.reset(); + inbox.expectedBodiesReceivedInAnyOrder("A", "B"); + Map<String, Object> replayed = chaos().replayFiles("inbox"); + assertThat(replayed).containsEntry("route", "inbox").containsEntry("files", 2); + MockEndpoint.assertIsSatisfied(context); + + // moved over the files there: the same two, so a replay is the same each time + await().atMost(10, TimeUnit.SECONDS).until(() -> new File(BASE, "inbox").listFiles(File::isFile).length == 0); + assertThat(ChaosAdvisor.takenFiles(done)).extracting(File::getName).containsExactly("a.txt", "b.txt"); + } + + @Test + void theFolderOfTheMoveOption() throws Exception { + MockEndpoint orders = getMockEndpoint("mock:orders"); + orders.expectedMessageCount(0); + assertThat(ChaosAdvisor.feed(context.getRoute("orders")).replay()).isNull(); + assertThatThrownBy(() -> chaos().replayFiles("orders")).hasMessageContaining("has taken no files yet"); + + write("orders/1.json", "{}"); + orders.reset(); + orders.expectedMessageCount(1); + MockEndpoint.assertIsSatisfied(context); + File done = new File(BASE, "orders/done"); + await().atMost(10, TimeUnit.SECONDS).until(() -> ChaosAdvisor.takenFiles(done).size() == 1); + assertThat(ChaosAdvisor.feed(context.getRoute("orders")).replay()).isEqualTo(done.getAbsolutePath()); + + orders.reset(); + orders.expectedMessageCount(1); + assertThat(chaos().replayFiles("orders")).containsEntry("files", 1); + MockEndpoint.assertIsSatisfied(context); + } + + @Test + void aRouteThatLeavesItsFilesIsNotReplayed() throws Exception { + getMockEndpoint("mock:kept").expectedBodiesReceived("C"); + MockEndpoint.assertIsSatisfied(context); + // noop: the route never takes a file of the same name again + assertThat(ChaosAdvisor.feed(context.getRoute("kept")).replay()).isNull(); + assertThatThrownBy(() -> chaos().replayFiles("kept")).hasMessageContaining("does not take files it moves"); + assertThatThrownBy(() -> chaos().replayFiles("nope")).hasMessageContaining("There is no route nope"); + } + + @Test + void theSuggestionsSayHowToReplay() throws Exception { + getMockEndpoint("mock:inbox").expectedMessageCount(2); + MockEndpoint.assertIsSatisfied(context); + await().atMost(10, TimeUnit.SECONDS) + .until(() -> ChaosAdvisor.takenFiles(new File(BASE, "inbox/.camel")).size() == 2); + + List<Object> list = (JsonArray) chaos().suggest().get("suggestions"); + JsonObject feed = list.stream().map(JsonObject.class::cast).filter(s -> "inbox".equals(s.getString("route"))) + .map(s -> (JsonObject) s.get("feed")).findFirst().orElseThrow(); + assertThat(feed.getString("replay")).endsWith(".camel"); + assertThat(feed.getInteger("files")).isEqualTo(2); + assertThat(feed.getString("replayRoute")).isEqualTo("inbox"); + JsonObject ship = list.stream().map(JsonObject.class::cast).filter(s -> "ship".equals(s.getString("route"))) + .map(s -> (JsonObject) s.get("feed")).findFirst().orElseThrow(); + // the route the file route sends to gets the files too + assertThat(ship.getString("hint")).startsWith("route inbox sends to it: the 2 files the route took"); + assertThat(ship.getString("replayRoute")).isEqualTo("inbox"); + // a route the file route sends to gets the files too + assertThat(ChaosAdvisor.replayVia(context)).containsEntry("inbox", "inbox").containsEntry("ship", "inbox") + .doesNotContainKey("kept"); + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-jbang-managing.adoc b/docs/user-manual/modules/ROOT/pages/camel-jbang-managing.adoc index 0b120ab261b6..7e3f459c1f27 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-jbang-managing.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-jbang-managing.adoc @@ -981,6 +981,11 @@ one on another host is not fed. A route that starts from `file:<folder>` is fed with files: each message is written as a file into the folder the route polls, named so the route takes it: `chaos-<time>.json`, numbered as `chaos-<time>-<n>.json` when the messages are repeated, with the extension of its `include` or `antInclude` filter, else of the body. A `CamelFileName` in `--header` names the files instead. +With `--body=replay:` nothing is sent: the files the route took, which it moved to `.camel` (or the folder of its `move` +option), are put back into its folder under their own names, and again while the experiment runs, as the route takes +them, so a rule with a rate meets enough messages. The route moves them over the ones there, so an experiment that is +run again has the same messages. A route with `noop=true` never takes a file of +the same name again, so it cannot be replayed. A feed to a remote system that is not stubbed, such as a Kafka broker, has a warning, as the messages go to that system: use a local one (`camel infra run kafka`), or stub it. A broker that `camel infra run` started on this machine is such a diff --git a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/ChaosAction.java b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/ChaosAction.java index eebe24a97ea2..4c91957d2a80 100644 --- a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/ChaosAction.java +++ b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/ChaosAction.java @@ -141,6 +141,16 @@ final class ChaosAction { if (reset) { answer.put("reset", true); } + } else if ("replay".equals(command)) { + String route = root.getString("route"); + if (isBlank(route)) { + output.error("The id of the route is required"); + return; + } + // the files a file route took are put back, so it takes the same files again (CAMEL-26817) + @SuppressWarnings("unchecked") + Map<String, Object> replayed = (Map<String, Object>) call(type, chaos, "replayFiles", route); + answer.putAll(replayed); } else if ("explore-start".equals(command)) { // an exploration (CAMEL-26773): camel-chaos keeps its journal and ends its rules at the deadline @SuppressWarnings("unchecked") @@ -174,8 +184,8 @@ final class ChaosAction { answer.putAll(ended); } else { output.error("Unknown chaos command: " + command - + " (add, check, change, remove, clear, load, suggest, explore-start, explore-finish," - + " experiment-begin or experiment-end)"); + + " (add, check, change, remove, clear, load, suggest, replay, explore-start," + + " explore-finish, experiment-begin or experiment-end)"); return; } } catch (IllegalArgumentException | IllegalStateException e) { diff --git a/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorChaosActionTest.java b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorChaosActionTest.java index 1f41b807f294..4d5c6896e8ac 100644 --- a/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorChaosActionTest.java +++ b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorChaosActionTest.java @@ -223,6 +223,21 @@ class LocalCliConnectorChaosActionTest { assertThat(reportRules()).isEmpty(); } + @Test + void replayPutsBackTheFilesARouteTook() throws Exception { + start(null, Map.of("camel.chaos.enabled", "true")); + + // the route is required, and it must be a file route that moves the files it took + assertThat(chaos("replay", null, null).error()).isEqualTo("The id of the route is required"); + JsonObject action = new JsonObject(); + action.put("action", "chaos"); + action.put("command", "replay"); + action.put("route", "start"); + CliActionOutcome replayed = CliActionOutcome.dispatch(connector, (JsonObject) Jsoner.deserialize(action.toJson())) + .get(10, TimeUnit.SECONDS); + assertThat(replayed.error()).startsWith("Route start does not take files it moves to a folder"); + } + @Test void anExplorationStartsAndFinishes() throws Exception { start(null, Map.of("camel.chaos.enabled", "true")); diff --git a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRun.java b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRun.java index ff4c2c91b557..6b7be582cf7a 100644 --- a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRun.java +++ b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRun.java @@ -107,14 +107,31 @@ public final class ChaosExperimentRun { * The traffic of an experiment, as camel cmd send sends it. * * @param endpoint the endpoint or the id of the route to send to - * @param body the body, file:<path>, or file:<folder> for a message per file + * @param body the body, file:<path>, file:<folder> for a message per file, or {@link #REPLAY} for + * the files a file route took, put back into its folder (the endpoint is then the route) * @param headers the headers as key=value lines, or null * @param repeat how many messages in all; 0 to send once (each file of a folder once) * @param interval the pause between two messages, such as 200ms or a range such as 100..500, or null */ public record Feed(String endpoint, String body, String headers, int repeat, String interval) { + + /** + * The files the file route took, which it moved to .camel, put back into its folder under their own names, so + * the route takes the same files each time the experiment is run (CAMEL-26817). They are put back again while + * the experiment runs, as the route takes them, so a rule with a rate meets enough messages. + */ + public static Feed replay(String route) { + return new Feed(route, REPLAY, null, 0, null); + } + + boolean isReplay() { + return REPLAY.equals(body); + } } + /** The body of a feed that puts back the files a file route took ({@link Feed#replay(String)}). */ + public static final String REPLAY = "replay:"; + /** The time, and a pause; the tests give a clock that does not wait. */ interface Clock { long now(); @@ -189,7 +206,9 @@ public final class ChaosExperimentRun { http = ChaosHttpFeed.target(status, route, feed.endpoint()); // a file route reads the files its filter takes: the messages are written as such files String from = route != null ? route.getString("from") : null; - String name = from != null && from.startsWith("file:") ? fileName(from, feed, clock.now()) : null; + // the files of a replay keep their own names + String name = from != null && from.startsWith("file:") && !feed.isReplay() + ? fileName(from, feed, clock.now()) : null; if (name != null) { feed = new Feed( feed.endpoint(), feed.body(), @@ -233,6 +252,8 @@ public final class ChaosExperimentRun { } } JsonObject traffic = null; + // the file route whose files are put back, again and again while the experiment runs + String replayRoute = null; long waited = 0; ChaosHttpFeed.Sender sender = null; Thread sending = null; @@ -241,6 +262,9 @@ public final class ChaosExperimentRun { // sent while the experiment waits for its verdict, and stopped when it ends sender = new ChaosHttpFeed.Sender(integration, http, feed, clock); sending = Thread.ofVirtual().name("chaos-http-feed").start(sender); + } else if (feed != null && feed.isReplay()) { + replayRoute = fedRoute != null ? fedRoute.getString("routeId") : feed.endpoint(); + traffic = replay(integration, replayRoute); } else if (feed != null) { traffic = traffic(integration, feed); } @@ -255,6 +279,14 @@ public final class ChaosExperimentRun { break; } clock.sleep(POLL); + if (replayRoute != null && "success".equals(traffic.getString("status"))) { + // a few files make a few messages, which a rule with a rate may miss: they are put back again while + // the experiment runs, those the route took since (a file still waiting is left as it is) + JsonObject again = replay(integration, replayRoute); + if ("success".equals(again.getString("status"))) { + traffic.put("files", traffic.getIntegerOrDefault("files", 0) + again.getIntegerOrDefault("files", 0)); + } + } } waited = clock.now() - start; } finally { @@ -388,6 +420,21 @@ public final class ChaosExperimentRun { return answer; } + /** Puts back the files the route took: the summary of the replay, or why it could not. */ + private static JsonObject replay(Integration integration, String route) { + JsonObject answer; + try { + answer = new JsonObject(integration.chaos("replay", root -> root.put("route", route))); + answer.put("status", "success"); + } catch (RuntimeException e) { + answer = new JsonObject(); + answer.put("status", "failed"); + answer.put("error", e.getMessage()); + } + answer.put("endpoint", route); + return answer; + } + /** Ends the experiment: what came of it, also when the exploration ended it at its deadline meanwhile. */ private static JsonObject end(Integration integration, int number, String note, Clock clock) throws InterruptedException { diff --git a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRunTest.java b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRunTest.java index dc3ccf83eca9..eab6f8ea9935 100644 --- a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRunTest.java +++ b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/common/ChaosExperimentRunTest.java @@ -410,6 +410,30 @@ class ChaosExperimentRunTest { assertThat(integration.sent.headers()).isEqualTo("country=DK\nCamelFileName=chaos-1000000.json"); } + @Test + void aReplayPutsBackTheFilesTheRouteTookAfterTheRulesAreAdded() throws Exception { + JsonObject status = statusWith(experiment("PENDING", true)); + status.put("routes", new JsonArray( + List.of(new JsonObject(Map.of("routeId", "invoices", "from", "file://invoices"))))); + integration.statuses.add(status); + integration.shows("PENDING", true); + integration.shows("PASS", true); + + JsonObject result = ChaosExperimentRun.run(integration, null, RULE, ChaosExperimentRun.Feed.replay("invoices"), + 60_000, null, clock); + + // nothing is sent: the files keep their names, so the route takes the same files each time + assertThat(integration.sent).isNull(); + // again while the experiment runs, so a rule with a rate meets enough messages + assertThat(integration.actions).extracting(a -> a.getString("command")) + .containsSubsequence("experiment-begin", "replay", "replay", "experiment-end"); + assertThat(integration.actions).filteredOn(a -> "replay".equals(a.getString("command"))) + .extracting(a -> a.getString("route")).containsOnly("invoices"); + JsonObject traffic = (JsonObject) result.get("traffic"); + assertThat(traffic.getString("status")).isEqualTo("success"); + assertThat(traffic.getString("endpoint")).isEqualTo("invoices"); + } + @Test void theRouteTheFeedReachesGoesWithTheExperiment() throws Exception { JsonObject status = statusWith(experiment("PENDING", true));
