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"
29 // IArvadosClient is the minimal Arvados API methods used by crunch-run.
30 type IArvadosClient interface {
31 Create(resourceType string, parameters arvadosclient.Dict, output interface{}) error
32 Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
33 Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
34 Call(method, resourceType, uuid, action string, parameters arvadosclient.Dict, output interface{}) error
35 Discovery(key string) (interface{}, error)
38 // ErrCancelled is the error returned when the container is cancelled.
39 var ErrCancelled = errors.New("Cancelled")
41 // IKeepClient is the minimal Keep API methods used by crunch-run.
42 type IKeepClient interface {
43 PutHB(hash string, buf []byte) (string, int, error)
44 ManifestFileReader(m manifest.Manifest, filename string) (keepclient.Reader, error)
47 // NewLogWriter is a factory function to create a new log writer.
48 type NewLogWriter func(name string) io.WriteCloser
50 type RunArvMount func(args []string, tok string) (*exec.Cmd, error)
52 type MkTempDir func(string, string) (string, error)
54 // ThinDockerClient is the minimal Docker client interface used by crunch-run.
55 type ThinDockerClient interface {
56 StopContainer(id string, timeout int) error
57 InspectImage(id string) (*dockerclient.ImageInfo, error)
58 LoadImage(reader io.Reader) error
59 CreateContainer(config *dockerclient.ContainerConfig, name string, authConfig *dockerclient.AuthConfig) (string, error)
60 StartContainer(id string, config *dockerclient.HostConfig) error
61 AttachContainer(id string, options *dockerclient.AttachOptions) (io.ReadCloser, error)
62 Wait(id string) <-chan dockerclient.WaitResult
63 RemoveImage(name string, force bool) ([]*dockerclient.ImageDelete, error)
66 // ContainerRunner is the main stateful struct used for a single execution of a
68 type ContainerRunner struct {
69 Docker ThinDockerClient
70 ArvClient IArvadosClient
73 dockerclient.ContainerConfig
74 dockerclient.HostConfig
80 CrunchLog *ThrottledLogger
82 Stderr *ThrottledLogger
83 LogCollection *CollectionWriter
90 CleanupTempDir []string
95 SigChan chan os.Signal
96 ArvMountExit chan error
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) SetupArvMountPoint(prefix string) (err error) {
243 if runner.ArvMountPoint == "" {
244 runner.ArvMountPoint, err = runner.MkTempDir("", prefix)
249 func (runner *ContainerRunner) SetupMounts() (err error) {
250 err = runner.SetupArvMountPoint("keep")
252 return fmt.Errorf("While creating keep mount temp dir: %v", err)
255 runner.CleanupTempDir = append(runner.CleanupTempDir, runner.ArvMountPoint)
259 arvMountCmd := []string{"--foreground", "--allow-other", "--read-write"}
261 if runner.Container.RuntimeConstraints.KeepCacheRAM > 0 {
262 arvMountCmd = append(arvMountCmd, "--file-cache", fmt.Sprintf("%d", runner.Container.RuntimeConstraints.KeepCacheRAM))
265 collectionPaths := []string{}
267 needCertMount := true
270 for bind, _ := range runner.Container.Mounts {
271 binds = append(binds, bind)
275 for _, bind := range binds {
276 mnt := runner.Container.Mounts[bind]
277 if bind == "stdout" {
278 // Is it a "file" mount kind?
279 if mnt.Kind != "file" {
280 return fmt.Errorf("Unsupported mount kind '%s' for stdout. Only 'file' is supported.", mnt.Kind)
283 // Does path start with OutputPath?
284 prefix := runner.Container.OutputPath
285 if !strings.HasSuffix(prefix, "/") {
288 if !strings.HasPrefix(mnt.Path, prefix) {
289 return fmt.Errorf("Stdout path does not start with OutputPath: %s, %s", mnt.Path, prefix)
293 if bind == "/etc/arvados/ca-certificates.crt" {
294 needCertMount = false
297 if strings.HasPrefix(bind, runner.Container.OutputPath+"/") && bind != runner.Container.OutputPath+"/" {
298 if mnt.Kind != "collection" {
299 return fmt.Errorf("Only mount points of kind 'collection' are supported underneath the output_path: %v", bind)
304 case mnt.Kind == "collection":
306 if mnt.UUID != "" && mnt.PortableDataHash != "" {
307 return fmt.Errorf("Cannot specify both 'uuid' and 'portable_data_hash' for a collection mount")
311 return fmt.Errorf("Writing to existing collections currently not permitted.")
314 src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.UUID)
315 } else if mnt.PortableDataHash != "" {
317 return fmt.Errorf("Can never write to a collection specified by portable data hash")
319 idx := strings.Index(mnt.PortableDataHash, "/")
321 mnt.Path = path.Clean(mnt.PortableDataHash[idx:])
322 mnt.PortableDataHash = mnt.PortableDataHash[0:idx]
323 runner.Container.Mounts[bind] = mnt
325 src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.PortableDataHash)
326 if mnt.Path != "" && mnt.Path != "." {
327 if strings.HasPrefix(mnt.Path, "./") {
328 mnt.Path = mnt.Path[2:]
329 } else if strings.HasPrefix(mnt.Path, "/") {
330 mnt.Path = mnt.Path[1:]
332 src += "/" + mnt.Path
335 src = fmt.Sprintf("%s/tmp%d", runner.ArvMountPoint, tmpcount)
336 arvMountCmd = append(arvMountCmd, "--mount-tmp")
337 arvMountCmd = append(arvMountCmd, fmt.Sprintf("tmp%d", tmpcount))
341 if bind == runner.Container.OutputPath {
342 runner.HostOutputDir = src
343 } else if strings.HasPrefix(bind, runner.Container.OutputPath+"/") {
344 return fmt.Errorf("Writable mount points are not permitted underneath the output_path: %v", bind)
346 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind))
348 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", src, bind))
350 collectionPaths = append(collectionPaths, src)
352 case mnt.Kind == "tmp" && bind == runner.Container.OutputPath:
353 runner.HostOutputDir, err = runner.MkTempDir("", "")
355 return fmt.Errorf("While creating mount temp dir: %v", err)
357 st, staterr := os.Stat(runner.HostOutputDir)
359 return fmt.Errorf("While Stat on temp dir: %v", staterr)
361 err = os.Chmod(runner.HostOutputDir, st.Mode()|os.ModeSetgid|0777)
363 return fmt.Errorf("While Chmod temp dir: %v", err)
365 runner.CleanupTempDir = append(runner.CleanupTempDir, runner.HostOutputDir)
366 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", runner.HostOutputDir, bind))
368 case mnt.Kind == "tmp":
369 runner.Binds = append(runner.Binds, bind)
371 case mnt.Kind == "json":
372 jsondata, err := json.Marshal(mnt.Content)
374 return fmt.Errorf("encoding json data: %v", err)
376 // Create a tempdir with a single file
377 // (instead of just a tempfile): this way we
378 // can ensure the file is world-readable
379 // inside the container, without having to
380 // make it world-readable on the docker host.
381 tmpdir, err := runner.MkTempDir("", "")
383 return fmt.Errorf("creating temp dir: %v", err)
385 runner.CleanupTempDir = append(runner.CleanupTempDir, tmpdir)
386 tmpfn := filepath.Join(tmpdir, "mountdata.json")
387 err = ioutil.WriteFile(tmpfn, jsondata, 0644)
389 return fmt.Errorf("writing temp file: %v", err)
391 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", tmpfn, bind))
395 if runner.HostOutputDir == "" {
396 return fmt.Errorf("Output path does not correspond to a writable mount point")
399 if wantAPI := runner.Container.RuntimeConstraints.API; needCertMount && wantAPI != nil && *wantAPI {
400 for _, certfile := range arvadosclient.CertFiles {
401 _, err := os.Stat(certfile)
403 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:/etc/arvados/ca-certificates.crt:ro", certfile))
410 arvMountCmd = append(arvMountCmd, "--mount-by-pdh", "by_id")
412 arvMountCmd = append(arvMountCmd, "--mount-by-id", "by_id")
414 arvMountCmd = append(arvMountCmd, runner.ArvMountPoint)
416 token, err := runner.ContainerToken()
418 return fmt.Errorf("could not get container token: %s", err)
421 runner.ArvMount, err = runner.RunArvMount(arvMountCmd, token)
423 return fmt.Errorf("While trying to start arv-mount: %v", err)
426 for _, p := range collectionPaths {
429 return fmt.Errorf("While checking that input files exist: %v", err)
436 func (runner *ContainerRunner) ProcessDockerAttach(containerReader io.Reader) {
437 // Handle docker log protocol
438 // https://docs.docker.com/engine/reference/api/docker_remote_api_v1.15/#attach-to-a-container
440 header := make([]byte, 8)
442 _, readerr := io.ReadAtLeast(containerReader, header, 8)
445 readsize := int64(header[7]) | (int64(header[6]) << 8) | (int64(header[5]) << 16) | (int64(header[4]) << 24)
448 _, readerr = io.CopyN(runner.Stdout, containerReader, readsize)
451 _, readerr = io.CopyN(runner.Stderr, containerReader, readsize)
456 if readerr != io.EOF {
457 runner.CrunchLog.Printf("While reading docker logs: %v", readerr)
460 closeerr := runner.Stdout.Close()
462 runner.CrunchLog.Printf("While closing stdout logs: %v", closeerr)
465 closeerr = runner.Stderr.Close()
467 runner.CrunchLog.Printf("While closing stderr logs: %v", closeerr)
470 if runner.statReporter != nil {
471 runner.statReporter.Stop()
472 closeerr = runner.statLogger.Close()
474 runner.CrunchLog.Printf("While closing crunchstat logs: %v", closeerr)
478 runner.loggingDone <- true
479 close(runner.loggingDone)
485 func (runner *ContainerRunner) StartCrunchstat() {
486 runner.statLogger = NewThrottledLogger(runner.NewLogWriter("crunchstat"))
487 runner.statReporter = &crunchstat.Reporter{
488 CID: runner.ContainerID,
489 Logger: log.New(runner.statLogger, "", 0),
490 CgroupParent: runner.expectCgroupParent,
491 CgroupRoot: runner.cgroupRoot,
492 PollPeriod: runner.statInterval,
494 runner.statReporter.Start()
497 // AttachLogs connects the docker container stdout and stderr logs to the
498 // Arvados logger which logs to Keep and the API server logs table.
499 func (runner *ContainerRunner) AttachStreams() (err error) {
501 runner.CrunchLog.Print("Attaching container streams")
503 var containerReader io.Reader
504 containerReader, err = runner.Docker.AttachContainer(runner.ContainerID,
505 &dockerclient.AttachOptions{Stream: true, Stdout: true, Stderr: true})
507 return fmt.Errorf("While attaching container stdout/stderr streams: %v", err)
510 runner.loggingDone = make(chan bool)
512 if stdoutMnt, ok := runner.Container.Mounts["stdout"]; ok {
513 stdoutPath := stdoutMnt.Path[len(runner.Container.OutputPath):]
514 index := strings.LastIndex(stdoutPath, "/")
516 subdirs := stdoutPath[:index]
518 st, err := os.Stat(runner.HostOutputDir)
520 return fmt.Errorf("While Stat on temp dir: %v", err)
522 stdoutPath := path.Join(runner.HostOutputDir, subdirs)
523 err = os.MkdirAll(stdoutPath, st.Mode()|os.ModeSetgid|0777)
525 return fmt.Errorf("While MkdirAll %q: %v", stdoutPath, err)
529 stdoutFile, err := os.Create(path.Join(runner.HostOutputDir, stdoutPath))
531 return fmt.Errorf("While creating stdout file: %v", err)
533 runner.Stdout = stdoutFile
535 runner.Stdout = NewThrottledLogger(runner.NewLogWriter("stdout"))
537 runner.Stderr = NewThrottledLogger(runner.NewLogWriter("stderr"))
539 go runner.ProcessDockerAttach(containerReader)
544 // CreateContainer creates the docker container.
545 func (runner *ContainerRunner) CreateContainer() error {
546 runner.CrunchLog.Print("Creating Docker container")
548 runner.ContainerConfig.Cmd = runner.Container.Command
549 if runner.Container.Cwd != "." {
550 runner.ContainerConfig.WorkingDir = runner.Container.Cwd
553 for k, v := range runner.Container.Environment {
554 runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v)
556 if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
557 tok, err := runner.ContainerToken()
561 runner.ContainerConfig.Env = append(runner.ContainerConfig.Env,
562 "ARVADOS_API_TOKEN="+tok,
563 "ARVADOS_API_HOST="+os.Getenv("ARVADOS_API_HOST"),
564 "ARVADOS_API_HOST_INSECURE="+os.Getenv("ARVADOS_API_HOST_INSECURE"),
566 runner.ContainerConfig.NetworkDisabled = false
568 runner.ContainerConfig.NetworkDisabled = true
572 runner.ContainerID, err = runner.Docker.CreateContainer(&runner.ContainerConfig, "", nil)
574 return fmt.Errorf("While creating container: %v", err)
577 runner.HostConfig = dockerclient.HostConfig{
579 CgroupParent: runner.setCgroupParent,
580 LogConfig: dockerclient.LogConfig{
585 return runner.AttachStreams()
588 // StartContainer starts the docker container created by CreateContainer.
589 func (runner *ContainerRunner) StartContainer() error {
590 runner.CrunchLog.Printf("Starting Docker container id '%s'", runner.ContainerID)
591 err := runner.Docker.StartContainer(runner.ContainerID, &runner.HostConfig)
593 return fmt.Errorf("could not start container: %v", err)
598 // WaitFinish waits for the container to terminate, capture the exit code, and
599 // close the stdout/stderr logging.
600 func (runner *ContainerRunner) WaitFinish() error {
601 runner.CrunchLog.Print("Waiting for container to finish")
603 result := runner.Docker.Wait(runner.ContainerID)
606 return fmt.Errorf("While waiting for container to finish: %v", wr.Error)
608 runner.ExitCode = &wr.ExitCode
610 // wait for stdout/stderr to complete
616 // HandleOutput sets the output, unmounts the FUSE mount, and deletes temporary directories
617 func (runner *ContainerRunner) CaptureOutput() error {
618 if runner.finalState != "Complete" {
622 if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
623 // Output may have been set directly by the container, so
624 // refresh the container record to check.
625 err := runner.ArvClient.Get("containers", runner.Container.UUID,
626 nil, &runner.Container)
630 if runner.Container.Output != "" {
631 // Container output is already set.
632 runner.OutputPDH = &runner.Container.Output
637 if runner.HostOutputDir == "" {
641 _, err := os.Stat(runner.HostOutputDir)
643 return fmt.Errorf("While checking host output path: %v", err)
646 var manifestText string
648 collectionMetafile := fmt.Sprintf("%s/.arvados#collection", runner.HostOutputDir)
649 _, err = os.Stat(collectionMetafile)
652 cw := CollectionWriter{0, runner.Kc, nil, nil, sync.Mutex{}}
653 manifestText, err = cw.WriteTree(runner.HostOutputDir, runner.CrunchLog.Logger)
655 return fmt.Errorf("While uploading output files: %v", err)
658 // FUSE mount directory
659 file, openerr := os.Open(collectionMetafile)
661 return fmt.Errorf("While opening FUSE metafile: %v", err)
665 var rec arvados.Collection
666 err = json.NewDecoder(file).Decode(&rec)
668 return fmt.Errorf("While reading FUSE metafile: %v", err)
670 manifestText = rec.ManifestText
673 // Pre-populate output from the configured mount points
675 for bind, _ := range runner.Container.Mounts {
676 binds = append(binds, bind)
680 for _, bind := range binds {
681 mnt := runner.Container.Mounts[bind]
683 bindSuffix := strings.TrimPrefix(bind, runner.Container.OutputPath)
685 if bindSuffix == bind || len(bindSuffix) <= 0 {
686 // either does not start with OutputPath or is OutputPath itself
690 if mnt.ExcludeFromOutput == true {
694 // append to manifest_text
695 m, err := runner.getCollectionManifestForPath(mnt, bindSuffix)
700 manifestText = manifestText + m
704 var response arvados.Collection
705 manifest := manifest.Manifest{Text: manifestText}
706 manifestText = manifest.Extract(".", ".").Text
707 err = runner.ArvClient.Create("collections",
709 "collection": arvadosclient.Dict{
711 "name": "output for " + runner.Container.UUID,
712 "manifest_text": manifestText}},
715 return fmt.Errorf("While creating output collection: %v", err)
717 runner.OutputPDH = &response.PortableDataHash
721 var outputCollections = make(map[string]arvados.Collection)
723 // Fetch the collection for the mnt.PortableDataHash
724 // Return the manifest_text fragment corresponding to the specified mnt.Path
725 // after making any required updates.
727 // If mnt.Path is not specified,
728 // return the entire manifest_text after replacing any "." with bindSuffix
729 // If mnt.Path corresponds to one stream,
730 // return the manifest_text for that stream after replacing that stream name with bindSuffix
731 // Otherwise, check if a filename in any one stream is being sought. Return the manifest_text
732 // for that stream after replacing stream name with bindSuffix minus the last word
733 // and the file name with last word of the bindSuffix
734 // Allowed path examples:
737 // "path":"/subdir1/subdir2"
738 // "path":"/subdir/filename" etc
739 func (runner *ContainerRunner) getCollectionManifestForPath(mnt arvados.Mount, bindSuffix string) (string, error) {
740 collection := outputCollections[mnt.PortableDataHash]
741 if collection.PortableDataHash == "" {
742 err := runner.ArvClient.Get("collections", mnt.PortableDataHash, nil, &collection)
744 return "", fmt.Errorf("While getting collection for %v: %v", mnt.PortableDataHash, err)
746 outputCollections[mnt.PortableDataHash] = collection
749 if collection.ManifestText == "" {
750 runner.CrunchLog.Printf("No manifest text for collection %v", collection.PortableDataHash)
754 mft := manifest.Manifest{Text: collection.ManifestText}
755 extracted := mft.Extract(mnt.Path, bindSuffix)
756 if extracted.Err != nil {
757 return "", fmt.Errorf("Error parsing manifest for %v: %v", mnt.PortableDataHash, extracted.Err.Error())
759 return extracted.Text, nil
762 func (runner *ContainerRunner) CleanupDirs() {
763 if runner.ArvMount != nil {
764 umount := exec.Command("fusermount", "-z", "-u", runner.ArvMountPoint)
765 umnterr := umount.Run()
767 runner.CrunchLog.Printf("While running fusermount: %v", umnterr)
770 mnterr := <-runner.ArvMountExit
772 runner.CrunchLog.Printf("Arv-mount exit error: %v", mnterr)
776 for _, tmpdir := range runner.CleanupTempDir {
777 rmerr := os.RemoveAll(tmpdir)
779 runner.CrunchLog.Printf("While cleaning up temporary directory %s: %v", tmpdir, rmerr)
784 // CommitLogs posts the collection containing the final container logs.
785 func (runner *ContainerRunner) CommitLogs() error {
786 runner.CrunchLog.Print(runner.finalState)
787 runner.CrunchLog.Close()
789 // Closing CrunchLog above allows it to be committed to Keep at this
790 // point, but re-open crunch log with ArvClient in case there are any
791 // other further (such as failing to write the log to Keep!) while
793 runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{runner.ArvClient, runner.Container.UUID,
796 if runner.LogsPDH != nil {
797 // If we have already assigned something to LogsPDH,
798 // we must be closing the re-opened log, which won't
799 // end up getting attached to the container record and
800 // therefore doesn't need to be saved as a collection
801 // -- it exists only to send logs to other channels.
805 mt, err := runner.LogCollection.ManifestText()
807 return fmt.Errorf("While creating log manifest: %v", err)
810 var response arvados.Collection
811 err = runner.ArvClient.Create("collections",
813 "collection": arvadosclient.Dict{
815 "name": "logs for " + runner.Container.UUID,
816 "manifest_text": mt}},
819 return fmt.Errorf("While creating log collection: %v", err)
821 runner.LogsPDH = &response.PortableDataHash
825 // UpdateContainerRunning updates the container state to "Running"
826 func (runner *ContainerRunner) UpdateContainerRunning() error {
827 runner.CancelLock.Lock()
828 defer runner.CancelLock.Unlock()
829 if runner.Cancelled {
832 return runner.ArvClient.Update("containers", runner.Container.UUID,
833 arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running"}}, nil)
836 // ContainerToken returns the api_token the container (and any
837 // arv-mount processes) are allowed to use.
838 func (runner *ContainerRunner) ContainerToken() (string, error) {
839 if runner.token != "" {
840 return runner.token, nil
843 var auth arvados.APIClientAuthorization
844 err := runner.ArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth)
848 runner.token = auth.APIToken
849 return runner.token, nil
852 // UpdateContainerComplete updates the container record state on API
853 // server to "Complete" or "Cancelled"
854 func (runner *ContainerRunner) UpdateContainerFinal() error {
855 update := arvadosclient.Dict{}
856 update["state"] = runner.finalState
857 if runner.LogsPDH != nil {
858 update["log"] = *runner.LogsPDH
860 if runner.finalState == "Complete" {
861 if runner.ExitCode != nil {
862 update["exit_code"] = *runner.ExitCode
864 if runner.OutputPDH != nil {
865 update["output"] = *runner.OutputPDH
868 return runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil)
871 // IsCancelled returns the value of Cancelled, with goroutine safety.
872 func (runner *ContainerRunner) IsCancelled() bool {
873 runner.CancelLock.Lock()
874 defer runner.CancelLock.Unlock()
875 return runner.Cancelled
878 // NewArvLogWriter creates an ArvLogWriter
879 func (runner *ContainerRunner) NewArvLogWriter(name string) io.WriteCloser {
880 return &ArvLogWriter{runner.ArvClient, runner.Container.UUID, name, runner.LogCollection.Open(name + ".txt")}
883 // Run the full container lifecycle.
884 func (runner *ContainerRunner) Run() (err error) {
885 runner.CrunchLog.Printf("Executing container '%s'", runner.Container.UUID)
887 hostname, hosterr := os.Hostname()
889 runner.CrunchLog.Printf("Error getting hostname '%v'", hosterr)
891 runner.CrunchLog.Printf("Executing on host '%s'", hostname)
894 // Clean up temporary directories _after_ finalizing
895 // everything (if we've made any by then)
896 defer runner.CleanupDirs()
898 runner.finalState = "Queued"
901 // checkErr prints e (unless it's nil) and sets err to
902 // e (unless err is already non-nil). Thus, if err
903 // hasn't already been assigned when Run() returns,
904 // this cleanup func will cause Run() to return the
905 // first non-nil error that is passed to checkErr().
906 checkErr := func(e error) {
910 runner.CrunchLog.Print(e)
916 // Log the error encountered in Run(), if any
919 if runner.finalState == "Queued" {
920 runner.CrunchLog.Close()
921 runner.UpdateContainerFinal()
925 if runner.IsCancelled() {
926 runner.finalState = "Cancelled"
927 // but don't return yet -- we still want to
928 // capture partial output and write logs
931 checkErr(runner.CaptureOutput())
932 checkErr(runner.CommitLogs())
933 checkErr(runner.UpdateContainerFinal())
935 // The real log is already closed, but then we opened
936 // a new one in case we needed to log anything while
938 runner.CrunchLog.Close()
941 err = runner.ArvClient.Get("containers", runner.Container.UUID, nil, &runner.Container)
943 err = fmt.Errorf("While getting container record: %v", err)
947 // setup signal handling
948 runner.SetupSignals()
950 // check for and/or load image
951 err = runner.LoadImage()
953 runner.finalState = "Cancelled"
954 err = fmt.Errorf("While loading container image: %v", err)
958 // set up FUSE mount and binds
959 err = runner.SetupMounts()
961 runner.finalState = "Cancelled"
962 err = fmt.Errorf("While setting up mounts: %v", err)
966 err = runner.CreateContainer()
971 runner.StartCrunchstat()
973 if runner.IsCancelled() {
977 err = runner.UpdateContainerRunning()
981 runner.finalState = "Cancelled"
983 err = runner.StartContainer()
988 err = runner.WaitFinish()
990 runner.finalState = "Complete"
995 // NewContainerRunner creates a new container runner.
996 func NewContainerRunner(api IArvadosClient,
998 docker ThinDockerClient,
999 containerUUID string) *ContainerRunner {
1001 cr := &ContainerRunner{ArvClient: api, Kc: kc, Docker: docker}
1002 cr.NewLogWriter = cr.NewArvLogWriter
1003 cr.RunArvMount = cr.ArvMountCmd
1004 cr.MkTempDir = ioutil.TempDir
1005 cr.LogCollection = &CollectionWriter{0, kc, nil, nil, sync.Mutex{}}
1006 cr.Container.UUID = containerUUID
1007 cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run"))
1008 cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0)
1013 statInterval := flag.Duration("crunchstat-interval", 10*time.Second, "sampling period for periodic resource usage reporting")
1014 cgroupRoot := flag.String("cgroup-root", "/sys/fs/cgroup", "path to sysfs cgroup tree")
1015 cgroupParent := flag.String("cgroup-parent", "docker", "name of container's parent cgroup (ignored if -cgroup-parent-subsystem is used)")
1016 cgroupParentSubsystem := flag.String("cgroup-parent-subsystem", "", "use current cgroup for given subsystem as parent cgroup for container")
1017 caCertsPath := flag.String("ca-certs", "", "Path to TLS root certificates")
1020 containerId := flag.Arg(0)
1022 if *caCertsPath != "" {
1023 arvadosclient.CertFiles = []string{*caCertsPath}
1026 api, err := arvadosclient.MakeArvadosClient()
1028 log.Fatalf("%s: %v", containerId, err)
1032 var kc *keepclient.KeepClient
1033 kc, err = keepclient.MakeKeepClient(api)
1035 log.Fatalf("%s: %v", containerId, err)
1039 var docker *dockerclient.DockerClient
1040 docker, err = dockerclient.NewDockerClient("unix:///var/run/docker.sock", nil)
1042 log.Fatalf("%s: %v", containerId, err)
1045 cr := NewContainerRunner(api, kc, docker, containerId)
1046 cr.statInterval = *statInterval
1047 cr.cgroupRoot = *cgroupRoot
1048 cr.expectCgroupParent = *cgroupParent
1049 if *cgroupParentSubsystem != "" {
1050 p := findCgroup(*cgroupParentSubsystem)
1051 cr.setCgroupParent = p
1052 cr.expectCgroupParent = p
1057 log.Fatalf("%s: %v", containerId, err)