closes #4467
[arvados.git] / services / keepstore / keepstore.go
1 package main
2
3 import (
4         "bufio"
5         "bytes"
6         "errors"
7         "flag"
8         "fmt"
9         "git.curoverse.com/arvados.git/sdk/go/keepclient"
10         "io/ioutil"
11         "log"
12         "net"
13         "net/http"
14         "os"
15         "os/signal"
16         "strings"
17         "syscall"
18         "time"
19 )
20
21 // ======================
22 // Configuration settings
23 //
24 // TODO(twp): make all of these configurable via command line flags
25 // and/or configuration file settings.
26
27 // Default TCP address on which to listen for requests.
28 // Initialized by the --listen flag.
29 const DEFAULT_ADDR = ":25107"
30
31 // A Keep "block" is 64MB.
32 const BLOCKSIZE = 64 * 1024 * 1024
33
34 // A Keep volume must have at least MIN_FREE_KILOBYTES available
35 // in order to permit writes.
36 const MIN_FREE_KILOBYTES = BLOCKSIZE / 1024
37
38 var PROC_MOUNTS = "/proc/mounts"
39
40 // enforce_permissions controls whether permission signatures
41 // should be enforced (affecting GET and DELETE requests).
42 // Initialized by the -enforce-permissions flag.
43 var enforce_permissions bool
44
45 // blob_signature_ttl is the time duration for which new permission
46 // signatures (returned by PUT requests) will be valid.
47 // Initialized by the -permission-ttl flag.
48 var blob_signature_ttl time.Duration
49
50 // data_manager_token represents the API token used by the
51 // Data Manager, and is required on certain privileged operations.
52 // Initialized by the -data-manager-token-file flag.
53 var data_manager_token string
54
55 // never_delete can be used to prevent the DELETE handler from
56 // actually deleting anything.
57 var never_delete = false
58
59 var maxBuffers = 128
60 var bufs *bufferPool
61
62 // ==========
63 // Error types.
64 //
65 type KeepError struct {
66         HTTPCode int
67         ErrMsg   string
68 }
69
70 var (
71         BadRequestError     = &KeepError{400, "Bad Request"}
72         UnauthorizedError   = &KeepError{401, "Unauthorized"}
73         CollisionError      = &KeepError{500, "Collision"}
74         RequestHashError    = &KeepError{422, "Hash mismatch in request"}
75         PermissionError     = &KeepError{403, "Forbidden"}
76         DiskHashError       = &KeepError{500, "Hash mismatch in stored data"}
77         ExpiredError        = &KeepError{401, "Expired permission signature"}
78         NotFoundError       = &KeepError{404, "Not Found"}
79         GenericError        = &KeepError{500, "Fail"}
80         FullError           = &KeepError{503, "Full"}
81         SizeRequiredError   = &KeepError{411, "Missing Content-Length"}
82         TooLongError        = &KeepError{413, "Block is too large"}
83         MethodDisabledError = &KeepError{405, "Method disabled"}
84 )
85
86 func (e *KeepError) Error() string {
87         return e.ErrMsg
88 }
89
90 // ========================
91 // Internal data structures
92 //
93 // These global variables are used by multiple parts of the
94 // program. They are good candidates for moving into their own
95 // packages.
96
97 // The Keep VolumeManager maintains a list of available volumes.
98 // Initialized by the --volumes flag (or by FindKeepVolumes).
99 var KeepVM VolumeManager
100
101 // The pull list manager and trash queue are threadsafe queues which
102 // support atomic update operations. The PullHandler and TrashHandler
103 // store results from Data Manager /pull and /trash requests here.
104 //
105 // See the Keep and Data Manager design documents for more details:
106 // https://arvados.org/projects/arvados/wiki/Keep_Design_Doc
107 // https://arvados.org/projects/arvados/wiki/Data_Manager_Design_Doc
108 //
109 var pullq *WorkQueue
110 var trashq *WorkQueue
111
112 var (
113         flagSerializeIO bool
114         flagReadonly    bool
115 )
116
117 type volumeSet []Volume
118
119 func (vs *volumeSet) Set(value string) error {
120         if dirs := strings.Split(value, ","); len(dirs) > 1 {
121                 log.Print("DEPRECATED: using comma-separated volume list.")
122                 for _, dir := range dirs {
123                         if err := vs.Set(dir); err != nil {
124                                 return err
125                         }
126                 }
127                 return nil
128         }
129         if len(value) == 0 || value[0] != '/' {
130                 return errors.New("Invalid volume: must begin with '/'.")
131         }
132         if _, err := os.Stat(value); err != nil {
133                 return err
134         }
135         *vs = append(*vs, &UnixVolume{
136                 root:      value,
137                 serialize: flagSerializeIO,
138                 readonly:  flagReadonly,
139         })
140         return nil
141 }
142
143 func (vs *volumeSet) String() string {
144         s := "["
145         for i, v := range *vs {
146                 if i > 0 {
147                         s = s + " "
148                 }
149                 s = s + v.String()
150         }
151         return s + "]"
152 }
153
154 // Discover adds a volume for every directory named "keep" that is
155 // located at the top level of a device- or tmpfs-backed mount point
156 // other than "/". It returns the number of volumes added.
157 func (vs *volumeSet) Discover() int {
158         added := 0
159         f, err := os.Open(PROC_MOUNTS)
160         if err != nil {
161                 log.Fatalf("opening %s: %s", PROC_MOUNTS, err)
162         }
163         scanner := bufio.NewScanner(f)
164         for scanner.Scan() {
165                 args := strings.Fields(scanner.Text())
166                 if err := scanner.Err(); err != nil {
167                         log.Fatalf("reading %s: %s", PROC_MOUNTS, err)
168                 }
169                 dev, mount := args[0], args[1]
170                 if mount == "/" {
171                         continue
172                 }
173                 if dev != "tmpfs" && !strings.HasPrefix(dev, "/dev/") {
174                         continue
175                 }
176                 keepdir := mount + "/keep"
177                 if st, err := os.Stat(keepdir); err != nil || !st.IsDir() {
178                         continue
179                 }
180                 // Set the -readonly flag (but only for this volume)
181                 // if the filesystem is mounted readonly.
182                 flagReadonlyWas := flagReadonly
183                 for _, fsopt := range strings.Split(args[3], ",") {
184                         if fsopt == "ro" {
185                                 flagReadonly = true
186                                 break
187                         }
188                         if fsopt == "rw" {
189                                 break
190                         }
191                 }
192                 vs.Set(keepdir)
193                 flagReadonly = flagReadonlyWas
194                 added++
195         }
196         return added
197 }
198
199 // TODO(twp): continue moving as much code as possible out of main
200 // so it can be effectively tested. Esp. handling and postprocessing
201 // of command line flags (identifying Keep volumes and initializing
202 // permission arguments).
203
204 func main() {
205         log.Println("keepstore starting, pid", os.Getpid())
206         defer log.Println("keepstore exiting, pid", os.Getpid())
207
208         var (
209                 data_manager_token_file string
210                 listen                  string
211                 blob_signing_key_file   string
212                 permission_ttl_sec      int
213                 volumes                 volumeSet
214                 pidfile                 string
215         )
216         flag.StringVar(
217                 &data_manager_token_file,
218                 "data-manager-token-file",
219                 "",
220                 "File with the API token used by the Data Manager. All DELETE "+
221                         "requests or GET /index requests must carry this token.")
222         flag.BoolVar(
223                 &enforce_permissions,
224                 "enforce-permissions",
225                 false,
226                 "Enforce permission signatures on requests.")
227         flag.StringVar(
228                 &listen,
229                 "listen",
230                 DEFAULT_ADDR,
231                 "Listening address, in the form \"host:port\". e.g., 10.0.1.24:8000. Omit the host part to listen on all interfaces.")
232         flag.BoolVar(
233                 &never_delete,
234                 "never-delete",
235                 false,
236                 "If set, nothing will be deleted. HTTP 405 will be returned "+
237                         "for valid DELETE requests.")
238         flag.StringVar(
239                 &blob_signing_key_file,
240                 "permission-key-file",
241                 "",
242                 "Synonym for -blob-signing-key-file.")
243         flag.StringVar(
244                 &blob_signing_key_file,
245                 "blob-signing-key-file",
246                 "",
247                 "File containing the secret key for generating and verifying "+
248                         "blob permission signatures.")
249         flag.IntVar(
250                 &permission_ttl_sec,
251                 "permission-ttl",
252                 0,
253                 "Synonym for -blob-signature-ttl.")
254         flag.IntVar(
255                 &permission_ttl_sec,
256                 "blob-signature-ttl",
257                 int(time.Duration(2*7*24*time.Hour).Seconds()),
258                 "Lifetime of blob permission signatures. "+
259                         "See services/api/config/application.default.yml.")
260         flag.BoolVar(
261                 &flagSerializeIO,
262                 "serialize",
263                 false,
264                 "Serialize read and write operations on the following volumes.")
265         flag.BoolVar(
266                 &flagReadonly,
267                 "readonly",
268                 false,
269                 "Do not write, delete, or touch anything on the following volumes.")
270         flag.Var(
271                 &volumes,
272                 "volumes",
273                 "Deprecated synonym for -volume.")
274         flag.Var(
275                 &volumes,
276                 "volume",
277                 "Local storage directory. Can be given more than once to add multiple directories. If none are supplied, the default is to use all directories named \"keep\" that exist in the top level directory of a mount point at startup time. Can be a comma-separated list, but this is deprecated: use multiple -volume arguments instead.")
278         flag.StringVar(
279                 &pidfile,
280                 "pid",
281                 "",
282                 "Path to write pid file during startup. This file is kept open and locked with LOCK_EX until keepstore exits, so `fuser -k pidfile` is one way to shut down. Exit immediately if there is an error opening, locking, or writing the pid file.")
283         flag.IntVar(
284                 &maxBuffers,
285                 "max-buffers",
286                 maxBuffers,
287                 fmt.Sprintf("Maximum RAM to use for data buffers, given in multiples of block size (%d MiB). When this limit is reached, HTTP requests requiring buffers (like GET and PUT) will wait for buffer space to be released.", BLOCKSIZE>>20))
288
289         flag.Parse()
290
291         if maxBuffers < 0 {
292                 log.Fatal("-max-buffers must be greater than zero.")
293         }
294         bufs = newBufferPool(maxBuffers, BLOCKSIZE)
295
296         if pidfile != "" {
297                 f, err := os.OpenFile(pidfile, os.O_RDWR | os.O_CREATE, 0777)
298                 if err != nil {
299                         log.Fatalf("open pidfile (%s): %s", pidfile, err)
300                 }
301                 err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX | syscall.LOCK_NB)
302                 if err != nil {
303                         log.Fatalf("flock pidfile (%s): %s", pidfile, err)
304                 }
305                 err = f.Truncate(0)
306                 if err != nil {
307                         log.Fatalf("truncate pidfile (%s): %s", pidfile, err)
308                 }
309                 _, err = fmt.Fprint(f, os.Getpid())
310                 if err != nil {
311                         log.Fatalf("write pidfile (%s): %s", pidfile, err)
312                 }
313                 err = f.Sync()
314                 if err != nil {
315                         log.Fatalf("sync pidfile (%s): %s", pidfile, err)
316                 }
317                 defer f.Close()
318                 defer os.Remove(pidfile)
319         }
320
321         if len(volumes) == 0 {
322                 if volumes.Discover() == 0 {
323                         log.Fatal("No volumes found.")
324                 }
325         }
326
327         for _, v := range volumes {
328                 log.Printf("Using volume %v (writable=%v)", v, v.Writable())
329         }
330
331         // Initialize data manager token and permission key.
332         // If these tokens are specified but cannot be read,
333         // raise a fatal error.
334         if data_manager_token_file != "" {
335                 if buf, err := ioutil.ReadFile(data_manager_token_file); err == nil {
336                         data_manager_token = strings.TrimSpace(string(buf))
337                 } else {
338                         log.Fatalf("reading data manager token: %s\n", err)
339                 }
340         }
341         if blob_signing_key_file != "" {
342                 if buf, err := ioutil.ReadFile(blob_signing_key_file); err == nil {
343                         PermissionSecret = bytes.TrimSpace(buf)
344                 } else {
345                         log.Fatalf("reading permission key: %s\n", err)
346                 }
347         }
348
349         blob_signature_ttl = time.Duration(permission_ttl_sec) * time.Second
350
351         if PermissionSecret == nil {
352                 if enforce_permissions {
353                         log.Fatal("-enforce-permissions requires a permission key")
354                 } else {
355                         log.Println("Running without a PermissionSecret. Block locators " +
356                                 "returned by this server will not be signed, and will be rejected " +
357                                 "by a server that enforces permissions.")
358                         log.Println("To fix this, use the -blob-signing-key-file flag " +
359                                 "to specify the file containing the permission key.")
360                 }
361         }
362
363         // Start a round-robin VolumeManager with the volumes we have found.
364         KeepVM = MakeRRVolumeManager(volumes)
365
366         // Tell the built-in HTTP server to direct all requests to the REST router.
367         loggingRouter := MakeLoggingRESTRouter()
368         http.HandleFunc("/", func(resp http.ResponseWriter, req *http.Request) {
369                 loggingRouter.ServeHTTP(resp, req)
370         })
371
372         // Set up a TCP listener.
373         listener, err := net.Listen("tcp", listen)
374         if err != nil {
375                 log.Fatal(err)
376         }
377
378         // Initialize Pull queue and worker
379         keepClient := &keepclient.KeepClient{
380                 Arvados:       nil,
381                 Want_replicas: 1,
382                 Using_proxy:   true,
383                 Client:        &http.Client{},
384         }
385
386         // Initialize the pullq and worker
387         pullq = NewWorkQueue()
388         go RunPullWorker(pullq, keepClient)
389
390         // Initialize the trashq and worker
391         trashq = NewWorkQueue()
392         go RunTrashWorker(trashq)
393
394         // Shut down the server gracefully (by closing the listener)
395         // if SIGTERM is received.
396         term := make(chan os.Signal, 1)
397         go func(sig <-chan os.Signal) {
398                 s := <-sig
399                 log.Println("caught signal:", s)
400                 listener.Close()
401         }(term)
402         signal.Notify(term, syscall.SIGTERM)
403         signal.Notify(term, syscall.SIGINT)
404
405         log.Println("listening at", listen)
406         srv := &http.Server{Addr: listen}
407         srv.Serve(listener)
408 }