package main
import (
+ "fmt"
"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
"git.curoverse.com/arvados.git/sdk/go/keepclient"
"log"
"os"
"os/exec"
"os/signal"
+ "strings"
"syscall"
)
type TaskDef struct {
- commands []string `json:"commands"`
- env map[string]string `json:"env"`
- stdin string `json:"stdin"`
- stdout string `json:"stdout"`
- vwd map[string]string `json:"vwd"`
- successCodes []int `json:"successCodes"`
- permanentFailCodes []int `json:"permanentFailCodes"`
- temporaryFailCodes []int `json:"temporaryFailCodes"`
+ command []string `json:"command"`
+ env map[string]string `json:"task.env"`
+ stdin string `json:"task.stdin"`
+ stdout string `json:"task.stdout"`
+ vwd map[string]string `json:"task.vwd"`
+ successCodes []int `json:"task.successCodes"`
+ permanentFailCodes []int `json:"task.permanentFailCodes"`
+ temporaryFailCodes []int `json:"task.temporaryFailCodes"`
}
type Tasks struct {
progress float32 `json:"sequence"`
}
-func setupDirectories(tmpdir, taskUuid string) (outdir string, err error) {
- err = os.Chdir(tmpdir)
- if err != nil {
- return "", err
- }
+type IArvadosClient interface {
+ Create(resourceType string, parameters arvadosclient.Dict, output interface{}) error
+ Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error)
+}
- err = os.Mkdir("tmpdir", 0700)
+func setupDirectories(crunchtmpdir, taskUuid string) (tmpdir, outdir string, err error) {
+ tmpdir = crunchtmpdir + "/tmpdir"
+ err = os.Mkdir(tmpdir, 0700)
if err != nil {
- return "", err
+ return "", "", err
}
- err = os.Mkdir(taskUuid, 0700)
+ outdir = crunchtmpdir + "/outdir"
+ err = os.Mkdir(outdir, 0700)
if err != nil {
- return "", err
+ return "", "", err
}
- os.Chdir(taskUuid)
- if err != nil {
- return "", err
- }
+ return tmpdir, outdir, nil
+}
- outdir, err = os.Getwd()
- if err != nil {
- return "", err
+func checkOutputFilename(outdir, fn string) error {
+ if strings.HasPrefix(fn, "/") || strings.HasSuffix(fn, "/") {
+ return fmt.Errorf("Path must not start or end with '/'")
+ }
+ if strings.Index("../", fn) != -1 {
+ return fmt.Errorf("Path must not contain '../'")
}
- return outdir, nil
+ sl := strings.LastIndex(fn, "/")
+ if sl != -1 {
+ os.MkdirAll(outdir+"/"+fn[0:sl], 0777)
+ }
+ return nil
}
-func setupCommand(cmd *exec.Cmd, taskp TaskDef, keepmount, outdir string) error {
- var err error
-
+func setupCommand(cmd *exec.Cmd, taskp TaskDef, outdir string, replacements map[string]string) (stdin, stdout string, err error) {
if taskp.vwd != nil {
for k, v := range taskp.vwd {
- os.Symlink(keepmount+"/"+v, outdir+"/"+k)
+ v = substitute(v, replacements)
+ err = checkOutputFilename(outdir, k)
+ if err != nil {
+ return "", "", err
+ }
+ os.Symlink(v, outdir+"/"+k)
}
}
if taskp.stdin != "" {
// Set up stdin redirection
- cmd.Stdin, err = os.Open(keepmount + "/" + taskp.stdin)
+ stdin = substitute(taskp.stdin, replacements)
+ cmd.Stdin, err = os.Open(stdin)
if err != nil {
- return err
+ return "", "", err
}
}
if taskp.stdout != "" {
+ err = checkOutputFilename(outdir, taskp.stdout)
+ if err != nil {
+ return "", "", err
+ }
// Set up stdout redirection
- cmd.Stdout, err = os.Create(outdir + "/" + taskp.stdout)
+ stdout = outdir + "/" + taskp.stdout
+ cmd.Stdout, err = os.Create(stdout)
if err != nil {
- return err
+ return "", "", err
}
} else {
cmd.Stdout = os.Stdout
// Set up subprocess environment
cmd.Env = os.Environ()
for k, v := range taskp.env {
+ v = substitute(v, replacements)
cmd.Env = append(cmd.Env, k+"="+v)
}
}
- return nil
+ return stdin, stdout, nil
}
-func setupSignals(cmd *exec.Cmd) {
+func setupSignals(cmd *exec.Cmd) chan os.Signal {
// Set up signal handlers
// Forward SIGINT, SIGTERM and SIGQUIT to inner process
sigChan := make(chan os.Signal, 1)
go func(sig <-chan os.Signal) {
catch := <-sig
- if cmd.Process != nil {
- cmd.Process.Signal(catch)
- }
+ cmd.Process.Signal(catch)
}(sigChan)
signal.Notify(sigChan, syscall.SIGTERM)
signal.Notify(sigChan, syscall.SIGINT)
signal.Notify(sigChan, syscall.SIGQUIT)
+ return sigChan
}
func inCodes(code int, codes []int) bool {
const TASK_TEMPFAIL = 111
-type TempFail struct{ InnerError error }
+type TempFail struct{ error }
type PermFail struct{}
-func (s TempFail) Error() string {
- return s.InnerError.Error()
-}
-
func (s PermFail) Error() string {
return "PermFail"
}
-func runner(api arvadosclient.IArvadosClient,
+func substitute(inp string, subst map[string]string) string {
+ for k, v := range subst {
+ inp = strings.Replace(inp, k, v, -1)
+ }
+ return inp
+}
+
+func runner(api IArvadosClient,
kc IKeepClient,
- jobUuid, taskUuid, tmpdir, keepmount string,
+ jobUuid, taskUuid, crunchtmpdir, keepmount string,
jobStruct Job, taskStruct Task) error {
var err error
}
}
- // Set up subprocess
- cmd := exec.Command(taskp.commands[0], taskp.commands[1:]...)
-
- var outdir string
- outdir, err = setupDirectories(tmpdir, taskUuid)
+ var tmpdir, outdir string
+ tmpdir, outdir, err = setupDirectories(crunchtmpdir, taskUuid)
if err != nil {
return TempFail{err}
}
+ replacements := map[string]string{
+ "$(task.tmpdir)": tmpdir,
+ "$(task.outdir)": outdir,
+ "$(task.keep)": keepmount}
+
+ // Set up subprocess
+ for k, v := range taskp.command {
+ taskp.command[k] = substitute(v, replacements)
+ }
+
+ cmd := exec.Command(taskp.command[0], taskp.command[1:]...)
+
cmd.Dir = outdir
- err = setupCommand(cmd, taskp, keepmount, outdir)
+ var stdin, stdout string
+ stdin, stdout, err = setupCommand(cmd, taskp, outdir, replacements)
if err != nil {
return err
}
- setupSignals(cmd)
-
// Run subprocess and wait for it to complete
- log.Printf("Running %v", cmd.Args)
+ if stdin != "" {
+ stdin = " < " + stdin
+ }
+ if stdout != "" {
+ stdout = " > " + stdout
+ }
+ log.Printf("Running %v%v%v", cmd.Args, stdin, stdout)
- err = cmd.Run()
+ err = cmd.Start()
+
+ signals := setupSignals(cmd)
+ err = cmd.Wait()
+ signal.Stop(signals)
if err != nil {
// Run() returns ExitError on non-zero exit code, but we handle
}
}
- const success = 1
- const permfail = 2
- const tempfail = 2
- var status int
+ var success bool
exitCode := cmd.ProcessState.Sys().(syscall.WaitStatus).ExitStatus()
log.Printf("Completed with exit code %v", exitCode)
- if inCodes(exitCode, taskp.successCodes) {
- status = success
- } else if inCodes(exitCode, taskp.permanentFailCodes) {
- status = permfail
+ if inCodes(exitCode, taskp.permanentFailCodes) {
+ success = false
} else if inCodes(exitCode, taskp.temporaryFailCodes) {
- return TempFail{nil}
- } else if cmd.ProcessState.Success() {
- status = success
+ return TempFail{fmt.Errorf("Process tempfail with exit code %v", exitCode)}
+ } else if inCodes(exitCode, taskp.successCodes) || cmd.ProcessState.Success() {
+ success = true
} else {
- status = permfail
+ success = false
}
// Upload output directory
map[string]interface{}{
"job_task": Task{
output: manifest,
- success: status == success,
+ success: success,
progress: 1}},
nil)
if err != nil {
return TempFail{err}
}
- if status == success {
+ if success {
return nil
} else {
return PermFail{}
}
func main() {
- syscall.Umask(0077)
-
api, err := arvadosclient.MakeArvadosClient()
if err != nil {
log.Fatal(err)
import (
"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
. "gopkg.in/check.v1"
+ "io"
"io/ioutil"
+ "log"
"os"
+ "syscall"
"testing"
+ "time"
)
// Gocheck boilerplate
return nil
}
-func (t ArvTestClient) Delete(resource string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error) {
- return nil
-}
-
func (t ArvTestClient) Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error) {
t.c.Check(resourceType, Equals, "job_tasks")
t.c.Check(parameters, DeepEquals, arvadosclient.Dict{"job_task": Task{
return nil
}
-func (t ArvTestClient) Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error) {
- return nil
-}
-
-func (t ArvTestClient) List(resource string, parameters arvadosclient.Dict, output interface{}) (err error) {
- return nil
-}
-
func (s *TestSuite) TestSimpleRun(c *C) {
tmpdir, _ := ioutil.TempDir("", "")
defer func() {
tmpdir,
"",
Job{script_parameters: Tasks{[]TaskDef{TaskDef{
- commands: []string{"echo", "foo"}}}}},
+ command: []string{"echo", "foo"}}}}},
Task{sequence: 0})
c.Check(err, IsNil)
-
}
func checkOutput(c *C, tmpdir string) {
- file, err := os.Open(tmpdir + "/zzzz-ot0gb-111111111111111/output.txt")
+ file, err := os.Open(tmpdir + "/outdir/output.txt")
c.Assert(err, IsNil)
data := make([]byte, 100)
var count int
- count, err = file.Read(data)
- c.Assert(err, IsNil)
- c.Check(string(data[0:count]), Equals, "foo\n")
+ err = nil
+ offset := 0
+ for err == nil {
+ count, err = file.Read(data[offset:])
+ offset += count
+ }
+ c.Assert(err, Equals, io.EOF)
+ c.Check(string(data[0:offset]), Equals, "foo\n")
}
func (s *TestSuite) TestSimpleRunSubtask(c *C) {
tmpdir,
"",
Job{script_parameters: Tasks{[]TaskDef{
- TaskDef{commands: []string{"echo", "bar"}},
- TaskDef{commands: []string{"echo", "foo"}}}}},
+ TaskDef{command: []string{"echo", "bar"}},
+ TaskDef{command: []string{"echo", "foo"}}}}},
Task{parameters: TaskDef{
- commands: []string{"echo", "foo"},
- stdout: "output.txt"},
+ command: []string{"echo", "foo"},
+ stdout: "output.txt"},
sequence: 1})
c.Check(err, IsNil)
tmpdir,
"",
Job{script_parameters: Tasks{[]TaskDef{TaskDef{
- commands: []string{"cat"},
- stdout: "output.txt",
- stdin: tmpfile.Name()}}}},
+ command: []string{"cat"},
+ stdout: "output.txt",
+ stdin: tmpfile.Name()}}}},
Task{sequence: 0})
c.Check(err, IsNil)
tmpdir,
"",
Job{script_parameters: Tasks{[]TaskDef{TaskDef{
- commands: []string{"/bin/sh", "-c", "echo $BAR"},
- stdout: "output.txt",
- env: map[string]string{"BAR": "foo"}}}}},
+ command: []string{"/bin/sh", "-c", "echo $BAR"},
+ stdout: "output.txt",
+ env: map[string]string{"BAR": "foo"}}}}},
Task{sequence: 0})
c.Check(err, IsNil)
+ checkOutput(c, tmpdir)
+}
+func (s *TestSuite) TestEnvSubstitute(c *C) {
+ tmpdir, _ := ioutil.TempDir("", "")
+ defer func() {
+ os.RemoveAll(tmpdir)
+ }()
+
+ err := runner(ArvTestClient{c, ". d3b07384d113edec49eaa6238ad5ff00+4 0:4:output.txt\n", true},
+ KeepTestClient{},
+ "zzzz-8i9sb-111111111111111",
+ "zzzz-ot0gb-111111111111111",
+ tmpdir,
+ "foo\n",
+ Job{script_parameters: Tasks{[]TaskDef{TaskDef{
+ command: []string{"/bin/sh", "-c", "echo $BAR"},
+ stdout: "output.txt",
+ env: map[string]string{"BAR": "$(task.keep)"}}}}},
+ Task{sequence: 0})
+ c.Check(err, IsNil)
+ checkOutput(c, tmpdir)
+}
+
+func (s *TestSuite) TestEnvReplace(c *C) {
+ tmpdir, _ := ioutil.TempDir("", "")
+ defer func() {
+ os.RemoveAll(tmpdir)
+ }()
+
+ err := runner(ArvTestClient{c, ". d3b07384d113edec49eaa6238ad5ff00+4 0:4:output.txt\n", true},
+ KeepTestClient{},
+ "zzzz-8i9sb-111111111111111",
+ "zzzz-ot0gb-111111111111111",
+ tmpdir,
+ "",
+ Job{script_parameters: Tasks{[]TaskDef{TaskDef{
+ command: []string{"/bin/sh", "-c", "echo $PATH"},
+ stdout: "output.txt",
+ env: map[string]string{"PATH": "foo"}}}}},
+ Task{sequence: 0})
+ c.Check(err, IsNil)
checkOutput(c, tmpdir)
}
return nil
}
-func (t SubtaskTestClient) Delete(resource string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error) {
- return nil
-}
-
func (t SubtaskTestClient) Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error) {
return nil
}
-func (t SubtaskTestClient) Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error) {
- return nil
-}
-
-func (t SubtaskTestClient) List(resource string, parameters arvadosclient.Dict, output interface{}) (err error) {
- return nil
-}
-
func (s *TestSuite) TestScheduleSubtask(c *C) {
api := SubtaskTestClient{c, []Task{
created_by_job_task_uuid: "zzzz-ot0gb-111111111111111",
sequence: 1,
parameters: TaskDef{
- commands: []string{"echo", "bar"}}},
+ command: []string{"echo", "bar"}}},
Task{job_uuid: "zzzz-8i9sb-111111111111111",
created_by_job_task_uuid: "zzzz-ot0gb-111111111111111",
sequence: 1,
parameters: TaskDef{
- commands: []string{"echo", "foo"}}}},
+ command: []string{"echo", "foo"}}}},
0}
tmpdir, _ := ioutil.TempDir("", "")
tmpdir,
"",
Job{script_parameters: Tasks{[]TaskDef{
- TaskDef{commands: []string{"echo", "bar"}},
- TaskDef{commands: []string{"echo", "foo"}}}}},
+ TaskDef{command: []string{"echo", "bar"}},
+ TaskDef{command: []string{"echo", "foo"}}}}},
Task{sequence: 0})
c.Check(err, IsNil)
tmpdir,
"",
Job{script_parameters: Tasks{[]TaskDef{TaskDef{
- commands: []string{"/bin/sh", "-c", "exit 1"}}}}},
+ command: []string{"/bin/sh", "-c", "exit 1"}}}}},
Task{sequence: 0})
c.Check(err, FitsTypeOf, PermFail{})
}
tmpdir,
"",
Job{script_parameters: Tasks{[]TaskDef{TaskDef{
- commands: []string{"/bin/sh", "-c", "exit 1"},
+ command: []string{"/bin/sh", "-c", "exit 1"},
successCodes: []int{0, 1}}}}},
Task{sequence: 0})
c.Check(err, IsNil)
tmpdir,
"",
Job{script_parameters: Tasks{[]TaskDef{TaskDef{
- commands: []string{"/bin/sh", "-c", "exit 0"},
+ command: []string{"/bin/sh", "-c", "exit 0"},
permanentFailCodes: []int{0, 1}}}}},
Task{sequence: 0})
c.Check(err, FitsTypeOf, PermFail{})
tmpdir,
"",
Job{script_parameters: Tasks{[]TaskDef{TaskDef{
- commands: []string{"/bin/sh", "-c", "exit 1"},
+ command: []string{"/bin/sh", "-c", "exit 1"},
temporaryFailCodes: []int{1}}}}},
Task{sequence: 0})
c.Check(err, FitsTypeOf, TempFail{})
tmpdir,
"",
Job{script_parameters: Tasks{[]TaskDef{TaskDef{
- commands: []string{"ls", "output.txt"},
+ command: []string{"ls", "output.txt"},
vwd: map[string]string{
"output.txt": tmpfile.Name()}}}}},
Task{sequence: 0})
c.Check(err, IsNil)
checkOutput(c, tmpdir)
}
+
+func (s *TestSuite) TestSubstitutionStdin(c *C) {
+ keepmount, _ := ioutil.TempDir("", "")
+ ioutil.WriteFile(keepmount+"/"+"file1.txt", []byte("foo\n"), 0600)
+ defer func() {
+ os.RemoveAll(keepmount)
+ }()
+
+ log.Print("Keepmount is ", keepmount)
+
+ tmpdir, _ := ioutil.TempDir("", "")
+ defer func() {
+ os.RemoveAll(tmpdir)
+ }()
+
+ log.Print("tmpdir is ", tmpdir)
+
+ err := runner(ArvTestClient{c,
+ ". d3b07384d113edec49eaa6238ad5ff00+4 0:4:output.txt\n", true},
+ KeepTestClient{},
+ "zzzz-8i9sb-111111111111111",
+ "zzzz-ot0gb-111111111111111",
+ tmpdir,
+ keepmount,
+ Job{script_parameters: Tasks{[]TaskDef{TaskDef{
+ command: []string{"cat"},
+ stdout: "output.txt",
+ stdin: "$(task.keep)/file1.txt"}}}},
+ Task{sequence: 0})
+ c.Check(err, IsNil)
+ checkOutput(c, tmpdir)
+}
+
+func (s *TestSuite) TestSubstitutionCommandLine(c *C) {
+ keepmount, _ := ioutil.TempDir("", "")
+ ioutil.WriteFile(keepmount+"/"+"file1.txt", []byte("foo\n"), 0600)
+ defer func() {
+ os.RemoveAll(keepmount)
+ }()
+
+ tmpdir, _ := ioutil.TempDir("", "")
+ defer func() {
+ os.RemoveAll(tmpdir)
+ }()
+
+ err := runner(ArvTestClient{c,
+ ". d3b07384d113edec49eaa6238ad5ff00+4 0:4:output.txt\n", true},
+ KeepTestClient{},
+ "zzzz-8i9sb-111111111111111",
+ "zzzz-ot0gb-111111111111111",
+ tmpdir,
+ keepmount,
+ Job{script_parameters: Tasks{[]TaskDef{TaskDef{
+ command: []string{"cat", "$(task.keep)/file1.txt"},
+ stdout: "output.txt"}}}},
+ Task{sequence: 0})
+ c.Check(err, IsNil)
+
+ checkOutput(c, tmpdir)
+}
+
+func (s *TestSuite) TestSignal(c *C) {
+ tmpdir, _ := ioutil.TempDir("", "")
+ defer func() {
+ os.RemoveAll(tmpdir)
+ }()
+
+ go func() {
+ time.Sleep(1 * time.Second)
+ self, _ := os.FindProcess(os.Getpid())
+ self.Signal(syscall.SIGINT)
+ }()
+
+ err := runner(ArvTestClient{c,
+ "", false},
+ KeepTestClient{},
+ "zzzz-8i9sb-111111111111111",
+ "zzzz-ot0gb-111111111111111",
+ tmpdir,
+ "",
+ Job{script_parameters: Tasks{[]TaskDef{TaskDef{
+ command: []string{"sleep", "4"}}}}},
+ Task{sequence: 0})
+ c.Check(err, FitsTypeOf, PermFail{})
+
+}
+
+func (s *TestSuite) TestQuoting(c *C) {
+ tmpdir, _ := ioutil.TempDir("", "")
+ defer func() {
+ os.RemoveAll(tmpdir)
+ }()
+
+ err := runner(ArvTestClient{c,
+ "./s\\040ub:dir d3b07384d113edec49eaa6238ad5ff00+4 0:4::e\\040vil\n", true},
+ KeepTestClient{},
+ "zzzz-8i9sb-111111111111111",
+ "zzzz-ot0gb-111111111111111",
+ tmpdir,
+ "",
+ Job{script_parameters: Tasks{[]TaskDef{TaskDef{
+ command: []string{"echo", "foo"},
+ stdout: "s ub:dir/:e vi\nl"}}}},
+ Task{sequence: 0})
+ c.Check(err, IsNil)
+}
projects / arvados.git / commitdiff