Dragonfly2

1
/*
2
 *     Copyright 2022 The Dragonfly Authors
3
 *
4
 * Licensed under the Apache License, Version 2.0 (the "License");
5
 * you may not use this file except in compliance with the License.
6
 * You may obtain a copy of the License at
7
 *
8
 *      http://www.apache.org/licenses/LICENSE-2.0
9
 *
10
 * Unless required by applicable law or agreed to in writing, software
11
 * distributed under the License is distributed on an "AS IS" BASIS,
12
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13
 * See the License for the specific language governing permissions and
14
 * limitations under the License.
15
 */
16

17
//go:generate mockgen -destination mocks/dfstore_mock.go -source dfstore.go -package mocks
18

19
package dfstore
20

21
import (
22
	"bytes"
23
	"context"
24
	"encoding/json"
25
	"errors"
26
	"fmt"
27
	"io"
28
	"mime/multipart"
29
	"net/http"
30
	"net/url"
31
	"path/filepath"
32
	"strconv"
33
	"strings"
34
	"time"
35

36
	"github.com/go-http-utils/headers"
37

38
	"d7y.io/dragonfly/v2/client/config"
39
	"d7y.io/dragonfly/v2/client/daemon/objectstorage"
40
	pkgobjectstorage "d7y.io/dragonfly/v2/pkg/objectstorage"
41
)
42

43
const (
44
	// DefaultGetObjectMetadatasLimit is the default limit of get object metadatas.
45
	DefaultGetObjectMetadatasLimit = 1000
46

47
	// MaxGetObjectMetadatasLimit is the max limit of get object metadatas.
48
	MaxGetObjectMetadatasLimit = 1000
49
	// DefaultPutObjectBufferSize is the buffer size of io.CopyBuffer
50
	DefaultPutObjectBufferSize = 64 * 1024 * 1024
51
)
52

53
// Dfstore is the interface used for object storage.
54
type Dfstore interface {
55
	// CreateBucketRequestWithContext returns *http.Request of create bucket.
56
	CreateBucketRequestWithContext(ctx context.Context, input *CreateBucketInput) (*http.Request, error)
57

58
	// CreateBucket creates bucket.
59
	CreateBucketWithContext(ctx context.Context, input *CreateBucketInput) error
60

61
	// GetObjectMetadataRequestWithContext returns *http.Request of getting object metadata.
62
	GetObjectMetadataRequestWithContext(ctx context.Context, input *GetObjectMetadataInput) (*http.Request, error)
63

64
	// GetObjectMetadataWithContext returns matedata of object.
65
	GetObjectMetadataWithContext(ctx context.Context, input *GetObjectMetadataInput) (*pkgobjectstorage.ObjectMetadata, error)
66

67
	// GetObjectRequestWithContext returns *http.Request of getting object.
68
	GetObjectRequestWithContext(ctx context.Context, input *GetObjectInput) (*http.Request, error)
69

70
	// GetObjectWithContext returns data of object.
71
	GetObjectWithContext(ctx context.Context, input *GetObjectInput) (io.ReadCloser, error)
72

73
	// GetObjectMetadatasRequestWithContext returns *http.Request of getting object metadatas.
74
	GetObjectMetadatasRequestWithContext(ctx context.Context, input *GetObjectMetadatasInput) (*http.Request, error)
75

76
	// GetObjectMetadatasWithContext returns list of object metadatas.
77
	GetObjectMetadatasWithContext(ctx context.Context, input *GetObjectMetadatasInput) (*pkgobjectstorage.ObjectMetadatas, error)
78

79
	// PutObjectRequestWithContext returns *http.Request of putting object.
80
	PutObjectRequestWithContext(ctx context.Context, input *PutObjectInput) (*http.Request, error)
81

82
	// PutObjectWithContext puts data of object.
83
	PutObjectWithContext(ctx context.Context, input *PutObjectInput) error
84

85
	// CopyObjectRequestWithContext returns *http.Request of copying object.
86
	CopyObjectRequestWithContext(ctx context.Context, input *CopyObjectInput) (*http.Request, error)
87

88
	// CopyObjectWithContext copy object from source to destination.
89
	CopyObjectWithContext(ctx context.Context, input *CopyObjectInput) error
90

91
	// DeleteObjectRequestWithContext returns *http.Request of deleting object.
92
	DeleteObjectRequestWithContext(ctx context.Context, input *DeleteObjectInput) (*http.Request, error)
93

94
	// DeleteObjectWithContext deletes data of object.
95
	DeleteObjectWithContext(ctx context.Context, input *DeleteObjectInput) error
96

97
	// IsObjectExistRequestWithContext returns *http.Request of heading object.
98
	IsObjectExistRequestWithContext(ctx context.Context, input *IsObjectExistInput) (*http.Request, error)
99

100
	// IsObjectExistWithContext returns whether the object exists.
101
	IsObjectExistWithContext(ctx context.Context, input *IsObjectExistInput) (bool, error)
102
}
103

