services/crunch-run/crunchrun.go

   1 package main
   2
   3 import (
   4         "encoding/json"
   5         "errors"
   6         "flag"
   7         "fmt"
   8         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
   9         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  10         "git.curoverse.com/arvados.git/sdk/go/manifest"
  11         "github.com/curoverse/dockerclient"
  12         "io"
  13         "io/ioutil"
  14         "log"
  15         "os"
  16         "os/exec"
  17         "os/signal"
  18         "strings"
  19         "sync"
  20         "syscall"
  21         "time"
  22 )
  23
  24 // IArvadosClient is the minimal Arvados API methods used by crunch-run.
  25 type IArvadosClient interface {
  26         Create(resourceType string, parameters arvadosclient.Dict, output interface{}) error
  27         Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
  28         Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error)
  29 }
  30
  31 // ErrCancelled is the error returned when the container is cancelled.
  32 var ErrCancelled = errors.New("Cancelled")
  33
  34 // IKeepClient is the minimal Keep API methods used by crunch-run.
  35 type IKeepClient interface {
  36         PutHB(hash string, buf []byte) (string, int, error)
  37         ManifestFileReader(m manifest.Manifest, filename string) (keepclient.ReadCloserWithLen, error)
  38 }
  39
  40 // Mount describes the mount points to create inside the container.
  41 type Mount struct {
  42         Kind             string `json:"kind"`
  43         Writable         bool   `json:"writable"`
  44         PortableDataHash string `json:"portable_data_hash"`
  45         UUID             string `json:"uuid"`
  46         DeviceType       string `json:"device_type"`
  47 }
  48
  49 // Collection record returned by the API server.
  50 type CollectionRecord struct {
  51         ManifestText     string `json:"manifest_text"`
  52         PortableDataHash string `json:"portable_data_hash"`
  53 }
  54
  55 // ContainerRecord is the container record returned by the API server.
  56 type ContainerRecord struct {
  57         UUID               string                 `json:"uuid"`
  58         Command            []string               `json:"command"`
  59         ContainerImage     string                 `json:"container_image"`
  60         Cwd                string                 `json:"cwd"`
  61         Environment        map[string]string      `json:"environment"`
  62         Mounts             map[string]Mount       `json:"mounts"`
  63         OutputPath         string                 `json:"output_path"`
  64         Priority           int                    `json:"priority"`
  65         RuntimeConstraints map[string]interface{} `json:"runtime_constraints"`
  66         State              string                 `json:"state"`
  67         Output             string                 `json:"output"`
  68 }
  69
  70 // NewLogWriter is a factory function to create a new log writer.
  71 type NewLogWriter func(name string) io.WriteCloser
  72
  73 type RunArvMount func([]string) (*exec.Cmd, error)
  74
  75 type MkTempDir func(string, string) (string, error)
  76
  77 // ThinDockerClient is the minimal Docker client interface used by crunch-run.
  78 type ThinDockerClient interface {
  79         StopContainer(id string, timeout int) error
  80         InspectImage(id string) (*dockerclient.ImageInfo, error)
  81         LoadImage(reader io.Reader) error
  82         CreateContainer(config *dockerclient.ContainerConfig, name string, authConfig *dockerclient.AuthConfig) (string, error)
  83         StartContainer(id string, config *dockerclient.HostConfig) error
  84         AttachContainer(id string, options *dockerclient.AttachOptions) (io.ReadCloser, error)
  85         Wait(id string) <-chan dockerclient.WaitResult
  86         RemoveImage(name string, force bool) ([]*dockerclient.ImageDelete, error)
  87 }
  88
  89 // ContainerRunner is the main stateful struct used for a single execution of a
  90 // container.
  91 type ContainerRunner struct {
  92         Docker    ThinDockerClient
  93         ArvClient IArvadosClient
  94         Kc        IKeepClient
  95         ContainerRecord
  96         dockerclient.ContainerConfig
  97         ContainerID string
  98         ExitCode    *int
  99         NewLogWriter
 100         loggingDone   chan bool
 101         CrunchLog     *ThrottledLogger
 102         Stdout        *ThrottledLogger
 103         Stderr        *ThrottledLogger
 104         LogCollection *CollectionWriter
 105         LogsPDH       *string
 106         RunArvMount
 107         MkTempDir
 108         ArvMount       *exec.Cmd
 109         ArvMountPoint  string
 110         HostOutputDir  string
 111         CleanupTempDir []string
 112         Binds          []string
 113         OutputPDH      *string
 114         CancelLock     sync.Mutex
 115         Cancelled      bool
 116         SigChan        chan os.Signal
 117         ArvMountExit   chan error
 118         finalState     string
 119 }
 120
 121 // SetupSignals sets up signal handling to gracefully terminate the underlying
 122 // Docker container and update state when receiving a TERM, INT or QUIT signal.
 123 func (runner *ContainerRunner) SetupSignals() error {
 124         runner.SigChan = make(chan os.Signal, 1)
 125         signal.Notify(runner.SigChan, syscall.SIGTERM)
 126         signal.Notify(runner.SigChan, syscall.SIGINT)
 127         signal.Notify(runner.SigChan, syscall.SIGQUIT)
 128
 129         go func(sig <-chan os.Signal) {
 130                 for _ = range sig {
 131                         if !runner.Cancelled {
 132                                 runner.CancelLock.Lock()
 133                                 runner.Cancelled = true
 134                                 if runner.ContainerID != "" {
 135                                         runner.Docker.StopContainer(runner.ContainerID, 10)
 136                                 }
 137                                 runner.CancelLock.Unlock()
 138                         }
 139                 }
 140         }(runner.SigChan)
 141
 142         return nil
 143 }
 144
 145 // LoadImage determines the docker image id from the container record and
 146 // checks if it is available in the local Docker image store.  If not, it loads
 147 // the image from Keep.
 148 func (runner *ContainerRunner) LoadImage() (err error) {
 149
 150         runner.CrunchLog.Printf("Fetching Docker image from collection '%s'", runner.ContainerRecord.ContainerImage)
 151
 152         var collection CollectionRecord
 153         err = runner.ArvClient.Get("collections", runner.ContainerRecord.ContainerImage, nil, &collection)
 154         if err != nil {
 155                 return fmt.Errorf("While getting container image collection: %v", err)
 156         }
 157         manifest := manifest.Manifest{Text: collection.ManifestText}
 158         var img, imageID string
 159         for ms := range manifest.StreamIter() {
 160                 img = ms.FileStreamSegments[0].Name
 161                 if !strings.HasSuffix(img, ".tar") {
 162                         return fmt.Errorf("First file in the container image collection does not end in .tar")
 163                 }
 164                 imageID = img[:len(img)-4]
 165         }
 166
 167         runner.CrunchLog.Printf("Using Docker image id '%s'", imageID)
 168
 169         _, err = runner.Docker.InspectImage(imageID)
 170         if err != nil {
 171                 runner.CrunchLog.Print("Loading Docker image from keep")
 172
 173                 var readCloser io.ReadCloser
 174                 readCloser, err = runner.Kc.ManifestFileReader(manifest, img)
 175                 if err != nil {
 176                         return fmt.Errorf("While creating ManifestFileReader for container image: %v", err)
 177                 }
 178
 179                 err = runner.Docker.LoadImage(readCloser)
 180                 if err != nil {
 181                         return fmt.Errorf("While loading container image into Docker: %v", err)
 182                 }
 183         } else {
 184                 runner.CrunchLog.Print("Docker image is available")
 185         }
 186
 187         runner.ContainerConfig.Image = imageID
 188
 189         return nil
 190 }
 191
 192 func (runner *ContainerRunner) ArvMountCmd(arvMountCmd []string) (c *exec.Cmd, err error) {
 193         c = exec.Command("arv-mount", arvMountCmd...)
 194         nt := NewThrottledLogger(runner.NewLogWriter("arv-mount"))
 195         c.Stdout = nt
 196         c.Stderr = nt
 197
 198         err = c.Start()
 199         if err != nil {
 200                 return nil, err
 201         }
 202
 203         statReadme := make(chan bool)
 204         runner.ArvMountExit = make(chan error)
 205
 206         keepStatting := true
 207         go func() {
 208                 for keepStatting {
 209                         time.Sleep(100 * time.Millisecond)
 210                         _, err = os.Stat(fmt.Sprintf("%s/by_id/README", runner.ArvMountPoint))
 211                         if err == nil {
 212                                 keepStatting = false
 213                                 statReadme <- true
 214                         }
 215                 }
 216                 close(statReadme)
 217         }()
 218
 219         go func() {
 220                 runner.ArvMountExit <- c.Wait()
 221                 close(runner.ArvMountExit)
 222         }()
 223
 224         select {
 225         case <-statReadme:
 226                 break
 227         case err := <-runner.ArvMountExit:
 228                 runner.ArvMount = nil
 229                 keepStatting = false
 230                 return nil, err
 231         }
 232
 233         return c, nil
 234 }
 235
 236 func (runner *ContainerRunner) SetupMounts() (err error) {
 237         runner.ArvMountPoint, err = runner.MkTempDir("", "keep")
 238         if err != nil {
 239                 return fmt.Errorf("While creating keep mount temp dir: %v", err)
 240         }
 241
 242         runner.CleanupTempDir = append(runner.CleanupTempDir, runner.ArvMountPoint)
 243
 244         pdhOnly := true
 245         tmpcount := 0
 246         arvMountCmd := []string{"--foreground", "--allow-other", "--read-write"}
 247         collectionPaths := []string{}
 248         runner.Binds = nil
 249
 250         for bind, mnt := range runner.ContainerRecord.Mounts {
 251                 if mnt.Kind == "collection" {
 252                         var src string
 253                         if mnt.UUID != "" && mnt.PortableDataHash != "" {
 254                                 return fmt.Errorf("Cannot specify both 'uuid' and 'portable_data_hash' for a collection mount")
 255                         }
 256                         if mnt.UUID != "" {
 257                                 if mnt.Writable {
 258                                         return fmt.Errorf("Writing to existing collections currently not permitted.")
 259                                 }
 260                                 pdhOnly = false
 261                                 src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.UUID)
 262                         } else if mnt.PortableDataHash != "" {
 263                                 if mnt.Writable {
 264                                         return fmt.Errorf("Can never write to a collection specified by portable data hash")
 265                                 }
 266                                 src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.PortableDataHash)
 267                         } else {
 268                                 src = fmt.Sprintf("%s/tmp%d", runner.ArvMountPoint, tmpcount)
 269                                 arvMountCmd = append(arvMountCmd, "--mount-tmp")
 270                                 arvMountCmd = append(arvMountCmd, fmt.Sprintf("tmp%d", tmpcount))
 271                                 tmpcount += 1
 272                         }
 273                         if mnt.Writable {
 274                                 if bind == runner.ContainerRecord.OutputPath {
 275                                         runner.HostOutputDir = src
 276                                 }
 277                                 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind))
 278                         } else {
 279                                 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", src, bind))
 280                         }
 281                         collectionPaths = append(collectionPaths, src)
 282                 } else if mnt.Kind == "tmp" {
 283                         if bind == runner.ContainerRecord.OutputPath {
 284                                 runner.HostOutputDir, err = runner.MkTempDir("", "")
 285                                 if err != nil {
 286                                         return fmt.Errorf("While creating mount temp dir: %v", err)
 287                                 }
 288                                 st, staterr := os.Stat(runner.HostOutputDir)
 289                                 if staterr != nil {
 290                                         return fmt.Errorf("While Stat on temp dir: %v", staterr)
 291                                 }
 292                                 err = os.Chmod(runner.HostOutputDir, st.Mode()|os.ModeSetgid|0777)
 293                                 if staterr != nil {
 294                                         return fmt.Errorf("While Chmod temp dir: %v", err)
 295                                 }
 296                                 runner.CleanupTempDir = append(runner.CleanupTempDir, runner.HostOutputDir)
 297                                 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", runner.HostOutputDir, bind))
 298                         } else {
 299                                 runner.Binds = append(runner.Binds, bind)
 300                         }
 301                 } else {
 302                         return fmt.Errorf("Unknown mount kind '%s'", mnt.Kind)
 303                 }
 304         }
 305
 306         if runner.HostOutputDir == "" {
 307                 return fmt.Errorf("Output path does not correspond to a writable mount point")
 308         }
 309
 310         if pdhOnly {
 311                 arvMountCmd = append(arvMountCmd, "--mount-by-pdh", "by_id")
 312         } else {
 313                 arvMountCmd = append(arvMountCmd, "--mount-by-id", "by_id")
 314         }
 315         arvMountCmd = append(arvMountCmd, runner.ArvMountPoint)
 316
 317         runner.ArvMount, err = runner.RunArvMount(arvMountCmd)
 318         if err != nil {
 319                 return fmt.Errorf("While trying to start arv-mount: %v", err)
 320         }
 321
 322         for _, p := range collectionPaths {
 323                 _, err = os.Stat(p)
 324                 if err != nil {
 325                         return fmt.Errorf("While checking that input files exist: %v", err)
 326                 }
 327         }
 328
 329         return nil
 330 }
 331
 332 func (runner *ContainerRunner) ProcessDockerAttach(containerReader io.Reader) {
 333         // Handle docker log protocol
 334         // https://docs.docker.com/engine/reference/api/docker_remote_api_v1.15/#attach-to-a-container
 335
 336         header := make([]byte, 8)
 337         for {
 338                 _, readerr := io.ReadAtLeast(containerReader, header, 8)
 339
 340                 if readerr == nil {
 341                         readsize := int64(header[7]) | (int64(header[6]) << 8) | (int64(header[5]) << 16) | (int64(header[4]) << 24)
 342                         if header[0] == 1 {
 343                                 // stdout
 344                                 _, readerr = io.CopyN(runner.Stdout, containerReader, readsize)
 345                         } else {
 346                                 // stderr
 347                                 _, readerr = io.CopyN(runner.Stderr, containerReader, readsize)
 348                         }
 349                 }
 350
 351                 if readerr != nil {
 352                         if readerr != io.EOF {
 353                                 runner.CrunchLog.Printf("While reading docker logs: %v", readerr)
 354                         }
 355
 356                         closeerr := runner.Stdout.Close()
 357                         if closeerr != nil {
 358                                 runner.CrunchLog.Printf("While closing stdout logs: %v", readerr)
 359                         }
 360
 361                         closeerr = runner.Stderr.Close()
 362                         if closeerr != nil {
 363                                 runner.CrunchLog.Printf("While closing stderr logs: %v", readerr)
 364                         }
 365
 366                         runner.loggingDone <- true
 367                         close(runner.loggingDone)
 368                         return
 369                 }
 370         }
 371 }
 372
 373 // AttachLogs connects the docker container stdout and stderr logs to the
 374 // Arvados logger which logs to Keep and the API server logs table.
 375 func (runner *ContainerRunner) AttachStreams() (err error) {
 376
 377         runner.CrunchLog.Print("Attaching container streams")
 378
 379         var containerReader io.Reader
 380         containerReader, err = runner.Docker.AttachContainer(runner.ContainerID,
 381                 &dockerclient.AttachOptions{Stream: true, Stdout: true, Stderr: true})
 382         if err != nil {
 383                 return fmt.Errorf("While attaching container logs: %v", err)
 384         }
 385
 386         runner.loggingDone = make(chan bool)
 387
 388         runner.Stdout = NewThrottledLogger(runner.NewLogWriter("stdout"))
 389         runner.Stderr = NewThrottledLogger(runner.NewLogWriter("stderr"))
 390
 391         go runner.ProcessDockerAttach(containerReader)
 392
 393         return nil
 394 }
 395
 396 // StartContainer creates the container and runs it.
 397 func (runner *ContainerRunner) StartContainer() (err error) {
 398         runner.CrunchLog.Print("Creating Docker container")
 399
 400         runner.CancelLock.Lock()
 401         defer runner.CancelLock.Unlock()
 402
 403         if runner.Cancelled {
 404                 return ErrCancelled
 405         }
 406
 407         runner.ContainerConfig.Cmd = runner.ContainerRecord.Command
 408         if runner.ContainerRecord.Cwd != "." {
 409                 runner.ContainerConfig.WorkingDir = runner.ContainerRecord.Cwd
 410         }
 411         for k, v := range runner.ContainerRecord.Environment {
 412                 runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v)
 413         }
 414         runner.ContainerConfig.NetworkDisabled = true
 415         runner.ContainerID, err = runner.Docker.CreateContainer(&runner.ContainerConfig, "", nil)
 416         if err != nil {
 417                 return fmt.Errorf("While creating container: %v", err)
 418         }
 419         hostConfig := &dockerclient.HostConfig{Binds: runner.Binds,
 420                 LogConfig: dockerclient.LogConfig{Type: "none"}}
 421
 422         runner.AttachStreams()
 423         if err != nil {
 424                 return fmt.Errorf("While attaching streams: %v", err)
 425                 return err
 426         }
 427
 428         runner.CrunchLog.Printf("Starting Docker container id '%s'", runner.ContainerID)
 429         err = runner.Docker.StartContainer(runner.ContainerID, hostConfig)
 430         if err != nil {
 431                 return fmt.Errorf("While starting container: %v", err)
 432         }
 433
 434         return nil
 435 }
 436
 437 // WaitFinish waits for the container to terminate, capture the exit code, and
 438 // close the stdout/stderr logging.
 439 func (runner *ContainerRunner) WaitFinish() error {
 440         runner.CrunchLog.Print("Waiting for container to finish")
 441
 442         result := runner.Docker.Wait(runner.ContainerID)
 443         wr := <-result
 444         if wr.Error != nil {
 445                 return fmt.Errorf("While waiting for container to finish: %v", wr.Error)
 446         }
 447         runner.ExitCode = &wr.ExitCode
 448
 449         // wait for stdout/stderr to complete
 450         <-runner.loggingDone
 451
 452         return nil
 453 }
 454
 455 // HandleOutput sets the output, unmounts the FUSE mount, and deletes temporary directories
 456 func (runner *ContainerRunner) CaptureOutput() error {
 457         if runner.finalState != "Complete" {
 458                 return nil
 459         }
 460
 461         if runner.HostOutputDir == "" {
 462                 return nil
 463         }
 464
 465         _, err := os.Stat(runner.HostOutputDir)
 466         if err != nil {
 467                 return fmt.Errorf("While checking host output path: %v", err)
 468         }
 469
 470         var manifestText string
 471
 472         collectionMetafile := fmt.Sprintf("%s/.arvados#collection", runner.HostOutputDir)
 473         _, err = os.Stat(collectionMetafile)
 474         if err != nil {
 475                 // Regular directory
 476                 cw := CollectionWriter{runner.Kc, nil, sync.Mutex{}}
 477                 manifestText, err = cw.WriteTree(runner.HostOutputDir, runner.CrunchLog.Logger)
 478                 if err != nil {
 479                         return fmt.Errorf("While uploading output files: %v", err)
 480                 }
 481         } else {
 482                 // FUSE mount directory
 483                 file, openerr := os.Open(collectionMetafile)
 484                 if openerr != nil {
 485                         return fmt.Errorf("While opening FUSE metafile: %v", err)
 486                 }
 487                 defer file.Close()
 488
 489                 rec := CollectionRecord{}
 490                 err = json.NewDecoder(file).Decode(&rec)
 491                 if err != nil {
 492                         return fmt.Errorf("While reading FUSE metafile: %v", err)
 493                 }
 494                 manifestText = rec.ManifestText
 495         }
 496
 497         var response CollectionRecord
 498         err = runner.ArvClient.Create("collections",
 499                 arvadosclient.Dict{
 500                         "collection": arvadosclient.Dict{
 501                                 "manifest_text": manifestText}},
 502                 &response)
 503         if err != nil {
 504                 return fmt.Errorf("While creating output collection: %v", err)
 505         }
 506
 507         runner.OutputPDH = new(string)
 508         *runner.OutputPDH = response.PortableDataHash
 509
 510         return nil
 511 }
 512
 513 func (runner *ContainerRunner) CleanupDirs() {
 514         if runner.ArvMount != nil {
 515                 umount := exec.Command("fusermount", "-z", "-u", runner.ArvMountPoint)
 516                 umnterr := umount.Run()
 517                 if umnterr != nil {
 518                         runner.CrunchLog.Printf("While running fusermount: %v", umnterr)
 519                 }
 520
 521                 mnterr := <-runner.ArvMountExit
 522                 if mnterr != nil {
 523                         runner.CrunchLog.Printf("Arv-mount exit error: %v", mnterr)
 524                 }
 525         }
 526
 527         for _, tmpdir := range runner.CleanupTempDir {
 528                 rmerr := os.RemoveAll(tmpdir)
 529                 if rmerr != nil {
 530                         runner.CrunchLog.Printf("While cleaning up temporary directory %s: %v", tmpdir, rmerr)
 531                 }
 532         }
 533 }
 534
 535 // CommitLogs posts the collection containing the final container logs.
 536 func (runner *ContainerRunner) CommitLogs() error {
 537         runner.CrunchLog.Print(runner.finalState)
 538         runner.CrunchLog.Close()
 539
 540         // Closing CrunchLog above allows it to be committed to Keep at this
 541         // point, but re-open crunch log with ArvClient in case there are any
 542         // other further (such as failing to write the log to Keep!) while
 543         // shutting down
 544         runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{runner.ArvClient, runner.ContainerRecord.UUID,
 545                 "crunch-run", nil})
 546
 547         mt, err := runner.LogCollection.ManifestText()
 548         if err != nil {
 549                 return fmt.Errorf("While creating log manifest: %v", err)
 550         }
 551
 552         var response CollectionRecord
 553         err = runner.ArvClient.Create("collections",
 554                 arvadosclient.Dict{
 555                         "collection": arvadosclient.Dict{
 556                                 "name":          "logs for " + runner.ContainerRecord.UUID,
 557                                 "manifest_text": mt}},
 558                 &response)
 559         if err != nil {
 560                 return fmt.Errorf("While creating log collection: %v", err)
 561         }
 562
 563         runner.LogsPDH = new(string)
 564         *runner.LogsPDH = response.PortableDataHash
 565
 566         return nil
 567 }
 568
 569 // UpdateContainerRecordRunning updates the container state to "Running"
 570 func (runner *ContainerRunner) UpdateContainerRecordRunning() error {
 571         return runner.ArvClient.Update("containers", runner.ContainerRecord.UUID,
 572                 arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running"}}, nil)
 573 }
 574
 575 // UpdateContainerRecordComplete updates the container record state on API
 576 // server to "Complete" or "Cancelled"
 577 func (runner *ContainerRunner) UpdateContainerRecordComplete() error {
 578         update := arvadosclient.Dict{}
 579         if runner.LogsPDH != nil {
 580                 update["log"] = *runner.LogsPDH
 581         }
 582         if runner.ExitCode != nil {
 583                 update["exit_code"] = *runner.ExitCode
 584         }
 585         if runner.OutputPDH != nil {
 586                 update["output"] = runner.OutputPDH
 587         }
 588
 589         update["state"] = runner.finalState
 590
 591         return runner.ArvClient.Update("containers", runner.ContainerRecord.UUID, arvadosclient.Dict{"container": update}, nil)
 592 }
 593
 594 // NewArvLogWriter creates an ArvLogWriter
 595 func (runner *ContainerRunner) NewArvLogWriter(name string) io.WriteCloser {
 596         return &ArvLogWriter{runner.ArvClient, runner.ContainerRecord.UUID, name, runner.LogCollection.Open(name + ".txt")}
 597 }
 598
 599 // Run the full container lifecycle.
 600 func (runner *ContainerRunner) Run() (err error) {
 601         runner.CrunchLog.Printf("Executing container '%s'", runner.ContainerRecord.UUID)
 602
 603         hostname, hosterr := os.Hostname()
 604         if hosterr != nil {
 605                 runner.CrunchLog.Printf("Error getting hostname '%v'", hosterr)
 606         } else {
 607                 runner.CrunchLog.Printf("Executing on host '%s'", runner.ContainerRecord.UUID, hostname)
 608         }
 609
 610         var runerr, waiterr error
 611
 612         defer func() {
 613                 if err != nil {
 614                         runner.CrunchLog.Print(err)
 615                 }
 616
 617                 if runner.Cancelled {
 618                         runner.finalState = "Cancelled"
 619                 } else {
 620                         runner.finalState = "Complete"
 621                 }
 622
 623                 // (6) capture output
 624                 outputerr := runner.CaptureOutput()
 625                 if outputerr != nil {
 626                         runner.CrunchLog.Print(outputerr)
 627                 }
 628
 629                 // (7) clean up temporary directories
 630                 runner.CleanupDirs()
 631
 632                 // (8) write logs
 633                 logerr := runner.CommitLogs()
 634                 if logerr != nil {
 635                         runner.CrunchLog.Print(logerr)
 636                 }
 637
 638                 // (9) update container record with results
 639                 updateerr := runner.UpdateContainerRecordComplete()
 640                 if updateerr != nil {
 641                         runner.CrunchLog.Print(updateerr)
 642                 }
 643
 644                 runner.CrunchLog.Close()
 645
 646                 if err == nil {
 647                         if runerr != nil {
 648                                 err = runerr
 649                         } else if waiterr != nil {
 650                                 err = runerr
 651                         } else if logerr != nil {
 652                                 err = logerr
 653                         } else if updateerr != nil {
 654                                 err = updateerr
 655                         }
 656                 }
 657         }()
 658
 659         err = runner.ArvClient.Get("containers", runner.ContainerRecord.UUID, nil, &runner.ContainerRecord)
 660         if err != nil {
 661                 return fmt.Errorf("While getting container record: %v", err)
 662         }
 663
 664         // (1) setup signal handling
 665         err = runner.SetupSignals()
 666         if err != nil {
 667                 return fmt.Errorf("While setting up signal handling: %v", err)
 668         }
 669
 670         // (2) check for and/or load image
 671         err = runner.LoadImage()
 672         if err != nil {
 673                 return fmt.Errorf("While loading container image: %v", err)
 674         }
 675
 676         // (3) set up FUSE mount and binds
 677         err = runner.SetupMounts()
 678         if err != nil {
 679                 return fmt.Errorf("While setting up mounts: %v", err)
 680         }
 681
 682         // (3) create and start container
 683         err = runner.StartContainer()
 684         if err != nil {
 685                 if err == ErrCancelled {
 686                         err = nil
 687                 }
 688                 return
 689         }
 690
 691         // (4) update container record state
 692         err = runner.UpdateContainerRecordRunning()
 693         if err != nil {
 694                 runner.CrunchLog.Print(err)
 695         }
 696
 697         // (5) wait for container to finish
 698         waiterr = runner.WaitFinish()
 699
 700         return
 701 }
 702
 703 // NewContainerRunner creates a new container runner.
 704 func NewContainerRunner(api IArvadosClient,
 705         kc IKeepClient,
 706         docker ThinDockerClient,
 707         containerUUID string) *ContainerRunner {
 708
 709         cr := &ContainerRunner{ArvClient: api, Kc: kc, Docker: docker}
 710         cr.NewLogWriter = cr.NewArvLogWriter
 711         cr.RunArvMount = cr.ArvMountCmd
 712         cr.MkTempDir = ioutil.TempDir
 713         cr.LogCollection = &CollectionWriter{kc, nil, sync.Mutex{}}
 714         cr.ContainerRecord.UUID = containerUUID
 715         cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run"))
 716         cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0)
 717         return cr
 718 }
 719
 720 func main() {
 721         flag.Parse()
 722
 723         containerId := flag.Arg(0)
 724
 725         api, err := arvadosclient.MakeArvadosClient()
 726         if err != nil {
 727                 log.Fatalf("%s: %v", containerId, err)
 728         }
 729         api.Retries = 8
 730
 731         var kc *keepclient.KeepClient
 732         kc, err = keepclient.MakeKeepClient(&api)
 733         if err != nil {
 734                 log.Fatalf("%s: %v", containerId, err)
 735         }
 736         kc.Retries = 4
 737
 738         var docker *dockerclient.DockerClient
 739         docker, err = dockerclient.NewDockerClient("unix:///var/run/docker.sock", nil)
 740         if err != nil {
 741                 log.Fatalf("%s: %v", containerId, err)
 742         }
 743
 744         cr := NewContainerRunner(api, kc, docker, containerId)
 745
 746         err = cr.Run()
 747         if err != nil {
 748                 log.Fatalf("%s: %v", containerId, err)
 749         }
 750
 751 }