X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/d47c009c2afd20b86cca083787d238c136e70abd..a4211acb465bd42869bf2a2f9fad6ff2c5e518e0:/services/keepproxy/keepproxy.go diff --git a/services/keepproxy/keepproxy.go b/services/keepproxy/keepproxy.go index 367a9c5604..ebdb6eb376 100644 --- a/services/keepproxy/keepproxy.go +++ b/services/keepproxy/keepproxy.go @@ -1,17 +1,20 @@ package main import ( - keepclient "git.curoverse.com/arvados.git/sdk/go/keepclient" - arvadosclient "git.curoverse.com/arvados.git/sdk/go/arvadosclient" + "errors" "flag" "fmt" + "git.curoverse.com/arvados.git/sdk/go/arvadosclient" + "git.curoverse.com/arvados.git/sdk/go/keepclient" "github.com/gorilla/mux" "io" + "io/ioutil" "log" "net" "net/http" "os" "os/signal" + "reflect" "sync" "syscall" "time" @@ -29,6 +32,7 @@ func main() { no_get bool no_put bool default_replicas int + timeout int64 pidfile string ) @@ -60,6 +64,12 @@ func main() { 2, "Default number of replicas to write if not specified by the client.") + flagset.Int64Var( + &timeout, + "timeout", + 15, + "Timeout on requests to internal Keep services (default 15 seconds)") + flagset.StringVar( &pidfile, "pid", @@ -80,22 +90,24 @@ func main() { if pidfile != "" { f, err := os.Create(pidfile) - if err == nil { - fmt.Fprint(f, os.Getpid()) - f.Close() - } else { - log.Printf("Error writing pid file (%s): %s", pidfile, err.Error()) + if err != nil { + log.Fatalf("Error writing pid file (%s): %s", pidfile, err.Error()) } + fmt.Fprint(f, os.Getpid()) + f.Close() + defer os.Remove(pidfile) } kc.Want_replicas = default_replicas + kc.Client.Timeout = time.Duration(timeout) * time.Second + listener, err = net.Listen("tcp", listen) if err != nil { log.Fatalf("Could not listen on %v", listen) } - go RefreshServicesList(&kc) + go RefreshServicesList(kc) // Shut down the server gracefully (by closing the listener) // if SIGTERM is received. @@ -108,26 +120,12 @@ func main() { signal.Notify(term, syscall.SIGTERM) signal.Notify(term, syscall.SIGINT) - if pidfile != "" { - f, err := os.Create(pidfile) - if err == nil { - fmt.Fprint(f, os.Getpid()) - f.Close() - } else { - log.Printf("Error writing pid file (%s): %s", pidfile, err.Error()) - } - } - - log.Printf("Arvados Keep proxy started listening on %v with server list %v", listener.Addr(), kc.ServiceRoots()) + log.Printf("Arvados Keep proxy started listening on %v", listener.Addr()) // Start listening for requests. - http.Serve(listener, MakeRESTRouter(!no_get, !no_put, &kc)) + http.Serve(listener, MakeRESTRouter(!no_get, !no_put, kc)) log.Println("shutting down") - - if pidfile != "" { - os.Remove(pidfile) - } } type ApiTokenCache struct { @@ -138,16 +136,25 @@ type ApiTokenCache struct { // Refresh the keep service list every five minutes. func RefreshServicesList(kc *keepclient.KeepClient) { + var previousRoots = []map[string]string{} + var delay time.Duration = 0 for { - time.Sleep(300 * time.Second) - oldservices := kc.ServiceRoots() - kc.DiscoverKeepServers() - newservices := kc.ServiceRoots() - s1 := fmt.Sprint(oldservices) - s2 := fmt.Sprint(newservices) - if s1 != s2 { - log.Printf("Updated server list to %v", s2) + time.Sleep(delay * time.Second) + delay = 300 + if err := kc.DiscoverKeepServers(); err != nil { + log.Println("Error retrieving services list:", err) + delay = 3 + continue } + newRoots := []map[string]string{kc.LocalRoots(), kc.GatewayRoots()} + if !reflect.DeepEqual(previousRoots, newRoots) { + log.Printf("Updated services list: locals %v gateways %v", newRoots[0], newRoots[1]) + } + if len(newRoots[0]) == 0 { + log.Print("WARNING: No local services. Retrying in 3 seconds.") + delay = 3 + } + previousRoots = newRoots } } @@ -235,6 +242,8 @@ type PutBlockHandler struct { type InvalidPathHandler struct{} +type OptionsHandler struct{} + // MakeRESTRouter // Returns a mux.Router that passes GET and PUT requests to the // appropriate handlers. @@ -249,14 +258,17 @@ func MakeRESTRouter( rest := mux.NewRouter() if enable_get { - rest.Handle(`/{hash:[0-9a-f]{32}}+{hints}`, + rest.Handle(`/{locator:[0-9a-f]{32}\+.*}`, GetBlockHandler{kc, t}).Methods("GET", "HEAD") - rest.Handle(`/{hash:[0-9a-f]{32}}`, GetBlockHandler{kc, t}).Methods("GET", "HEAD") + rest.Handle(`/{locator:[0-9a-f]{32}}`, GetBlockHandler{kc, t}).Methods("GET", "HEAD") } if enable_put { - rest.Handle(`/{hash:[0-9a-f]{32}}+{hints}`, PutBlockHandler{kc, t}).Methods("PUT") - rest.Handle(`/{hash:[0-9a-f]{32}}`, PutBlockHandler{kc, t}).Methods("PUT") + rest.Handle(`/{locator:[0-9a-f]{32}\+.*}`, PutBlockHandler{kc, t}).Methods("PUT") + rest.Handle(`/{locator:[0-9a-f]{32}}`, PutBlockHandler{kc, t}).Methods("PUT") + rest.Handle(`/`, PutBlockHandler{kc, t}).Methods("POST") + rest.Handle(`/{any}`, OptionsHandler{}).Methods("OPTIONS") + rest.Handle(`/`, OptionsHandler{}).Methods("OPTIONS") } rest.NotFoundHandler = InvalidPathHandler{} @@ -264,26 +276,49 @@ func MakeRESTRouter( return rest } +func SetCorsHeaders(resp http.ResponseWriter) { + resp.Header().Set("Access-Control-Allow-Methods", "GET, HEAD, POST, PUT, OPTIONS") + resp.Header().Set("Access-Control-Allow-Origin", "*") + resp.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Length, Content-Type, X-Keep-Desired-Replicas") + resp.Header().Set("Access-Control-Max-Age", "86486400") +} + func (this InvalidPathHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) { log.Printf("%s: %s %s unroutable", GetRemoteAddress(req), req.Method, req.URL.Path) http.Error(resp, "Bad request", http.StatusBadRequest) } -func (this GetBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) { +func (this OptionsHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) { + log.Printf("%s: %s %s", GetRemoteAddress(req), req.Method, req.URL.Path) + SetCorsHeaders(resp) +} - kc := *this.KeepClient +var BadAuthorizationHeader = errors.New("Missing or invalid Authorization header") +var ContentLengthMismatch = errors.New("Actual length != expected content length") +var MethodNotSupported = errors.New("Method not supported") - hash := mux.Vars(req)["hash"] - hints := mux.Vars(req)["hints"] +func (this GetBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) { + SetCorsHeaders(resp) - locator := keepclient.MakeLocator2(hash, hints) + locator := mux.Vars(req)["locator"] + var err error + var status int + var expectLength, responseLength int64 + var proxiedURI = "-" + + defer func() { + log.Println(GetRemoteAddress(req), req.Method, req.URL.Path, status, expectLength, responseLength, proxiedURI, err) + if status != http.StatusOK { + http.Error(resp, err.Error(), status) + } + }() - log.Printf("%s: %s %s", GetRemoteAddress(req), req.Method, hash) + kc := *this.KeepClient var pass bool var tok string if pass, tok = CheckAuthorizationHeader(kc, this.ApiTokenCache, req); !pass { - http.Error(resp, "Missing or invalid Authorization header", http.StatusForbidden) + status, err = http.StatusForbidden, BadAuthorizationHeader return } @@ -293,79 +328,97 @@ func (this GetBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques kc.Arvados = &arvclient var reader io.ReadCloser - var err error - var blocklen int64 - if req.Method == "GET" { - reader, blocklen, _, err = kc.AuthorizedGet(hash, locator.Signature, locator.Timestamp) - defer reader.Close() - } else if req.Method == "HEAD" { - blocklen, _, err = kc.AuthorizedAsk(hash, locator.Signature, locator.Timestamp) + switch req.Method { + case "HEAD": + expectLength, proxiedURI, err = kc.Ask(locator) + case "GET": + reader, expectLength, proxiedURI, err = kc.Get(locator) + if reader != nil { + defer reader.Close() + } + default: + status, err = http.StatusNotImplemented, MethodNotSupported + return } - if blocklen > 0 { - resp.Header().Set("Content-Length", fmt.Sprint(blocklen)) + if expectLength == -1 { + log.Println("Warning:", GetRemoteAddress(req), req.Method, proxiedURI, "Content-Length not provided") } switch err { case nil: - if reader != nil { - n, err2 := io.Copy(resp, reader) - if n != blocklen { - log.Printf("%s: %s %s mismatched return %v with Content-Length %v error %v", GetRemoteAddress(req), req.Method, hash, n, blocklen, err2) - } else if err2 == nil { - log.Printf("%s: %s %s success returned %v bytes", GetRemoteAddress(req), req.Method, hash, n) - } else { - log.Printf("%s: %s %s returned %v bytes error %v", GetRemoteAddress(req), req.Method, hash, n, err.Error()) + status = http.StatusOK + resp.Header().Set("Content-Length", fmt.Sprint(expectLength)) + switch req.Method { + case "HEAD": + responseLength = 0 + case "GET": + responseLength, err = io.Copy(resp, reader) + if err == nil && expectLength > -1 && responseLength != expectLength { + err = ContentLengthMismatch } - } else { - log.Printf("%s: %s %s success", GetRemoteAddress(req), req.Method, hash) } case keepclient.BlockNotFound: - http.Error(resp, "Not found", http.StatusNotFound) + status = http.StatusNotFound default: - http.Error(resp, err.Error(), http.StatusBadGateway) - } - - if err != nil { - log.Printf("%s: %s %s error %s", GetRemoteAddress(req), req.Method, hash, err.Error()) + status = http.StatusBadGateway } } +var LengthRequiredError = errors.New(http.StatusText(http.StatusLengthRequired)) +var LengthMismatchError = errors.New("Locator size hint does not match Content-Length header") + func (this PutBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) { + SetCorsHeaders(resp) kc := *this.KeepClient + var err error + var expectLength int64 = -1 + var status = http.StatusInternalServerError + var wroteReplicas int + var locatorOut string = "-" + + defer func() { + log.Println(GetRemoteAddress(req), req.Method, req.URL.Path, status, expectLength, kc.Want_replicas, wroteReplicas, locatorOut, err) + if status != http.StatusOK { + http.Error(resp, err.Error(), status) + } + }() - hash := mux.Vars(req)["hash"] - hints := mux.Vars(req)["hints"] - - locator := keepclient.MakeLocator2(hash, hints) + locatorIn := mux.Vars(req)["locator"] - var contentLength int64 = -1 if req.Header.Get("Content-Length") != "" { - _, err := fmt.Sscanf(req.Header.Get("Content-Length"), "%d", &contentLength) + _, err := fmt.Sscanf(req.Header.Get("Content-Length"), "%d", &expectLength) if err != nil { - resp.Header().Set("Content-Length", fmt.Sprintf("%d", contentLength)) + resp.Header().Set("Content-Length", fmt.Sprintf("%d", expectLength)) } } - log.Printf("%s: %s %s Content-Length %v", GetRemoteAddress(req), req.Method, hash, contentLength) - - if contentLength < 1 { - http.Error(resp, "Must include Content-Length header", http.StatusLengthRequired) + if expectLength < 0 { + err = LengthRequiredError + status = http.StatusLengthRequired return } - if locator.Size > 0 && int64(locator.Size) != contentLength { - http.Error(resp, "Locator size hint does not match Content-Length header", http.StatusBadRequest) - return + if locatorIn != "" { + var loc *keepclient.Locator + if loc, err = keepclient.MakeLocator(locatorIn); err != nil { + status = http.StatusBadRequest + return + } else if loc.Size > 0 && int64(loc.Size) != expectLength { + err = LengthMismatchError + status = http.StatusBadRequest + return + } } var pass bool var tok string if pass, tok = CheckAuthorizationHeader(kc, this.ApiTokenCache, req); !pass { - http.Error(resp, "Missing or invalid Authorization header", http.StatusForbidden) + err = BadAuthorizationHeader + status = http.StatusForbidden return } @@ -384,44 +437,42 @@ func (this PutBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques } // Now try to put the block through - hash, replicas, err := kc.PutHR(hash, req.Body, contentLength) + if locatorIn == "" { + if bytes, err := ioutil.ReadAll(req.Body); err != nil { + err = errors.New(fmt.Sprintf("Error reading request body: %s", err)) + status = http.StatusInternalServerError + return + } else { + locatorOut, wroteReplicas, err = kc.PutB(bytes) + } + } else { + locatorOut, wroteReplicas, err = kc.PutHR(locatorIn, req.Body, expectLength) + } // Tell the client how many successful PUTs we accomplished - resp.Header().Set(keepclient.X_Keep_Replicas_Stored, fmt.Sprintf("%d", replicas)) + resp.Header().Set(keepclient.X_Keep_Replicas_Stored, fmt.Sprintf("%d", wroteReplicas)) switch err { case nil: - // Default will return http.StatusOK - log.Printf("%s: %s %s finished, stored %v replicas (desired %v)", GetRemoteAddress(req), req.Method, hash, replicas, kc.Want_replicas) - n, err2 := io.WriteString(resp, hash) - if err2 != nil { - log.Printf("%s: wrote %v bytes to response body and got error %v", n, err2.Error()) - } + status = http.StatusOK + _, err = io.WriteString(resp, locatorOut) case keepclient.OversizeBlockError: // Too much data - http.Error(resp, fmt.Sprintf("Exceeded maximum blocksize %d", keepclient.BLOCKSIZE), http.StatusRequestEntityTooLarge) + status = http.StatusRequestEntityTooLarge case keepclient.InsufficientReplicasError: - if replicas > 0 { + if wroteReplicas > 0 { // At least one write is considered success. The // client can decide if getting less than the number of // replications it asked for is a fatal error. - // Default will return http.StatusOK - n, err2 := io.WriteString(resp, hash) - if err2 != nil { - log.Printf("%s: wrote %v bytes to response body and got error %v", n, err2.Error()) - } + status = http.StatusOK + _, err = io.WriteString(resp, locatorOut) } else { - http.Error(resp, "", http.StatusServiceUnavailable) + status = http.StatusServiceUnavailable } default: - http.Error(resp, err.Error(), http.StatusBadGateway) - } - - if err != nil { - log.Printf("%s: %s %s stored %v replicas (desired %v) got error %v", GetRemoteAddress(req), req.Method, hash, replicas, kc.Want_replicas, err.Error()) + status = http.StatusBadGateway } - }