1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
23 "git.curoverse.com/arvados.git/sdk/go/arvados"
24 "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
25 "git.curoverse.com/arvados.git/sdk/go/config"
26 "git.curoverse.com/arvados.git/sdk/go/health"
27 "git.curoverse.com/arvados.git/sdk/go/httpserver"
28 "git.curoverse.com/arvados.git/sdk/go/keepclient"
29 log "github.com/Sirupsen/logrus"
30 "github.com/coreos/go-systemd/daemon"
31 "github.com/ghodss/yaml"
32 "github.com/gorilla/mux"
43 Timeout arvados.Duration
46 ManagementToken string
49 func DefaultConfig() *Config {
52 Timeout: arvados.Duration(15 * time.Second),
61 const rfc3339NanoFixed = "2006-01-02T15:04:05.000000000Z07:00"
64 log.SetFormatter(&log.JSONFormatter{
65 TimestampFormat: rfc3339NanoFixed,
68 cfg := DefaultConfig()
70 flagset := flag.NewFlagSet("keepproxy", flag.ExitOnError)
73 const deprecated = " (DEPRECATED -- use config file instead)"
74 flagset.StringVar(&cfg.Listen, "listen", cfg.Listen, "Local port to listen on."+deprecated)
75 flagset.BoolVar(&cfg.DisableGet, "no-get", cfg.DisableGet, "Disable GET operations."+deprecated)
76 flagset.BoolVar(&cfg.DisablePut, "no-put", cfg.DisablePut, "Disable PUT operations."+deprecated)
77 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)
78 flagset.StringVar(&cfg.PIDFile, "pid", cfg.PIDFile, "Path to write pid file."+deprecated)
79 timeoutSeconds := flagset.Int("timeout", int(time.Duration(cfg.Timeout)/time.Second), "Timeout (in seconds) on requests to internal Keep services."+deprecated)
80 flagset.StringVar(&cfg.ManagementToken, "management-token", cfg.ManagementToken, "Authorization token to be included in all health check requests.")
83 const defaultCfgPath = "/etc/arvados/keepproxy/keepproxy.yml"
84 flagset.StringVar(&cfgPath, "config", defaultCfgPath, "Configuration file `path`")
85 dumpConfig := flagset.Bool("dump-config", false, "write current configuration to stdout and exit")
86 getVersion := flagset.Bool("version", false, "Print version information and exit.")
87 flagset.Parse(os.Args[1:])
89 // Print version information if requested
91 fmt.Printf("keepproxy %s\n", version)
95 err := config.LoadFile(cfg, cfgPath)
97 h := os.Getenv("ARVADOS_API_HOST")
98 t := os.Getenv("ARVADOS_API_TOKEN")
99 if h == "" || t == "" || !os.IsNotExist(err) || cfgPath != defaultCfgPath {
102 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.")
103 cfg.Client.APIHost = h
104 cfg.Client.AuthToken = t
105 if regexp.MustCompile("^(?i:1|yes|true)$").MatchString(os.Getenv("ARVADOS_API_HOST_INSECURE")) {
106 cfg.Client.Insecure = true
108 if y, err := yaml.Marshal(cfg); err == nil && !*dumpConfig {
109 log.Print("Current configuration:\n", string(y))
111 cfg.Timeout = arvados.Duration(time.Duration(*timeoutSeconds) * time.Second)
115 log.Fatal(config.DumpAndExit(cfg))
118 log.Printf("keepproxy %s started", version)
120 arv, err := arvadosclient.New(&cfg.Client)
122 log.Fatalf("Error setting up arvados client %s", err.Error())
126 keepclient.DebugPrintf = log.Printf
128 kc, err := keepclient.MakeKeepClient(arv)
130 log.Fatalf("Error setting up keep client %s", err.Error())
132 keepclient.RefreshServiceDiscoveryOnSIGHUP()
134 if cfg.PIDFile != "" {
135 f, err := os.Create(cfg.PIDFile)
140 err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
142 log.Fatalf("flock(%s): %s", cfg.PIDFile, err)
144 defer os.Remove(cfg.PIDFile)
147 log.Fatalf("truncate(%s): %s", cfg.PIDFile, err)
149 _, err = fmt.Fprint(f, os.Getpid())
151 log.Fatalf("write(%s): %s", cfg.PIDFile, err)
155 log.Fatal("sync(%s): %s", cfg.PIDFile, err)
159 if cfg.DefaultReplicas > 0 {
160 kc.Want_replicas = cfg.DefaultReplicas
163 listener, err = net.Listen("tcp", cfg.Listen)
165 log.Fatalf("listen(%s): %s", cfg.Listen, err)
167 if _, err := daemon.SdNotify(false, "READY=1"); err != nil {
168 log.Printf("Error notifying init daemon: %v", err)
170 log.Println("Listening at", listener.Addr())
172 // Shut down the server gracefully (by closing the listener)
173 // if SIGTERM is received.
174 term := make(chan os.Signal, 1)
175 go func(sig <-chan os.Signal) {
177 log.Println("caught signal:", s)
180 signal.Notify(term, syscall.SIGTERM)
181 signal.Notify(term, syscall.SIGINT)
183 // Start serving requests.
184 router = MakeRESTRouter(!cfg.DisableGet, !cfg.DisablePut, kc, time.Duration(cfg.Timeout), cfg.ManagementToken)
185 http.Serve(listener, httpserver.AddRequestIDs(httpserver.LogRequests(router)))
187 log.Println("shutting down")
190 type ApiTokenCache struct {
191 tokens map[string]int64
196 // Cache the token and set an expire time. If we already have an expire time
197 // on the token, it is not updated.
198 func (this *ApiTokenCache) RememberToken(token string) {
200 defer this.lock.Unlock()
202 now := time.Now().Unix()
203 if this.tokens[token] == 0 {
204 this.tokens[token] = now + this.expireTime
208 // Check if the cached token is known and still believed to be valid.
209 func (this *ApiTokenCache) RecallToken(token string) bool {
211 defer this.lock.Unlock()
213 now := time.Now().Unix()
214 if this.tokens[token] == 0 {
217 } else if now < this.tokens[token] {
218 // Token is known and still valid
222 this.tokens[token] = 0
227 func GetRemoteAddress(req *http.Request) string {
228 if xff := req.Header.Get("X-Forwarded-For"); xff != "" {
229 return xff + "," + req.RemoteAddr
231 return req.RemoteAddr
234 func CheckAuthorizationHeader(kc *keepclient.KeepClient, cache *ApiTokenCache, req *http.Request) (pass bool, tok string) {
235 parts := strings.SplitN(req.Header.Get("Authorization"), " ", 2)
236 if len(parts) < 2 || !(parts[0] == "OAuth2" || parts[0] == "Bearer") || len(parts[1]) == 0 {
241 // Tokens are validated differently depending on what kind of
242 // operation is being performed. For example, tokens in
243 // collection-sharing links permit GET requests, but not
246 if req.Method == "GET" || req.Method == "HEAD" {
252 if cache.RecallToken(op + ":" + tok) {
253 // Valid in the cache, short circuit
261 err = arv.Call("HEAD", "keep_services", "", "accessible", nil, nil)
263 err = arv.Call("HEAD", "users", "", "current", nil, nil)
266 log.Printf("%s: CheckAuthorizationHeader error: %v", GetRemoteAddress(req), err)
270 // Success! Update cache
271 cache.RememberToken(op + ":" + tok)
276 type proxyHandler struct {
278 *keepclient.KeepClient
280 timeout time.Duration
281 transport *http.Transport
284 // MakeRESTRouter returns an http.Handler that passes GET and PUT
285 // requests to the appropriate handlers.
286 func MakeRESTRouter(enable_get bool, enable_put bool, kc *keepclient.KeepClient, timeout time.Duration, mgmtToken string) http.Handler {
287 rest := mux.NewRouter()
289 transport := *(http.DefaultTransport.(*http.Transport))
290 transport.DialContext = (&net.Dialer{
291 Timeout: keepclient.DefaultConnectTimeout,
292 KeepAlive: keepclient.DefaultKeepAlive,
295 transport.TLSClientConfig = arvadosclient.MakeTLSConfig(kc.Arvados.ApiInsecure)
296 transport.TLSHandshakeTimeout = keepclient.DefaultTLSHandshakeTimeout
302 transport: &transport,
303 ApiTokenCache: &ApiTokenCache{
304 tokens: make(map[string]int64),
310 rest.HandleFunc(`/{locator:[0-9a-f]{32}\+.*}`, h.Get).Methods("GET", "HEAD")
311 rest.HandleFunc(`/{locator:[0-9a-f]{32}}`, h.Get).Methods("GET", "HEAD")
314 rest.HandleFunc(`/index`, h.Index).Methods("GET")
316 // List blocks whose hash has the given prefix
317 rest.HandleFunc(`/index/{prefix:[0-9a-f]{0,32}}`, h.Index).Methods("GET")
321 rest.HandleFunc(`/{locator:[0-9a-f]{32}\+.*}`, h.Put).Methods("PUT")
322 rest.HandleFunc(`/{locator:[0-9a-f]{32}}`, h.Put).Methods("PUT")
323 rest.HandleFunc(`/`, h.Put).Methods("POST")
324 rest.HandleFunc(`/{any}`, h.Options).Methods("OPTIONS")
325 rest.HandleFunc(`/`, h.Options).Methods("OPTIONS")
328 rest.Handle("/_health/{check}", &health.Handler{
333 rest.NotFoundHandler = InvalidPathHandler{}
337 var errLoopDetected = errors.New("loop detected")
339 func (*proxyHandler) checkLoop(resp http.ResponseWriter, req *http.Request) error {
340 if via := req.Header.Get("Via"); strings.Index(via, " "+viaAlias) >= 0 {
341 log.Printf("proxy loop detected (request has Via: %q): perhaps keepproxy is misidentified by gateway config as an external client, or its keep_services record does not have service_type=proxy?", via)
342 http.Error(resp, errLoopDetected.Error(), http.StatusInternalServerError)
343 return errLoopDetected
348 func SetCorsHeaders(resp http.ResponseWriter) {
349 resp.Header().Set("Access-Control-Allow-Methods", "GET, HEAD, POST, PUT, OPTIONS")
350 resp.Header().Set("Access-Control-Allow-Origin", "*")
351 resp.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Length, Content-Type, X-Keep-Desired-Replicas")
352 resp.Header().Set("Access-Control-Max-Age", "86486400")
355 type InvalidPathHandler struct{}
357 func (InvalidPathHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
358 log.Printf("%s: %s %s unroutable", GetRemoteAddress(req), req.Method, req.URL.Path)
359 http.Error(resp, "Bad request", http.StatusBadRequest)
362 func (h *proxyHandler) Options(resp http.ResponseWriter, req *http.Request) {
363 log.Printf("%s: %s %s", GetRemoteAddress(req), req.Method, req.URL.Path)
367 var BadAuthorizationHeader = errors.New("Missing or invalid Authorization header")
368 var ContentLengthMismatch = errors.New("Actual length != expected content length")
369 var MethodNotSupported = errors.New("Method not supported")
371 var removeHint, _ = regexp.Compile("\\+K@[a-z0-9]{5}(\\+|$)")
373 func (h *proxyHandler) Get(resp http.ResponseWriter, req *http.Request) {
374 if err := h.checkLoop(resp, req); err != nil {
378 resp.Header().Set("Via", req.Proto+" "+viaAlias)
380 locator := mux.Vars(req)["locator"]
383 var expectLength, responseLength int64
387 log.Println(GetRemoteAddress(req), req.Method, req.URL.Path, status, expectLength, responseLength, proxiedURI, err)
388 if status != http.StatusOK {
389 http.Error(resp, err.Error(), status)
393 kc := h.makeKeepClient(req)
397 if pass, tok = CheckAuthorizationHeader(kc, h.ApiTokenCache, req); !pass {
398 status, err = http.StatusForbidden, BadAuthorizationHeader
402 // Copy ArvadosClient struct and use the client's API token
403 arvclient := *kc.Arvados
404 arvclient.ApiToken = tok
405 kc.Arvados = &arvclient
407 var reader io.ReadCloser
409 locator = removeHint.ReplaceAllString(locator, "$1")
413 expectLength, proxiedURI, err = kc.Ask(locator)
415 reader, expectLength, proxiedURI, err = kc.Get(locator)
420 status, err = http.StatusNotImplemented, MethodNotSupported
424 if expectLength == -1 {
425 log.Println("Warning:", GetRemoteAddress(req), req.Method, proxiedURI, "Content-Length not provided")
428 switch respErr := err.(type) {
430 status = http.StatusOK
431 resp.Header().Set("Content-Length", fmt.Sprint(expectLength))
436 responseLength, err = io.Copy(resp, reader)
437 if err == nil && expectLength > -1 && responseLength != expectLength {
438 err = ContentLengthMismatch
441 case keepclient.Error:
442 if respErr == keepclient.BlockNotFound {
443 status = http.StatusNotFound
444 } else if respErr.Temporary() {
445 status = http.StatusBadGateway
450 status = http.StatusInternalServerError
454 var LengthRequiredError = errors.New(http.StatusText(http.StatusLengthRequired))
455 var LengthMismatchError = errors.New("Locator size hint does not match Content-Length header")
457 func (h *proxyHandler) Put(resp http.ResponseWriter, req *http.Request) {
458 if err := h.checkLoop(resp, req); err != nil {
462 resp.Header().Set("Via", "HTTP/1.1 "+viaAlias)
464 kc := h.makeKeepClient(req)
467 var expectLength int64
468 var status = http.StatusInternalServerError
469 var wroteReplicas int
470 var locatorOut string = "-"
473 log.Println(GetRemoteAddress(req), req.Method, req.URL.Path, status, expectLength, kc.Want_replicas, wroteReplicas, locatorOut, err)
474 if status != http.StatusOK {
475 http.Error(resp, err.Error(), status)
479 locatorIn := mux.Vars(req)["locator"]
481 _, err = fmt.Sscanf(req.Header.Get("Content-Length"), "%d", &expectLength)
482 if err != nil || expectLength < 0 {
483 err = LengthRequiredError
484 status = http.StatusLengthRequired
489 var loc *keepclient.Locator
490 if loc, err = keepclient.MakeLocator(locatorIn); err != nil {
491 status = http.StatusBadRequest
493 } else if loc.Size > 0 && int64(loc.Size) != expectLength {
494 err = LengthMismatchError
495 status = http.StatusBadRequest
502 if pass, tok = CheckAuthorizationHeader(kc, h.ApiTokenCache, req); !pass {
503 err = BadAuthorizationHeader
504 status = http.StatusForbidden
508 // Copy ArvadosClient struct and use the client's API token
509 arvclient := *kc.Arvados
510 arvclient.ApiToken = tok
511 kc.Arvados = &arvclient
513 // Check if the client specified the number of replicas
514 if req.Header.Get("X-Keep-Desired-Replicas") != "" {
516 _, err := fmt.Sscanf(req.Header.Get(keepclient.X_Keep_Desired_Replicas), "%d", &r)
522 // Now try to put the block through
524 bytes, err2 := ioutil.ReadAll(req.Body)
526 _ = errors.New(fmt.Sprintf("Error reading request body: %s", err2))
527 status = http.StatusInternalServerError
530 locatorOut, wroteReplicas, err = kc.PutB(bytes)
532 locatorOut, wroteReplicas, err = kc.PutHR(locatorIn, req.Body, expectLength)
535 // Tell the client how many successful PUTs we accomplished
536 resp.Header().Set(keepclient.X_Keep_Replicas_Stored, fmt.Sprintf("%d", wroteReplicas))
540 status = http.StatusOK
541 _, err = io.WriteString(resp, locatorOut)
543 case keepclient.OversizeBlockError:
545 status = http.StatusRequestEntityTooLarge
547 case keepclient.InsufficientReplicasError:
548 if wroteReplicas > 0 {
549 // At least one write is considered success. The
550 // client can decide if getting less than the number of
551 // replications it asked for is a fatal error.
552 status = http.StatusOK
553 _, err = io.WriteString(resp, locatorOut)
555 status = http.StatusServiceUnavailable
559 status = http.StatusBadGateway
563 // ServeHTTP implementation for IndexHandler
564 // Supports only GET requests for /index/{prefix:[0-9a-f]{0,32}}
565 // For each keep server found in LocalRoots:
566 // Invokes GetIndex using keepclient
567 // Expects "complete" response (terminating with blank new line)
568 // Aborts on any errors
569 // Concatenates responses from all those keep servers and returns
570 func (h *proxyHandler) Index(resp http.ResponseWriter, req *http.Request) {
573 prefix := mux.Vars(req)["prefix"]
578 if status != http.StatusOK {
579 http.Error(resp, err.Error(), status)
583 kc := h.makeKeepClient(req)
584 ok, token := CheckAuthorizationHeader(kc, h.ApiTokenCache, req)
586 status, err = http.StatusForbidden, BadAuthorizationHeader
590 // Copy ArvadosClient struct and use the client's API token
591 arvclient := *kc.Arvados
592 arvclient.ApiToken = token
593 kc.Arvados = &arvclient
595 // Only GET method is supported
596 if req.Method != "GET" {
597 status, err = http.StatusNotImplemented, MethodNotSupported
601 // Get index from all LocalRoots and write to resp
603 for uuid := range kc.LocalRoots() {
604 reader, err = kc.GetIndex(uuid, prefix)
606 status = http.StatusBadGateway
610 _, err = io.Copy(resp, reader)
612 status = http.StatusBadGateway
617 // Got index from all the keep servers and wrote to resp
618 status = http.StatusOK
619 resp.Write([]byte("\n"))
622 func (h *proxyHandler) makeKeepClient(req *http.Request) *keepclient.KeepClient {
624 kc.HTTPClient = &proxyClient{
625 client: &http.Client{
627 Transport: h.transport,
630 requestID: req.Header.Get("X-Request-Id"),