projects
/
arvados.git
/ blobdiff
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Merge branch 'master' into 13937-keepstore-prometheus
[arvados.git]
/
services
/
keepstore
/
handlers.go
diff --git
a/services/keepstore/handlers.go
b/services/keepstore/handlers.go
index c31ab9c2e38fde497f451e95c2a99437735e4455..51dd73a513c1d4c729a6743aaabe0cefa1202c4b 100644
(file)
--- a/
services/keepstore/handlers.go
+++ b/
services/keepstore/handlers.go
@@
-4,13
+4,6
@@
package main
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"
import (
"container/list"
"context"
@@
-27,29
+20,37
@@
import (
"sync"
"time"
"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"
"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
)
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.
}
// 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(
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}`,
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")
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")
@@
-87,10
+88,14
@@
func MakeRESTRouter() http.Handler {
rtr.NotFoundHandler = http.HandlerFunc(BadRequestHandler)
rtr.limiter = httpserver.NewRequestLimiter(theConfig.MaxRequests, rtr)
rtr.NotFoundHandler = http.HandlerFunc(BadRequestHandler)
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)
-
stack := httpserver.Instrument(nil
, nil,
+
instrumented := httpserver.Instrument(rtr.metrics.reg
, nil,
httpserver.AddRequestIDs(httpserver.LogRequests(nil, rtr.limiter)))
httpserver.AddRequestIDs(httpserver.LogRequests(nil, rtr.limiter)))
- return
stack.ServeAPI(stack
)
+ return
instrumented.ServeAPI(theConfig.ManagementToken, instrumented
)
}
// BadRequestHandler is a HandleFunc to address bad requests.
}
// BadRequestHandler is a HandleFunc to address bad requests.
@@
-98,11
+103,16
@@
func BadRequestHandler(w http.ResponseWriter, r *http.Request) {
http.Error(w, BadRequestError.Error(), BadRequestError.HTTPCode)
}
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()
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 {
if theConfig.RequireSignatures {
locator := req.URL.Path[1:] // strip leading slash
if err := VerifySignature(locator, GetAPIToken(req)); err != nil {
@@
-177,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()
ctx, cancel := contextForResponse(context.TODO(), resp)
defer cancel()
@@
-666,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 !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.
continue
}
// Check the file checksum.
@@
-826,7
+840,7
@@
func IsValidLocator(loc string) bool {
return validLocatorRe.MatchString(loc)
}
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
// GetAPIToken returns the OAuth2 token from the Authorization
// header of a HTTP request, or an empty string if no matching
@@
-834,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 {
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 ""
}
}
return ""