Merge, don't replace RemoteClusters refs #13996
[arvados.git] / services / crunch-run / background.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4
5 package main
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 )
25
26 // procinfo is saved in each process's lockfile.
27 type procinfo struct {
28         UUID string
29         PID  int
30 }
31
32 // Detach acquires a lock for the given uuid, and starts the current
33 // program as a child process (with -no-detach prepended to the given
34 // arguments so the child knows not to detach again). The lock is
35 // passed along to the child process.
36 //
37 // Stdout and stderr in the child process are sent to the systemd
38 // journal using the systemd-cat program.
39 func Detach(uuid string, args []string, stdout, stderr io.Writer) int {
40         return exitcode(stderr, detach(uuid, args, stdout, stderr))
41 }
42 func detach(uuid string, args []string, stdout, stderr io.Writer) error {
43         lockfile, err := func() (*os.File, error) {
44                 // We must hold the dir-level lock between
45                 // opening/creating the lockfile and acquiring LOCK_EX
46                 // on it, to avoid racing with the ListProcess's
47                 // alive-checking and garbage collection.
48                 dirlock, err := lockall()
49                 if err != nil {
50                         return nil, err
51                 }
52                 defer dirlock.Close()
53                 lockfilename := filepath.Join(lockdir, lockprefix+uuid+locksuffix)
54                 lockfile, err := os.OpenFile(lockfilename, os.O_CREATE|os.O_RDWR, 0700)
55                 if err != nil {
56                         return nil, fmt.Errorf("open %s: %s", lockfilename, err)
57                 }
58                 err = syscall.Flock(int(lockfile.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
59                 if err != nil {
60                         lockfile.Close()
61                         return nil, fmt.Errorf("lock %s: %s", lockfilename, err)
62                 }
63                 return lockfile, nil
64         }()
65         if err != nil {
66                 return err
67         }
68         defer lockfile.Close()
69         lockfile.Truncate(0)
70
71         cmd := exec.Command("systemd-cat", append([]string{"--identifier=crunch-run", args[0], "-no-detach"}, args[1:]...)...)
72         // Child inherits lockfile.
73         cmd.ExtraFiles = []*os.File{lockfile}
74         // Ensure child isn't interrupted even if we receive signals
75         // from parent (sshd) while sending lockfile content to
76         // caller.
77         cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
78         err = cmd.Start()
79         if err != nil {
80                 return fmt.Errorf("exec %s: %s", cmd.Path, err)
81         }
82
83         w := io.MultiWriter(stdout, lockfile)
84         return json.NewEncoder(w).Encode(procinfo{
85                 UUID: uuid,
86                 PID:  cmd.Process.Pid,
87         })
88 }
89
90 // KillProcess finds the crunch-run process corresponding to the given
91 // uuid, and sends the given signal to it. It then waits up to 1
92 // second for the process to die. It returns 0 if the process is
93 // successfully killed or didn't exist in the first place.
94 func KillProcess(uuid string, signal syscall.Signal, stdout, stderr io.Writer) int {
95         return exitcode(stderr, kill(uuid, signal, stdout, stderr))
96 }
97
98 func kill(uuid string, signal syscall.Signal, stdout, stderr io.Writer) error {
99         path := filepath.Join(lockdir, lockprefix+uuid+locksuffix)
100         f, err := os.Open(path)
101         if os.IsNotExist(err) {
102                 return nil
103         } else if err != nil {
104                 return fmt.Errorf("open %s: %s", path, err)
105         }
106         defer f.Close()
107
108         var pi procinfo
109         err = json.NewDecoder(f).Decode(&pi)
110         if err != nil {
111                 return fmt.Errorf("decode %s: %s\n", path, err)
112         }
113
114         if pi.UUID != uuid || pi.PID == 0 {
115                 return fmt.Errorf("%s: bogus procinfo: %+v", path, pi)
116         }
117
118         proc, err := os.FindProcess(pi.PID)
119         if err != nil {
120                 // FindProcess should have succeeded, even if the
121                 // process does not exist.
122                 return fmt.Errorf("%s: find process %d: %s", uuid, pi.PID, err)
123         }
124
125         // Send the requested signal once, then send signal 0 a few
126         // times.  When proc.Signal() returns an error (process no
127         // longer exists), return success.  If that doesn't happen
128         // within 1 second, return an error.
129         err = proc.Signal(signal)
130         for deadline := time.Now().Add(time.Second); err == nil && time.Now().Before(deadline); time.Sleep(time.Second / 100) {
131                 err = proc.Signal(syscall.Signal(0))
132         }
133         if err == nil {
134                 // Reached deadline without a proc.Signal() error.
135                 return fmt.Errorf("%s: pid %d: sent signal %d (%s) but process is still alive", uuid, pi.PID, signal, signal)
136         }
137         fmt.Fprintf(stderr, "%s: pid %d: %s\n", uuid, pi.PID, err)
138         return nil
139 }
140
141 // List UUIDs of active crunch-run processes.
142 func ListProcesses(stdout, stderr io.Writer) int {
143         // filepath.Walk does not follow symlinks, so we must walk
144         // lockdir+"/." in case lockdir itself is a symlink.
145         walkdir := lockdir + "/."
146         return exitcode(stderr, filepath.Walk(walkdir, func(path string, info os.FileInfo, err error) error {
147                 if info.IsDir() && path != walkdir {
148                         return filepath.SkipDir
149                 }
150                 if name := info.Name(); name == brokenfile {
151                         fmt.Fprintln(stdout, "broken")
152                         return nil
153                 } else if !strings.HasPrefix(name, lockprefix) || !strings.HasSuffix(name, locksuffix) {
154                         return nil
155                 }
156                 if info.Size() == 0 {
157                         // race: process has opened/locked but hasn't yet written pid/uuid
158                         return nil
159                 }
160
161                 f, err := os.Open(path)
162                 if err != nil {
163                         return nil
164                 }
165                 defer f.Close()
166
167                 // Ensure other processes don't acquire this lockfile
168                 // after we have decided it is abandoned but before we
169                 // have deleted it.
170                 dirlock, err := lockall()
171                 if err != nil {
172                         return err
173                 }
174                 err = syscall.Flock(int(f.Fd()), syscall.LOCK_SH|syscall.LOCK_NB)
175                 if err == nil {
176                         // lockfile is stale
177                         err := os.Remove(path)
178                         dirlock.Close()
179                         if err != nil {
180                                 fmt.Fprintf(stderr, "unlink %s: %s\n", f.Name(), err)
181                         }
182                         return nil
183                 }
184                 dirlock.Close()
185
186                 var pi procinfo
187                 err = json.NewDecoder(f).Decode(&pi)
188                 if err != nil {
189                         fmt.Fprintf(stderr, "%s: %s\n", path, err)
190                         return nil
191                 }
192                 if pi.UUID == "" || pi.PID == 0 {
193                         fmt.Fprintf(stderr, "%s: bogus procinfo: %+v", path, pi)
194                         return nil
195                 }
196
197                 fmt.Fprintln(stdout, pi.UUID)
198                 return nil
199         }))
200 }
201
202 // If err is nil, return 0 ("success"); otherwise, print err to stderr
203 // and return 1.
204 func exitcode(stderr io.Writer, err error) int {
205         if err != nil {
206                 fmt.Fprintln(stderr, err)
207                 return 1
208         }
209         return 0
210 }
211
212 // Acquire a dir-level lock. Must be held while creating or deleting
213 // container-specific lockfiles, to avoid races during the intervals
214 // when those container-specific lockfiles are open but not locked.
215 //
216 // Caller releases the lock by closing the returned file.
217 func lockall() (*os.File, error) {
218         lockfile := filepath.Join(lockdir, lockprefix+"all"+locksuffix)
219         f, err := os.OpenFile(lockfile, os.O_CREATE|os.O_RDWR, 0700)
220         if err != nil {
221                 return nil, fmt.Errorf("open %s: %s", lockfile, err)
222         }
223         err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX)
224         if err != nil {
225                 f.Close()
226                 return nil, fmt.Errorf("lock %s: %s", lockfile, err)
227         }
228         return f, nil
229 }