"errors"
"flag"
"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"
"git.curoverse.com/arvados.git/sdk/go/manifest"
ManifestFileReader(m manifest.Manifest, filename string) (keepclient.ReadCloserWithLen, error)
}
-// Mount describes the mount points to create inside the container.
-type Mount struct {
- Kind string `json:"kind"`
- Writable bool `json:"writable"`
- PortableDataHash string `json:"portable_data_hash"`
- UUID string `json:"uuid"`
- DeviceType string `json:"device_type"`
- Path string `json:"path"`
-}
-
-// Collection record returned by the API server.
-type CollectionRecord struct {
- ManifestText string `json:"manifest_text"`
- PortableDataHash string `json:"portable_data_hash"`
-}
-
-type RuntimeConstraints struct {
- API *bool
-}
-
-// ContainerRecord is the container record returned by the API server.
-type ContainerRecord struct {
- UUID string `json:"uuid"`
- Command []string `json:"command"`
- ContainerImage string `json:"container_image"`
- Cwd string `json:"cwd"`
- Environment map[string]string `json:"environment"`
- Mounts map[string]Mount `json:"mounts"`
- OutputPath string `json:"output_path"`
- Priority int `json:"priority"`
- RuntimeConstraints RuntimeConstraints `json:"runtime_constraints"`
- State string `json:"state"`
- Output string `json:"output"`
-}
-
-// APIClientAuthorization is an arvados#api_client_authorization resource.
-type APIClientAuthorization struct {
- UUID string `json:"uuid"`
- APIToken string `json:"api_token"`
-}
-
// NewLogWriter is a factory function to create a new log writer.
type NewLogWriter func(name string) io.WriteCloser
Docker ThinDockerClient
ArvClient IArvadosClient
Kc IKeepClient
- ContainerRecord
+ arvados.Container
dockerclient.ContainerConfig
dockerclient.HostConfig
token string
signal.Notify(runner.SigChan, syscall.SIGQUIT)
go func(sig <-chan os.Signal) {
- for _ = range sig {
+ for range sig {
if !runner.Cancelled {
runner.CancelLock.Lock()
runner.Cancelled = true
// the image from Keep.
func (runner *ContainerRunner) LoadImage() (err error) {
- runner.CrunchLog.Printf("Fetching Docker image from collection '%s'", runner.ContainerRecord.ContainerImage)
+ runner.CrunchLog.Printf("Fetching Docker image from collection '%s'", runner.Container.ContainerImage)
- var collection CollectionRecord
- err = runner.ArvClient.Get("collections", runner.ContainerRecord.ContainerImage, nil, &collection)
+ var collection arvados.Collection
+ err = runner.ArvClient.Get("collections", runner.Container.ContainerImage, nil, &collection)
if err != nil {
return fmt.Errorf("While getting container image collection: %v", err)
}
collectionPaths := []string{}
runner.Binds = nil
- for bind, mnt := range runner.ContainerRecord.Mounts {
+ for bind, mnt := range runner.Container.Mounts {
if bind == "stdout" {
// Is it a "file" mount kind?
if mnt.Kind != "file" {
}
// Does path start with OutputPath?
- prefix := runner.ContainerRecord.OutputPath
+ prefix := runner.Container.OutputPath
if !strings.HasSuffix(prefix, "/") {
prefix += "/"
}
tmpcount += 1
}
if mnt.Writable {
- if bind == runner.ContainerRecord.OutputPath {
+ if bind == runner.Container.OutputPath {
runner.HostOutputDir = src
}
runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind))
}
collectionPaths = append(collectionPaths, src)
} else if mnt.Kind == "tmp" {
- if bind == runner.ContainerRecord.OutputPath {
+ if bind == runner.Container.OutputPath {
runner.HostOutputDir, err = runner.MkTempDir("", "")
if err != nil {
return fmt.Errorf("While creating mount temp dir: %v", err)
runner.loggingDone = make(chan bool)
- if stdoutMnt, ok := runner.ContainerRecord.Mounts["stdout"]; ok {
- stdoutPath := stdoutMnt.Path[len(runner.ContainerRecord.OutputPath):]
+ if stdoutMnt, ok := runner.Container.Mounts["stdout"]; ok {
+ stdoutPath := stdoutMnt.Path[len(runner.Container.OutputPath):]
index := strings.LastIndex(stdoutPath, "/")
if index > 0 {
subdirs := stdoutPath[:index]
func (runner *ContainerRunner) CreateContainer() error {
runner.CrunchLog.Print("Creating Docker container")
- runner.ContainerConfig.Cmd = runner.ContainerRecord.Command
- if runner.ContainerRecord.Cwd != "." {
- runner.ContainerConfig.WorkingDir = runner.ContainerRecord.Cwd
+ runner.ContainerConfig.Cmd = runner.Container.Command
+ if runner.Container.Cwd != "." {
+ runner.ContainerConfig.WorkingDir = runner.Container.Cwd
}
- for k, v := range runner.ContainerRecord.Environment {
+ for k, v := range runner.Container.Environment {
runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v)
}
- if wantAPI := runner.ContainerRecord.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
+ if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
tok, err := runner.ContainerToken()
if err != nil {
return err
"ARVADOS_API_HOST="+os.Getenv("ARVADOS_API_HOST"),
"ARVADOS_API_HOST_INSECURE="+os.Getenv("ARVADOS_API_HOST_INSECURE"),
)
+ runner.ContainerConfig.NetworkDisabled = false
+ } else {
+ runner.ContainerConfig.NetworkDisabled = true
}
- runner.ContainerConfig.NetworkDisabled = true
-
var err error
runner.ContainerID, err = runner.Docker.CreateContainer(&runner.ContainerConfig, "", nil)
if err != nil {
}
defer file.Close()
- rec := CollectionRecord{}
+ var rec arvados.Collection
err = json.NewDecoder(file).Decode(&rec)
if err != nil {
return fmt.Errorf("While reading FUSE metafile: %v", err)
manifestText = rec.ManifestText
}
- var response CollectionRecord
+ var response arvados.Collection
err = runner.ArvClient.Create("collections",
arvadosclient.Dict{
"collection": arvadosclient.Dict{
// point, but re-open crunch log with ArvClient in case there are any
// other further (such as failing to write the log to Keep!) while
// shutting down
- runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{runner.ArvClient, runner.ContainerRecord.UUID,
+ runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{runner.ArvClient, runner.Container.UUID,
"crunch-run", nil})
if runner.LogsPDH != nil {
return fmt.Errorf("While creating log manifest: %v", err)
}
- var response CollectionRecord
+ var response arvados.Collection
err = runner.ArvClient.Create("collections",
arvadosclient.Dict{
"collection": arvadosclient.Dict{
- "name": "logs for " + runner.ContainerRecord.UUID,
+ "name": "logs for " + runner.Container.UUID,
"manifest_text": mt}},
&response)
if err != nil {
return nil
}
-// UpdateContainerRecordRunning updates the container state to "Running"
-func (runner *ContainerRunner) UpdateContainerRecordRunning() error {
+// UpdateContainerRunning updates the container state to "Running"
+func (runner *ContainerRunner) UpdateContainerRunning() error {
runner.CancelLock.Lock()
defer runner.CancelLock.Unlock()
if runner.Cancelled {
return ErrCancelled
}
- return runner.ArvClient.Update("containers", runner.ContainerRecord.UUID,
+ return runner.ArvClient.Update("containers", runner.Container.UUID,
arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running"}}, nil)
}
return runner.token, nil
}
- var auth APIClientAuthorization
- err := runner.ArvClient.Call("GET", "containers", runner.ContainerRecord.UUID, "auth", nil, &auth)
+ var auth arvados.APIClientAuthorization
+ err := runner.ArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth)
if err != nil {
return "", err
}
return runner.token, nil
}
-// UpdateContainerRecordComplete updates the container record state on API
+// UpdateContainerComplete updates the container record state on API
// server to "Complete" or "Cancelled"
-func (runner *ContainerRunner) UpdateContainerRecordFinal() error {
+func (runner *ContainerRunner) UpdateContainerFinal() error {
update := arvadosclient.Dict{}
update["state"] = runner.finalState
if runner.finalState == "Complete" {
update["output"] = *runner.OutputPDH
}
}
- return runner.ArvClient.Update("containers", runner.ContainerRecord.UUID, arvadosclient.Dict{"container": update}, nil)
+ return runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil)
}
// IsCancelled returns the value of Cancelled, with goroutine safety.
// NewArvLogWriter creates an ArvLogWriter
func (runner *ContainerRunner) NewArvLogWriter(name string) io.WriteCloser {
- return &ArvLogWriter{runner.ArvClient, runner.ContainerRecord.UUID, name, runner.LogCollection.Open(name + ".txt")}
+ return &ArvLogWriter{runner.ArvClient, runner.Container.UUID, name, runner.LogCollection.Open(name + ".txt")}
}
// Run the full container lifecycle.
func (runner *ContainerRunner) Run() (err error) {
- runner.CrunchLog.Printf("Executing container '%s'", runner.ContainerRecord.UUID)
+ runner.CrunchLog.Printf("Executing container '%s'", runner.Container.UUID)
hostname, hosterr := os.Hostname()
if hosterr != nil {
checkErr(err)
if runner.finalState == "Queued" {
- runner.UpdateContainerRecordFinal()
+ runner.UpdateContainerFinal()
return
}
checkErr(runner.CaptureOutput())
checkErr(runner.CommitLogs())
- checkErr(runner.UpdateContainerRecordFinal())
+ checkErr(runner.UpdateContainerFinal())
// The real log is already closed, but then we opened
// a new one in case we needed to log anything while
runner.CrunchLog.Close()
}()
- err = runner.ArvClient.Get("containers", runner.ContainerRecord.UUID, nil, &runner.ContainerRecord)
+ err = runner.ArvClient.Get("containers", runner.Container.UUID, nil, &runner.Container)
if err != nil {
err = fmt.Errorf("While getting container record: %v", err)
return
return
}
- err = runner.UpdateContainerRecordRunning()
+ err = runner.UpdateContainerRunning()
if err != nil {
return
}
cr.RunArvMount = cr.ArvMountCmd
cr.MkTempDir = ioutil.TempDir
cr.LogCollection = &CollectionWriter{kc, nil, sync.Mutex{}}
- cr.ContainerRecord.UUID = containerUUID
+ cr.Container.UUID = containerUUID
cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run"))
cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0)
return cr
projects / arvados.git / blobdiff