20240: Change locking mode to EXCLUSIVE.

[arvados.git] / services / keep-web / s3.go

diff --git a/services/keep-web/s3.go b/services/keep-web/s3.go
index d92828e066eeb9b61c9b25e38be543989ca540a1..f98efd8fdfcdf39febe29e84610aea474ccda02d 100644 (file)
--- a/services/keep-web/s3.go
+++ b/services/keep-web/s3.go
@@ -14,6 +14,7 @@ import (
        "fmt"
        "hash"
        "io"
+       "mime"
        "net/http"
        "net/textproto"
        "net/url"
@@ -26,9 +27,7 @@ import (
        "time"
 
        "git.arvados.org/arvados.git/sdk/go/arvados"
-       "git.arvados.org/arvados.git/sdk/go/arvadosclient"
        "git.arvados.org/arvados.git/sdk/go/ctxlog"
-       "git.arvados.org/arvados.git/sdk/go/keepclient"
        "github.com/AdRoll/goamz/s3"
 )
 
@@ -311,33 +310,18 @@ func (h *handler) serveS3(w http.ResponseWriter, r *http.Request) bool {
                return false
        }
 
-       var err error
-       var fs arvados.CustomFileSystem
-       var arvclient *arvadosclient.ArvadosClient
-       if r.Method == http.MethodGet || r.Method == http.MethodHead {
-               // Use a single session (cached FileSystem) across
-               // multiple read requests.
-               var sess *cachedSession
-               fs, sess, err = h.Cache.GetSession(token)
-               if err != nil {
-                       s3ErrorResponse(w, InternalError, err.Error(), r.URL.Path, http.StatusInternalServerError)
-                       return true
-               }
-               arvclient = sess.arvadosclient
-       } else {
+       fs, sess, tokenUser, err := h.Cache.GetSession(token)
+       if err != nil {
+               s3ErrorResponse(w, InternalError, err.Error(), r.URL.Path, http.StatusInternalServerError)
+               return true
+       }
+       readfs := fs
+       if writeMethod[r.Method] {
                // Create a FileSystem for this request, to avoid
                // exposing incomplete write operations to concurrent
                // requests.
-               var kc *keepclient.KeepClient
-               var release func()
-               var client *arvados.Client
-               arvclient, kc, client, release, err = h.getClients(r.Header.Get("X-Request-Id"), token)
-               if err != nil {
-                       s3ErrorResponse(w, InternalError, err.Error(), r.URL.Path, http.StatusInternalServerError)
-                       return true
-               }
-               defer release()
-               fs = client.SiteFileSystem(kc)
+               client := sess.client.WithRequestID(r.Header.Get("X-Request-Id"))
+               fs = client.SiteFileSystem(sess.keepclient)
                fs.ForwardSlashNameSubstitution(h.Cluster.Collections.ForwardSlashNameSubstitution)
        }
 
@@ -417,12 +401,11 @@ func (h *handler) serveS3(w http.ResponseWriter, r *http.Request) bool {
                        return true
                }
 
-               tokenUser, err := h.Cache.GetTokenUser(token)
                if !h.userPermittedToUploadOrDownload(r.Method, tokenUser) {
                        http.Error(w, "Not permitted", http.StatusForbidden)
                        return true
                }
-               h.logUploadOrDownload(r, arvclient, fs, fspath, nil, tokenUser)
+               h.logUploadOrDownload(r, sess.arvadosclient, fs, fspath, nil, tokenUser)
 
                // shallow copy r, and change URL path
                r := *r
@@ -513,12 +496,11 @@ func (h *handler) serveS3(w http.ResponseWriter, r *http.Request) bool {
                        }
                        defer f.Close()
 
-                       tokenUser, err := h.Cache.GetTokenUser(token)
                        if !h.userPermittedToUploadOrDownload(r.Method, tokenUser) {
                                http.Error(w, "Not permitted", http.StatusForbidden)
                                return true
                        }
-                       h.logUploadOrDownload(r, arvclient, fs, fspath, nil, tokenUser)
+                       h.logUploadOrDownload(r, sess.arvadosclient, fs, fspath, nil, tokenUser)
 
                        _, err = io.Copy(f, r.Body)
                        if err != nil {
@@ -533,14 +515,12 @@ func (h *handler) serveS3(w http.ResponseWriter, r *http.Request) bool {
                                return true
                        }
                }
-               err = fs.Sync()
+               err = h.syncCollection(fs, readfs, fspath)
                if err != nil {
                        err = fmt.Errorf("sync failed: %w", err)
                        s3ErrorResponse(w, InternalError, err.Error(), r.URL.Path, http.StatusInternalServerError)
                        return true
                }
-               // Ensure a subsequent read operation will see the changes.
-               h.Cache.ResetSession(token)
                w.WriteHeader(http.StatusOK)
                return true
        case r.Method == http.MethodDelete:
@@ -587,14 +567,12 @@ func (h *handler) serveS3(w http.ResponseWriter, r *http.Request) bool {
                        s3ErrorResponse(w, InvalidArgument, err.Error(), r.URL.Path, http.StatusBadRequest)
                        return true
                }
-               err = fs.Sync()
+               err = h.syncCollection(fs, readfs, fspath)
                if err != nil {
                        err = fmt.Errorf("sync failed: %w", err)
                        s3ErrorResponse(w, InternalError, err.Error(), r.URL.Path, http.StatusInternalServerError)
                        return true
                }
-               // Ensure a subsequent read operation will see the changes.
-               h.Cache.ResetSession(token)
                w.WriteHeader(http.StatusNoContent)
                return true
        default:
@@ -603,7 +581,43 @@ func (h *handler) serveS3(w http.ResponseWriter, r *http.Request) bool {
        }
 }
 
+// Save modifications to the indicated collection in srcfs, then (if
+// successful) ensure they are also reflected in dstfs.
+func (h *handler) syncCollection(srcfs, dstfs arvados.CustomFileSystem, path string) error {
+       coll, _ := h.determineCollection(srcfs, path)
+       if coll == nil || coll.UUID == "" {
+               return errors.New("could not determine collection to sync")
+       }
+       d, err := srcfs.OpenFile("by_id/"+coll.UUID, os.O_RDWR, 0777)
+       if err != nil {
+               return err
+       }
+       defer d.Close()
+       err = d.Sync()
+       if err != nil {
+               return err
+       }
+       snap, err := d.Snapshot()
+       if err != nil {
+               return err
+       }
+       dstd, err := dstfs.OpenFile("by_id/"+coll.UUID, os.O_RDWR, 0777)
+       if err != nil {
+               return err
+       }
+       defer dstd.Close()
+       return dstd.Splice(snap)
+}
+
 func setFileInfoHeaders(header http.Header, fs arvados.CustomFileSystem, path string) error {
+       maybeEncode := func(s string) string {
+               for _, c := range s {
+                       if c > '\u007f' || c < ' ' {
+                               return mime.BEncoding.Encode("UTF-8", s)
+                       }
+               }
+               return s
+       }
        path = strings.TrimSuffix(path, "/")
        var props map[string]interface{}
        for {
@@ -636,9 +650,9 @@ func setFileInfoHeaders(header http.Header, fs arvados.CustomFileSystem, path st
                }
                k = "x-amz-meta-" + k
                if s, ok := v.(string); ok {
-                       header.Set(k, s)
+                       header.Set(k, maybeEncode(s))
                } else if j, err := json.Marshal(v); err == nil {
-                       header.Set(k, string(j))
+                       header.Set(k, maybeEncode(string(j)))
                }
        }
        return nil