104
// dfstore provides object storage function.
105
type dfstore struct {
106
	endpoint   string
107
	httpClient *http.Client
108
}
109

110
// Option is a functional option for configuring the dfstore.
111
type Option func(dfs *dfstore)
112

113
// WithHTTPClient set http client for dfstore.
114
func WithHTTPClient(client *http.Client) Option {
115
	return func(dfs *dfstore) {
116
		dfs.httpClient = client
117
	}
118
}
119

120
// New dfstore instance.
121
func New(endpoint string, options ...Option) Dfstore {
122
	dfs := &dfstore{
123
		endpoint:   endpoint,
124
		httpClient: http.DefaultClient,
125
	}
126

127
	for _, opt := range options {
128
		opt(dfs)
129
	}
130

131
	return dfs
132
}
133

134
// GetObjectMetadataInput is used to construct request of getting object metadata.
135
type GetObjectMetadataInput struct {
136
	// BucketName is bucket name.
137
	BucketName string
138

139
	// ObjectKey is object key.
140
	ObjectKey string
141
}
142

143
// Validate validates GetObjectMetadataInput fields.
144
func (i *GetObjectMetadataInput) Validate() error {
145
	if i.BucketName == "" {
146
		return errors.New("invalid BucketName")
147
	}
148

149
	if i.ObjectKey == "" {
150
		return errors.New("invalid ObjectKey")
151
	}
152

153
	return nil
154
}
155

156
// GetObjectMetadataRequestWithContext returns *http.Request of getting object metadata.
157
func (dfs *dfstore) GetObjectMetadataRequestWithContext(ctx context.Context, input *GetObjectMetadataInput) (*http.Request, error) {
158
	if err := input.Validate(); err != nil {
159
		return nil, err
160
	}
161

162
	u, err := url.Parse(dfs.endpoint)
163
	if err != nil {
164
		return nil, err
165
	}
166

167
	u.Path = filepath.Join("buckets", input.BucketName, "objects", input.ObjectKey)
168

169
	if strings.HasSuffix(input.ObjectKey, "/") {
170
		u.Path += "/"
171
	}
172

173
	req, err := http.NewRequestWithContext(ctx, http.MethodHead, u.String(), nil)
174
	if err != nil {
175
		return nil, err
176
	}
177

178
	return req, nil
179
}
180

