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"
"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")
// 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.
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 {
}
}
-// 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()
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
}
}
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.
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
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 ""
projects / arvados.git / blobdiff