21200: copied branch from arvados-workbench2 Arvados-DCO-1.1-Signed-off-by: Lisa...
[arvados.git] / lib / crunchrun / background.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4
5 package crunchrun
6
7 import (
8         "encoding/json"
9         "fmt"
10         "io"
11         "os"
12         "os/exec"
13         "path/filepath"
14         "strings"
15         "syscall"
16         "time"
17 )
18
19 var (
20         lockdir    = "/var/lock"
21         lockprefix = "crunch-run-"
22         locksuffix = ".lock"
23         brokenfile = "crunch-run-broken"
24         pricesfile = "crunch-run-prices.json"
25 )
26
27 // procinfo is saved in each process's lockfile.
28 type procinfo struct {
29         UUID string
30         PID  int
31 }
32
33 // Detach acquires a lock for the given uuid, and starts the current
34 // program as a child process (with -no-detach prepended to the given
35 // arguments so the child knows not to detach again). The lock is
36 // passed along to the child process.
37 //
38 // Stdout and stderr in the child process are sent to the systemd
39 // journal using the systemd-cat program.
40 func Detach(uuid string, prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) int {
41         return exitcode(stderr, detach(uuid, prog, args, stdin, stdout))
42 }
43 func detach(uuid string, prog string, args []string, stdin io.Reader, stdout io.Writer) error {
44         lockfile, err := func() (*os.File, error) {
45                 // We must hold the dir-level lock between
46                 // opening/creating the lockfile and acquiring LOCK_EX
47                 // on it, to avoid racing with the ListProcess's
48                 // alive-checking and garbage collection.
49                 dirlock, err := lockall()
50                 if err != nil {
51                         return nil, err
52                 }
53                 defer dirlock.Close()
54                 lockfilename := filepath.Join(lockdir, lockprefix+uuid+locksuffix)
55                 lockfile, err := os.OpenFile(lockfilename, os.O_CREATE|os.O_RDWR, 0700)
56                 if err != nil {
57                         return nil, fmt.Errorf("open %s: %s", lockfilename, err)
58                 }
59                 err = syscall.Flock(int(lockfile.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
60                 if err != nil {
61                         lockfile.Close()
62                         return nil, fmt.Errorf("lock %s: %s", lockfilename, err)
63                 }
64                 return lockfile, nil
65         }()
66         if err != nil {
67                 return err
68         }
69         defer lockfile.Close()
70         lockfile.Truncate(0)
71
72         execargs := append([]string{"-no-detach"}, args...)
73         if strings.HasSuffix(prog, " crunch-run") {
74                 // invoked as "/path/to/arvados-server crunch-run"
75                 // (see arvados/lib/cmd.Multi)
76                 execargs = append([]string{strings.TrimSuffix(prog, " crunch-run"), "crunch-run"}, execargs...)
77         } else {
78                 // invoked as "/path/to/crunch-run"
79                 execargs = append([]string{prog}, execargs...)
80         }
81         if _, err := exec.LookPath("systemd-cat"); err == nil {
82                 execargs = append([]string{
83                         // Here, if the inner systemd-cat can't exec
84                         // crunch-run, it writes an error message to
85                         // stderr, and the outer systemd-cat writes it
86                         // to the journal where the operator has a
87                         // chance to discover it. (If we only used one
88                         // systemd-cat command, it would be up to us
89                         // to report the error -- but we are going to
90                         // detach and exit, not wait for something to
91                         // appear on stderr.)  Note these systemd-cat
92                         // calls don't result in additional processes
93                         // -- they just connect stderr/stdout to
94                         // sockets and call exec().
95                         "systemd-cat", "--identifier=crunch-run",
96                         "systemd-cat", "--identifier=crunch-run",
97                 }, execargs...)
98         }
99
100         cmd := exec.Command(execargs[0], execargs[1:]...)
101         // Child inherits lockfile.
102         cmd.ExtraFiles = []*os.File{lockfile}
103         // Ensure child isn't interrupted even if we receive signals
104         // from parent (sshd) while sending lockfile content to
105         // caller.
106         cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
107         // We need to manage our own OS pipe here to ensure the child
108         // process reads all of our stdin pipe before we return.
109         piper, pipew, err := os.Pipe()
110         if err != nil {
111                 return err
112         }
113         defer pipew.Close()
114         cmd.Stdin = piper
115         err = cmd.Start()
116         if err != nil {
117                 return fmt.Errorf("exec %s: %s", cmd.Path, err)
118         }
119         _, err = io.Copy(pipew, stdin)
120         if err != nil {
121                 return err
122         }
123         err = pipew.Close()
124         if err != nil {
125                 return err
126         }
127
128         w := io.MultiWriter(stdout, lockfile)
129         return json.NewEncoder(w).Encode(procinfo{
130                 UUID: uuid,
131                 PID:  cmd.Process.Pid,
132         })
133 }
134
135 // KillProcess finds the crunch-run process corresponding to the given
136 // uuid, and sends the given signal to it. It then waits up to 1
137 // second for the process to die. It returns 0 if the process is
138 // successfully killed or didn't exist in the first place.
139 func KillProcess(uuid string, signal syscall.Signal, stdout, stderr io.Writer) int {
140         return exitcode(stderr, kill(uuid, signal, stdout, stderr))
141 }
142
143 func kill(uuid string, signal syscall.Signal, stdout, stderr io.Writer) error {
144         path := filepath.Join(lockdir, lockprefix+uuid+locksuffix)
145         f, err := os.Open(path)
146         if os.IsNotExist(err) {
147                 return nil
148         } else if err != nil {
149                 return fmt.Errorf("open %s: %s", path, err)
150         }
151         defer f.Close()
152
153         var pi procinfo
154         err = json.NewDecoder(f).Decode(&pi)
155         if err != nil {
156                 return fmt.Errorf("decode %s: %s", path, err)
157         }
158
159         if pi.UUID != uuid || pi.PID == 0 {
160                 return fmt.Errorf("%s: bogus procinfo: %+v", path, pi)
161         }
162
163         proc, err := os.FindProcess(pi.PID)
164         if err != nil {
165                 // FindProcess should have succeeded, even if the
166                 // process does not exist.
167                 return fmt.Errorf("%s: find process %d: %s", uuid, pi.PID, err)
168         }
169
170         // Send the requested signal once, then send signal 0 a few
171         // times.  When proc.Signal() returns an error (process no
172         // longer exists), return success.  If that doesn't happen
173         // within 1 second, return an error.
174         err = proc.Signal(signal)
175         for deadline := time.Now().Add(time.Second); err == nil && time.Now().Before(deadline); time.Sleep(time.Second / 100) {
176                 err = proc.Signal(syscall.Signal(0))
177         }
178         if err == nil {
179                 // Reached deadline without a proc.Signal() error.
180                 return fmt.Errorf("%s: pid %d: sent signal %d (%s) but process is still alive", uuid, pi.PID, signal, signal)
181         }
182         fmt.Fprintf(stderr, "%s: pid %d: %s\n", uuid, pi.PID, err)
183         return nil
184 }
185
186 // ListProcesses lists UUIDs of active crunch-run processes.
187 func ListProcesses(stdin io.Reader, stdout, stderr io.Writer) int {
188         if buf, err := io.ReadAll(stdin); err == nil && len(buf) > 0 {
189                 // write latest pricing data to disk where
190                 // current/future crunch-run processes can load it
191                 fnm := filepath.Join(lockdir, pricesfile)
192                 fnmtmp := fmt.Sprintf("%s~%d", fnm, os.Getpid())
193                 err := os.WriteFile(fnmtmp, buf, 0777)
194                 if err != nil {
195                         fmt.Fprintf(stderr, "error writing price data to %s: %s", fnmtmp, err)
196                 } else if err = os.Rename(fnmtmp, fnm); err != nil {
197                         fmt.Fprintf(stderr, "error renaming %s to %s: %s", fnmtmp, fnm, err)
198                         os.Remove(fnmtmp)
199                 }
200         }
201         // filepath.Walk does not follow symlinks, so we must walk
202         // lockdir+"/." in case lockdir itself is a symlink.
203         walkdir := lockdir + "/."
204         return exitcode(stderr, filepath.Walk(walkdir, func(path string, info os.FileInfo, err error) error {
205                 if info.IsDir() && path != walkdir {
206                         return filepath.SkipDir
207                 }
208                 if name := info.Name(); name == brokenfile {
209                         fmt.Fprintln(stdout, "broken")
210                         return nil
211                 } else if !strings.HasPrefix(name, lockprefix) || !strings.HasSuffix(name, locksuffix) {
212                         return nil
213                 }
214                 if info.Size() == 0 {
215                         // race: process has opened/locked but hasn't yet written pid/uuid
216                         return nil
217                 }
218
219                 f, err := os.Open(path)
220                 if err != nil {
221                         return nil
222                 }
223                 defer f.Close()
224
225                 // Ensure other processes don't acquire this lockfile
226                 // after we have decided it is abandoned but before we
227                 // have deleted it.
228                 dirlock, err := lockall()
229                 if err != nil {
230                         return err
231                 }
232                 err = syscall.Flock(int(f.Fd()), syscall.LOCK_SH|syscall.LOCK_NB)
233                 if err == nil {
234                         // lockfile is stale
235                         err := os.Remove(path)
236                         dirlock.Close()
237                         if err != nil {
238                                 fmt.Fprintf(stderr, "unlink %s: %s\n", f.Name(), err)
239                         }
240                         return nil
241                 }
242                 dirlock.Close()
243
244                 var pi procinfo
245                 err = json.NewDecoder(f).Decode(&pi)
246                 if err != nil {
247                         fmt.Fprintf(stderr, "%s: %s\n", path, err)
248                         return nil
249                 }
250                 if pi.UUID == "" || pi.PID == 0 {
251                         fmt.Fprintf(stderr, "%s: bogus procinfo: %+v", path, pi)
252                         return nil
253                 }
254
255                 proc, err := os.FindProcess(pi.PID)
256                 if err != nil {
257                         // FindProcess should have succeeded, even if the
258                         // process does not exist.
259                         fmt.Fprintf(stderr, "%s: find process %d: %s", path, pi.PID, err)
260                         return nil
261                 }
262                 err = proc.Signal(syscall.SIGUSR2)
263                 if err != nil {
264                         // Process is dead, even though lockfile was
265                         // still locked. Most likely a stuck arv-mount
266                         // process that inherited the lock from
267                         // crunch-run. Report container UUID as
268                         // "stale".
269                         fmt.Fprintln(stdout, pi.UUID, "stale")
270                         return nil
271                 }
272
273                 fmt.Fprintln(stdout, pi.UUID)
274                 return nil
275         }))
276 }
277
278 // If err is nil, return 0 ("success"); otherwise, print err to stderr
279 // and return 1.
280 func exitcode(stderr io.Writer, err error) int {
281         if err != nil {
282                 fmt.Fprintln(stderr, err)
283                 return 1
284         }
285         return 0
286 }
287
288 // Acquire a dir-level lock. Must be held while creating or deleting
289 // container-specific lockfiles, to avoid races during the intervals
290 // when those container-specific lockfiles are open but not locked.
291 //
292 // Caller releases the lock by closing the returned file.
293 func lockall() (*os.File, error) {
294         lockfile := filepath.Join(lockdir, lockprefix+"all"+locksuffix)
295         f, err := os.OpenFile(lockfile, os.O_CREATE|os.O_RDWR, 0700)
296         if err != nil {
297                 return nil, fmt.Errorf("open %s: %s", lockfile, err)
298         }
299         err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX)
300         if err != nil {
301                 f.Close()
302                 return nil, fmt.Errorf("lock %s: %s", lockfile, err)
303         }
304         return f, nil
305 }