181
// GetObjectMetadataWithContext returns metadata of object.
182
func (dfs *dfstore) GetObjectMetadataWithContext(ctx context.Context, input *GetObjectMetadataInput) (*pkgobjectstorage.ObjectMetadata, error) {
183
	req, err := dfs.GetObjectMetadataRequestWithContext(ctx, input)
184
	if err != nil {
185
		return nil, err
186
	}
187

188
	resp, err := dfs.httpClient.Do(req)
189
	if err != nil {
190
		return nil, err
191
	}
192
	defer resp.Body.Close()
193

194
	if resp.StatusCode/100 != 2 {
195
		return nil, fmt.Errorf("bad response status %s", resp.Status)
196
	}
197

198
	contentLength, err := strconv.ParseInt(resp.Header.Get(headers.ContentLength), 10, 64)
199
	if err != nil {
200
		return nil, err
201
	}
202

203
	lastModifiedTime, err := time.Parse(http.TimeFormat, resp.Header.Get(config.HeaderDragonflyObjectMetaLastModifiedTime))
204
	if err != nil {
205
		return nil, err
206
	}
207

208
	return &pkgobjectstorage.ObjectMetadata{
209
		ContentDisposition: resp.Header.Get(headers.ContentDisposition),
210
		ContentEncoding:    resp.Header.Get(headers.ContentEncoding),
211
		ContentLanguage:    resp.Header.Get(headers.ContentLanguage),
212
		ContentLength:      int64(contentLength),
213
		ContentType:        resp.Header.Get(headers.ContentType),
214
		ETag:               resp.Header.Get(headers.ContentType),
215
		Digest:             resp.Header.Get(config.HeaderDragonflyObjectMetaDigest),
216
		LastModifiedTime:   lastModifiedTime,
217
		StorageClass:       resp.Header.Get(config.HeaderDragonflyObjectMetaStorageClass),
218
	}, nil
219
}
220

221
// GetObjectMetadatasInput is used to construct request of getting object metadatas.
222
type GetObjectMetadatasInput struct {
223
	// BucketName is the bucket name.
224
	BucketName string
225

226
	// Prefix filters the objects by their key's prefix.
227
	Prefix string
228

229
	// Marker is used for pagination, indicating the object key to start listing from.
230
	Marker string
231

232
	// Delimiter is used to create a hierarchical structure, simulating directories in the listing results.
233
	Delimiter string
234

235
	// Limit specifies the maximum number of objects to be returned in a single listing request.
236
	Limit int64
237
}
238

239
// Convert converts GetObjectMetadatasInput fields.
240
func (i *GetObjectMetadatasInput) Convert() {
241
	if i.Limit == 0 {
242
		i.Limit = DefaultGetObjectMetadatasLimit
243
	}
244

245
	if i.Limit > MaxGetObjectMetadatasLimit {
246
		i.Limit = DefaultGetObjectMetadatasLimit
247
	}
248
}
249

250
// Validate validates GetObjectMetadatasInput fields.
251
func (i *GetObjectMetadatasInput) Validate() error {
252
	if i.BucketName == "" {
253
		return errors.New("invalid BucketName")
254
	}
255

256
	if i.Limit <= 0 {
257
		return errors.New("invalid limit")
258
	}
259

260
	return nil
261
}
262

263
// GetObjectMetadatasRequestWithContext returns *http.Request of getting object metadatas.
264
func (dfs *dfstore) GetObjectMetadatasRequestWithContext(ctx context.Context, input *GetObjectMetadatasInput) (*http.Request, error) {
265
	// Convert input fields.
266
	input.Convert()
267

268
	if err := input.Validate(); err != nil {
269
		return nil, err
270
	}
271

272
	u, err := url.Parse(dfs.endpoint)
273
	if err != nil {
274
		return nil, err
275
	}
276

277
	u.Path = filepath.Join("buckets", input.BucketName, "metadatas")
278

279
	query := u.Query()
280
	if input.Prefix != "" {
281
		query.Set("prefix", input.Prefix)
282
	}
283

284
	if input.Marker != "" {
285
		query.Set("marker", input.Marker)
286
	}
287

288
	if input.Delimiter != "" {
289
		query.Set("delimiter", input.Delimiter)
290
	}
291

292
	if input.Limit != 0 {
293
		query.Set("limit", fmt.Sprint(input.Limit))
294
	}
295

296
	u.RawQuery = query.Encode()
297

298
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
299
	if err != nil {
300
		return nil, err
301
	}
302

303
	return req, nil
304
}
305

