8 "git.curoverse.com/arvados.git/lib/crunchstat"
9 "git.curoverse.com/arvados.git/sdk/go/arvados"
10 "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
11 "git.curoverse.com/arvados.git/sdk/go/keepclient"
12 "git.curoverse.com/arvados.git/sdk/go/manifest"
13 "github.com/curoverse/dockerclient"
28 // IArvadosClient is the minimal Arvados API methods used by crunch-run.
29 type IArvadosClient interface {
30 Create(resourceType string, parameters arvadosclient.Dict, output interface{}) error
31 Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
32 Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
33 Call(method, resourceType, uuid, action string, parameters arvadosclient.Dict, output interface{}) error
34 Discovery(key string) (interface{}, error)
37 // ErrCancelled is the error returned when the container is cancelled.
38 var ErrCancelled = errors.New("Cancelled")
40 // IKeepClient is the minimal Keep API methods used by crunch-run.
41 type IKeepClient interface {
42 PutHB(hash string, buf []byte) (string, int, error)
43 ManifestFileReader(m manifest.Manifest, filename string) (keepclient.ReadCloserWithLen, error)
46 // NewLogWriter is a factory function to create a new log writer.
47 type NewLogWriter func(name string) io.WriteCloser
49 type RunArvMount func(args []string, tok string) (*exec.Cmd, error)
51 type MkTempDir func(string, string) (string, error)
53 // ThinDockerClient is the minimal Docker client interface used by crunch-run.
54 type ThinDockerClient interface {
55 StopContainer(id string, timeout int) error
56 InspectImage(id string) (*dockerclient.ImageInfo, error)
57 LoadImage(reader io.Reader) error
58 CreateContainer(config *dockerclient.ContainerConfig, name string, authConfig *dockerclient.AuthConfig) (string, error)
59 StartContainer(id string, config *dockerclient.HostConfig) error
60 AttachContainer(id string, options *dockerclient.AttachOptions) (io.ReadCloser, error)
61 Wait(id string) <-chan dockerclient.WaitResult
62 RemoveImage(name string, force bool) ([]*dockerclient.ImageDelete, error)
65 // ContainerRunner is the main stateful struct used for a single execution of a
67 type ContainerRunner struct {
68 Docker ThinDockerClient
69 ArvClient IArvadosClient
72 dockerclient.ContainerConfig
73 dockerclient.HostConfig
79 CrunchLog *ThrottledLogger
81 Stderr *ThrottledLogger
82 LogCollection *CollectionWriter
89 CleanupTempDir []string
94 SigChan chan os.Signal
95 ArvMountExit chan error
97 trashLifetime time.Duration
99 statLogger io.WriteCloser
100 statReporter *crunchstat.Reporter
101 statInterval time.Duration
103 // What we expect the container's cgroup parent to be.
104 expectCgroupParent string
105 // What we tell docker to use as the container's cgroup
106 // parent. Note: Ideally we would use the same field for both
107 // expectCgroupParent and setCgroupParent, and just make it
108 // default to "docker". However, when using docker < 1.10 with
109 // systemd, specifying a non-empty cgroup parent (even the
110 // default value "docker") hits a docker bug
111 // (https://github.com/docker/docker/issues/17126). Using two
112 // separate fields makes it possible to use the "expect cgroup
113 // parent to be X" feature even on sites where the "specify
114 // cgroup parent" feature breaks.
115 setCgroupParent string
118 // SetupSignals sets up signal handling to gracefully terminate the underlying
119 // Docker container and update state when receiving a TERM, INT or QUIT signal.
120 func (runner *ContainerRunner) SetupSignals() {
121 runner.SigChan = make(chan os.Signal, 1)
122 signal.Notify(runner.SigChan, syscall.SIGTERM)
123 signal.Notify(runner.SigChan, syscall.SIGINT)
124 signal.Notify(runner.SigChan, syscall.SIGQUIT)
126 go func(sig <-chan os.Signal) {
128 if !runner.Cancelled {
129 runner.CancelLock.Lock()
130 runner.Cancelled = true
131 if runner.ContainerID != "" {
132 runner.Docker.StopContainer(runner.ContainerID, 10)
134 runner.CancelLock.Unlock()
140 // LoadImage determines the docker image id from the container record and
141 // checks if it is available in the local Docker image store. If not, it loads
142 // the image from Keep.
143 func (runner *ContainerRunner) LoadImage() (err error) {
145 runner.CrunchLog.Printf("Fetching Docker image from collection '%s'", runner.Container.ContainerImage)
147 var collection arvados.Collection
148 err = runner.ArvClient.Get("collections", runner.Container.ContainerImage, nil, &collection)
150 return fmt.Errorf("While getting container image collection: %v", err)
152 manifest := manifest.Manifest{Text: collection.ManifestText}
153 var img, imageID string
154 for ms := range manifest.StreamIter() {
155 img = ms.FileStreamSegments[0].Name
156 if !strings.HasSuffix(img, ".tar") {
157 return fmt.Errorf("First file in the container image collection does not end in .tar")
159 imageID = img[:len(img)-4]
162 runner.CrunchLog.Printf("Using Docker image id '%s'", imageID)
164 _, err = runner.Docker.InspectImage(imageID)
166 runner.CrunchLog.Print("Loading Docker image from keep")
168 var readCloser io.ReadCloser
169 readCloser, err = runner.Kc.ManifestFileReader(manifest, img)
171 return fmt.Errorf("While creating ManifestFileReader for container image: %v", err)
174 err = runner.Docker.LoadImage(readCloser)
176 return fmt.Errorf("While loading container image into Docker: %v", err)
179 runner.CrunchLog.Print("Docker image is available")
182 runner.ContainerConfig.Image = imageID
187 func (runner *ContainerRunner) ArvMountCmd(arvMountCmd []string, token string) (c *exec.Cmd, err error) {
188 c = exec.Command("arv-mount", arvMountCmd...)
190 // Copy our environment, but override ARVADOS_API_TOKEN with
191 // the container auth token.
193 for _, s := range os.Environ() {
194 if !strings.HasPrefix(s, "ARVADOS_API_TOKEN=") {
195 c.Env = append(c.Env, s)
198 c.Env = append(c.Env, "ARVADOS_API_TOKEN="+token)
200 nt := NewThrottledLogger(runner.NewLogWriter("arv-mount"))
209 statReadme := make(chan bool)
210 runner.ArvMountExit = make(chan error)
215 time.Sleep(100 * time.Millisecond)
216 _, err = os.Stat(fmt.Sprintf("%s/by_id/README", runner.ArvMountPoint))
226 runner.ArvMountExit <- c.Wait()
227 close(runner.ArvMountExit)
233 case err := <-runner.ArvMountExit:
234 runner.ArvMount = nil
242 func (runner *ContainerRunner) SetupMounts() (err error) {
243 runner.ArvMountPoint, err = runner.MkTempDir("", "keep")
245 return fmt.Errorf("While creating keep mount temp dir: %v", err)
248 runner.CleanupTempDir = append(runner.CleanupTempDir, runner.ArvMountPoint)
252 arvMountCmd := []string{"--foreground", "--allow-other", "--read-write"}
253 collectionPaths := []string{}
256 for bind, mnt := range runner.Container.Mounts {
257 if bind == "stdout" {
258 // Is it a "file" mount kind?
259 if mnt.Kind != "file" {
260 return fmt.Errorf("Unsupported mount kind '%s' for stdout. Only 'file' is supported.", mnt.Kind)
263 // Does path start with OutputPath?
264 prefix := runner.Container.OutputPath
265 if !strings.HasSuffix(prefix, "/") {
268 if !strings.HasPrefix(mnt.Path, prefix) {
269 return fmt.Errorf("Stdout path does not start with OutputPath: %s, %s", mnt.Path, prefix)
274 case mnt.Kind == "collection":
276 if mnt.UUID != "" && mnt.PortableDataHash != "" {
277 return fmt.Errorf("Cannot specify both 'uuid' and 'portable_data_hash' for a collection mount")
281 return fmt.Errorf("Writing to existing collections currently not permitted.")
284 src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.UUID)
285 } else if mnt.PortableDataHash != "" {
287 return fmt.Errorf("Can never write to a collection specified by portable data hash")
289 src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.PortableDataHash)
291 src = fmt.Sprintf("%s/tmp%d", runner.ArvMountPoint, tmpcount)
292 arvMountCmd = append(arvMountCmd, "--mount-tmp")
293 arvMountCmd = append(arvMountCmd, fmt.Sprintf("tmp%d", tmpcount))
297 if bind == runner.Container.OutputPath {
298 runner.HostOutputDir = src
300 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind))
302 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", src, bind))
304 collectionPaths = append(collectionPaths, src)
306 case mnt.Kind == "tmp" && bind == runner.Container.OutputPath:
307 runner.HostOutputDir, err = runner.MkTempDir("", "")
309 return fmt.Errorf("While creating mount temp dir: %v", err)
311 st, staterr := os.Stat(runner.HostOutputDir)
313 return fmt.Errorf("While Stat on temp dir: %v", staterr)
315 err = os.Chmod(runner.HostOutputDir, st.Mode()|os.ModeSetgid|0777)
317 return fmt.Errorf("While Chmod temp dir: %v", err)
319 runner.CleanupTempDir = append(runner.CleanupTempDir, runner.HostOutputDir)
320 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", runner.HostOutputDir, bind))
322 case mnt.Kind == "tmp":
323 runner.Binds = append(runner.Binds, bind)
325 case mnt.Kind == "json":
326 jsondata, err := json.Marshal(mnt.Content)
328 return fmt.Errorf("encoding json data: %v", err)
330 // Create a tempdir with a single file
331 // (instead of just a tempfile): this way we
332 // can ensure the file is world-readable
333 // inside the container, without having to
334 // make it world-readable on the docker host.
335 tmpdir, err := runner.MkTempDir("", "")
337 return fmt.Errorf("creating temp dir: %v", err)
339 runner.CleanupTempDir = append(runner.CleanupTempDir, tmpdir)
340 tmpfn := filepath.Join(tmpdir, "mountdata.json")
341 err = ioutil.WriteFile(tmpfn, jsondata, 0644)
343 return fmt.Errorf("writing temp file: %v", err)
345 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", tmpfn, bind))
349 if runner.HostOutputDir == "" {
350 return fmt.Errorf("Output path does not correspond to a writable mount point")
354 arvMountCmd = append(arvMountCmd, "--mount-by-pdh", "by_id")
356 arvMountCmd = append(arvMountCmd, "--mount-by-id", "by_id")
358 arvMountCmd = append(arvMountCmd, runner.ArvMountPoint)
360 if runner.Container.RuntimeConstraints.KeepCacheRAM > 0 {
361 arvMountCmd = append(arvMountCmd, "--file-cache", fmt.Sprintf("%d", runner.Container.RuntimeConstraints.KeepCacheRAM))
364 token, err := runner.ContainerToken()
366 return fmt.Errorf("could not get container token: %s", err)
369 runner.ArvMount, err = runner.RunArvMount(arvMountCmd, token)
371 return fmt.Errorf("While trying to start arv-mount: %v", err)
374 for _, p := range collectionPaths {
377 return fmt.Errorf("While checking that input files exist: %v", err)
384 func (runner *ContainerRunner) ProcessDockerAttach(containerReader io.Reader) {
385 // Handle docker log protocol
386 // https://docs.docker.com/engine/reference/api/docker_remote_api_v1.15/#attach-to-a-container
388 header := make([]byte, 8)
390 _, readerr := io.ReadAtLeast(containerReader, header, 8)
393 readsize := int64(header[7]) | (int64(header[6]) << 8) | (int64(header[5]) << 16) | (int64(header[4]) << 24)
396 _, readerr = io.CopyN(runner.Stdout, containerReader, readsize)
399 _, readerr = io.CopyN(runner.Stderr, containerReader, readsize)
404 if readerr != io.EOF {
405 runner.CrunchLog.Printf("While reading docker logs: %v", readerr)
408 closeerr := runner.Stdout.Close()
410 runner.CrunchLog.Printf("While closing stdout logs: %v", closeerr)
413 closeerr = runner.Stderr.Close()
415 runner.CrunchLog.Printf("While closing stderr logs: %v", closeerr)
418 if runner.statReporter != nil {
419 runner.statReporter.Stop()
420 closeerr = runner.statLogger.Close()
422 runner.CrunchLog.Printf("While closing crunchstat logs: %v", closeerr)
426 runner.loggingDone <- true
427 close(runner.loggingDone)
433 func (runner *ContainerRunner) StartCrunchstat() {
434 runner.statLogger = NewThrottledLogger(runner.NewLogWriter("crunchstat"))
435 runner.statReporter = &crunchstat.Reporter{
436 CID: runner.ContainerID,
437 Logger: log.New(runner.statLogger, "", 0),
438 CgroupParent: runner.expectCgroupParent,
439 CgroupRoot: runner.cgroupRoot,
440 PollPeriod: runner.statInterval,
442 runner.statReporter.Start()
445 // AttachLogs connects the docker container stdout and stderr logs to the
446 // Arvados logger which logs to Keep and the API server logs table.
447 func (runner *ContainerRunner) AttachStreams() (err error) {
449 runner.CrunchLog.Print("Attaching container streams")
451 var containerReader io.Reader
452 containerReader, err = runner.Docker.AttachContainer(runner.ContainerID,
453 &dockerclient.AttachOptions{Stream: true, Stdout: true, Stderr: true})
455 return fmt.Errorf("While attaching container stdout/stderr streams: %v", err)
458 runner.loggingDone = make(chan bool)
460 if stdoutMnt, ok := runner.Container.Mounts["stdout"]; ok {
461 stdoutPath := stdoutMnt.Path[len(runner.Container.OutputPath):]
462 index := strings.LastIndex(stdoutPath, "/")
464 subdirs := stdoutPath[:index]
466 st, err := os.Stat(runner.HostOutputDir)
468 return fmt.Errorf("While Stat on temp dir: %v", err)
470 stdoutPath := path.Join(runner.HostOutputDir, subdirs)
471 err = os.MkdirAll(stdoutPath, st.Mode()|os.ModeSetgid|0777)
473 return fmt.Errorf("While MkdirAll %q: %v", stdoutPath, err)
477 stdoutFile, err := os.Create(path.Join(runner.HostOutputDir, stdoutPath))
479 return fmt.Errorf("While creating stdout file: %v", err)
481 runner.Stdout = stdoutFile
483 runner.Stdout = NewThrottledLogger(runner.NewLogWriter("stdout"))
485 runner.Stderr = NewThrottledLogger(runner.NewLogWriter("stderr"))
487 go runner.ProcessDockerAttach(containerReader)
492 // CreateContainer creates the docker container.
493 func (runner *ContainerRunner) CreateContainer() error {
494 runner.CrunchLog.Print("Creating Docker container")
496 runner.ContainerConfig.Cmd = runner.Container.Command
497 if runner.Container.Cwd != "." {
498 runner.ContainerConfig.WorkingDir = runner.Container.Cwd
501 for k, v := range runner.Container.Environment {
502 runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v)
504 if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
505 tok, err := runner.ContainerToken()
509 runner.ContainerConfig.Env = append(runner.ContainerConfig.Env,
510 "ARVADOS_API_TOKEN="+tok,
511 "ARVADOS_API_HOST="+os.Getenv("ARVADOS_API_HOST"),
512 "ARVADOS_API_HOST_INSECURE="+os.Getenv("ARVADOS_API_HOST_INSECURE"),
514 runner.ContainerConfig.NetworkDisabled = false
516 runner.ContainerConfig.NetworkDisabled = true
520 runner.ContainerID, err = runner.Docker.CreateContainer(&runner.ContainerConfig, "", nil)
522 return fmt.Errorf("While creating container: %v", err)
525 runner.HostConfig = dockerclient.HostConfig{
527 CgroupParent: runner.setCgroupParent,
528 LogConfig: dockerclient.LogConfig{
533 return runner.AttachStreams()
536 // StartContainer starts the docker container created by CreateContainer.
537 func (runner *ContainerRunner) StartContainer() error {
538 runner.CrunchLog.Printf("Starting Docker container id '%s'", runner.ContainerID)
539 err := runner.Docker.StartContainer(runner.ContainerID, &runner.HostConfig)
541 return fmt.Errorf("could not start container: %v", err)
546 // WaitFinish waits for the container to terminate, capture the exit code, and
547 // close the stdout/stderr logging.
548 func (runner *ContainerRunner) WaitFinish() error {
549 runner.CrunchLog.Print("Waiting for container to finish")
551 result := runner.Docker.Wait(runner.ContainerID)
554 return fmt.Errorf("While waiting for container to finish: %v", wr.Error)
556 runner.ExitCode = &wr.ExitCode
558 // wait for stdout/stderr to complete
564 // HandleOutput sets the output, unmounts the FUSE mount, and deletes temporary directories
565 func (runner *ContainerRunner) CaptureOutput() error {
566 if runner.finalState != "Complete" {
570 if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
571 // Output may have been set directly by the container, so
572 // refresh the container record to check.
573 err := runner.ArvClient.Get("containers", runner.Container.UUID,
574 nil, &runner.Container)
578 if runner.Container.Output != "" {
579 // Container output is already set.
580 runner.OutputPDH = &runner.Container.Output
585 if runner.HostOutputDir == "" {
589 _, err := os.Stat(runner.HostOutputDir)
591 return fmt.Errorf("While checking host output path: %v", err)
594 var manifestText string
596 collectionMetafile := fmt.Sprintf("%s/.arvados#collection", runner.HostOutputDir)
597 _, err = os.Stat(collectionMetafile)
600 cw := CollectionWriter{runner.Kc, nil, sync.Mutex{}}
601 manifestText, err = cw.WriteTree(runner.HostOutputDir, runner.CrunchLog.Logger)
603 return fmt.Errorf("While uploading output files: %v", err)
606 // FUSE mount directory
607 file, openerr := os.Open(collectionMetafile)
609 return fmt.Errorf("While opening FUSE metafile: %v", err)
613 var rec arvados.Collection
614 err = json.NewDecoder(file).Decode(&rec)
616 return fmt.Errorf("While reading FUSE metafile: %v", err)
618 manifestText = rec.ManifestText
621 var response arvados.Collection
622 err = runner.ArvClient.Create("collections",
624 "collection": arvadosclient.Dict{
625 "expires_at": time.Now().Add(runner.trashLifetime).Format(time.RFC3339),
626 "name": "output for " + runner.Container.UUID,
627 "manifest_text": manifestText}},
630 return fmt.Errorf("While creating output collection: %v", err)
632 runner.OutputPDH = &response.PortableDataHash
636 func (runner *ContainerRunner) loadDiscoveryVars() {
637 tl, err := runner.ArvClient.Discovery("defaultTrashLifetime")
639 log.Fatalf("getting defaultTrashLifetime from discovery document: %s", err)
641 runner.trashLifetime = time.Duration(tl.(float64)) * time.Second
644 func (runner *ContainerRunner) CleanupDirs() {
645 if runner.ArvMount != nil {
646 umount := exec.Command("fusermount", "-z", "-u", runner.ArvMountPoint)
647 umnterr := umount.Run()
649 runner.CrunchLog.Printf("While running fusermount: %v", umnterr)
652 mnterr := <-runner.ArvMountExit
654 runner.CrunchLog.Printf("Arv-mount exit error: %v", mnterr)
658 for _, tmpdir := range runner.CleanupTempDir {
659 rmerr := os.RemoveAll(tmpdir)
661 runner.CrunchLog.Printf("While cleaning up temporary directory %s: %v", tmpdir, rmerr)
666 // CommitLogs posts the collection containing the final container logs.
667 func (runner *ContainerRunner) CommitLogs() error {
668 runner.CrunchLog.Print(runner.finalState)
669 runner.CrunchLog.Close()
671 // Closing CrunchLog above allows it to be committed to Keep at this
672 // point, but re-open crunch log with ArvClient in case there are any
673 // other further (such as failing to write the log to Keep!) while
675 runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{runner.ArvClient, runner.Container.UUID,
678 if runner.LogsPDH != nil {
679 // If we have already assigned something to LogsPDH,
680 // we must be closing the re-opened log, which won't
681 // end up getting attached to the container record and
682 // therefore doesn't need to be saved as a collection
683 // -- it exists only to send logs to other channels.
687 mt, err := runner.LogCollection.ManifestText()
689 return fmt.Errorf("While creating log manifest: %v", err)
692 var response arvados.Collection
693 err = runner.ArvClient.Create("collections",
695 "collection": arvadosclient.Dict{
696 "expires_at": time.Now().Add(runner.trashLifetime).Format(time.RFC3339),
697 "name": "logs for " + runner.Container.UUID,
698 "manifest_text": mt}},
701 return fmt.Errorf("While creating log collection: %v", err)
703 runner.LogsPDH = &response.PortableDataHash
707 // UpdateContainerRunning updates the container state to "Running"
708 func (runner *ContainerRunner) UpdateContainerRunning() error {
709 runner.CancelLock.Lock()
710 defer runner.CancelLock.Unlock()
711 if runner.Cancelled {
714 return runner.ArvClient.Update("containers", runner.Container.UUID,
715 arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running"}}, nil)
718 // ContainerToken returns the api_token the container (and any
719 // arv-mount processes) are allowed to use.
720 func (runner *ContainerRunner) ContainerToken() (string, error) {
721 if runner.token != "" {
722 return runner.token, nil
725 var auth arvados.APIClientAuthorization
726 err := runner.ArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth)
730 runner.token = auth.APIToken
731 return runner.token, nil
734 // UpdateContainerComplete updates the container record state on API
735 // server to "Complete" or "Cancelled"
736 func (runner *ContainerRunner) UpdateContainerFinal() error {
737 update := arvadosclient.Dict{}
738 update["state"] = runner.finalState
739 if runner.finalState == "Complete" {
740 if runner.LogsPDH != nil {
741 update["log"] = *runner.LogsPDH
743 if runner.ExitCode != nil {
744 update["exit_code"] = *runner.ExitCode
746 if runner.OutputPDH != nil {
747 update["output"] = *runner.OutputPDH
750 return runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil)
753 // IsCancelled returns the value of Cancelled, with goroutine safety.
754 func (runner *ContainerRunner) IsCancelled() bool {
755 runner.CancelLock.Lock()
756 defer runner.CancelLock.Unlock()
757 return runner.Cancelled
760 // NewArvLogWriter creates an ArvLogWriter
761 func (runner *ContainerRunner) NewArvLogWriter(name string) io.WriteCloser {
762 return &ArvLogWriter{runner.ArvClient, runner.Container.UUID, name, runner.LogCollection.Open(name + ".txt")}
765 // Run the full container lifecycle.
766 func (runner *ContainerRunner) Run() (err error) {
767 runner.CrunchLog.Printf("Executing container '%s'", runner.Container.UUID)
769 hostname, hosterr := os.Hostname()
771 runner.CrunchLog.Printf("Error getting hostname '%v'", hosterr)
773 runner.CrunchLog.Printf("Executing on host '%s'", hostname)
776 // Clean up temporary directories _after_ finalizing
777 // everything (if we've made any by then)
778 defer runner.CleanupDirs()
780 runner.finalState = "Queued"
783 // checkErr prints e (unless it's nil) and sets err to
784 // e (unless err is already non-nil). Thus, if err
785 // hasn't already been assigned when Run() returns,
786 // this cleanup func will cause Run() to return the
787 // first non-nil error that is passed to checkErr().
788 checkErr := func(e error) {
792 runner.CrunchLog.Print(e)
798 // Log the error encountered in Run(), if any
801 if runner.finalState == "Queued" {
802 runner.UpdateContainerFinal()
806 if runner.IsCancelled() {
807 runner.finalState = "Cancelled"
808 // but don't return yet -- we still want to
809 // capture partial output and write logs
812 checkErr(runner.CaptureOutput())
813 checkErr(runner.CommitLogs())
814 checkErr(runner.UpdateContainerFinal())
816 // The real log is already closed, but then we opened
817 // a new one in case we needed to log anything while
819 runner.CrunchLog.Close()
822 err = runner.ArvClient.Get("containers", runner.Container.UUID, nil, &runner.Container)
824 err = fmt.Errorf("While getting container record: %v", err)
828 // setup signal handling
829 runner.SetupSignals()
831 // check for and/or load image
832 err = runner.LoadImage()
834 err = fmt.Errorf("While loading container image: %v", err)
838 // set up FUSE mount and binds
839 err = runner.SetupMounts()
841 err = fmt.Errorf("While setting up mounts: %v", err)
845 err = runner.CreateContainer()
850 runner.StartCrunchstat()
852 if runner.IsCancelled() {
856 err = runner.UpdateContainerRunning()
860 runner.finalState = "Cancelled"
862 err = runner.StartContainer()
867 err = runner.WaitFinish()
869 runner.finalState = "Complete"
874 // NewContainerRunner creates a new container runner.
875 func NewContainerRunner(api IArvadosClient,
877 docker ThinDockerClient,
878 containerUUID string) *ContainerRunner {
880 cr := &ContainerRunner{ArvClient: api, Kc: kc, Docker: docker}
881 cr.NewLogWriter = cr.NewArvLogWriter
882 cr.RunArvMount = cr.ArvMountCmd
883 cr.MkTempDir = ioutil.TempDir
884 cr.LogCollection = &CollectionWriter{kc, nil, sync.Mutex{}}
885 cr.Container.UUID = containerUUID
886 cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run"))
887 cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0)
888 cr.loadDiscoveryVars()
893 statInterval := flag.Duration("crunchstat-interval", 10*time.Second, "sampling period for periodic resource usage reporting")
894 cgroupRoot := flag.String("cgroup-root", "/sys/fs/cgroup", "path to sysfs cgroup tree")
895 cgroupParent := flag.String("cgroup-parent", "docker", "name of container's parent cgroup (ignored if -cgroup-parent-subsystem is used)")
896 cgroupParentSubsystem := flag.String("cgroup-parent-subsystem", "", "use current cgroup for given subsystem as parent cgroup for container")
899 containerId := flag.Arg(0)
901 api, err := arvadosclient.MakeArvadosClient()
903 log.Fatalf("%s: %v", containerId, err)
907 var kc *keepclient.KeepClient
908 kc, err = keepclient.MakeKeepClient(api)
910 log.Fatalf("%s: %v", containerId, err)
914 var docker *dockerclient.DockerClient
915 docker, err = dockerclient.NewDockerClient("unix:///var/run/docker.sock", nil)
917 log.Fatalf("%s: %v", containerId, err)
920 cr := NewContainerRunner(api, kc, docker, containerId)
921 cr.statInterval = *statInterval
922 cr.cgroupRoot = *cgroupRoot
923 cr.expectCgroupParent = *cgroupParent
924 if *cgroupParentSubsystem != "" {
925 p := findCgroup(*cgroupParentSubsystem)
926 cr.setCgroupParent = p
927 cr.expectCgroupParent = p
932 log.Fatalf("%s: %v", containerId, err)