X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/f17da3a5f54ccaad3ec4f38dedac8b6c50a5cb0a..5b863886118890cc81b728a3a606ea823c836f2b:/services/keepstore/handlers.go diff --git a/services/keepstore/handlers.go b/services/keepstore/handlers.go index adaaa361e9..51dd73a513 100644 --- a/services/keepstore/handlers.go +++ b/services/keepstore/handlers.go @@ -1,11 +1,8 @@ -package main - -// REST handlers for Keep are implemented here. +// Copyright (C) The Arvados Authors. All rights reserved. // -// GetBlockHandler (GET /locator) -// PutBlockHandler (PUT /locator) -// IndexHandler (GET /index, GET /index/prefix) -// StatusHandler (GET /status.json) +// SPDX-License-Identifier: AGPL-3.0 + +package main import ( "container/list" @@ -13,7 +10,6 @@ import ( "crypto/md5" "encoding/json" "fmt" - "github.com/gorilla/mux" "io" "net/http" "os" @@ -24,55 +20,82 @@ import ( "sync" "time" + "git.curoverse.com/arvados.git/sdk/go/arvados" + "git.curoverse.com/arvados.git/sdk/go/health" "git.curoverse.com/arvados.git/sdk/go/httpserver" - log "github.com/Sirupsen/logrus" + "github.com/gorilla/mux" + "github.com/prometheus/client_golang/prometheus" ) type router struct { *mux.Router - limiter httpserver.RequestCounter + limiter httpserver.RequestCounter + cluster *arvados.Cluster + remoteProxy remoteProxy + metrics *nodeMetrics } // MakeRESTRouter returns a new router that forwards all Keep requests // to the appropriate handlers. -func MakeRESTRouter() *router { - rest := mux.NewRouter() - rtr := &router{Router: rest} +func MakeRESTRouter(cluster *arvados.Cluster, reg *prometheus.Registry) http.Handler { + rtr := &router{ + Router: mux.NewRouter(), + cluster: cluster, + metrics: &nodeMetrics{reg: reg}, + } - rest.HandleFunc( - `/{hash:[0-9a-f]{32}}`, GetBlockHandler).Methods("GET", "HEAD") - rest.HandleFunc( + rtr.HandleFunc( + `/{hash:[0-9a-f]{32}}`, rtr.handleGET).Methods("GET", "HEAD") + rtr.HandleFunc( `/{hash:[0-9a-f]{32}}+{hints}`, - GetBlockHandler).Methods("GET", "HEAD") + rtr.handleGET).Methods("GET", "HEAD") - rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, PutBlockHandler).Methods("PUT") - rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, DeleteHandler).Methods("DELETE") + rtr.HandleFunc(`/{hash:[0-9a-f]{32}}`, rtr.handlePUT).Methods("PUT") + rtr.HandleFunc(`/{hash:[0-9a-f]{32}}`, DeleteHandler).Methods("DELETE") // List all blocks stored here. Privileged client only. - rest.HandleFunc(`/index`, IndexHandler).Methods("GET", "HEAD") + rtr.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") + rtr.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") + rtr.HandleFunc(`/debug.json`, rtr.DebugHandler).Methods("GET", "HEAD") // List volumes: path, device number, bytes used/avail. - rest.HandleFunc(`/status.json`, rtr.StatusHandler).Methods("GET", "HEAD") + rtr.HandleFunc(`/status.json`, rtr.StatusHandler).Methods("GET", "HEAD") + + // List mounts: UUID, readonly, tier, device ID, ... + rtr.HandleFunc(`/mounts`, rtr.MountsHandler).Methods("GET") + rtr.HandleFunc(`/mounts/{uuid}/blocks`, rtr.IndexHandler).Methods("GET") + rtr.HandleFunc(`/mounts/{uuid}/blocks/`, rtr.IndexHandler).Methods("GET") // Replace the current pull queue. - rest.HandleFunc(`/pull`, PullHandler).Methods("PUT") + rtr.HandleFunc(`/pull`, PullHandler).Methods("PUT") // Replace the current trash queue. - rest.HandleFunc(`/trash`, TrashHandler).Methods("PUT") + rtr.HandleFunc(`/trash`, TrashHandler).Methods("PUT") // Untrash moves blocks from trash back into store - rest.HandleFunc(`/untrash/{hash:[0-9a-f]{32}}`, UntrashHandler).Methods("PUT") + rtr.HandleFunc(`/untrash/{hash:[0-9a-f]{32}}`, UntrashHandler).Methods("PUT") + + rtr.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) + rtr.NotFoundHandler = http.HandlerFunc(BadRequestHandler) - return rtr + rtr.limiter = httpserver.NewRequestLimiter(theConfig.MaxRequests, rtr) + rtr.metrics.setupBufferPoolMetrics(bufs) + rtr.metrics.setupWorkQueueMetrics(pullq, "pull") + rtr.metrics.setupWorkQueueMetrics(trashq, "trash") + rtr.metrics.setupRequestMetrics(rtr.limiter) + + instrumented := httpserver.Instrument(rtr.metrics.reg, nil, + httpserver.AddRequestIDs(httpserver.LogRequests(nil, rtr.limiter))) + return instrumented.ServeAPI(theConfig.ManagementToken, instrumented) } // BadRequestHandler is a HandleFunc to address bad requests. @@ -80,11 +103,16 @@ func BadRequestHandler(w http.ResponseWriter, r *http.Request) { http.Error(w, BadRequestError.Error(), BadRequestError.HTTPCode) } -// GetBlockHandler is a HandleFunc to address Get block requests. -func GetBlockHandler(resp http.ResponseWriter, req *http.Request) { +func (rtr *router) handleGET(resp http.ResponseWriter, req *http.Request) { ctx, cancel := contextForResponse(context.TODO(), resp) defer cancel() + locator := req.URL.Path[1:] + if strings.Contains(locator, "+R") && !strings.Contains(locator, "+A") { + rtr.remoteProxy.Get(ctx, resp, req, rtr.cluster) + return + } + if theConfig.RequireSignatures { locator := req.URL.Path[1:] // strip leading slash if err := VerifySignature(locator, GetAPIToken(req)); err != nil { @@ -159,8 +187,7 @@ func getBufferWithContext(ctx context.Context, bufs *bufferPool, bufSize int) ([ } } -// PutBlockHandler is a HandleFunc to address Put block requests. -func PutBlockHandler(resp http.ResponseWriter, req *http.Request) { +func (rtr *router) handlePUT(resp http.ResponseWriter, req *http.Request) { ctx, cancel := contextForResponse(context.TODO(), resp) defer cancel() @@ -222,18 +249,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 @@ -249,9 +292,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"` } @@ -271,6 +322,7 @@ type NodeStatus struct { TrashQueue WorkQueueStatus RequestsCurrent int RequestsMax int + Version string } var st NodeStatus @@ -306,6 +358,7 @@ func (rtr *router) StatusHandler(resp http.ResponseWriter, req *http.Request) { // populate the given NodeStatus struct with current values. func (rtr *router) readNodeStatus(st *NodeStatus) { + st.Version = version vols := KeepVM.AllReadable() if cap(st.Volumes) < len(vols) { st.Volumes = make([]*volumeStatusEnt, len(vols)) @@ -462,6 +515,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 `json:"mount_uuid"` } // PullHandler processes "PUT /pull" requests for the data manager. @@ -494,10 +550,13 @@ func PullHandler(resp http.ResponseWriter, req *http.Request) { pullq.ReplaceQueue(plist) } -// TrashRequest consists of a block locator and it's Mtime +// TrashRequest consists of a block locator and its Mtime type TrashRequest struct { Locator string `json:"locator"` BlockMtime int64 `json:"block_mtime"` + + // Target mount, or "" for "everywhere" + MountUUID string `json:"mount_uuid"` } // TrashHandler processes /trash requests. @@ -616,6 +675,11 @@ func GetBlock(ctx context.Context, hash string, buf []byte, resp http.ResponseWr if !os.IsNotExist(err) { log.Printf("%s: Get(%s): %s", vol, hash, err) } + // If some volume returns a transient error, return it to the caller + // instead of "Not found" so it can retry. + if err == VolumeBusyError { + errorToCaller = err.(*KeepError) + } continue } // Check the file checksum. @@ -776,7 +840,7 @@ func IsValidLocator(loc string) bool { return validLocatorRe.MatchString(loc) } -var authRe = regexp.MustCompile(`^OAuth2\s+(.*)`) +var authRe = regexp.MustCompile(`^(OAuth2|Bearer)\s+(.*)`) // GetAPIToken returns the OAuth2 token from the Authorization // header of a HTTP request, or an empty string if no matching @@ -784,7 +848,7 @@ var authRe = regexp.MustCompile(`^OAuth2\s+(.*)`) func GetAPIToken(req *http.Request) string { if auth, ok := req.Header["Authorization"]; ok { if match := authRe.FindStringSubmatch(auth[0]); match != nil { - return match[1] + return match[2] } } return ""