X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/df591042778408d03d410d5c22a669d85652d1ea..e3f324c979fe45803e2bd42aa8f6d715ea23edb5:/services/keepstore/handlers.go diff --git a/services/keepstore/handlers.go b/services/keepstore/handlers.go index 258604ce59..9a4d02df85 100644 --- a/services/keepstore/handlers.go +++ b/services/keepstore/handlers.go @@ -4,13 +4,6 @@ package main -// REST handlers for Keep are implemented here. -// -// GetBlockHandler (GET /locator) -// PutBlockHandler (PUT /locator) -// IndexHandler (GET /index, GET /index/prefix) -// StatusHandler (GET /status.json) - import ( "container/list" "context" @@ -27,29 +20,37 @@ import ( "sync" "time" - "github.com/gorilla/mux" - + "git.curoverse.com/arvados.git/sdk/go/arvados" "git.curoverse.com/arvados.git/sdk/go/health" "git.curoverse.com/arvados.git/sdk/go/httpserver" + "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() http.Handler { - rtr := &router{Router: mux.NewRouter()} +func MakeRESTRouter(cluster *arvados.Cluster, reg *prometheus.Registry) http.Handler { + rtr := &router{ + Router: mux.NewRouter(), + cluster: cluster, + metrics: &nodeMetrics{reg: reg}, + } rtr.HandleFunc( - `/{hash:[0-9a-f]{32}}`, GetBlockHandler).Methods("GET", "HEAD") + `/{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") - rtr.HandleFunc(`/{hash:[0-9a-f]{32}}`, PutBlockHandler).Methods("PUT") + 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. rtr.HandleFunc(`/index`, rtr.IndexHandler).Methods("GET", "HEAD") @@ -86,17 +87,15 @@ func MakeRESTRouter() http.Handler { // 400 Bad Request. rtr.NotFoundHandler = http.HandlerFunc(BadRequestHandler) - theConfig.metrics.setup() - rtr.limiter = httpserver.NewRequestLimiter(theConfig.MaxRequests, rtr) - - mux := http.NewServeMux() - mux.Handle("/", theConfig.metrics.Instrument( - httpserver.AddRequestIDs(httpserver.LogRequests(rtr.limiter)))) - mux.HandleFunc("/metrics.json", theConfig.metrics.exportJSON) - mux.Handle("/metrics", theConfig.metrics.exportProm) - - return mux + 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. @@ -104,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 { @@ -183,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() @@ -274,13 +277,19 @@ func (rtr *router) IndexHandler(resp http.ResponseWriter, req *http.Request) { 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 - // disconnected and this error will never be - // reported to the client -- but it will - // appear in our own error log. - http.Error(resp, err.Error(), http.StatusInternalServerError) + // We can't send an error message to the + // client because we might have already sent + // headers and index content. All we can do is + // log the error in our own logs, and (in + // cases where headers haven't been sent yet) + // set a 500 status. + // + // If headers have already been sent, the + // client must notice the lack of trailing + // newline as an indication that the response + // is incomplete. + log.Printf("index error from volume %s: %s", v, err) + http.Error(resp, "", http.StatusInternalServerError) return } } @@ -514,7 +523,7 @@ type PullRequest struct { Servers []string `json:"servers"` // Destination mount, or "" for "anywhere" - MountUUID string + MountUUID string `json:"mount_uuid"` } // PullHandler processes "PUT /pull" requests for the data manager. @@ -547,13 +556,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 + MountUUID string `json:"mount_uuid"` } // TrashHandler processes /trash requests. @@ -672,6 +681,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. @@ -832,7 +846,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 @@ -840,7 +854,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 ""