1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
package objectstore
import (
"context"
"fmt"
"io"
"io/ioutil"
"net/http"
"gitlab.com/gitlab-org/labkit/mask"
"gitlab.com/gitlab-org/gitlab/workhorse/internal/helper/httptransport"
)
var httpClient = &http.Client{
Transport: httptransport.New(),
}
// Object represents an object on a S3 compatible Object Store service.
// It can be used as io.WriteCloser for uploading an object
type Object struct {
// putURL is a presigned URL for PutObject
putURL string
// deleteURL is a presigned URL for RemoveObject
deleteURL string
putHeaders map[string]string
size int64
etag string
metrics bool
*uploader
}
type StatusCodeError error
// NewObject opens an HTTP connection to Object Store and returns an Object pointer that can be used for uploading.
func NewObject(putURL, deleteURL string, putHeaders map[string]string, size int64) (*Object, error) {
return newObject(putURL, deleteURL, putHeaders, size, true)
}
func newObject(putURL, deleteURL string, putHeaders map[string]string, size int64, metrics bool) (*Object, error) {
o := &Object{
putURL: putURL,
deleteURL: deleteURL,
putHeaders: putHeaders,
size: size,
metrics: metrics,
}
o.uploader = newETagCheckUploader(o, metrics)
return o, nil
}
func (o *Object) Upload(ctx context.Context, r io.Reader) error {
// we should prevent pr.Close() otherwise it may shadow error set with pr.CloseWithError(err)
req, err := http.NewRequest(http.MethodPut, o.putURL, ioutil.NopCloser(r))
if err != nil {
return fmt.Errorf("PUT %q: %v", mask.URL(o.putURL), err)
}
req.ContentLength = o.size
for k, v := range o.putHeaders {
req.Header.Set(k, v)
}
resp, err := httpClient.Do(req)
if err != nil {
return fmt.Errorf("PUT request %q: %v", mask.URL(o.putURL), err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
if o.metrics {
objectStorageUploadRequestsInvalidStatus.Inc()
}
return StatusCodeError(fmt.Errorf("PUT request %v returned: %s", mask.URL(o.putURL), resp.Status))
}
o.etag = extractETag(resp.Header.Get("ETag"))
return nil
}
func (o *Object) ETag() string {
return o.etag
}
func (o *Object) Abort() {
o.Delete()
}
func (o *Object) Delete() {
deleteURL(o.deleteURL)
}
|
