X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/4f4f6f5fe9367d5cd6c57070fd8a223efc87cb21..c4e6d3c9a8a55460c4ee663e66ea1093c6088d4f:/services/keepstore/handlers.go diff --git a/services/keepstore/handlers.go b/services/keepstore/handlers.go index ed0d8d7a5b..2d90aba14e 100644 --- a/services/keepstore/handlers.go +++ b/services/keepstore/handlers.go @@ -1,3 +1,7 @@ +// Copyright (C) The Arvados Authors. All rights reserved. +// +// SPDX-License-Identifier: AGPL-3.0 + package main // REST handlers for Keep are implemented here. @@ -13,7 +17,6 @@ import ( "crypto/md5" "encoding/json" "fmt" - "github.com/gorilla/mux" "io" "net/http" "os" @@ -24,14 +27,23 @@ import ( "sync" "time" + "github.com/gorilla/mux" + + "git.curoverse.com/arvados.git/sdk/go/health" + "git.curoverse.com/arvados.git/sdk/go/httpserver" log "github.com/Sirupsen/logrus" ) -// MakeRESTRouter returns a new mux.Router that forwards all Keep -// requests to the appropriate handlers. -// -func MakeRESTRouter() *mux.Router { +type router struct { + *mux.Router + limiter httpserver.RequestCounter +} + +// MakeRESTRouter returns a new router that forwards all Keep requests +// to the appropriate handlers. +func MakeRESTRouter() *router { rest := mux.NewRouter() + rtr := &router{Router: rest} rest.HandleFunc( `/{hash:[0-9a-f]{32}}`, GetBlockHandler).Methods("GET", "HEAD") @@ -42,16 +54,21 @@ func MakeRESTRouter() *mux.Router { rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, PutBlockHandler).Methods("PUT") rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, DeleteHandler).Methods("DELETE") // List all blocks stored here. Privileged client only. - rest.HandleFunc(`/index`, IndexHandler).Methods("GET", "HEAD") + rest.HandleFunc(`/index`, rtr.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") + rest.HandleFunc(`/index/{prefix:[0-9a-f]{0,32}}`, rtr.IndexHandler).Methods("GET", "HEAD") // Internals/debugging info (runtime.MemStats) - rest.HandleFunc(`/debug.json`, DebugHandler).Methods("GET", "HEAD") + rest.HandleFunc(`/debug.json`, rtr.DebugHandler).Methods("GET", "HEAD") // List volumes: path, device number, bytes used/avail. - rest.HandleFunc(`/status.json`, StatusHandler).Methods("GET", "HEAD") + rest.HandleFunc(`/status.json`, rtr.StatusHandler).Methods("GET", "HEAD") + + // List mounts: UUID, readonly, tier, device ID, ... + rest.HandleFunc(`/mounts`, rtr.MountsHandler).Methods("GET") + rest.HandleFunc(`/mounts/{uuid}/blocks`, rtr.IndexHandler).Methods("GET") + rest.HandleFunc(`/mounts/{uuid}/blocks/`, rtr.IndexHandler).Methods("GET") // Replace the current pull queue. rest.HandleFunc(`/pull`, PullHandler).Methods("PUT") @@ -62,11 +79,16 @@ func MakeRESTRouter() *mux.Router { // Untrash moves blocks from trash back into store rest.HandleFunc(`/untrash/{hash:[0-9a-f]{32}}`, UntrashHandler).Methods("PUT") + rest.Handle("/_health/{check}", &health.Handler{ + Token: theConfig.ManagementToken, + Prefix: "/_health/", + }).Methods("GET") + // Any request which does not match any of these routes gets // 400 Bad Request. rest.NotFoundHandler = http.HandlerFunc(BadRequestHandler) - return rest + return rtr } // BadRequestHandler is a HandleFunc to address bad requests. @@ -216,18 +238,34 @@ func PutBlockHandler(resp http.ResponseWriter, req *http.Request) { resp.Write([]byte(returnHash + "\n")) } -// IndexHandler is a HandleFunc to address /index and /index/{prefix} requests. -func IndexHandler(resp http.ResponseWriter, req *http.Request) { - // Reject unauthorized requests. +// IndexHandler responds to "/index", "/index/{prefix}", and +// "/mounts/{uuid}/blocks" requests. +func (rtr *router) IndexHandler(resp http.ResponseWriter, req *http.Request) { if !IsSystemAuth(GetAPIToken(req)) { http.Error(resp, UnauthorizedError.Error(), UnauthorizedError.HTTPCode) return } prefix := mux.Vars(req)["prefix"] + if prefix == "" { + req.ParseForm() + prefix = req.Form.Get("prefix") + } - for _, vol := range KeepVM.AllReadable() { - if err := vol.IndexTo(prefix, resp); err != nil { + uuid := mux.Vars(req)["uuid"] + + var vols []Volume + if uuid == "" { + vols = KeepVM.AllReadable() + } else if v := KeepVM.Lookup(uuid, false); v == nil { + http.Error(resp, "mount not found", http.StatusNotFound) + return + } else { + vols = []Volume{v} + } + + for _, v := range vols { + if err := v.IndexTo(prefix, resp); err != nil { // The only errors returned by IndexTo are // write errors returned by resp.Write(), // which probably means the client has @@ -243,9 +281,17 @@ func IndexHandler(resp http.ResponseWriter, req *http.Request) { resp.Write([]byte{'\n'}) } +// MountsHandler responds to "GET /mounts" requests. +func (rtr *router) MountsHandler(resp http.ResponseWriter, req *http.Request) { + err := json.NewEncoder(resp).Encode(KeepVM.Mounts()) + if err != nil { + http.Error(resp, err.Error(), http.StatusInternalServerError) + } +} + // PoolStatus struct type PoolStatus struct { - Alloc uint64 `json:"BytesAllocated"` + Alloc uint64 `json:"BytesAllocatedCumulative"` Cap int `json:"BuffersMax"` Len int `json:"BuffersInUse"` } @@ -259,17 +305,19 @@ type volumeStatusEnt struct { // NodeStatus struct type NodeStatus struct { - Volumes []*volumeStatusEnt - BufferPool PoolStatus - PullQueue WorkQueueStatus - TrashQueue WorkQueueStatus + Volumes []*volumeStatusEnt + BufferPool PoolStatus + PullQueue WorkQueueStatus + TrashQueue WorkQueueStatus + RequestsCurrent int + RequestsMax int } var st NodeStatus var stLock sync.Mutex // DebugHandler addresses /debug.json requests. -func DebugHandler(resp http.ResponseWriter, req *http.Request) { +func (rtr *router) DebugHandler(resp http.ResponseWriter, req *http.Request) { type debugStats struct { MemStats runtime.MemStats } @@ -282,9 +330,9 @@ func DebugHandler(resp http.ResponseWriter, req *http.Request) { } // StatusHandler addresses /status.json requests. -func StatusHandler(resp http.ResponseWriter, req *http.Request) { +func (rtr *router) StatusHandler(resp http.ResponseWriter, req *http.Request) { stLock.Lock() - readNodeStatus(&st) + rtr.readNodeStatus(&st) jstat, err := json.Marshal(&st) stLock.Unlock() if err == nil { @@ -297,7 +345,7 @@ func StatusHandler(resp http.ResponseWriter, req *http.Request) { } // populate the given NodeStatus struct with current values. -func readNodeStatus(st *NodeStatus) { +func (rtr *router) readNodeStatus(st *NodeStatus) { vols := KeepVM.AllReadable() if cap(st.Volumes) < len(vols) { st.Volumes = make([]*volumeStatusEnt, len(vols)) @@ -320,6 +368,10 @@ func readNodeStatus(st *NodeStatus) { st.BufferPool.Len = bufs.Len() st.PullQueue = getWorkQueueStatus(pullq) st.TrashQueue = getWorkQueueStatus(trashq) + if rtr.limiter != nil { + st.RequestsCurrent = rtr.limiter.Current() + st.RequestsMax = rtr.limiter.Max() + } } // return a WorkQueueStatus for the given queue. If q is nil (which @@ -450,6 +502,9 @@ func DeleteHandler(resp http.ResponseWriter, req *http.Request) { type PullRequest struct { Locator string `json:"locator"` Servers []string `json:"servers"` + + // Destination mount, or "" for "anywhere" + MountUUID string } // PullHandler processes "PUT /pull" requests for the data manager. @@ -486,6 +541,9 @@ func PullHandler(resp http.ResponseWriter, req *http.Request) { type TrashRequest struct { Locator string `json:"locator"` BlockMtime int64 `json:"block_mtime"` + + // Target mount, or "" for "everywhere" + MountUUID string } // TrashHandler processes /trash requests.