Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pkg/api/controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -3549,7 +3549,7 @@ func (c *Controller) CopyObject(w http.ResponseWriter, r *http.Request, body api
}

// copy entry
entry, err := c.Catalog.CopyEntry(ctx, repository, srcRef, srcPath, repository, branch, destPath, false, nil, opts...)
entry, err := c.Catalog.CopyEntry(ctx, repository, srcRef, srcPath, repository, branch, destPath, false, nil, "", opts...)
if c.handleAPIError(ctx, w, r, err) {
return
}
Expand Down
12 changes: 7 additions & 5 deletions pkg/catalog/catalog.go
Original file line number Diff line number Diff line change
Expand Up @@ -3062,7 +3062,7 @@ func (c *Catalog) PrepareGCUncommitted(ctx context.Context, repositoryID string,
// not delete the underlying object.
// ErrCannotClone is returned if any of the clone conditions are not satisfied.
func (c *Catalog) cloneEntry(ctx context.Context, srcRepo *Repository, srcRef string, srcEntry *DBEntry, destRepository, destBranch, destPath string,
replaceSrcMetadata bool, metadata Metadata, opts ...graveler.SetOptionsFunc,
replaceSrcMetadata bool, metadata Metadata, contentType string, opts ...graveler.SetOptionsFunc,
) (*DBEntry, error) {
// validate clone conditions
if srcRepo.Name != destRepository {
Expand All @@ -3087,6 +3087,7 @@ func (c *Catalog) cloneEntry(ctx context.Context, srcRepo *Repository, srcRef st

if replaceSrcMetadata {
dstEntry.Metadata = metadata
dstEntry.ContentType = contentType
}

dstEntry.Metadata[apiutil.CloneMetadataKey] = srcMetaValue
Expand All @@ -3099,9 +3100,9 @@ func (c *Catalog) cloneEntry(ctx context.Context, srcRepo *Repository, srcRef st

// CopyEntry copy entry information by using the block adapter to make a copy of the data to a new physical address or clone the entry if possible.
// if clone is possible, the entry will be cloned with the same physical address and metadata.
// if replaceSrcMetadata is true, the metadata will be replaced with the provided metadata.
// if replaceSrcMetadata is false, the metadata will be copied from the source entry.
func (c *Catalog) CopyEntry(ctx context.Context, srcRepository, srcRef, srcPath, destRepository, destBranch, destPath string, replaceSrcMetadata bool, metadata Metadata, opts ...graveler.SetOptionsFunc) (*DBEntry, error) {
// if replaceSrcMetadata is true, the metadata and content type will be replaced with the provided metadata and content type.
// if replaceSrcMetadata is false, the metadata and content type will be copied from the source entry.
func (c *Catalog) CopyEntry(ctx context.Context, srcRepository, srcRef, srcPath, destRepository, destBranch, destPath string, replaceSrcMetadata bool, metadata Metadata, contentType string, opts ...graveler.SetOptionsFunc) (*DBEntry, error) {
// copyObjectFull copy data from srcEntry's physical address (if set) or srcPath into destPath
// fetch src entry if needed - optimization in case we already have the entry
srcEntry, err := c.GetEntry(ctx, srcRepository, srcRef, srcPath, GetEntryParams{})
Expand Down Expand Up @@ -3131,7 +3132,7 @@ func (c *Catalog) CopyEntry(ctx context.Context, srcRepository, srcRef, srcPath,
options := graveler.NewSetOptions(opts)
// Clone entry
if options.Shallow {
return c.cloneEntry(ctx, srcRepo, srcRef, srcEntry, destRepository, destBranch, destPath, replaceSrcMetadata, metadata, opts...)
return c.cloneEntry(ctx, srcRepo, srcRef, srcEntry, destRepository, destBranch, destPath, replaceSrcMetadata, metadata, contentType, opts...)
}

// copy data to a new physical address
Expand All @@ -3142,6 +3143,7 @@ func (c *Catalog) CopyEntry(ctx context.Context, srcRepository, srcRef, srcPath,

if replaceSrcMetadata {
dstEntry.Metadata = metadata
dstEntry.ContentType = contentType
}

srcObject := block.ObjectPointer{
Expand Down
58 changes: 58 additions & 0 deletions pkg/gateway/handler_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"

Expand Down Expand Up @@ -195,3 +196,60 @@ func TestContextCancellation(t *testing.T) {
// Verify that the status code is 499 (Client Closed Request)
assert.Equal(t, httputil.HttpStatusClientClosedRequest, result.StatusCode, "Expected status code 499 for cancelled context")
}

func TestCopyObjectReplaceContentType(t *testing.T) {
t.Setenv("USE_STORAGE_NAMESPACE", "mem://"+repoName)
h, _ := testutil.GetBasicHandler(t, &testutil.FakeAuthService{
BareDomain: "example.com",
AccessKeyID: esti.DefaultAdminAccessKeyID,
SecretAccessKey: esti.DefaultAdminSecretAccessKey,
UserID: "65867",
Region: "MockRegion",
}, repoName)
const srcPath = "/" + repoName + "/main/data.json"

serve := func(t *testing.T, req *http.Request) *http.Response {
t.Helper()
creds := aws.Credentials{
AccessKeyID: esti.DefaultAdminAccessKeyID,
SecretAccessKey: esti.DefaultAdminSecretAccessKey,
}
req.Host = "host.domain.com"
req.Header.Set("X-Amz-Content-Sha256", "UNSIGNED-PAYLOAD")
err := v4.NewSigner().SignHTTP(t.Context(), creds, req, "UNSIGNED-PAYLOAD", "s3", "us-east-1", time.Now())
require.NoError(t, err)
rr := httptest.NewRecorder()
h.ServeHTTP(rr, req)
return rr.Result()
}

putReq := httptest.NewRequest(http.MethodPut, srcPath, strings.NewReader("{}"))
putReq.Header.Set("Content-Length", "2")
putReq.Header.Set("Content-Type", "text/plain")
putReq.Header.Set("X-Amz-Meta-K", "v")
require.Equal(t, http.StatusOK, serve(t, putReq).StatusCode)

testCases := []struct {
name string
directive string
expectedContentType string
}{
{name: "replace", directive: "REPLACE", expectedContentType: "application/json"},
{name: "copy", directive: "COPY", expectedContentType: "text/plain"},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
destPath := "/" + repoName + "/main/copy-" + tc.name + ".json"
copyReq := httptest.NewRequest(http.MethodPut, destPath, nil)
copyReq.Header.Set("X-Amz-Copy-Source", srcPath)
copyReq.Header.Set("X-Amz-Metadata-Directive", tc.directive)
copyReq.Header.Set("Content-Type", "application/json")
copyReq.Header.Set("X-Amz-Meta-K", "w")
require.Equal(t, http.StatusOK, serve(t, copyReq).StatusCode)

headResult := serve(t, httptest.NewRequest(http.MethodHead, destPath, nil))
require.Equal(t, http.StatusOK, headResult.StatusCode)
require.Equal(t, tc.expectedContentType, headResult.Header.Get("Content-Type"))
})
}
}
2 changes: 1 addition & 1 deletion pkg/gateway/operations/putobject.go
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,7 @@ func handleCopy(w http.ResponseWriter, req *http.Request, o *PathOperation, copy
}
replaceMetadata := shouldReplaceMetadata(req)

entry, err := o.Catalog.CopyEntry(ctx, srcPath.Repo, srcPath.Reference, srcPath.Path, repository, branch, o.Path, replaceMetadata, metadata)
entry, err := o.Catalog.CopyEntry(ctx, srcPath.Repo, srcPath.Reference, srcPath.Path, repository, branch, o.Path, replaceMetadata, metadata, req.Header.Get("Content-Type"))
if err != nil {
o.Log(req).WithError(err).Error("could create a copy")
if errors.Is(err, block.ErrDataNotFound) || errors.Is(err, graveler.ErrNotFound) {
Expand Down
10 changes: 7 additions & 3 deletions pkg/gateway/testutil/gateway_setup.go
Original file line number Diff line number Diff line change
Expand Up @@ -44,9 +44,6 @@ func GetBasicHandler(t *testing.T, authService *FakeAuthService, repoName string

multipartTracker := multipart.NewTracker(store)

blockstoreType, _ := os.LookupEnv(testutil.EnvKeyUseBlockAdapter)
blockAdapter := testutil.NewBlockAdapterByType(t, blockstoreType)

cfg := &config.ConfigImpl{}
_, err = config.NewConfig("", cfg)
testutil.MustDo(t, "config", err)
Expand All @@ -62,6 +59,13 @@ func GetBasicHandler(t *testing.T, authService *FakeAuthService, repoName string
_ = c.Close()
})

// The in-memory adapter keeps its data per instance, so share the catalog's adapter with the
// handler to let catalog operations (such as copy) read objects written through the gateway.
blockAdapter := c.BlockAdapter
if blockstoreType, _ := os.LookupEnv(testutil.EnvKeyUseBlockAdapter); blockstoreType != "" && blockstoreType != block.BlockstoreTypeMem {
blockAdapter = testutil.NewBlockAdapterByType(t, blockstoreType)
}

storageNamespace := os.Getenv("USE_STORAGE_NAMESPACE")
if storageNamespace == "" {
storageNamespace = "replay"
Expand Down
Loading