306
// GetObjectMetadatasWithContext returns *http.Request of getting object metadatas.
307
func (dfs *dfstore) GetObjectMetadatasWithContext(ctx context.Context, input *GetObjectMetadatasInput) (*pkgobjectstorage.ObjectMetadatas, error) {
308
	req, err := dfs.GetObjectMetadatasRequestWithContext(ctx, input)
309
	if err != nil {
310
		return nil, err
311
	}
312

313
	resp, err := dfs.httpClient.Do(req)
314
	if err != nil {
315
		return nil, err
316
	}
317
	defer resp.Body.Close()
318

319
	if resp.StatusCode/100 != 2 {
320
		return nil, fmt.Errorf("bad response status %s", resp.Status)
321
	}
322

323
	var metadatas pkgobjectstorage.ObjectMetadatas
324
	if err := json.NewDecoder(resp.Body).Decode(&metadatas); err != nil {
325
		return nil, err
326
	}
327

328
	return &metadatas, nil
329
}
330

331
// GetObjectInput is used to construct request of getting object.
332
type GetObjectInput struct {
333
	// BucketName is bucket name.
334
	BucketName string
335

336
	// ObjectKey is object key.
337
	ObjectKey string
338

339
	// Filter is used to generate a unique Task ID by
340
	// filtering unnecessary query params in the URL,
341
	// it is separated by & character.
342
	Filter string
343

344
	// Range is the HTTP range header.
345
	Range string
346
}
347

348
// Validate validates GetObjectInput fields.
349
func (i *GetObjectInput) Validate() error {
350
	if i.BucketName == "" {
351
		return errors.New("invalid BucketName")
352
	}
353

354
	if i.ObjectKey == "" {
355
		return errors.New("invalid ObjectKey")
356
	}
357

358
	return nil
359
}
360

361
// GetObjectRequestWithContext returns *http.Request of getting object.
362
func (dfs *dfstore) GetObjectRequestWithContext(ctx context.Context, input *GetObjectInput) (*http.Request, error) {
363
	if err := input.Validate(); err != nil {
364
		return nil, err
365
	}
366

367
	u, err := url.Parse(dfs.endpoint)
368
	if err != nil {
369
		return nil, err
370
	}
371

372
	u.Path = filepath.Join("buckets", input.BucketName, "objects", input.ObjectKey)
373

374
	if strings.HasSuffix(input.ObjectKey, "/") {
375
		u.Path += "/"
376
	}
377

378
	query := u.Query()
379
	if input.Filter != "" {
380
		query.Set("filter", input.Filter)
381
	}
382
	u.RawQuery = query.Encode()
383

384
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
385
	if err != nil {
386
		return nil, err
387
	}
388

389
	if input.Range != "" {
390
		req.Header.Set(headers.Range, input.Range)
391
	}
392

393
	return req, nil
394
}
395

396
// GetObjectWithContext returns data of object.
397
func (dfs *dfstore) GetObjectWithContext(ctx context.Context, input *GetObjectInput) (io.ReadCloser, error) {
398
	req, err := dfs.GetObjectRequestWithContext(ctx, input)
399
	if err != nil {
400
		return nil, err
401
	}
402

403
	resp, err := dfs.httpClient.Do(req)
404
	if err != nil {
405
		return nil, err
406
	}
407

408
	if resp.StatusCode/100 != 2 {
409
		return nil, fmt.Errorf("bad response status %s", resp.Status)
410
	}
411

412
	return resp.Body, nil
413
}
414

415
// PutObjectInput is used to construct request of putting object.
416
type PutObjectInput struct {
417
	// BucketName is bucket name.
418
	BucketName string
419

420
	// ObjectKey is object key.
421
	ObjectKey string
422

423
	// Filter is used to generate a unique Task ID by
424
	// filtering unnecessary query params in the URL,
425
	// it is separated by & character.
426
	Filter string
427

428
	// Mode is the mode in which the backend is written,
429
	// including WriteBack and AsyncWriteBack.
430
	Mode int
431

432
	// MaxReplicas is the maximum number of
433
	// replicas of an object cache in seed peers.
434
	MaxReplicas int
435

436
	// Reader is reader of object.
437
	Reader io.Reader
438
}
439

