services/keepstore/router.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package keepstore
   6
   7 import (
   8         "encoding/json"
   9         "errors"
  10         "fmt"
  11         "io"
  12         "net/http"
  13         "os"
  14         "strconv"
  15         "strings"
  16         "sync/atomic"
  17
  18         "git.arvados.org/arvados.git/lib/service"
  19         "git.arvados.org/arvados.git/sdk/go/arvados"
  20         "git.arvados.org/arvados.git/sdk/go/auth"
  21         "git.arvados.org/arvados.git/sdk/go/httpserver"
  22         "git.arvados.org/arvados.git/sdk/go/keepclient"
  23         "github.com/gorilla/mux"
  24 )
  25
  26 type router struct {
  27         http.Handler
  28         keepstore *keepstore
  29         puller    *puller
  30         trasher   *trasher
  31 }
  32
  33 func newRouter(keepstore *keepstore, puller *puller, trasher *trasher) service.Handler {
  34         rtr := &router{
  35                 keepstore: keepstore,
  36                 puller:    puller,
  37                 trasher:   trasher,
  38         }
  39         adminonly := func(h http.HandlerFunc) http.HandlerFunc {
  40                 return auth.RequireLiteralToken(keepstore.cluster.SystemRootToken, h).ServeHTTP
  41         }
  42
  43         r := mux.NewRouter()
  44         // Without SkipClean(true), the gorilla/mux package responds
  45         // to "PUT //foo" with a 301 redirect to "/foo", which causes
  46         // a (Fetch Standard compliant) client to repeat the request
  47         // as "GET /foo", which is clearly inappropriate in this case.
  48         // It's less confusing if we just return 400.
  49         r.SkipClean(true)
  50         locatorPath := `/{locator:[0-9a-f]{32}.*}`
  51         get := r.Methods(http.MethodGet, http.MethodHead).Subrouter()
  52         get.HandleFunc(locatorPath, rtr.handleBlockRead)
  53         get.HandleFunc("/"+locatorPath, rtr.handleBlockRead) // for compatibility -- see TestBlockRead_DoubleSlash
  54         get.HandleFunc(`/index`, adminonly(rtr.handleIndex))
  55         get.HandleFunc(`/index/{prefix:[0-9a-f]{0,32}}`, adminonly(rtr.handleIndex))
  56         get.HandleFunc(`/mounts`, adminonly(rtr.handleMounts))
  57         get.HandleFunc(`/mounts/{uuid}/blocks`, adminonly(rtr.handleIndex))
  58         get.HandleFunc(`/mounts/{uuid}/blocks/{prefix:[0-9a-f]{0,32}}`, adminonly(rtr.handleIndex))
  59         put := r.Methods(http.MethodPut).Subrouter()
  60         put.HandleFunc(locatorPath, rtr.handleBlockWrite)
  61         put.HandleFunc(`/pull`, adminonly(rtr.handlePullList))
  62         put.HandleFunc(`/trash`, adminonly(rtr.handleTrashList))
  63         put.HandleFunc(`/untrash`+locatorPath, adminonly(rtr.handleUntrash))
  64         touch := r.Methods("TOUCH").Subrouter()
  65         touch.HandleFunc(locatorPath, adminonly(rtr.handleBlockTouch))
  66         delete := r.Methods(http.MethodDelete).Subrouter()
  67         delete.HandleFunc(locatorPath, adminonly(rtr.handleBlockTrash))
  68         options := r.Methods(http.MethodOptions).Subrouter()
  69         options.NewRoute().PathPrefix(`/`).HandlerFunc(rtr.handleOptions)
  70         r.NotFoundHandler = http.HandlerFunc(rtr.handleBadRequest)
  71         r.MethodNotAllowedHandler = http.HandlerFunc(rtr.handleBadRequest)
  72         rtr.Handler = corsHandler(auth.LoadToken(r))
  73         return rtr
  74 }
  75
  76 func (rtr *router) CheckHealth() error {
  77         return nil
  78 }
  79
  80 func (rtr *router) Done() <-chan struct{} {
  81         return nil
  82 }
  83
  84 func (rtr *router) handleBlockRead(w http.ResponseWriter, req *http.Request) {
  85         // Intervening proxies must not return a cached GET response
  86         // to a prior request if a X-Keep-Signature request header has
  87         // been added or changed.
  88         w.Header().Add("Vary", keepclient.XKeepSignature)
  89         var localLocator func(string)
  90         if strings.SplitN(req.Header.Get(keepclient.XKeepSignature), ",", 2)[0] == "local" {
  91                 localLocator = func(locator string) {
  92                         w.Header().Set(keepclient.XKeepLocator, locator)
  93                 }
  94         }
  95         out := w
  96         if req.Method == http.MethodHead {
  97                 out = discardWrite{ResponseWriter: w}
  98         } else if li, err := getLocatorInfo(mux.Vars(req)["locator"]); err != nil {
  99                 rtr.handleError(w, req, err)
 100                 return
 101         } else if li.size == 0 && li.hash != "d41d8cd98f00b204e9800998ecf8427e" {
 102                 // GET {hash} (with no size hint) is not allowed
 103                 // because we can't report md5 mismatches.
 104                 rtr.handleError(w, req, errMethodNotAllowed)
 105                 return
 106         }
 107         n, err := rtr.keepstore.BlockRead(req.Context(), arvados.BlockReadOptions{
 108                 Locator:      mux.Vars(req)["locator"],
 109                 WriteTo:      out,
 110                 LocalLocator: localLocator,
 111         })
 112         if err != nil && (n == 0 || req.Method == http.MethodHead) {
 113                 rtr.handleError(w, req, err)
 114                 return
 115         }
 116 }
 117
 118 func (rtr *router) handleBlockWrite(w http.ResponseWriter, req *http.Request) {
 119         dataSize, _ := strconv.Atoi(req.Header.Get("Content-Length"))
 120         replicas, _ := strconv.Atoi(req.Header.Get(keepclient.XKeepDesiredReplicas))
 121         resp, err := rtr.keepstore.BlockWrite(req.Context(), arvados.BlockWriteOptions{
 122                 Hash:           mux.Vars(req)["locator"],
 123                 Reader:         req.Body,
 124                 DataSize:       dataSize,
 125                 RequestID:      req.Header.Get("X-Request-Id"),
 126                 StorageClasses: trimSplit(req.Header.Get(keepclient.XKeepStorageClasses), ","),
 127                 Replicas:       replicas,
 128         })
 129         if err != nil {
 130                 rtr.handleError(w, req, err)
 131                 return
 132         }
 133         w.Header().Set(keepclient.XKeepReplicasStored, fmt.Sprintf("%d", resp.Replicas))
 134         scc := ""
 135         for k, n := range resp.StorageClasses {
 136                 if n > 0 {
 137                         if scc != "" {
 138                                 scc += ", "
 139                         }
 140                         scc += fmt.Sprintf("%s=%d", k, n)
 141                 }
 142         }
 143         w.Header().Set(keepclient.XKeepStorageClassesConfirmed, scc)
 144         w.WriteHeader(http.StatusOK)
 145         fmt.Fprintln(w, resp.Locator)
 146 }
 147
 148 func (rtr *router) handleBlockTouch(w http.ResponseWriter, req *http.Request) {
 149         err := rtr.keepstore.BlockTouch(req.Context(), mux.Vars(req)["locator"])
 150         rtr.handleError(w, req, err)
 151 }
 152
 153 func (rtr *router) handleBlockTrash(w http.ResponseWriter, req *http.Request) {
 154         err := rtr.keepstore.BlockTrash(req.Context(), mux.Vars(req)["locator"])
 155         rtr.handleError(w, req, err)
 156 }
 157
 158 func (rtr *router) handleMounts(w http.ResponseWriter, req *http.Request) {
 159         json.NewEncoder(w).Encode(rtr.keepstore.Mounts())
 160 }
 161
 162 func (rtr *router) handleIndex(w http.ResponseWriter, req *http.Request) {
 163         httpserver.ExemptFromDeadline(req)
 164         prefix := req.FormValue("prefix")
 165         if prefix == "" {
 166                 prefix = mux.Vars(req)["prefix"]
 167         }
 168         cw := &countingWriter{writer: w}
 169         err := rtr.keepstore.Index(req.Context(), indexOptions{
 170                 MountUUID: mux.Vars(req)["uuid"],
 171                 Prefix:    prefix,
 172                 WriteTo:   cw,
 173         })
 174         if err != nil && cw.n.Load() == 0 {
 175                 // Nothing was written, so it's not too late to report
 176                 // an error via http response header. (Otherwise, all
 177                 // we can do is omit the trailing newline below to
 178                 // indicate something went wrong.)
 179                 rtr.handleError(w, req, err)
 180                 return
 181         }
 182         if err == nil {
 183                 // A trailing blank line signals to the caller that
 184                 // the response is complete.
 185                 w.Write([]byte("\n"))
 186         }
 187 }
 188
 189 func (rtr *router) handlePullList(w http.ResponseWriter, req *http.Request) {
 190         var pl []PullListItem
 191         err := json.NewDecoder(req.Body).Decode(&pl)
 192         if err != nil {
 193                 rtr.handleError(w, req, err)
 194                 return
 195         }
 196         req.Body.Close()
 197         if len(pl) > 0 && len(pl[0].Locator) == 32 {
 198                 rtr.handleError(w, req, httpserver.ErrorWithStatus(errors.New("rejecting pull list containing a locator without a size hint -- this probably means keep-balance needs to be upgraded"), http.StatusBadRequest))
 199                 return
 200         }
 201         rtr.puller.SetPullList(pl)
 202 }
 203
 204 func (rtr *router) handleTrashList(w http.ResponseWriter, req *http.Request) {
 205         var tl []TrashListItem
 206         err := json.NewDecoder(req.Body).Decode(&tl)
 207         if err != nil {
 208                 rtr.handleError(w, req, err)
 209                 return
 210         }
 211         req.Body.Close()
 212         rtr.trasher.SetTrashList(tl)
 213 }
 214
 215 func (rtr *router) handleUntrash(w http.ResponseWriter, req *http.Request) {
 216         err := rtr.keepstore.BlockUntrash(req.Context(), mux.Vars(req)["locator"])
 217         rtr.handleError(w, req, err)
 218 }
 219
 220 func (rtr *router) handleBadRequest(w http.ResponseWriter, req *http.Request) {
 221         http.Error(w, "Bad Request", http.StatusBadRequest)
 222 }
 223
 224 func (rtr *router) handleOptions(w http.ResponseWriter, req *http.Request) {
 225 }
 226
 227 func (rtr *router) handleError(w http.ResponseWriter, req *http.Request, err error) {
 228         if req.Context().Err() != nil {
 229                 w.WriteHeader(499)
 230                 return
 231         }
 232         if err == nil {
 233                 return
 234         } else if os.IsNotExist(err) {
 235                 w.WriteHeader(http.StatusNotFound)
 236         } else if statusErr := interface{ HTTPStatus() int }(nil); errors.As(err, &statusErr) {
 237                 w.WriteHeader(statusErr.HTTPStatus())
 238         } else {
 239                 w.WriteHeader(http.StatusInternalServerError)
 240         }
 241         fmt.Fprintln(w, err.Error())
 242 }
 243
 244 type countingWriter struct {
 245         writer io.Writer
 246         n      atomic.Int64
 247 }
 248
 249 func (cw *countingWriter) Write(p []byte) (int, error) {
 250         n, err := cw.writer.Write(p)
 251         cw.n.Add(int64(n))
 252         return n, err
 253 }
 254
 255 // Split s by sep, trim whitespace from each part, and drop empty
 256 // parts.
 257 func trimSplit(s, sep string) []string {
 258         var r []string
 259         for _, part := range strings.Split(s, sep) {
 260                 part = strings.TrimSpace(part)
 261                 if part != "" {
 262                         r = append(r, part)
 263                 }
 264         }
 265         return r
 266 }
 267
 268 // setSizeOnWrite sets the Content-Length header to the given size on
 269 // first write.
 270 type setSizeOnWrite struct {
 271         http.ResponseWriter
 272         size  int
 273         wrote bool
 274 }
 275
 276 func (ss *setSizeOnWrite) Write(p []byte) (int, error) {
 277         if !ss.wrote {
 278                 ss.Header().Set("Content-Length", fmt.Sprintf("%d", ss.size))
 279                 ss.wrote = true
 280         }
 281         return ss.ResponseWriter.Write(p)
 282 }
 283
 284 func (ss *setSizeOnWrite) Unwrap() http.ResponseWriter {
 285         return ss.ResponseWriter
 286 }
 287
 288 type discardWrite struct {
 289         http.ResponseWriter
 290 }
 291
 292 func (discardWrite) Write(p []byte) (int, error) {
 293         return len(p), nil
 294 }
 295
 296 func (dw discardWrite) Unwrap() http.ResponseWriter {
 297         return dw.ResponseWriter
 298 }
 299
 300 func corsHandler(h http.Handler) http.Handler {
 301         return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 302                 SetCORSHeaders(w)
 303                 h.ServeHTTP(w, r)
 304         })
 305 }
 306
 307 var corsHeaders = map[string]string{
 308         "Access-Control-Allow-Methods":  "GET, HEAD, PUT, OPTIONS",
 309         "Access-Control-Allow-Origin":   "*",
 310         "Access-Control-Allow-Headers":  "Authorization, Content-Length, Content-Type, " + keepclient.XKeepDesiredReplicas + ", " + keepclient.XKeepSignature + ", " + keepclient.XKeepStorageClasses,
 311         "Access-Control-Expose-Headers": keepclient.XKeepLocator + ", " + keepclient.XKeepReplicasStored + ", " + keepclient.XKeepStorageClassesConfirmed,
 312         "Access-Control-Max-Age":        "86486400",
 313 }
 314
 315 func SetCORSHeaders(w http.ResponseWriter) {
 316         for k, v := range corsHeaders {
 317                 w.Header().Set(k, v)
 318         }
 319 }