Merge pull request #540 from wgerlach/master

various AWE updates

Author: wgerlach

Dockerfile

@@ -2,6 +2,12 @@
 # docker build --force-rm --no-cache --rm -t mgrast/awe .
 # docker tag mgrast/awe mgrast/awe:${TAG}
 
+# skycore push mgrast/awe:${TAG}
+# docker create --name awe-temporary mgrast/awe:${TAG}
+# docker cp awe-temporary:/go/bin/awe-worker .
+# docker cp awe-temporary:/go/bin/awe-submitter .
+# docker rm awe-temporary
+
 
 FROM golang:1.7.6-alpine
 

awe-submitter/awe-submitter.go

@@ -7,9 +7,15 @@ import (
 	"github.com/MG-RAST/AWE/lib/core/cwl"
 	cwl_types "github.com/MG-RAST/AWE/lib/core/cwl/types"
 	"github.com/MG-RAST/AWE/lib/logger"
+	//"github.com/MG-RAST/AWE/lib/logger/event"
+	"bytes"
 	"github.com/MG-RAST/AWE/lib/shock"
 	"github.com/davecgh/go-spew/spew"
 	"gopkg.in/yaml.v2"
+	"io"
+	"io/ioutil"
+	"mime/multipart"
+	"net/http"
 	"net/url"
 	"os"
 	"path"
@@ -54,6 +60,8 @@ func uploadFile(file *cwl_types.File, inputfile_path string) (err error) {
 
 	file_path := file.Path
 
+	basename := path.Base(file_path)
+
 	if file_path == "" {
 		return
 	}
@@ -75,35 +83,39 @@ func uploadFile(file *cwl_types.File, inputfile_path string) (err error) {
 	}
 	spew.Dump(node)
 
-	file.Location_url, err = url.Parse(conf.SERVER_URL + "/node/" + node.Id + "?download")
+	file.Location_url, err = url.Parse(conf.SHOCK_URL + "/node/" + node.Id + "?download")
 	if err != nil {
 		return
 	}
 
 	file.Location = file.Location_url.String()
 	file.Path = ""
+	file.Basename = basename
 
 	return
 }
 
-func processInputData(native interface{}, inputfile_path string) (err error) {
+func processInputData(native interface{}, inputfile_path string) (count int, err error) {
+
 	fmt.Printf("(processInputData) start\n")
 	defer fmt.Printf("(processInputData) end\n")
 	switch native.(type) {
 	case *cwl.Job_document:
-
+		fmt.Printf("found Job_document\n")
 		job_doc_ptr := native.(*cwl.Job_document)
 
 		job_doc := *job_doc_ptr
 
-		for key, value := range job_doc {
+		for _, value := range job_doc {
 
-			fmt.Printf("recurse into key: %s\n", key)
-			err = processInputData(value, inputfile_path)
+			id := value.GetId()
+			fmt.Printf("recurse into key: %s\n", id)
+			var sub_count int
+			sub_count, err = processInputData(value, inputfile_path)
 			if err != nil {
 				return
 			}
-
+			count += sub_count
 		}
 
 		return
@@ -123,7 +135,7 @@ func processInputData(native interface{}, inputfile_path string) (err error) {
 		if err != nil {
 			return
 		}
-
+		count += 1
 		return
 	default:
 		spew.Dump(native)
@@ -135,10 +147,19 @@ func processInputData(native interface{}, inputfile_path string) (err error) {
 }
 
 func main() {
+	err := main_wrapper()
+	if err != nil {
+		println(err.Error())
+		os.Exit(1)
+	}
+	os.Exit(0)
+}
+
+func main_wrapper() (err error) {
 
 	conf.LOG_OUTPUT = "console"
 
-	err := conf.Init_conf("submitter")
+	err = conf.Init_conf("submitter")
 
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "ERROR: error reading conf file: "+err.Error())
@@ -147,23 +168,24 @@ func main() {
 
 	logger.Initialize("client")
 
-	if conf.CWL_JOB == "" {
-		logger.Error("cwl job file missing")
-		time.Sleep(time.Second)
-		os.Exit(1)
+	for _, value := range conf.ARGS {
+		println(value)
 	}
 
-	inputfile_path := path.Dir(conf.CWL_JOB)
+	job_file := conf.ARGS[0]
+	workflow_file := conf.ARGS[1]
+
+	inputfile_path := path.Dir(job_file)
 	fmt.Printf("job path: %s\n", inputfile_path) // needed to resolve relative paths
 
-	job_doc, err := cwl.ParseJob(conf.CWL_JOB)
+	job_doc, err := cwl.ParseJobFile(job_file)
 	if err != nil {
 		logger.Error("error parsing cwl job: %v", err)
 		time.Sleep(time.Second)
 		os.Exit(1)
 	}
 
-	fmt.Println("Job input fter reading from file:")
+	fmt.Println("Job input after reading from file:")
 	spew.Dump(*job_doc)
 
 	data, err := yaml.Marshal(*job_doc)
@@ -172,15 +194,23 @@ func main() {
 		os.Exit(1)
 	}
 
-	fmt.Printf("yaml:\n%s\n", string(data[:]))
+	job_doc_string := string(data[:])
+	fmt.Printf("job_doc_string: \"%s\"\n", job_doc_string)
+	if job_doc_string == "" {
+		fmt.Println("job_doc_string is empty")
+		os.Exit(1)
+	}
+
+	fmt.Printf("yaml:\n%s\n", job_doc_string)
 
 	// process input files
 
-	err = processInputData(job_doc, inputfile_path)
+	upload_count, err := processInputData(job_doc, inputfile_path)
 	if err != nil {
 		fmt.Printf("error: %s", err.Error())
 		os.Exit(1)
 	}
+	fmt.Printf("%d files have been uploaded\n", upload_count)
 	time.Sleep(2)
 
 	fmt.Println("------------Job input after parsing:")
@@ -192,4 +222,111 @@ func main() {
 
 	fmt.Printf("yaml:\n%s\n", string(data[:]))
 
+	// job submission example:
+	// curl -X POST -F [email protected] -F cwl=@/Users/wolfganggerlach/awe_data/pipeline/CWL/PackedWorkflow/preprocess-fasta.workflow.cwl http://localhost:8001/job
+
+	//var b bytes.Buffer
+	//w := multipart.NewWriter(&b)
+	err = SubmitCWLJobToAWE(workflow_file, job_file, &data)
+	if err != nil {
+		return
+	}
+
+	return
+}
+
+func SubmitCWLJobToAWE(workflow_file string, job_file string, data *[]byte) (err error) {
+	multipart := NewMultipartWriter()
+	err = multipart.AddFile("cwl", workflow_file)
+	if err != nil {
+		return
+	}
+	err = multipart.AddDataAsFile("job", job_file, data)
+	if err != nil {
+		return
+	}
+	response, err := multipart.Send("POST", conf.SERVER_URL+"/job")
+	if err != nil {
+		return
+	}
+	responseData, err := ioutil.ReadAll(response.Body)
+	if err != nil {
+		return
+	}
+	responseString := string(responseData)
+
+	fmt.Println(responseString)
+	return
+
+}
+
+type MultipartWriter struct {
+	b bytes.Buffer
+	w *multipart.Writer
+}
+
+func NewMultipartWriter() *MultipartWriter {
+	m := &MultipartWriter{}
+	m.w = multipart.NewWriter(&m.b)
+	return m
+}
+
+func (m *MultipartWriter) Send(method string, url string) (response *http.Response, err error) {
+	m.w.Close()
+	fmt.Println("------------")
+	spew.Dump(m.w)
+	fmt.Println("------------")
+
+	req, err := http.NewRequest(method, url, &m.b)
+	if err != nil {
+		return
+	}
+	// Don't forget to set the content type, this will contain the boundary.
+	req.Header.Set("Content-Type", m.w.FormDataContentType())
+
+	// Submit the request
+	client := &http.Client{}
+	fmt.Printf("%s %s\n\n", method, url)
+	response, err = client.Do(req)
+	if err != nil {
+		return
+	}
+
+	// Check the response
+	//if response.StatusCode != http.StatusOK {
+	//	err = fmt.Errorf("bad status: %s", response.Status)
+	//}
+	return
+
+}
+
+func (m *MultipartWriter) AddDataAsFile(fieldname string, filepath string, data *[]byte) (err error) {
+
+	fw, err := m.w.CreateFormFile(fieldname, filepath)
+	if err != nil {
+		return
+	}
+	_, err = fw.Write(*data)
+	if err != nil {
+		return
+	}
+	return
+}
+
+func (m *MultipartWriter) AddFile(fieldname string, filepath string) (err error) {
+
+	f, err := os.Open(filepath)
+	if err != nil {
+		return
+	}
+	defer f.Close()
+	fw, err := m.w.CreateFormFile(fieldname, filepath)
+	if err != nil {
+		return
+	}
+	if _, err = io.Copy(fw, f); err != nil {
+		return
+	}
+
+	return
 }

awe-worker/awe-worker.go

@@ -78,7 +78,9 @@ func main() {
 		os.Exit(1)
 	}
 
-	var self *core.Client
+	core.SetClientProfile(profile)
+	self := core.Self
+	//var self *core.Client
 	if worker.Client_mode == "online" {
 		if conf.SERVER_URL == "" {
 			fmt.Fprintf(os.Stderr, "AWE server url not configured or is empty. Please check the [Client]serverurl field in the configuration file.\n")
@@ -89,20 +91,17 @@ func main() {
 			os.Exit(1)
 		}
 
-		self, err = worker.RegisterWithAuth(conf.SERVER_URL, profile)
+		err = worker.RegisterWithAuth(conf.SERVER_URL, profile)
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "fail to register: %s\n", err.Error())
 			logger.Error("fail to register: %s\n", err.Error())
 			os.Exit(1)
 		}
 
-	} else {
-		self = core.NewClient()
 	}
 
-	core.SetClientProfile(self)
 	if worker.Client_mode == "online" {
-		fmt.Printf("Client registered, name=%s, id=%s\n", self.Name, self.Id)
+		fmt.Printf("Client registered, name=%s, id=%s\n", self.WorkerRuntime.Name, self.Id)
 		logger.Event(event.CLIENT_REGISTRATION, "clientid="+self.Id)
 	}
 
@@ -114,7 +113,7 @@ func main() {
 			time.Sleep(time.Second)
 			os.Exit(1)
 		}
-		job_doc, err := cwl.ParseJob(conf.CWL_JOB)
+		job_doc, err := cwl.ParseJobFile(conf.CWL_JOB)
 		if err != nil {
 			logger.Error("error parsing cwl job: %v", err)
 			time.Sleep(time.Second)
@@ -126,13 +125,13 @@ func main() {
 
 		os.Getwd() //https://golang.org/pkg/os/#Getwd
 
-		workunit := &core.Workunit{Id: "00000000-0000-0000-0000-000000000000_0_0", CWL: core.NewCWL_workunit()}
+		workunit := &core.Workunit{Id: "00000000-0000-0000-0000-000000000000_0_0", CWL_workunit: core.NewCWL_workunit()}
 
-		workunit.CWL.Job_input = job_doc
-		workunit.CWL.Job_input_filename = conf.CWL_JOB
+		workunit.CWL_workunit.Job_input = job_doc
+		workunit.CWL_workunit.Job_input_filename = conf.CWL_JOB
 
-		workunit.CWL.CWL_tool_filename = conf.CWL_TOOL
-		workunit.CWL.CWL_tool = &cwl.CommandLineTool{} // TODO parsing and testing ?
+		workunit.CWL_workunit.CWL_tool_filename = conf.CWL_TOOL
+		workunit.CWL_workunit.CWL_tool = &cwl.CommandLineTool{} // TODO parsing and testing ?
 
 		current_working_directory, err := os.Getwd()
 		if err != nil {
@@ -142,13 +141,15 @@ func main() {
 		}
 		workunit.WorkPath = current_working_directory
 
-		workunit.Cmd = &core.Command{}
-		workunit.Cmd.Local = true // this makes sure the working directory is not deleted
-		workunit.Cmd.Name = "/usr/bin/cwl-runner"
+		cmd := &core.Command{}
+		cmd.Local = true // this makes sure the working directory is not deleted
+		cmd.Name = "/usr/bin/cwl-runner"
 
-		workunit.Cmd.ArgsArray = []string{"--leave-outputs", "--leave-tmpdir", "--tmp-outdir-prefix", "./tmp/", "--tmpdir-prefix", "./tmp/", "--disable-pull", "--rm-container", "--on-error", "stop", workunit.CWL.CWL_tool_filename, workunit.CWL.Job_input_filename}
+		cmd.ArgsArray = []string{"--leave-outputs", "--leave-tmpdir", "--tmp-outdir-prefix", "./tmp/", "--tmpdir-prefix", "./tmp/", "--disable-pull", "--rm-container", "--on-error", "stop", workunit.CWL_workunit.CWL_tool_filename, workunit.CWL_workunit.Job_input_filename}
 
-		workunit.WorkPerf = core.NewWorkPerf(workunit.Id)
+		workunit.Cmd = cmd
+
+		workunit.WorkPerf = core.NewWorkPerf()
 		workunit.WorkPerf.Checkout = time.Now().Unix()
 
 		logger.Debug(1, "injecting cwl job into worker...")