440
// Validate validates PutObjectInput fields.
441
func (i *PutObjectInput) Validate() error {
442
	if i.BucketName == "" {
443
		return errors.New("invalid BucketName")
444
	}
445

446
	if i.ObjectKey == "" {
447
		return errors.New("invalid ObjectKey")
448
	}
449

450
	if i.Mode != objectstorage.WriteBack && i.Mode != objectstorage.AsyncWriteBack {
451
		return errors.New("invalid Mode")
452
	}
453

454
	if i.MaxReplicas < 0 || i.MaxReplicas > 100 {
455
		return errors.New("invalid MaxReplicas")
456
	}
457

458
	return nil
459
}
460

461
// PutObjectRequestWithContext returns *http.Request of putting object.
462
func (dfs *dfstore) PutObjectRequestWithContext(ctx context.Context, input *PutObjectInput) (*http.Request, error) {
463
	if err := input.Validate(); err != nil {
464
		return nil, err
465
	}
466

467
	body := &bytes.Buffer{}
468
	writer := multipart.NewWriter(body)
469

470
	// AsyncWriteBack mode is used by default.
471
	if err := writer.WriteField("mode", fmt.Sprint(input.Mode)); err != nil {
472
		return nil, err
473
	}
474

475
	if input.Filter != "" {
476
		if err := writer.WriteField("filter", input.Filter); err != nil {
477
			return nil, err
478
		}
479
	}
480

481
	if input.MaxReplicas > 0 {
482
		if err := writer.WriteField("maxReplicas", fmt.Sprint(input.MaxReplicas)); err != nil {
483
			return nil, err
484
		}
485
	}
486

487
	part, err := writer.CreateFormFile("file", filepath.Base(input.ObjectKey))
488
	if err != nil {
489
		return nil, err
490
	}
491

492
	buf := make([]byte, DefaultPutObjectBufferSize)
493
	if _, err := io.CopyBuffer(part, input.Reader, buf); err != nil {
494
		return nil, err
495
	}
496

497
	if err := writer.Close(); err != nil {
498
		return nil, err
499
	}
500

501
	u, err := url.Parse(dfs.endpoint)
502
	if err != nil {
503
		return nil, err
504
	}
505

506
	u.Path = filepath.Join("buckets", input.BucketName, "objects", input.ObjectKey)
507

508
	if strings.HasSuffix(input.ObjectKey, "/") {
509
		u.Path += "/"
510
	}
511

512
	req, err := http.NewRequestWithContext(ctx, http.MethodPut, u.String(), body)
513
	if err != nil {
514
		return nil, err
515
	}
516
	req.Header.Add(headers.ContentType, writer.FormDataContentType())
517

518
	return req, nil
519
}
520

521
// PutObjectWithContext puts data of object.
522
func (dfs *dfstore) PutObjectWithContext(ctx context.Context, input *PutObjectInput) error {
523
	req, err := dfs.PutObjectRequestWithContext(ctx, input)
524
	if err != nil {
525
		return err
526
	}
527

528
	resp, err := http.DefaultClient.Do(req)
529
	if err != nil {
530
		return err
531
	}
532
	defer resp.Body.Close()
533

534
	if resp.StatusCode/100 != 2 {
535
		return fmt.Errorf("bad response status %s", resp.Status)
536
	}
537

538
	return nil
539
}
540

541
// CopyObjectInput is used to construct request of copying object.
542
type CopyObjectInput struct {
543
	// BucketName is bucket name.
544
	BucketName string
545

546
	// SourceObjectKey is the key of object to be copied.
547
	SourceObjectKey string
548

549
	// DestinationObjectKey is the object key of the destination.
550
	DestinationObjectKey string
551
}
552

553
// Validate validates CopyObjectInput fields.
554
func (i *CopyObjectInput) Validate() error {
555
	if i.BucketName == "" {
556
		return errors.New("invalid BucketName")
557
	}
558

559
	if i.SourceObjectKey == "" {
560
		return errors.New("invalid SourceObjectKey")
561
	}
562

563
	if i.DestinationObjectKey == "" {
564
		return errors.New("invalid DestinationObjectKey")
565
	}
566

567
	return nil
568
}
569

