Skip to content
New issue

Have a question about this project? # for a free GitHub account to open an issue and contact its maintainers and the community.

#

By clicking “#”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? # to your account

feat: task runner #67

Merged
merged 1 commit into from
Jun 25, 2024
Merged

feat: task runner #67

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
36 changes: 28 additions & 8 deletions src/main/java/io/kestra/plugin/spark/AbstractSubmit.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,15 @@
import io.kestra.core.models.tasks.RunnableTask;
import io.kestra.core.models.tasks.Task;
import io.kestra.core.models.tasks.runners.ScriptService;
import io.kestra.core.models.tasks.runners.TaskRunner;
import io.kestra.core.runners.RunContext;
import io.kestra.plugin.scripts.exec.scripts.models.DockerOptions;
import io.kestra.plugin.scripts.exec.scripts.models.RunnerType;
import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput;
import io.kestra.plugin.scripts.exec.scripts.runners.CommandsWrapper;
import io.kestra.plugin.scripts.runner.docker.Docker;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.Valid;
import lombok.*;
import lombok.experimental.SuperBuilder;
import org.apache.commons.io.IOUtils;
Expand Down Expand Up @@ -103,26 +106,41 @@ public abstract class AbstractSubmit extends Task implements RunnableTask<Script
)
protected Map<String, String> env;

@Builder.Default
@Schema(
title = "Script runner to use."
title = "Script runner to use.",
description = "Deprecated - use 'taskRunner' instead."
)
@PluginProperty
@NotNull
@NotEmpty
protected RunnerType runner = RunnerType.PROCESS;
protected RunnerType runner;

@Schema(
title = "Deprecated, use 'taskRunner' instead"
)
@PluginProperty
@Deprecated
private DockerOptions docker;

@Schema(
title = "Docker options when using the `DOCKER` runner.",
defaultValue = "{image=" + DEFAULT_IMAGE + ", pullPolicy=ALWAYS}"
title = "The task runner to use.",
description = "Task runners are provided by plugins, each have their own properties."
)
@PluginProperty
@Builder.Default
protected DockerOptions docker = DockerOptions.builder().build();
@Valid
private TaskRunner taskRunner = Docker.INSTANCE;

@Schema(title = "The task runner container image, only used if the task runner is container-based.")
@PluginProperty(dynamic = true)
@Builder.Default
private String containerImage = DEFAULT_IMAGE;

abstract protected void configure(RunContext runContext, SparkLauncher spark) throws Exception;

protected DockerOptions injectDefaults(DockerOptions original) {
if (original == null) {
return null;
}

var builder = original.toBuilder();
if (original.getImage() == null) {
builder.image(DEFAULT_IMAGE);
Expand Down Expand Up @@ -165,6 +183,8 @@ public ScriptOutput run(RunContext runContext) throws Exception {
.withEnv(this.envs(runContext))
.withRunnerType(this.runner)
.withDockerOptions(injectDefaults(this.getDocker()))
.withTaskRunner(this.taskRunner)
.withContainerImage(this.containerImage)
.withCommands(ScriptService.scriptCommands(
List.of("/bin/sh", "-c"),
List.of(),
Expand Down