1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
16 "github.com/sirupsen/logrus"
19 // remoteRunner handles the starting and stopping of a crunch-run
20 // process on a remote machine.
21 type remoteRunner struct {
24 envJSON json.RawMessage
28 timeoutTERM time.Duration
29 timeoutSignal time.Duration
30 onUnkillable func(uuid string) // callback invoked when giving up on SIGTERM
31 onKilled func(uuid string) // callback invoked when process exits after SIGTERM
32 logger logrus.FieldLogger
34 stopping bool // true if Stop() has been called
35 givenup bool // true if timeoutTERM has been reached
36 closed chan struct{} // channel is closed if Close() has been called
39 // newRemoteRunner returns a new remoteRunner. Caller should ensure
40 // Close() is called to release resources.
41 func newRemoteRunner(uuid string, wkr *worker) *remoteRunner {
42 // Send the instance type record as a JSON doc so crunch-run
44 var instJSON bytes.Buffer
45 enc := json.NewEncoder(&instJSON)
46 enc.SetIndent("", " ")
47 if err := enc.Encode(wkr.instType); err != nil {
50 env := map[string]string{
51 "ARVADOS_API_HOST": wkr.wp.arvClient.APIHost,
52 "ARVADOS_API_TOKEN": wkr.wp.arvClient.AuthToken,
53 "InstanceType": instJSON.String(),
54 "GatewayAddress": net.JoinHostPort(wkr.instance.Address(), "0"),
55 "GatewayAuthSecret": wkr.wp.gatewayAuthSecret(uuid),
57 if wkr.wp.arvClient.Insecure {
58 env["ARVADOS_API_HOST_INSECURE"] = "1"
60 envJSON, err := json.Marshal(env)
66 executor: wkr.executor,
68 runnerCmd: wkr.wp.runnerCmd,
69 runnerArgs: wkr.wp.runnerArgs,
70 remoteUser: wkr.instance.RemoteUser(),
71 timeoutTERM: wkr.wp.timeoutTERM,
72 timeoutSignal: wkr.wp.timeoutSignal,
73 onUnkillable: wkr.onUnkillable,
74 onKilled: wkr.onKilled,
75 logger: wkr.logger.WithField("ContainerUUID", uuid),
76 closed: make(chan struct{}),
81 // Start a crunch-run process on the remote host.
83 // Start does not return any error encountered. The caller should
84 // assume the remote process _might_ have started, at least until it
85 // probes the worker and finds otherwise.
86 func (rr *remoteRunner) Start() {
87 cmd := rr.runnerCmd + " --detach --stdin-env"
88 for _, arg := range rr.runnerArgs {
89 cmd += " '" + strings.Replace(arg, "'", "'\\''", -1) + "'"
91 cmd += " '" + rr.uuid + "'"
92 if rr.remoteUser != "root" {
95 stdin := bytes.NewBuffer(rr.envJSON)
96 stdout, stderr, err := rr.executor.Execute(nil, cmd, stdin)
98 rr.logger.WithField("stdout", string(stdout)).
99 WithField("stderr", string(stderr)).
101 Error("error starting crunch-run process")
104 rr.logger.Info("crunch-run process started")
107 // Close abandons the remote process (if any) and releases
108 // resources. Close must not be called more than once.
109 func (rr *remoteRunner) Close() {
113 // Kill starts a background task to kill the remote process, first
114 // trying SIGTERM until reaching timeoutTERM, then calling
117 // SIGKILL is not used. It would merely kill the crunch-run supervisor
118 // and thereby make the docker container, arv-mount, etc. invisible to
119 // us without actually stopping them.
121 // Once Kill has been called, calling it again has no effect.
122 func (rr *remoteRunner) Kill(reason string) {
127 rr.logger.WithField("Reason", reason).Info("killing crunch-run process")
129 termDeadline := time.Now().Add(rr.timeoutTERM)
130 t := time.NewTicker(rr.timeoutSignal)
136 case time.Now().After(termDeadline):
137 rr.logger.Debug("giving up")
139 rr.onUnkillable(rr.uuid)
142 rr.kill(syscall.SIGTERM)
148 func (rr *remoteRunner) kill(sig syscall.Signal) {
149 logger := rr.logger.WithField("Signal", int(sig))
150 logger.Info("sending signal")
151 cmd := fmt.Sprintf(rr.runnerCmd+" --kill %d %s", sig, rr.uuid)
152 if rr.remoteUser != "root" {
155 stdout, stderr, err := rr.executor.Execute(nil, cmd, nil)
157 logger.WithFields(logrus.Fields{
158 "stderr": string(stderr),
159 "stdout": string(stdout),
161 }).Info("kill attempt unsuccessful")
167 func (rr *remoteRunner) isClosed() bool {