X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/0f535af4fbcb7a9f6ca1e25675b31cc9656d88f3..4fd89ed7b10f0860a6030c25e44d4df45a087b2e:/services/keepproxy/keepproxy.go diff --git a/services/keepproxy/keepproxy.go b/services/keepproxy/keepproxy.go index 8cfaa90458..76a8a1551f 100644 --- a/services/keepproxy/keepproxy.go +++ b/services/keepproxy/keepproxy.go @@ -4,9 +4,6 @@ import ( "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" @@ -14,98 +11,135 @@ import ( "net/http" "os" "os/signal" - "reflect" "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/ghodss/yaml" + "github.com/gorilla/mux" ) -// Default TCP address on which to listen for requests. -// Override with -listen. -const DefaultAddr = ":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 - ) + cfg := DefaultConfig() flagset := flag.NewFlagSet("keepproxy", flag.ExitOnError) - - flagset.StringVar( - &listen, - "listen", - DefaultAddr, - "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") - + 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`") + dumpConfig := flagset.Bool("dump-config", false, "write current configuration to stdout and exit") flagset.Parse(os.Args[1:]) - if pidfile != "" { - f, err := os.Create(pidfile) - if err != nil { - log.Fatalf("Error writing pid file (%s): %s", pidfile, err.Error()) + 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) } - fmt.Fprint(f, os.Getpid()) - f.Close() - defer os.Remove(pidfile) + 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 y, err := yaml.Marshal(cfg); err == nil && !*dumpConfig { + log.Print("Current configuration:\n", string(y)) + } + cfg.Timeout = arvados.Duration(time.Duration(*timeoutSeconds) * time.Second) + } + + if *dumpConfig { + log.Fatal(config.DumpAndExit(cfg)) } - arv, err := arvadosclient.MakeArvadosClient() + arv, err := arvadosclient.New(&cfg.Client) if err != nil { - log.Fatalf("setting up arvados client: %v", err) + 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("setting up keep client: %v", err) + log.Fatalf("Error setting up keep client %s", err.Error()) } - kc.Want_replicas = default_replicas - kc.Client.Timeout = time.Duration(timeout) * time.Second - go RefreshServicesList(kc, 5*time.Minute, 3*time.Second) - listener, err = net.Listen("tcp", listen) + 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.Fatal("sync(%s): %s", cfg.PIDFile, err) + } + } + + 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", cfg.Listen) if err != nil { - log.Fatalf("Could not listen on %v", listen) + log.Fatalf("listen(%s): %s", cfg.Listen, err) + } + if _, err := daemon.SdNotify(false, "READY=1"); err != nil { + log.Printf("Error notifying init daemon: %v", err) } - log.Printf("Arvados Keep proxy started listening on %v", listener.Addr()) + log.Println("Listening at", listener.Addr()) // Shut down the server gracefully (by closing the listener) // if SIGTERM is received. @@ -119,7 +153,7 @@ func main() { signal.Notify(term, syscall.SIGINT) // Start serving requests. - http.Serve(listener, MakeRESTRouter(!no_get, !no_put, kc)) + http.Serve(listener, MakeRESTRouter(!cfg.DisableGet, !cfg.DisablePut, kc)) log.Println("shutting down") } @@ -130,42 +164,6 @@ type ApiTokenCache struct { expireTime int64 } -// Refresh the keep service list on SIGHUP; when the given interval -// has elapsed since the last refresh; and (if the last refresh -// failed) the given errInterval has elapsed. -func RefreshServicesList(kc *keepclient.KeepClient, interval, errInterval time.Duration) { - var previousRoots = []map[string]string{} - - timer := time.NewTimer(interval) - gotHUP := make(chan os.Signal, 1) - signal.Notify(gotHUP, syscall.SIGHUP) - - for { - select { - case <-gotHUP: - case <-timer.C: - } - timer.Reset(interval) - - if err := kc.DiscoverKeepServers(); err != nil { - log.Println("Error retrieving services list: %v (retrying in %v)", err, errInterval) - timer.Reset(errInterval) - 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]) - previousRoots = newRoots - } - - if len(newRoots[0]) == 0 { - log.Printf("WARNING: No local services (retrying in %v)", errInterval) - timer.Reset(errInterval) - } - } -} - // 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) { @@ -204,7 +202,7 @@ func GetRemoteAddress(req *http.Request) string { 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, "" @@ -217,7 +215,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 } @@ -334,7 +332,7 @@ func (this GetBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques var pass bool var tok string - if pass, tok = CheckAuthorizationHeader(kc, this.ApiTokenCache, req); !pass { + if pass, tok = CheckAuthorizationHeader(&kc, this.ApiTokenCache, req); !pass { status, err = http.StatusForbidden, BadAuthorizationHeader return } @@ -365,7 +363,7 @@ func (this GetBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques log.Println("Warning:", GetRemoteAddress(req), req.Method, proxiedURI, "Content-Length not provided") } - switch err { + switch respErr := err.(type) { case nil: status = http.StatusOK resp.Header().Set("Content-Length", fmt.Sprint(expectLength)) @@ -378,10 +376,16 @@ func (this GetBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques err = ContentLengthMismatch } } - case keepclient.BlockNotFound: - status = http.StatusNotFound + case keepclient.Error: + if respErr == keepclient.BlockNotFound { + status = http.StatusNotFound + } else if respErr.Temporary() { + status = http.StatusBadGateway + } else { + status = 422 + } default: - status = http.StatusBadGateway + status = http.StatusInternalServerError } } @@ -393,7 +397,7 @@ func (this PutBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques kc := *this.KeepClient var err error - var expectLength int64 = -1 + var expectLength int64 var status = http.StatusInternalServerError var wroteReplicas int var locatorOut string = "-" @@ -407,15 +411,8 @@ func (this PutBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques locatorIn := mux.Vars(req)["locator"] - if req.Header.Get("Content-Length") != "" { - _, err := fmt.Sscanf(req.Header.Get("Content-Length"), "%d", &expectLength) - if err != nil { - resp.Header().Set("Content-Length", fmt.Sprintf("%d", expectLength)) - } - - } - - if expectLength < 0 { + _, err = fmt.Sscanf(req.Header.Get("Content-Length"), "%d", &expectLength) + if err != nil || expectLength < 0 { err = LengthRequiredError status = http.StatusLengthRequired return @@ -435,7 +432,7 @@ func (this PutBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques var pass bool var tok string - if pass, tok = CheckAuthorizationHeader(kc, this.ApiTokenCache, req); !pass { + if pass, tok = CheckAuthorizationHeader(&kc, this.ApiTokenCache, req); !pass { err = BadAuthorizationHeader status = http.StatusForbidden return @@ -450,7 +447,7 @@ 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 } } @@ -518,7 +515,7 @@ func (handler IndexHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques kc := *handler.KeepClient - ok, token := CheckAuthorizationHeader(kc, handler.ApiTokenCache, req) + ok, token := CheckAuthorizationHeader(&kc, handler.ApiTokenCache, req) if !ok { status, err = http.StatusForbidden, BadAuthorizationHeader return