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 runner.HostOutputDir == "" {
574 _, err := os.Stat(runner.HostOutputDir)
576 return fmt.Errorf("While checking host output path: %v", err)
579 var manifestText string
581 collectionMetafile := fmt.Sprintf("%s/.arvados#collection", runner.HostOutputDir)
582 _, err = os.Stat(collectionMetafile)
585 cw := CollectionWriter{runner.Kc, nil, sync.Mutex{}}
586 manifestText, err = cw.WriteTree(runner.HostOutputDir, runner.CrunchLog.Logger)
588 return fmt.Errorf("While uploading output files: %v", err)
591 // FUSE mount directory
592 file, openerr := os.Open(collectionMetafile)
594 return fmt.Errorf("While opening FUSE metafile: %v", err)
598 var rec arvados.Collection
599 err = json.NewDecoder(file).Decode(&rec)
601 return fmt.Errorf("While reading FUSE metafile: %v", err)
603 manifestText = rec.ManifestText
606 var response arvados.Collection
607 err = runner.ArvClient.Create("collections",
609 "collection": arvadosclient.Dict{
610 "expires_at": time.Now().Add(runner.trashLifetime).Format(time.RFC3339),
611 "name": "output for " + runner.Container.UUID,
612 "manifest_text": manifestText}},
615 return fmt.Errorf("While creating output collection: %v", err)
617 runner.OutputPDH = &response.PortableDataHash
621 func (runner *ContainerRunner) loadDiscoveryVars() {
622 tl, err := runner.ArvClient.Discovery("defaultTrashLifetime")
624 log.Fatalf("getting defaultTrashLifetime from discovery document: %s", err)
626 runner.trashLifetime = time.Duration(tl.(float64)) * time.Second
629 func (runner *ContainerRunner) CleanupDirs() {
630 if runner.ArvMount != nil {
631 umount := exec.Command("fusermount", "-z", "-u", runner.ArvMountPoint)
632 umnterr := umount.Run()
634 runner.CrunchLog.Printf("While running fusermount: %v", umnterr)
637 mnterr := <-runner.ArvMountExit
639 runner.CrunchLog.Printf("Arv-mount exit error: %v", mnterr)
643 for _, tmpdir := range runner.CleanupTempDir {
644 rmerr := os.RemoveAll(tmpdir)
646 runner.CrunchLog.Printf("While cleaning up temporary directory %s: %v", tmpdir, rmerr)
651 // CommitLogs posts the collection containing the final container logs.
652 func (runner *ContainerRunner) CommitLogs() error {
653 runner.CrunchLog.Print(runner.finalState)
654 runner.CrunchLog.Close()
656 // Closing CrunchLog above allows it to be committed to Keep at this
657 // point, but re-open crunch log with ArvClient in case there are any
658 // other further (such as failing to write the log to Keep!) while
660 runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{runner.ArvClient, runner.Container.UUID,
663 if runner.LogsPDH != nil {
664 // If we have already assigned something to LogsPDH,
665 // we must be closing the re-opened log, which won't
666 // end up getting attached to the container record and
667 // therefore doesn't need to be saved as a collection
668 // -- it exists only to send logs to other channels.
672 mt, err := runner.LogCollection.ManifestText()
674 return fmt.Errorf("While creating log manifest: %v", err)
677 var response arvados.Collection
678 err = runner.ArvClient.Create("collections",
680 "collection": arvadosclient.Dict{
681 "expires_at": time.Now().Add(runner.trashLifetime).Format(time.RFC3339),
682 "name": "logs for " + runner.Container.UUID,
683 "manifest_text": mt}},
686 return fmt.Errorf("While creating log collection: %v", err)
688 runner.LogsPDH = &response.PortableDataHash
692 // UpdateContainerRunning updates the container state to "Running"
693 func (runner *ContainerRunner) UpdateContainerRunning() error {
694 runner.CancelLock.Lock()
695 defer runner.CancelLock.Unlock()
696 if runner.Cancelled {
699 return runner.ArvClient.Update("containers", runner.Container.UUID,
700 arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running"}}, nil)
703 // ContainerToken returns the api_token the container (and any
704 // arv-mount processes) are allowed to use.
705 func (runner *ContainerRunner) ContainerToken() (string, error) {
706 if runner.token != "" {
707 return runner.token, nil
710 var auth arvados.APIClientAuthorization
711 err := runner.ArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth)
715 runner.token = auth.APIToken
716 return runner.token, nil
719 // UpdateContainerComplete updates the container record state on API
720 // server to "Complete" or "Cancelled"
721 func (runner *ContainerRunner) UpdateContainerFinal() error {
722 update := arvadosclient.Dict{}
723 update["state"] = runner.finalState
724 if runner.finalState == "Complete" {
725 if runner.LogsPDH != nil {
726 update["log"] = *runner.LogsPDH
728 if runner.ExitCode != nil {
729 update["exit_code"] = *runner.ExitCode
731 if runner.OutputPDH != nil {
732 update["output"] = *runner.OutputPDH
735 return runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil)
738 // IsCancelled returns the value of Cancelled, with goroutine safety.
739 func (runner *ContainerRunner) IsCancelled() bool {
740 runner.CancelLock.Lock()
741 defer runner.CancelLock.Unlock()
742 return runner.Cancelled
745 // NewArvLogWriter creates an ArvLogWriter
746 func (runner *ContainerRunner) NewArvLogWriter(name string) io.WriteCloser {
747 return &ArvLogWriter{runner.ArvClient, runner.Container.UUID, name, runner.LogCollection.Open(name + ".txt")}
750 // Run the full container lifecycle.
751 func (runner *ContainerRunner) Run() (err error) {
752 runner.CrunchLog.Printf("Executing container '%s'", runner.Container.UUID)
754 hostname, hosterr := os.Hostname()
756 runner.CrunchLog.Printf("Error getting hostname '%v'", hosterr)
758 runner.CrunchLog.Printf("Executing on host '%s'", hostname)
761 // Clean up temporary directories _after_ finalizing
762 // everything (if we've made any by then)
763 defer runner.CleanupDirs()
765 runner.finalState = "Queued"
768 // checkErr prints e (unless it's nil) and sets err to
769 // e (unless err is already non-nil). Thus, if err
770 // hasn't already been assigned when Run() returns,
771 // this cleanup func will cause Run() to return the
772 // first non-nil error that is passed to checkErr().
773 checkErr := func(e error) {
777 runner.CrunchLog.Print(e)
783 // Log the error encountered in Run(), if any
786 if runner.finalState == "Queued" {
787 runner.UpdateContainerFinal()
791 if runner.IsCancelled() {
792 runner.finalState = "Cancelled"
793 // but don't return yet -- we still want to
794 // capture partial output and write logs
797 checkErr(runner.CaptureOutput())
798 checkErr(runner.CommitLogs())
799 checkErr(runner.UpdateContainerFinal())
801 // The real log is already closed, but then we opened
802 // a new one in case we needed to log anything while
804 runner.CrunchLog.Close()
807 err = runner.ArvClient.Get("containers", runner.Container.UUID, nil, &runner.Container)
809 err = fmt.Errorf("While getting container record: %v", err)
813 // setup signal handling
814 runner.SetupSignals()
816 // check for and/or load image
817 err = runner.LoadImage()
819 err = fmt.Errorf("While loading container image: %v", err)
823 // set up FUSE mount and binds
824 err = runner.SetupMounts()
826 err = fmt.Errorf("While setting up mounts: %v", err)
830 err = runner.CreateContainer()
835 runner.StartCrunchstat()
837 if runner.IsCancelled() {
841 err = runner.UpdateContainerRunning()
845 runner.finalState = "Cancelled"
847 err = runner.StartContainer()
852 err = runner.WaitFinish()
854 runner.finalState = "Complete"
859 // NewContainerRunner creates a new container runner.
860 func NewContainerRunner(api IArvadosClient,
862 docker ThinDockerClient,
863 containerUUID string) *ContainerRunner {
865 cr := &ContainerRunner{ArvClient: api, Kc: kc, Docker: docker}
866 cr.NewLogWriter = cr.NewArvLogWriter
867 cr.RunArvMount = cr.ArvMountCmd
868 cr.MkTempDir = ioutil.TempDir
869 cr.LogCollection = &CollectionWriter{kc, nil, sync.Mutex{}}
870 cr.Container.UUID = containerUUID
871 cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run"))
872 cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0)
873 cr.loadDiscoveryVars()
878 statInterval := flag.Duration("crunchstat-interval", 10*time.Second, "sampling period for periodic resource usage reporting")
879 cgroupRoot := flag.String("cgroup-root", "/sys/fs/cgroup", "path to sysfs cgroup tree")
880 cgroupParent := flag.String("cgroup-parent", "docker", "name of container's parent cgroup (ignored if -cgroup-parent-subsystem is used)")
881 cgroupParentSubsystem := flag.String("cgroup-parent-subsystem", "", "use current cgroup for given subsystem as parent cgroup for container")
884 containerId := flag.Arg(0)
886 api, err := arvadosclient.MakeArvadosClient()
888 log.Fatalf("%s: %v", containerId, err)
892 var kc *keepclient.KeepClient
893 kc, err = keepclient.MakeKeepClient(api)
895 log.Fatalf("%s: %v", containerId, err)
899 var docker *dockerclient.DockerClient
900 docker, err = dockerclient.NewDockerClient("unix:///var/run/docker.sock", nil)
902 log.Fatalf("%s: %v", containerId, err)
905 cr := NewContainerRunner(api, kc, docker, containerId)
906 cr.statInterval = *statInterval
907 cr.cgroupRoot = *cgroupRoot
908 cr.expectCgroupParent = *cgroupParent
909 if *cgroupParentSubsystem != "" {
910 p := findCgroup(*cgroupParentSubsystem)
911 cr.setCgroupParent = p
912 cr.expectCgroupParent = p
917 log.Fatalf("%s: %v", containerId, err)