10383: Merge branch 'master' into 10383-arv-put-incremental-upload

[arvados.git] / services / keepstore / handlers.go

diff --git a/services/keepstore/handlers.go b/services/keepstore/handlers.go
index 2c680d355e1268caa1d953dbb5353e3e68715f83..b51009ea4d78348fe53bab01c13070d950e6aa42 100644 (file)
--- a/services/keepstore/handlers.go
+++ b/services/keepstore/handlers.go
@@ -46,6 +46,9 @@ func MakeRESTRouter() *mux.Router {
        // Privileged client only.
        rest.HandleFunc(`/index/{prefix:[0-9a-f]{0,32}}`, IndexHandler).Methods("GET", "HEAD")
 
+       // Internals/debugging info (runtime.MemStats)
+       rest.HandleFunc(`/debug.json`, DebugHandler).Methods("GET", "HEAD")
+
        // List volumes: path, device number, bytes used/avail.
        rest.HandleFunc(`/status.json`, StatusHandler).Methods("GET", "HEAD")
 
@@ -239,18 +242,6 @@ func IndexHandler(resp http.ResponseWriter, req *http.Request) {
        resp.Write([]byte{'\n'})
 }
 
-// StatusHandler
-//     Responds to /status.json requests with the current node status,
-//     described in a JSON structure.
-//
-//     The data given in a status.json response includes:
-//        volumes - a list of Keep volumes currently in use by this server
-//          each volume is an object with the following fields:
-//            * mount_point
-//            * device_num (an integer identifying the underlying filesystem)
-//            * bytes_free
-//            * bytes_used
-
 // PoolStatus struct
 type PoolStatus struct {
        Alloc uint64 `json:"BytesAllocated"`
@@ -258,18 +249,37 @@ type PoolStatus struct {
        Len   int    `json:"BuffersInUse"`
 }
 
+type volumeStatusEnt struct {
+       Label         string
+       Status        *VolumeStatus `json:",omitempty"`
+       VolumeStats   *ioStats      `json:",omitempty"`
+       InternalStats interface{}   `json:",omitempty"`
+}
+
 // NodeStatus struct
 type NodeStatus struct {
-       Volumes    []*VolumeStatus `json:"volumes"`
+       Volumes    []*volumeStatusEnt
        BufferPool PoolStatus
        PullQueue  WorkQueueStatus
        TrashQueue WorkQueueStatus
-       Memory     runtime.MemStats
 }
 
 var st NodeStatus
 var stLock sync.Mutex
 
+// DebugHandler addresses /debug.json requests.
+func DebugHandler(resp http.ResponseWriter, req *http.Request) {
+       type debugStats struct {
+               MemStats runtime.MemStats
+       }
+       var ds debugStats
+       runtime.ReadMemStats(&ds.MemStats)
+       err := json.NewEncoder(resp).Encode(&ds)
+       if err != nil {
+               http.Error(resp, err.Error(), 500)
+       }
+}
+
 // StatusHandler addresses /status.json requests.
 func StatusHandler(resp http.ResponseWriter, req *http.Request) {
        stLock.Lock()
@@ -289,20 +299,26 @@ func StatusHandler(resp http.ResponseWriter, req *http.Request) {
 func readNodeStatus(st *NodeStatus) {
        vols := KeepVM.AllReadable()
        if cap(st.Volumes) < len(vols) {
-               st.Volumes = make([]*VolumeStatus, len(vols))
+               st.Volumes = make([]*volumeStatusEnt, len(vols))
        }
        st.Volumes = st.Volumes[:0]
        for _, vol := range vols {
-               if s := vol.Status(); s != nil {
-                       st.Volumes = append(st.Volumes, s)
+               var internalStats interface{}
+               if vol, ok := vol.(InternalStatser); ok {
+                       internalStats = vol.InternalStats()
                }
+               st.Volumes = append(st.Volumes, &volumeStatusEnt{
+                       Label:         vol.String(),
+                       Status:        vol.Status(),
+                       InternalStats: internalStats,
+                       //VolumeStats: KeepVM.VolumeStats(vol),
+               })
        }
        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)
 }
 
 // return a WorkQueueStatus for the given queue. If q is nil (which
@@ -647,8 +663,10 @@ func PutBlock(ctx context.Context, block []byte, hash string) (int, error) {
        // 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 n, err := CompareAndTouch(hash, block); err == nil || err == CollisionError {
+       if n, err := CompareAndTouch(ctx, hash, block); err == nil || err == CollisionError {
                return n, err
+       } else if ctx.Err() != nil {
+               return 0, ErrClientDisconnect
        }
 
        // Choose a Keep volume to write to.
@@ -699,10 +717,13 @@ func PutBlock(ctx context.Context, block []byte, hash string) (int, error) {
 // the relevant block's modification time in order to protect it from
 // premature garbage collection. Otherwise, it returns a non-nil
 // error.
-func CompareAndTouch(hash string, buf []byte) (int, error) {
+func CompareAndTouch(ctx context.Context, hash string, buf []byte) (int, error) {
        var bestErr error = NotFoundError
        for _, vol := range KeepVM.AllWritable() {
-               if err := vol.Compare(hash, buf); err == CollisionError {
+               err := vol.Compare(ctx, hash, buf)
+               if ctx.Err() != nil {
+                       return 0, ctx.Err()
+               } else if 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