570
// CopyObjectWithContext copy object from source to destination.
571
func (dfs *dfstore) CopyObjectWithContext(ctx context.Context, input *CopyObjectInput) error {
572
	req, err := dfs.CopyObjectRequestWithContext(ctx, input)
573
	if err != nil {
574
		return err
575
	}
576

577
	resp, err := http.DefaultClient.Do(req)
578
	if err != nil {
579
		return err
580
	}
581
	defer resp.Body.Close()
582

583
	if resp.StatusCode/100 != 2 {
584
		return fmt.Errorf("bad response status %s", resp.Status)
585
	}
586

587
	return nil
588
}
589

590
// CopyObjectRequestWithContext returns *http.Request of copying object.
591
func (dfs *dfstore) CopyObjectRequestWithContext(ctx context.Context, input *CopyObjectInput) (*http.Request, error) {
592
	if err := input.Validate(); err != nil {
593
		return nil, err
594
	}
595

596
	body := &bytes.Buffer{}
597
	writer := multipart.NewWriter(body)
598

599
	if err := writer.WriteField("source_object_key", input.SourceObjectKey); err != nil {
600
		return nil, err
601
	}
602

603
	if err := writer.Close(); err != nil {
604
		return nil, err
605
	}
606

607
	u, err := url.Parse(dfs.endpoint)
608
	if err != nil {
609
		return nil, err
610
	}
611

612
	u.Path = filepath.Join("buckets", input.BucketName, "objects", input.DestinationObjectKey)
613

614
	query := u.Query()
615

616
	u.RawQuery = query.Encode()
617

618
	req, err := http.NewRequestWithContext(ctx, http.MethodPut, u.String(), body)
619
	if err != nil {
620
		return nil, err
621
	}
622

623
	req.Header.Add(headers.ContentType, writer.FormDataContentType())
624
	req.Header.Add(config.HeaderDragonflyObjectOperation, fmt.Sprint(objectstorage.CopyOperation))
625
	return req, nil
626
}
627

628
// CreateBucketInput is used to construct request of creating bucket.
629
type CreateBucketInput struct {
630
	// BucketName is bucket name.
631
	BucketName string
632
}
633

634
// Validate validates CreateBucketInput fields.
635
func (i *CreateBucketInput) Validate() error {
636
	if i.BucketName == "" {
637
		return errors.New("invalid BucketName")
638
	}
639

640
	return nil
641
}
642

643
// CreateBucketWithContext creates bucket.
644
func (dfs *dfstore) CreateBucketWithContext(ctx context.Context, input *CreateBucketInput) error {
645
	req, err := dfs.CreateBucketRequestWithContext(ctx, input)
646
	if err != nil {
647
		return err
648
	}
649

650
	resp, err := http.DefaultClient.Do(req)
651
	if err != nil {
652
		return err
653
	}
654
	defer resp.Body.Close()
655

656
	if resp.StatusCode/100 != 2 {
657
		return fmt.Errorf("bad response status %s", resp.Status)
658
	}
659

660
	return nil
661
}
662

663
// CreateBucketRequestWithContext returns *http.Request of creating bucket.
664
func (dfs *dfstore) CreateBucketRequestWithContext(ctx context.Context, input *CreateBucketInput) (*http.Request, error) {
665
	if err := input.Validate(); err != nil {
666
		return nil, err
667
	}
668

669
	u, err := url.Parse(dfs.endpoint)
670
	if err != nil {
671
		return nil, err
672
	}
673

674
	u.Path = filepath.Join("buckets", input.BucketName)
675

676
	query := u.Query()
677

678
	u.RawQuery = query.Encode()
679

680
	return http.NewRequestWithContext(ctx, http.MethodPost, u.String(), nil)
681
}
682

683
// DeleteObjectInput is used to construct request of deleting object.
684
type DeleteObjectInput struct {
685
	// BucketName is bucket name.
686
	BucketName string
687

688
	// ObjectKey is object key.
689
	ObjectKey string
690
}
691

