services/keepproxy/keepproxy.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package main
   6
   7 import (
   8         "errors"
   9         "flag"
  10         "fmt"
  11         "io"
  12         "io/ioutil"
  13         "net"
  14         "net/http"
  15         "os"
  16         "os/signal"
  17         "regexp"
  18         "strings"
  19         "syscall"
  20         "time"
  21
  22         "git.arvados.org/arvados.git/lib/cmd"
  23         "git.arvados.org/arvados.git/lib/config"
  24         "git.arvados.org/arvados.git/sdk/go/arvados"
  25         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
  26         "git.arvados.org/arvados.git/sdk/go/health"
  27         "git.arvados.org/arvados.git/sdk/go/httpserver"
  28         "git.arvados.org/arvados.git/sdk/go/keepclient"
  29         "github.com/coreos/go-systemd/daemon"
  30         "github.com/ghodss/yaml"
  31         "github.com/gorilla/mux"
  32         lru "github.com/hashicorp/golang-lru"
  33         log "github.com/sirupsen/logrus"
  34 )
  35
  36 var version = "dev"
  37
  38 var (
  39         listener net.Listener
  40         router   http.Handler
  41 )
  42
  43 const rfc3339NanoFixed = "2006-01-02T15:04:05.000000000Z07:00"
  44
  45 func configure(logger log.FieldLogger, args []string) (*arvados.Cluster, error) {
  46         prog := args[0]
  47         flags := flag.NewFlagSet(prog, flag.ContinueOnError)
  48
  49         dumpConfig := flags.Bool("dump-config", false, "write current configuration to stdout and exit")
  50         getVersion := flags.Bool("version", false, "Print version information and exit.")
  51
  52         loader := config.NewLoader(os.Stdin, logger)
  53         loader.SetupFlags(flags)
  54         args = loader.MungeLegacyConfigArgs(logger, args[1:], "-legacy-keepproxy-config")
  55
  56         if ok, code := cmd.ParseFlags(flags, prog, args, "", os.Stderr); !ok {
  57                 os.Exit(code)
  58         } else if *getVersion {
  59                 fmt.Printf("keepproxy %s\n", version)
  60                 return nil, nil
  61         }
  62
  63         cfg, err := loader.Load()
  64         if err != nil {
  65                 return nil, err
  66         }
  67         cluster, err := cfg.GetCluster("")
  68         if err != nil {
  69                 return nil, err
  70         }
  71
  72         if *dumpConfig {
  73                 out, err := yaml.Marshal(cfg)
  74                 if err != nil {
  75                         return nil, err
  76                 }
  77                 if _, err := os.Stdout.Write(out); err != nil {
  78                         return nil, err
  79                 }
  80                 return nil, nil
  81         }
  82         return cluster, nil
  83 }
  84
  85 func main() {
  86         logger := log.New()
  87         logger.Formatter = &log.JSONFormatter{
  88                 TimestampFormat: rfc3339NanoFixed,
  89         }
  90
  91         cluster, err := configure(logger, os.Args)
  92         if err != nil {
  93                 log.Fatal(err)
  94         }
  95         if cluster == nil {
  96                 return
  97         }
  98
  99         log.Printf("keepproxy %s started", version)
 100
 101         if err := run(logger, cluster); err != nil {
 102                 log.Fatal(err)
 103         }
 104
 105         log.Println("shutting down")
 106 }
 107
 108 func run(logger log.FieldLogger, cluster *arvados.Cluster) error {
 109         client, err := arvados.NewClientFromConfig(cluster)
 110         if err != nil {
 111                 return err
 112         }
 113         client.AuthToken = cluster.SystemRootToken
 114
 115         arv, err := arvadosclient.New(client)
 116         if err != nil {
 117                 return fmt.Errorf("Error setting up arvados client %v", err)
 118         }
 119
 120         // If a config file is available, use the keepstores defined there
 121         // instead of the legacy autodiscover mechanism via the API server
 122         for k := range cluster.Services.Keepstore.InternalURLs {
 123                 arv.KeepServiceURIs = append(arv.KeepServiceURIs, strings.TrimRight(k.String(), "/"))
 124         }
 125
 126         if cluster.SystemLogs.LogLevel == "debug" {
 127                 keepclient.DebugPrintf = log.Printf
 128         }
 129         kc, err := keepclient.MakeKeepClient(arv)
 130         if err != nil {
 131                 return fmt.Errorf("Error setting up keep client %v", err)
 132         }
 133         keepclient.RefreshServiceDiscoveryOnSIGHUP()
 134
 135         if cluster.Collections.DefaultReplication > 0 {
 136                 kc.Want_replicas = cluster.Collections.DefaultReplication
 137         }
 138
 139         var listen arvados.URL
 140         for listen = range cluster.Services.Keepproxy.InternalURLs {
 141                 break
 142         }
 143
 144         var lErr error
 145         listener, lErr = net.Listen("tcp", listen.Host)
 146         if lErr != nil {
 147                 return fmt.Errorf("listen(%s): %v", listen.Host, lErr)
 148         }
 149
 150         if _, err := daemon.SdNotify(false, "READY=1"); err != nil {
 151                 log.Printf("Error notifying init daemon: %v", err)
 152         }
 153         log.Println("listening at", listener.Addr())
 154
 155         // Shut down the server gracefully (by closing the listener)
 156         // if SIGTERM is received.
 157         term := make(chan os.Signal, 1)
 158         go func(sig <-chan os.Signal) {
 159                 s := <-sig
 160                 log.Println("caught signal:", s)
 161                 listener.Close()
 162         }(term)
 163         signal.Notify(term, syscall.SIGTERM)
 164         signal.Notify(term, syscall.SIGINT)
 165
 166         // Start serving requests.
 167         router, err = MakeRESTRouter(kc, time.Duration(keepclient.DefaultProxyRequestTimeout), cluster, logger)
 168         if err != nil {
 169                 return err
 170         }
 171         return http.Serve(listener, httpserver.AddRequestIDs(httpserver.LogRequests(router)))
 172 }
 173
 174 type TokenCacheEntry struct {
 175         expire int64
 176         user   *arvados.User
 177 }
 178
 179 type APITokenCache struct {
 180         tokens     *lru.TwoQueueCache
 181         expireTime int64
 182 }
 183
 184 // RememberToken caches the token and set an expire time.  If the
 185 // token is already in the cache, it is not updated.
 186 func (cache *APITokenCache) RememberToken(token string, user *arvados.User) {
 187         now := time.Now().Unix()
 188         _, ok := cache.tokens.Get(token)
 189         if !ok {
 190                 cache.tokens.Add(token, TokenCacheEntry{
 191                         expire: now + cache.expireTime,
 192                         user:   user,
 193                 })
 194         }
 195 }
 196
 197 // RecallToken checks if the cached token is known and still believed to be
 198 // valid.
 199 func (cache *APITokenCache) RecallToken(token string) (bool, *arvados.User) {
 200         val, ok := cache.tokens.Get(token)
 201         if !ok {
 202                 return false, nil
 203         }
 204
 205         cacheEntry := val.(TokenCacheEntry)
 206         now := time.Now().Unix()
 207         if now < cacheEntry.expire {
 208                 // Token is known and still valid
 209                 return true, cacheEntry.user
 210         } else {
 211                 // Token is expired
 212                 cache.tokens.Remove(token)
 213                 return false, nil
 214         }
 215 }
 216
 217 // GetRemoteAddress returns a string with the remote address for the request.
 218 // If the X-Forwarded-For header is set and has a non-zero length, it returns a
 219 // string made from a comma separated list of all the remote addresses,
 220 // starting with the one(s) from the X-Forwarded-For header.
 221 func GetRemoteAddress(req *http.Request) string {
 222         if xff := req.Header.Get("X-Forwarded-For"); xff != "" {
 223                 return xff + "," + req.RemoteAddr
 224         }
 225         return req.RemoteAddr
 226 }
 227
 228 func (h *proxyHandler) CheckAuthorizationHeader(req *http.Request) (pass bool, tok string, user *arvados.User) {
 229         parts := strings.SplitN(req.Header.Get("Authorization"), " ", 2)
 230         if len(parts) < 2 || !(parts[0] == "OAuth2" || parts[0] == "Bearer") || len(parts[1]) == 0 {
 231                 return false, "", nil
 232         }
 233         tok = parts[1]
 234
 235         // Tokens are validated differently depending on what kind of
 236         // operation is being performed. For example, tokens in
 237         // collection-sharing links permit GET requests, but not
 238         // PUT requests.
 239         var op string
 240         if req.Method == "GET" || req.Method == "HEAD" {
 241                 op = "read"
 242         } else {
 243                 op = "write"
 244         }
 245
 246         if ok, user := h.APITokenCache.RecallToken(op + ":" + tok); ok {
 247                 // Valid in the cache, short circuit
 248                 return true, tok, user
 249         }
 250
 251         var err error
 252         arv := *h.KeepClient.Arvados
 253         arv.ApiToken = tok
 254         arv.RequestID = req.Header.Get("X-Request-Id")
 255         user = &arvados.User{}
 256         userCurrentError := arv.Call("GET", "users", "", "current", nil, user)
 257         err = userCurrentError
 258         if err != nil && op == "read" {
 259                 apiError, ok := err.(arvadosclient.APIServerError)
 260                 if ok && apiError.HttpStatusCode == http.StatusForbidden {
 261                         // If it was a scoped "sharing" token it will
 262                         // return 403 instead of 401 for the current
 263                         // user check.  If it is a download operation
 264                         // and they have permission to read the
 265                         // keep_services table, we can allow it.
 266                         err = arv.Call("HEAD", "keep_services", "", "accessible", nil, nil)
 267                 }
 268         }
 269         if err != nil {
 270                 log.Printf("%s: CheckAuthorizationHeader error: %v", GetRemoteAddress(req), err)
 271                 return false, "", nil
 272         }
 273
 274         if userCurrentError == nil && user.IsAdmin {
 275                 // checking userCurrentError is probably redundant,
 276                 // IsAdmin would be false anyway. But can't hurt.
 277                 if op == "read" && !h.cluster.Collections.KeepproxyPermission.Admin.Download {
 278                         return false, "", nil
 279                 }
 280                 if op == "write" && !h.cluster.Collections.KeepproxyPermission.Admin.Upload {
 281                         return false, "", nil
 282                 }
 283         } else {
 284                 if op == "read" && !h.cluster.Collections.KeepproxyPermission.User.Download {
 285                         return false, "", nil
 286                 }
 287                 if op == "write" && !h.cluster.Collections.KeepproxyPermission.User.Upload {
 288                         return false, "", nil
 289                 }
 290         }
 291
 292         // Success!  Update cache
 293         h.APITokenCache.RememberToken(op+":"+tok, user)
 294
 295         return true, tok, user
 296 }
 297
 298 // We need to make a private copy of the default http transport early
 299 // in initialization, then make copies of our private copy later. It
 300 // won't be safe to copy http.DefaultTransport itself later, because
 301 // its private mutexes might have already been used. (Without this,
 302 // the test suite sometimes panics "concurrent map writes" in
 303 // net/http.(*Transport).removeIdleConnLocked().)
 304 var defaultTransport = *(http.DefaultTransport.(*http.Transport))
 305
 306 type proxyHandler struct {
 307         http.Handler
 308         *keepclient.KeepClient
 309         *APITokenCache
 310         timeout   time.Duration
 311         transport *http.Transport
 312         logger    log.FieldLogger
 313         cluster   *arvados.Cluster
 314 }
 315
 316 // MakeRESTRouter returns an http.Handler that passes GET and PUT
 317 // requests to the appropriate handlers.
 318 func MakeRESTRouter(kc *keepclient.KeepClient, timeout time.Duration, cluster *arvados.Cluster, logger log.FieldLogger) (http.Handler, error) {
 319         rest := mux.NewRouter()
 320
 321         transport := defaultTransport
 322         transport.DialContext = (&net.Dialer{
 323                 Timeout:   keepclient.DefaultConnectTimeout,
 324                 KeepAlive: keepclient.DefaultKeepAlive,
 325                 DualStack: true,
 326         }).DialContext
 327         transport.TLSClientConfig = arvadosclient.MakeTLSConfig(kc.Arvados.ApiInsecure)
 328         transport.TLSHandshakeTimeout = keepclient.DefaultTLSHandshakeTimeout
 329
 330         cacheQ, err := lru.New2Q(500)
 331         if err != nil {
 332                 return nil, fmt.Errorf("Error from lru.New2Q: %v", err)
 333         }
 334
 335         h := &proxyHandler{
 336                 Handler:    rest,
 337                 KeepClient: kc,
 338                 timeout:    timeout,
 339                 transport:  &transport,
 340                 APITokenCache: &APITokenCache{
 341                         tokens:     cacheQ,
 342                         expireTime: 300,
 343                 },
 344                 logger:  logger,
 345                 cluster: cluster,
 346         }
 347
 348         rest.HandleFunc(`/{locator:[0-9a-f]{32}\+.*}`, h.Get).Methods("GET", "HEAD")
 349         rest.HandleFunc(`/{locator:[0-9a-f]{32}}`, h.Get).Methods("GET", "HEAD")
 350
 351         // List all blocks
 352         rest.HandleFunc(`/index`, h.Index).Methods("GET")
 353
 354         // List blocks whose hash has the given prefix
 355         rest.HandleFunc(`/index/{prefix:[0-9a-f]{0,32}}`, h.Index).Methods("GET")
 356
 357         rest.HandleFunc(`/{locator:[0-9a-f]{32}\+.*}`, h.Put).Methods("PUT")
 358         rest.HandleFunc(`/{locator:[0-9a-f]{32}}`, h.Put).Methods("PUT")
 359         rest.HandleFunc(`/`, h.Put).Methods("POST")
 360         rest.HandleFunc(`/{any}`, h.Options).Methods("OPTIONS")
 361         rest.HandleFunc(`/`, h.Options).Methods("OPTIONS")
 362
 363         rest.Handle("/_health/{check}", &health.Handler{
 364                 Token:  cluster.ManagementToken,
 365                 Prefix: "/_health/",
 366         }).Methods("GET")
 367
 368         rest.NotFoundHandler = InvalidPathHandler{}
 369         return h, nil
 370 }
 371
 372 var errLoopDetected = errors.New("loop detected")
 373
 374 func (h *proxyHandler) checkLoop(resp http.ResponseWriter, req *http.Request) error {
 375         if via := req.Header.Get("Via"); strings.Index(via, " "+viaAlias) >= 0 {
 376                 h.logger.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)
 377                 http.Error(resp, errLoopDetected.Error(), http.StatusInternalServerError)
 378                 return errLoopDetected
 379         }
 380         return nil
 381 }
 382
 383 func SetCorsHeaders(resp http.ResponseWriter) {
 384         resp.Header().Set("Access-Control-Allow-Methods", "GET, HEAD, POST, PUT, OPTIONS")
 385         resp.Header().Set("Access-Control-Allow-Origin", "*")
 386         resp.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Length, Content-Type, X-Keep-Desired-Replicas")
 387         resp.Header().Set("Access-Control-Max-Age", "86486400")
 388 }
 389
 390 type InvalidPathHandler struct{}
 391
 392 func (InvalidPathHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 393         log.Printf("%s: %s %s unroutable", GetRemoteAddress(req), req.Method, req.URL.Path)
 394         http.Error(resp, "Bad request", http.StatusBadRequest)
 395 }
 396
 397 func (h *proxyHandler) Options(resp http.ResponseWriter, req *http.Request) {
 398         log.Printf("%s: %s %s", GetRemoteAddress(req), req.Method, req.URL.Path)
 399         SetCorsHeaders(resp)
 400 }
 401
 402 var errBadAuthorizationHeader = errors.New("Missing or invalid Authorization header, or method not allowed")
 403 var errContentLengthMismatch = errors.New("Actual length != expected content length")
 404 var errMethodNotSupported = errors.New("Method not supported")
 405
 406 var removeHint, _ = regexp.Compile("\\+K@[a-z0-9]{5}(\\+|$)")
 407
 408 func (h *proxyHandler) Get(resp http.ResponseWriter, req *http.Request) {
 409         if err := h.checkLoop(resp, req); err != nil {
 410                 return
 411         }
 412         SetCorsHeaders(resp)
 413         resp.Header().Set("Via", req.Proto+" "+viaAlias)
 414
 415         locator := mux.Vars(req)["locator"]
 416         var err error
 417         var status int
 418         var expectLength, responseLength int64
 419         var proxiedURI = "-"
 420
 421         defer func() {
 422                 log.Println(GetRemoteAddress(req), req.Method, req.URL.Path, status, expectLength, responseLength, proxiedURI, err)
 423                 if status != http.StatusOK {
 424                         http.Error(resp, err.Error(), status)
 425                 }
 426         }()
 427
 428         kc := h.makeKeepClient(req)
 429
 430         var pass bool
 431         var tok string
 432         var user *arvados.User
 433         if pass, tok, user = h.CheckAuthorizationHeader(req); !pass {
 434                 status, err = http.StatusForbidden, errBadAuthorizationHeader
 435                 return
 436         }
 437
 438         // Copy ArvadosClient struct and use the client's API token
 439         arvclient := *kc.Arvados
 440         arvclient.ApiToken = tok
 441         kc.Arvados = &arvclient
 442
 443         var reader io.ReadCloser
 444
 445         locator = removeHint.ReplaceAllString(locator, "$1")
 446
 447         if locator != "" {
 448                 parts := strings.SplitN(locator, "+", 3)
 449                 if len(parts) >= 2 {
 450                         logger := h.logger
 451                         if user != nil {
 452                                 logger = logger.WithField("user_uuid", user.UUID).
 453                                         WithField("user_full_name", user.FullName)
 454                         }
 455                         logger.WithField("locator", fmt.Sprintf("%s+%s", parts[0], parts[1])).Infof("Block download")
 456                 }
 457         }
 458
 459         switch req.Method {
 460         case "HEAD":
 461                 expectLength, proxiedURI, err = kc.Ask(locator)
 462         case "GET":
 463                 reader, expectLength, proxiedURI, err = kc.Get(locator)
 464                 if reader != nil {
 465                         defer reader.Close()
 466                 }
 467         default:
 468                 status, err = http.StatusNotImplemented, errMethodNotSupported
 469                 return
 470         }
 471
 472         if expectLength == -1 {
 473                 log.Println("Warning:", GetRemoteAddress(req), req.Method, proxiedURI, "Content-Length not provided")
 474         }
 475
 476         switch respErr := err.(type) {
 477         case nil:
 478                 status = http.StatusOK
 479                 resp.Header().Set("Content-Length", fmt.Sprint(expectLength))
 480                 switch req.Method {
 481                 case "HEAD":
 482                         responseLength = 0
 483                 case "GET":
 484                         responseLength, err = io.Copy(resp, reader)
 485                         if err == nil && expectLength > -1 && responseLength != expectLength {
 486                                 err = errContentLengthMismatch
 487                         }
 488                 }
 489         case keepclient.Error:
 490                 if respErr == keepclient.BlockNotFound {
 491                         status = http.StatusNotFound
 492                 } else if respErr.Temporary() {
 493                         status = http.StatusBadGateway
 494                 } else {
 495                         status = 422
 496                 }
 497         default:
 498                 status = http.StatusInternalServerError
 499         }
 500 }
 501
 502 var errLengthRequired = errors.New(http.StatusText(http.StatusLengthRequired))
 503 var errLengthMismatch = errors.New("Locator size hint does not match Content-Length header")
 504
 505 func (h *proxyHandler) Put(resp http.ResponseWriter, req *http.Request) {
 506         if err := h.checkLoop(resp, req); err != nil {
 507                 return
 508         }
 509         SetCorsHeaders(resp)
 510         resp.Header().Set("Via", "HTTP/1.1 "+viaAlias)
 511
 512         kc := h.makeKeepClient(req)
 513
 514         var err error
 515         var expectLength int64
 516         var status = http.StatusInternalServerError
 517         var wroteReplicas int
 518         var locatorOut string = "-"
 519
 520         defer func() {
 521                 log.Println(GetRemoteAddress(req), req.Method, req.URL.Path, status, expectLength, kc.Want_replicas, wroteReplicas, locatorOut, err)
 522                 if status != http.StatusOK {
 523                         http.Error(resp, err.Error(), status)
 524                 }
 525         }()
 526
 527         locatorIn := mux.Vars(req)["locator"]
 528
 529         // Check if the client specified storage classes
 530         if req.Header.Get("X-Keep-Storage-Classes") != "" {
 531                 var scl []string
 532                 for _, sc := range strings.Split(req.Header.Get("X-Keep-Storage-Classes"), ",") {
 533                         scl = append(scl, strings.Trim(sc, " "))
 534                 }
 535                 kc.SetStorageClasses(scl)
 536         }
 537
 538         _, err = fmt.Sscanf(req.Header.Get("Content-Length"), "%d", &expectLength)
 539         if err != nil || expectLength < 0 {
 540                 err = errLengthRequired
 541                 status = http.StatusLengthRequired
 542                 return
 543         }
 544
 545         if locatorIn != "" {
 546                 var loc *keepclient.Locator
 547                 if loc, err = keepclient.MakeLocator(locatorIn); err != nil {
 548                         status = http.StatusBadRequest
 549                         return
 550                 } else if loc.Size > 0 && int64(loc.Size) != expectLength {
 551                         err = errLengthMismatch
 552                         status = http.StatusBadRequest
 553                         return
 554                 }
 555         }
 556
 557         var pass bool
 558         var tok string
 559         var user *arvados.User
 560         if pass, tok, user = h.CheckAuthorizationHeader(req); !pass {
 561                 err = errBadAuthorizationHeader
 562                 status = http.StatusForbidden
 563                 return
 564         }
 565
 566         // Copy ArvadosClient struct and use the client's API token
 567         arvclient := *kc.Arvados
 568         arvclient.ApiToken = tok
 569         kc.Arvados = &arvclient
 570
 571         // Check if the client specified the number of replicas
 572         if desiredReplicas := req.Header.Get(keepclient.XKeepDesiredReplicas); desiredReplicas != "" {
 573                 var r int
 574                 _, err := fmt.Sscanf(desiredReplicas, "%d", &r)
 575                 if err == nil {
 576                         kc.Want_replicas = r
 577                 }
 578         }
 579
 580         // Now try to put the block through
 581         if locatorIn == "" {
 582                 bytes, err2 := ioutil.ReadAll(req.Body)
 583                 if err2 != nil {
 584                         err = fmt.Errorf("Error reading request body: %s", err2)
 585                         status = http.StatusInternalServerError
 586                         return
 587                 }
 588                 locatorOut, wroteReplicas, err = kc.PutB(bytes)
 589         } else {
 590                 locatorOut, wroteReplicas, err = kc.PutHR(locatorIn, req.Body, expectLength)
 591         }
 592
 593         if locatorOut != "" {
 594                 parts := strings.SplitN(locatorOut, "+", 3)
 595                 if len(parts) >= 2 {
 596                         logger := h.logger
 597                         if user != nil {
 598                                 logger = logger.WithField("user_uuid", user.UUID).
 599                                         WithField("user_full_name", user.FullName)
 600                         }
 601                         logger.WithField("locator", fmt.Sprintf("%s+%s", parts[0], parts[1])).Infof("Block upload")
 602                 }
 603         }
 604
 605         // Tell the client how many successful PUTs we accomplished
 606         resp.Header().Set(keepclient.XKeepReplicasStored, fmt.Sprintf("%d", wroteReplicas))
 607
 608         switch err.(type) {
 609         case nil:
 610                 status = http.StatusOK
 611                 if len(kc.StorageClasses) > 0 {
 612                         // A successful PUT request with storage classes means that all
 613                         // storage classes were fulfilled, so the client will get a
 614                         // confirmation via the X-Storage-Classes-Confirmed header.
 615                         hdr := ""
 616                         isFirst := true
 617                         for _, sc := range kc.StorageClasses {
 618                                 if isFirst {
 619                                         hdr = fmt.Sprintf("%s=%d", sc, wroteReplicas)
 620                                         isFirst = false
 621                                 } else {
 622                                         hdr += fmt.Sprintf(", %s=%d", sc, wroteReplicas)
 623                                 }
 624                         }
 625                         resp.Header().Set(keepclient.XKeepStorageClassesConfirmed, hdr)
 626                 }
 627                 _, err = io.WriteString(resp, locatorOut)
 628         case keepclient.OversizeBlockError:
 629                 // Too much data
 630                 status = http.StatusRequestEntityTooLarge
 631         case keepclient.InsufficientReplicasError:
 632                 status = http.StatusServiceUnavailable
 633         default:
 634                 status = http.StatusBadGateway
 635         }
 636 }
 637
 638 // ServeHTTP implementation for IndexHandler
 639 // Supports only GET requests for /index/{prefix:[0-9a-f]{0,32}}
 640 // For each keep server found in LocalRoots:
 641 //   Invokes GetIndex using keepclient
 642 //   Expects "complete" response (terminating with blank new line)
 643 //   Aborts on any errors
 644 // Concatenates responses from all those keep servers and returns
 645 func (h *proxyHandler) Index(resp http.ResponseWriter, req *http.Request) {
 646         SetCorsHeaders(resp)
 647
 648         prefix := mux.Vars(req)["prefix"]
 649         var err error
 650         var status int
 651
 652         defer func() {
 653                 if status != http.StatusOK {
 654                         http.Error(resp, err.Error(), status)
 655                 }
 656         }()
 657
 658         kc := h.makeKeepClient(req)
 659         ok, token, _ := h.CheckAuthorizationHeader(req)
 660         if !ok {
 661                 status, err = http.StatusForbidden, errBadAuthorizationHeader
 662                 return
 663         }
 664
 665         // Copy ArvadosClient struct and use the client's API token
 666         arvclient := *kc.Arvados
 667         arvclient.ApiToken = token
 668         kc.Arvados = &arvclient
 669
 670         // Only GET method is supported
 671         if req.Method != "GET" {
 672                 status, err = http.StatusNotImplemented, errMethodNotSupported
 673                 return
 674         }
 675
 676         // Get index from all LocalRoots and write to resp
 677         var reader io.Reader
 678         for uuid := range kc.LocalRoots() {
 679                 reader, err = kc.GetIndex(uuid, prefix)
 680                 if err != nil {
 681                         status = http.StatusBadGateway
 682                         return
 683                 }
 684
 685                 _, err = io.Copy(resp, reader)
 686                 if err != nil {
 687                         status = http.StatusBadGateway
 688                         return
 689                 }
 690         }
 691
 692         // Got index from all the keep servers and wrote to resp
 693         status = http.StatusOK
 694         resp.Write([]byte("\n"))
 695 }
 696
 697 func (h *proxyHandler) makeKeepClient(req *http.Request) *keepclient.KeepClient {
 698         kc := *h.KeepClient
 699         kc.RequestID = req.Header.Get("X-Request-Id")
 700         kc.HTTPClient = &proxyClient{
 701                 client: &http.Client{
 702                         Timeout:   h.timeout,
 703                         Transport: h.transport,
 704                 },
 705                 proto: req.Proto,
 706         }
 707         return &kc
 708 }