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 token, err := runner.ContainerToken()
362 return fmt.Errorf("could not get container token: %s", err)
365 runner.ArvMount, err = runner.RunArvMount(arvMountCmd, token)
367 return fmt.Errorf("While trying to start arv-mount: %v", err)
370 for _, p := range collectionPaths {
373 return fmt.Errorf("While checking that input files exist: %v", err)
380 func (runner *ContainerRunner) ProcessDockerAttach(containerReader io.Reader) {
381 // Handle docker log protocol
382 // https://docs.docker.com/engine/reference/api/docker_remote_api_v1.15/#attach-to-a-container
384 header := make([]byte, 8)
386 _, readerr := io.ReadAtLeast(containerReader, header, 8)
389 readsize := int64(header[7]) | (int64(header[6]) << 8) | (int64(header[5]) << 16) | (int64(header[4]) << 24)
392 _, readerr = io.CopyN(runner.Stdout, containerReader, readsize)
395 _, readerr = io.CopyN(runner.Stderr, containerReader, readsize)
400 if readerr != io.EOF {
401 runner.CrunchLog.Printf("While reading docker logs: %v", readerr)
404 closeerr := runner.Stdout.Close()
406 runner.CrunchLog.Printf("While closing stdout logs: %v", closeerr)
409 closeerr = runner.Stderr.Close()
411 runner.CrunchLog.Printf("While closing stderr logs: %v", closeerr)
414 if runner.statReporter != nil {
415 runner.statReporter.Stop()
416 closeerr = runner.statLogger.Close()
418 runner.CrunchLog.Printf("While closing crunchstat logs: %v", closeerr)
422 runner.loggingDone <- true
423 close(runner.loggingDone)
429 func (runner *ContainerRunner) StartCrunchstat() {
430 runner.statLogger = NewThrottledLogger(runner.NewLogWriter("crunchstat"))
431 runner.statReporter = &crunchstat.Reporter{
432 CID: runner.ContainerID,
433 Logger: log.New(runner.statLogger, "", 0),
434 CgroupParent: runner.expectCgroupParent,
435 CgroupRoot: runner.cgroupRoot,
436 PollPeriod: runner.statInterval,
438 runner.statReporter.Start()
441 // AttachLogs connects the docker container stdout and stderr logs to the
442 // Arvados logger which logs to Keep and the API server logs table.
443 func (runner *ContainerRunner) AttachStreams() (err error) {
445 runner.CrunchLog.Print("Attaching container streams")
447 var containerReader io.Reader
448 containerReader, err = runner.Docker.AttachContainer(runner.ContainerID,
449 &dockerclient.AttachOptions{Stream: true, Stdout: true, Stderr: true})
451 return fmt.Errorf("While attaching container stdout/stderr streams: %v", err)
454 runner.loggingDone = make(chan bool)
456 if stdoutMnt, ok := runner.Container.Mounts["stdout"]; ok {
457 stdoutPath := stdoutMnt.Path[len(runner.Container.OutputPath):]
458 index := strings.LastIndex(stdoutPath, "/")
460 subdirs := stdoutPath[:index]
462 st, err := os.Stat(runner.HostOutputDir)
464 return fmt.Errorf("While Stat on temp dir: %v", err)
466 stdoutPath := path.Join(runner.HostOutputDir, subdirs)
467 err = os.MkdirAll(stdoutPath, st.Mode()|os.ModeSetgid|0777)
469 return fmt.Errorf("While MkdirAll %q: %v", stdoutPath, err)
473 stdoutFile, err := os.Create(path.Join(runner.HostOutputDir, stdoutPath))
475 return fmt.Errorf("While creating stdout file: %v", err)
477 runner.Stdout = stdoutFile
479 runner.Stdout = NewThrottledLogger(runner.NewLogWriter("stdout"))
481 runner.Stderr = NewThrottledLogger(runner.NewLogWriter("stderr"))
483 go runner.ProcessDockerAttach(containerReader)
488 // CreateContainer creates the docker container.
489 func (runner *ContainerRunner) CreateContainer() error {
490 runner.CrunchLog.Print("Creating Docker container")
492 runner.ContainerConfig.Cmd = runner.Container.Command
493 if runner.Container.Cwd != "." {
494 runner.ContainerConfig.WorkingDir = runner.Container.Cwd
497 for k, v := range runner.Container.Environment {
498 runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v)
500 if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
501 tok, err := runner.ContainerToken()
505 runner.ContainerConfig.Env = append(runner.ContainerConfig.Env,
506 "ARVADOS_API_TOKEN="+tok,
507 "ARVADOS_API_HOST="+os.Getenv("ARVADOS_API_HOST"),
508 "ARVADOS_API_HOST_INSECURE="+os.Getenv("ARVADOS_API_HOST_INSECURE"),
510 runner.ContainerConfig.NetworkDisabled = false
512 runner.ContainerConfig.NetworkDisabled = true
516 runner.ContainerID, err = runner.Docker.CreateContainer(&runner.ContainerConfig, "", nil)
518 return fmt.Errorf("While creating container: %v", err)
521 runner.HostConfig = dockerclient.HostConfig{
523 CgroupParent: runner.setCgroupParent,
524 LogConfig: dockerclient.LogConfig{
529 return runner.AttachStreams()
532 // StartContainer starts the docker container created by CreateContainer.
533 func (runner *ContainerRunner) StartContainer() error {
534 runner.CrunchLog.Printf("Starting Docker container id '%s'", runner.ContainerID)
535 err := runner.Docker.StartContainer(runner.ContainerID, &runner.HostConfig)
537 return fmt.Errorf("could not start container: %v", err)
542 // WaitFinish waits for the container to terminate, capture the exit code, and
543 // close the stdout/stderr logging.
544 func (runner *ContainerRunner) WaitFinish() error {
545 runner.CrunchLog.Print("Waiting for container to finish")
547 result := runner.Docker.Wait(runner.ContainerID)
550 return fmt.Errorf("While waiting for container to finish: %v", wr.Error)
552 runner.ExitCode = &wr.ExitCode
554 // wait for stdout/stderr to complete
560 // HandleOutput sets the output, unmounts the FUSE mount, and deletes temporary directories
561 func (runner *ContainerRunner) CaptureOutput() error {
562 if runner.finalState != "Complete" {
566 if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
567 // Output may have been set directly by the container, so
568 // refresh the container record to check.
569 err := runner.ArvClient.Get("containers", runner.Container.UUID,
570 nil, &runner.Container)
574 if runner.Container.Output != "" {
575 // Container output is already set.
576 runner.OutputPDH = &runner.Container.Output
581 if runner.HostOutputDir == "" {
585 _, err := os.Stat(runner.HostOutputDir)
587 return fmt.Errorf("While checking host output path: %v", err)
590 var manifestText string
592 collectionMetafile := fmt.Sprintf("%s/.arvados#collection", runner.HostOutputDir)
593 _, err = os.Stat(collectionMetafile)
596 cw := CollectionWriter{runner.Kc, nil, sync.Mutex{}}
597 manifestText, err = cw.WriteTree(runner.HostOutputDir, runner.CrunchLog.Logger)
599 return fmt.Errorf("While uploading output files: %v", err)
602 // FUSE mount directory
603 file, openerr := os.Open(collectionMetafile)
605 return fmt.Errorf("While opening FUSE metafile: %v", err)
609 var rec arvados.Collection
610 err = json.NewDecoder(file).Decode(&rec)
612 return fmt.Errorf("While reading FUSE metafile: %v", err)
614 manifestText = rec.ManifestText
617 var response arvados.Collection
618 err = runner.ArvClient.Create("collections",
620 "collection": arvadosclient.Dict{
621 "expires_at": time.Now().Add(runner.trashLifetime).Format(time.RFC3339),
622 "name": "output for " + runner.Container.UUID,
623 "manifest_text": manifestText}},
626 return fmt.Errorf("While creating output collection: %v", err)
628 runner.OutputPDH = &response.PortableDataHash
632 func (runner *ContainerRunner) loadDiscoveryVars() {
633 tl, err := runner.ArvClient.Discovery("defaultTrashLifetime")
635 log.Fatalf("getting defaultTrashLifetime from discovery document: %s", err)
637 runner.trashLifetime = time.Duration(tl.(float64)) * time.Second
640 func (runner *ContainerRunner) CleanupDirs() {
641 if runner.ArvMount != nil {
642 umount := exec.Command("fusermount", "-z", "-u", runner.ArvMountPoint)
643 umnterr := umount.Run()
645 runner.CrunchLog.Printf("While running fusermount: %v", umnterr)
648 mnterr := <-runner.ArvMountExit
650 runner.CrunchLog.Printf("Arv-mount exit error: %v", mnterr)
654 for _, tmpdir := range runner.CleanupTempDir {
655 rmerr := os.RemoveAll(tmpdir)
657 runner.CrunchLog.Printf("While cleaning up temporary directory %s: %v", tmpdir, rmerr)
662 // CommitLogs posts the collection containing the final container logs.
663 func (runner *ContainerRunner) CommitLogs() error {
664 runner.CrunchLog.Print(runner.finalState)
665 runner.CrunchLog.Close()
667 // Closing CrunchLog above allows it to be committed to Keep at this
668 // point, but re-open crunch log with ArvClient in case there are any
669 // other further (such as failing to write the log to Keep!) while
671 runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{runner.ArvClient, runner.Container.UUID,
674 if runner.LogsPDH != nil {
675 // If we have already assigned something to LogsPDH,
676 // we must be closing the re-opened log, which won't
677 // end up getting attached to the container record and
678 // therefore doesn't need to be saved as a collection
679 // -- it exists only to send logs to other channels.
683 mt, err := runner.LogCollection.ManifestText()
685 return fmt.Errorf("While creating log manifest: %v", err)
688 var response arvados.Collection
689 err = runner.ArvClient.Create("collections",
691 "collection": arvadosclient.Dict{
692 "expires_at": time.Now().Add(runner.trashLifetime).Format(time.RFC3339),
693 "name": "logs for " + runner.Container.UUID,
694 "manifest_text": mt}},
697 return fmt.Errorf("While creating log collection: %v", err)
699 runner.LogsPDH = &response.PortableDataHash
703 // UpdateContainerRunning updates the container state to "Running"
704 func (runner *ContainerRunner) UpdateContainerRunning() error {
705 runner.CancelLock.Lock()
706 defer runner.CancelLock.Unlock()
707 if runner.Cancelled {
710 return runner.ArvClient.Update("containers", runner.Container.UUID,
711 arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running"}}, nil)
714 // ContainerToken returns the api_token the container (and any
715 // arv-mount processes) are allowed to use.
716 func (runner *ContainerRunner) ContainerToken() (string, error) {
717 if runner.token != "" {
718 return runner.token, nil
721 var auth arvados.APIClientAuthorization
722 err := runner.ArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth)
726 runner.token = auth.APIToken
727 return runner.token, nil
730 // UpdateContainerComplete updates the container record state on API
731 // server to "Complete" or "Cancelled"
732 func (runner *ContainerRunner) UpdateContainerFinal() error {
733 update := arvadosclient.Dict{}
734 update["state"] = runner.finalState
735 if runner.finalState == "Complete" {
736 if runner.LogsPDH != nil {
737 update["log"] = *runner.LogsPDH
739 if runner.ExitCode != nil {
740 update["exit_code"] = *runner.ExitCode
742 if runner.OutputPDH != nil {
743 update["output"] = *runner.OutputPDH
746 return runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil)
749 // IsCancelled returns the value of Cancelled, with goroutine safety.
750 func (runner *ContainerRunner) IsCancelled() bool {
751 runner.CancelLock.Lock()
752 defer runner.CancelLock.Unlock()
753 return runner.Cancelled
756 // NewArvLogWriter creates an ArvLogWriter
757 func (runner *ContainerRunner) NewArvLogWriter(name string) io.WriteCloser {
758 return &ArvLogWriter{runner.ArvClient, runner.Container.UUID, name, runner.LogCollection.Open(name + ".txt")}
761 // Run the full container lifecycle.
762 func (runner *ContainerRunner) Run() (err error) {
763 runner.CrunchLog.Printf("Executing container '%s'", runner.Container.UUID)
765 hostname, hosterr := os.Hostname()
767 runner.CrunchLog.Printf("Error getting hostname '%v'", hosterr)
769 runner.CrunchLog.Printf("Executing on host '%s'", hostname)
772 // Clean up temporary directories _after_ finalizing
773 // everything (if we've made any by then)
774 defer runner.CleanupDirs()
776 runner.finalState = "Queued"
779 // checkErr prints e (unless it's nil) and sets err to
780 // e (unless err is already non-nil). Thus, if err
781 // hasn't already been assigned when Run() returns,
782 // this cleanup func will cause Run() to return the
783 // first non-nil error that is passed to checkErr().
784 checkErr := func(e error) {
788 runner.CrunchLog.Print(e)
794 // Log the error encountered in Run(), if any
797 if runner.finalState == "Queued" {
798 runner.UpdateContainerFinal()
802 if runner.IsCancelled() {
803 runner.finalState = "Cancelled"
804 // but don't return yet -- we still want to
805 // capture partial output and write logs
808 checkErr(runner.CaptureOutput())
809 checkErr(runner.CommitLogs())
810 checkErr(runner.UpdateContainerFinal())
812 // The real log is already closed, but then we opened
813 // a new one in case we needed to log anything while
815 runner.CrunchLog.Close()
818 err = runner.ArvClient.Get("containers", runner.Container.UUID, nil, &runner.Container)
820 err = fmt.Errorf("While getting container record: %v", err)
824 // setup signal handling
825 runner.SetupSignals()
827 // check for and/or load image
828 err = runner.LoadImage()
830 err = fmt.Errorf("While loading container image: %v", err)
834 // set up FUSE mount and binds
835 err = runner.SetupMounts()
837 err = fmt.Errorf("While setting up mounts: %v", err)
841 err = runner.CreateContainer()
846 runner.StartCrunchstat()
848 if runner.IsCancelled() {
852 err = runner.UpdateContainerRunning()
856 runner.finalState = "Cancelled"
858 err = runner.StartContainer()
863 err = runner.WaitFinish()
865 runner.finalState = "Complete"
870 // NewContainerRunner creates a new container runner.
871 func NewContainerRunner(api IArvadosClient,
873 docker ThinDockerClient,
874 containerUUID string) *ContainerRunner {
876 cr := &ContainerRunner{ArvClient: api, Kc: kc, Docker: docker}
877 cr.NewLogWriter = cr.NewArvLogWriter
878 cr.RunArvMount = cr.ArvMountCmd
879 cr.MkTempDir = ioutil.TempDir
880 cr.LogCollection = &CollectionWriter{kc, nil, sync.Mutex{}}
881 cr.Container.UUID = containerUUID
882 cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run"))
883 cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0)
884 cr.loadDiscoveryVars()
889 statInterval := flag.Duration("crunchstat-interval", 10*time.Second, "sampling period for periodic resource usage reporting")
890 cgroupRoot := flag.String("cgroup-root", "/sys/fs/cgroup", "path to sysfs cgroup tree")
891 cgroupParent := flag.String("cgroup-parent", "docker", "name of container's parent cgroup (ignored if -cgroup-parent-subsystem is used)")
892 cgroupParentSubsystem := flag.String("cgroup-parent-subsystem", "", "use current cgroup for given subsystem as parent cgroup for container")
895 containerId := flag.Arg(0)
897 api, err := arvadosclient.MakeArvadosClient()
899 log.Fatalf("%s: %v", containerId, err)
903 var kc *keepclient.KeepClient
904 kc, err = keepclient.MakeKeepClient(api)
906 log.Fatalf("%s: %v", containerId, err)
910 var docker *dockerclient.DockerClient
911 docker, err = dockerclient.NewDockerClient("unix:///var/run/docker.sock", nil)
913 log.Fatalf("%s: %v", containerId, err)
916 cr := NewContainerRunner(api, kc, docker, containerId)
917 cr.statInterval = *statInterval
918 cr.cgroupRoot = *cgroupRoot
919 cr.expectCgroupParent = *cgroupParent
920 if *cgroupParentSubsystem != "" {
921 p := findCgroup(*cgroupParentSubsystem)
922 cr.setCgroupParent = p
923 cr.expectCgroupParent = p
928 log.Fatalf("%s: %v", containerId, err)