3 // REST handlers for Keep are implemented here.
5 // GetBlockHandler (GET /locator)
6 // PutBlockHandler (PUT /locator)
7 // IndexHandler (GET /index, GET /index/prefix)
8 // StatusHandler (GET /status.json)
16 "github.com/gorilla/mux"
29 // MakeRESTRouter returns a new mux.Router that forwards all Keep
30 // requests to the appropriate handlers.
32 func MakeRESTRouter() *mux.Router {
33 rest := mux.NewRouter()
36 `/{hash:[0-9a-f]{32}}`, GetBlockHandler).Methods("GET", "HEAD")
38 `/{hash:[0-9a-f]{32}}+{hints}`,
39 GetBlockHandler).Methods("GET", "HEAD")
41 rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, PutBlockHandler).Methods("PUT")
42 rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, DeleteHandler).Methods("DELETE")
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
56 rest.HandleFunc(`/index`, IndexHandler).Methods("GET", "HEAD")
58 `/index/{prefix:[0-9a-f]{0,32}}`, IndexHandler).Methods("GET", "HEAD")
59 rest.HandleFunc(`/status.json`, StatusHandler).Methods("GET", "HEAD")
61 // Any request which does not match any of these routes gets
63 rest.NotFoundHandler = http.HandlerFunc(BadRequestHandler)
68 func BadRequestHandler(w http.ResponseWriter, r *http.Request) {
69 http.Error(w, BadRequestError.Error(), BadRequestError.HTTPCode)
72 // FindKeepVolumes scans all mounted volumes on the system for Keep
73 // volumes, and returns a list of matching paths.
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
79 func FindKeepVolumes() []string {
80 vols := make([]string, 0)
82 if f, err := os.Open(PROC_MOUNTS); err != nil {
83 log.Fatalf("opening %s: %s\n", PROC_MOUNTS, err)
85 scanner := bufio.NewScanner(f)
87 args := strings.Fields(scanner.Text())
88 dev, mount := args[0], args[1]
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)
97 if err := scanner.Err(); err != nil {
104 func GetBlockHandler(resp http.ResponseWriter, req *http.Request) {
105 hash := mux.Vars(req)["hash"]
107 log.Printf("%s %s", req.Method, hash)
109 hints := mux.Vars(req)["hints"]
111 // Parse the locator string and hints from the request.
112 // TODO(twp): implement a Locator type.
113 var signature, timestamp string
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 {
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.
126 // Unknown format; not a valid locator.
127 http.Error(resp, BadRequestError.Error(), BadRequestError.HTTPCode)
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)
139 } else if IsExpired(timestamp) {
140 http.Error(resp, ExpiredError.Error(), ExpiredError.HTTPCode)
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)
151 block, err := GetBlock(hash)
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.
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)
165 http.Error(resp, err.Error(), err.(*KeepError).HTTPCode)
169 resp.Header().Set("X-Block-Size", fmt.Sprintf("%d", len(block)))
171 _, err = resp.Write(block)
173 log.Printf("GetBlockHandler: writing response: %s", err)
179 func PutBlockHandler(resp http.ResponseWriter, req *http.Request) {
180 // Garbage collect after each PUT. Fixes #2865.
181 // See also GetBlockHandler.
184 hash := mux.Vars(req)["hash"]
186 log.Printf("%s %s", req.Method, hash)
188 // Read the block data to be stored.
189 // If the request exceeds BLOCKSIZE bytes, issue a HTTP 500 error.
191 if req.ContentLength > BLOCKSIZE {
192 http.Error(resp, TooLongError.Error(), TooLongError.HTTPCode)
196 buf := make([]byte, req.ContentLength)
197 nread, err := io.ReadFull(req.Body, buf)
199 http.Error(resp, err.Error(), 500)
200 } else if int64(nread) < req.ContentLength {
201 http.Error(resp, "request truncated", 500)
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)
212 resp.Write([]byte(return_hash + "\n"))
214 ke := err.(*KeepError)
215 http.Error(resp, ke.Error(), ke.HTTPCode)
222 // A HandleFunc to address /index and /index/{prefix} requests.
224 func IndexHandler(resp http.ResponseWriter, req *http.Request) {
225 prefix := mux.Vars(req)["prefix"]
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 ||
233 data_manager_token != api_token {
234 http.Error(resp, PermissionError.Error(), PermissionError.HTTPCode)
238 for _, vol := range KeepVM.Volumes() {
239 index = index + vol.Index(prefix)
241 resp.Write([]byte(index))
245 // Responds to /status.json requests with the current node status,
246 // described in a JSON structure.
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:
252 // * device_num (an integer identifying the underlying filesystem)
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"`
263 type NodeStatus struct {
264 Volumes []*VolumeStatus `json:"volumes"`
267 func StatusHandler(resp http.ResponseWriter, req *http.Request) {
268 st := GetNodeStatus()
269 if jstat, err := json.Marshal(st); err == nil {
272 log.Printf("json.Marshal: %s\n", err)
273 log.Printf("NodeStatus = %v\n", st)
274 http.Error(resp, err.Error(), 500)
279 // Returns a NodeStatus struct describing this Keep
280 // node's current status.
282 func GetNodeStatus() *NodeStatus {
283 st := new(NodeStatus)
285 st.Volumes = make([]*VolumeStatus, len(KeepVM.Volumes()))
286 for i, vol := range KeepVM.Volumes() {
287 st.Volumes[i] = vol.Status()
293 // Returns a VolumeStatus describing the requested volume.
295 func GetVolumeStatus(volume string) *VolumeStatus {
296 var fs syscall.Statfs_t
299 if fi, err := os.Stat(volume); err == nil {
300 devnum = fi.Sys().(*syscall.Stat_t).Dev
302 log.Printf("GetVolumeStatus: os.Stat: %s\n", err)
306 err := syscall.Statfs(volume, &fs)
308 log.Printf("GetVolumeStatus: statfs: %s\n", err)
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}
319 // DeleteHandler processes DELETE requests.
321 // DELETE /{hash:[0-9a-f]{32} will delete the block with the specified hash
322 // from all connected volumes.
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.
329 // Upon receiving a valid request from an authorized user,
330 // DeleteHandler deletes all copies of the specified block on local
335 // The response body consists of the JSON message
337 // {"copies_deleted":d,"copies_failed":f}
339 // where d and f are integers representing the number of blocks that
340 // were successfully and unsuccessfully deleted.
342 // * If any blocks were successfully deleted (copies_deleted > 0), the
343 // HTTP response code is 200 OK.
345 // * If no blocks were found at all (copies_deleted == copies_failed
346 // == 0), the response code is 404 Not Found.
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
352 func DeleteHandler(resp http.ResponseWriter, req *http.Request) {
353 hash := mux.Vars(req)["hash"]
354 log.Printf("%s %s", req.Method, hash)
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)
363 // Delete copies of this block from all available volumes. Report
364 // how many blocks were successfully and unsuccessfully
367 Deleted int `json:"copies_deleted"`
368 Failed int `json:"copies_failed"`
370 for _, vol := range KeepVM.Volumes() {
371 if err := vol.Delete(hash); err == nil {
373 } else if os.IsNotExist(err) {
377 log.Println("DeleteHandler:", err)
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)
389 resp.WriteHeader(http.StatusOK)
393 log.Printf("json.Marshal: %s\n", err)
394 log.Printf("result = %v\n", result)
395 http.Error(resp, err.Error(), 500)
399 func GetBlock(hash string) ([]byte, error) {
400 // Attempt to read the requested hash from a keep volume.
401 error_to_caller := NotFoundError
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
410 case os.IsNotExist(err):
413 log.Printf("GetBlock: reading %s: %s\n", hash, err)
416 // Double check the file checksum.
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.
425 log.Printf("%s: checksum mismatch for request %s (actual %s)\n",
427 error_to_caller = DiskHashError
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",
439 if error_to_caller != NotFoundError {
440 log.Printf("%s: checksum mismatch, no good copy found\n", hash)
442 return nil, error_to_caller
445 /* PutBlock(block, hash)
446 Stores the BLOCK (identified by the content id HASH) in Keep.
448 The MD5 checksum of the block must be identical to the content id HASH.
449 If not, an error is returned.
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.
454 On success, PutBlock returns nil.
455 On failure, it returns a KeepError with one of the following codes:
458 A different block with the same hash already exists on this
461 The MD5 hash of the BLOCK does not match the argument HASH.
463 There was not enough space left in any Keep volume to store
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.
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
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 {
488 return CollisionError
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!
499 for _, vol := range KeepVM.Volumes() {
500 err := vol.Put(hash, block)
502 return nil // success!
504 if err != FullError {
505 // The volume is not full but the write did not succeed.
506 // Report the error and continue trying.
508 log.Printf("%s: Write(%s): %s\n", vol, hash, err)
513 log.Printf("all Keep volumes full")
516 log.Printf("all Keep volumes failed")
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.
527 func IsValidLocator(loc string) bool {
528 match, err := regexp.MatchString(`^[0-9a-f]{32}$`, loc)
532 log.Printf("IsValidLocator: %s\n", err)
536 // GetApiToken returns the OAuth2 token from the Authorization
537 // header of a HTTP request, or an empty string if no matching
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 {
543 } else if match := pat.FindStringSubmatch(auth[0]); match != nil {
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)
556 log.Printf("IsExpired: %s\n", err)
559 return time.Unix(ts, 0).Before(time.Now())
562 // CanDelete returns true if the user identified by api_token is
563 // allowed to delete blocks.
564 func CanDelete(api_token string) bool {
568 // Blocks may be deleted only when Keep has been configured with a
570 if data_manager_token == "" {
573 if api_token == data_manager_token {
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