aboutsummaryrefslogtreecommitdiff
path: root/weed/s3api
diff options
context:
space:
mode:
Diffstat (limited to 'weed/s3api')
-rw-r--r--weed/s3api/filer_multipart.go27
-rw-r--r--weed/s3api/filer_util.go30
-rw-r--r--weed/s3api/s3api_bucket_handlers.go17
-rw-r--r--weed/s3api/s3api_handlers.go5
-rw-r--r--weed/s3api/s3api_object_multipart_handlers.go15
-rw-r--r--weed/s3api/s3api_objects_list_handlers.go14
6 files changed, 57 insertions, 51 deletions
diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go
index d39e821d0..e6af085fd 100644
--- a/weed/s3api/filer_multipart.go
+++ b/weed/s3api/filer_multipart.go
@@ -1,6 +1,7 @@
package s3api
import (
+ "context"
"encoding/xml"
"fmt"
"path/filepath"
@@ -21,11 +22,11 @@ type InitiateMultipartUploadResult struct {
s3.CreateMultipartUploadOutput
}
-func (s3a *S3ApiServer) createMultipartUpload(input *s3.CreateMultipartUploadInput) (output *InitiateMultipartUploadResult, code ErrorCode) {
+func (s3a *S3ApiServer) createMultipartUpload(ctx context.Context, input *s3.CreateMultipartUploadInput) (output *InitiateMultipartUploadResult, code ErrorCode) {
uploadId, _ := uuid.NewV4()
uploadIdString := uploadId.String()
- if err := s3a.mkdir(s3a.genUploadsFolder(*input.Bucket), uploadIdString, func(entry *filer_pb.Entry) {
+ if err := s3a.mkdir(ctx, s3a.genUploadsFolder(*input.Bucket), uploadIdString, func(entry *filer_pb.Entry) {
if entry.Extended == nil {
entry.Extended = make(map[string][]byte)
}
@@ -51,11 +52,11 @@ type CompleteMultipartUploadResult struct {
s3.CompleteMultipartUploadOutput
}
-func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploadInput) (output *CompleteMultipartUploadResult, code ErrorCode) {
+func (s3a *S3ApiServer) completeMultipartUpload(ctx context.Context, input *s3.CompleteMultipartUploadInput) (output *CompleteMultipartUploadResult, code ErrorCode) {
uploadDirectory := s3a.genUploadsFolder(*input.Bucket) + "/" + *input.UploadId
- entries, err := s3a.list(uploadDirectory, "", "", false, 0)
+ entries, err := s3a.list(ctx, uploadDirectory, "", "", false, 0)
if err != nil {
glog.Errorf("completeMultipartUpload %s %s error: %v", *input.Bucket, *input.UploadId, err)
return nil, ErrNoSuchUpload
@@ -90,7 +91,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa
}
dirName = fmt.Sprintf("%s/%s/%s", s3a.option.BucketsPath, *input.Bucket, dirName)
- err = s3a.mkFile(dirName, entryName, finalParts)
+ err = s3a.mkFile(ctx, dirName, entryName, finalParts)
if err != nil {
glog.Errorf("completeMultipartUpload %s/%s error: %v", dirName, entryName, err)
@@ -105,22 +106,22 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa
},
}
- if err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, false, true); err != nil {
+ if err = s3a.rm(ctx, s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, false, true); err != nil {
glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err)
}
return
}
-func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput) (output *s3.AbortMultipartUploadOutput, code ErrorCode) {
+func (s3a *S3ApiServer) abortMultipartUpload(ctx context.Context, input *s3.AbortMultipartUploadInput) (output *s3.AbortMultipartUploadOutput, code ErrorCode) {
- exists, err := s3a.exists(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true)
+ exists, err := s3a.exists(ctx, s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true)
if err != nil {
glog.V(1).Infof("bucket %s abort upload %s: %v", *input.Bucket, *input.UploadId, err)
return nil, ErrNoSuchUpload
}
if exists {
- err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true, true)
+ err = s3a.rm(ctx, s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true, true)
}
if err != nil {
glog.V(1).Infof("bucket %s remove upload %s: %v", *input.Bucket, *input.UploadId, err)
@@ -135,7 +136,7 @@ type ListMultipartUploadsResult struct {
s3.ListMultipartUploadsOutput
}
-func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput) (output *ListMultipartUploadsResult, code ErrorCode) {
+func (s3a *S3ApiServer) listMultipartUploads(ctx context.Context, input *s3.ListMultipartUploadsInput) (output *ListMultipartUploadsResult, code ErrorCode) {
output = &ListMultipartUploadsResult{
ListMultipartUploadsOutput: s3.ListMultipartUploadsOutput{
@@ -148,7 +149,7 @@ func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput
},
}
- entries, err := s3a.list(s3a.genUploadsFolder(*input.Bucket), *input.Prefix, *input.KeyMarker, true, int(*input.MaxUploads))
+ entries, err := s3a.list(ctx, s3a.genUploadsFolder(*input.Bucket), *input.Prefix, *input.KeyMarker, true, int(*input.MaxUploads))
if err != nil {
glog.Errorf("listMultipartUploads %s error: %v", *input.Bucket, err)
return
@@ -172,7 +173,7 @@ type ListPartsResult struct {
s3.ListPartsOutput
}
-func (s3a *S3ApiServer) listObjectParts(input *s3.ListPartsInput) (output *ListPartsResult, code ErrorCode) {
+func (s3a *S3ApiServer) listObjectParts(ctx context.Context, input *s3.ListPartsInput) (output *ListPartsResult, code ErrorCode) {
output = &ListPartsResult{
ListPartsOutput: s3.ListPartsOutput{
Bucket: input.Bucket,
@@ -183,7 +184,7 @@ func (s3a *S3ApiServer) listObjectParts(input *s3.ListPartsInput) (output *ListP
},
}
- entries, err := s3a.list(s3a.genUploadsFolder(*input.Bucket)+"/"+*input.UploadId,
+ entries, err := s3a.list(ctx, s3a.genUploadsFolder(*input.Bucket)+"/"+*input.UploadId,
"", fmt.Sprintf("%04d.part", *input.PartNumberMarker), false, int(*input.MaxParts))
if err != nil {
glog.Errorf("listObjectParts %s %s error: %v", *input.Bucket, *input.UploadId, err)
diff --git a/weed/s3api/filer_util.go b/weed/s3api/filer_util.go
index 40c5a3e26..4f2c1578b 100644
--- a/weed/s3api/filer_util.go
+++ b/weed/s3api/filer_util.go
@@ -10,8 +10,8 @@ import (
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
)
-func (s3a *S3ApiServer) mkdir(parentDirectoryPath string, dirName string, fn func(entry *filer_pb.Entry)) error {
- return s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+func (s3a *S3ApiServer) mkdir(ctx context.Context, parentDirectoryPath string, dirName string, fn func(entry *filer_pb.Entry)) error {
+ return s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
entry := &filer_pb.Entry{
Name: dirName,
@@ -35,7 +35,7 @@ func (s3a *S3ApiServer) mkdir(parentDirectoryPath string, dirName string, fn fun
}
glog.V(1).Infof("mkdir: %v", request)
- if _, err := client.CreateEntry(context.Background(), request); err != nil {
+ if _, err := client.CreateEntry(ctx, request); err != nil {
return fmt.Errorf("mkdir %s/%s: %v", parentDirectoryPath, dirName, err)
}
@@ -43,8 +43,8 @@ func (s3a *S3ApiServer) mkdir(parentDirectoryPath string, dirName string, fn fun
})
}
-func (s3a *S3ApiServer) mkFile(parentDirectoryPath string, fileName string, chunks []*filer_pb.FileChunk) error {
- return s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+func (s3a *S3ApiServer) mkFile(ctx context.Context, parentDirectoryPath string, fileName string, chunks []*filer_pb.FileChunk) error {
+ return s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
entry := &filer_pb.Entry{
Name: fileName,
@@ -65,7 +65,7 @@ func (s3a *S3ApiServer) mkFile(parentDirectoryPath string, fileName string, chun
}
glog.V(1).Infof("create file: %s/%s", parentDirectoryPath, fileName)
- if _, err := client.CreateEntry(context.Background(), request); err != nil {
+ if _, err := client.CreateEntry(ctx, request); err != nil {
return fmt.Errorf("create file %s/%s: %v", parentDirectoryPath, fileName, err)
}
@@ -73,9 +73,9 @@ func (s3a *S3ApiServer) mkFile(parentDirectoryPath string, fileName string, chun
})
}
-func (s3a *S3ApiServer) list(parentDirectoryPath, prefix, startFrom string, inclusive bool, limit int) (entries []*filer_pb.Entry, err error) {
+func (s3a *S3ApiServer) list(ctx context.Context, parentDirectoryPath, prefix, startFrom string, inclusive bool, limit int) (entries []*filer_pb.Entry, err error) {
- err = s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+ err = s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.ListEntriesRequest{
Directory: parentDirectoryPath,
@@ -86,7 +86,7 @@ func (s3a *S3ApiServer) list(parentDirectoryPath, prefix, startFrom string, incl
}
glog.V(4).Infof("read directory: %v", request)
- resp, err := client.ListEntries(context.Background(), request)
+ resp, err := client.ListEntries(ctx, request)
if err != nil {
return fmt.Errorf("list dir %v: %v", parentDirectoryPath, err)
}
@@ -100,11 +100,9 @@ func (s3a *S3ApiServer) list(parentDirectoryPath, prefix, startFrom string, incl
}
-func (s3a *S3ApiServer) rm(parentDirectoryPath string, entryName string, isDirectory, isDeleteData, isRecursive bool) error {
+func (s3a *S3ApiServer) rm(ctx context.Context, parentDirectoryPath string, entryName string, isDirectory, isDeleteData, isRecursive bool) error {
- return s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-
- ctx := context.Background()
+ return s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.DeleteEntryRequest{
Directory: parentDirectoryPath,
@@ -123,11 +121,9 @@ func (s3a *S3ApiServer) rm(parentDirectoryPath string, entryName string, isDirec
}
-func (s3a *S3ApiServer) exists(parentDirectoryPath string, entryName string, isDirectory bool) (exists bool, err error) {
-
- err = s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+func (s3a *S3ApiServer) exists(ctx context.Context, parentDirectoryPath string, entryName string, isDirectory bool) (exists bool, err error) {
- ctx := context.Background()
+ err = s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.LookupDirectoryEntryRequest{
Directory: parentDirectoryPath,
diff --git a/weed/s3api/s3api_bucket_handlers.go b/weed/s3api/s3api_bucket_handlers.go
index 35aa85493..492d94616 100644
--- a/weed/s3api/s3api_bucket_handlers.go
+++ b/weed/s3api/s3api_bucket_handlers.go
@@ -31,7 +31,7 @@ func (s3a *S3ApiServer) ListBucketsHandler(w http.ResponseWriter, r *http.Reques
var response ListAllMyBucketsResult
- entries, err := s3a.list(s3a.option.BucketsPath, "", "", false, math.MaxInt32)
+ entries, err := s3a.list(context.Background(), s3a.option.BucketsPath, "", "", false, math.MaxInt32)
if err != nil {
writeErrorResponse(w, ErrInternalError, r.URL)
@@ -65,7 +65,7 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
bucket := vars["bucket"]
// create the folder for bucket, but lazily create actual collection
- if err := s3a.mkdir(s3a.option.BucketsPath, bucket, nil); err != nil {
+ if err := s3a.mkdir(context.Background(), s3a.option.BucketsPath, bucket, nil); err != nil {
writeErrorResponse(w, ErrInternalError, r.URL)
return
}
@@ -78,9 +78,8 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque
vars := mux.Vars(r)
bucket := vars["bucket"]
- err := s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-
- ctx := context.Background()
+ ctx := context.Background()
+ err := s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
// delete collection
deleteCollectionRequest := &filer_pb.DeleteCollectionRequest{
@@ -95,7 +94,7 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque
return nil
})
- err = s3a.rm(s3a.option.BucketsPath, bucket, true, false, true)
+ err = s3a.rm(ctx, s3a.option.BucketsPath, bucket, true, false, true)
if err != nil {
writeErrorResponse(w, ErrInternalError, r.URL)
@@ -110,7 +109,9 @@ func (s3a *S3ApiServer) HeadBucketHandler(w http.ResponseWriter, r *http.Request
vars := mux.Vars(r)
bucket := vars["bucket"]
- err := s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+ ctx := context.Background()
+
+ err := s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.LookupDirectoryEntryRequest{
Directory: s3a.option.BucketsPath,
@@ -118,7 +119,7 @@ func (s3a *S3ApiServer) HeadBucketHandler(w http.ResponseWriter, r *http.Request
}
glog.V(1).Infof("lookup bucket: %v", request)
- if _, err := client.LookupDirectoryEntry(context.Background(), request); err != nil {
+ if _, err := client.LookupDirectoryEntry(ctx, request); err != nil {
return fmt.Errorf("lookup bucket %s/%s: %v", s3a.option.BucketsPath, bucket, err)
}
diff --git a/weed/s3api/s3api_handlers.go b/weed/s3api/s3api_handlers.go
index 5d92085cc..5a63648ca 100644
--- a/weed/s3api/s3api_handlers.go
+++ b/weed/s3api/s3api_handlers.go
@@ -2,6 +2,7 @@ package s3api
import (
"bytes"
+ "context"
"encoding/base64"
"encoding/xml"
"fmt"
@@ -35,9 +36,9 @@ func encodeResponse(response interface{}) []byte {
return bytesBuffer.Bytes()
}
-func (s3a *S3ApiServer) withFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
+func (s3a *S3ApiServer) withFilerClient(ctx context.Context, fn func(filer_pb.SeaweedFilerClient) error) error {
- grpcConnection, err := util.GrpcDial(s3a.option.FilerGrpcAddress, s3a.option.GrpcDialOption)
+ grpcConnection, err := util.GrpcDial(ctx, s3a.option.FilerGrpcAddress, s3a.option.GrpcDialOption)
if err != nil {
return fmt.Errorf("fail to dial %s: %v", s3a.option.FilerGrpcAddress, err)
}
diff --git a/weed/s3api/s3api_object_multipart_handlers.go b/weed/s3api/s3api_object_multipart_handlers.go
index 267d126c5..6643cb105 100644
--- a/weed/s3api/s3api_object_multipart_handlers.go
+++ b/weed/s3api/s3api_object_multipart_handlers.go
@@ -1,6 +1,7 @@
package s3api
import (
+ "context"
"fmt"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/s3"
@@ -25,7 +26,7 @@ func (s3a *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http
bucket = vars["bucket"]
object = vars["object"]
- response, errCode := s3a.createMultipartUpload(&s3.CreateMultipartUploadInput{
+ response, errCode := s3a.createMultipartUpload(context.Background(), &s3.CreateMultipartUploadInput{
Bucket: aws.String(bucket),
Key: aws.String(object),
})
@@ -50,7 +51,7 @@ func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r
// Get upload id.
uploadID, _, _, _ := getObjectResources(r.URL.Query())
- response, errCode := s3a.completeMultipartUpload(&s3.CompleteMultipartUploadInput{
+ response, errCode := s3a.completeMultipartUpload(context.Background(), &s3.CompleteMultipartUploadInput{
Bucket: aws.String(bucket),
Key: aws.String(object),
UploadId: aws.String(uploadID),
@@ -76,7 +77,7 @@ func (s3a *S3ApiServer) AbortMultipartUploadHandler(w http.ResponseWriter, r *ht
// Get upload id.
uploadID, _, _, _ := getObjectResources(r.URL.Query())
- response, errCode := s3a.abortMultipartUpload(&s3.AbortMultipartUploadInput{
+ response, errCode := s3a.abortMultipartUpload(context.Background(), &s3.AbortMultipartUploadInput{
Bucket: aws.String(bucket),
Key: aws.String(object),
UploadId: aws.String(uploadID),
@@ -111,7 +112,7 @@ func (s3a *S3ApiServer) ListMultipartUploadsHandler(w http.ResponseWriter, r *ht
}
}
- response, errCode := s3a.listMultipartUploads(&s3.ListMultipartUploadsInput{
+ response, errCode := s3a.listMultipartUploads(context.Background(), &s3.ListMultipartUploadsInput{
Bucket: aws.String(bucket),
Delimiter: aws.String(delimiter),
EncodingType: aws.String(encodingType),
@@ -148,7 +149,7 @@ func (s3a *S3ApiServer) ListObjectPartsHandler(w http.ResponseWriter, r *http.Re
return
}
- response, errCode := s3a.listObjectParts(&s3.ListPartsInput{
+ response, errCode := s3a.listObjectParts(context.Background(), &s3.ListPartsInput{
Bucket: aws.String(bucket),
Key: aws.String(object),
MaxParts: aws.Int64(int64(maxParts)),
@@ -174,8 +175,10 @@ func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Requ
rAuthType := getRequestAuthType(r)
+ ctx := context.Background()
+
uploadID := r.URL.Query().Get("uploadId")
- exists, err := s3a.exists(s3a.genUploadsFolder(bucket), uploadID, true)
+ exists, err := s3a.exists(ctx, s3a.genUploadsFolder(bucket), uploadID, true)
if !exists {
writeErrorResponse(w, ErrNoSuchUpload, r.URL)
return
diff --git a/weed/s3api/s3api_objects_list_handlers.go b/weed/s3api/s3api_objects_list_handlers.go
index 927416e0f..a685802d2 100644
--- a/weed/s3api/s3api_objects_list_handlers.go
+++ b/weed/s3api/s3api_objects_list_handlers.go
@@ -44,7 +44,9 @@ func (s3a *S3ApiServer) ListObjectsV2Handler(w http.ResponseWriter, r *http.Requ
marker = startAfter
}
- response, err := s3a.listFilerEntries(bucket, originalPrefix, maxKeys, marker)
+ ctx := context.Background()
+
+ response, err := s3a.listFilerEntries(ctx, bucket, originalPrefix, maxKeys, marker)
if err != nil {
writeErrorResponse(w, ErrInternalError, r.URL)
@@ -62,6 +64,8 @@ func (s3a *S3ApiServer) ListObjectsV1Handler(w http.ResponseWriter, r *http.Requ
vars := mux.Vars(r)
bucket := vars["bucket"]
+ ctx := context.Background()
+
originalPrefix, marker, delimiter, maxKeys := getListObjectsV1Args(r.URL.Query())
if maxKeys < 0 {
@@ -73,7 +77,7 @@ func (s3a *S3ApiServer) ListObjectsV1Handler(w http.ResponseWriter, r *http.Requ
return
}
- response, err := s3a.listFilerEntries(bucket, originalPrefix, maxKeys, marker)
+ response, err := s3a.listFilerEntries(ctx, bucket, originalPrefix, maxKeys, marker)
if err != nil {
writeErrorResponse(w, ErrInternalError, r.URL)
@@ -83,13 +87,13 @@ func (s3a *S3ApiServer) ListObjectsV1Handler(w http.ResponseWriter, r *http.Requ
writeSuccessResponseXML(w, encodeResponse(response))
}
-func (s3a *S3ApiServer) listFilerEntries(bucket, originalPrefix string, maxKeys int, marker string) (response ListBucketResult, err error) {
+func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, bucket, originalPrefix string, maxKeys int, marker string) (response ListBucketResult, err error) {
// convert full path prefix into directory name and prefix for entry name
dir, prefix := filepath.Split(originalPrefix)
// check filer
- err = s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+ err = s3a.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.ListEntriesRequest{
Directory: fmt.Sprintf("%s/%s/%s", s3a.option.BucketsPath, bucket, dir),
@@ -99,7 +103,7 @@ func (s3a *S3ApiServer) listFilerEntries(bucket, originalPrefix string, maxKeys
InclusiveStartFrom: false,
}
- resp, err := client.ListEntries(context.Background(), request)
+ resp, err := client.ListEntries(ctx, request)
if err != nil {
return fmt.Errorf("list buckets: %v", err)
}
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 22:01:47 +0000