Fix: source entrypoint with custom shell (#751)

* fix: adding exec_entrypoint script

* Revert "fix: adding exec_entrypoint script"

This reverts commit ce373c2.

* feat: add cmds for custom shell

* refactor: rename entrypoint cmds and script

* remove: join template func

ext/scheduler/airflow/dag/compiler_test.go

@@ -157,36 +157,48 @@ func (m mockPluginRepo) GetByName(name string) (*plugin.Plugin, error) {
 func setupPluginRepo() mockPluginRepo {
 	execUnit := new(mock.YamlMod)
 	execUnit.On("PluginInfo").Return(&plugin.Info{
-		Name:       "bq-bq",
-		Image:      "example.io/namespace/bq2bq-executor:latest",
-		Entrypoint: "python3 /opt/bumblebee/main.py",
+		Name:  "bq-bq",
+		Image: "example.io/namespace/bq2bq-executor:latest",
+		Entrypoint: plugin.Entrypoint{
+			Shell:  "/bin/bash",
+			Script: "python3 /opt/bumblebee/main.py",
+		},
 	}, nil)
 
 	transporterHook := "transporter"
 	hookUnit := new(mock.YamlMod)
 	hookUnit.On("PluginInfo").Return(&plugin.Info{
-		Name:       transporterHook,
-		HookType:   plugin.HookTypePre,
-		Image:      "example.io/namespace/transporter-executor:latest",
-		Entrypoint: "java -cp /opt/transporter/transporter.jar:/opt/transporter/jolokia-jvm-agent.jar -javaagent:jolokia-jvm-agent.jar=port=7777,host=0.0.0.0 com.gojek.transporter.Main",
-		DependsOn:  []string{"predator"},
+		Name:     transporterHook,
+		HookType: plugin.HookTypePre,
+		Image:    "example.io/namespace/transporter-executor:latest",
+		Entrypoint: plugin.Entrypoint{
+			Shell:  "/bin/sh",
+			Script: "java -cp /opt/transporter/transporter.jar:/opt/transporter/jolokia-jvm-agent.jar -javaagent:jolokia-jvm-agent.jar=port=7777,host=0.0.0.0 com.gojek.transporter.Main",
+		},
+		DependsOn: []string{"predator"},
 	}, nil)
 
 	predatorHook := "predator"
 	hookUnit2 := new(mock.YamlMod)
 	hookUnit2.On("PluginInfo").Return(&plugin.Info{
-		Name:       predatorHook,
-		HookType:   plugin.HookTypePost,
-		Image:      "example.io/namespace/predator-image:latest",
-		Entrypoint: "predator ${SUB_COMMAND} -s ${PREDATOR_URL} -u \"${BQ_PROJECT}.${BQ_DATASET}.${BQ_TABLE}\"",
+		Name:     predatorHook,
+		HookType: plugin.HookTypePost,
+		Image:    "example.io/namespace/predator-image:latest",
+		Entrypoint: plugin.Entrypoint{
+			Shell:  "/bin/sh",
+			Script: "predator ${SUB_COMMAND} -s ${PREDATOR_URL} -u \"${BQ_PROJECT}.${BQ_DATASET}.${BQ_TABLE}\"",
+		},
 	}, nil)
 
 	hookUnit3 := new(mock.YamlMod)
 	hookUnit3.On("PluginInfo").Return(&plugin.Info{
-		Name:       "failureHook",
-		HookType:   plugin.HookTypeFail,
-		Image:      "example.io/namespace/failure-hook-image:latest",
-		Entrypoint: "sleep 5",
+		Name:     "failureHook",
+		HookType: plugin.HookTypeFail,
+		Image:    "example.io/namespace/failure-hook-image:latest",
+		Entrypoint: plugin.Entrypoint{
+			Shell:  "/bin/sh",
+			Script: "sleep 5",
+		},
 	}, nil)
 
 	repo := mockPluginRepo{plugins: []*plugin.Plugin{

ext/scheduler/airflow/dag/dag.py.tmpl

@@ -102,12 +102,12 @@ IMAGE_PULL_POLICY = "IfNotPresent"
 INIT_CONTAINER_IMAGE = "odpf/optimus:{{.Version}}"
 INIT_CONTAINER_ENTRYPOINT = "/opt/entrypoint_init_container.sh"
 
-def get_entrypoint_cmd(plugin_entrypoint):
+def get_entrypoint_cmd(plugin_entrypoint_script):
     path_config = JOB_DIR + "/in/.env"
     path_secret = JOB_DIR + "/in/.secret"
     entrypoint = "set -o allexport; source {path_config}; set +o allexport; cat {path_config}; ".format(path_config=path_config)
     entrypoint += "set -o allexport; source {path_secret}; set +o allexport; ".format(path_secret=path_secret)
-    return entrypoint + plugin_entrypoint
+    return entrypoint + plugin_entrypoint_script
 
 volume = k8s.V1Volume(
     name='asset-volume',
@@ -150,8 +150,8 @@ init_container = k8s.V1Container(
     image_pull_policy=IMAGE_PULL_POLICY,
     namespace=conf.get('kubernetes', 'namespace', fallback="default"),
     image={{  .Task.Image | quote}},
-    cmds=["/bin/sh"],
-    arguments=["-c", get_entrypoint_cmd("""{{.Task.Entrypoint}} """)],
+    cmds=["{{.Task.Entrypoint.Shell}}", "-c"],
+    arguments=[get_entrypoint_cmd("""{{.Task.Entrypoint.Script}} """)],
     name="{{  .Task.Name | replace "_" "-" }}",
     task_id={{ .Task.Name | quote}},
     get_logs=True,
@@ -193,8 +193,8 @@ hook_{{$hookName}} = SuperKubernetesPodOperator(
     image_pull_policy=IMAGE_PULL_POLICY,
     namespace=conf.get('kubernetes', 'namespace', fallback="default"),
     image="{{ $t.Image }}",
-    cmds=["/bin/sh"],
-    arguments=["-c", get_entrypoint_cmd("""{{ $t.Entrypoint }} """)],
+    cmds=["{{$t.Entrypoint.Shell}}", "-c"],
+    arguments=[get_entrypoint_cmd("""{{$t.Entrypoint.Script}} """)],
     name="hook_{{ $t.Name | replace "_" "-" }}",
     task_id="hook_{{ $t.Name }}",
     get_logs=True,

ext/scheduler/airflow/dag/expected_dag.py

@@ -72,12 +72,12 @@
 INIT_CONTAINER_IMAGE = "odpf/optimus:dev"
 INIT_CONTAINER_ENTRYPOINT = "/opt/entrypoint_init_container.sh"
 
-def get_entrypoint_cmd(plugin_entrypoint):
+def get_entrypoint_cmd(plugin_entrypoint_script):
     path_config = JOB_DIR + "/in/.env"
     path_secret = JOB_DIR + "/in/.secret"
     entrypoint = "set -o allexport; source {path_config}; set +o allexport; cat {path_config}; ".format(path_config=path_config)
     entrypoint += "set -o allexport; source {path_secret}; set +o allexport; ".format(path_secret=path_secret)
-    return entrypoint + plugin_entrypoint
+    return entrypoint + plugin_entrypoint_script
 
 volume = k8s.V1Volume(
     name='asset-volume',
@@ -119,8 +119,8 @@ def get_entrypoint_cmd(plugin_entrypoint):
     image_pull_policy=IMAGE_PULL_POLICY,
     namespace=conf.get('kubernetes', 'namespace', fallback="default"),
     image="example.io/namespace/bq2bq-executor:latest",
-    cmds=["/bin/sh"],
-    arguments=["-c", get_entrypoint_cmd("""python3 /opt/bumblebee/main.py """)],
+    cmds=["/bin/bash", "-c"],
+    arguments=[get_entrypoint_cmd("""python3 /opt/bumblebee/main.py """)],
     name="bq-bq",
     task_id="bq-bq",
     get_logs=True,
@@ -156,8 +156,8 @@ def get_entrypoint_cmd(plugin_entrypoint):
     image_pull_policy=IMAGE_PULL_POLICY,
     namespace=conf.get('kubernetes', 'namespace', fallback="default"),
     image="example.io/namespace/transporter-executor:latest",
-    cmds=["/bin/sh"],
-    arguments=["-c", get_entrypoint_cmd("""java -cp /opt/transporter/transporter.jar:/opt/transporter/jolokia-jvm-agent.jar -javaagent:jolokia-jvm-agent.jar=port=7777,host=0.0.0.0 com.gojek.transporter.Main """)],
+    cmds=["/bin/sh", "-c"],
+    arguments=[get_entrypoint_cmd("""java -cp /opt/transporter/transporter.jar:/opt/transporter/jolokia-jvm-agent.jar -javaagent:jolokia-jvm-agent.jar=port=7777,host=0.0.0.0 com.gojek.transporter.Main """)],
     name="hook_transporter",
     task_id="hook_transporter",
     get_logs=True,
@@ -189,8 +189,8 @@ def get_entrypoint_cmd(plugin_entrypoint):
     image_pull_policy=IMAGE_PULL_POLICY,
     namespace=conf.get('kubernetes', 'namespace', fallback="default"),
     image="example.io/namespace/predator-image:latest",
-    cmds=["/bin/sh"],
-    arguments=["-c", get_entrypoint_cmd("""predator ${SUB_COMMAND} -s ${PREDATOR_URL} -u "${BQ_PROJECT}.${BQ_DATASET}.${BQ_TABLE}" """)],
+    cmds=["/bin/sh", "-c"],
+    arguments=[get_entrypoint_cmd("""predator ${SUB_COMMAND} -s ${PREDATOR_URL} -u "${BQ_PROJECT}.${BQ_DATASET}.${BQ_TABLE}" """)],
     name="hook_predator",
     task_id="hook_predator",
     get_logs=True,
@@ -222,8 +222,8 @@ def get_entrypoint_cmd(plugin_entrypoint):
     image_pull_policy=IMAGE_PULL_POLICY,
     namespace=conf.get('kubernetes', 'namespace', fallback="default"),
     image="example.io/namespace/failure-hook-image:latest",
-    cmds=["/bin/sh"],
-    arguments=["-c", get_entrypoint_cmd("""sleep 5 """)],
+    cmds=["/bin/sh", "-c"],
+    arguments=[get_entrypoint_cmd("""sleep 5 """)],
     name="hook_failureHook",
     task_id="hook_failureHook",
     get_logs=True,

ext/scheduler/airflow/dag/models.go

@@ -33,7 +33,7 @@ type TemplateContext struct {
 type Task struct {
 	Name       string
 	Image      string
-	Entrypoint string
+	Entrypoint plugin.Entrypoint
 }
 
 func PrepareTask(job *scheduler.Job, pluginRepo PluginRepo) (Task, error) {
@@ -54,7 +54,7 @@ func PrepareTask(job *scheduler.Job, pluginRepo PluginRepo) (Task, error) {
 type Hook struct {
 	Name       string
 	Image      string
-	Entrypoint string
+	Entrypoint plugin.Entrypoint
 	IsFailHook bool
 }
 

plugin/yaml/plugin.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"strings"
 
 	"github.com/hashicorp/go-hclog"
 	"github.com/spf13/afero"
@@ -104,6 +105,14 @@ func NewPluginSpec(pluginPath string) (*PluginSpec, error) {
 	if err := yaml.UnmarshalStrict(pluginBytes, &plugin); err != nil {
 		return &plugin, err
 	}
+	// default values
+	if plugin.Info.Entrypoint.Shell == "" {
+		plugin.Info.Entrypoint.Shell = "/bin/sh"
+	}
+
+	// standardize script value
+	script := plugin.Info.Entrypoint.Script
+	plugin.Info.Entrypoint.Script = strings.ReplaceAll(script, "\n", "; ")
 	return &plugin, nil
 }
 

plugin/yaml/plugin_test.go

@@ -16,7 +16,7 @@ import (
 type mockYamlMod struct {
 	Name          string
 	Image         string
-	Entrypoint    string
+	Entrypoint    plugin.Entrypoint
 	PluginVersion string
 	PluginType    string
 }
@@ -51,10 +51,13 @@ func TestYamlPlugin(t *testing.T) {
 	testYamlPluginPath := "tests/sample_plugin.yaml" // success
 	testYamlPluginName := "bq2bqtest"
 	expectedInfo := &plugin.Info{
-		Name:          "bq2bqtest",
-		Description:   "Testing",
-		Image:         "docker.io/odpf/optimus-task-bq2bq-executor:latest",
-		Entrypoint:    "sleep 100",
+		Name:        "bq2bqtest",
+		Description: "Testing",
+		Image:       "docker.io/odpf/optimus-task-bq2bq-executor:latest",
+		Entrypoint: plugin.Entrypoint{
+			Shell:  "/bin/bash",
+			Script: "sleep 100; sleep 150",
+		},
 		PluginType:    "task",
 		PluginMods:    []plugin.Mod{"cli"},
 		PluginVersion: "latest",
@@ -168,12 +171,20 @@ func TestYamlPlugin(t *testing.T) {
 			assert.NoError(t, err)
 			assert.NotEmpty(t, repo.GetAll())
 		})
+		t.Run("should use default when entrypoint cmds is empty", func(t *testing.T) {
+			pluginSpec, err := yaml.NewPluginSpec("tests/sample_plugin_without_shell.yaml")
+			assert.NoError(t, err)
+			assert.NotEmpty(t, pluginSpec)
+			assert.Equal(t, "/bin/sh", pluginSpec.Entrypoint.Shell)
+		})
 		t.Run("should returns error when load yaml when same name exists", func(t *testing.T) {
 			repoWithBinayPlugin := models.NewPluginRepository()
 			err := repoWithBinayPlugin.AddYaml(&mockYamlMod{
-				Name:          testYamlPluginName,
-				Image:         "sdsd",
-				Entrypoint:    "sleep 100",
+				Name:  testYamlPluginName,
+				Image: "sdsd",
+				Entrypoint: plugin.Entrypoint{
+					Script: "sleep 100",
+				},
 				PluginVersion: "asdasd",
 				PluginType:    plugin.TypeTask.String(),
 			})

plugin/yaml/tests/sample_plugin.yaml

@@ -6,7 +6,11 @@ pluginmods:
   - dependencyresolver
 pluginversion: latest
 image: docker.io/odpf/optimus-task-bq2bq-executor:latest
-entrypoint: "sleep 100"
+entrypoint:
+  shell: "/bin/bash"
+  script: |-
+    sleep 100
+    sleep 150
 
 questions:
   - name: PROJECT

plugin/yaml/tests/sample_plugin_without_shell.yaml

@@ -0,0 +1,24 @@
+name: bq2bqtest
+description: Testing
+plugintype: task
+pluginmods:
+  - cli
+  - dependencyresolver
+pluginversion: latest
+image: docker.io/odpf/optimus-task-bq2bq-executor:latest
+entrypoint:
+  script: "sleep 100"
+
+questions:
+  - name: PROJECT
+    prompt: Project ID
+    regexp: ^[a-zA-Z0-9_\-]+$
+    minlength: 3
+
+defaultconfig:
+- name: TEST
+  value: "{{.test}}"
+
+defaultassets:
+  - name: query.sql
+    value: Select * from "project.dataset.table";

plugin/yaml/tests/sample_plugin_without_version.yaml

@@ -5,7 +5,9 @@ pluginmods:
   - cli
 pluginversion: ""
 image: ""
-entrypoint: "sleep 100"
+entrypoint:
+  shell: "/bin/bash"
+  script: "sleep 100"
 
 questions:
   - name: PROJECT

sdk/plugin/mock/plugin.go

@@ -34,8 +34,10 @@ func (p *MockYamlMod) PluginInfo() *plugin.Info {
 		DependsOn:     nil,
 		HookType:      "",
 		Image:         "gcr.io/bq-plugin:dev",
-		Entrypoint:    "sleep 60",
-		PluginMods:    []plugin.Mod{plugin.ModTypeCLI},
+		Entrypoint: plugin.Entrypoint{
+			Script: "sleep 60",
+		},
+		PluginMods: []plugin.Mod{plugin.ModTypeCLI},
 	}
 }
 

sdk/plugin/plugin.go

@@ -37,6 +37,11 @@ func (m Mod) String() string {
 	return string(m)
 }
 
+type Entrypoint struct {
+	Shell  string
+	Script string
+}
+
 type Info struct {
 	// Name should as simple as possible with no special characters
 	// should start with a character, better if all lowercase
@@ -52,7 +57,7 @@ type Info struct {
 	Image string
 
 	// Entrypoint command which will be used to execute the plugin
-	Entrypoint string
+	Entrypoint Entrypoint
 
 	// DependsOn returns list of hooks this should be executed after
 	DependsOn []string `yaml:",omitempty"`
@@ -78,8 +83,8 @@ func (info *Info) Validate() error {
 	}
 
 	// entrypoint is a required field
-	if info.Entrypoint == "" {
-		return errors.New("entrypoint cannot be empty")
+	if info.Entrypoint.Script == "" {
+		return errors.New("entrypoint script cannot be empty")
 	}
 
 	switch info.PluginType {