"sync"
"time"
- "git.curoverse.com/arvados.git/sdk/go/arvados"
- "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
- "git.curoverse.com/arvados.git/sdk/go/auth"
- "git.curoverse.com/arvados.git/sdk/go/keepclient"
+ "git.arvados.org/arvados.git/sdk/go/arvados"
+ "git.arvados.org/arvados.git/sdk/go/arvadosclient"
+ "git.arvados.org/arvados.git/sdk/go/auth"
+ "git.arvados.org/arvados.git/sdk/go/keepclient"
)
type remoteProxy struct {
mtx sync.Mutex
}
-func (rp *remoteProxy) Get(ctx context.Context, w http.ResponseWriter, r *http.Request, cluster *arvados.Cluster) {
+func (rp *remoteProxy) Get(ctx context.Context, w http.ResponseWriter, r *http.Request, cluster *arvados.Cluster, volmgr *RRVolumeManager) {
+ // Intervening proxies must not return a cached GET response
+ // to a prior request if a X-Keep-Signature request header has
+ // been added or changed.
+ w.Header().Add("Vary", "X-Keep-Signature")
+
token := GetAPIToken(r)
if token == "" {
http.Error(w, "no token provided in Authorization header", http.StatusUnauthorized)
return
}
- if sign := r.Header.Get("X-Keep-Signature"); sign != "" {
+ if strings.SplitN(r.Header.Get("X-Keep-Signature"), ",", 2)[0] == "local" {
buf, err := getBufferWithContext(ctx, bufs, BlockSize)
if err != nil {
http.Error(w, err.Error(), http.StatusServiceUnavailable)
Token: token,
Buffer: buf[:0],
ResponseWriter: w,
+ Context: ctx,
+ Cluster: cluster,
+ VolumeManager: volmgr,
}
- defer rrc.Flush(ctx)
+ defer rrc.Close()
w = rrc
}
var remoteClient *keepclient.KeepClient
remoteID := part[1:6]
remote, ok := cluster.RemoteClusters[remoteID]
if !ok {
- http.Error(w, "remote cluster not configured", http.StatusBadGateway)
+ http.Error(w, "remote cluster not configured", http.StatusBadRequest)
return
}
kc, err := rp.remoteClient(remoteID, remote, token)
// local volume, adds a response header with a locally-signed locator,
// and finally writes the data through.
type remoteResponseCacher struct {
- Locator string
- Token string
- Buffer []byte
+ Locator string
+ Token string
+ Buffer []byte
+ Context context.Context
+ Cluster *arvados.Cluster
+ VolumeManager *RRVolumeManager
http.ResponseWriter
statusCode int
}
rrc.statusCode = statusCode
}
-func (rrc *remoteResponseCacher) Flush(ctx context.Context) {
+func (rrc *remoteResponseCacher) Close() error {
if rrc.statusCode == 0 {
rrc.statusCode = http.StatusOK
} else if rrc.statusCode != http.StatusOK {
rrc.ResponseWriter.WriteHeader(rrc.statusCode)
rrc.ResponseWriter.Write(rrc.Buffer)
- return
+ return nil
+ }
+ _, err := PutBlock(rrc.Context, rrc.VolumeManager, rrc.Buffer, rrc.Locator[:32], nil)
+ if rrc.Context.Err() != nil {
+ // If caller hung up, log that instead of subsequent/misleading errors.
+ http.Error(rrc.ResponseWriter, rrc.Context.Err().Error(), http.StatusGatewayTimeout)
+ return err
}
- _, err := PutBlock(ctx, rrc.Buffer, rrc.Locator[:32])
if err == RequestHashError {
http.Error(rrc.ResponseWriter, "checksum mismatch in remote response", http.StatusBadGateway)
- return
+ return err
}
if err, ok := err.(*KeepError); ok {
http.Error(rrc.ResponseWriter, err.Error(), err.HTTPCode)
- return
+ return err
}
if err != nil {
http.Error(rrc.ResponseWriter, err.Error(), http.StatusBadGateway)
- return
+ return err
}
unsigned := localOrRemoteSignature.ReplaceAllLiteralString(rrc.Locator, "")
- signed := SignLocator(unsigned, rrc.Token, time.Now().Add(theConfig.BlobSignatureTTL.Duration()))
+ expiry := time.Now().Add(rrc.Cluster.Collections.BlobSigningTTL.Duration())
+ signed := SignLocator(rrc.Cluster, unsigned, rrc.Token, expiry)
if signed == unsigned {
- http.Error(rrc.ResponseWriter, "could not sign locator", http.StatusInternalServerError)
- return
+ err = errors.New("could not sign locator")
+ http.Error(rrc.ResponseWriter, err.Error(), http.StatusInternalServerError)
+ return err
}
rrc.Header().Set("X-Keep-Locator", signed)
rrc.ResponseWriter.WriteHeader(rrc.statusCode)
- rrc.ResponseWriter.Write(rrc.Buffer)
+ _, err = rrc.ResponseWriter.Write(rrc.Buffer)
+ return err
}