d5a2a3591712d7ab4df3e084c93764665be7f0ee
[arvados.git] / services / keep / src / keep / handlers.go
1 package main
2
3 // REST handlers for Keep are implemented here.
4 //
5 // GetBlockHandler (GET /locator)
6 // PutBlockHandler (PUT /locator)
7 // IndexHandler    (GET /index, GET /index/prefix)
8 // StatusHandler   (GET /status.json)
9
10 import (
11         "bufio"
12         "bytes"
13         "crypto/md5"
14         "encoding/json"
15         "fmt"
16         "github.com/gorilla/mux"
17         "io"
18         "log"
19         "net/http"
20         "os"
21         "regexp"
22         "runtime"
23         "strconv"
24         "strings"
25         "syscall"
26         "time"
27 )
28
29 // MakeRESTRouter returns a new mux.Router that forwards all Keep
30 // requests to the appropriate handlers.
31 //
32 func MakeRESTRouter() *mux.Router {
33         rest := mux.NewRouter()
34
35         rest.HandleFunc(
36                 `/{hash:[0-9a-f]{32}}`, GetBlockHandler).Methods("GET", "HEAD")
37         rest.HandleFunc(
38                 `/{hash:[0-9a-f]{32}}+{hints}`,
39                 GetBlockHandler).Methods("GET", "HEAD")
40
41         rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, PutBlockHandler).Methods("PUT")
42         rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, DeleteHandler).Methods("DELETE")
43
44         // For IndexHandler we support:
45         //   /index           - returns all locators
46         //   /index/{prefix}  - returns all locators that begin with {prefix}
47         //      {prefix} is a string of hexadecimal digits between 0 and 32 digits.
48         //      If {prefix} is the empty string, return an index of all locators
49         //      (so /index and /index/ behave identically)
50         //      A client may supply a full 32-digit locator string, in which
51         //      case the server will return an index with either zero or one
52         //      entries. This usage allows a client to check whether a block is
53         //      present, and its size and upload time, without retrieving the
54         //      entire block.
55         //
56         rest.HandleFunc(`/index`, IndexHandler).Methods("GET", "HEAD")
57         rest.HandleFunc(
58                 `/index/{prefix:[0-9a-f]{0,32}}`, IndexHandler).Methods("GET", "HEAD")
59         rest.HandleFunc(`/status.json`, StatusHandler).Methods("GET", "HEAD")
60
61         // Any request which does not match any of these routes gets
62         // 400 Bad Request.
63         rest.NotFoundHandler = http.HandlerFunc(BadRequestHandler)
64
65         return rest
66 }
67
68 func BadRequestHandler(w http.ResponseWriter, r *http.Request) {
69         http.Error(w, BadRequestError.Error(), BadRequestError.HTTPCode)
70 }
71
72 // FindKeepVolumes scans all mounted volumes on the system for Keep
73 // volumes, and returns a list of matching paths.
74 //
75 // A device is assumed to be a Keep volume if it is a normal or tmpfs
76 // volume and has a "/keep" directory directly underneath the mount
77 // point.
78 //
79 func FindKeepVolumes() []string {
80         vols := make([]string, 0)
81
82         if f, err := os.Open(PROC_MOUNTS); err != nil {
83                 log.Fatalf("opening %s: %s\n", PROC_MOUNTS, err)
84         } else {
85                 scanner := bufio.NewScanner(f)
86                 for scanner.Scan() {
87                         args := strings.Fields(scanner.Text())
88                         dev, mount := args[0], args[1]
89                         if mount != "/" &&
90                                 (dev == "tmpfs" || strings.HasPrefix(dev, "/dev/")) {
91                                 keep := mount + "/keep"
92                                 if st, err := os.Stat(keep); err == nil && st.IsDir() {
93                                         vols = append(vols, keep)
94                                 }
95                         }
96                 }
97                 if err := scanner.Err(); err != nil {
98                         log.Fatal(err)
99                 }
100         }
101         return vols
102 }
103
104 func GetBlockHandler(resp http.ResponseWriter, req *http.Request) {
105         hash := mux.Vars(req)["hash"]
106
107         log.Printf("%s %s", req.Method, hash)
108
109         hints := mux.Vars(req)["hints"]
110
111         // Parse the locator string and hints from the request.
112         // TODO(twp): implement a Locator type.
113         var signature, timestamp string
114         if hints != "" {
115                 signature_pat, _ := regexp.Compile("^A([[:xdigit:]]+)@([[:xdigit:]]{8})$")
116                 for _, hint := range strings.Split(hints, "+") {
117                         if match, _ := regexp.MatchString("^[[:digit:]]+$", hint); match {
118                                 // Server ignores size hints
119                         } else if m := signature_pat.FindStringSubmatch(hint); m != nil {
120                                 signature = m[1]
121                                 timestamp = m[2]
122                         } else if match, _ := regexp.MatchString("^[[:upper:]]", hint); match {
123                                 // Any unknown hint that starts with an uppercase letter is
124                                 // presumed to be valid and ignored, to permit forward compatibility.
125                         } else {
126                                 // Unknown format; not a valid locator.
127                                 http.Error(resp, BadRequestError.Error(), BadRequestError.HTTPCode)
128                                 return
129                         }
130                 }
131         }
132
133         // If permission checking is in effect, verify this
134         // request's permission signature.
135         if enforce_permissions {
136                 if signature == "" || timestamp == "" {
137                         http.Error(resp, PermissionError.Error(), PermissionError.HTTPCode)
138                         return
139                 } else if IsExpired(timestamp) {
140                         http.Error(resp, ExpiredError.Error(), ExpiredError.HTTPCode)
141                         return
142                 } else {
143                         req_locator := req.URL.Path[1:] // strip leading slash
144                         if !VerifySignature(req_locator, GetApiToken(req)) {
145                                 http.Error(resp, PermissionError.Error(), PermissionError.HTTPCode)
146                                 return
147                         }
148                 }
149         }
150
151         block, err := GetBlock(hash)
152
153         // Garbage collect after each GET. Fixes #2865.
154         // TODO(twp): review Keep memory usage and see if there's
155         // a better way to do this than blindly garbage collecting
156         // after every block.
157         defer runtime.GC()
158
159         if err != nil {
160                 // This type assertion is safe because the only errors
161                 // GetBlock can return are DiskHashError or NotFoundError.
162                 if err == NotFoundError {
163                         log.Printf("%s: not found, giving up\n", hash)
164                 }
165                 http.Error(resp, err.Error(), err.(*KeepError).HTTPCode)
166                 return
167         }
168
169         resp.Header().Set("X-Block-Size", fmt.Sprintf("%d", len(block)))
170
171         _, err = resp.Write(block)
172         if err != nil {
173                 log.Printf("GetBlockHandler: writing response: %s", err)
174         }
175
176         return
177 }
178
179 func PutBlockHandler(resp http.ResponseWriter, req *http.Request) {
180         // Garbage collect after each PUT. Fixes #2865.
181         // See also GetBlockHandler.
182         defer runtime.GC()
183
184         hash := mux.Vars(req)["hash"]
185
186         log.Printf("%s %s", req.Method, hash)
187
188         // Read the block data to be stored.
189         // If the request exceeds BLOCKSIZE bytes, issue a HTTP 500 error.
190         //
191         if req.ContentLength > BLOCKSIZE {
192                 http.Error(resp, TooLongError.Error(), TooLongError.HTTPCode)
193                 return
194         }
195
196         buf := make([]byte, req.ContentLength)
197         nread, err := io.ReadFull(req.Body, buf)
198         if err != nil {
199                 http.Error(resp, err.Error(), 500)
200         } else if int64(nread) < req.ContentLength {
201                 http.Error(resp, "request truncated", 500)
202         } else {
203                 if err := PutBlock(buf, hash); err == nil {
204                         // Success; add a size hint, sign the locator if
205                         // possible, and return it to the client.
206                         return_hash := fmt.Sprintf("%s+%d", hash, len(buf))
207                         api_token := GetApiToken(req)
208                         if PermissionSecret != nil && api_token != "" {
209                                 expiry := time.Now().Add(permission_ttl)
210                                 return_hash = SignLocator(return_hash, api_token, expiry)
211                         }
212                         resp.Write([]byte(return_hash + "\n"))
213                 } else {
214                         ke := err.(*KeepError)
215                         http.Error(resp, ke.Error(), ke.HTTPCode)
216                 }
217         }
218         return
219 }
220
221 // IndexHandler
222 //     A HandleFunc to address /index and /index/{prefix} requests.
223 //
224 func IndexHandler(resp http.ResponseWriter, req *http.Request) {
225         prefix := mux.Vars(req)["prefix"]
226
227         // Only the data manager may issue /index requests,
228         // and only if enforce_permissions is enabled.
229         // All other requests return 403 Forbidden.
230         api_token := GetApiToken(req)
231         if !enforce_permissions ||
232                 api_token == "" ||
233                 data_manager_token != api_token {
234                 http.Error(resp, PermissionError.Error(), PermissionError.HTTPCode)
235                 return
236         }
237         var index string
238         for _, vol := range KeepVM.Volumes() {
239                 index = index + vol.Index(prefix)
240         }
241         resp.Write([]byte(index))
242 }
243
244 // StatusHandler
245 //     Responds to /status.json requests with the current node status,
246 //     described in a JSON structure.
247 //
248 //     The data given in a status.json response includes:
249 //        volumes - a list of Keep volumes currently in use by this server
250 //          each volume is an object with the following fields:
251 //            * mount_point
252 //            * device_num (an integer identifying the underlying filesystem)
253 //            * bytes_free
254 //            * bytes_used
255 //
256 type VolumeStatus struct {
257         MountPoint string `json:"mount_point"`
258         DeviceNum  uint64 `json:"device_num"`
259         BytesFree  uint64 `json:"bytes_free"`
260         BytesUsed  uint64 `json:"bytes_used"`
261 }
262
263 type NodeStatus struct {
264         Volumes []*VolumeStatus `json:"volumes"`
265 }
266
267 func StatusHandler(resp http.ResponseWriter, req *http.Request) {
268         st := GetNodeStatus()
269         if jstat, err := json.Marshal(st); err == nil {
270                 resp.Write(jstat)
271         } else {
272                 log.Printf("json.Marshal: %s\n", err)
273                 log.Printf("NodeStatus = %v\n", st)
274                 http.Error(resp, err.Error(), 500)
275         }
276 }
277
278 // GetNodeStatus
279 //     Returns a NodeStatus struct describing this Keep
280 //     node's current status.
281 //
282 func GetNodeStatus() *NodeStatus {
283         st := new(NodeStatus)
284
285         st.Volumes = make([]*VolumeStatus, len(KeepVM.Volumes()))
286         for i, vol := range KeepVM.Volumes() {
287                 st.Volumes[i] = vol.Status()
288         }
289         return st
290 }
291
292 // GetVolumeStatus
293 //     Returns a VolumeStatus describing the requested volume.
294 //
295 func GetVolumeStatus(volume string) *VolumeStatus {
296         var fs syscall.Statfs_t
297         var devnum uint64
298
299         if fi, err := os.Stat(volume); err == nil {
300                 devnum = fi.Sys().(*syscall.Stat_t).Dev
301         } else {
302                 log.Printf("GetVolumeStatus: os.Stat: %s\n", err)
303                 return nil
304         }
305
306         err := syscall.Statfs(volume, &fs)
307         if err != nil {
308                 log.Printf("GetVolumeStatus: statfs: %s\n", err)
309                 return nil
310         }
311         // These calculations match the way df calculates disk usage:
312         // "free" space is measured by fs.Bavail, but "used" space
313         // uses fs.Blocks - fs.Bfree.
314         free := fs.Bavail * uint64(fs.Bsize)
315         used := (fs.Blocks - fs.Bfree) * uint64(fs.Bsize)
316         return &VolumeStatus{volume, devnum, free, used}
317 }
318
319 // DeleteHandler processes DELETE requests.
320 //
321 // DELETE /{hash:[0-9a-f]{32} will delete the block with the specified hash
322 // from all connected volumes.
323 //
324 // Only the Data Manager, or an Arvados admin with scope "all", are
325 // allowed to issue DELETE requests.  If a DELETE request is not
326 // authenticated or is issued by a non-admin user, the server returns
327 // a PermissionError.
328 //
329 // Upon receiving a valid request from an authorized user,
330 // DeleteHandler deletes all copies of the specified block on local
331 // volumes.
332 //
333 // Response format:
334 //
335 // The response body consists of the JSON message
336 //
337 //    {"copies_deleted":d,"copies_failed":f}
338 //
339 // where d and f are integers representing the number of blocks that
340 // were successfully and unsuccessfully deleted.
341 //
342 //   * If any blocks were successfully deleted (copies_deleted > 0), the
343 //     HTTP response code is 200 OK.
344 //
345 //   * If no blocks were found at all (copies_deleted == copies_failed
346 //     == 0), the response code is 404 Not Found.
347 //
348 //   * If blocks were found but none could be deleted (copies_deleted
349 //     == 0 and copies_failed > 0), the response code is 405 Method Not
350 //     Allowed.
351 //
352 func DeleteHandler(resp http.ResponseWriter, req *http.Request) {
353         hash := mux.Vars(req)["hash"]
354         log.Printf("%s %s", req.Method, hash)
355
356         // Confirm that this user is an admin and has a token with unlimited scope.
357         var tok = GetApiToken(req)
358         if tok == "" || !CanDelete(tok) {
359                 http.Error(resp, PermissionError.Error(), PermissionError.HTTPCode)
360                 return
361         }
362
363         // Delete copies of this block from all available volumes.  Report
364         // how many blocks were successfully and unsuccessfully
365         // deleted.
366         var result struct {
367                 Deleted int `json:"copies_deleted"`
368                 Failed  int `json:"copies_failed"`
369         }
370         for _, vol := range KeepVM.Volumes() {
371                 if err := vol.Delete(hash); err == nil {
372                         result.Deleted++
373                 } else if os.IsNotExist(err) {
374                         continue
375                 } else {
376                         result.Failed++
377                         log.Println("DeleteHandler:", err)
378                 }
379         }
380
381         if j, err := json.Marshal(result); err == nil {
382                 if result.Deleted == 0 && result.Failed == 0 {
383                         // If no blocks were found, HTTP 404
384                         resp.WriteHeader(http.StatusNotFound)
385                 } else if result.Deleted == 0 && result.Failed > 0 {
386                         // If all delete attempts failed, HTTP 405
387                         resp.WriteHeader(http.StatusMethodNotAllowed)
388                 } else {
389                         resp.WriteHeader(http.StatusOK)
390                 }
391                 resp.Write(j)
392         } else {
393                 log.Printf("json.Marshal: %s\n", err)
394                 log.Printf("result = %v\n", result)
395                 http.Error(resp, err.Error(), 500)
396         }
397 }
398
399 func GetBlock(hash string) ([]byte, error) {
400         // Attempt to read the requested hash from a keep volume.
401         error_to_caller := NotFoundError
402
403         for _, vol := range KeepVM.Volumes() {
404                 if buf, err := vol.Get(hash); err != nil {
405                         // IsNotExist is an expected error and may be ignored.
406                         // (If all volumes report IsNotExist, we return a NotFoundError)
407                         // All other errors should be logged but we continue trying to
408                         // read.
409                         switch {
410                         case os.IsNotExist(err):
411                                 continue
412                         default:
413                                 log.Printf("GetBlock: reading %s: %s\n", hash, err)
414                         }
415                 } else {
416                         // Double check the file checksum.
417                         //
418                         filehash := fmt.Sprintf("%x", md5.Sum(buf))
419                         if filehash != hash {
420                                 // TODO(twp): this condition probably represents a bad disk and
421                                 // should raise major alarm bells for an administrator: e.g.
422                                 // they should be sent directly to an event manager at high
423                                 // priority or logged as urgent problems.
424                                 //
425                                 log.Printf("%s: checksum mismatch for request %s (actual %s)\n",
426                                         vol, hash, filehash)
427                                 error_to_caller = DiskHashError
428                         } else {
429                                 // Success!
430                                 if error_to_caller != NotFoundError {
431                                         log.Printf("%s: checksum mismatch for request %s but a good copy was found on another volume and returned\n",
432                                                 vol, hash)
433                                 }
434                                 return buf, nil
435                         }
436                 }
437         }
438
439         if error_to_caller != NotFoundError {
440                 log.Printf("%s: checksum mismatch, no good copy found\n", hash)
441         }
442         return nil, error_to_caller
443 }
444
445 /* PutBlock(block, hash)
446    Stores the BLOCK (identified by the content id HASH) in Keep.
447
448    The MD5 checksum of the block must be identical to the content id HASH.
449    If not, an error is returned.
450
451    PutBlock stores the BLOCK on the first Keep volume with free space.
452    A failure code is returned to the user only if all volumes fail.
453
454    On success, PutBlock returns nil.
455    On failure, it returns a KeepError with one of the following codes:
456
457    500 Collision
458           A different block with the same hash already exists on this
459           Keep server.
460    422 MD5Fail
461           The MD5 hash of the BLOCK does not match the argument HASH.
462    503 Full
463           There was not enough space left in any Keep volume to store
464           the object.
465    500 Fail
466           The object could not be stored for some other reason (e.g.
467           all writes failed). The text of the error message should
468           provide as much detail as possible.
469 */
470
471 func PutBlock(block []byte, hash string) error {
472         // Check that BLOCK's checksum matches HASH.
473         blockhash := fmt.Sprintf("%x", md5.Sum(block))
474         if blockhash != hash {
475                 log.Printf("%s: MD5 checksum %s did not match request", hash, blockhash)
476                 return RequestHashError
477         }
478
479         // If we already have a block on disk under this identifier, return
480         // success (but check for MD5 collisions).
481         // The only errors that GetBlock can return are DiskHashError and NotFoundError.
482         // In either case, we want to write our new (good) block to disk,
483         // so there is nothing special to do if err != nil.
484         if oldblock, err := GetBlock(hash); err == nil {
485                 if bytes.Compare(block, oldblock) == 0 {
486                         return nil
487                 } else {
488                         return CollisionError
489                 }
490         }
491
492         // Choose a Keep volume to write to.
493         // If this volume fails, try all of the volumes in order.
494         vol := KeepVM.Choose()
495         if err := vol.Put(hash, block); err == nil {
496                 return nil // success!
497         } else {
498                 allFull := true
499                 for _, vol := range KeepVM.Volumes() {
500                         err := vol.Put(hash, block)
501                         if err == nil {
502                                 return nil // success!
503                         }
504                         if err != FullError {
505                                 // The volume is not full but the write did not succeed.
506                                 // Report the error and continue trying.
507                                 allFull = false
508                                 log.Printf("%s: Write(%s): %s\n", vol, hash, err)
509                         }
510                 }
511
512                 if allFull {
513                         log.Printf("all Keep volumes full")
514                         return FullError
515                 } else {
516                         log.Printf("all Keep volumes failed")
517                         return GenericError
518                 }
519         }
520 }
521
522 // IsValidLocator
523 //     Return true if the specified string is a valid Keep locator.
524 //     When Keep is extended to support hash types other than MD5,
525 //     this should be updated to cover those as well.
526 //
527 func IsValidLocator(loc string) bool {
528         match, err := regexp.MatchString(`^[0-9a-f]{32}$`, loc)
529         if err == nil {
530                 return match
531         }
532         log.Printf("IsValidLocator: %s\n", err)
533         return false
534 }
535
536 // GetApiToken returns the OAuth2 token from the Authorization
537 // header of a HTTP request, or an empty string if no matching
538 // token is found.
539 func GetApiToken(req *http.Request) string {
540         if auth, ok := req.Header["Authorization"]; ok {
541                 if pat, err := regexp.Compile(`^OAuth2\s+(.*)`); err != nil {
542                         log.Println(err)
543                 } else if match := pat.FindStringSubmatch(auth[0]); match != nil {
544                         return match[1]
545                 }
546         }
547         return ""
548 }
549
550 // IsExpired returns true if the given Unix timestamp (expressed as a
551 // hexadecimal string) is in the past, or if timestamp_hex cannot be
552 // parsed as a hexadecimal string.
553 func IsExpired(timestamp_hex string) bool {
554         ts, err := strconv.ParseInt(timestamp_hex, 16, 0)
555         if err != nil {
556                 log.Printf("IsExpired: %s\n", err)
557                 return true
558         }
559         return time.Unix(ts, 0).Before(time.Now())
560 }
561
562 // CanDelete returns true if the user identified by api_token is
563 // allowed to delete blocks.
564 func CanDelete(api_token string) bool {
565         if api_token == "" {
566                 return false
567         }
568         // Blocks may be deleted only when Keep has been configured with a
569         // data manager.
570         if data_manager_token == "" {
571                 return false
572         }
573         if api_token == data_manager_token {
574                 return true
575         }
576         // TODO(twp): look up api_token with the API server
577         // return true if is_admin is true and if the token
578         // has unlimited scope
579         return false
580 }