support putifabsent

esti/s3_gateway_test.go

@@ -4,9 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/aws/aws-sdk-go-v2/service/s3"
-	"github.com/go-openapi/swag"
 	"io"
 	"math/rand"
 	"net/http"
@@ -16,6 +13,11 @@ import (
 	"testing"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/smithy-go/middleware"
+	"github.com/go-openapi/swag"
+
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
 	"github.com/minio/minio-go/v7/pkg/tags"
@@ -182,6 +184,82 @@ func TestS3UploadAndDownload(t *testing.T) {
 	}
 }
 
+func setHTTPHeaders(ifNoneMatch string) func(*middleware.Stack) error {
+	return func(stack *middleware.Stack) error {
+		return stack.Build.Add(middleware.BuildMiddlewareFunc("AddIfNoneMatchHeader", func(
+			ctx context.Context, in middleware.BuildInput, next middleware.BuildHandler,
+		) (
+			middleware.BuildOutput, middleware.Metadata, error,
+		) {
+			if req, ok := in.Request.(*http.Request); ok {
+				// Add the If-None-Match header
+				req.Header.Set("If-None-Match", ifNoneMatch)
+				fmt.Printf("Set If-None-Match header: %s\n", ifNoneMatch) // Debug logging
+			}
+			// Continue with the next middleware handler
+			return next.HandleBuild(ctx, in)
+		}), middleware.Before)
+	}
+}
+func TestS3IfNoneMatch(t *testing.T) {
+	const parallelism = 10
+
+	ctx, _, repo := setupTest(t)
+	defer tearDownTest(repo)
+
+	s3Endpoint := viper.GetString("s3_endpoint")
+	s3Client := createS3Client(s3Endpoint, t)
+
+	type TestCase struct {
+		Path        string
+		Content     string
+		IfNoneMatch string
+		ExpectError bool
+	}
+
+	testCases := []TestCase{
+		{Path: "main/object1", Content: "data", IfNoneMatch: "", ExpectError: false},
+		{Path: "main/object1", Content: "data", IfNoneMatch: "*", ExpectError: true},
+		{Path: "main/object2", Content: "data", IfNoneMatch: "*", ExpectError: false},
+		{Path: "main/object2", Content: "data", IfNoneMatch: "", ExpectError: false},
+		{Path: "main/object3", Content: "data", IfNoneMatch: "hi", ExpectError: true},
+	}
+
+	objects := make(chan TestCase, parallelism*2)
+	wg := sync.WaitGroup{}
+
+	wg.Add(parallelism)
+	for i := 0; i < parallelism; i++ {
+		go func() {
+			defer wg.Done()
+			for tc := range objects {
+				// Create the PutObject request
+				input := &s3.PutObjectInput{
+					Bucket: aws.String(repo),
+					Key:    aws.String(tc.Path),
+					Body:   strings.NewReader(tc.Content),
+				}
+				fmt.Printf("Sending PutObject request for Path: %s with If-None-Match: %s\n", tc.Path, tc.IfNoneMatch) // Debug logging
+				_, err := s3Client.PutObject(ctx, input, s3.WithAPIOptions(setHTTPHeaders(tc.IfNoneMatch)))
+
+				if tc.ExpectError {
+					require.Error(t, err, "was expecting an error")
+				} else {
+					require.NoError(t, err, "wasn't expecting error")
+				}
+			}
+		}()
+	}
+	// Enqueue test cases
+	for _, tc := range testCases {
+		objects <- tc
+	}
+	close(objects)
+
+	// Wait for all workers to finish
+	wg.Wait()
+}
+
 func verifyObjectInfo(t *testing.T, got minio.ObjectInfo, expectedSize int) {
 	if got.Err != nil {
 		t.Errorf("%s: %s", got.Key, got.Err)

pkg/gateway/operations/operation_utils.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/treeverse/lakefs/pkg/catalog"
+	"github.com/treeverse/lakefs/pkg/graveler"
 	"github.com/treeverse/lakefs/pkg/logging"
 )
 
@@ -40,7 +41,7 @@ func shouldReplaceMetadata(req *http.Request) bool {
 	return req.Header.Get(amzMetadataDirectiveHeaderPrefix) == "REPLACE"
 }
 
-func (o *PathOperation) finishUpload(req *http.Request, mTime *time.Time, checksum, physicalAddress string, size int64, relative bool, metadata map[string]string, contentType string) error {
+func (o *PathOperation) finishUpload(req *http.Request, mTime *time.Time, checksum, physicalAddress string, size int64, relative bool, metadata map[string]string, contentType string, allowOverWrite bool) error {
 	var writeTime time.Time
 	if mTime == nil {
 		writeTime = time.Now()
@@ -59,7 +60,7 @@ func (o *PathOperation) finishUpload(req *http.Request, mTime *time.Time, checks
 		ContentType(contentType).
 		Build()
 
-	err := o.Catalog.CreateEntry(req.Context(), o.Repository.Name, o.Reference, entry)
+	err := o.Catalog.CreateEntry(req.Context(), o.Repository.Name, o.Reference, entry, graveler.WithIfAbsent(!allowOverWrite))
 	if err != nil {
 		o.Log(req).WithError(err).Error("could not update metadata")
 		return err

pkg/gateway/operations/postobject.go

@@ -94,6 +94,15 @@ func (controller *PostObject) HandleCompleteMultipartUpload(w http.ResponseWrite
 		_ = o.EncodeError(w, req, err, gatewayErrors.Codes.ToAPIErr(gatewayErrors.ErrInternalError))
 		return
 	}
+	allowOverWrite, err := o.checkIfAbsent(req)
+	if errors.Is(err, gatewayErrors.ErrPreconditionFailed) {
+		_ = o.EncodeError(w, req, err, gatewayErrors.Codes.ToAPIErr(gatewayErrors.ErrPreconditionFailed))
+		return
+	}
+	if errors.Is(err, gatewayErrors.ErrNotImplemented) {
+		_ = o.EncodeError(w, req, err, gatewayErrors.Codes.ToAPIErr(gatewayErrors.ErrNotImplemented))
+		return
+	}
 	objName := multiPart.PhysicalAddress
 	req = req.WithContext(logging.AddFields(req.Context(), logging.Fields{logging.PhysicalAddressFieldKey: objName}))
 	xmlMultipartComplete, err := io.ReadAll(req.Body)
@@ -124,7 +133,11 @@ func (controller *PostObject) HandleCompleteMultipartUpload(w http.ResponseWrite
 		return
 	}
 	checksum := strings.Split(resp.ETag, "-")[0]
-	err = o.finishUpload(req, resp.MTime, checksum, objName, resp.ContentLength, true, multiPart.Metadata, multiPart.ContentType)
+	err = o.finishUpload(req, resp.MTime, checksum, objName, resp.ContentLength, true, multiPart.Metadata, multiPart.ContentType, allowOverWrite)
+	if errors.Is(err, graveler.ErrPreconditionFailed) {
+		_ = o.EncodeError(w, req, err, gatewayErrors.Codes.ToAPIErr(gatewayErrors.ErrPreconditionFailed))
+		return
+	}
 	if errors.Is(err, graveler.ErrWriteToProtectedBranch) {
 		_ = o.EncodeError(w, req, err, gatewayErrors.Codes.ToAPIErr(gatewayErrors.ErrWriteToProtectedBranch))
 		return

pkg/gateway/operations/putobject.go

@@ -2,6 +2,7 @@
 
 import (
 	"errors"
+	"fmt"
 	"net/http"
 	"net/url"
 	"strconv"
@@ -20,6 +21,7 @@
 )
 
 const (
+	IfNoneMatchHeader     = "If-None-Match"
 	CopySourceHeader      = "x-amz-copy-source"
 	CopySourceRangeHeader = "x-amz-copy-source-range"
 	QueryParamUploadID    = "uploadId"
@@ -30,7 +32,6 @@
 
 func (controller *PutObject) RequiredPermissions(req *http.Request, repoID, _, destPath string) (permissions.Node, error) {
 	copySource := req.Header.Get(CopySourceHeader)
-
 	if len(copySource) == 0 {
 		return permissions.Node{
 			Permission: permissions.Permission{
@@ -298,6 +299,15 @@
 	o.Incr("put_object", o.Principal, o.Repository.Name, o.Reference)
 	storageClass := StorageClassFromHeader(req.Header)
 	opts := block.PutOpts{StorageClass: storageClass}
+	allowOverWrite, err := o.checkIfAbsent(req)
+	if errors.Is(err, gatewayErrors.ErrPreconditionFailed) {
+		_ = o.EncodeError(w, req, err, gatewayErrors.Codes.ToAPIErr(gatewayErrors.ErrPreconditionFailed))
+		return
+	}
+	if errors.Is(err, gatewayErrors.ErrNotImplemented) {
+		_ = o.EncodeError(w, req, err, gatewayErrors.Codes.ToAPIErr(gatewayErrors.ErrNotImplemented))
+		return
+	}
 	address := o.PathProvider.NewPath()
 	blob, err := upload.WriteBlob(req.Context(), o.BlockStore, o.Repository.StorageNamespace, address, req.Body, req.ContentLength, opts)
 	if err != nil {
@@ -309,7 +319,11 @@
 	// write metadata
 	metadata := amzMetaAsMetadata(req)
 	contentType := req.Header.Get("Content-Type")
-	err = o.finishUpload(req, &blob.CreationDate, blob.Checksum, blob.PhysicalAddress, blob.Size, true, metadata, contentType)
+	err = o.finishUpload(req, &blob.CreationDate, blob.Checksum, blob.PhysicalAddress, blob.Size, true, metadata, contentType, allowOverWrite)
+	if errors.Is(err, graveler.ErrPreconditionFailed) {
+		_ = o.EncodeError(w, req, err, gatewayErrors.Codes.ToAPIErr(gatewayErrors.ErrPreconditionFailed))
+		return
+	}
 	if errors.Is(err, graveler.ErrWriteToProtectedBranch) {
 		_ = o.EncodeError(w, req, err, gatewayErrors.Codes.ToAPIErr(gatewayErrors.ErrWriteToProtectedBranch))
 		return
@@ -325,3 +339,26 @@
 	o.SetHeader(w, "ETag", httputil.ETag(blob.Checksum))
 	w.WriteHeader(http.StatusOK)
 }
+
+func (o *PathOperation) checkIfAbsent(req *http.Request) (bool, error) {
+	for key, values := range req.Header {
+		for _, value := range values {
+			fmt.Printf("HEADER: %s = %s\n", key, value)
+		}
+	}
+	Header := req.Header.Get(IfNoneMatchHeader)
+	fmt.Println("HEADER: ", Header)
+	if Header == "" {
+		return true, nil
+	}
+	if Header == "*" {
+		_, err := o.Catalog.GetEntry(req.Context(), o.Repository.Name, o.Reference, o.Path, catalog.GetEntryParams{})
+		if err == nil {
+			return false, gatewayErrors.ErrPreconditionFailed
+		}
+		if !errors.Is(err, graveler.ErrNotFound) {
+			return false, gatewayErrors.ErrInternalError
+		}
+	}
+	return false, gatewayErrors.ErrNotImplemented
+}