692
// Validate validates DeleteObjectInput fields.
693
func (i *DeleteObjectInput) Validate() error {
694
	if i.BucketName == "" {
695
		return errors.New("invalid BucketName")
696
	}
697

698
	if i.ObjectKey == "" {
699
		return errors.New("invalid ObjectKey")
700
	}
701

702
	return nil
703
}
704

705
// DeleteObjectRequestWithContext returns *http.Request of deleting object.
706
func (dfs *dfstore) DeleteObjectRequestWithContext(ctx context.Context, input *DeleteObjectInput) (*http.Request, error) {
707
	if err := input.Validate(); err != nil {
708
		return nil, err
709
	}
710

711
	u, err := url.Parse(dfs.endpoint)
712
	if err != nil {
713
		return nil, err
714
	}
715

716
	u.Path = filepath.Join("buckets", input.BucketName, "objects", input.ObjectKey)
717

718
	if strings.HasSuffix(input.ObjectKey, "/") {
719
		u.Path += "/"
720
	}
721

722
	return http.NewRequestWithContext(ctx, http.MethodDelete, u.String(), nil)
723
}
724

725
// DeleteObjectWithContext deletes data of object.
726
func (dfs *dfstore) DeleteObjectWithContext(ctx context.Context, input *DeleteObjectInput) error {
727
	req, err := dfs.DeleteObjectRequestWithContext(ctx, input)
728
	if err != nil {
729
		return err
730
	}
731

732
	resp, err := http.DefaultClient.Do(req)
733
	if err != nil {
734
		return err
735
	}
736
	defer resp.Body.Close()
737

738
	if resp.StatusCode/100 != 2 {
739
		return fmt.Errorf("bad response status %s", resp.Status)
740
	}
741

742
	return nil
743
}
744

745
// IsObjectExistInput is used to construct request of heading object.
746
type IsObjectExistInput struct {
747
	// BucketName is bucket name.
748
	BucketName string
749

750
	// ObjectKey is object key.
751
	ObjectKey string
752
}
753

754
// Validate validates IsObjectExistInput fields.
755
func (i *IsObjectExistInput) Validate() error {
756
	if i.BucketName == "" {
757
		return errors.New("invalid BucketName")
758
	}
759

760
	if i.ObjectKey == "" {
761
		return errors.New("invalid ObjectKey")
762
	}
763

764
	return nil
765
}
766

767
// IsObjectExistRequestWithContext returns *http.Request of heading object.
768
func (dfs *dfstore) IsObjectExistRequestWithContext(ctx context.Context, input *IsObjectExistInput) (*http.Request, error) {
769
	if err := input.Validate(); err != nil {
770
		return nil, err
771
	}
772

773
	u, err := url.Parse(dfs.endpoint)
774
	if err != nil {
775
		return nil, err
776
	}
777

778
	u.Path = filepath.Join("buckets", input.BucketName, "objects", input.ObjectKey)
779

780
	if strings.HasSuffix(input.ObjectKey, "/") {
781
		u.Path += "/"
782
	}
783

784
	return http.NewRequestWithContext(ctx, http.MethodHead, u.String(), nil)
785
}
786

787
// IsObjectExistWithContext returns whether the object exists.
788
func (dfs *dfstore) IsObjectExistWithContext(ctx context.Context, input *IsObjectExistInput) (bool, error) {
789
	req, err := dfs.IsObjectExistRequestWithContext(ctx, input)
790
	if err != nil {
791
		return false, err
792
	}
793

794
	resp, err := http.DefaultClient.Do(req)
795
	if err != nil {
796
		return false, err
797
	}
798
	defer resp.Body.Close()
799

800
	if resp.StatusCode == http.StatusNotFound {
801
		return false, nil
802
	}
803

804
	if resp.StatusCode/100 != 2 {
805
		return false, fmt.Errorf("bad response status %s", resp.Status)
806
	}
807

808
	return true, nil
809
}
810