1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
16 "git.arvados.org/arvados.git/lib/crunchrun"
17 "github.com/sirupsen/logrus"
20 // remoteRunner handles the starting and stopping of a crunch-run
21 // process on a remote machine.
22 type remoteRunner struct {
25 configJSON json.RawMessage
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
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
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
45 var instJSON bytes.Buffer
46 enc := json.NewEncoder(&instJSON)
47 enc.SetIndent("", " ")
48 if err := enc.Encode(wkr.instType); err != nil {
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),
59 if wkr.wp.arvClient.Insecure {
60 configData.Env["ARVADOS_API_HOST_INSECURE"] = "1"
62 if bufs := wkr.wp.cluster.Containers.LocalKeepBlobBuffersPerVCPU; bufs > 0 {
63 configData.Cluster = wkr.wp.cluster
64 configData.KeepBuffers = bufs * wkr.instType.VCPUs
66 configJSON, err := json.Marshal(configData)
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{}),
87 // Start a crunch-run process on the remote host.
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) + "'"
97 cmd += " '" + rr.uuid + "'"
98 if rr.remoteUser != "root" {
101 stdin := bytes.NewBuffer(rr.configJSON)
102 stdout, stderr, err := rr.executor.Execute(nil, cmd, stdin)
104 rr.logger.WithField("stdout", string(stdout)).
105 WithField("stderr", string(stderr)).
107 Error("error starting crunch-run process")
110 rr.logger.Info("crunch-run process started")
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() {
119 // Kill starts a background task to kill the remote process, first
120 // trying SIGTERM until reaching timeoutTERM, then calling
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.
127 // Once Kill has been called, calling it again has no effect.
128 func (rr *remoteRunner) Kill(reason string) {
133 rr.logger.WithField("Reason", reason).Info("killing crunch-run process")
135 termDeadline := time.Now().Add(rr.timeoutTERM)
136 t := time.NewTicker(rr.timeoutSignal)
142 case time.Now().After(termDeadline):
143 rr.logger.Debug("giving up")
145 rr.onUnkillable(rr.uuid)
148 rr.kill(syscall.SIGTERM)
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" {
161 stdout, stderr, err := rr.executor.Execute(nil, cmd, nil)
163 logger.WithFields(logrus.Fields{
164 "stderr": string(stderr),
165 "stdout": string(stdout),
167 }).Info("kill attempt unsuccessful")
173 func (rr *remoteRunner) isClosed() bool {