Using couler-proj/couler (#3075)

* migrate couler-proj/couler

* remove old code

* update

* fix ci

* update

* update

* remove unnecessary file

* debug ci

* update

* test ci

* fix ci

* fix ci

* fix ci

* fix invalide env

* escape couler env value

Author: Yancey0623

docker/dev/build.sh

@@ -29,10 +29,6 @@ echo "Build sqlflowserver, sqlflow, and step into $SQLFLOW_BIN ..."
 go generate ./...
 GOBIN=$SQLFLOW_BIN go install ./...
 
-echo "Build $SQLFLOWPATH/python/couler into $SQLFLOW_BIN ..."
-cd $SQLFLOWPATH/python/couler
-python setup.py bdist_wheel -q --dist-dir $SQLFLOW_BIN > /dev/null
-
 echo "Build Fluid ..."
 cd $SQLFLOW_BIN
 if [[ ! -d fluid ]]; then
@@ -88,6 +84,15 @@ git fetch origin # The residual local repo might not be on a branch.
 git checkout 5dc6421f562ea447e501fa355a48a6ee89856a1d
 python setup.py bdist_wheel -q --dist-dir $SQLFLOW_BIN > /dev/null
 
+echo "Build couler-proj/couler ..."
+cd $SQLFLOW_BIN
+if [[ ! -d couler ]]; then
+    git clone https://github.com/couler-proj/couler.git
+fi
+cd couler
+git checkout 9c93f72791ebf5411fc5bec7d68890de753d8431 
+python setup.py bdist_wheel -q --dist-dir $SQLFLOW_BIN > /dev/null
+
 echo "Convert tutorials from Markdown to IPython notebooks ..."
 mkdir -p $SQLFLOW_BIN/tutorial
 for file in $SQLFLOWPATH/doc/tutorial/*.md; do

go/codegen/experimental/codegen_couler.go

@@ -131,9 +131,9 @@ func CodeGenCouler(stepList []*stepContext, session *pb.Session) (string, error)
 func GetPyFuncBody(program string, funcName string) (string, error) {
 	const coulerGetPyFuncCodeImpl = `
 %s
-
-import couler.pyfunc as pyfunc
-print(pyfunc.body(%s))
+from couler.core.utils import body as pybody
+import sys
+print(pybody(%s), file=sys.stderr)
 `
 
 	tmpFile, err := ioutil.TempFile("/tmp", "sqlflow-couler-tmp")
@@ -157,20 +157,24 @@ print(pyfunc.body(%s))
 	if err := cmd.Run(); err != nil {
 		return "", fmt.Errorf("%v: %s\nCode is:\n%s", err, stderr, coulerCode)
 	}
-	return strings.TrimSpace(stdout.String()), nil
+	return strings.TrimSpace(stderr.String()), nil
 }
 
 const coulerCodeTmpl = `
 import couler.argo as couler
-import couler.pyfunc as pyfunc
+from couler.core.utils import body as pybody
 from os import path
 import json
 import re
 
 datasource = "{{ .DataSource }}"
+workflow_ttl = {{.WorkflowTTL}}
 
+# it's bug of the couler project, that needs "" on integer environment variable value to avoid the 
+# workflow failed: "invalid spec: cannot convert int64 to string"
+# issue: https://github.com/couler-proj/couler/issues/108
 step_envs = dict()
-{{range $k, $v := .StepEnvs}}step_envs["{{$k}}"] = '''{{$v}}'''
+{{range $k, $v := .StepEnvs}}step_envs["{{$k}}"] = '''"{{$v}}"'''
 {{end}}
 
 sqlflow_secret = None
@@ -184,8 +188,6 @@ resources = None
 if '''{{.Resources}}''' != "":
     resources=json.loads('''{{.Resources}}''')
 
-couler.clean_workflow_after_seconds_finished({{.WorkflowTTL}})
-
 step_log_file = "{{.StepLogFile}}"
 step_exit_time_wait = {{.StepExitTimeWait}}
 
@@ -194,7 +196,7 @@ step_exit_time_wait = {{.StepExitTimeWait}}
 
 codes = [
     "python <<EOF",
-    pyfunc.body(step_entry_{{.StepIndex}}),
+    pybody(step_entry_{{.StepIndex}}),
     "EOF",
 ]
 
@@ -216,6 +218,7 @@ if step_log_file:
     ]
 
 couler.run_script(image="{{.Image}}", command="bash", source="\n".join(codes), env=step_envs, resources=resources)
+couler.config_workflow(time_to_clean=workflow_ttl)
 {{end}}
 `
 

go/workflow/couler/codegen.go

@@ -17,6 +17,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"os"
 	"os/exec"
 	"reflect"
@@ -46,8 +47,6 @@ func fillEnvFromSession(envs *map[string]string, session *pb.Session) {
 	fillMapIfValueNotEmpty(*envs, "SQLFLOW_HADOOP_USER", session.HdfsUser)
 	fillMapIfValueNotEmpty(*envs, "SQLFLOW_HADOOP_PASS", session.HdfsUser)
 	fillMapIfValueNotEmpty(*envs, "SQLFLOW_submitter", session.Submitter)
-	fillMapIfValueNotEmpty(*envs, "SQLFLOW_WORKFLOW_STEP_LOG_FILE", os.Getenv("SQLFLOW_WORKFLOW_STEP_LOG_FILE"))
-	fillMapIfValueNotEmpty(*envs, "SQLFLOW_WORKFLOW_EXIT_TIME_WAIT", os.Getenv("SQLFLOW_WORKFLOW_EXIT_TIME_WAIT"))
 }
 
 // GetStepEnvs returns a map of envs used for couler workflow.
@@ -100,6 +99,14 @@ func GetSecret() (string, string, error) {
 	return name, string(value), nil
 }
 
+func escapeStepSQL(sql string) string {
+	sql = strings.Replace(sql, `\`, `\\\`, -1)
+	sql = strings.Replace(sql, `"`, `\\\"`, -1)
+	sql = strings.Replace(sql, "`", "\\`", -1)
+	sql = strings.Replace(sql, "$", "\\$", -1)
+	return sql
+}
+
 // GenFiller generates Filler to fill the template
 func GenFiller(programIR []ir.SQLFlowStmt, session *pb.Session) (*Filler, error) {
 	stepEnvs, err := GetStepEnvs(session)
@@ -123,13 +130,14 @@ func GenFiller(programIR []ir.SQLFlowStmt, session *pb.Session) (*Filler, error)
 	stepLogFile := os.Getenv("SQLFLOW_WORKFLOW_STEP_LOG_FILE")
 
 	r := &Filler{
-		DataSource:  session.DbConnStr,
-		StepEnvs:    stepEnvs,
-		WorkflowTTL: workflowTTL,
-		SecretName:  secretName,
-		SecretData:  secretData,
-		Resources:   os.Getenv(envResource),
-		StepLogFile: stepLogFile,
+		DataSource:      session.DbConnStr,
+		StepEnvs:        stepEnvs,
+		WorkflowTTL:     workflowTTL,
+		SecretName:      secretName,
+		SecretData:      secretData,
+		Resources:       os.Getenv(envResource),
+		StepLogFile:     stepLogFile,
+		ClusterConfigFn: os.Getenv("SQLFLOW_WORKFLOW_CLUSTER_CONFIG"),
 	}
 	// NOTE(yancey1989): does not use ModelImage here since the Predict statement
 	// does not contain the ModelImage field in SQL Program IR.
@@ -138,11 +146,12 @@ func GenFiller(programIR []ir.SQLFlowStmt, session *pb.Session) (*Filler, error)
 	}
 
 	for _, sqlIR := range programIR {
+		escapedSQL := escapeStepSQL(sqlIR.GetOriginalSQL())
 		switch i := sqlIR.(type) {
 		case *ir.NormalStmt, *ir.PredictStmt, *ir.ExplainStmt, *ir.EvaluateStmt:
 			// TODO(typhoonzero): get model image used when training.
 			sqlStmt := &sqlStatement{
-				OriginalSQL: sqlIR.GetOriginalSQL(), IsExtendedSQL: sqlIR.IsExtended(),
+				OriginalSQL: escapedSQL, IsExtendedSQL: sqlIR.IsExtended(),
 				DockerImage: defaultDockerImage}
 			r.SQLStatements = append(r.SQLStatements, sqlStmt)
 		case *ir.TrainStmt:
@@ -158,20 +167,20 @@ func GenFiller(programIR []ir.SQLFlowStmt, session *pb.Session) (*Filler, error)
 				r.SQLStatements = append(r.SQLStatements, sqlStmt)
 			} else {
 				sqlStmt := &sqlStatement{
-					OriginalSQL: sqlIR.GetOriginalSQL(), IsExtendedSQL: sqlIR.IsExtended(),
+					OriginalSQL: escapedSQL, IsExtendedSQL: sqlIR.IsExtended(),
 					DockerImage: stepImage}
 				r.SQLStatements = append(r.SQLStatements, sqlStmt)
 			}
 		case *ir.ShowTrainStmt, *ir.OptimizeStmt:
 			sqlStmt := &sqlStatement{
-				OriginalSQL:   sqlIR.GetOriginalSQL(),
+				OriginalSQL:   escapedSQL,
 				IsExtendedSQL: sqlIR.IsExtended(),
 				DockerImage:   defaultDockerImage,
 			}
 			r.SQLStatements = append(r.SQLStatements, sqlStmt)
 		case *ir.RunStmt:
 			sqlStmt := &sqlStatement{
-				OriginalSQL:   sqlIR.GetOriginalSQL(),
+				OriginalSQL:   escapedSQL,
 				IsExtendedSQL: sqlIR.IsExtended(),
 				DockerImage:   i.ImageName,
 				Select:        i.Select,
@@ -199,16 +208,15 @@ func GenCode(programIR []ir.SQLFlowStmt, session *pb.Session) (string, error) {
 
 // GenYAML translate the Couler program into Argo YAML
 func GenYAML(coulerProgram string) (string, error) {
-	cmdline := bytes.Buffer{}
-	fmt.Fprintf(&cmdline, "couler run --mode argo --workflow_name sqlflow ")
-	if c := os.Getenv("SQLFLOW_WORKFLOW_CLUSTER_CONFIG"); len(c) > 0 {
-		fmt.Fprintf(&cmdline, "--cluster_config %s ", c)
+	file, e := ioutil.TempFile("/tmp", "sqlflow.py")
+	if e != nil {
+		return "", e
 	}
-	fmt.Fprintf(&cmdline, "--file -")
-
-	coulerExec := strings.Split(cmdline.String(), " ")
-	// execute command: `cat sqlflow.couler | couler run --mode argo --workflow_name sqlflow --file -`
-	cmd := exec.Command(coulerExec[0], coulerExec[1:]...)
+	defer os.Remove(file.Name())
+	if e := ioutil.WriteFile(file.Name(), []byte(coulerProgram), 0644); e != nil {
+		return "", e
+	}
+	cmd := exec.Command("python", file.Name())
 	cmd.Env = append(os.Environ())
 	cmd.Stdin = strings.NewReader(coulerProgram)
 	out, err := cmd.CombinedOutput()

go/workflow/couler/codegen_test.go

@@ -14,6 +14,7 @@
 package couler
 
 import (
+	"fmt"
 	"io/ioutil"
 	"os"
 	"regexp"
@@ -31,11 +32,11 @@ class K8s(object):
     def __init__(self):
         pass
 
-    def with_pod(self, template):
+    def config_pod(self, template):
         self._with_tolerations(template)
         return template
 
-    def with_workflow_spec(self, spec):
+    def config_workflow(self, spec):
         spec["hostNetwork"] = True
         return spec
 
@@ -67,9 +68,12 @@ spec:
       container:
         image: docker/whalesay
         command:
-          - bash
-          - -c
           - 'echo "SQLFlow bridges AI and SQL engine."'
+        env:
+          - name: NVIDIA_VISIBLE_DEVICES
+            value: ""
+          - name: NVIDIA_DRIVER_CAPABILITIES
+            value: ""
       tolerations:
         - effect: NoSchedule
           key: key
@@ -81,6 +85,7 @@ spec:
 var testCoulerProgram = `
 import couler.argo as couler
 couler.run_container(image="docker/whalesay", command='echo "SQLFlow bridges AI and SQL engine."')
+couler.config_workflow(cluster_config_file="%s")
 `
 
 func TestCoulerCodegen(t *testing.T) {
@@ -92,20 +97,15 @@ func TestCoulerCodegen(t *testing.T) {
 	defer os.Unsetenv("SQLFLOW_OSS_AK")
 	code, err := GenCode(sqlIR, &pb.Session{})
 	a.NoError(err)
-
-	r, e := regexp.Compile(`steps.sqlflow\(sql=r'''(.*);''', `)
-	a.NoError(e)
-	a.Equal(r.FindStringSubmatch(code)[1], "SELECT * FROM iris.train limit 10")
-	a.True(strings.Contains(code, `step_envs["SQLFLOW_OSS_AK"] = '''oss_key'''`))
+	a.True(strings.Contains(code, `SELECT * FROM iris.train limit 10`))
+	a.True(strings.Contains(code, `step_envs["SQLFLOW_OSS_AK"] = '''"%s"''' % escape_env('''oss_key''')`))
 	a.False(strings.Contains(code, `step_envs["SQLFLOW_WORKFLOW_SECRET"]`))
-	a.True(strings.Contains(code, `couler.clean_workflow_after_seconds_finished(86400)`))
-	a.True(strings.Contains(code, `couler.secret(secret_data, name="sqlflow-secret", dry_run=True)`))
+	a.True(strings.Contains(code, `couler.create_secret(secret_data, name="sqlflow-secret", dry_run=True)`))
 	a.True(strings.Contains(code, `resources=json.loads('''{"memory": "32Mi", "cpu": "100m"}''')`))
 
-	_, e = GenYAML(code)
 	yaml, e := GenYAML(code)
 	a.NoError(e)
-	r, e = regexp.Compile(`step -e "(.*);"`)
+	r, e := regexp.Compile(`step -e "(.*);"`)
 	a.NoError(e)
 	a.Equal("SELECT * FROM iris.train limit 10", r.FindStringSubmatch(yaml)[1])
 	a.NoError(e)
@@ -116,7 +116,6 @@ func TestCoulerCodegen(t *testing.T) {
 	a.NoError(err)
 	r, e = regexp.Compile(`step_log_file = "(.*)"`)
 	a.True(strings.Contains(code, `step_log_file = "/home/admin/logs/step.log"`))
-	a.True(strings.Contains(code, "log_file=step_log_file"))
 
 	yaml, e = GenYAML(code)
 	a.NoError(e)
@@ -125,10 +124,6 @@ func TestCoulerCodegen(t *testing.T) {
 	a.Equal("/home/admin/logs", r.FindStringSubmatch(yaml)[1])
 	a.Equal("/home/admin/logs/step.log", r.FindStringSubmatch(yaml)[3])
 	a.NoError(e)
-
-	r, e = regexp.Compile("- name: SQLFLOW_WORKFLOW_STEP_LOG_FILE\n.*value: '(.*)'")
-	a.NoError(e)
-	a.Equal("/home/admin/logs/step.log", r.FindStringSubmatch(yaml)[1])
 }
 
 func TestCoulerCodegenSpecialChars(t *testing.T) {
@@ -158,8 +153,9 @@ func TestStringInStringSQL(t *testing.T) {
 	a.NoError(err)
 	yaml, e := GenYAML(code)
 	a.NoError(e)
-	expect := `validation.select=\\\"select * from iris.train where\
-            \ name like \\\\\\\"Versicolor\\\\\\\";\\\"`
+	expect := `validation.select=\\\"select * from iris.train where\`
+	a.True(strings.Contains(yaml, expect))
+	expect = `\ name like \\\\\\\"Versicolor\\\\\\\";\\\"`
 	a.True(strings.Contains(yaml, expect))
 }
 
@@ -177,11 +173,8 @@ func TestCompileCoulerProgram(t *testing.T) {
 	a.NoError(e)
 	defer os.Remove(cfFileName)
 
-	os.Setenv("SQLFLOW_WORKFLOW_CLUSTER_CONFIG", cfFileName)
-	defer os.Unsetenv("SQLFLOW_WORKFLOW_CLUSTER_CONFIG")
-	out, e := GenYAML(testCoulerProgram)
+	out, e := GenYAML(fmt.Sprintf(testCoulerProgram, cfFileName))
 	a.NoError(e)
-
 	a.Equal(expectedArgoYAML, out)
 }