projects
/
arvados.git
/ blobdiff
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
13994: Proxy to remote cluster if +R hint given.
[arvados.git]
/
services
/
keepstore
/
handlers.go
diff --git
a/services/keepstore/handlers.go
b/services/keepstore/handlers.go
index 2d90aba14e4d36ddf41f035282b66100967d783b..2426c9cbdacd4044a4c2fab06f4ee51edef2b4e9 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"
@@
-29,66
+22,74
@@
import (
"github.com/gorilla/mux"
"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"
- log "github.com/Sirupsen/logrus"
)
type router struct {
*mux.Router
)
type router struct {
*mux.Router
- limiter httpserver.RequestCounter
+ limiter httpserver.RequestCounter
+ cluster *arvados.Cluster
+ remoteProxy remoteProxy
}
// 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() *router {
- rest := mux.NewRouter()
- rtr := &router{Router: rest}
+func MakeRESTRouter(cluster *arvados.Cluster) http.Handler {
+ rtr := &router{
+ Router: mux.NewRouter(),
+ cluster: cluster,
+ }
- r
est
.HandleFunc(
- `/{hash:[0-9a-f]{32}}`,
GetBlockHandler
).Methods("GET", "HEAD")
- r
est
.HandleFunc(
+ r
tr
.HandleFunc(
+ `/{hash:[0-9a-f]{32}}`,
rtr.handleGET
).Methods("GET", "HEAD")
+ r
tr
.HandleFunc(
`/{hash:[0-9a-f]{32}}+{hints}`,
`/{hash:[0-9a-f]{32}}+{hints}`,
-
GetBlockHandler
).Methods("GET", "HEAD")
+
rtr.handleGET
).Methods("GET", "HEAD")
- r
est.HandleFunc(`/{hash:[0-9a-f]{32}}`, PutBlockHandler
).Methods("PUT")
- r
est
.HandleFunc(`/{hash:[0-9a-f]{32}}`, DeleteHandler).Methods("DELETE")
+ r
tr.HandleFunc(`/{hash:[0-9a-f]{32}}`, rtr.handlePUT
).Methods("PUT")
+ r
tr
.HandleFunc(`/{hash:[0-9a-f]{32}}`, DeleteHandler).Methods("DELETE")
// List all blocks stored here. Privileged client only.
// List all blocks stored here. Privileged client only.
- r
est
.HandleFunc(`/index`, rtr.IndexHandler).Methods("GET", "HEAD")
+ r
tr
.HandleFunc(`/index`, rtr.IndexHandler).Methods("GET", "HEAD")
// List blocks stored here whose hash has the given prefix.
// Privileged client only.
// List blocks stored here whose hash has the given prefix.
// Privileged client only.
- r
est
.HandleFunc(`/index/{prefix:[0-9a-f]{0,32}}`, rtr.IndexHandler).Methods("GET", "HEAD")
+ r
tr
.HandleFunc(`/index/{prefix:[0-9a-f]{0,32}}`, rtr.IndexHandler).Methods("GET", "HEAD")
// Internals/debugging info (runtime.MemStats)
// Internals/debugging info (runtime.MemStats)
- r
est
.HandleFunc(`/debug.json`, rtr.DebugHandler).Methods("GET", "HEAD")
+ r
tr
.HandleFunc(`/debug.json`, rtr.DebugHandler).Methods("GET", "HEAD")
// List volumes: path, device number, bytes used/avail.
// List volumes: path, device number, bytes used/avail.
- r
est
.HandleFunc(`/status.json`, rtr.StatusHandler).Methods("GET", "HEAD")
+ r
tr
.HandleFunc(`/status.json`, rtr.StatusHandler).Methods("GET", "HEAD")
// List mounts: UUID, readonly, tier, device ID, ...
// List mounts: UUID, readonly, tier, device ID, ...
- r
est
.HandleFunc(`/mounts`, rtr.MountsHandler).Methods("GET")
- r
est
.HandleFunc(`/mounts/{uuid}/blocks`, rtr.IndexHandler).Methods("GET")
- r
est
.HandleFunc(`/mounts/{uuid}/blocks/`, rtr.IndexHandler).Methods("GET")
+ r
tr
.HandleFunc(`/mounts`, rtr.MountsHandler).Methods("GET")
+ r
tr
.HandleFunc(`/mounts/{uuid}/blocks`, rtr.IndexHandler).Methods("GET")
+ r
tr
.HandleFunc(`/mounts/{uuid}/blocks/`, rtr.IndexHandler).Methods("GET")
// Replace the current pull queue.
// Replace the current pull queue.
- r
est
.HandleFunc(`/pull`, PullHandler).Methods("PUT")
+ r
tr
.HandleFunc(`/pull`, PullHandler).Methods("PUT")
// Replace the current trash queue.
// Replace the current trash queue.
- r
est
.HandleFunc(`/trash`, TrashHandler).Methods("PUT")
+ r
tr
.HandleFunc(`/trash`, TrashHandler).Methods("PUT")
// Untrash moves blocks from trash back into store
// Untrash moves blocks from trash back into store
- r
est
.HandleFunc(`/untrash/{hash:[0-9a-f]{32}}`, UntrashHandler).Methods("PUT")
+ r
tr
.HandleFunc(`/untrash/{hash:[0-9a-f]{32}}`, UntrashHandler).Methods("PUT")
- r
est
.Handle("/_health/{check}", &health.Handler{
+ r
tr
.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.
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)
+
+ rtr.limiter = httpserver.NewRequestLimiter(theConfig.MaxRequests, rtr)
- return rtr
+ stack := httpserver.Instrument(nil, nil,
+ httpserver.AddRequestIDs(httpserver.LogRequests(nil, rtr.limiter)))
+ return stack.ServeAPI(stack)
}
// BadRequestHandler is a HandleFunc to address bad requests.
}
// BadRequestHandler is a HandleFunc to address bad requests.
@@
-96,11
+97,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(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 {
@@
-175,8
+181,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()
@@
-311,6
+316,7
@@
type NodeStatus struct {
TrashQueue WorkQueueStatus
RequestsCurrent int
RequestsMax int
TrashQueue WorkQueueStatus
RequestsCurrent int
RequestsMax int
+ Version string
}
var st NodeStatus
}
var st NodeStatus
@@
-346,6
+352,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) {
// 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))
vols := KeepVM.AllReadable()
if cap(st.Volumes) < len(vols) {
st.Volumes = make([]*volumeStatusEnt, len(vols))
@@
-504,7
+511,7
@@
type PullRequest struct {
Servers []string `json:"servers"`
// Destination mount, or "" for "anywhere"
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.
}
// PullHandler processes "PUT /pull" requests for the data manager.
@@
-537,13
+544,13
@@
func PullHandler(resp http.ResponseWriter, req *http.Request) {
pullq.ReplaceQueue(plist)
}
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"
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.
}
// TrashHandler processes /trash requests.
@@
-822,7
+829,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
@@
-830,7
+837,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 ""