diff --git a/AGENTS.md b/AGENTS.md index 9a0cd2ca..27b9c660 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -110,7 +110,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/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..07e2ebfb --- /dev/null +++ b/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/CommandsTrigger.java @@ -0,0 +1,320 @@ +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.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.*; +import lombok.experimental.SuperBuilder; + +import java.time.Duration; +import java.time.Instant; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +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 }})" + """ + ) + } +) +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); + + // The trigger is rebuilt on every poll, so the compiled condition has to live in a static. + // Conditions can be templated, so keep it bounded and drop the least recently used entry. + private static final int MAX_CACHED_CONDITIONS = 64; + private static final Map CONDITION_PATTERNS = Collections.synchronizedMap( + new LinkedHashMap<>(16, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry eldest) { + return size() > MAX_CACHED_CONDITIONS; + } + } + ); + + @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 = """ + 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); + + @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 = shouldEmit(runContext, context, edgeEnabled, matched); + + if (!emit) { + return Optional.empty(); + } + + 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()) + .type(Commands.class.getName()) + .containerImage(this.containerImage) + .commands(this.commands) + .build(); + + String renderedCondition = runContext.render(this.exitCondition) + .as(String.class) + .filter(condition -> !condition.isBlank()) + .orElseThrow(() -> new IllegalArgumentException("exitCondition must render to a non-empty value")); + + 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 { + return conditionPattern(cond).matcher(haystack).find(); + } catch (Exception invalidRegex) { + return haystack.contains(cond); + } + } + + static Pattern conditionPattern(String condition) { + return CONDITION_PATTERNS.computeIfAbsent(condition, Pattern::compile); + } + + 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..6b4c06d9 --- /dev/null +++ b/plugin-script-r/src/main/java/io/kestra/plugin/scripts/r/ScriptTrigger.java @@ -0,0 +1,316 @@ +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.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.*; +import lombok.experimental.SuperBuilder; + +import java.time.Duration; +import java.time.Instant; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Optional; +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 }})" + """ + ) + } +) +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); + + // The trigger is rebuilt on every poll, so the compiled condition has to live in a static. + // Conditions can be templated, so keep it bounded and drop the least recently used entry. + private static final int MAX_CACHED_CONDITIONS = 64; + private static final Map CONDITION_PATTERNS = Collections.synchronizedMap( + new LinkedHashMap<>(16, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry eldest) { + return size() > MAX_CACHED_CONDITIONS; + } + } + ); + + @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 = """ + 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); + + @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 = shouldEmit(runContext, context, edgeEnabled, matched); + + if (!emit) { + return Optional.empty(); + } + + return Optional.of( + TriggerService.generateExecution(this, conditionContext, context, output) + ); + } + + 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()) + .type(Script.class.getName()) + .containerImage(this.containerImage) + .script(this.script) + .build(); + + String renderedCondition = runContext.render(this.exitCondition) + .as(String.class) + .filter(condition -> !condition.isBlank()) + .orElseThrow(() -> new IllegalArgumentException("exitCondition must render to a non-empty value")); + + 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 { + return conditionPattern(cond).matcher(haystack).find(); + } catch (Exception invalidRegex) { + return haystack.contains(cond); + } + } + + static Pattern conditionPattern(String condition) { + return CONDITION_PATTERNS.computeIfAbsent(condition, Pattern::compile); + } + + 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/main/resources/doc/io.kestra.plugin.scripts.r.md b/plugin-script-r/src/main/resources/doc/io.kestra.plugin.scripts.r.md index 1205201c..402eefc6 100644 --- a/plugin-script-r/src/main/resources/doc/io.kestra.plugin.scripts.r.md +++ b/plugin-script-r/src/main/resources/doc/io.kestra.plugin.scripts.r.md @@ -11,3 +11,24 @@ Run R scripts for statistical computing and data analysis inside a container as `Script` runs inline R code defined in the `script` property — best for short, flow-specific logic. `Commands` runs shell commands (e.g., `Rscript main.R`) against script files; use it when your code lives in [namespace files](https://kestra.io/docs/concepts/namespace-files) shared across flows, or is cloned from a Git repository with a preceding `Clone` task. Install CRAN packages in `beforeCommands` with `Rscript -e 'install.packages(c("dplyr", "ggplot2"), repos="https://cloud.r-project.org")'`. For reproducible environments, use a `renv.lock` file via namespace files and restore with `Rscript -e 'renv::restore()'` — or build a custom `containerImage` with packages pre-installed to avoid per-run install time. + +## Triggers + +### ScriptTrigger + +Polls on an interval by running an inline R script the same way the `Script` task does, and starts an execution when `exitCondition` matches. The script runs in a fresh container on every poll, so keep it quick — and remember that installing CRAN packages on each poll is rarely what you want; prefer an image that already has them. + +Required properties: +- `script`: inline R script body +- `exitCondition`: either `exit N`, which matches when the script exits with code N, or a regex (with substring fallback) matched against the vars the script emits with `::{"outputs":{...}}::` + +Optional: +- `interval`: time between polls, defaults to `PT60S` +- `edge`: defaults to `true`, so the trigger fires only when the condition changes from not matching to matching. The previous result is kept in the namespace KV store under a key starting with `trigger-edge-`. Set to `false` to fire on every matching poll +- `containerImage`: defaults to `r-base` + +The trigger outputs are available as `{{ trigger.timestamp }}`, `{{ trigger.condition }}`, `{{ trigger.exitCode }}` and `{{ trigger.vars }}`. A failed run has no vars, so only an `exit N` condition can match a failure. + +### CommandsTrigger + +Same behavior as `ScriptTrigger`, but runs a list of shell commands the way the `Commands` task does. Required properties are `commands` and `exitCondition`; `interval`, `edge` and `containerImage` work as above. 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..930a15c3 --- /dev/null +++ b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/CommandsTriggerTest.java @@ -0,0 +1,138 @@ +package io.kestra.plugin.scripts.r; + +import java.util.List; +import java.util.Map; +import java.util.Optional; + +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.IdUtils; +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-" + IdUtils.create()) + .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-" + IdUtils.create()) + .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-" + IdUtils.create()) + .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)); + } + + // Drives the real evaluate() wiring rather than the edge arithmetic on its own, so a + // regression in how evaluate() consults the stored state is caught here. + @Test + void edgeMode_preventsConsecutiveEmit() throws Exception { + CommandsTrigger trigger = CommandsTrigger.builder() + .id("commands-edge-" + IdUtils.create()) + .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); + + assertThat( + "first matching poll should emit", + trigger.evaluate(context.getKey(), context.getValue()).isPresent(), + is(true) + ); + assertThat( + "a condition that stays matched should not emit again in edge mode", + trigger.evaluate(context.getKey(), context.getValue()).isPresent(), + is(false) + ); + } + + @Test + void edgeDisabled_emitsOnEveryMatchingPoll() throws Exception { + CommandsTrigger trigger = CommandsTrigger.builder() + .id("commands-noedge-" + IdUtils.create()) + .type(CommandsTrigger.class.getName()) + .exitCondition(Property.ofValue("exit 1")) + .edge(Property.ofValue(false)) + .containerImage(Property.ofValue("r-base")) + .commands(Property.ofValue(List.of("Rscript -e 'quit(status = 1)'"))) + .build(); + + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + + assertThat(trigger.evaluate(context.getKey(), context.getValue()).isPresent(), is(true)); + assertThat( + "edge=false should emit on every matching poll", + trigger.evaluate(context.getKey(), context.getValue()).isPresent(), + is(true) + ); + } +} diff --git a/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/EdgeStateTest.java b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/EdgeStateTest.java new file mode 100644 index 00000000..b8e6badb --- /dev/null +++ b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/EdgeStateTest.java @@ -0,0 +1,194 @@ +package io.kestra.plugin.scripts.r; + +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-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..d7cd9404 --- /dev/null +++ b/plugin-script-r/src/test/java/io/kestra/plugin/scripts/r/ScriptTriggerTest.java @@ -0,0 +1,77 @@ +package io.kestra.plugin.scripts.r; + +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.Map; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; + +/** + * Unit tests for ScriptTrigger's condition-matching logic. + * + * These tests exercise matchesCondition via the Output model without requiring an R + * runtime, which may not be available on all CI machines. Edge mode is covered in + * EdgeStateTest, and 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 invalidRegex_shouldFallBackToSubstringMatch() { + assertThat(trigger.matchesCondition(output("a[b", 0, Map.of("k", "xa[by"))), is(true)); + assertThat(trigger.matchesCondition(output("a[b", 0, Map.of("k", "nothing"))), is(false)); + } +}