"regexp"
"runtime"
"strconv"
- "strings"
- "syscall"
+ "sync"
"time"
)
rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, PutBlockHandler).Methods("PUT")
rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, DeleteHandler).Methods("DELETE")
-
- // For IndexHandler we support:
- // /index - returns all locators
- // /index/{prefix} - returns all locators that begin with {prefix}
- // {prefix} is a string of hexadecimal digits between 0 and 32 digits.
- // If {prefix} is the empty string, return an index of all locators
- // (so /index and /index/ behave identically)
- // A client may supply a full 32-digit locator string, in which
- // case the server will return an index with either zero or one
- // entries. This usage allows a client to check whether a block is
- // present, and its size and upload time, without retrieving the
- // entire block.
- //
+ // List all blocks stored here. Privileged client only.
rest.HandleFunc(`/index`, IndexHandler).Methods("GET", "HEAD")
- rest.HandleFunc(
- `/index/{prefix:[0-9a-f]{0,32}}`, IndexHandler).Methods("GET", "HEAD")
+ // List blocks stored here whose hash has the given prefix.
+ // Privileged client only.
+ rest.HandleFunc(`/index/{prefix:[0-9a-f]{0,32}}`, IndexHandler).Methods("GET", "HEAD")
+
+ // List volumes: path, device number, bytes used/avail.
rest.HandleFunc(`/status.json`, StatusHandler).Methods("GET", "HEAD")
- // The PullHandler and TrashHandler process "PUT /pull" and "PUT
- // /trash" requests from Data Manager. These requests instruct
- // Keep to replicate or delete blocks; see
- // https://arvados.org/projects/arvados/wiki/Keep_Design_Doc
- // for more details.
- //
- // Each handler parses the JSON list of block management requests
- // in the message body, and replaces any existing pull queue or
- // trash queue with their contentes.
- //
+ // Replace the current pull queue.
rest.HandleFunc(`/pull`, PullHandler).Methods("PUT")
+
+ // Replace the current trash queue.
rest.HandleFunc(`/trash`, TrashHandler).Methods("PUT")
// Any request which does not match any of these routes gets
}
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)
-
- // Garbage collect after each GET. Fixes #2865.
- // TODO(twp): review Keep memory usage and see if there's
- // a better way to do this than blindly garbage collecting
- // after every block.
- defer runtime.GC()
-
+ block, err := GetBlock(mux.Vars(req)["hash"], false)
if err != nil {
// This type assertion is safe because the only errors
// GetBlock can return are DiskHashError or NotFoundError.
http.Error(resp, err.Error(), err.(*KeepError).HTTPCode)
return
}
+ defer bufs.Put(block)
- resp.Header().Set("Content-Length", fmt.Sprintf("%d", len(block)))
-
- _, err = resp.Write(block)
-
- return
+ resp.Header().Set("Content-Length", strconv.Itoa(len(block)))
+ resp.Header().Set("Content-Type", "application/octet-stream")
+ resp.Write(block)
}
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
return
}
- buf := make([]byte, req.ContentLength)
- nread, err := io.ReadFull(req.Body, buf)
+ buf := bufs.Get(int(req.ContentLength))
+ _, err := io.ReadFull(req.Body, buf)
if err != nil {
http.Error(resp, err.Error(), 500)
- return
- } else if int64(nread) < req.ContentLength {
- http.Error(resp, "request truncated", 500)
+ bufs.Put(buf)
return
}
err = PutBlock(buf, hash)
+ bufs.Put(buf)
+
if err != nil {
ke := err.(*KeepError)
http.Error(resp, ke.Error(), ke.HTTPCode)
// Success; add a size hint, sign the locator if possible, and
// return it to the client.
- return_hash := fmt.Sprintf("%s+%d", hash, len(buf))
+ return_hash := fmt.Sprintf("%s+%d", hash, req.ContentLength)
api_token := GetApiToken(req)
if PermissionSecret != nil && api_token != "" {
expiry := time.Now().Add(blob_signature_ttl)
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
BytesUsed uint64 `json:"bytes_used"`
}
+type PoolStatus struct {
+ Alloc uint64 `json:"BytesAllocated"`
+ Cap int `json:"BuffersMax"`
+ Len int `json:"BuffersInUse"`
+}
+
+type WorkQueueStatus struct {
+ InProgress int
+ Outstanding int
+ Queued int
+}
+
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("NodeStatus = %v\n", &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))
+ }
+ st.Volumes = st.Volumes[:0]
+ for _, vol := range vols {
+ if s := vol.Status(); s != nil {
+ st.Volumes = append(st.Volumes, s)
+ }
}
- return st
+ st.BufferPool.Alloc = bufs.Alloc()
+ st.BufferPool.Cap = bufs.Cap()
+ st.BufferPool.Len = bufs.Len()
+ readWorkQueueStatus(&st.PullQueue, pullq)
+ readWorkQueueStatus(&st.TrashQueue, trashq)
+ runtime.ReadMemStats(&st.Memory)
}
-// 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
+// Populate a WorkQueueStatus. This is not atomic, so race conditions
+// can cause InProgress + Queued != Outstanding.
+func readWorkQueueStatus(st *WorkQueueStatus, q *WorkQueue) {
+ if q == nil {
+ // This should only happen during tests.
+ *st = WorkQueueStatus{}
+ return
}
-
- err := syscall.Statfs(volume, &fs)
- if err != nil {
- log.Printf("GetVolumeStatus: statfs: %s\n", err)
- return nil
- }
- // 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}
+ st.InProgress = q.CountInProgress()
+ st.Outstanding = q.CountOutstanding()
+ st.Queued = q.CountQueued()
}
// DeleteHandler processes DELETE requests.
var pr []PullRequest
r := json.NewDecoder(req.Body)
if err := r.Decode(&pr); err != nil {
- http.Error(resp, BadRequestError.Error(), BadRequestError.HTTPCode)
+ http.Error(resp, err.Error(), BadRequestError.HTTPCode)
return
}
var trash []TrashRequest
r := json.NewDecoder(req.Body)
if err := r.Decode(&trash); err != nil {
- http.Error(resp, BadRequestError.Error(), BadRequestError.HTTPCode)
+ http.Error(resp, err.Error(), BadRequestError.HTTPCode)
return
}
log.Printf("%s: checksum mismatch for request %s (actual %s)\n",
vol, hash, filehash)
error_to_caller = DiskHashError
+ bufs.Put(buf)
continue
}
if error_to_caller == DiskHashError {
error_to_caller = GenericError
log.Printf("%s: Touch %s failed: %s",
vol, hash, error_to_caller)
+ bufs.Put(buf)
continue
}
}
// 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
}
}
+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]
}
}
projects / arvados.git / blobdiff