X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/d9f2aaaa6b5762f448276ce96b6994245062a4c8..5b863886118890cc81b728a3a606ea823c836f2b:/services/keepstore/handlers.go diff --git a/services/keepstore/handlers.go b/services/keepstore/handlers.go index a6798a9f72..51dd73a513 100644 --- a/services/keepstore/handlers.go +++ b/services/keepstore/handlers.go @@ -1,20 +1,16 @@ -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" + "context" "crypto/md5" "encoding/json" "fmt" - "github.com/gorilla/mux" "io" - "log" "net/http" "os" "regexp" @@ -23,45 +19,83 @@ import ( "strings" "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" + "github.com/gorilla/mux" + "github.com/prometheus/client_golang/prometheus" ) -// MakeRESTRouter returns a new mux.Router that forwards all Keep -// requests to the appropriate handlers. -// -func MakeRESTRouter() *mux.Router { - rest := mux.NewRouter() +type router struct { + *mux.Router + 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(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) + rtr.HandleFunc(`/debug.json`, rtr.DebugHandler).Methods("GET", "HEAD") // List volumes: path, device number, bytes used/avail. - rest.HandleFunc(`/status.json`, 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 rest + 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. @@ -69,9 +103,17 @@ 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) { - if enforcePermissions { +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 { http.Error(resp, err.Error(), err.(*KeepError).HTTPCode) @@ -86,14 +128,14 @@ func GetBlockHandler(resp http.ResponseWriter, req *http.Request) { // isn't here, we can return 404 now instead of waiting for a // buffer. - buf, err := getBufferForResponseWriter(resp, bufs, BlockSize) + buf, err := getBufferWithContext(ctx, bufs, BlockSize) if err != nil { http.Error(resp, err.Error(), http.StatusServiceUnavailable) return } defer bufs.Put(buf) - size, err := GetBlock(mux.Vars(req)["hash"], buf, resp) + size, err := GetBlock(ctx, mux.Vars(req)["hash"], buf, resp) if err != nil { code := http.StatusInternalServerError if err, ok := err.(*KeepError); ok { @@ -108,24 +150,33 @@ func GetBlockHandler(resp http.ResponseWriter, req *http.Request) { resp.Write(buf[:size]) } +// Return a new context that gets cancelled by resp's CloseNotifier. +func contextForResponse(parent context.Context, resp http.ResponseWriter) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(parent) + if cn, ok := resp.(http.CloseNotifier); ok { + go func(c <-chan bool) { + select { + case <-c: + theConfig.debugLogf("cancel context") + cancel() + case <-ctx.Done(): + } + }(cn.CloseNotify()) + } + return ctx, cancel +} + // Get a buffer from the pool -- but give up and return a non-nil -// error if resp implements http.CloseNotifier and tells us that the -// client has disconnected before we get a buffer. -func getBufferForResponseWriter(resp http.ResponseWriter, bufs *bufferPool, bufSize int) ([]byte, error) { - var closeNotifier <-chan bool - if resp, ok := resp.(http.CloseNotifier); ok { - closeNotifier = resp.CloseNotify() - } - var buf []byte +// error if ctx ends before we get a buffer. +func getBufferWithContext(ctx context.Context, bufs *bufferPool, bufSize int) ([]byte, error) { bufReady := make(chan []byte) go func() { bufReady <- bufs.Get(bufSize) - close(bufReady) }() select { - case buf = <-bufReady: + case buf := <-bufReady: return buf, nil - case <-closeNotifier: + case <-ctx.Done(): go func() { // Even if closeNotifier happened first, we // need to keep waiting for our buf so we can @@ -136,8 +187,10 @@ func getBufferForResponseWriter(resp http.ResponseWriter, bufs *bufferPool, bufS } } -// 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() + hash := mux.Vars(req)["hash"] // Detect as many error conditions as possible before reading @@ -159,7 +212,7 @@ func PutBlockHandler(resp http.ResponseWriter, req *http.Request) { return } - buf, err := getBufferForResponseWriter(resp, bufs, int(req.ContentLength)) + buf, err := getBufferWithContext(ctx, bufs, int(req.ContentLength)) if err != nil { http.Error(resp, err.Error(), http.StatusServiceUnavailable) return @@ -172,12 +225,15 @@ func PutBlockHandler(resp http.ResponseWriter, req *http.Request) { return } - replication, err := PutBlock(buf, hash) + replication, err := PutBlock(ctx, buf, hash) bufs.Put(buf) if err != nil { - ke := err.(*KeepError) - http.Error(resp, ke.Error(), ke.HTTPCode) + code := http.StatusInternalServerError + if err, ok := err.(*KeepError); ok { + code = err.HTTPCode + } + http.Error(resp, err.Error(), code) return } @@ -185,26 +241,42 @@ func PutBlockHandler(resp http.ResponseWriter, req *http.Request) { // return it to the client. returnHash := fmt.Sprintf("%s+%d", hash, req.ContentLength) apiToken := GetAPIToken(req) - if PermissionSecret != nil && apiToken != "" { - expiry := time.Now().Add(blobSignatureTTL) + if theConfig.blobSigningKey != nil && apiToken != "" { + expiry := time.Now().Add(theConfig.BlobSignatureTTL.Duration()) returnHash = SignLocator(returnHash, apiToken, expiry) } resp.Header().Set("X-Keep-Replicas-Stored", strconv.Itoa(replication)) 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. - if !IsDataManagerToken(GetAPIToken(req)) { +// 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 @@ -220,41 +292,59 @@ func IndexHandler(resp http.ResponseWriter, req *http.Request) { resp.Write([]byte{'\n'}) } -// StatusHandler -// Responds to /status.json requests with the current node status, -// described in a JSON structure. -// -// The data given in a status.json response includes: -// volumes - a list of Keep volumes currently in use by this server -// each volume is an object with the following fields: -// * mount_point -// * device_num (an integer identifying the underlying filesystem) -// * bytes_free -// * bytes_used +// 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"` } +type volumeStatusEnt struct { + Label string + Status *VolumeStatus `json:",omitempty"` + VolumeStats *ioStats `json:",omitempty"` + InternalStats interface{} `json:",omitempty"` +} + // NodeStatus struct type NodeStatus struct { - Volumes []*VolumeStatus `json:"volumes"` - BufferPool PoolStatus - PullQueue WorkQueueStatus - TrashQueue WorkQueueStatus - Memory runtime.MemStats + Volumes []*volumeStatusEnt + BufferPool PoolStatus + PullQueue WorkQueueStatus + TrashQueue WorkQueueStatus + RequestsCurrent int + RequestsMax int + Version string } var st NodeStatus var stLock sync.Mutex +// DebugHandler addresses /debug.json requests. +func (rtr *router) DebugHandler(resp http.ResponseWriter, req *http.Request) { + type debugStats struct { + MemStats runtime.MemStats + } + var ds debugStats + runtime.ReadMemStats(&ds.MemStats) + err := json.NewEncoder(resp).Encode(&ds) + if err != nil { + http.Error(resp, err.Error(), 500) + } +} + // StatusHandler addresses /status.json requests. -func StatusHandler(resp http.ResponseWriter, req *http.Request) { +func (rtr *router) StatusHandler(resp http.ResponseWriter, req *http.Request) { stLock.Lock() - readNodeStatus(&st) + rtr.readNodeStatus(&st) jstat, err := json.Marshal(&st) stLock.Unlock() if err == nil { @@ -267,23 +357,34 @@ func StatusHandler(resp http.ResponseWriter, req *http.Request) { } // populate the given NodeStatus struct with current values. -func readNodeStatus(st *NodeStatus) { +func (rtr *router) readNodeStatus(st *NodeStatus) { + st.Version = version vols := KeepVM.AllReadable() if cap(st.Volumes) < len(vols) { - st.Volumes = make([]*VolumeStatus, len(vols)) + st.Volumes = make([]*volumeStatusEnt, len(vols)) } st.Volumes = st.Volumes[:0] for _, vol := range vols { - if s := vol.Status(); s != nil { - st.Volumes = append(st.Volumes, s) + var internalStats interface{} + if vol, ok := vol.(InternalStatser); ok { + internalStats = vol.InternalStats() } + st.Volumes = append(st.Volumes, &volumeStatusEnt{ + Label: vol.String(), + Status: vol.Status(), + InternalStats: internalStats, + //VolumeStats: KeepVM.VolumeStats(vol), + }) } st.BufferPool.Alloc = bufs.Alloc() st.BufferPool.Cap = bufs.Cap() st.BufferPool.Len = bufs.Len() st.PullQueue = getWorkQueueStatus(pullq) st.TrashQueue = getWorkQueueStatus(trashq) - runtime.ReadMemStats(&st.Memory) + if rtr.limiter != nil { + st.RequestsCurrent = rtr.limiter.Current() + st.RequestsMax = rtr.limiter.Max() + } } // return a WorkQueueStatus for the given queue. If q is nil (which @@ -334,7 +435,7 @@ func DeleteHandler(resp http.ResponseWriter, req *http.Request) { return } - if neverDelete { + if !theConfig.EnableDelete { http.Error(resp, MethodDisabledError.Error(), MethodDisabledError.HTTPCode) return } @@ -414,12 +515,15 @@ 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. func PullHandler(resp http.ResponseWriter, req *http.Request) { // Reject unauthorized requests. - if !IsDataManagerToken(GetAPIToken(req)) { + if !IsSystemAuth(GetAPIToken(req)) { http.Error(resp, UnauthorizedError.Error(), UnauthorizedError.HTTPCode) return } @@ -446,16 +550,19 @@ 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. func TrashHandler(resp http.ResponseWriter, req *http.Request) { // Reject unauthorized requests. - if !IsDataManagerToken(GetAPIToken(req)) { + if !IsSystemAuth(GetAPIToken(req)) { http.Error(resp, UnauthorizedError.Error(), UnauthorizedError.HTTPCode) return } @@ -485,7 +592,7 @@ func TrashHandler(resp http.ResponseWriter, req *http.Request) { // UntrashHandler processes "PUT /untrash/{hash:[0-9a-f]{32}}" requests for the data manager. func UntrashHandler(resp http.ResponseWriter, req *http.Request) { // Reject unauthorized requests. - if !IsDataManagerToken(GetAPIToken(req)) { + if !IsSystemAuth(GetAPIToken(req)) { http.Error(resp, UnauthorizedError.Error(), UnauthorizedError.HTTPCode) return } @@ -548,12 +655,17 @@ func UntrashHandler(resp http.ResponseWriter, req *http.Request) { // If the block found does not have the correct MD5 hash, returns // DiskHashError. // -func GetBlock(hash string, buf []byte, resp http.ResponseWriter) (int, error) { +func GetBlock(ctx context.Context, hash string, buf []byte, resp http.ResponseWriter) (int, error) { // Attempt to read the requested hash from a keep volume. errorToCaller := NotFoundError for _, vol := range KeepVM.AllReadable() { - size, err := vol.Get(hash, buf) + size, err := vol.Get(ctx, hash, buf) + select { + case <-ctx.Done(): + return 0, ErrClientDisconnect + default: + } if err != nil { // IsNotExist is an expected error and may be // ignored. All other errors are logged. In @@ -563,6 +675,11 @@ func GetBlock(hash string, buf []byte, resp http.ResponseWriter) (int, error) { 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. @@ -587,7 +704,7 @@ func GetBlock(hash string, buf []byte, resp http.ResponseWriter) (int, error) { // PutBlock Stores the BLOCK (identified by the content id HASH) in Keep. // -// PutBlock(block, hash) +// PutBlock(ctx, block, hash) // Stores the BLOCK (identified by the content id HASH) in Keep. // // The MD5 checksum of the block must be identical to the content id HASH. @@ -612,7 +729,7 @@ func GetBlock(hash string, buf []byte, resp http.ResponseWriter) (int, error) { // all writes failed). The text of the error message should // provide as much detail as possible. // -func PutBlock(block []byte, hash string) (int, error) { +func PutBlock(ctx context.Context, block []byte, hash string) (int, error) { // Check that BLOCK's checksum matches HASH. blockhash := fmt.Sprintf("%x", md5.Sum(block)) if blockhash != hash { @@ -623,16 +740,21 @@ func PutBlock(block []byte, hash string) (int, error) { // If we already have this data, it's intact on disk, and we // can update its timestamp, return success. If we have // different data with the same hash, return failure. - if n, err := CompareAndTouch(hash, block); err == nil || err == CollisionError { + if n, err := CompareAndTouch(ctx, hash, block); err == nil || err == CollisionError { return n, err + } else if ctx.Err() != nil { + return 0, ErrClientDisconnect } // Choose a Keep volume to write to. // If this volume fails, try all of the volumes in order. if vol := KeepVM.NextWritable(); vol != nil { - if err := vol.Put(hash, block); err == nil { + if err := vol.Put(ctx, hash, block); err == nil { return vol.Replication(), nil // success! } + if ctx.Err() != nil { + return 0, ErrClientDisconnect + } } writables := KeepVM.AllWritable() @@ -643,7 +765,10 @@ func PutBlock(block []byte, hash string) (int, error) { allFull := true for _, vol := range writables { - err := vol.Put(hash, block) + err := vol.Put(ctx, hash, block) + if ctx.Err() != nil { + return 0, ErrClientDisconnect + } if err == nil { return vol.Replication(), nil // success! } @@ -669,10 +794,13 @@ func PutBlock(block []byte, hash string) (int, error) { // the relevant block's modification time in order to protect it from // premature garbage collection. Otherwise, it returns a non-nil // error. -func CompareAndTouch(hash string, buf []byte) (int, error) { +func CompareAndTouch(ctx context.Context, hash string, buf []byte) (int, error) { var bestErr error = NotFoundError for _, vol := range KeepVM.AllWritable() { - if err := vol.Compare(hash, buf); err == CollisionError { + err := vol.Compare(ctx, hash, buf) + if ctx.Err() != nil { + return 0, ctx.Err() + } else if err == CollisionError { // Stop if we have a block with same hash but // different content. (It will be impossible // to tell which one is wanted if we have @@ -712,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 @@ -720,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 "" @@ -746,7 +874,7 @@ func CanDelete(apiToken string) bool { } // Blocks may be deleted only when Keep has been configured with a // data manager. - if IsDataManagerToken(apiToken) { + if IsSystemAuth(apiToken) { return true } // TODO(twp): look up apiToken with the API server @@ -755,8 +883,8 @@ func CanDelete(apiToken string) bool { return false } -// IsDataManagerToken returns true if apiToken represents the data -// manager's token. -func IsDataManagerToken(apiToken string) bool { - return dataManagerToken != "" && apiToken == dataManagerToken +// IsSystemAuth returns true if the given token is allowed to perform +// system level actions like deleting data. +func IsSystemAuth(token string) bool { + return token != "" && token == theConfig.systemAuthToken }