X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/e41260f6a5b14753fcad27fbd0e0515a051a4671..9a34f5ed292fb8d2f262e4c23e758cd689d81db7:/services/keepstore/handlers.go diff --git a/services/keepstore/handlers.go b/services/keepstore/handlers.go index 16c39c2c88..e6129a7376 100644 --- a/services/keepstore/handlers.go +++ b/services/keepstore/handlers.go @@ -8,7 +8,6 @@ package main // StatusHandler (GET /status.json) import ( - "bytes" "container/list" "crypto/md5" "encoding/json" @@ -21,8 +20,7 @@ import ( "regexp" "runtime" "strconv" - "strings" - "syscall" + "sync" "time" ) @@ -67,51 +65,15 @@ func BadRequestHandler(w http.ResponseWriter, r *http.Request) { } func GetBlockHandler(resp http.ResponseWriter, req *http.Request) { - hash := mux.Vars(req)["hash"] - - hints := mux.Vars(req)["hints"] - - // Parse the locator string and hints from the request. - // TODO(twp): implement a Locator type. - var signature, timestamp string - if hints != "" { - signature_pat, _ := regexp.Compile("^A([[:xdigit:]]+)@([[:xdigit:]]{8})$") - for _, hint := range strings.Split(hints, "+") { - if match, _ := regexp.MatchString("^[[:digit:]]+$", hint); match { - // Server ignores size hints - } else if m := signature_pat.FindStringSubmatch(hint); m != nil { - signature = m[1] - timestamp = m[2] - } else if match, _ := regexp.MatchString("^[[:upper:]]", hint); match { - // Any unknown hint that starts with an uppercase letter is - // presumed to be valid and ignored, to permit forward compatibility. - } else { - // Unknown format; not a valid locator. - http.Error(resp, BadRequestError.Error(), BadRequestError.HTTPCode) - return - } - } - } - - // If permission checking is in effect, verify this - // request's permission signature. if enforce_permissions { - if signature == "" || timestamp == "" { - http.Error(resp, PermissionError.Error(), PermissionError.HTTPCode) - return - } else if IsExpired(timestamp) { - http.Error(resp, ExpiredError.Error(), ExpiredError.HTTPCode) + locator := req.URL.Path[1:] // strip leading slash + if err := VerifySignature(locator, GetApiToken(req)); err != nil { + http.Error(resp, err.Error(), err.(*KeepError).HTTPCode) return - } else { - req_locator := req.URL.Path[1:] // strip leading slash - if !VerifySignature(req_locator, GetApiToken(req)) { - http.Error(resp, PermissionError.Error(), PermissionError.HTTPCode) - return - } } } - block, err := GetBlock(hash, false) + block, err := GetBlock(mux.Vars(req)["hash"]) if err != nil { // This type assertion is safe because the only errors // GetBlock can return are DiskHashError or NotFoundError. @@ -126,10 +88,6 @@ func GetBlockHandler(resp http.ResponseWriter, req *http.Request) { } func PutBlockHandler(resp http.ResponseWriter, req *http.Request) { - // Garbage collect after each PUT. Fixes #2865. - // See also GetBlockHandler. - defer runtime.GC() - hash := mux.Vars(req)["hash"] // Detect as many error conditions as possible before reading @@ -203,6 +161,9 @@ func IndexHandler(resp http.ResponseWriter, req *http.Request) { return } } + // An empty line at EOF is the only way the client can be + // assured the entire index was received. + resp.Write([]byte{'\n'}) } // StatusHandler @@ -224,60 +185,66 @@ type VolumeStatus struct { BytesUsed uint64 `json:"bytes_used"` } +type PoolStatus struct { + Alloc uint64 `json:"BytesAllocated"` + Cap int `json:"BuffersMax"` + Len int `json:"BuffersInUse"` +} + type NodeStatus struct { - Volumes []*VolumeStatus `json:"volumes"` + Volumes []*VolumeStatus `json:"volumes"` + BufferPool PoolStatus + PullQueue WorkQueueStatus + TrashQueue WorkQueueStatus + Memory runtime.MemStats } +var st NodeStatus +var stLock sync.Mutex + func StatusHandler(resp http.ResponseWriter, req *http.Request) { - st := GetNodeStatus() - if jstat, err := json.Marshal(st); err == nil { + stLock.Lock() + readNodeStatus(&st) + jstat, err := json.Marshal(&st) + stLock.Unlock() + if err == nil { resp.Write(jstat) } else { - log.Printf("json.Marshal: %s\n", err) - log.Printf("NodeStatus = %v\n", st) + log.Printf("json.Marshal: %s", err) + log.Printf("NodeStatus = %v", &st) http.Error(resp, err.Error(), 500) } } -// GetNodeStatus -// Returns a NodeStatus struct describing this Keep -// node's current status. -// -func GetNodeStatus() *NodeStatus { - st := new(NodeStatus) - - st.Volumes = make([]*VolumeStatus, len(KeepVM.AllReadable())) - for i, vol := range KeepVM.AllReadable() { - st.Volumes[i] = vol.Status() +// populate the given NodeStatus struct with current values. +func readNodeStatus(st *NodeStatus) { + vols := KeepVM.AllReadable() + if cap(st.Volumes) < len(vols) { + st.Volumes = make([]*VolumeStatus, len(vols)) } - return st -} - -// GetVolumeStatus -// Returns a VolumeStatus describing the requested volume. -// -func GetVolumeStatus(volume string) *VolumeStatus { - var fs syscall.Statfs_t - var devnum uint64 - - if fi, err := os.Stat(volume); err == nil { - devnum = fi.Sys().(*syscall.Stat_t).Dev - } else { - log.Printf("GetVolumeStatus: os.Stat: %s\n", err) - return nil + st.Volumes = st.Volumes[:0] + for _, vol := range vols { + if s := vol.Status(); s != nil { + st.Volumes = append(st.Volumes, s) + } } + st.BufferPool.Alloc = bufs.Alloc() + st.BufferPool.Cap = bufs.Cap() + st.BufferPool.Len = bufs.Len() + st.PullQueue = getWorkQueueStatus(pullq) + st.TrashQueue = getWorkQueueStatus(trashq) + runtime.ReadMemStats(&st.Memory) +} - err := syscall.Statfs(volume, &fs) - if err != nil { - log.Printf("GetVolumeStatus: statfs: %s\n", err) - return nil +// return a WorkQueueStatus for the given queue. If q is nil (which +// should never happen except in test suites), return a zero status +// value instead of crashing. +func getWorkQueueStatus(q *WorkQueue) WorkQueueStatus { + if q == nil { + // This should only happen during tests. + return WorkQueueStatus{} } - // These calculations match the way df calculates disk usage: - // "free" space is measured by fs.Bavail, but "used" space - // uses fs.Blocks - fs.Bfree. - free := fs.Bavail * uint64(fs.Bsize) - used := (fs.Blocks - fs.Bfree) * uint64(fs.Bsize) - return &VolumeStatus{volume, devnum, free, used} + return q.Status() } // DeleteHandler processes DELETE requests. @@ -354,7 +321,7 @@ func DeleteHandler(resp http.ResponseWriter, req *http.Request) { if body, err := json.Marshal(result); err == nil { resp.Write(body) } else { - log.Printf("json.Marshal: %s (result = %v)\n", err, result) + log.Printf("json.Marshal: %s (result = %v)", err, result) http.Error(resp, err.Error(), 500) } } @@ -474,10 +441,7 @@ func TrashHandler(resp http.ResponseWriter, req *http.Request) { // which volume to check for fetching blocks, storing blocks, etc. // ============================== -// GetBlock fetches and returns the block identified by "hash". If -// the update_timestamp argument is true, GetBlock also updates the -// block's file modification time (for the sake of PutBlock, which -// must update the file's timestamp when the block already exists). +// GetBlock fetches and returns the block identified by "hash". // // On success, GetBlock returns a byte slice with the block data, and // a nil error. @@ -488,22 +452,11 @@ func TrashHandler(resp http.ResponseWriter, req *http.Request) { // DiskHashError. // -func GetBlock(hash string, update_timestamp bool) ([]byte, error) { +func GetBlock(hash string) ([]byte, error) { // Attempt to read the requested hash from a keep volume. error_to_caller := NotFoundError - var vols []Volume - if update_timestamp { - // Pointless to find the block on an unwritable volume - // because Touch() will fail -- this is as good as - // "not found" for purposes of callers who need to - // update_timestamp. - vols = KeepVM.AllWritable() - } else { - vols = KeepVM.AllReadable() - } - - for _, vol := range vols { + for _, vol := range KeepVM.AllReadable() { buf, err := vol.Get(hash) if err != nil { // IsNotExist is an expected error and may be @@ -512,7 +465,7 @@ func GetBlock(hash string, update_timestamp bool) ([]byte, error) { // volumes. If all volumes report IsNotExist, // we return a NotFoundError. if !os.IsNotExist(err) { - log.Printf("GetBlock: reading %s: %s\n", hash, err) + log.Printf("%s: Get(%s): %s", vol, hash, err) } continue } @@ -522,7 +475,7 @@ func GetBlock(hash string, update_timestamp bool) ([]byte, error) { if filehash != hash { // TODO: Try harder to tell a sysadmin about // this. - log.Printf("%s: checksum mismatch for request %s (actual %s)\n", + log.Printf("%s: checksum mismatch for request %s (actual %s)", vol, hash, filehash) error_to_caller = DiskHashError bufs.Put(buf) @@ -532,15 +485,6 @@ func GetBlock(hash string, update_timestamp bool) ([]byte, error) { log.Printf("%s: checksum mismatch for request %s but a good copy was found on another volume and returned", vol, hash) } - if update_timestamp { - if err := vol.Touch(hash); err != nil { - error_to_caller = GenericError - log.Printf("%s: Touch %s failed: %s", - vol, hash, error_to_caller) - bufs.Put(buf) - continue - } - } return buf, nil } return nil, error_to_caller @@ -580,21 +524,11 @@ func PutBlock(block []byte, hash string) error { return RequestHashError } - // If we already have a block on disk under this identifier, return - // success (but check for MD5 collisions). While fetching the block, - // update its timestamp. - // The only errors that GetBlock can return are DiskHashError and NotFoundError. - // In either case, we want to write our new (good) block to disk, - // so there is nothing special to do if err != nil. - // - if oldblock, err := GetBlock(hash, true); err == nil { - defer bufs.Put(oldblock) - if bytes.Compare(block, oldblock) == 0 { - // The block already exists; return success. - return nil - } else { - return CollisionError - } + // If we already have this data, it's intact on disk, and we + // can update its timestamp, return success. If we have + // different data with the same hash, return failure. + if err := CompareAndTouch(hash, block); err == nil || err == CollisionError { + return err } // Choose a Keep volume to write to. @@ -622,7 +556,7 @@ func PutBlock(block []byte, hash string) error { // write did not succeed. Report the // error and continue trying. allFull = false - log.Printf("%s: Write(%s): %s\n", vol, hash, err) + log.Printf("%s: Write(%s): %s", vol, hash, err) } } @@ -635,28 +569,62 @@ func PutBlock(block []byte, hash string) error { } } +// CompareAndTouch returns nil if one of the volumes already has the +// given content and it successfully updates the relevant block's +// modification time in order to protect it from premature garbage +// collection. +func CompareAndTouch(hash string, buf []byte) error { + var bestErr error = NotFoundError + for _, vol := range KeepVM.AllWritable() { + if err := vol.Compare(hash, buf); err == CollisionError { + // Stop if we have a block with same hash but + // different content. (It will be impossible + // to tell which one is wanted if we have + // both, so there's no point writing it even + // on a different volume.) + log.Printf("%s: Compare(%s): %s", vol, hash, err) + return err + } else if os.IsNotExist(err) { + // Block does not exist. This is the only + // "normal" error: we don't log anything. + continue + } else if err != nil { + // Couldn't open file, data is corrupt on + // disk, etc.: log this abnormal condition, + // and try the next volume. + log.Printf("%s: Compare(%s): %s", vol, hash, err) + continue + } + if err := vol.Touch(hash); err != nil { + log.Printf("%s: Touch %s failed: %s", vol, hash, err) + bestErr = err + continue + } + // Compare and Touch both worked --> done. + return nil + } + return bestErr +} + +var validLocatorRe = regexp.MustCompile(`^[0-9a-f]{32}$`) + // IsValidLocator // Return true if the specified string is a valid Keep locator. // When Keep is extended to support hash types other than MD5, // this should be updated to cover those as well. // func IsValidLocator(loc string) bool { - match, err := regexp.MatchString(`^[0-9a-f]{32}$`, loc) - if err == nil { - return match - } - log.Printf("IsValidLocator: %s\n", err) - return false + return validLocatorRe.MatchString(loc) } +var authRe = regexp.MustCompile(`^OAuth2\s+(.*)`) + // GetApiToken returns the OAuth2 token from the Authorization // header of a HTTP request, or an empty string if no matching // token is found. func GetApiToken(req *http.Request) string { if auth, ok := req.Header["Authorization"]; ok { - if pat, err := regexp.Compile(`^OAuth2\s+(.*)`); err != nil { - log.Println(err) - } else if match := pat.FindStringSubmatch(auth[0]); match != nil { + if match := authRe.FindStringSubmatch(auth[0]); match != nil { return match[1] } } @@ -669,7 +637,7 @@ func GetApiToken(req *http.Request) string { func IsExpired(timestamp_hex string) bool { ts, err := strconv.ParseInt(timestamp_hex, 16, 0) if err != nil { - log.Printf("IsExpired: %s\n", err) + log.Printf("IsExpired: %s", err) return true } return time.Unix(ts, 0).Before(time.Now())