"crypto/md5"
"encoding/json"
"fmt"
- "github.com/gorilla/mux"
"io"
- "log"
"net/http"
"os"
"regexp"
"strings"
"sync"
"time"
+
+ "github.com/gorilla/mux"
+
+ "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")
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`, 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")
// 400 Bad Request.
rest.NotFoundHandler = http.HandlerFunc(BadRequestHandler)
- return rest
+ return rtr
}
// BadRequestHandler is a HandleFunc to address bad requests.
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
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
+// 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 {
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"`
- BufferPool PoolStatus
- PullQueue WorkQueueStatus
- TrashQueue WorkQueueStatus
- Memory runtime.MemStats
+ 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 (rtr *router) 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) {
+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 {
}
// 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([]*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)
+ 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
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.
type TrashRequest struct {
Locator string `json:"locator"`
BlockMtime int64 `json:"block_mtime"`
+
+ // Target mount, or "" for "everywhere"
+ MountUUID string
}
// TrashHandler processes /trash requests.
// 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.
// 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
projects / arvados.git / blobdiff