lib/dispatchcloud/worker/runner.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package worker
   6
   7 import (
   8         "bytes"
   9         "encoding/json"
  10         "fmt"
  11         "net"
  12         "strings"
  13         "syscall"
  14         "time"
  15
  16         "git.arvados.org/arvados.git/lib/crunchrun"
  17         "github.com/sirupsen/logrus"
  18 )
  19
  20 // remoteRunner handles the starting and stopping of a crunch-run
  21 // process on a remote machine.
  22 type remoteRunner struct {
  23         uuid          string
  24         executor      Executor
  25         configJSON    json.RawMessage
  26         runnerCmd     string
  27         runnerArgs    []string
  28         remoteUser    string
  29         timeoutTERM   time.Duration
  30         timeoutSignal time.Duration
  31         onUnkillable  func(uuid string) // callback invoked when giving up on SIGTERM
  32         onKilled      func(uuid string) // callback invoked when process exits after SIGTERM
  33         logger        logrus.FieldLogger
  34
  35         stopping bool          // true if Stop() has been called
  36         givenup  bool          // true if timeoutTERM has been reached
  37         closed   chan struct{} // channel is closed if Close() has been called
  38 }
  39
  40 // newRemoteRunner returns a new remoteRunner. Caller should ensure
  41 // Close() is called to release resources.
  42 func newRemoteRunner(uuid string, wkr *worker) *remoteRunner {
  43         // Send the instance type record as a JSON doc so crunch-run
  44         // can log it.
  45         var instJSON bytes.Buffer
  46         enc := json.NewEncoder(&instJSON)
  47         enc.SetIndent("", "    ")
  48         if err := enc.Encode(wkr.instType); err != nil {
  49                 panic(err)
  50         }
  51         var configData crunchrun.ConfigData
  52         configData.Env = map[string]string{
  53                 "ARVADOS_API_HOST":  wkr.wp.arvClient.APIHost,
  54                 "ARVADOS_API_TOKEN": wkr.wp.arvClient.AuthToken,
  55                 "InstanceType":      instJSON.String(),
  56                 "GatewayAddress":    net.JoinHostPort(wkr.instance.Address(), "0"),
  57                 "GatewayAuthSecret": wkr.wp.gatewayAuthSecret(uuid),
  58         }
  59         if wkr.wp.arvClient.Insecure {
  60                 configData.Env["ARVADOS_API_HOST_INSECURE"] = "1"
  61         }
  62         if bufs := wkr.wp.cluster.Containers.LocalKeepBlobBuffersPerVCPU; bufs > 0 {
  63                 configData.Cluster = wkr.wp.cluster
  64                 configData.KeepBuffers = bufs * wkr.instType.VCPUs
  65         }
  66         configJSON, err := json.Marshal(configData)
  67         if err != nil {
  68                 panic(err)
  69         }
  70         rr := &remoteRunner{
  71                 uuid:          uuid,
  72                 executor:      wkr.executor,
  73                 configJSON:    configJSON,
  74                 runnerCmd:     wkr.wp.runnerCmd,
  75                 runnerArgs:    wkr.wp.runnerArgs,
  76                 remoteUser:    wkr.instance.RemoteUser(),
  77                 timeoutTERM:   wkr.wp.timeoutTERM,
  78                 timeoutSignal: wkr.wp.timeoutSignal,
  79                 onUnkillable:  wkr.onUnkillable,
  80                 onKilled:      wkr.onKilled,
  81                 logger:        wkr.logger.WithField("ContainerUUID", uuid),
  82                 closed:        make(chan struct{}),
  83         }
  84         return rr
  85 }
  86
  87 // Start a crunch-run process on the remote host.
  88 //
  89 // Start does not return any error encountered. The caller should
  90 // assume the remote process _might_ have started, at least until it
  91 // probes the worker and finds otherwise.
  92 func (rr *remoteRunner) Start() {
  93         cmd := rr.runnerCmd + " --detach --stdin-config"
  94         for _, arg := range rr.runnerArgs {
  95                 cmd += " '" + strings.Replace(arg, "'", "'\\''", -1) + "'"
  96         }
  97         cmd += " '" + rr.uuid + "'"
  98         if rr.remoteUser != "root" {
  99                 cmd = "sudo " + cmd
 100         }
 101         stdin := bytes.NewBuffer(rr.configJSON)
 102         stdout, stderr, err := rr.executor.Execute(nil, cmd, stdin)
 103         if err != nil {
 104                 rr.logger.WithField("stdout", string(stdout)).
 105                         WithField("stderr", string(stderr)).
 106                         WithError(err).
 107                         Error("error starting crunch-run process")
 108                 return
 109         }
 110         rr.logger.Info("crunch-run process started")
 111 }
 112
 113 // Close abandons the remote process (if any) and releases
 114 // resources. Close must not be called more than once.
 115 func (rr *remoteRunner) Close() {
 116         close(rr.closed)
 117 }
 118
 119 // Kill starts a background task to kill the remote process, first
 120 // trying SIGTERM until reaching timeoutTERM, then calling
 121 // onUnkillable().
 122 //
 123 // SIGKILL is not used. It would merely kill the crunch-run supervisor
 124 // and thereby make the docker container, arv-mount, etc. invisible to
 125 // us without actually stopping them.
 126 //
 127 // Once Kill has been called, calling it again has no effect.
 128 func (rr *remoteRunner) Kill(reason string) {
 129         if rr.stopping {
 130                 return
 131         }
 132         rr.stopping = true
 133         rr.logger.WithField("Reason", reason).Info("killing crunch-run process")
 134         go func() {
 135                 termDeadline := time.Now().Add(rr.timeoutTERM)
 136                 t := time.NewTicker(rr.timeoutSignal)
 137                 defer t.Stop()
 138                 for range t.C {
 139                         switch {
 140                         case rr.isClosed():
 141                                 return
 142                         case time.Now().After(termDeadline):
 143                                 rr.logger.Debug("giving up")
 144                                 rr.givenup = true
 145                                 rr.onUnkillable(rr.uuid)
 146                                 return
 147                         default:
 148                                 rr.kill(syscall.SIGTERM)
 149                         }
 150                 }
 151         }()
 152 }
 153
 154 func (rr *remoteRunner) kill(sig syscall.Signal) {
 155         logger := rr.logger.WithField("Signal", int(sig))
 156         logger.Info("sending signal")
 157         cmd := fmt.Sprintf(rr.runnerCmd+" --kill %d %s", sig, rr.uuid)
 158         if rr.remoteUser != "root" {
 159                 cmd = "sudo " + cmd
 160         }
 161         stdout, stderr, err := rr.executor.Execute(nil, cmd, nil)
 162         if err != nil {
 163                 logger.WithFields(logrus.Fields{
 164                         "stderr": string(stderr),
 165                         "stdout": string(stdout),
 166                         "error":  err,
 167                 }).Info("kill attempt unsuccessful")
 168                 return
 169         }
 170         rr.onKilled(rr.uuid)
 171 }
 172
 173 func (rr *remoteRunner) isClosed() bool {
 174         select {
 175         case <-rr.closed:
 176                 return true
 177         default:
 178                 return false
 179         }
 180 }