Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`
Expand All @@ -108,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`
Expand Down
3 changes: 3 additions & 0 deletions plugin-script-perl/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Original file line number Diff line number Diff line change
@@ -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<CommandsTrigger.Output> {

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<String> 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<List<String>> 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<String> 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<Boolean> edge = Property.ofValue(true);

@Builder.Default
@Getter(AccessLevel.NONE)
private final AtomicBoolean lastMatched = new AtomicBoolean(false);

@Override
public Optional<Execution> 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<String, Object> 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<String, Object> 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<String, Object> vars;
}
}
Loading
Loading