diff --git a/.github/workflows/lakectl-compatibility-tests.yaml b/.github/workflows/lakectl-compatibility-tests.yaml index 3db33363a61..bfe19ffd2a5 100644 --- a/.github/workflows/lakectl-compatibility-tests.yaml +++ b/.github/workflows/lakectl-compatibility-tests.yaml @@ -125,6 +125,7 @@ jobs: env: AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }} AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }} + AWS_REGION: us-east-1 ESTI_AWS_ACCESS_KEY_ID: ${{ secrets.ESTI_AWS_ACCESS_KEY_ID }} ESTI_AWS_SECRET_ACCESS_KEY: ${{ secrets.ESTI_AWS_SECRET_ACCESS_KEY }} ESTI_BINARIES_DIR: ${{ github.workspace }}/bin diff --git a/esti/adapter_utils.go b/esti/adapter_utils.go index bcfaa4d84ed..1f150ef8646 100644 --- a/esti/adapter_utils.go +++ b/esti/adapter_utils.go @@ -35,6 +35,9 @@ type ClientAdapter interface { // Download returns a Reader to download data from physicalAddress. The Close method // of that Reader can return errors! Download(ctx context.Context, physicalAddress *url.URL) (io.ReadCloser, error) + + // Delete removes the object at physicalAddress from the backing store. + Delete(ctx context.Context, physicalAddress *url.URL) error } type AdapterFactory map[string]func(context.Context) (ClientAdapter, error) @@ -110,3 +113,16 @@ func (s *s3Adapter) Download(ctx context.Context, physicalAddress *url.URL) (io. } return getObjectResponse.Body, nil } + +func (s *s3Adapter) Delete(ctx context.Context, physicalAddress *url.URL) error { + if physicalAddress.Scheme != s3Scheme { + return fmt.Errorf("%s: %w", s3Scheme, helpers.ErrUnsupportedProtocol) + } + bucket := physicalAddress.Hostname() + key := strings.TrimPrefix(physicalAddress.Path, "/") + _, err := s.svc.DeleteObject(ctx, &s3.DeleteObjectInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + }) + return err +} diff --git a/esti/lakectl_test.go b/esti/lakectl_test.go index f47d02d8b8f..cce15b9e028 100644 --- a/esti/lakectl_test.go +++ b/esti/lakectl_test.go @@ -2,11 +2,13 @@ package esti import ( "fmt" + "net/url" "os" "strings" "testing" "time" + "github.com/go-openapi/swag" "github.com/spf13/viper" "github.com/stretchr/testify/require" "github.com/treeverse/lakefs/pkg/api/apigen" @@ -162,6 +164,62 @@ func TestLakectlPreSignUpload(t *testing.T) { }) } +func TestLakectlFsUploadLarge(t *testing.T) { + RequireBlockstoreType(t, block.BlockstoreTypeS3) + ctx := t.Context() + + repoName := GenerateUniqueRepositoryName() + storage := GenerateUniqueStorageNamespace(repoName) + vars := map[string]string{ + "REPO": repoName, + "STORAGE": storage, + "BRANCH": mainBranch, + } + RunCmdAndVerifySuccessWithFile(t, Lakectl()+" repo create lakefs://"+repoName+" "+storage, false, "lakectl_repo_create", vars) + + // S3 rejects single-part uploads above 5GiB, so one byte more forces + // the multipart path (#9284). A sparse file keeps disk usage low. + const largeFileSize = 5<<30 + 1 + f, err := os.CreateTemp("", "lakectl-large-upload-") + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, os.Remove(f.Name())) + }) + require.NoError(t, f.Truncate(largeFileSize)) + require.NoError(t, f.Close()) + + uploadAndVerify := func(t *testing.T, objPath, extraArgs string) { + t.Helper() + // deleting the repository does not remove its data, so remove the large object from the storage + t.Cleanup(func() { + statResp, err := client.StatObjectWithResponse(ctx, repoName, mainBranch, &apigen.StatObjectParams{Path: objPath}) + require.NoError(t, err) + if statResp.JSON200 == nil { + return + } + physicalAddress, err := url.Parse(statResp.JSON200.PhysicalAddress) + require.NoError(t, err) + adapter, err := NewAdapter(ctx, physicalAddress.Scheme) + require.NoError(t, err) + require.NoError(t, adapter.Delete(ctx, physicalAddress)) + }) + + RunCmdAndVerifyContainsText(t, Lakectl()+" fs upload -s "+f.Name()+" lakefs://"+repoName+"/"+mainBranch+"/"+objPath+extraArgs, false, fmt.Sprintf("Size: %d bytes", largeFileSize), vars) + + statResp, err := client.StatObjectWithResponse(ctx, repoName, mainBranch, &apigen.StatObjectParams{Path: objPath}) + require.NoError(t, err) + require.NotNil(t, statResp.JSON200, "stat %s: %s", objPath, statResp.Status()) + require.Equal(t, int64(largeFileSize), swag.Int64Value(statResp.JSON200.SizeBytes)) + } + + t.Run("pre-sign", func(t *testing.T) { + uploadAndVerify(t, "large_file_presign", " --pre-sign") + }) + t.Run("no pre-sign", func(t *testing.T) { + uploadAndVerify(t, "large_file", "") + }) +} + func TestLakectlCommit(t *testing.T) { repoName := GenerateUniqueRepositoryName() storage := GenerateUniqueStorageNamespace(repoName)