X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/fe45b1b66c730f2546d78a7899375707c0816518..38c4ce8bd5aefed6784d457ed3caf28f279d6de4:/services/keepproxy/keepproxy.go diff --git a/services/keepproxy/keepproxy.go b/services/keepproxy/keepproxy.go index 65f7a42cd9..b6c8bd66aa 100644 --- a/services/keepproxy/keepproxy.go +++ b/services/keepproxy/keepproxy.go @@ -1,3 +1,7 @@ +// Copyright (C) The Arvados Authors. All rights reserved. +// +// SPDX-License-Identifier: AGPL-3.0 + package main import ( @@ -6,7 +10,6 @@ import ( "fmt" "io" "io/ioutil" - "log" "net" "net/http" "os" @@ -20,12 +23,17 @@ import ( "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/health" + "git.curoverse.com/arvados.git/sdk/go/httpserver" "git.curoverse.com/arvados.git/sdk/go/keepclient" + log "github.com/Sirupsen/logrus" "github.com/coreos/go-systemd/daemon" "github.com/ghodss/yaml" "github.com/gorilla/mux" ) +var version = "dev" + type Config struct { Client arvados.Client Listen string @@ -35,6 +43,7 @@ type Config struct { Timeout arvados.Duration PIDFile string Debug bool + ManagementToken string } func DefaultConfig() *Config { @@ -49,7 +58,13 @@ var ( router http.Handler ) +const rfc3339NanoFixed = "2006-01-02T15:04:05.000000000Z07:00" + func main() { + log.SetFormatter(&log.JSONFormatter{ + TimestampFormat: rfc3339NanoFixed, + }) + cfg := DefaultConfig() flagset := flag.NewFlagSet("keepproxy", flag.ExitOnError) @@ -62,13 +77,21 @@ func main() { 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) + flagset.StringVar(&cfg.ManagementToken, "management-token", cfg.ManagementToken, "Authorization token to be included in all health check requests.") 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") + getVersion := flagset.Bool("version", false, "Print version information and exit.") flagset.Parse(os.Args[1:]) + // Print version information if requested + if *getVersion { + fmt.Printf("keepproxy %s\n", version) + return + } + err := config.LoadFile(cfg, cfgPath) if err != nil { h := os.Getenv("ARVADOS_API_HOST") @@ -92,6 +115,8 @@ func main() { log.Fatal(config.DumpAndExit(cfg)) } + log.Printf("keepproxy %s started", version) + arv, err := arvadosclient.New(&cfg.Client) if err != nil { log.Fatalf("Error setting up arvados client %s", err.Error()) @@ -104,6 +129,7 @@ func main() { if err != nil { log.Fatalf("Error setting up keep client %s", err.Error()) } + keepclient.RefreshServiceDiscoveryOnSIGHUP() if cfg.PIDFile != "" { f, err := os.Create(cfg.PIDFile) @@ -133,8 +159,6 @@ func main() { if cfg.DefaultReplicas > 0 { kc.Want_replicas = cfg.DefaultReplicas } - kc.Client.(*http.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 { @@ -157,8 +181,8 @@ func main() { signal.Notify(term, syscall.SIGINT) // Start serving requests. - router = MakeRESTRouter(!cfg.DisableGet, !cfg.DisablePut, kc) - http.Serve(listener, router) + router = MakeRESTRouter(!cfg.DisableGet, !cfg.DisablePut, kc, time.Duration(cfg.Timeout), cfg.ManagementToken) + http.Serve(listener, httpserver.AddRequestIDs(httpserver.LogRequests(nil, router))) log.Println("shutting down") } @@ -208,48 +232,83 @@ func GetRemoteAddress(req *http.Request) 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 == "" { + parts := strings.SplitN(req.Header.Get("Authorization"), " ", 2) + if len(parts) < 2 || !(parts[0] == "OAuth2" || parts[0] == "Bearer") || len(parts[1]) == 0 { return false, "" } + tok = parts[1] - _, err := fmt.Sscanf(auth, "OAuth2 %s", &tok) - if err != nil { - // Scanning error - return false, "" + // Tokens are validated differently depending on what kind of + // operation is being performed. For example, tokens in + // collection-sharing links permit GET requests, but not + // PUT requests. + var op string + if req.Method == "GET" || req.Method == "HEAD" { + op = "read" + } else { + op = "write" } - if cache.RecallToken(tok) { + if cache.RecallToken(op + ":" + tok) { // Valid in the cache, short circuit return true, tok } + var err error arv := *kc.Arvados arv.ApiToken = tok - if err := arv.Call("HEAD", "users", "", "current", nil, nil); err != nil { + arv.RequestID = req.Header.Get("X-Request-Id") + if op == "read" { + err = arv.Call("HEAD", "keep_services", "", "accessible", nil, nil) + } else { + err = arv.Call("HEAD", "users", "", "current", nil, nil) + } + if err != nil { log.Printf("%s: CheckAuthorizationHeader error: %v", GetRemoteAddress(req), err) return false, "" } // Success! Update cache - cache.RememberToken(tok) + cache.RememberToken(op + ":" + tok) return true, tok } +// We need to make a private copy of the default http transport early +// in initialization, then make copies of our private copy later. It +// won't be safe to copy http.DefaultTransport itself later, because +// its private mutexes might have already been used. (Without this, +// the test suite sometimes panics "concurrent map writes" in +// net/http.(*Transport).removeIdleConnLocked().) +var defaultTransport = *(http.DefaultTransport.(*http.Transport)) + type proxyHandler struct { http.Handler *keepclient.KeepClient *ApiTokenCache + timeout time.Duration + transport *http.Transport } // MakeRESTRouter returns an http.Handler that passes GET and PUT // requests to the appropriate handlers. -func MakeRESTRouter(enable_get bool, enable_put bool, kc *keepclient.KeepClient) http.Handler { +func MakeRESTRouter(enable_get bool, enable_put bool, kc *keepclient.KeepClient, timeout time.Duration, mgmtToken string) http.Handler { rest := mux.NewRouter() + + transport := defaultTransport + transport.DialContext = (&net.Dialer{ + Timeout: keepclient.DefaultConnectTimeout, + KeepAlive: keepclient.DefaultKeepAlive, + DualStack: true, + }).DialContext + transport.TLSClientConfig = arvadosclient.MakeTLSConfig(kc.Arvados.ApiInsecure) + transport.TLSHandshakeTimeout = keepclient.DefaultTLSHandshakeTimeout + h := &proxyHandler{ Handler: rest, KeepClient: kc, + timeout: timeout, + transport: &transport, ApiTokenCache: &ApiTokenCache{ tokens: make(map[string]int64), expireTime: 300, @@ -275,6 +334,11 @@ func MakeRESTRouter(enable_get bool, enable_put bool, kc *keepclient.KeepClient) rest.HandleFunc(`/`, h.Options).Methods("OPTIONS") } + rest.Handle("/_health/{check}", &health.Handler{ + Token: mgmtToken, + Prefix: "/_health/", + }).Methods("GET") + rest.NotFoundHandler = InvalidPathHandler{} return h } @@ -335,12 +399,11 @@ func (h *proxyHandler) Get(resp http.ResponseWriter, req *http.Request) { } }() - kc := *h.KeepClient - kc.Client = &proxyClient{client: kc.Client, proto: req.Proto} + kc := h.makeKeepClient(req) var pass bool var tok string - if pass, tok = CheckAuthorizationHeader(&kc, h.ApiTokenCache, req); !pass { + if pass, tok = CheckAuthorizationHeader(kc, h.ApiTokenCache, req); !pass { status, err = http.StatusForbidden, BadAuthorizationHeader return } @@ -407,8 +470,7 @@ func (h *proxyHandler) Put(resp http.ResponseWriter, req *http.Request) { SetCorsHeaders(resp) resp.Header().Set("Via", "HTTP/1.1 "+viaAlias) - kc := *h.KeepClient - kc.Client = &proxyClient{client: kc.Client, proto: req.Proto} + kc := h.makeKeepClient(req) var err error var expectLength int64 @@ -425,6 +487,15 @@ func (h *proxyHandler) Put(resp http.ResponseWriter, req *http.Request) { locatorIn := mux.Vars(req)["locator"] + // Check if the client specified storage classes + if req.Header.Get("X-Keep-Storage-Classes") != "" { + var scl []string + for _, sc := range strings.Split(req.Header.Get("X-Keep-Storage-Classes"), ",") { + scl = append(scl, strings.Trim(sc, " ")) + } + kc.StorageClasses = scl + } + _, err = fmt.Sscanf(req.Header.Get("Content-Length"), "%d", &expectLength) if err != nil || expectLength < 0 { err = LengthRequiredError @@ -446,7 +517,7 @@ func (h *proxyHandler) Put(resp http.ResponseWriter, req *http.Request) { var pass bool var tok string - if pass, tok = CheckAuthorizationHeader(&kc, h.ApiTokenCache, req); !pass { + if pass, tok = CheckAuthorizationHeader(kc, h.ApiTokenCache, req); !pass { err = BadAuthorizationHeader status = http.StatusForbidden return @@ -468,13 +539,13 @@ func (h *proxyHandler) Put(resp http.ResponseWriter, req *http.Request) { // Now try to put the block through if locatorIn == "" { - if bytes, err := ioutil.ReadAll(req.Body); err != nil { - err = errors.New(fmt.Sprintf("Error reading request body: %s", err)) + bytes, err2 := ioutil.ReadAll(req.Body) + if err2 != nil { + _ = errors.New(fmt.Sprintf("Error reading request body: %s", err2)) status = http.StatusInternalServerError return - } else { - locatorOut, wroteReplicas, err = kc.PutB(bytes) } + locatorOut, wroteReplicas, err = kc.PutB(bytes) } else { locatorOut, wroteReplicas, err = kc.PutHR(locatorIn, req.Body, expectLength) } @@ -527,9 +598,8 @@ func (h *proxyHandler) Index(resp http.ResponseWriter, req *http.Request) { } }() - kc := *h.KeepClient - - ok, token := CheckAuthorizationHeader(&kc, h.ApiTokenCache, req) + kc := h.makeKeepClient(req) + ok, token := CheckAuthorizationHeader(kc, h.ApiTokenCache, req) if !ok { status, err = http.StatusForbidden, BadAuthorizationHeader return @@ -566,3 +636,16 @@ func (h *proxyHandler) Index(resp http.ResponseWriter, req *http.Request) { status = http.StatusOK resp.Write([]byte("\n")) } + +func (h *proxyHandler) makeKeepClient(req *http.Request) *keepclient.KeepClient { + kc := *h.KeepClient + kc.RequestID = req.Header.Get("X-Request-Id") + kc.HTTPClient = &proxyClient{ + client: &http.Client{ + Timeout: h.timeout, + Transport: h.transport, + }, + proto: req.Proto, + } + return &kc +}