From 9a8257423032e2069b2837576d7c9504eb80e61f Mon Sep 17 00:00:00 2001 From: Abhishek Date: Wed, 16 Sep 2026 20:10:17 +0530 Subject: [PATCH 1/6] feat/r-script-commands-triggers --- AGENTS.md | 2 + plugin-script-r/build.gradle | 3 + .../plugin/scripts/r/CommandsTrigger.java | 294 ++++++++++++++++++ .../plugin/scripts/r/ScriptTrigger.java | 290 +++++++++++++++++ .../r/CommandsTriggerConditionTest.java | 65 ++++ .../plugin/scripts/r/CommandsTriggerTest.java | 115 +++++++ .../scripts/r/ScriptTriggerConditionTest.java | 59 ++++ .../plugin/scripts/r/ScriptTriggerTest.java | 95 ++++++ 8 files changed, 923 insertions(+) create mode 100644 plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/CommandsTrigger.java create mode 100644 plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/ScriptTrigger.java create mode 100644 plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerConditionTest.java create mode 100644 plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerTest.java create mode 100644 plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerConditionTest.java create mode 100644 plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerTest.java diff --git a/AGENTS.md b/AGENTS.md index 84c7163d..f7458d3f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -108,7 +108,9 @@ This is a **multi-module** plugin with 19 submodules: **plugin-script-r:** - `io.kestra.plugin.scripts.r.Commands` +- `io.kestra.plugin.scripts.r.CommandsTrigger` - `io.kestra.plugin.scripts.r.Script` +- `io.kestra.plugin.scripts.r.ScriptTrigger` **plugin-script-ruby:** - `io.kestra.plugin.scripts.ruby.Commands` diff --git a/plugin-script-r/build.gradle b/plugin-script-r/build.gradle index cefb63f3..acccc504 100644 --- a/plugin-script-r/build.gradle +++ b/plugin-script-r/build.gradle @@ -16,4 +16,7 @@ dependencies { implementation project(':plugin-script') testImplementation project(path: ':plugin-script', configuration: 'testOutput') + + testImplementation group: "io.kestra", name: "scheduler", version: kestraVersion + testImplementation group: "io.kestra", name: "worker", version: kestraVersion } diff --git a/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/CommandsTrigger.java b/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/CommandsTrigger.java new file mode 100644 index 00000000..432d5f21 --- /dev/null +++ b/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/CommandsTrigger.java @@ -0,0 +1,294 @@ +package io.kestra.plugin.scripts.r; + +import io.kestra.core.models.annotations.Example; +import io.kestra.core.models.annotations.Plugin; +import io.kestra.core.models.annotations.PluginProperty; +import io.kestra.core.models.conditions.ConditionContext; +import io.kestra.core.models.executions.Execution; +import io.kestra.core.models.property.Property; +import io.kestra.core.models.tasks.RunnableTaskException; +import io.kestra.core.models.tasks.runners.TaskException; +import io.kestra.core.models.triggers.*; +import io.kestra.core.runners.RunContext; +import io.kestra.plugin.scripts.exec.TriggerRunContext; +import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput; +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotNull; +import lombok.*; +import lombok.experimental.SuperBuilder; + +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +@SuperBuilder +@ToString +@EqualsAndHashCode +@Getter +@NoArgsConstructor +@Schema( + title = "Trigger a flow when R commands match a condition", + description = "Polls by running R commands in a container (default image 'r-base') and starts the flow when their result matches the condition." +) +@Plugin( + examples = { + @Example( + title = "Trigger when an R command fails.", + full = true, + code = """ + id: r_commands_trigger + namespace: company.team + + triggers: + - id: on_fail + type: io.kestra.plugin.scripts.r.CommandsTrigger + interval: PT5S + exitCondition: "exit 1" + commands: + - Rscript -e 'stop("boom")' + + tasks: + - id: log + type: io.kestra.plugin.core.log.Log + message: "Triggered with exitCode={{ trigger.exitCode }} (condition={{ trigger.condition }})" + """ + ) + } +) +// TODO: extract shared trigger logic (evaluate, matchesCondition, extractFailure, Output) +// into an AbstractScriptTrigger in plugin-script to reduce duplication across Shell, Node, Ruby, R, etc. +public class CommandsTrigger extends AbstractTrigger + implements PollingTriggerInterface, TriggerOutput { + + private static final String DEFAULT_IMAGE = "r-base"; + + private static final Pattern EXIT_CONDITION_PATTERN = + Pattern.compile("^\\s*exit\\s+(\\d+)\\s*$", Pattern.CASE_INSENSITIVE); + + @Schema( + title = "Docker image used to execute the commands", + description = """ + Container image used by the underlying Commands task to run R commands. + Defaults to 'r-base'. + """ + ) + @Builder.Default + @PluginProperty(group = "execution") + protected Property containerImage = Property.ofValue(DEFAULT_IMAGE); + + @Schema( + title = "R commands to execute", + description = "Commands executed in order on each poll." + ) + @NotNull + @PluginProperty(group = "main") + protected Property> commands; + + @Schema( + title = "Condition to match", + description = """ + Condition evaluated after execution. + + Supported forms: + - 'exit N' + - regex / substring matched against vars + logs + """ + ) + @NotNull + @PluginProperty(group = "main") + protected Property exitCondition; + + @Schema( + title = "Check interval", + description = "Interval between polling evaluations." + ) + @Builder.Default + @PluginProperty(group = "execution") + private final Duration interval = Duration.ofSeconds(60); + + @Schema( + title = "Edge trigger mode", + description = """ + If true, the trigger emits only on a transition from 'not matching' to 'matching' (anti-spam). + If false, the trigger emits on every poll where the condition matches. + """ + ) + @Builder.Default + @PluginProperty(group = "advanced") + protected Property edge = Property.ofValue(true); + + // Known limitation: in-memory only — resets when the trigger is rehydrated (e.g. after restart), + // so edge mode may re-fire once after a scheduler restart. + @Builder.Default + @Getter(AccessLevel.NONE) + private final AtomicBoolean lastMatched = new AtomicBoolean(false); + + @Override + public Optional evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception { + RunContext runContext = conditionContext.getRunContext(); + boolean edgeEnabled = runContext.render(this.edge).as(Boolean.class).orElse(true); + + Output out; + try { + out = runOnce(runContext); + } catch (Exception e) { + runContext.logger().warn("Trigger evaluation failed, returning empty result to avoid blocking the scheduler", e); + return Optional.empty(); + } + + boolean matched = matchesCondition(out); + + boolean emit = edgeEnabled + ? (!lastMatched.getAndSet(matched) && matched) + : matched; + + if (!emit) { + return Optional.empty(); + } + + return Optional.of( + TriggerService.generateExecution(this, conditionContext, context, out) + ); + } + + private Output runOnce(RunContext runContext) throws Exception { + Commands task = Commands.builder() + .id(this.getId()) + .type(Commands.class.getName()) + .containerImage(this.containerImage) + .commands(this.commands) + .build(); + + String renderedCondition = runContext.render(this.exitCondition) + .as(String.class) + .orElse(""); + + try { + ScriptOutput taskOutput = task.run(TriggerRunContext.forEmbeddedTask(runContext, task)); + + return new Output( + Instant.now(), + renderedCondition, + safeExitCode(taskOutput), + safeVars(taskOutput) + ); + } catch (RunnableTaskException e) { + ExtractedFailure failure = extractFailure(e); + return new Output( + Instant.now(), + renderedCondition, + failure.exitCode, + null + ); + } + } + + boolean matchesCondition(Output out) { + String cond = out.getCondition() == null ? "" : out.getCondition().trim(); + + Matcher exitMatcher = EXIT_CONDITION_PATTERN.matcher(cond); + + if (exitMatcher.matches()) { + int expected = Integer.parseInt(exitMatcher.group(1)); + return out.getExitCode() != null && out.getExitCode() == expected; + } + + String haystack = buildHaystack(out); + if (haystack.isEmpty() || cond.isEmpty()) { + return false; + } + + try { + // Guard against catastrophic backtracking (ReDoS) from user-supplied patterns + var pattern = Pattern.compile(cond); + var future = CompletableFuture.supplyAsync( + () -> pattern.matcher(haystack).find() + ); + return future.get(5, TimeUnit.SECONDS); + } catch (TimeoutException te) { + return haystack.contains(cond); + } catch (Exception e) { + return haystack.contains(cond); + } + } + + private String buildHaystack(Output out) { + if (out.getVars() == null || out.getVars().isEmpty()) { + return ""; + } + // Map.toString() produces {key=value, ...} — intentional for substring/regex matching. + return out.getVars().toString(); + } + + private Integer safeExitCode(ScriptOutput taskOutput) { + try { + return taskOutput.getExitCode(); + } catch (Exception ignored) { + return null; + } + } + + private Map safeVars(ScriptOutput taskOutput) { + try { + return taskOutput.getVars(); + } catch (Exception ignored) { + return null; + } + } + + private record ExtractedFailure(Integer exitCode) {} + + private ExtractedFailure extractFailure(RunnableTaskException e) { + Integer exitCode = null; + + Throwable cur = e.getCause(); + while (cur != null) { + if (cur instanceof TaskException te) { + exitCode = te.getExitCode(); + break; + } + cur = cur.getCause(); + } + + return new ExtractedFailure(exitCode); + } + + @Data + @AllArgsConstructor + public static class Output implements io.kestra.core.models.tasks.Output { + @Schema( + title = "Poll timestamp", + description = "Timestamp when this trigger evaluation occurred." + ) + private Instant timestamp; + + @Schema( + title = "Rendered condition", + description = "Rendered value of the exitCondition property for this poll." + ) + private String condition; + + @Schema( + title = "Commands exit code", + description = "Exit code returned by the R process (may be null if not available)." + ) + private Integer exitCode; + + @Schema( + title = "Commands vars", + description = """ + Vars produced by the task (e.g. via ::{"outputs":{...}}:: convention). This is the main structured + way to evaluate non-exit conditions on successful runs. + """ + ) + private Map vars; + } +} diff --git a/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/ScriptTrigger.java b/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/ScriptTrigger.java new file mode 100644 index 00000000..34c99e45 --- /dev/null +++ b/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/ScriptTrigger.java @@ -0,0 +1,290 @@ +package io.kestra.plugin.scripts.r; + +import io.kestra.core.models.annotations.Example; +import io.kestra.core.models.annotations.Plugin; +import io.kestra.core.models.annotations.PluginProperty; +import io.kestra.core.models.conditions.ConditionContext; +import io.kestra.core.models.enums.MonacoLanguages; +import io.kestra.core.models.executions.Execution; +import io.kestra.core.models.property.Property; +import io.kestra.core.models.tasks.RunnableTaskException; +import io.kestra.core.models.tasks.runners.TaskException; +import io.kestra.core.models.triggers.*; +import io.kestra.core.runners.RunContext; +import io.kestra.plugin.scripts.exec.TriggerRunContext; +import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput; +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotNull; +import lombok.*; +import lombok.experimental.SuperBuilder; + +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +@SuperBuilder +@ToString +@EqualsAndHashCode +@Getter +@NoArgsConstructor +@Schema( + title = "Trigger a flow when an R script matches a condition", + description = "Polls by running an inline R script in a container (default image 'r-base') and starts the flow when its result matches the condition." +) +@Plugin( + examples = { + @Example( + title = "Trigger when the script fails with exit code 1.", + full = true, + code = """ + id: r_script_trigger + namespace: company.team + + triggers: + - id: script_failure + type: io.kestra.plugin.scripts.r.ScriptTrigger + interval: PT10S + exitCondition: "exit 1" + edge: true + script: | + stop("boom") + + tasks: + - id: log + type: io.kestra.plugin.core.log.Log + message: "Triggered with exitCode={{ trigger.exitCode }} (condition={{ trigger.condition }})" + """ + ) + } +) +// TODO: extract shared trigger logic (evaluate, matchesCondition, extractFailure, Output) +// into an AbstractScriptTrigger in plugin-script to reduce duplication across Shell, Node, Ruby, R, etc. +public class ScriptTrigger extends AbstractTrigger + implements PollingTriggerInterface, TriggerOutput { + + private static final String DEFAULT_IMAGE = "r-base"; + + private static final Pattern EXIT_CONDITION_PATTERN = + Pattern.compile("^\\s*exit\\s+(\\d+)\\s*$", Pattern.CASE_INSENSITIVE); + + @Schema( + title = "Container image for script execution", + description = "Image used by the Script task to run the inline R script; defaults to 'r-base'. Provide an image that includes the required CRAN packages, or install them in the script itself." + ) + @Builder.Default + @PluginProperty(group = "execution") + protected Property containerImage = Property.ofValue(DEFAULT_IMAGE); + + @Schema( + title = "Inline R script", + description = "Multi-line R script executed on each poll, with the same semantics as the R Script task." + ) + @NotNull + @PluginProperty(language = MonacoLanguages.R, group = "main") + protected Property script; + + @Schema( + title = "Condition to match", + description = """ + Condition evaluated after each execution. The trigger emits only when it matches. + 'exit N' compares the exit code, otherwise the string is used as a regex + (or substring fallback) against emitted vars and failure logs. + """ + ) + @NotNull + @PluginProperty(group = "main") + protected Property exitCondition; + + @Schema( + title = "Check interval", + description = "Interval between polling evaluations." + ) + @Builder.Default + @PluginProperty(group = "execution") + private final Duration interval = Duration.ofSeconds(60); + + @Schema( + title = "Edge trigger mode", + description = """ + If true, the trigger emits only on a transition from 'not matching' to 'matching' (anti-spam). + If false, the trigger emits on every poll where the condition matches. + """ + ) + @Builder.Default + @PluginProperty(group = "advanced") + protected Property edge = Property.ofValue(true); + + // Known limitation: in-memory only — resets when the trigger is rehydrated (e.g. after restart), + // so edge mode may re-fire once after a scheduler restart. + @Getter(AccessLevel.NONE) + @Builder.Default + private final AtomicBoolean lastMatched = new AtomicBoolean(false); + + @Override + public Optional evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception { + RunContext runContext = conditionContext.getRunContext(); + boolean edgeEnabled = runContext.render(this.edge).as(Boolean.class).orElse(true); + + Output output; + try { + output = runOnce(runContext); + } catch (Exception e) { + runContext.logger().warn("Trigger evaluation failed, returning empty result to avoid blocking the scheduler", e); + return Optional.empty(); + } + + boolean matched = matchesCondition(output); + + boolean emit = edgeEnabled + ? (!lastMatched.getAndSet(matched) && matched) + : matched; + + if (!emit) { + return Optional.empty(); + } + + return Optional.of( + TriggerService.generateExecution(this, conditionContext, context, output) + ); + } + + private Output runOnce(RunContext runContext) throws Exception { + Script task = Script.builder() + .id(this.getId()) + .type(Script.class.getName()) + .containerImage(this.containerImage) + .script(this.script) + .build(); + + String renderedCondition = runContext.render(this.exitCondition) + .as(String.class) + .orElse(""); + + try { + ScriptOutput taskOutput = task.run(TriggerRunContext.forEmbeddedTask(runContext, task)); + + return new Output( + Instant.now(), + renderedCondition, + safeExitCode(taskOutput), + safeVars(taskOutput) + ); + } catch (RunnableTaskException e) { + ExtractedFailure failure = extractFailure(e); + return new Output( + Instant.now(), + renderedCondition, + failure.exitCode, + null + ); + } + } + + boolean matchesCondition(Output out) { + String cond = out.getCondition() == null ? "" : out.getCondition().trim(); + + Matcher exitMatcher = EXIT_CONDITION_PATTERN.matcher(cond); + + if (exitMatcher.matches()) { + int expected = Integer.parseInt(exitMatcher.group(1)); + return out.getExitCode() != null && out.getExitCode() == expected; + } + + String haystack = buildHaystack(out); + if (haystack.isEmpty() || cond.isEmpty()) { + return false; + } + + try { + // Guard against catastrophic backtracking (ReDoS) from user-supplied patterns + var pattern = Pattern.compile(cond); + var future = CompletableFuture.supplyAsync( + () -> pattern.matcher(haystack).find() + ); + return future.get(5, TimeUnit.SECONDS); + } catch (TimeoutException te) { + return haystack.contains(cond); + } catch (Exception e) { + return haystack.contains(cond); + } + } + + private String buildHaystack(Output out) { + if (out.getVars() == null || out.getVars().isEmpty()) { + return ""; + } + // Map.toString() produces {key=value, ...} — intentional for substring/regex matching. + return out.getVars().toString(); + } + + private Integer safeExitCode(ScriptOutput output) { + try { + return output.getExitCode(); + } catch (Exception ignored) { + return null; + } + } + + private Map safeVars(ScriptOutput output) { + try { + return output.getVars(); + } catch (Exception ignored) { + return null; + } + } + + private record ExtractedFailure(Integer exitCode) {} + + private ExtractedFailure extractFailure(RunnableTaskException e) { + Integer exitCode = null; + + Throwable cur = e.getCause(); + while (cur != null) { + if (cur instanceof TaskException te) { + exitCode = te.getExitCode(); + break; + } + cur = cur.getCause(); + } + + return new ExtractedFailure(exitCode); + } + + @Data + @AllArgsConstructor + public static class Output implements io.kestra.core.models.tasks.Output { + @Schema( + title = "Poll timestamp", + description = "Timestamp when this trigger evaluation occurred." + ) + private Instant timestamp; + + @Schema( + title = "Rendered condition", + description = "Rendered value of the exitCondition property for this poll." + ) + private String condition; + + @Schema( + title = "Script exit code", + description = "Exit code returned by the R process (may be null if not available)." + ) + private Integer exitCode; + + @Schema( + title = "Script vars", + description = """ + Vars produced by the task (e.g. via ::{"outputs":{...}}:: convention). This is the main structured + way to evaluate non-exit conditions on successful runs. + """ + ) + private Map vars; + } +} diff --git a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerConditionTest.java b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerConditionTest.java new file mode 100644 index 00000000..2a120568 --- /dev/null +++ b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerConditionTest.java @@ -0,0 +1,65 @@ +package io.kestra.plugin.scripts.r; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.time.Instant; +import java.util.Map; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; + +class CommandsTriggerConditionTest { + + private final CommandsTrigger trigger = CommandsTrigger.builder().build(); + + private CommandsTrigger.Output output(String condition, Integer exitCode, Map vars) { + return new CommandsTrigger.Output(Instant.now(), condition, exitCode, vars); + } + + @ParameterizedTest + @CsvSource({ + "exit 0, 0, true", + "exit 1, 1, true", + "EXIT 1, 1, true", + "exit 0, 1, false", + "exit 1, 0, false", + "exit 42, 42, true", + }) + void exitCodeCondition(String condition, int exitCode, boolean expected) { + assertThat(trigger.matchesCondition(output(condition, exitCode, null)), is(expected)); + } + + @Test + void exitCondition_nullExitCode_doesNotMatch() { + assertThat(trigger.matchesCondition(output("exit 1", null, null)), is(false)); + } + + @Test + void substringMatch_inVars() { + assertThat(trigger.matchesCondition( + output("toto", 0, Map.of("key", "toto"))), is(true)); + } + + @Test + void regexMatch_inVars() { + assertThat(trigger.matchesCondition( + output("status=\\w+", 0, Map.of("status", "status=ready"))), is(true)); + } + + @Test + void noMatch_emptyHaystack() { + assertThat(trigger.matchesCondition(output("something", 0, null)), is(false)); + } + + @Test + void noMatch_emptyCondition() { + assertThat(trigger.matchesCondition(output("", 0, Map.of("k", "v"))), is(false)); + } + + @Test + void nullCondition_doesNotMatch() { + assertThat(trigger.matchesCondition(output(null, 0, Map.of("k", "v"))), is(false)); + } +} diff --git a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerTest.java b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerTest.java new file mode 100644 index 00000000..a99cbca4 --- /dev/null +++ b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerTest.java @@ -0,0 +1,115 @@ +package io.kestra.plugin.scripts.r; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.jupiter.api.Test; + +import io.kestra.core.junit.annotations.KestraTest; +import io.kestra.core.models.executions.Execution; +import io.kestra.core.models.property.Property; +import io.kestra.core.runners.RunContextFactory; +import io.kestra.core.utils.TestsUtils; + +import jakarta.inject.Inject; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.notNullValue; + +@KestraTest +class CommandsTriggerTest { + @Inject + private RunContextFactory runContextFactory; + + @Test + void commandsTrigger_shouldTriggerOnImplicitFailureExit1() throws Exception { + CommandsTrigger trigger = CommandsTrigger.builder() + .id("commands-trigger") + .type(CommandsTrigger.class.getName()) + .exitCondition(Property.ofValue("exit 1")) + .edge(Property.ofValue(true)) + .containerImage(Property.ofValue("r-base")) + .commands(Property.ofValue(List.of("Rscript -e 'quit(status = 1)'"))) + .build(); + + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + Optional execution = trigger.evaluate(context.getKey(), context.getValue()); + + assertThat(execution.isPresent(), is(true)); + + Map triggerVars = execution.get().getTrigger().getVariables(); + assertThat("condition should be present", triggerVars.get("condition"), is("exit 1")); + assertThat("exitCode should be present", triggerVars.get("exitCode"), notNullValue()); + assertThat("exitCode should be 1", triggerVars.get("exitCode"), is(1)); + assertThat("timestamp should be present", triggerVars.get("timestamp"), notNullValue()); + } + + @Test + void commandsTrigger_shouldTriggerOnStdoutMatchUsingStructuredOutputs() throws Exception { + CommandsTrigger trigger = CommandsTrigger.builder() + .id("commands-stdout-match-trigger") + .type(CommandsTrigger.class.getName()) + .exitCondition(Property.ofValue("toto")) + .edge(Property.ofValue(true)) + .containerImage(Property.ofValue("r-base")) + .commands(Property.ofValue(List.of("echo '::{\"outputs\":{\"listing\":\"toto\"}}::'"))) + .build(); + + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + Optional execution = trigger.evaluate(context.getKey(), context.getValue()); + + assertThat(execution.isPresent(), is(true)); + + Map triggerVars = execution.get().getTrigger().getVariables(); + assertThat("condition should be present", triggerVars.get("condition"), is("toto")); + assertThat("exitCode should be present", triggerVars.get("exitCode"), notNullValue()); + assertThat("exitCode should be 0", triggerVars.get("exitCode"), is(0)); + assertThat("timestamp should be present", triggerVars.get("timestamp"), notNullValue()); + assertThat("vars should be present", triggerVars.get("vars"), notNullValue()); + } + + @Test + void commandsTrigger_shouldNotEmitWhenConditionDoesNotMatch() throws Exception { + CommandsTrigger trigger = CommandsTrigger.builder() + .id("commands-no-match-trigger") + .type(CommandsTrigger.class.getName()) + .exitCondition(Property.ofValue("exit 1")) + .edge(Property.ofValue(true)) + .containerImage(Property.ofValue("r-base")) + .commands(Property.ofValue(List.of("Rscript -e 'quit(status = 0)'"))) + .build(); + + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + Optional execution = trigger.evaluate(context.getKey(), context.getValue()); + + assertThat("successful run should not match 'exit 1'", execution.isPresent(), is(false)); + } + + @Test + void edgeMode_preventsConsecutiveEmit() { + AtomicBoolean lastMatched = new AtomicBoolean(false); + + // First match: transition false->true => should emit + boolean matched1 = true; + boolean emit1 = !lastMatched.getAndSet(matched1) && matched1; + assertThat("first match should emit", emit1, is(true)); + + // Second consecutive match: true->true => should NOT emit + boolean matched2 = true; + boolean emit2 = !lastMatched.getAndSet(matched2) && matched2; + assertThat("consecutive match should NOT emit in edge mode", emit2, is(false)); + + // Non-match: true->false => should not emit + boolean matched3 = false; + boolean emit3 = !lastMatched.getAndSet(matched3) && matched3; + assertThat("non-match should not emit", emit3, is(false)); + + // Match again after non-match: false->true => should emit + boolean matched4 = true; + boolean emit4 = !lastMatched.getAndSet(matched4) && matched4; + assertThat("match after non-match should emit", emit4, is(true)); + } +} diff --git a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerConditionTest.java b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerConditionTest.java new file mode 100644 index 00000000..9779f859 --- /dev/null +++ b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerConditionTest.java @@ -0,0 +1,59 @@ +package io.kestra.plugin.scripts.r; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.time.Instant; +import java.util.Map; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; + +class ScriptTriggerConditionTest { + + private final ScriptTrigger trigger = ScriptTrigger.builder().build(); + + private ScriptTrigger.Output output(String condition, Integer exitCode, Map vars) { + return new ScriptTrigger.Output(Instant.now(), condition, exitCode, vars); + } + + @ParameterizedTest + @CsvSource({ + "exit 0, 0, true", + "exit 1, 1, true", + "EXIT 1, 1, true", + "exit 0, 1, false", + "exit 1, 0, false", + "exit 42, 42, true", + }) + void exitCodeCondition(String condition, int exitCode, boolean expected) { + assertThat(trigger.matchesCondition(output(condition, exitCode, null)), is(expected)); + } + + @Test + void exitCondition_nullExitCode_doesNotMatch() { + assertThat(trigger.matchesCondition(output("exit 1", null, null)), is(false)); + } + + @Test + void substringMatch_inVars() { + assertThat(trigger.matchesCondition( + output("toto", 0, Map.of("key", "toto"))), is(true)); + } + + @Test + void noMatch_emptyHaystack() { + assertThat(trigger.matchesCondition(output("something", 0, null)), is(false)); + } + + @Test + void noMatch_emptyCondition() { + assertThat(trigger.matchesCondition(output("", 0, Map.of("k", "v"))), is(false)); + } + + @Test + void nullCondition_doesNotMatch() { + assertThat(trigger.matchesCondition(output(null, 0, Map.of("k", "v"))), is(false)); + } +} diff --git a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerTest.java b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerTest.java new file mode 100644 index 00000000..745e8bff --- /dev/null +++ b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerTest.java @@ -0,0 +1,95 @@ +package io.kestra.plugin.scripts.r; + +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; + +/** + * Unit tests for ScriptTrigger's condition-matching logic and edge mode. + * + * These tests exercise matchesCondition via the Output model without requiring an R + * runtime, which may not be available on all CI machines. + * Integration coverage against an actual R runtime lives in CommandsTriggerTest. + */ +class ScriptTriggerTest { + + private final ScriptTrigger trigger = ScriptTrigger.builder().build(); + + private ScriptTrigger.Output output(String condition, Integer exitCode, Map vars) { + return new ScriptTrigger.Output(Instant.now(), condition, exitCode, vars); + } + + @Test + void exitCodeCondition_shouldMatchWhenExitCodeEquals() { + assertThat(trigger.matchesCondition(output("exit 1", 1, null)), is(true)); + } + + @Test + void exitCodeCondition_shouldNotMatchWhenExitCodeDiffers() { + assertThat(trigger.matchesCondition(output("exit 1", 127, null)), is(false)); + } + + @Test + void exitCodeCondition_shouldNotMatchWhenExitCodeIsNull() { + assertThat(trigger.matchesCondition(output("exit 1", null, null)), is(false)); + } + + @Test + void substringCondition_shouldMatchAgainstVars() { + assertThat(trigger.matchesCondition(output("toto", 0, Map.of("listing", "toto"))), is(true)); + } + + @Test + void substringCondition_shouldNotMatchWhenAbsent() { + assertThat(trigger.matchesCondition(output("toto", 0, Map.of("listing", "something_else"))), is(false)); + } + + @Test + void regexCondition_shouldMatchAgainstVars() { + assertThat(trigger.matchesCondition(output("status=\\w+", 0, Map.of("status", "status=ready"))), is(true)); + } + + @Test + void emptyCondition_shouldNotMatch() { + assertThat(trigger.matchesCondition(output("", 0, null)), is(false)); + } + + @Test + void nullCondition_shouldNotMatch() { + assertThat(trigger.matchesCondition(output(null, 0, null)), is(false)); + } + + @Test + void exitZeroCondition_shouldMatchSuccessfulExecution() { + assertThat(trigger.matchesCondition(output("exit 0", 0, null)), is(true)); + } + + @Test + void edgeMode_shouldEmitOnFirstMatch() { + var lastMatched = new AtomicBoolean(false); + boolean matched = true; + boolean emit = !lastMatched.getAndSet(matched) && matched; + assertThat("first match should emit", emit, is(true)); + } + + @Test + void edgeMode_shouldSuppressConsecutiveMatches() { + var lastMatched = new AtomicBoolean(true); + boolean matched = true; + boolean emit = !lastMatched.getAndSet(matched) && matched; + assertThat("consecutive match should not emit in edge mode", emit, is(false)); + } + + @Test + void edgeMode_shouldEmitAgainAfterNonMatch() { + var lastMatched = new AtomicBoolean(false); + boolean matched = true; + boolean emit = !lastMatched.getAndSet(matched) && matched; + assertThat("match after non-match should emit", emit, is(true)); + } +} From 61ad5d185e8003b3ab6385cc7b7a98fa633f19ef Mon Sep 17 00:00:00 2001 From: Abhishek Date: Sun, 20 Sep 2026 15:12:22 +0530 Subject: [PATCH 2/6] feat/perl-script-commands-triggers --- AGENTS.md | 2 + plugin-script-perl/build.gradle | 3 + .../plugin/scripts/perl/CommandsTrigger.java | 275 +++++++++++++++++ .../plugin/scripts/perl/ScriptTrigger.java | 276 ++++++++++++++++++ .../perl/CommandsTriggerConditionTest.java | 65 +++++ .../scripts/perl/CommandsTriggerTest.java | 136 +++++++++ .../perl/ScriptTriggerConditionTest.java | 65 +++++ .../scripts/perl/ScriptTriggerTest.java | 95 ++++++ 8 files changed, 917 insertions(+) create mode 100644 plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java create mode 100644 plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/ScriptTrigger.java create mode 100644 plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerConditionTest.java create mode 100644 plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerTest.java create mode 100644 plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerConditionTest.java create mode 100644 plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerTest.java diff --git a/AGENTS.md b/AGENTS.md index f7458d3f..e8e370a6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -91,7 +91,9 @@ This is a **multi-module** plugin with 19 submodules: **plugin-script-perl:** - `io.kestra.plugin.scripts.perl.Commands` +- `io.kestra.plugin.scripts.perl.CommandsTrigger` - `io.kestra.plugin.scripts.perl.Script` +- `io.kestra.plugin.scripts.perl.ScriptTrigger` **plugin-script-php:** - `io.kestra.plugin.scripts.php.Commands` diff --git a/plugin-script-perl/build.gradle b/plugin-script-perl/build.gradle index 3bddfca5..2f8a80e6 100644 --- a/plugin-script-perl/build.gradle +++ b/plugin-script-perl/build.gradle @@ -16,4 +16,7 @@ dependencies { implementation project(':plugin-script') testImplementation project(path: ':plugin-script', configuration: 'testOutput') + + testImplementation group: "io.kestra", name: "scheduler", version: kestraVersion + testImplementation group: "io.kestra", name: "worker", version: kestraVersion } diff --git a/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java new file mode 100644 index 00000000..f4e2deb5 --- /dev/null +++ b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java @@ -0,0 +1,275 @@ +package io.kestra.plugin.scripts.perl; + +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import io.kestra.core.models.annotations.Example; +import io.kestra.core.models.annotations.Plugin; +import io.kestra.core.models.conditions.ConditionContext; +import io.kestra.core.models.executions.Execution; +import io.kestra.core.models.property.Property; +import io.kestra.core.models.tasks.RunnableTaskException; +import io.kestra.core.models.tasks.runners.TaskException; +import io.kestra.core.models.triggers.AbstractTrigger; +import io.kestra.core.models.triggers.PollingTriggerInterface; +import io.kestra.core.models.triggers.TriggerContext; +import io.kestra.core.models.triggers.TriggerOutput; +import io.kestra.core.models.triggers.TriggerService; +import io.kestra.core.runners.RunContext; +import io.kestra.plugin.scripts.exec.TriggerRunContext; +import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput; + +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotNull; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.ToString; +import lombok.experimental.SuperBuilder; +import io.kestra.core.models.annotations.PluginProperty; + +@SuperBuilder +@ToString +@EqualsAndHashCode +@Getter +@NoArgsConstructor +@Schema( + title = "Trigger a flow when Perl commands match a condition", + description = "Polls and triggers a flow by running Perl commands within a script container." +) +@Plugin( + examples = { + @Example( + title = "Trigger when commands fail with an implicit error (exit 1).", + full = true, + code = """ + id: commands_trigger + namespace: company.team + + triggers: + - id: commands_failure + type: io.kestra.plugin.scripts.perl.CommandsTrigger + interval: PT10S + exitCondition: "exit 1" + edge: true + containerImage: perl + commands: + - perl missing.pl + + tasks: + - id: log + type: io.kestra.plugin.core.log.Log + message: "Triggered with exitCode={{ trigger.exitCode }} (condition={{ trigger.condition }})" + """ + ) + } +) +public class CommandsTrigger extends AbstractTrigger + implements PollingTriggerInterface, TriggerOutput { + + private static final String DEFAULT_IMAGE = "perl"; + private static final Pattern EXIT_CONDITION_PATTERN = Pattern.compile("^\\s*exit\\s+(\\d+)\\s*$", Pattern.CASE_INSENSITIVE); + + @Schema( + title = "Docker image used to execute the commands", + description = """ + Container image used by the underlying Commands task to run Perl commands. + Defaults to 'perl'. + """ + ) + @Builder.Default + @PluginProperty(group = "execution") + protected Property containerImage = Property.ofValue(DEFAULT_IMAGE); + + @Schema( + title = "Perl commands to execute", + description = "Commands executed on each poll (same semantics as the Perl Commands task)." + ) + @NotNull + @PluginProperty(group = "main") + protected Property> commands; + + @Schema( + title = "Condition to match", + description = """ + Condition evaluated after each commands execution. The trigger emits an event only when this condition matches. + + Supported forms: + - 'exit N' (example: 'exit 1'): matches when the process exit code equals N. + - Any other string: treated as a regex (or substring if regex is invalid) matched against: + - the task 'vars' (when commands emit ::{"outputs":...}::), + - and error logs when the task fails (TaskException). + """ + ) + @NotNull + @PluginProperty(group = "main") + protected Property exitCondition; + + @Schema( + title = "Check interval", + description = "Interval between polling evaluations." + ) + @Builder.Default + @PluginProperty(group = "execution") + private final Duration interval = Duration.ofSeconds(60); + + @Schema( + title = "Edge trigger mode", + description = """ + If true, the trigger emits only on a transition from 'not matching' to 'matching' (anti-spam). + If false, the trigger emits on every poll where the condition matches. + """ + ) + @Builder.Default + @PluginProperty(group = "advanced") + protected Property edge = Property.ofValue(true); + + @Builder.Default + @Getter(AccessLevel.NONE) + private final AtomicBoolean lastMatched = new AtomicBoolean(false); + + @Override + public Optional evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception { + RunContext runContext = conditionContext.getRunContext(); + boolean renderedEdge = runContext.render(this.edge).as(Boolean.class).orElse(true); + + Output out; + try { + out = runOnce(runContext); + } catch (Exception e) { + runContext.logger().warn("Trigger evaluation failed, returning empty result to avoid blocking the scheduler", e); + return Optional.empty(); + } + + boolean matched = matchesCondition(out); + + boolean emit = renderedEdge + ? (!lastMatched.getAndSet(matched) && matched) + : matched; + + if (!emit) { + return Optional.empty(); + } + + return Optional.of(TriggerService.generateExecution(this, conditionContext, context, out)); + } + + private Output runOnce(RunContext runContext) throws Exception { + Commands task = Commands.builder() + .id(this.getId()) + .type(Commands.class.getName()) + .containerImage(this.containerImage) + .commands(this.commands) + .build(); + + String renderedCondition = runContext.render(this.exitCondition).as(String.class).orElse(""); + + try { + ScriptOutput taskOutput = task.run(TriggerRunContext.forEmbeddedTask(runContext, task)); + Integer exitCode = safeExitCode(taskOutput); + Map vars = safeVars(taskOutput); + + return new Output(Instant.now(), renderedCondition, exitCode, vars); + } catch (RunnableTaskException e) { + ExtractedFailure failure = extractFailure(e); + return new Output(Instant.now(), renderedCondition, failure.exitCode, null); + } + } + + boolean matchesCondition(Output out) { + String cond = out.getCondition() == null ? "" : out.getCondition().trim(); + + Matcher exitMatcher = EXIT_CONDITION_PATTERN.matcher(cond); + if (exitMatcher.matches()) { + int expected = Integer.parseInt(exitMatcher.group(1)); + return out.getExitCode() != null && out.getExitCode() == expected; + } + + String haystack = buildHaystack(out); + if (haystack.isEmpty() || cond.isEmpty()) { + return false; + } + + try { + return Pattern.compile(cond).matcher(haystack).find(); + } catch (Exception invalidRegex) { + return haystack.contains(cond); + } + } + + private String buildHaystack(Output out) { + if (out.getVars() == null || out.getVars().isEmpty()) { + return ""; + } + return out.getVars().toString(); + } + + private Integer safeExitCode(ScriptOutput taskOutput) { + try { + return taskOutput.getExitCode(); + } catch (Exception ignored) { + return null; + } + } + + private Map safeVars(ScriptOutput taskOutput) { + try { + return taskOutput.getVars(); + } catch (Exception ignored) { + return null; + } + } + + private record ExtractedFailure(Integer exitCode) { + } + + private ExtractedFailure extractFailure(RunnableTaskException e) { + Integer exitCode = null; + + Throwable cur = e.getCause(); + while (cur != null) { + if (cur instanceof TaskException te) { + exitCode = te.getExitCode(); + break; + } + cur = cur.getCause(); + } + + return new ExtractedFailure(exitCode); + } + + @Data + @AllArgsConstructor + public static class Output implements io.kestra.core.models.tasks.Output { + @Schema(title = "Timestamp of the event that fired the trigger") + private Instant timestamp; + + @Schema( + title = "Rendered condition", + description = "Rendered value of the exitCondition property for this poll." + ) + private String condition; + + @Schema( + title = "Commands exit code", + description = "Exit code returned by the Perl process (may be null if not available)." + ) + private Integer exitCode; + + @Schema( + title = "Commands vars", + description = "Vars produced by the task (e.g. via ::{\"outputs\":{...}}:: convention)." + ) + private Map vars; + } +} diff --git a/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/ScriptTrigger.java b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/ScriptTrigger.java new file mode 100644 index 00000000..9594d02f --- /dev/null +++ b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/ScriptTrigger.java @@ -0,0 +1,276 @@ +package io.kestra.plugin.scripts.perl; + +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import io.kestra.core.models.annotations.Example; +import io.kestra.core.models.annotations.Plugin; +import io.kestra.core.models.annotations.PluginProperty; +import io.kestra.core.models.conditions.ConditionContext; +import io.kestra.core.models.executions.Execution; +import io.kestra.core.models.property.Property; +import io.kestra.core.models.tasks.RunnableTaskException; +import io.kestra.core.models.tasks.runners.TaskException; +import io.kestra.core.models.triggers.AbstractTrigger; +import io.kestra.core.models.triggers.PollingTriggerInterface; +import io.kestra.core.models.triggers.TriggerContext; +import io.kestra.core.models.triggers.TriggerOutput; +import io.kestra.core.models.triggers.TriggerService; +import io.kestra.core.runners.RunContext; +import io.kestra.plugin.scripts.exec.TriggerRunContext; +import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput; + +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotNull; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.ToString; +import lombok.experimental.SuperBuilder; + +@SuperBuilder +@ToString +@EqualsAndHashCode +@Getter +@NoArgsConstructor +@Schema( + title = "Trigger on Perl script condition", + description = "Polls by running an inline Perl script in a container (default image perl) and emits when exitCondition matches. Supports edge mode to emit only on transitions and polls every 60s by default. Accepts 'exit N' or a regex (fallback substring) matched against emitted vars and failure logs." +) +@Plugin( + examples = { + @Example( + title = "Trigger when the script fails with an implicit error (exit 1).", + full = true, + code = """ + id: script_trigger + namespace: company.team + + triggers: + - id: script_failure + type: io.kestra.plugin.scripts.perl.ScriptTrigger + interval: PT10S + exitCondition: "exit 1" + edge: true + containerImage: perl + script: | + # This fails with a non-zero exit code. + exit 1; + + tasks: + - id: log + type: io.kestra.plugin.core.log.Log + message: "Triggered with exitCode={{ trigger.exitCode }} (condition={{ trigger.condition }})" + """ + ) + } +) +public class ScriptTrigger extends AbstractTrigger + implements PollingTriggerInterface, TriggerOutput { + + private static final String DEFAULT_IMAGE = "perl"; + private static final Pattern EXIT_CONDITION_PATTERN = Pattern.compile("^\\s*exit\\s+(\\d+)\\s*$", Pattern.CASE_INSENSITIVE); + + @Schema( + title = "Container image for script execution", + description = """ + Image used by the Script task to run the inline Perl script; defaults to 'perl'. + Provide an image that includes the Perl runtime and any required CPAN modules. + """ + ) + @Builder.Default + @PluginProperty(group = "execution") + protected Property containerImage = Property.ofValue(DEFAULT_IMAGE); + + @Schema( + title = "Inline Perl script", + description = """ + Multi-line Perl script executed on each poll, with the same semantics as the Perl Script task. + """ + ) + @NotNull + @PluginProperty(group = "main") + protected Property script; + + @Schema( + title = "Condition to match", + description = """ + Rendered condition evaluated after each execution; the trigger emits only when it matches. + 'exit N' compares the exit code, otherwise the string is used as a regex (or substring fallback) against emitted vars (from ::{"outputs":...}::) and failure logs. + """ + ) + @NotNull + @PluginProperty(group = "main") + protected Property exitCondition; + + @Schema( + title = "Check interval", + description = """ + Interval between polls; default PT60S. The scheduler uses this to schedule the next evaluation. + """ + ) + @Builder.Default + @PluginProperty(group = "execution") + private final Duration interval = Duration.ofSeconds(60); + + @Schema( + title = "Edge trigger mode", + description = """ + When true (default), emit only on a transition from not matching to matching. When false, emit on every poll that matches. + """ + ) + @Builder.Default + @PluginProperty(group = "advanced") + protected Property edge = Property.ofValue(true); + + @Builder.Default + @Getter(AccessLevel.NONE) + private final AtomicBoolean lastMatched = new AtomicBoolean(false); + + @Override + public Optional evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception { + RunContext runContext = conditionContext.getRunContext(); + boolean renderedEdge = runContext.render(this.edge).as(Boolean.class).orElse(true); + + Output out; + try { + out = runOnce(runContext); + } catch (Exception e) { + runContext.logger().warn("Trigger evaluation failed, returning empty result to avoid blocking the scheduler", e); + return Optional.empty(); + } + + boolean matched = matchesCondition(out); + + boolean emit = renderedEdge + ? (!lastMatched.getAndSet(matched) && matched) + : matched; + + if (!emit) { + return Optional.empty(); + } + + return Optional.of(TriggerService.generateExecution(this, conditionContext, context, out)); + } + + private Output runOnce(RunContext runContext) throws Exception { + Script task = Script.builder() + .id(this.getId()) + .type(Script.class.getName()) + .containerImage(this.containerImage) + .script(this.script) + .build(); + + String renderedExitCondition = runContext.render(this.exitCondition).as(String.class).orElse(""); + + try { + ScriptOutput taskOutput = task.run(TriggerRunContext.forEmbeddedTask(runContext, task)); + Integer exitCode = safeExitCode(taskOutput); + Map vars = safeVars(taskOutput); + + return new Output(Instant.now(), renderedExitCondition, exitCode, vars); + } catch (RunnableTaskException e) { + ExtractedFailure failure = extractFailure(e); + return new Output(Instant.now(), renderedExitCondition, failure.exitCode, null); + } + } + + boolean matchesCondition(Output out) { + String cond = out.getCondition() == null ? "" : out.getCondition().trim(); + + Matcher exitMatcher = EXIT_CONDITION_PATTERN.matcher(cond); + if (exitMatcher.matches()) { + int expected = Integer.parseInt(exitMatcher.group(1)); + return out.getExitCode() != null && out.getExitCode() == expected; + } + + String haystack = buildHaystack(out); + if (haystack.isEmpty() || cond.isEmpty()) { + return false; + } + + try { + return Pattern.compile(cond).matcher(haystack).find(); + } catch (Exception invalidRegex) { + return haystack.contains(cond); + } + } + + private String buildHaystack(Output out) { + if (out.getVars() == null || out.getVars().isEmpty()) { + return ""; + } + return out.getVars().toString(); + } + + private Integer safeExitCode(ScriptOutput taskOutput) { + try { + return taskOutput.getExitCode(); + } catch (Exception ignored) { + return null; + } + } + + private Map safeVars(ScriptOutput taskOutput) { + try { + return taskOutput.getVars(); + } catch (Exception ignored) { + return null; + } + } + + private record ExtractedFailure(Integer exitCode) { + } + + private ExtractedFailure extractFailure(RunnableTaskException e) { + Integer exitCode = null; + + Throwable cur = e.getCause(); + while (cur != null) { + if (cur instanceof TaskException te) { + exitCode = te.getExitCode(); + break; + } + cur = cur.getCause(); + } + + return new ExtractedFailure(exitCode); + } + + @Data + @AllArgsConstructor + public static class Output implements io.kestra.core.models.tasks.Output { + @Schema(title = "Timestamp of the event that fired the trigger") + private Instant timestamp; + + @Schema( + title = "Rendered condition", + description = "Rendered value of the exitCondition property for this poll." + ) + private String condition; + + @Schema( + title = "Script exit code", + description = "Exit code returned by the Perl process (may be null if not available)." + ) + private Integer exitCode; + + @Schema( + title = "Script vars", + description = """ + Vars produced by the task (e.g. via ::{"outputs":{...}}:: convention). This is the main structured + way to evaluate non-exit conditions on successful runs. + """ + ) + private Map vars; + } +} diff --git a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerConditionTest.java b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerConditionTest.java new file mode 100644 index 00000000..efde42a0 --- /dev/null +++ b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerConditionTest.java @@ -0,0 +1,65 @@ +package io.kestra.plugin.scripts.perl; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.time.Instant; +import java.util.Map; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; + +class CommandsTriggerConditionTest { + + private final CommandsTrigger trigger = CommandsTrigger.builder().build(); + + private CommandsTrigger.Output output(String condition, Integer exitCode, Map vars) { + return new CommandsTrigger.Output(Instant.now(), condition, exitCode, vars); + } + + @ParameterizedTest + @CsvSource({ + "exit 0, 0, true", + "exit 1, 1, true", + "EXIT 1, 1, true", + "exit 0, 1, false", + "exit 1, 0, false", + "exit 42, 42, true", + }) + void exitCodeCondition(String condition, int exitCode, boolean expected) { + assertThat(trigger.matchesCondition(output(condition, exitCode, null)), is(expected)); + } + + @Test + void exitCondition_nullExitCode_doesNotMatch() { + assertThat(trigger.matchesCondition(output("exit 1", null, null)), is(false)); + } + + @Test + void substringMatch_inVars() { + assertThat(trigger.matchesCondition( + output("toto", 0, Map.of("key", "toto"))), is(true)); + } + + @Test + void regexMatch_inVars() { + assertThat(trigger.matchesCondition( + output("status=\\w+", 0, Map.of("status", "status=ready"))), is(true)); + } + + @Test + void noMatch_emptyHaystack() { + assertThat(trigger.matchesCondition(output("something", 0, null)), is(false)); + } + + @Test + void noMatch_emptyCondition() { + assertThat(trigger.matchesCondition(output("", 0, Map.of("k", "v"))), is(false)); + } + + @Test + void nullCondition_doesNotMatch() { + assertThat(trigger.matchesCondition(output(null, 0, Map.of("k", "v"))), is(false)); + } +} diff --git a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerTest.java b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerTest.java new file mode 100644 index 00000000..79ea87c1 --- /dev/null +++ b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerTest.java @@ -0,0 +1,136 @@ +package io.kestra.plugin.scripts.perl; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.jupiter.api.Test; + +import io.kestra.core.junit.annotations.KestraTest; +import io.kestra.core.models.executions.Execution; +import io.kestra.core.models.property.Property; +import io.kestra.core.runners.RunContextFactory; +import io.kestra.core.utils.TestsUtils; + +import jakarta.inject.Inject; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.notNullValue; + +@KestraTest +class CommandsTriggerTest { + @Inject + private RunContextFactory runContextFactory; + + @Test + void commandsTrigger_shouldTriggerOnImplicitFailureExit1() throws Exception { + CommandsTrigger trigger = CommandsTrigger.builder() + .id("commands-trigger") + .type(CommandsTrigger.class.getName()) + .exitCondition(Property.ofValue("exit 1")) + .edge(Property.ofValue(true)) + .containerImage(Property.ofValue("perl:latest")) + .commands(Property.ofValue(List.of("perl -e 'exit 1;'"))) + .build(); + + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + Optional execution = trigger.evaluate(context.getKey(), context.getValue()); + + assertThat(execution.isPresent(), is(true)); + + Map triggerVars = execution.get().getTrigger().getVariables(); + assertThat("condition should be present", triggerVars.get("condition"), is("exit 1")); + assertThat("exitCode should be present", triggerVars.get("exitCode"), notNullValue()); + assertThat("exitCode should be 1", triggerVars.get("exitCode"), is(1)); + assertThat("timestamp should be present", triggerVars.get("timestamp"), notNullValue()); + } + + @Test + void commandsTrigger_shouldTriggerOnStdoutMatchUsingStructuredOutputs() throws Exception { + CommandsTrigger trigger = CommandsTrigger.builder() + .id("commands-stdout-match-trigger") + .type(CommandsTrigger.class.getName()) + .exitCondition(Property.ofValue("toto")) + .edge(Property.ofValue(true)) + .containerImage(Property.ofValue("perl:latest")) + .commands(Property.ofValue(List.of("echo '::{\"outputs\":{\"listing\":\"toto\"}}::'"))) + .build(); + + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + Optional execution = trigger.evaluate(context.getKey(), context.getValue()); + + assertThat(execution.isPresent(), is(true)); + + Map triggerVars = execution.get().getTrigger().getVariables(); + assertThat("condition should be present", triggerVars.get("condition"), is("toto")); + assertThat("exitCode should be present", triggerVars.get("exitCode"), notNullValue()); + assertThat("exitCode should be 0", triggerVars.get("exitCode"), is(0)); + assertThat("timestamp should be present", triggerVars.get("timestamp"), notNullValue()); + assertThat("vars should be present", triggerVars.get("vars"), notNullValue()); + } + + @Test + void commandsTrigger_shouldNotEmitWhenConditionDoesNotMatch() throws Exception { + CommandsTrigger trigger = CommandsTrigger.builder() + .id("commands-no-match-trigger") + .type(CommandsTrigger.class.getName()) + .exitCondition(Property.ofValue("exit 1")) + .edge(Property.ofValue(true)) + .containerImage(Property.ofValue("perl:latest")) + .commands(Property.ofValue(List.of("perl -e 'exit 0;'"))) + .build(); + + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + Optional execution = trigger.evaluate(context.getKey(), context.getValue()); + + assertThat("successful run should not match 'exit 1'", execution.isPresent(), is(false)); + } + + @Test + void commandsTrigger_shouldMatchRegexAgainstStructuredOutputs() throws Exception { + CommandsTrigger trigger = CommandsTrigger.builder() + .id("commands-regex-trigger") + .type(CommandsTrigger.class.getName()) + .exitCondition(Property.ofValue("status=\\w+")) + .edge(Property.ofValue(true)) + .containerImage(Property.ofValue("perl:latest")) + .commands(Property.ofValue(List.of("echo '::{\"outputs\":{\"status\":\"status=ready\"}}::'"))) + .build(); + + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + Optional execution = trigger.evaluate(context.getKey(), context.getValue()); + + assertThat("Regex condition should match", execution.isPresent(), is(true)); + + Map triggerVars = execution.get().getTrigger().getVariables(); + assertThat("exitCode should be 0", triggerVars.get("exitCode"), is(0)); + assertThat("vars should be present", triggerVars.get("vars"), notNullValue()); + } + + @Test + void edgeMode_preventsConsecutiveEmit() { + AtomicBoolean lastMatched = new AtomicBoolean(false); + + // First match: transition false->true => should emit + boolean matched1 = true; + boolean emit1 = !lastMatched.getAndSet(matched1) && matched1; + assertThat("first match should emit", emit1, is(true)); + + // Second consecutive match: true->true => should NOT emit + boolean matched2 = true; + boolean emit2 = !lastMatched.getAndSet(matched2) && matched2; + assertThat("consecutive match should NOT emit in edge mode", emit2, is(false)); + + // Non-match: true->false => should not emit + boolean matched3 = false; + boolean emit3 = !lastMatched.getAndSet(matched3) && matched3; + assertThat("non-match should not emit", emit3, is(false)); + + // Match again after non-match: false->true => should emit + boolean matched4 = true; + boolean emit4 = !lastMatched.getAndSet(matched4) && matched4; + assertThat("match after non-match should emit", emit4, is(true)); + } +} diff --git a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerConditionTest.java b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerConditionTest.java new file mode 100644 index 00000000..ecbc5aa8 --- /dev/null +++ b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerConditionTest.java @@ -0,0 +1,65 @@ +package io.kestra.plugin.scripts.perl; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.time.Instant; +import java.util.Map; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; + +class ScriptTriggerConditionTest { + + private final ScriptTrigger trigger = ScriptTrigger.builder().build(); + + private ScriptTrigger.Output output(String condition, Integer exitCode, Map vars) { + return new ScriptTrigger.Output(Instant.now(), condition, exitCode, vars); + } + + @ParameterizedTest + @CsvSource({ + "exit 0, 0, true", + "exit 1, 1, true", + "EXIT 1, 1, true", + "exit 0, 1, false", + "exit 1, 0, false", + "exit 42, 42, true", + }) + void exitCodeCondition(String condition, int exitCode, boolean expected) { + assertThat(trigger.matchesCondition(output(condition, exitCode, null)), is(expected)); + } + + @Test + void exitCondition_nullExitCode_doesNotMatch() { + assertThat(trigger.matchesCondition(output("exit 1", null, null)), is(false)); + } + + @Test + void substringMatch_inVars() { + assertThat(trigger.matchesCondition( + output("toto", 0, Map.of("key", "toto"))), is(true)); + } + + @Test + void regexMatch_inVars() { + assertThat(trigger.matchesCondition( + output("status=\\w+", 0, Map.of("status", "status=ready"))), is(true)); + } + + @Test + void noMatch_emptyHaystack() { + assertThat(trigger.matchesCondition(output("something", 0, null)), is(false)); + } + + @Test + void noMatch_emptyCondition() { + assertThat(trigger.matchesCondition(output("", 0, Map.of("k", "v"))), is(false)); + } + + @Test + void nullCondition_doesNotMatch() { + assertThat(trigger.matchesCondition(output(null, 0, Map.of("k", "v"))), is(false)); + } +} diff --git a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerTest.java b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerTest.java new file mode 100644 index 00000000..4ae863ca --- /dev/null +++ b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerTest.java @@ -0,0 +1,95 @@ +package io.kestra.plugin.scripts.perl; + +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; + +/** + * Unit tests for ScriptTrigger's condition-matching logic and edge mode. + * + * These tests exercise matchesCondition via the Output model without requiring a Perl + * runtime, which may not be available on all CI machines. + * Integration coverage against an actual Perl runtime lives in CommandsTriggerTest. + */ +class ScriptTriggerTest { + + private final ScriptTrigger trigger = ScriptTrigger.builder().build(); + + private ScriptTrigger.Output output(String condition, Integer exitCode, Map vars) { + return new ScriptTrigger.Output(Instant.now(), condition, exitCode, vars); + } + + @Test + void exitCodeCondition_shouldMatchWhenExitCodeEquals() { + assertThat(trigger.matchesCondition(output("exit 1", 1, null)), is(true)); + } + + @Test + void exitCodeCondition_shouldNotMatchWhenExitCodeDiffers() { + assertThat(trigger.matchesCondition(output("exit 1", 127, null)), is(false)); + } + + @Test + void exitCodeCondition_shouldNotMatchWhenExitCodeIsNull() { + assertThat(trigger.matchesCondition(output("exit 1", null, null)), is(false)); + } + + @Test + void substringCondition_shouldMatchAgainstVars() { + assertThat(trigger.matchesCondition(output("toto", 0, Map.of("listing", "toto"))), is(true)); + } + + @Test + void substringCondition_shouldNotMatchWhenAbsent() { + assertThat(trigger.matchesCondition(output("toto", 0, Map.of("listing", "something_else"))), is(false)); + } + + @Test + void regexCondition_shouldMatchAgainstVars() { + assertThat(trigger.matchesCondition(output("status=\\w+", 0, Map.of("status", "status=ready"))), is(true)); + } + + @Test + void emptyCondition_shouldNotMatch() { + assertThat(trigger.matchesCondition(output("", 0, null)), is(false)); + } + + @Test + void nullCondition_shouldNotMatch() { + assertThat(trigger.matchesCondition(output(null, 0, null)), is(false)); + } + + @Test + void exitZeroCondition_shouldMatchSuccessfulExecution() { + assertThat(trigger.matchesCondition(output("exit 0", 0, null)), is(true)); + } + + @Test + void edgeMode_shouldEmitOnFirstMatch() { + var lastMatched = new AtomicBoolean(false); + boolean matched = true; + boolean emit = !lastMatched.getAndSet(matched) && matched; + assertThat("first match should emit", emit, is(true)); + } + + @Test + void edgeMode_shouldSuppressConsecutiveMatches() { + var lastMatched = new AtomicBoolean(true); + boolean matched = true; + boolean emit = !lastMatched.getAndSet(matched) && matched; + assertThat("consecutive match should not emit in edge mode", emit, is(false)); + } + + @Test + void edgeMode_shouldEmitAgainAfterNonMatch() { + var lastMatched = new AtomicBoolean(false); + boolean matched = true; + boolean emit = !lastMatched.getAndSet(matched) && matched; + assertThat("match after non-match should emit", emit, is(true)); + } +} From 054c674bbaf8789d152df888667948089905133b Mon Sep 17 00:00:00 2001 From: jymaire Date: Tue, 22 Sep 2026 16:27:09 +0200 Subject: [PATCH 3/6] chore(r): drop R triggers, keep them in #438 The R ScriptTrigger/CommandsTrigger in this branch are byte-for-byte identical to PR #438, which is dedicated to the R module. Remove them here so #446 is scoped to the Perl triggers and #438 owns the R work. Co-Authored-By: Claude Opus 4.8 --- AGENTS.md | 2 - plugin-script-r/build.gradle | 3 - .../plugin/scripts/r/CommandsTrigger.java | 294 ------------------ .../plugin/scripts/r/ScriptTrigger.java | 290 ----------------- .../r/CommandsTriggerConditionTest.java | 65 ---- .../plugin/scripts/r/CommandsTriggerTest.java | 115 ------- .../scripts/r/ScriptTriggerConditionTest.java | 59 ---- .../plugin/scripts/r/ScriptTriggerTest.java | 95 ------ 8 files changed, 923 deletions(-) delete mode 100644 plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/CommandsTrigger.java delete mode 100644 plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/ScriptTrigger.java delete mode 100644 plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerConditionTest.java delete mode 100644 plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerTest.java delete mode 100644 plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerConditionTest.java delete mode 100644 plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerTest.java diff --git a/AGENTS.md b/AGENTS.md index e8e370a6..98b1e0df 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -110,9 +110,7 @@ This is a **multi-module** plugin with 19 submodules: **plugin-script-r:** - `io.kestra.plugin.scripts.r.Commands` -- `io.kestra.plugin.scripts.r.CommandsTrigger` - `io.kestra.plugin.scripts.r.Script` -- `io.kestra.plugin.scripts.r.ScriptTrigger` **plugin-script-ruby:** - `io.kestra.plugin.scripts.ruby.Commands` diff --git a/plugin-script-r/build.gradle b/plugin-script-r/build.gradle index acccc504..cefb63f3 100644 --- a/plugin-script-r/build.gradle +++ b/plugin-script-r/build.gradle @@ -16,7 +16,4 @@ dependencies { implementation project(':plugin-script') testImplementation project(path: ':plugin-script', configuration: 'testOutput') - - testImplementation group: "io.kestra", name: "scheduler", version: kestraVersion - testImplementation group: "io.kestra", name: "worker", version: kestraVersion } diff --git a/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/CommandsTrigger.java b/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/CommandsTrigger.java deleted file mode 100644 index 432d5f21..00000000 --- a/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/CommandsTrigger.java +++ /dev/null @@ -1,294 +0,0 @@ -package io.kestra.plugin.scripts.r; - -import io.kestra.core.models.annotations.Example; -import io.kestra.core.models.annotations.Plugin; -import io.kestra.core.models.annotations.PluginProperty; -import io.kestra.core.models.conditions.ConditionContext; -import io.kestra.core.models.executions.Execution; -import io.kestra.core.models.property.Property; -import io.kestra.core.models.tasks.RunnableTaskException; -import io.kestra.core.models.tasks.runners.TaskException; -import io.kestra.core.models.triggers.*; -import io.kestra.core.runners.RunContext; -import io.kestra.plugin.scripts.exec.TriggerRunContext; -import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput; -import io.swagger.v3.oas.annotations.media.Schema; -import jakarta.validation.constraints.NotNull; -import lombok.*; -import lombok.experimental.SuperBuilder; - -import java.time.Duration; -import java.time.Instant; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.regex.Matcher; -import java.util.regex.Pattern; - -@SuperBuilder -@ToString -@EqualsAndHashCode -@Getter -@NoArgsConstructor -@Schema( - title = "Trigger a flow when R commands match a condition", - description = "Polls by running R commands in a container (default image 'r-base') and starts the flow when their result matches the condition." -) -@Plugin( - examples = { - @Example( - title = "Trigger when an R command fails.", - full = true, - code = """ - id: r_commands_trigger - namespace: company.team - - triggers: - - id: on_fail - type: io.kestra.plugin.scripts.r.CommandsTrigger - interval: PT5S - exitCondition: "exit 1" - commands: - - Rscript -e 'stop("boom")' - - tasks: - - id: log - type: io.kestra.plugin.core.log.Log - message: "Triggered with exitCode={{ trigger.exitCode }} (condition={{ trigger.condition }})" - """ - ) - } -) -// TODO: extract shared trigger logic (evaluate, matchesCondition, extractFailure, Output) -// into an AbstractScriptTrigger in plugin-script to reduce duplication across Shell, Node, Ruby, R, etc. -public class CommandsTrigger extends AbstractTrigger - implements PollingTriggerInterface, TriggerOutput { - - private static final String DEFAULT_IMAGE = "r-base"; - - private static final Pattern EXIT_CONDITION_PATTERN = - Pattern.compile("^\\s*exit\\s+(\\d+)\\s*$", Pattern.CASE_INSENSITIVE); - - @Schema( - title = "Docker image used to execute the commands", - description = """ - Container image used by the underlying Commands task to run R commands. - Defaults to 'r-base'. - """ - ) - @Builder.Default - @PluginProperty(group = "execution") - protected Property containerImage = Property.ofValue(DEFAULT_IMAGE); - - @Schema( - title = "R commands to execute", - description = "Commands executed in order on each poll." - ) - @NotNull - @PluginProperty(group = "main") - protected Property> commands; - - @Schema( - title = "Condition to match", - description = """ - Condition evaluated after execution. - - Supported forms: - - 'exit N' - - regex / substring matched against vars + logs - """ - ) - @NotNull - @PluginProperty(group = "main") - protected Property exitCondition; - - @Schema( - title = "Check interval", - description = "Interval between polling evaluations." - ) - @Builder.Default - @PluginProperty(group = "execution") - private final Duration interval = Duration.ofSeconds(60); - - @Schema( - title = "Edge trigger mode", - description = """ - If true, the trigger emits only on a transition from 'not matching' to 'matching' (anti-spam). - If false, the trigger emits on every poll where the condition matches. - """ - ) - @Builder.Default - @PluginProperty(group = "advanced") - protected Property edge = Property.ofValue(true); - - // Known limitation: in-memory only — resets when the trigger is rehydrated (e.g. after restart), - // so edge mode may re-fire once after a scheduler restart. - @Builder.Default - @Getter(AccessLevel.NONE) - private final AtomicBoolean lastMatched = new AtomicBoolean(false); - - @Override - public Optional evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception { - RunContext runContext = conditionContext.getRunContext(); - boolean edgeEnabled = runContext.render(this.edge).as(Boolean.class).orElse(true); - - Output out; - try { - out = runOnce(runContext); - } catch (Exception e) { - runContext.logger().warn("Trigger evaluation failed, returning empty result to avoid blocking the scheduler", e); - return Optional.empty(); - } - - boolean matched = matchesCondition(out); - - boolean emit = edgeEnabled - ? (!lastMatched.getAndSet(matched) && matched) - : matched; - - if (!emit) { - return Optional.empty(); - } - - return Optional.of( - TriggerService.generateExecution(this, conditionContext, context, out) - ); - } - - private Output runOnce(RunContext runContext) throws Exception { - Commands task = Commands.builder() - .id(this.getId()) - .type(Commands.class.getName()) - .containerImage(this.containerImage) - .commands(this.commands) - .build(); - - String renderedCondition = runContext.render(this.exitCondition) - .as(String.class) - .orElse(""); - - try { - ScriptOutput taskOutput = task.run(TriggerRunContext.forEmbeddedTask(runContext, task)); - - return new Output( - Instant.now(), - renderedCondition, - safeExitCode(taskOutput), - safeVars(taskOutput) - ); - } catch (RunnableTaskException e) { - ExtractedFailure failure = extractFailure(e); - return new Output( - Instant.now(), - renderedCondition, - failure.exitCode, - null - ); - } - } - - boolean matchesCondition(Output out) { - String cond = out.getCondition() == null ? "" : out.getCondition().trim(); - - Matcher exitMatcher = EXIT_CONDITION_PATTERN.matcher(cond); - - if (exitMatcher.matches()) { - int expected = Integer.parseInt(exitMatcher.group(1)); - return out.getExitCode() != null && out.getExitCode() == expected; - } - - String haystack = buildHaystack(out); - if (haystack.isEmpty() || cond.isEmpty()) { - return false; - } - - try { - // Guard against catastrophic backtracking (ReDoS) from user-supplied patterns - var pattern = Pattern.compile(cond); - var future = CompletableFuture.supplyAsync( - () -> pattern.matcher(haystack).find() - ); - return future.get(5, TimeUnit.SECONDS); - } catch (TimeoutException te) { - return haystack.contains(cond); - } catch (Exception e) { - return haystack.contains(cond); - } - } - - private String buildHaystack(Output out) { - if (out.getVars() == null || out.getVars().isEmpty()) { - return ""; - } - // Map.toString() produces {key=value, ...} — intentional for substring/regex matching. - return out.getVars().toString(); - } - - private Integer safeExitCode(ScriptOutput taskOutput) { - try { - return taskOutput.getExitCode(); - } catch (Exception ignored) { - return null; - } - } - - private Map safeVars(ScriptOutput taskOutput) { - try { - return taskOutput.getVars(); - } catch (Exception ignored) { - return null; - } - } - - private record ExtractedFailure(Integer exitCode) {} - - private ExtractedFailure extractFailure(RunnableTaskException e) { - Integer exitCode = null; - - Throwable cur = e.getCause(); - while (cur != null) { - if (cur instanceof TaskException te) { - exitCode = te.getExitCode(); - break; - } - cur = cur.getCause(); - } - - return new ExtractedFailure(exitCode); - } - - @Data - @AllArgsConstructor - public static class Output implements io.kestra.core.models.tasks.Output { - @Schema( - title = "Poll timestamp", - description = "Timestamp when this trigger evaluation occurred." - ) - private Instant timestamp; - - @Schema( - title = "Rendered condition", - description = "Rendered value of the exitCondition property for this poll." - ) - private String condition; - - @Schema( - title = "Commands exit code", - description = "Exit code returned by the R process (may be null if not available)." - ) - private Integer exitCode; - - @Schema( - title = "Commands vars", - description = """ - Vars produced by the task (e.g. via ::{"outputs":{...}}:: convention). This is the main structured - way to evaluate non-exit conditions on successful runs. - """ - ) - private Map vars; - } -} diff --git a/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/ScriptTrigger.java b/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/ScriptTrigger.java deleted file mode 100644 index 34c99e45..00000000 --- a/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/ScriptTrigger.java +++ /dev/null @@ -1,290 +0,0 @@ -package io.kestra.plugin.scripts.r; - -import io.kestra.core.models.annotations.Example; -import io.kestra.core.models.annotations.Plugin; -import io.kestra.core.models.annotations.PluginProperty; -import io.kestra.core.models.conditions.ConditionContext; -import io.kestra.core.models.enums.MonacoLanguages; -import io.kestra.core.models.executions.Execution; -import io.kestra.core.models.property.Property; -import io.kestra.core.models.tasks.RunnableTaskException; -import io.kestra.core.models.tasks.runners.TaskException; -import io.kestra.core.models.triggers.*; -import io.kestra.core.runners.RunContext; -import io.kestra.plugin.scripts.exec.TriggerRunContext; -import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput; -import io.swagger.v3.oas.annotations.media.Schema; -import jakarta.validation.constraints.NotNull; -import lombok.*; -import lombok.experimental.SuperBuilder; - -import java.time.Duration; -import java.time.Instant; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.regex.Matcher; -import java.util.regex.Pattern; - -@SuperBuilder -@ToString -@EqualsAndHashCode -@Getter -@NoArgsConstructor -@Schema( - title = "Trigger a flow when an R script matches a condition", - description = "Polls by running an inline R script in a container (default image 'r-base') and starts the flow when its result matches the condition." -) -@Plugin( - examples = { - @Example( - title = "Trigger when the script fails with exit code 1.", - full = true, - code = """ - id: r_script_trigger - namespace: company.team - - triggers: - - id: script_failure - type: io.kestra.plugin.scripts.r.ScriptTrigger - interval: PT10S - exitCondition: "exit 1" - edge: true - script: | - stop("boom") - - tasks: - - id: log - type: io.kestra.plugin.core.log.Log - message: "Triggered with exitCode={{ trigger.exitCode }} (condition={{ trigger.condition }})" - """ - ) - } -) -// TODO: extract shared trigger logic (evaluate, matchesCondition, extractFailure, Output) -// into an AbstractScriptTrigger in plugin-script to reduce duplication across Shell, Node, Ruby, R, etc. -public class ScriptTrigger extends AbstractTrigger - implements PollingTriggerInterface, TriggerOutput { - - private static final String DEFAULT_IMAGE = "r-base"; - - private static final Pattern EXIT_CONDITION_PATTERN = - Pattern.compile("^\\s*exit\\s+(\\d+)\\s*$", Pattern.CASE_INSENSITIVE); - - @Schema( - title = "Container image for script execution", - description = "Image used by the Script task to run the inline R script; defaults to 'r-base'. Provide an image that includes the required CRAN packages, or install them in the script itself." - ) - @Builder.Default - @PluginProperty(group = "execution") - protected Property containerImage = Property.ofValue(DEFAULT_IMAGE); - - @Schema( - title = "Inline R script", - description = "Multi-line R script executed on each poll, with the same semantics as the R Script task." - ) - @NotNull - @PluginProperty(language = MonacoLanguages.R, group = "main") - protected Property script; - - @Schema( - title = "Condition to match", - description = """ - Condition evaluated after each execution. The trigger emits only when it matches. - 'exit N' compares the exit code, otherwise the string is used as a regex - (or substring fallback) against emitted vars and failure logs. - """ - ) - @NotNull - @PluginProperty(group = "main") - protected Property exitCondition; - - @Schema( - title = "Check interval", - description = "Interval between polling evaluations." - ) - @Builder.Default - @PluginProperty(group = "execution") - private final Duration interval = Duration.ofSeconds(60); - - @Schema( - title = "Edge trigger mode", - description = """ - If true, the trigger emits only on a transition from 'not matching' to 'matching' (anti-spam). - If false, the trigger emits on every poll where the condition matches. - """ - ) - @Builder.Default - @PluginProperty(group = "advanced") - protected Property edge = Property.ofValue(true); - - // Known limitation: in-memory only — resets when the trigger is rehydrated (e.g. after restart), - // so edge mode may re-fire once after a scheduler restart. - @Getter(AccessLevel.NONE) - @Builder.Default - private final AtomicBoolean lastMatched = new AtomicBoolean(false); - - @Override - public Optional evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception { - RunContext runContext = conditionContext.getRunContext(); - boolean edgeEnabled = runContext.render(this.edge).as(Boolean.class).orElse(true); - - Output output; - try { - output = runOnce(runContext); - } catch (Exception e) { - runContext.logger().warn("Trigger evaluation failed, returning empty result to avoid blocking the scheduler", e); - return Optional.empty(); - } - - boolean matched = matchesCondition(output); - - boolean emit = edgeEnabled - ? (!lastMatched.getAndSet(matched) && matched) - : matched; - - if (!emit) { - return Optional.empty(); - } - - return Optional.of( - TriggerService.generateExecution(this, conditionContext, context, output) - ); - } - - private Output runOnce(RunContext runContext) throws Exception { - Script task = Script.builder() - .id(this.getId()) - .type(Script.class.getName()) - .containerImage(this.containerImage) - .script(this.script) - .build(); - - String renderedCondition = runContext.render(this.exitCondition) - .as(String.class) - .orElse(""); - - try { - ScriptOutput taskOutput = task.run(TriggerRunContext.forEmbeddedTask(runContext, task)); - - return new Output( - Instant.now(), - renderedCondition, - safeExitCode(taskOutput), - safeVars(taskOutput) - ); - } catch (RunnableTaskException e) { - ExtractedFailure failure = extractFailure(e); - return new Output( - Instant.now(), - renderedCondition, - failure.exitCode, - null - ); - } - } - - boolean matchesCondition(Output out) { - String cond = out.getCondition() == null ? "" : out.getCondition().trim(); - - Matcher exitMatcher = EXIT_CONDITION_PATTERN.matcher(cond); - - if (exitMatcher.matches()) { - int expected = Integer.parseInt(exitMatcher.group(1)); - return out.getExitCode() != null && out.getExitCode() == expected; - } - - String haystack = buildHaystack(out); - if (haystack.isEmpty() || cond.isEmpty()) { - return false; - } - - try { - // Guard against catastrophic backtracking (ReDoS) from user-supplied patterns - var pattern = Pattern.compile(cond); - var future = CompletableFuture.supplyAsync( - () -> pattern.matcher(haystack).find() - ); - return future.get(5, TimeUnit.SECONDS); - } catch (TimeoutException te) { - return haystack.contains(cond); - } catch (Exception e) { - return haystack.contains(cond); - } - } - - private String buildHaystack(Output out) { - if (out.getVars() == null || out.getVars().isEmpty()) { - return ""; - } - // Map.toString() produces {key=value, ...} — intentional for substring/regex matching. - return out.getVars().toString(); - } - - private Integer safeExitCode(ScriptOutput output) { - try { - return output.getExitCode(); - } catch (Exception ignored) { - return null; - } - } - - private Map safeVars(ScriptOutput output) { - try { - return output.getVars(); - } catch (Exception ignored) { - return null; - } - } - - private record ExtractedFailure(Integer exitCode) {} - - private ExtractedFailure extractFailure(RunnableTaskException e) { - Integer exitCode = null; - - Throwable cur = e.getCause(); - while (cur != null) { - if (cur instanceof TaskException te) { - exitCode = te.getExitCode(); - break; - } - cur = cur.getCause(); - } - - return new ExtractedFailure(exitCode); - } - - @Data - @AllArgsConstructor - public static class Output implements io.kestra.core.models.tasks.Output { - @Schema( - title = "Poll timestamp", - description = "Timestamp when this trigger evaluation occurred." - ) - private Instant timestamp; - - @Schema( - title = "Rendered condition", - description = "Rendered value of the exitCondition property for this poll." - ) - private String condition; - - @Schema( - title = "Script exit code", - description = "Exit code returned by the R process (may be null if not available)." - ) - private Integer exitCode; - - @Schema( - title = "Script vars", - description = """ - Vars produced by the task (e.g. via ::{"outputs":{...}}:: convention). This is the main structured - way to evaluate non-exit conditions on successful runs. - """ - ) - private Map vars; - } -} diff --git a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerConditionTest.java b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerConditionTest.java deleted file mode 100644 index 2a120568..00000000 --- a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerConditionTest.java +++ /dev/null @@ -1,65 +0,0 @@ -package io.kestra.plugin.scripts.r; - -import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.CsvSource; - -import java.time.Instant; -import java.util.Map; - -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.is; - -class CommandsTriggerConditionTest { - - private final CommandsTrigger trigger = CommandsTrigger.builder().build(); - - private CommandsTrigger.Output output(String condition, Integer exitCode, Map vars) { - return new CommandsTrigger.Output(Instant.now(), condition, exitCode, vars); - } - - @ParameterizedTest - @CsvSource({ - "exit 0, 0, true", - "exit 1, 1, true", - "EXIT 1, 1, true", - "exit 0, 1, false", - "exit 1, 0, false", - "exit 42, 42, true", - }) - void exitCodeCondition(String condition, int exitCode, boolean expected) { - assertThat(trigger.matchesCondition(output(condition, exitCode, null)), is(expected)); - } - - @Test - void exitCondition_nullExitCode_doesNotMatch() { - assertThat(trigger.matchesCondition(output("exit 1", null, null)), is(false)); - } - - @Test - void substringMatch_inVars() { - assertThat(trigger.matchesCondition( - output("toto", 0, Map.of("key", "toto"))), is(true)); - } - - @Test - void regexMatch_inVars() { - assertThat(trigger.matchesCondition( - output("status=\\w+", 0, Map.of("status", "status=ready"))), is(true)); - } - - @Test - void noMatch_emptyHaystack() { - assertThat(trigger.matchesCondition(output("something", 0, null)), is(false)); - } - - @Test - void noMatch_emptyCondition() { - assertThat(trigger.matchesCondition(output("", 0, Map.of("k", "v"))), is(false)); - } - - @Test - void nullCondition_doesNotMatch() { - assertThat(trigger.matchesCondition(output(null, 0, Map.of("k", "v"))), is(false)); - } -} diff --git a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerTest.java b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerTest.java deleted file mode 100644 index a99cbca4..00000000 --- a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerTest.java +++ /dev/null @@ -1,115 +0,0 @@ -package io.kestra.plugin.scripts.r; - -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.atomic.AtomicBoolean; - -import org.junit.jupiter.api.Test; - -import io.kestra.core.junit.annotations.KestraTest; -import io.kestra.core.models.executions.Execution; -import io.kestra.core.models.property.Property; -import io.kestra.core.runners.RunContextFactory; -import io.kestra.core.utils.TestsUtils; - -import jakarta.inject.Inject; - -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.is; -import static org.hamcrest.Matchers.notNullValue; - -@KestraTest -class CommandsTriggerTest { - @Inject - private RunContextFactory runContextFactory; - - @Test - void commandsTrigger_shouldTriggerOnImplicitFailureExit1() throws Exception { - CommandsTrigger trigger = CommandsTrigger.builder() - .id("commands-trigger") - .type(CommandsTrigger.class.getName()) - .exitCondition(Property.ofValue("exit 1")) - .edge(Property.ofValue(true)) - .containerImage(Property.ofValue("r-base")) - .commands(Property.ofValue(List.of("Rscript -e 'quit(status = 1)'"))) - .build(); - - var context = TestsUtils.mockTrigger(runContextFactory, trigger); - Optional execution = trigger.evaluate(context.getKey(), context.getValue()); - - assertThat(execution.isPresent(), is(true)); - - Map triggerVars = execution.get().getTrigger().getVariables(); - assertThat("condition should be present", triggerVars.get("condition"), is("exit 1")); - assertThat("exitCode should be present", triggerVars.get("exitCode"), notNullValue()); - assertThat("exitCode should be 1", triggerVars.get("exitCode"), is(1)); - assertThat("timestamp should be present", triggerVars.get("timestamp"), notNullValue()); - } - - @Test - void commandsTrigger_shouldTriggerOnStdoutMatchUsingStructuredOutputs() throws Exception { - CommandsTrigger trigger = CommandsTrigger.builder() - .id("commands-stdout-match-trigger") - .type(CommandsTrigger.class.getName()) - .exitCondition(Property.ofValue("toto")) - .edge(Property.ofValue(true)) - .containerImage(Property.ofValue("r-base")) - .commands(Property.ofValue(List.of("echo '::{\"outputs\":{\"listing\":\"toto\"}}::'"))) - .build(); - - var context = TestsUtils.mockTrigger(runContextFactory, trigger); - Optional execution = trigger.evaluate(context.getKey(), context.getValue()); - - assertThat(execution.isPresent(), is(true)); - - Map triggerVars = execution.get().getTrigger().getVariables(); - assertThat("condition should be present", triggerVars.get("condition"), is("toto")); - assertThat("exitCode should be present", triggerVars.get("exitCode"), notNullValue()); - assertThat("exitCode should be 0", triggerVars.get("exitCode"), is(0)); - assertThat("timestamp should be present", triggerVars.get("timestamp"), notNullValue()); - assertThat("vars should be present", triggerVars.get("vars"), notNullValue()); - } - - @Test - void commandsTrigger_shouldNotEmitWhenConditionDoesNotMatch() throws Exception { - CommandsTrigger trigger = CommandsTrigger.builder() - .id("commands-no-match-trigger") - .type(CommandsTrigger.class.getName()) - .exitCondition(Property.ofValue("exit 1")) - .edge(Property.ofValue(true)) - .containerImage(Property.ofValue("r-base")) - .commands(Property.ofValue(List.of("Rscript -e 'quit(status = 0)'"))) - .build(); - - var context = TestsUtils.mockTrigger(runContextFactory, trigger); - Optional execution = trigger.evaluate(context.getKey(), context.getValue()); - - assertThat("successful run should not match 'exit 1'", execution.isPresent(), is(false)); - } - - @Test - void edgeMode_preventsConsecutiveEmit() { - AtomicBoolean lastMatched = new AtomicBoolean(false); - - // First match: transition false->true => should emit - boolean matched1 = true; - boolean emit1 = !lastMatched.getAndSet(matched1) && matched1; - assertThat("first match should emit", emit1, is(true)); - - // Second consecutive match: true->true => should NOT emit - boolean matched2 = true; - boolean emit2 = !lastMatched.getAndSet(matched2) && matched2; - assertThat("consecutive match should NOT emit in edge mode", emit2, is(false)); - - // Non-match: true->false => should not emit - boolean matched3 = false; - boolean emit3 = !lastMatched.getAndSet(matched3) && matched3; - assertThat("non-match should not emit", emit3, is(false)); - - // Match again after non-match: false->true => should emit - boolean matched4 = true; - boolean emit4 = !lastMatched.getAndSet(matched4) && matched4; - assertThat("match after non-match should emit", emit4, is(true)); - } -} diff --git a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerConditionTest.java b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerConditionTest.java deleted file mode 100644 index 9779f859..00000000 --- a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerConditionTest.java +++ /dev/null @@ -1,59 +0,0 @@ -package io.kestra.plugin.scripts.r; - -import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.CsvSource; - -import java.time.Instant; -import java.util.Map; - -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.is; - -class ScriptTriggerConditionTest { - - private final ScriptTrigger trigger = ScriptTrigger.builder().build(); - - private ScriptTrigger.Output output(String condition, Integer exitCode, Map vars) { - return new ScriptTrigger.Output(Instant.now(), condition, exitCode, vars); - } - - @ParameterizedTest - @CsvSource({ - "exit 0, 0, true", - "exit 1, 1, true", - "EXIT 1, 1, true", - "exit 0, 1, false", - "exit 1, 0, false", - "exit 42, 42, true", - }) - void exitCodeCondition(String condition, int exitCode, boolean expected) { - assertThat(trigger.matchesCondition(output(condition, exitCode, null)), is(expected)); - } - - @Test - void exitCondition_nullExitCode_doesNotMatch() { - assertThat(trigger.matchesCondition(output("exit 1", null, null)), is(false)); - } - - @Test - void substringMatch_inVars() { - assertThat(trigger.matchesCondition( - output("toto", 0, Map.of("key", "toto"))), is(true)); - } - - @Test - void noMatch_emptyHaystack() { - assertThat(trigger.matchesCondition(output("something", 0, null)), is(false)); - } - - @Test - void noMatch_emptyCondition() { - assertThat(trigger.matchesCondition(output("", 0, Map.of("k", "v"))), is(false)); - } - - @Test - void nullCondition_doesNotMatch() { - assertThat(trigger.matchesCondition(output(null, 0, Map.of("k", "v"))), is(false)); - } -} diff --git a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerTest.java b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerTest.java deleted file mode 100644 index 745e8bff..00000000 --- a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerTest.java +++ /dev/null @@ -1,95 +0,0 @@ -package io.kestra.plugin.scripts.r; - -import org.junit.jupiter.api.Test; - -import java.time.Instant; -import java.util.Map; -import java.util.concurrent.atomic.AtomicBoolean; - -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.is; - -/** - * Unit tests for ScriptTrigger's condition-matching logic and edge mode. - * - * These tests exercise matchesCondition via the Output model without requiring an R - * runtime, which may not be available on all CI machines. - * Integration coverage against an actual R runtime lives in CommandsTriggerTest. - */ -class ScriptTriggerTest { - - private final ScriptTrigger trigger = ScriptTrigger.builder().build(); - - private ScriptTrigger.Output output(String condition, Integer exitCode, Map vars) { - return new ScriptTrigger.Output(Instant.now(), condition, exitCode, vars); - } - - @Test - void exitCodeCondition_shouldMatchWhenExitCodeEquals() { - assertThat(trigger.matchesCondition(output("exit 1", 1, null)), is(true)); - } - - @Test - void exitCodeCondition_shouldNotMatchWhenExitCodeDiffers() { - assertThat(trigger.matchesCondition(output("exit 1", 127, null)), is(false)); - } - - @Test - void exitCodeCondition_shouldNotMatchWhenExitCodeIsNull() { - assertThat(trigger.matchesCondition(output("exit 1", null, null)), is(false)); - } - - @Test - void substringCondition_shouldMatchAgainstVars() { - assertThat(trigger.matchesCondition(output("toto", 0, Map.of("listing", "toto"))), is(true)); - } - - @Test - void substringCondition_shouldNotMatchWhenAbsent() { - assertThat(trigger.matchesCondition(output("toto", 0, Map.of("listing", "something_else"))), is(false)); - } - - @Test - void regexCondition_shouldMatchAgainstVars() { - assertThat(trigger.matchesCondition(output("status=\\w+", 0, Map.of("status", "status=ready"))), is(true)); - } - - @Test - void emptyCondition_shouldNotMatch() { - assertThat(trigger.matchesCondition(output("", 0, null)), is(false)); - } - - @Test - void nullCondition_shouldNotMatch() { - assertThat(trigger.matchesCondition(output(null, 0, null)), is(false)); - } - - @Test - void exitZeroCondition_shouldMatchSuccessfulExecution() { - assertThat(trigger.matchesCondition(output("exit 0", 0, null)), is(true)); - } - - @Test - void edgeMode_shouldEmitOnFirstMatch() { - var lastMatched = new AtomicBoolean(false); - boolean matched = true; - boolean emit = !lastMatched.getAndSet(matched) && matched; - assertThat("first match should emit", emit, is(true)); - } - - @Test - void edgeMode_shouldSuppressConsecutiveMatches() { - var lastMatched = new AtomicBoolean(true); - boolean matched = true; - boolean emit = !lastMatched.getAndSet(matched) && matched; - assertThat("consecutive match should not emit in edge mode", emit, is(false)); - } - - @Test - void edgeMode_shouldEmitAgainAfterNonMatch() { - var lastMatched = new AtomicBoolean(false); - boolean matched = true; - boolean emit = !lastMatched.getAndSet(matched) && matched; - assertThat("match after non-match should emit", emit, is(true)); - } -} From 00e0c16ac42028c4c976575eed3401ecf5d6de91 Mon Sep 17 00:00:00 2001 From: jymaire Date: Wed, 23 Sep 2026 16:02:20 +0200 Subject: [PATCH 4/6] fix(perl): guard trigger exitCondition regex against ReDoS Run the user-supplied exitCondition regex with a 5s timeout and fall back to substring matching, as the Ruby/Shell/Node/Bun triggers do, so a catastrophic-backtracking pattern can no longer hang the scheduler poll thread. Document the in-memory edge-state limitation and add condition tests for pathological and invalid regexes. Co-Authored-By: Claude Opus 5.5 --- .../plugin/scripts/perl/CommandsTrigger.java | 19 +++++++++++++++++-- .../plugin/scripts/perl/ScriptTrigger.java | 19 +++++++++++++++++-- .../perl/CommandsTriggerConditionTest.java | 18 ++++++++++++++++++ .../perl/ScriptTriggerConditionTest.java | 18 ++++++++++++++++++ 4 files changed, 70 insertions(+), 4 deletions(-) diff --git a/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java index f4e2deb5..f0e06dc6 100644 --- a/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java +++ b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java @@ -6,6 +6,9 @@ import java.util.Map; import java.util.Optional; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -74,6 +77,8 @@ ) } ) +// TODO: extract shared trigger logic (evaluate, matchesCondition, extractFailure, Output) +// into an AbstractScriptTrigger in plugin-script to reduce duplication across Shell, Node, Ruby, etc. public class CommandsTrigger extends AbstractTrigger implements PollingTriggerInterface, TriggerOutput { @@ -134,6 +139,8 @@ public class CommandsTrigger extends AbstractTrigger @PluginProperty(group = "advanced") protected Property edge = Property.ofValue(true); + // Known limitation: in-memory only — resets when the trigger is rehydrated (e.g. after restart), + // so edge mode may re-fire once after a scheduler restart. @Builder.Default @Getter(AccessLevel.NONE) private final AtomicBoolean lastMatched = new AtomicBoolean(false); @@ -201,8 +208,15 @@ boolean matchesCondition(Output out) { } try { - return Pattern.compile(cond).matcher(haystack).find(); - } catch (Exception invalidRegex) { + // Guard against catastrophic backtracking (ReDoS) from user-supplied patterns + var pattern = Pattern.compile(cond); + var future = CompletableFuture.supplyAsync( + () -> pattern.matcher(haystack).find() + ); + return future.get(5, TimeUnit.SECONDS); + } catch (TimeoutException te) { + return haystack.contains(cond); + } catch (Exception e) { return haystack.contains(cond); } } @@ -211,6 +225,7 @@ private String buildHaystack(Output out) { if (out.getVars() == null || out.getVars().isEmpty()) { return ""; } + // Map.toString() produces {key=value, ...} — intentional for substring/regex matching. return out.getVars().toString(); } diff --git a/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/ScriptTrigger.java b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/ScriptTrigger.java index 9594d02f..db63bf48 100644 --- a/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/ScriptTrigger.java +++ b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/ScriptTrigger.java @@ -5,6 +5,9 @@ import java.util.Map; import java.util.Optional; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -74,6 +77,8 @@ ) } ) +// TODO: extract shared trigger logic (evaluate, matchesCondition, extractFailure, Output) +// into an AbstractScriptTrigger in plugin-script to reduce duplication across Shell, Node, Ruby, etc. public class ScriptTrigger extends AbstractTrigger implements PollingTriggerInterface, TriggerOutput { @@ -132,6 +137,8 @@ When true (default), emit only on a transition from not matching to matching. Wh @PluginProperty(group = "advanced") protected Property edge = Property.ofValue(true); + // Known limitation: in-memory only — resets when the trigger is rehydrated (e.g. after restart), + // so edge mode may re-fire once after a scheduler restart. @Builder.Default @Getter(AccessLevel.NONE) private final AtomicBoolean lastMatched = new AtomicBoolean(false); @@ -199,8 +206,15 @@ boolean matchesCondition(Output out) { } try { - return Pattern.compile(cond).matcher(haystack).find(); - } catch (Exception invalidRegex) { + // Guard against catastrophic backtracking (ReDoS) from user-supplied patterns + var pattern = Pattern.compile(cond); + var future = CompletableFuture.supplyAsync( + () -> pattern.matcher(haystack).find() + ); + return future.get(5, TimeUnit.SECONDS); + } catch (TimeoutException te) { + return haystack.contains(cond); + } catch (Exception e) { return haystack.contains(cond); } } @@ -209,6 +223,7 @@ private String buildHaystack(Output out) { if (out.getVars() == null || out.getVars().isEmpty()) { return ""; } + // Map.toString() produces {key=value, ...} — intentional for substring/regex matching. return out.getVars().toString(); } diff --git a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerConditionTest.java b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerConditionTest.java index efde42a0..faa02d9e 100644 --- a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerConditionTest.java +++ b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerConditionTest.java @@ -4,11 +4,13 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; +import java.time.Duration; import java.time.Instant; import java.util.Map; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.is; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; class CommandsTriggerConditionTest { @@ -62,4 +64,20 @@ void noMatch_emptyCondition() { void nullCondition_doesNotMatch() { assertThat(trigger.matchesCondition(output(null, 0, Map.of("k", "v"))), is(false)); } + + @Test + void catastrophicBacktrackingRegex_fallsBackToSubstring_withoutHanging() { + String condition = "(a+)+$"; + String value = "a".repeat(40) + "!"; + + assertTimeoutPreemptively(Duration.ofSeconds(15), () -> + assertThat(trigger.matchesCondition(output(condition, 0, Map.of("k", value))), is(false)) + ); + } + + @Test + void invalidRegex_fallsBackToSubstring() { + assertThat(trigger.matchesCondition( + output("[unclosed", 0, Map.of("k", "value with [unclosed inside"))), is(true)); + } } diff --git a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerConditionTest.java b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerConditionTest.java index ecbc5aa8..ec12c424 100644 --- a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerConditionTest.java +++ b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerConditionTest.java @@ -4,11 +4,13 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; +import java.time.Duration; import java.time.Instant; import java.util.Map; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.is; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; class ScriptTriggerConditionTest { @@ -62,4 +64,20 @@ void noMatch_emptyCondition() { void nullCondition_doesNotMatch() { assertThat(trigger.matchesCondition(output(null, 0, Map.of("k", "v"))), is(false)); } + + @Test + void catastrophicBacktrackingRegex_fallsBackToSubstring_withoutHanging() { + String condition = "(a+)+$"; + String value = "a".repeat(40) + "!"; + + assertTimeoutPreemptively(Duration.ofSeconds(15), () -> + assertThat(trigger.matchesCondition(output(condition, 0, Map.of("k", value))), is(false)) + ); + } + + @Test + void invalidRegex_fallsBackToSubstring() { + assertThat(trigger.matchesCondition( + output("[unclosed", 0, Map.of("k", "value with [unclosed inside"))), is(true)); + } } From f61311387deb29144b6f8c9bdc78189ce6185635 Mon Sep 17 00:00:00 2001 From: jymaire Date: Wed, 23 Sep 2026 16:21:28 +0200 Subject: [PATCH 5/6] fix(perl): make ReDoS test hit the timeout and fix CommandsTrigger example (a+)+$ is memoized by the JDK 25 regex engine and fails instantly, so the test never reached the 5s guard; use (.*a){20}$ and assert the elapsed time. The CommandsTrigger example ran `perl missing.pl`, which exits 2 and never matched `exit 1`; use `perl -e 'exit 1'` instead. Co-Authored-By: Claude Opus 5.5 --- .../io/kestra/plugin/scripts/perl/CommandsTrigger.java | 4 ++-- .../scripts/perl/CommandsTriggerConditionTest.java | 10 +++++++++- .../scripts/perl/ScriptTriggerConditionTest.java | 10 +++++++++- 3 files changed, 20 insertions(+), 4 deletions(-) diff --git a/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java index f0e06dc6..c43485b4 100644 --- a/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java +++ b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java @@ -53,7 +53,7 @@ @Plugin( examples = { @Example( - title = "Trigger when commands fail with an implicit error (exit 1).", + title = "Trigger when the command explicitly exits with code 1.", full = true, code = """ id: commands_trigger @@ -67,7 +67,7 @@ edge: true containerImage: perl commands: - - perl missing.pl + - perl -e 'exit 1' tasks: - id: log diff --git a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerConditionTest.java b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerConditionTest.java index faa02d9e..8ff27013 100644 --- a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerConditionTest.java +++ b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerConditionTest.java @@ -9,6 +9,7 @@ import java.util.Map; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; import static org.hamcrest.Matchers.is; import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; @@ -67,12 +68,19 @@ void nullCondition_doesNotMatch() { @Test void catastrophicBacktrackingRegex_fallsBackToSubstring_withoutHanging() { - String condition = "(a+)+$"; + // (a+)+$ is memoized by the JDK 25 regex engine and returns near-instantly, so it no longer + // exercises the 5s timeout guard; (.*a){20}$ still triggers catastrophic backtracking there. + String condition = "(.*a){20}$"; String value = "a".repeat(40) + "!"; + long start = System.nanoTime(); assertTimeoutPreemptively(Duration.ofSeconds(15), () -> assertThat(trigger.matchesCondition(output(condition, 0, Map.of("k", value))), is(false)) ); + long elapsedMs = Duration.ofNanos(System.nanoTime() - start).toMillis(); + + // Confirms the 5s timeout guard actually tripped rather than a fast regex miss. + assertThat(elapsedMs, greaterThanOrEqualTo(4000L)); } @Test diff --git a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerConditionTest.java b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerConditionTest.java index ec12c424..2f3d0c74 100644 --- a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerConditionTest.java +++ b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerConditionTest.java @@ -9,6 +9,7 @@ import java.util.Map; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; import static org.hamcrest.Matchers.is; import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; @@ -67,12 +68,19 @@ void nullCondition_doesNotMatch() { @Test void catastrophicBacktrackingRegex_fallsBackToSubstring_withoutHanging() { - String condition = "(a+)+$"; + // (a+)+$ is memoized by the JDK 25 regex engine and returns near-instantly, so it no longer + // exercises the 5s timeout guard; (.*a){20}$ still triggers catastrophic backtracking there. + String condition = "(.*a){20}$"; String value = "a".repeat(40) + "!"; + long start = System.nanoTime(); assertTimeoutPreemptively(Duration.ofSeconds(15), () -> assertThat(trigger.matchesCondition(output(condition, 0, Map.of("k", value))), is(false)) ); + long elapsedMs = Duration.ofNanos(System.nanoTime() - start).toMillis(); + + // Confirms the 5s timeout guard actually tripped rather than a fast regex miss. + assertThat(elapsedMs, greaterThanOrEqualTo(4000L)); } @Test From 0ba7de7c01ab8608d06ca65e0442ad84fc0813eb Mon Sep 17 00:00:00 2001 From: jymaire Date: Wed, 23 Sep 2026 16:56:07 +0200 Subject: [PATCH 6/6] fix(perl): persist trigger edge state in the namespace KV store The in-memory lastMatched flag was rebuilt with the trigger on every poll, so edge mode fired on every matching poll. Keep the previous result in the namespace KV store instead, as the Bun/.NET/PowerShell triggers do. Replaces the tautological AtomicBoolean edge tests with EdgeStateTest and an evaluate-level test that polls through a serialized copy. Refs #449 Co-Authored-By: Claude Opus 5.5 --- .../plugin/scripts/perl/CommandsTrigger.java | 40 +++- .../plugin/scripts/perl/ScriptTrigger.java | 45 ++-- .../scripts/perl/CommandsTriggerTest.java | 52 +++-- .../plugin/scripts/perl/EdgeStateTest.java | 194 ++++++++++++++++++ .../scripts/perl/ScriptTriggerTest.java | 27 +-- 5 files changed, 281 insertions(+), 77 deletions(-) create mode 100644 plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/EdgeStateTest.java diff --git a/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java index c43485b4..b896cf41 100644 --- a/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java +++ b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/CommandsTrigger.java @@ -5,7 +5,6 @@ import java.util.List; import java.util.Map; import java.util.Optional; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -25,12 +24,13 @@ import io.kestra.core.models.triggers.TriggerOutput; import io.kestra.core.models.triggers.TriggerService; import io.kestra.core.runners.RunContext; +import io.kestra.core.storages.kv.KVStore; +import io.kestra.core.storages.kv.KVValueAndMetadata; import io.kestra.plugin.scripts.exec.TriggerRunContext; import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput; import io.swagger.v3.oas.annotations.media.Schema; import jakarta.validation.constraints.NotNull; -import lombok.AccessLevel; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -132,6 +132,7 @@ public class CommandsTrigger extends AbstractTrigger title = "Edge trigger mode", description = """ If true, the trigger emits only on a transition from 'not matching' to 'matching' (anti-spam). + The previous result is kept in the namespace KV store, keyed by flow and trigger id. If false, the trigger emits on every poll where the condition matches. """ ) @@ -139,12 +140,6 @@ public class CommandsTrigger extends AbstractTrigger @PluginProperty(group = "advanced") protected Property edge = Property.ofValue(true); - // Known limitation: in-memory only — resets when the trigger is rehydrated (e.g. after restart), - // so edge mode may re-fire once after a scheduler restart. - @Builder.Default - @Getter(AccessLevel.NONE) - private final AtomicBoolean lastMatched = new AtomicBoolean(false); - @Override public Optional evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception { RunContext runContext = conditionContext.getRunContext(); @@ -160,9 +155,7 @@ public Optional evaluate(ConditionContext conditionContext, TriggerCo boolean matched = matchesCondition(out); - boolean emit = renderedEdge - ? (!lastMatched.getAndSet(matched) && matched) - : matched; + boolean emit = shouldEmit(runContext, context, renderedEdge, matched); if (!emit) { return Optional.empty(); @@ -171,6 +164,31 @@ public Optional evaluate(ConditionContext conditionContext, TriggerCo return Optional.of(TriggerService.generateExecution(this, conditionContext, context, out)); } + boolean shouldEmit(RunContext runContext, TriggerContext context, boolean edge, boolean matched) throws Exception { + if (!edge) { + return matched; + } + + // A polling trigger is rebuilt from the flow definition (and serialized to a worker) on + // every poll, so the previous result cannot live in a field. It is kept in the namespace + // KV store instead and advanced on every poll. + KVStore kvStore = runContext.namespaceKv(context.getNamespace()); + String key = edgeStateKey(context); + + boolean previouslyMatched = kvStore.getValue(key) + .map(value -> Boolean.parseBoolean(String.valueOf(value.value()))) + .orElse(false); + kvStore.put(key, new KVValueAndMetadata(null, matched)); + + return matched && !previouslyMatched; + } + + // Length prefixed so that the pairs ("a-b", "c") and ("a", "b-c") can never share a key. + // Flow and trigger ids only use characters that are valid in a KV key. + static String edgeStateKey(TriggerContext context) { + return "trigger-edge-" + context.getFlowId().length() + "-" + context.getFlowId() + "-" + context.getTriggerId(); + } + private Output runOnce(RunContext runContext) throws Exception { Commands task = Commands.builder() .id(this.getId()) diff --git a/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/ScriptTrigger.java b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/ScriptTrigger.java index db63bf48..7ede0504 100644 --- a/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/ScriptTrigger.java +++ b/plugin-script-perl/src/main/java/io/kestra/plugin/scripts/perl/ScriptTrigger.java @@ -4,7 +4,6 @@ import java.time.Instant; import java.util.Map; import java.util.Optional; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -25,12 +24,13 @@ import io.kestra.core.models.triggers.TriggerOutput; import io.kestra.core.models.triggers.TriggerService; import io.kestra.core.runners.RunContext; +import io.kestra.core.storages.kv.KVStore; +import io.kestra.core.storages.kv.KVValueAndMetadata; import io.kestra.plugin.scripts.exec.TriggerRunContext; import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput; import io.swagger.v3.oas.annotations.media.Schema; import jakarta.validation.constraints.NotNull; -import lombok.AccessLevel; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -47,7 +47,7 @@ @NoArgsConstructor @Schema( title = "Trigger on Perl script condition", - description = "Polls by running an inline Perl script in a container (default image perl) and emits when exitCondition matches. Supports edge mode to emit only on transitions and polls every 60s by default. Accepts 'exit N' or a regex (fallback substring) matched against emitted vars and failure logs." + description = "Polls by running an inline Perl script in a container (default image perl) and emits when exitCondition matches. Edge mode (the default) emits only on a transition from not matching to matching, remembering the previous result in the namespace KV store. Polls every 60s by default. Accepts 'exit N' or a regex (fallback substring) matched against emitted vars and failure logs." ) @Plugin( examples = { @@ -130,19 +130,15 @@ public class ScriptTrigger extends AbstractTrigger @Schema( title = "Edge trigger mode", description = """ - When true (default), emit only on a transition from not matching to matching. When false, emit on every poll that matches. + When true (default), emit only on a transition from not matching to matching, so a condition that \ + stays true does not fire on every poll. The previous result is kept in the namespace KV store, keyed \ + by flow and trigger id. When false, emit on every poll that matches. """ ) @Builder.Default @PluginProperty(group = "advanced") protected Property edge = Property.ofValue(true); - // Known limitation: in-memory only — resets when the trigger is rehydrated (e.g. after restart), - // so edge mode may re-fire once after a scheduler restart. - @Builder.Default - @Getter(AccessLevel.NONE) - private final AtomicBoolean lastMatched = new AtomicBoolean(false); - @Override public Optional evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception { RunContext runContext = conditionContext.getRunContext(); @@ -158,9 +154,7 @@ public Optional evaluate(ConditionContext conditionContext, TriggerCo boolean matched = matchesCondition(out); - boolean emit = renderedEdge - ? (!lastMatched.getAndSet(matched) && matched) - : matched; + boolean emit = shouldEmit(runContext, context, renderedEdge, matched); if (!emit) { return Optional.empty(); @@ -169,6 +163,31 @@ public Optional evaluate(ConditionContext conditionContext, TriggerCo return Optional.of(TriggerService.generateExecution(this, conditionContext, context, out)); } + boolean shouldEmit(RunContext runContext, TriggerContext context, boolean edge, boolean matched) throws Exception { + if (!edge) { + return matched; + } + + // A polling trigger is rebuilt from the flow definition (and serialized to a worker) on + // every poll, so the previous result cannot live in a field. It is kept in the namespace + // KV store instead and advanced on every poll. + KVStore kvStore = runContext.namespaceKv(context.getNamespace()); + String key = edgeStateKey(context); + + boolean previouslyMatched = kvStore.getValue(key) + .map(value -> Boolean.parseBoolean(String.valueOf(value.value()))) + .orElse(false); + kvStore.put(key, new KVValueAndMetadata(null, matched)); + + return matched && !previouslyMatched; + } + + // Length prefixed so that the pairs ("a-b", "c") and ("a", "b-c") can never share a key. + // Flow and trigger ids only use characters that are valid in a KV key. + static String edgeStateKey(TriggerContext context) { + return "trigger-edge-" + context.getFlowId().length() + "-" + context.getFlowId() + "-" + context.getTriggerId(); + } + private Output runOnce(RunContext runContext) throws Exception { Script task = Script.builder() .id(this.getId()) diff --git a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerTest.java b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerTest.java index 79ea87c1..412710f9 100644 --- a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerTest.java +++ b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/CommandsTriggerTest.java @@ -3,7 +3,6 @@ import java.util.List; import java.util.Map; import java.util.Optional; -import java.util.concurrent.atomic.AtomicBoolean; import org.junit.jupiter.api.Test; @@ -11,6 +10,8 @@ import io.kestra.core.models.executions.Execution; import io.kestra.core.models.property.Property; import io.kestra.core.runners.RunContextFactory; +import io.kestra.core.serializers.JacksonMapper; +import io.kestra.core.utils.IdUtils; import io.kestra.core.utils.TestsUtils; import jakarta.inject.Inject; @@ -27,7 +28,7 @@ class CommandsTriggerTest { @Test void commandsTrigger_shouldTriggerOnImplicitFailureExit1() throws Exception { CommandsTrigger trigger = CommandsTrigger.builder() - .id("commands-trigger") + .id("commands-trigger-" + IdUtils.create()) .type(CommandsTrigger.class.getName()) .exitCondition(Property.ofValue("exit 1")) .edge(Property.ofValue(true)) @@ -50,7 +51,7 @@ void commandsTrigger_shouldTriggerOnImplicitFailureExit1() throws Exception { @Test void commandsTrigger_shouldTriggerOnStdoutMatchUsingStructuredOutputs() throws Exception { CommandsTrigger trigger = CommandsTrigger.builder() - .id("commands-stdout-match-trigger") + .id("commands-stdout-match-trigger-" + IdUtils.create()) .type(CommandsTrigger.class.getName()) .exitCondition(Property.ofValue("toto")) .edge(Property.ofValue(true)) @@ -74,7 +75,7 @@ void commandsTrigger_shouldTriggerOnStdoutMatchUsingStructuredOutputs() throws E @Test void commandsTrigger_shouldNotEmitWhenConditionDoesNotMatch() throws Exception { CommandsTrigger trigger = CommandsTrigger.builder() - .id("commands-no-match-trigger") + .id("commands-no-match-trigger-" + IdUtils.create()) .type(CommandsTrigger.class.getName()) .exitCondition(Property.ofValue("exit 1")) .edge(Property.ofValue(true)) @@ -91,7 +92,7 @@ void commandsTrigger_shouldNotEmitWhenConditionDoesNotMatch() throws Exception { @Test void commandsTrigger_shouldMatchRegexAgainstStructuredOutputs() throws Exception { CommandsTrigger trigger = CommandsTrigger.builder() - .id("commands-regex-trigger") + .id("commands-regex-trigger-" + IdUtils.create()) .type(CommandsTrigger.class.getName()) .exitCondition(Property.ofValue("status=\\w+")) .edge(Property.ofValue(true)) @@ -110,27 +111,24 @@ void commandsTrigger_shouldMatchRegexAgainstStructuredOutputs() throws Exception } @Test - void edgeMode_preventsConsecutiveEmit() { - AtomicBoolean lastMatched = new AtomicBoolean(false); - - // First match: transition false->true => should emit - boolean matched1 = true; - boolean emit1 = !lastMatched.getAndSet(matched1) && matched1; - assertThat("first match should emit", emit1, is(true)); - - // Second consecutive match: true->true => should NOT emit - boolean matched2 = true; - boolean emit2 = !lastMatched.getAndSet(matched2) && matched2; - assertThat("consecutive match should NOT emit in edge mode", emit2, is(false)); - - // Non-match: true->false => should not emit - boolean matched3 = false; - boolean emit3 = !lastMatched.getAndSet(matched3) && matched3; - assertThat("non-match should not emit", emit3, is(false)); - - // Match again after non-match: false->true => should emit - boolean matched4 = true; - boolean emit4 = !lastMatched.getAndSet(matched4) && matched4; - assertThat("match after non-match should emit", emit4, is(true)); + void commandsTrigger_edgeModeShouldSuppressSecondEmission() throws Exception { + CommandsTrigger trigger = CommandsTrigger.builder() + .id("commands-edge-trigger-" + IdUtils.create()) + .type(CommandsTrigger.class.getName()) + .exitCondition(Property.ofValue("exit 1")) + .edge(Property.ofValue(true)) + .containerImage(Property.ofValue("perl:latest")) + .commands(Property.ofValue(List.of("perl -e 'exit 1;'"))) + .build(); + + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + Optional first = trigger.evaluate(context.getKey(), context.getValue()); + assertThat("First evaluation should fire", first.isPresent(), is(true)); + + // The second poll runs on a copy that went through the worker's serialize/deserialize round trip. + CommandsTrigger nextPoll = JacksonMapper.ofJson().readValue(JacksonMapper.ofJson().writeValueAsString(trigger), CommandsTrigger.class); + context = TestsUtils.mockTrigger(runContextFactory, nextPoll); + Optional second = nextPoll.evaluate(context.getKey(), context.getValue()); + assertThat("Edge mode should suppress repeated emission", second.isPresent(), is(false)); } } diff --git a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/EdgeStateTest.java b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/EdgeStateTest.java new file mode 100644 index 00000000..4be30426 --- /dev/null +++ b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/EdgeStateTest.java @@ -0,0 +1,194 @@ +package io.kestra.plugin.scripts.perl; + +import java.time.ZonedDateTime; +import java.util.List; + +import org.junit.jupiter.api.Test; + +import io.kestra.core.junit.annotations.KestraTest; +import io.kestra.core.models.property.Property; +import io.kestra.core.models.triggers.Trigger; +import io.kestra.core.models.triggers.TriggerContext; +import io.kestra.core.runners.RunContext; +import io.kestra.core.runners.RunContextFactory; +import io.kestra.core.serializers.JacksonMapper; +import io.kestra.core.utils.IdUtils; +import io.kestra.core.utils.TestsUtils; + +import jakarta.inject.Inject; + +import static io.kestra.core.tenant.TenantService.MAIN_TENANT; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.not; + +/** + * Edge mode keeps its previous result in the namespace KV store. None of these tests need Docker: + * they drive shouldEmit directly, so only the state handling is under test. + * + * The KV store used by the test config is a local folder that survives between runs, so every + * test uses a unique trigger id. + */ +@KestraTest +class EdgeStateTest { + @Inject + private RunContextFactory runContextFactory; + + private static ScriptTrigger scriptTrigger(String id) { + return ScriptTrigger.builder() + .id(id) + .type(ScriptTrigger.class.getName()) + .exitCondition(Property.ofValue("exit 1")) + .script(Property.ofValue("unused")) + .build(); + } + + private static CommandsTrigger commandsTrigger(String id) { + return CommandsTrigger.builder() + .id(id) + .type(CommandsTrigger.class.getName()) + .exitCondition(Property.ofValue("exit 1")) + .commands(Property.ofValue(List.of("unused"))) + .build(); + } + + // The scheduler hands every poll a freshly deserialized trigger, so each poll here does the same. + private static T freshCopy(T trigger, Class type) throws Exception { + return JacksonMapper.ofJson().readValue(JacksonMapper.ofJson().writeValueAsString(trigger), type); + } + + private static TriggerContext contextFor(String flowId, String triggerId, String namespace) { + return Trigger.builder() + .triggerId(triggerId) + .flowId(flowId) + .tenantId(MAIN_TENANT) + .namespace(namespace) + .date(ZonedDateTime.now()) + .build(); + } + + @Test + void scriptTrigger_edgeEmitsOnlyOnTransitionAcrossFreshInstances() throws Exception { + ScriptTrigger original = scriptTrigger("script-edge-" + IdUtils.create()); + var mock = TestsUtils.mockTrigger(runContextFactory, original); + RunContext runContext = mock.getKey().getRunContext(); + TriggerContext context = mock.getValue(); + + boolean[] polls = { false, true, true, true, false, false, true }; + boolean[] expected = { false, true, false, false, false, false, true }; + + for (int i = 0; i < polls.length; i++) { + boolean emit = freshCopy(original, ScriptTrigger.class).shouldEmit(runContext, context, true, polls[i]); + assertThat("poll " + i + " (matched=" + polls[i] + ")", emit, is(expected[i])); + } + } + + @Test + void commandsTrigger_edgeEmitsOnlyOnTransitionAcrossFreshInstances() throws Exception { + CommandsTrigger original = commandsTrigger("commands-edge-" + IdUtils.create()); + var mock = TestsUtils.mockTrigger(runContextFactory, original); + RunContext runContext = mock.getKey().getRunContext(); + TriggerContext context = mock.getValue(); + + boolean[] polls = { false, true, true, true, false, false, true }; + boolean[] expected = { false, true, false, false, false, false, true }; + + for (int i = 0; i < polls.length; i++) { + boolean emit = freshCopy(original, CommandsTrigger.class).shouldEmit(runContext, context, true, polls[i]); + assertThat("poll " + i + " (matched=" + polls[i] + ")", emit, is(expected[i])); + } + } + + @Test + void scriptTrigger_edgeDisabledEmitsOnEveryMatchAndKeepsNoState() throws Exception { + ScriptTrigger original = scriptTrigger("script-noedge-" + IdUtils.create()); + var mock = TestsUtils.mockTrigger(runContextFactory, original); + RunContext runContext = mock.getKey().getRunContext(); + TriggerContext context = mock.getValue(); + + assertThat(original.shouldEmit(runContext, context, false, true), is(true)); + assertThat(original.shouldEmit(runContext, context, false, true), is(true)); + assertThat(original.shouldEmit(runContext, context, false, false), is(false)); + + assertThat( + "edge=false must not write to the KV store", + runContext.namespaceKv(context.getNamespace()).getValue(ScriptTrigger.edgeStateKey(context)).isPresent(), + is(false) + ); + } + + @Test + void commandsTrigger_edgeDisabledEmitsOnEveryMatchAndKeepsNoState() throws Exception { + CommandsTrigger original = commandsTrigger("commands-noedge-" + IdUtils.create()); + var mock = TestsUtils.mockTrigger(runContextFactory, original); + RunContext runContext = mock.getKey().getRunContext(); + TriggerContext context = mock.getValue(); + + assertThat(original.shouldEmit(runContext, context, false, true), is(true)); + assertThat(original.shouldEmit(runContext, context, false, true), is(true)); + assertThat(original.shouldEmit(runContext, context, false, false), is(false)); + + assertThat( + "edge=false must not write to the KV store", + runContext.namespaceKv(context.getNamespace()).getValue(CommandsTrigger.edgeStateKey(context)).isPresent(), + is(false) + ); + } + + @Test + void scriptTrigger_stateIsScopedToFlowAndTrigger() throws Exception { + String triggerId = "script-scope-" + IdUtils.create(); + ScriptTrigger trigger = scriptTrigger(triggerId); + var mock = TestsUtils.mockTrigger(runContextFactory, trigger); + RunContext runContext = mock.getKey().getRunContext(); + String namespace = mock.getValue().getNamespace(); + + TriggerContext flowA = contextFor("flow-a", triggerId, namespace); + TriggerContext flowB = contextFor("flow-b", triggerId, namespace); + TriggerContext otherTriggerInFlowA = contextFor("flow-a", triggerId + "-other", namespace); + + assertThat(trigger.shouldEmit(runContext, flowA, true, true), is(true)); + assertThat("same trigger id in another flow has its own state", trigger.shouldEmit(runContext, flowB, true, true), is(true)); + assertThat("another trigger in the same flow has its own state", trigger.shouldEmit(runContext, otherTriggerInFlowA, true, true), is(true)); + + assertThat(trigger.shouldEmit(runContext, flowA, true, true), is(false)); + assertThat(trigger.shouldEmit(runContext, flowB, true, true), is(false)); + } + + @Test + void commandsTrigger_stateIsScopedToFlowAndTrigger() throws Exception { + String triggerId = "commands-scope-" + IdUtils.create(); + CommandsTrigger trigger = commandsTrigger(triggerId); + var mock = TestsUtils.mockTrigger(runContextFactory, trigger); + RunContext runContext = mock.getKey().getRunContext(); + String namespace = mock.getValue().getNamespace(); + + TriggerContext flowA = contextFor("flow-a", triggerId, namespace); + TriggerContext flowB = contextFor("flow-b", triggerId, namespace); + TriggerContext otherTriggerInFlowA = contextFor("flow-a", triggerId + "-other", namespace); + + assertThat(trigger.shouldEmit(runContext, flowA, true, true), is(true)); + assertThat("same trigger id in another flow has its own state", trigger.shouldEmit(runContext, flowB, true, true), is(true)); + assertThat("another trigger in the same flow has its own state", trigger.shouldEmit(runContext, otherTriggerInFlowA, true, true), is(true)); + + assertThat(trigger.shouldEmit(runContext, flowA, true, true), is(false)); + assertThat(trigger.shouldEmit(runContext, flowB, true, true), is(false)); + } + + @Test + void edgeStateKey_doesNotCollideWhenIdsShareHyphens() { + // "a-b" + "c" and "a" + "b-c" would both read "a-b-c" without the length prefix. + TriggerContext first = contextFor("a-b", "c", "company.team"); + TriggerContext second = contextFor("a", "b-c", "company.team"); + + assertThat(ScriptTrigger.edgeStateKey(first), is(not(ScriptTrigger.edgeStateKey(second)))); + assertThat(CommandsTrigger.edgeStateKey(first), is(not(CommandsTrigger.edgeStateKey(second)))); + } + + @Test + void edgeStateKey_isAValidKvKey() { + String key = ScriptTrigger.edgeStateKey(contextFor("my_flow-1", "my_trigger-2", "company.team")); + + assertThat(key.matches("[a-zA-Z0-9][a-zA-Z0-9._-]*"), is(true)); + } +} diff --git a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerTest.java b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerTest.java index 4ae863ca..3e080d5a 100644 --- a/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerTest.java +++ b/plugin-script-perl/src/test/java/io/kestra/plugin/scripts/perl/ScriptTriggerTest.java @@ -4,13 +4,12 @@ import java.time.Instant; import java.util.Map; -import java.util.concurrent.atomic.AtomicBoolean; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.is; /** - * Unit tests for ScriptTrigger's condition-matching logic and edge mode. + * Unit tests for ScriptTrigger's condition-matching logic. Edge mode is covered by EdgeStateTest. * * These tests exercise matchesCondition via the Output model without requiring a Perl * runtime, which may not be available on all CI machines. @@ -68,28 +67,4 @@ void nullCondition_shouldNotMatch() { void exitZeroCondition_shouldMatchSuccessfulExecution() { assertThat(trigger.matchesCondition(output("exit 0", 0, null)), is(true)); } - - @Test - void edgeMode_shouldEmitOnFirstMatch() { - var lastMatched = new AtomicBoolean(false); - boolean matched = true; - boolean emit = !lastMatched.getAndSet(matched) && matched; - assertThat("first match should emit", emit, is(true)); - } - - @Test - void edgeMode_shouldSuppressConsecutiveMatches() { - var lastMatched = new AtomicBoolean(true); - boolean matched = true; - boolean emit = !lastMatched.getAndSet(matched) && matched; - assertThat("consecutive match should not emit in edge mode", emit, is(false)); - } - - @Test - void edgeMode_shouldEmitAgainAfterNonMatch() { - var lastMatched = new AtomicBoolean(false); - boolean matched = true; - boolean emit = !lastMatched.getAndSet(matched) && matched; - assertThat("match after non-match should emit", emit, is(true)); - } }