Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Pass contentLength to the worker #791

Merged
merged 5 commits into from
Dec 6, 2023
Merged

Pass contentLength to the worker #791

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6628848
Pass contentLength to the worker
mike76-dev Dec 5, 2023
a1c9518
Fix cluster_test.go
mike76-dev Dec 5, 2023
f0ccab8
Put contentLength in upload options
mike76-dev Dec 6, 2023
ca9c4a1
Remove redundant code
mike76-dev Dec 6, 2023
6da810d
Address comments
mike76-dev Dec 6, 2023
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
8 changes: 4 additions & 4 deletions internal/testing/cluster_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1982,7 +1982,7 @@ func TestMultipartUploads(t *testing.T) {
// correctly.
putPart := func(partNum int, offset int, data []byte) string {
t.Helper()
res, err := w.UploadMultipartUploadPart(context.Background(), bytes.NewReader(data), api.DefaultBucketName, objPath, mpr.UploadID, partNum, api.UploadMultipartUploadPartOptions{EncryptionOffset: offset})
res, err := w.UploadMultipartUploadPart(context.Background(), bytes.NewReader(data), api.DefaultBucketName, objPath, mpr.UploadID, partNum, int64(len(data)), api.UploadMultipartUploadPartOptions{EncryptionOffset: offset})
tt.OK(err)
if res.ETag == "" {
t.Fatal("expected non-empty ETag")
Expand Down Expand Up @@ -2294,21 +2294,21 @@ func TestMultipartUploadWrappedByPartialSlabs(t *testing.T) {

// upload a part that is a partial slab
part3Data := bytes.Repeat([]byte{3}, int(slabSize)/4)
resp3, err := w.UploadMultipartUploadPart(context.Background(), bytes.NewReader(part3Data), api.DefaultBucketName, objPath, mpr.UploadID, 3, api.UploadMultipartUploadPartOptions{
resp3, err := w.UploadMultipartUploadPart(context.Background(), bytes.NewReader(part3Data), api.DefaultBucketName, objPath, mpr.UploadID, 3, int64(len(part3Data)), api.UploadMultipartUploadPartOptions{
EncryptionOffset: int(slabSize + slabSize/4),
})
tt.OK(err)

// upload a part that is exactly a full slab
part2Data := bytes.Repeat([]byte{2}, int(slabSize))
resp2, err := w.UploadMultipartUploadPart(context.Background(), bytes.NewReader(part2Data), api.DefaultBucketName, objPath, mpr.UploadID, 2, api.UploadMultipartUploadPartOptions{
resp2, err := w.UploadMultipartUploadPart(context.Background(), bytes.NewReader(part2Data), api.DefaultBucketName, objPath, mpr.UploadID, 2, int64(len(part2Data)), api.UploadMultipartUploadPartOptions{
EncryptionOffset: int(slabSize / 4),
})
tt.OK(err)

// upload another part the same size as the first one
part1Data := bytes.Repeat([]byte{1}, int(slabSize)/4)
resp1, err := w.UploadMultipartUploadPart(context.Background(), bytes.NewReader(part1Data), api.DefaultBucketName, objPath, mpr.UploadID, 1, api.UploadMultipartUploadPartOptions{
resp1, err := w.UploadMultipartUploadPart(context.Background(), bytes.NewReader(part1Data), api.DefaultBucketName, objPath, mpr.UploadID, 1, int64(len(part1Data)), api.UploadMultipartUploadPartOptions{
EncryptionOffset: 0,
})
tt.OK(err)
Expand Down
2 changes: 1 addition & 1 deletion s3/backend.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -401,7 +401,7 @@ func (s *s3) CreateMultipartUpload(ctx context.Context, bucket, key string, meta
}

func (s *s3) UploadPart(ctx context.Context, bucket, object string, id gofakes3.UploadID, partNumber int, contentLength int64, input io.Reader) (*gofakes3.UploadPartResult, error) {
res, err := s.w.UploadMultipartUploadPart(ctx, input, bucket, object, string(id), partNumber, api.UploadMultipartUploadPartOptions{
res, err := s.w.UploadMultipartUploadPart(ctx, input, bucket, object, string(id), partNumber, contentLength, api.UploadMultipartUploadPartOptions{
DisablePreshardingEncryption: true,
})
if err != nil {
Expand Down
2 changes: 1 addition & 1 deletion s3/s3.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ type bus interface {
type worker interface {
GetObject(ctx context.Context, bucket, path string, opts api.DownloadObjectOptions) (*api.GetObjectResponse, error)
UploadObject(ctx context.Context, r io.Reader, bucket, path string, opts api.UploadObjectOptions) (*api.UploadObjectResponse, error)
UploadMultipartUploadPart(ctx context.Context, r io.Reader, bucket, path, uploadID string, partNumber int, opts api.UploadMultipartUploadPartOptions) (*api.UploadMultipartUploadPartResponse, error)
UploadMultipartUploadPart(ctx context.Context, r io.Reader, bucket, path, uploadID string, partNumber int, contentLength int64, opts api.UploadMultipartUploadPartOptions) (*api.UploadMultipartUploadPartResponse, error)
}

func (l *gofakes3Logger) Print(level gofakes3.LogLevel, v ...interface{}) {
Expand Down
3 changes: 2 additions & 1 deletion worker/client/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,7 @@ func (c *Client) State() (state api.WorkerStateResponse, err error) {
}

// UploadMultipartUploadPart uploads part of the data for a multipart upload.
func (c *Client) UploadMultipartUploadPart(ctx context.Context, r io.Reader, bucket, path, uploadID string, partNumber int, opts api.UploadMultipartUploadPartOptions) (*api.UploadMultipartUploadPartResponse, error) {
func (c *Client) UploadMultipartUploadPart(ctx context.Context, r io.Reader, bucket, path, uploadID string, partNumber int, contentLength int64, opts api.UploadMultipartUploadPartOptions) (*api.UploadMultipartUploadPartResponse, error) {
path = api.ObjectPathEscape(path)
c.c.Custom("PUT", fmt.Sprintf("/multipart/%s", path), []byte{}, nil)

Expand All @@ -194,6 +194,7 @@ func (c *Client) UploadMultipartUploadPart(ctx context.Context, r io.Reader, buc
panic(err)
}
req.SetBasicAuth("", c.c.WithContext(ctx).Password)
req.Header.Add("X-Content-Length", fmt.Sprintf("%d", contentLength))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Try setting req.ContentLength = contentLength instead. That way the Content-Length header will be set.
But contentLength should be part of opts and only be set if opts.ContentLength != 0.

It would also help us a lot if you did the same for PutObject in backend.go since that also seems to have a size that is currently not used.

Copy link
Member

@ChrisSchinnerl ChrisSchinnerl Dec 6, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you want to go the extra mile you can extend the client methods to try and cast r.(io.Seeker) iff opts.ContentLength = 0 and then seek to the end and back to the beginning to determine the content length of the input. That would make sure that calling the client methods with an os.File or bytes.Reader will set the content length implicitly.

Copy link
Contributor Author

@mike76-dev mike76-dev Dec 6, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That would make sure that calling the client methods with an os.File or bytes.Reader will set the content length implicitly.

So, no need to set req.ContentLength = <offset> in this case?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, in that case we don't set it and just use the seeker.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh sry I misunderstood your question. We still need to set the content length. Just not from the input but from the offset of the seeker.

Please also make sure to handle all errors

resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, err
Expand Down