package main
import (
+ "encoding/json"
"fmt"
+ "git.curoverse.com/arvados.git/sdk/go/arvados"
"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
"git.curoverse.com/arvados.git/sdk/go/keepclient"
+ "io"
+ "io/ioutil"
"log"
"os"
"os/exec"
Env map[string]string `json:"task.env"`
Stdin string `json:"task.stdin"`
Stdout string `json:"task.stdout"`
+ Stderr string `json:"task.stderr"`
Vwd map[string]string `json:"task.vwd"`
SuccessCodes []int `json:"task.successCodes"`
PermanentFailCodes []int `json:"task.permanentFailCodes"`
TemporaryFailCodes []int `json:"task.temporaryFailCodes"`
+ KeepTmpOutput bool `json:"task.keepTmpOutput"`
}
type Tasks struct {
Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error)
}
-func setupDirectories(crunchtmpdir, taskUuid string) (tmpdir, outdir string, err error) {
+func setupDirectories(crunchtmpdir, taskUuid string, keepTmp bool) (tmpdir, outdir string, err error) {
tmpdir = crunchtmpdir + "/tmpdir"
err = os.Mkdir(tmpdir, 0700)
if err != nil {
return "", "", err
}
- outdir = crunchtmpdir + "/outdir"
- err = os.Mkdir(outdir, 0700)
- if err != nil {
- return "", "", err
+ if keepTmp {
+ outdir = os.Getenv("TASK_KEEPMOUNT_TMP")
+ } else {
+ outdir = crunchtmpdir + "/outdir"
+ err = os.Mkdir(outdir, 0700)
+ if err != nil {
+ return "", "", err
+ }
}
return tmpdir, outdir, nil
return nil
}
-func setupCommand(cmd *exec.Cmd, taskp TaskDef, outdir string, replacements map[string]string) (stdin, stdout string, err error) {
+func copyFile(dst, src string) error {
+ in, err := os.Open(src)
+ if err != nil {
+ return err
+ }
+ defer in.Close()
+
+ out, err := os.Create(dst)
+ if err != nil {
+ return err
+ }
+ defer out.Close()
+
+ _, err = io.Copy(out, in)
+ return err
+}
+
+func setupCommand(cmd *exec.Cmd, taskp TaskDef, outdir string, replacements map[string]string) (stdin, stdout, stderr string, err error) {
if taskp.Vwd != nil {
for k, v := range taskp.Vwd {
v = substitute(v, replacements)
err = checkOutputFilename(outdir, k)
if err != nil {
- return "", "", err
+ return "", "", "", err
+ }
+ if taskp.KeepTmpOutput {
+ err = copyFile(v, outdir+"/"+k)
+ } else {
+ err = os.Symlink(v, outdir+"/"+k)
+ }
+ if err != nil {
+ return "", "", "", err
}
- os.Symlink(v, outdir+"/"+k)
}
}
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
+ return "", "", "", err
}
// Set up stdout redirection
stdout = outdir + "/" + taskp.Stdout
cmd.Stdout, err = os.Create(stdout)
if err != nil {
- return "", "", err
+ return "", "", "", err
}
} else {
cmd.Stdout = os.Stdout
}
+ if taskp.Stderr != "" {
+ err = checkOutputFilename(outdir, taskp.Stderr)
+ if err != nil {
+ return "", "", "", err
+ }
+ // Set up stderr redirection
+ stderr = outdir + "/" + taskp.Stderr
+ cmd.Stderr, err = os.Create(stderr)
+ if err != nil {
+ return "", "", "", err
+ }
+ } else {
+ cmd.Stderr = os.Stderr
+ }
+
if taskp.Env != nil {
// Set up subprocess environment
cmd.Env = os.Environ()
cmd.Env = append(cmd.Env, k+"="+v)
}
}
- return stdin, stdout, nil
+ return stdin, stdout, stderr, nil
}
+// Set up signal handlers. Go sends signal notifications to a "signal
+// channel".
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)
signal.Notify(sigChan, syscall.SIGTERM)
signal.Notify(sigChan, syscall.SIGINT)
return inp
}
+func getKeepTmp(outdir string) (manifest string, err error) {
+ fn, err := os.Open(outdir + "/" + ".arvados#collection")
+ if err != nil {
+ return "", err
+ }
+ defer fn.Close()
+
+ buf, err := ioutil.ReadAll(fn)
+ if err != nil {
+ return "", err
+ }
+ collection := arvados.Collection{}
+ err = json.Unmarshal(buf, &collection)
+ return collection.ManifestText, err
+}
+
func runner(api IArvadosClient,
kc IKeepClient,
jobUuid, taskUuid, crunchtmpdir, keepmount string,
}
var tmpdir, outdir string
- tmpdir, outdir, err = setupDirectories(crunchtmpdir, taskUuid)
+ tmpdir, outdir, err = setupDirectories(crunchtmpdir, taskUuid, taskp.KeepTmpOutput)
if err != nil {
return TempFail{err}
}
"$(task.outdir)": outdir,
"$(task.keep)": keepmount}
+ log.Printf("crunchrunner: $(task.tmpdir)=%v", tmpdir)
+ log.Printf("crunchrunner: $(task.outdir)=%v", outdir)
+ log.Printf("crunchrunner: $(task.keep)=%v", keepmount)
+
// Set up subprocess
for k, v := range taskp.Command {
taskp.Command[k] = substitute(v, replacements)
cmd.Dir = outdir
- var stdin, stdout string
- stdin, stdout, err = setupCommand(cmd, taskp, outdir, replacements)
+ var stdin, stdout, stderr string
+ stdin, stdout, stderr, err = setupCommand(cmd, taskp, outdir, replacements)
if err != nil {
return err
}
if stdout != "" {
stdout = " > " + stdout
}
- log.Printf("Running %v%v%v", cmd.Args, stdin, stdout)
+ if stderr != "" {
+ stderr = " 2> " + stderr
+ }
+ log.Printf("Running %v%v%v%v", cmd.Args, stdin, stdout, stderr)
var caughtSignal os.Signal
- {
- sigChan := setupSignals(cmd)
- defer signal.Stop(sigChan)
+ sigChan := setupSignals(cmd)
- err = cmd.Start()
- if err != nil {
- return TempFail{err}
+ err = cmd.Start()
+ if err != nil {
+ signal.Stop(sigChan)
+ return TempFail{err}
+ }
+
+ finishedSignalNotify := make(chan struct{})
+ go func(sig <-chan os.Signal) {
+ for sig := range sig {
+ caughtSignal = sig
+ cmd.Process.Signal(caughtSignal)
}
+ close(finishedSignalNotify)
+ }(sigChan)
- go func(sig <-chan os.Signal) {
- for sig := range sig {
- caughtSignal = sig
- cmd.Process.Signal(caughtSignal)
- }
- }(sigChan)
+ err = cmd.Wait()
+ signal.Stop(sigChan)
- err = cmd.Wait()
- }
+ close(sigChan)
+ <-finishedSignalNotify
if caughtSignal != nil {
log.Printf("Caught signal %v", caughtSignal)
}
// Upload output directory
- manifest, err := WriteTree(kc, outdir)
+ var manifest string
+ if taskp.KeepTmpOutput {
+ manifest, err = getKeepTmp(outdir)
+ } else {
+ manifest, err = WriteTree(kc, outdir)
+ }
if err != nil {
return TempFail{err}
}
}
var kc IKeepClient
- kc, err = keepclient.MakeKeepClient(&api)
+ kc, err = keepclient.MakeKeepClient(api)
if err != nil {
log.Fatal(err)
}
projects / arvados.git / blobdiff