X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/a04ea95e79c60ed2a54eaec5b5c2e235fe39ef9a..a5445a2ae553b0723d8579462bcb48855f71a17c:/services/keepproxy/keepproxy.go diff --git a/services/keepproxy/keepproxy.go b/services/keepproxy/keepproxy.go index b8c30d113e..816de29da8 100644 --- a/services/keepproxy/keepproxy.go +++ b/services/keepproxy/keepproxy.go @@ -1,11 +1,10 @@ package main import ( + "encoding/json" + "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" @@ -13,99 +12,129 @@ import ( "net/http" "os" "os/signal" + "regexp" "sync" "syscall" "time" + + "git.curoverse.com/arvados.git/sdk/go/arvados" + "git.curoverse.com/arvados.git/sdk/go/arvadosclient" + "git.curoverse.com/arvados.git/sdk/go/config" + "git.curoverse.com/arvados.git/sdk/go/keepclient" + "github.com/coreos/go-systemd/daemon" + "github.com/gorilla/mux" ) -// Default TCP address on which to listen for requests. -// Initialized by the -listen flag. -const DEFAULT_ADDR = ":25107" +type Config struct { + Client arvados.Client + Listen string + DisableGet bool + DisablePut bool + DefaultReplicas int + Timeout arvados.Duration + PIDFile string + Debug bool +} + +func DefaultConfig() *Config { + return &Config{ + Listen: ":25107", + Timeout: arvados.Duration(15 * time.Second), + } +} var listener net.Listener func main() { - var ( - listen string - no_get bool - no_put bool - default_replicas int - timeout int64 - pidfile string - ) - - flagset := flag.NewFlagSet("default", flag.ExitOnError) - - flagset.StringVar( - &listen, - "listen", - DEFAULT_ADDR, - "Interface on which to listen for requests, in the format "+ - "ipaddr:port. e.g. -listen=10.0.1.24:8000. Use -listen=:port "+ - "to listen on all network interfaces.") - - flagset.BoolVar( - &no_get, - "no-get", - false, - "If set, disable GET operations") - - flagset.BoolVar( - &no_put, - "no-put", - false, - "If set, disable PUT operations") - - flagset.IntVar( - &default_replicas, - "default-replicas", - 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", - "", - "Path to write pid file") - + cfg := DefaultConfig() + + flagset := flag.NewFlagSet("keepproxy", flag.ExitOnError) + flagset.Usage = usage + + const deprecated = " (DEPRECATED -- use config file instead)" + flagset.StringVar(&cfg.Listen, "listen", cfg.Listen, "Local port to listen on."+deprecated) + flagset.BoolVar(&cfg.DisableGet, "no-get", cfg.DisableGet, "Disable GET operations."+deprecated) + flagset.BoolVar(&cfg.DisablePut, "no-put", cfg.DisablePut, "Disable PUT operations."+deprecated) + flagset.IntVar(&cfg.DefaultReplicas, "default-replicas", cfg.DefaultReplicas, "Default number of replicas to write if not specified by the client. If 0, use site default."+deprecated) + flagset.StringVar(&cfg.PIDFile, "pid", cfg.PIDFile, "Path to write pid file."+deprecated) + timeoutSeconds := flagset.Int("timeout", int(time.Duration(cfg.Timeout)/time.Second), "Timeout (in seconds) on requests to internal Keep services."+deprecated) + + var cfgPath string + const defaultCfgPath = "/etc/arvados/keepproxy/keepproxy.yml" + flagset.StringVar(&cfgPath, "config", defaultCfgPath, "Configuration file `path`") flagset.Parse(os.Args[1:]) - arv, err := arvadosclient.MakeArvadosClient() + err := config.LoadFile(cfg, cfgPath) + if err != nil { + h := os.Getenv("ARVADOS_API_HOST") + t := os.Getenv("ARVADOS_API_TOKEN") + if h == "" || t == "" || !os.IsNotExist(err) || cfgPath != defaultCfgPath { + log.Fatal(err) + } + log.Print("DEPRECATED: No config file found, but ARVADOS_API_HOST and ARVADOS_API_TOKEN environment variables are set. Please use a config file instead.") + cfg.Client.APIHost = h + cfg.Client.AuthToken = t + if regexp.MustCompile("^(?i:1|yes|true)$").MatchString(os.Getenv("ARVADOS_API_HOST_INSECURE")) { + cfg.Client.Insecure = true + } + if j, err := json.MarshalIndent(cfg, "", " "); err == nil { + log.Print("Current configuration:\n", string(j)) + } + cfg.Timeout = arvados.Duration(time.Duration(*timeoutSeconds) * time.Second) + } + + arv, err := arvadosclient.New(&cfg.Client) if err != nil { log.Fatalf("Error setting up arvados client %s", err.Error()) } - kc, err := keepclient.MakeKeepClient(&arv) + if cfg.Debug { + keepclient.DebugPrintf = log.Printf + } + kc, err := keepclient.MakeKeepClient(arv) if err != nil { log.Fatalf("Error setting up keep client %s", err.Error()) } - if pidfile != "" { - f, err := os.Create(pidfile) + if cfg.PIDFile != "" { + f, err := os.Create(cfg.PIDFile) + if err != nil { + log.Fatal(err) + } + defer f.Close() + err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) + if err != nil { + log.Fatalf("flock(%s): %s", cfg.PIDFile, err) + } + defer os.Remove(cfg.PIDFile) + err = f.Truncate(0) + if err != nil { + log.Fatalf("truncate(%s): %s", cfg.PIDFile, err) + } + _, err = fmt.Fprint(f, os.Getpid()) + if err != nil { + log.Fatalf("write(%s): %s", cfg.PIDFile, err) + } + err = f.Sync() if err != nil { - log.Fatalf("Error writing pid file (%s): %s", pidfile, err.Error()) + log.Fatal("sync(%s): %s", cfg.PIDFile, err) } - fmt.Fprint(f, os.Getpid()) - f.Close() - defer os.Remove(pidfile) } - kc.Want_replicas = default_replicas - - kc.Client.Timeout = time.Duration(timeout) * time.Second + if cfg.DefaultReplicas > 0 { + kc.Want_replicas = cfg.DefaultReplicas + } + kc.Client.Timeout = time.Duration(cfg.Timeout) + go kc.RefreshServices(5*time.Minute, 3*time.Second) - listener, err = net.Listen("tcp", listen) + listener, err = net.Listen("tcp", cfg.Listen) if err != nil { - log.Fatalf("Could not listen on %v", listen) + log.Fatalf("listen(%s): %s", cfg.Listen, err) } - - go RefreshServicesList(&kc) + if _, err := daemon.SdNotify("READY=1"); err != nil { + log.Printf("Error notifying init daemon: %v", err) + } + log.Println("Listening at", listener.Addr()) // Shut down the server gracefully (by closing the listener) // if SIGTERM is received. @@ -118,10 +147,8 @@ func main() { signal.Notify(term, syscall.SIGTERM) signal.Notify(term, syscall.SIGINT) - log.Printf("Arvados Keep proxy started listening on %v with server list %v", listener.Addr(), kc.ServiceRoots()) - - // Start listening for requests. - http.Serve(listener, MakeRESTRouter(!no_get, !no_put, &kc)) + // Start serving requests. + http.Serve(listener, MakeRESTRouter(!cfg.DisableGet, !cfg.DisablePut, kc)) log.Println("shutting down") } @@ -132,21 +159,6 @@ type ApiTokenCache struct { expireTime int64 } -// Refresh the keep service list every five minutes. -func RefreshServicesList(kc *keepclient.KeepClient) { - 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) - } - } -} - // Cache the token and set an expire time. If we already have an expire time // on the token, it is not updated. func (this *ApiTokenCache) RememberToken(token string) { @@ -179,17 +191,13 @@ func (this *ApiTokenCache) RecallToken(token string) bool { } func GetRemoteAddress(req *http.Request) string { - if realip := req.Header.Get("X-Real-IP"); realip != "" { - if forwarded := req.Header.Get("X-Forwarded-For"); forwarded != realip { - return fmt.Sprintf("%s (X-Forwarded-For %s)", realip, forwarded) - } else { - return realip - } + if xff := req.Header.Get("X-Forwarded-For"); xff != "" { + return xff + "," + req.RemoteAddr } return req.RemoteAddr } -func CheckAuthorizationHeader(kc keepclient.KeepClient, cache *ApiTokenCache, req *http.Request) (pass bool, tok string) { +func CheckAuthorizationHeader(kc *keepclient.KeepClient, cache *ApiTokenCache, req *http.Request) (pass bool, tok string) { var auth string if auth = req.Header.Get("Authorization"); auth == "" { return false, "" @@ -202,7 +210,7 @@ func CheckAuthorizationHeader(kc keepclient.KeepClient, cache *ApiTokenCache, re } if cache.RecallToken(tok) { - // Valid in the cache, short circut + // Valid in the cache, short circuit return true, tok } @@ -229,6 +237,11 @@ type PutBlockHandler struct { *ApiTokenCache } +type IndexHandler struct { + *keepclient.KeepClient + *ApiTokenCache +} + type InvalidPathHandler struct{} type OptionsHandler struct{} @@ -247,14 +260,20 @@ 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") + + // List all blocks + rest.Handle(`/index`, IndexHandler{kc, t}).Methods("GET") + + // List blocks whose hash has the given prefix + rest.Handle(`/index/{prefix:[0-9a-f]{0,32}}`, IndexHandler{kc, t}).Methods("GET") } 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") @@ -282,22 +301,34 @@ func (this OptionsHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request SetCorsHeaders(resp) } -func (this GetBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) { - SetCorsHeaders(resp) +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") - kc := *this.KeepClient +var removeHint, _ = regexp.Compile("\\+K@[a-z0-9]{5}(\\+|$)") - 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 begin", 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) + if pass, tok = CheckAuthorizationHeader(&kc, this.ApiTokenCache, req); !pass { + status, err = http.StatusForbidden, BadAuthorizationHeader return } @@ -307,92 +338,98 @@ 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) + locator = removeHint.ReplaceAllString(locator, "$1") + + 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() } - } else if req.Method == "HEAD" { - blocklen, _, err = kc.AuthorizedAsk(hash, locator.Signature, locator.Timestamp) + default: + status, err = http.StatusNotImplemented, MethodNotSupported + return } - if blocklen == -1 { - log.Printf("%s: %s %s Keep server did not return Content-Length", - GetRemoteAddress(req), req.Method, hash) + if expectLength == -1 { + log.Println("Warning:", GetRemoteAddress(req), req.Method, proxiedURI, "Content-Length not provided") } - var status = 0 - switch err { + switch respErr := err.(type) { case nil: status = http.StatusOK - resp.Header().Set("Content-Length", fmt.Sprint(blocklen)) - if reader != nil { - n, err2 := io.Copy(resp, reader) - if blocklen > -1 && n != blocklen { - log.Printf("%s: %s %s %v %v mismatched copy size expected Content-Length: %v", - GetRemoteAddress(req), req.Method, hash, status, n, blocklen) - } else if err2 == nil { - log.Printf("%s: %s %s %v %v", - GetRemoteAddress(req), req.Method, hash, status, n) - } else { - log.Printf("%s: %s %s %v %v copy error: %v", - GetRemoteAddress(req), req.Method, hash, status, n, err2.Error()) + 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 } + } + case keepclient.Error: + if respErr == keepclient.BlockNotFound { + status = http.StatusNotFound + } else if respErr.Temporary() { + status = http.StatusBadGateway } else { - log.Printf("%s: %s %s %v 0", GetRemoteAddress(req), req.Method, hash, status) + status = 422 } - case keepclient.BlockNotFound: - status = http.StatusNotFound - http.Error(resp, "Not Found", http.StatusNotFound) default: - status = http.StatusBadGateway - http.Error(resp, err.Error(), http.StatusBadGateway) - } - - if err != nil { - log.Printf("%s: %s %s %v error: %v", - GetRemoteAddress(req), req.Method, hash, status, err.Error()) + status = http.StatusInternalServerError } } +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 - - hash := mux.Vars(req)["hash"] - hints := mux.Vars(req)["hints"] - - locator := keepclient.MakeLocator2(hash, hints) - - var contentLength int64 = -1 - if req.Header.Get("Content-Length") != "" { - _, err := fmt.Sscanf(req.Header.Get("Content-Length"), "%d", &contentLength) - if err != nil { - resp.Header().Set("Content-Length", fmt.Sprintf("%d", contentLength)) + var err error + var expectLength int64 + 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) } + }() - } - - log.Printf("%s: %s %s Content-Length %v", GetRemoteAddress(req), req.Method, hash, contentLength) + locatorIn := mux.Vars(req)["locator"] - if contentLength < 0 { - http.Error(resp, "Must include Content-Length header", http.StatusLengthRequired) + _, err = fmt.Sscanf(req.Header.Get("Content-Length"), "%d", &expectLength) + if err != nil || 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) + if pass, tok = CheckAuthorizationHeader(&kc, this.ApiTokenCache, req); !pass { + err = BadAuthorizationHeader + status = http.StatusForbidden return } @@ -405,63 +442,108 @@ func (this PutBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques if req.Header.Get("X-Keep-Desired-Replicas") != "" { var r int _, err := fmt.Sscanf(req.Header.Get(keepclient.X_Keep_Desired_Replicas), "%d", &r) - if err != nil { + if err == nil { kc.Want_replicas = r } } // Now try to put the block through - var replicas int - var put_err error - if hash == "" { + if locatorIn == "" { if bytes, err := ioutil.ReadAll(req.Body); err != nil { - msg := fmt.Sprintf("Error reading request body: %s", err) - log.Printf(msg) - http.Error(resp, msg, http.StatusInternalServerError) + err = errors.New(fmt.Sprintf("Error reading request body: %s", err)) + status = http.StatusInternalServerError return } else { - hash, replicas, put_err = kc.PutB(bytes) + locatorOut, wroteReplicas, err = kc.PutB(bytes) } } else { - hash, replicas, put_err = kc.PutHR(hash, req.Body, contentLength) + 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 put_err { + 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, put_err.Error(), http.StatusServiceUnavailable) + status = http.StatusServiceUnavailable } default: - http.Error(resp, put_err.Error(), http.StatusBadGateway) + status = http.StatusBadGateway + } +} + +// ServeHTTP implementation for IndexHandler +// Supports only GET requests for /index/{prefix:[0-9a-f]{0,32}} +// For each keep server found in LocalRoots: +// Invokes GetIndex using keepclient +// Expects "complete" response (terminating with blank new line) +// Aborts on any errors +// Concatenates responses from all those keep servers and returns +func (handler IndexHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) { + SetCorsHeaders(resp) + + prefix := mux.Vars(req)["prefix"] + var err error + var status int + + defer func() { + if status != http.StatusOK { + http.Error(resp, err.Error(), status) + } + }() + + kc := *handler.KeepClient + + ok, token := CheckAuthorizationHeader(&kc, handler.ApiTokenCache, req) + if !ok { + status, err = http.StatusForbidden, BadAuthorizationHeader + return + } + + // Copy ArvadosClient struct and use the client's API token + arvclient := *kc.Arvados + arvclient.ApiToken = token + kc.Arvados = &arvclient + + // Only GET method is supported + if req.Method != "GET" { + status, err = http.StatusNotImplemented, MethodNotSupported + return } - if put_err != nil { - log.Printf("%s: %s %s stored %v replicas (desired %v) got error %v", GetRemoteAddress(req), req.Method, hash, replicas, kc.Want_replicas, put_err.Error()) + // Get index from all LocalRoots and write to resp + var reader io.Reader + for uuid := range kc.LocalRoots() { + reader, err = kc.GetIndex(uuid, prefix) + if err != nil { + status = http.StatusBadGateway + return + } + + _, err = io.Copy(resp, reader) + if err != nil { + status = http.StatusBadGateway + return + } } + // Got index from all the keep servers and wrote to resp + status = http.StatusOK + resp.Write([]byte("\n")) }