aboutsummaryrefslogtreecommitdiffstats
path: root/vendor/github.com/minio/minio-go/v7/api-putobject-snowball.go
blob: eb4da41471b83db765fb818fe97363497756c653 (plain) (blame)
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
/*
 * MinIO Go Library for Amazon S3 Compatible Cloud Storage
 * Copyright 2021 MinIO, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package minio

import (
	"archive/tar"
	"bufio"
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/klauspost/compress/s2"
)

// SnowballOptions contains options for PutObjectsSnowball calls.
type SnowballOptions struct {
	// Opts is options applied to all objects.
	Opts PutObjectOptions

	// Processing options:

	// InMemory specifies that all objects should be collected in memory
	// before they are uploaded.
	// If false a temporary file will be created.
	InMemory bool

	// Compress enabled content compression before upload.
	// Compression will typically reduce memory and network usage,
	// Compression can safely be enabled with MinIO hosts.
	Compress bool

	// SkipErrs if enabled will skip any errors while reading the
	// object content while creating the snowball archive
	SkipErrs bool
}

// SnowballObject contains information about a single object to be added to the snowball.
type SnowballObject struct {
	// Key is the destination key, including prefix.
	Key string

	// Size is the content size of this object.
	Size int64

	// Modtime to apply to the object.
	// If Modtime is the zero value current time will be used.
	ModTime time.Time

	// Content of the object.
	// Exactly 'Size' number of bytes must be provided.
	Content io.Reader

	// VersionID of the object; if empty, a new versionID will be generated
	VersionID string

	// Headers contains more options for this object upload, the same as you
	// would include in a regular PutObject operation, such as user metadata
	// and content-disposition, expires, ..
	Headers http.Header

	// Close will be called when an object has finished processing.
	// Note that if PutObjectsSnowball returns because of an error,
	// objects not consumed from the input will NOT have been closed.
	// Leave as nil for no callback.
	Close func()
}

type nopReadSeekCloser struct {
	io.ReadSeeker
}

func (n nopReadSeekCloser) Close() error {
	return nil
}

// This is available as io.ReadSeekCloser from go1.16
type readSeekCloser interface {
	io.Reader
	io.Closer
	io.Seeker
}

// PutObjectsSnowball will put multiple objects with a single put call.
// A (compressed) TAR file will be created which will contain multiple objects.
// The key for each object will be used for the destination in the specified bucket.
// Total size should be < 5TB.
// This function blocks until 'objs' is closed and the content has been uploaded.
func (c Client) PutObjectsSnowball(ctx context.Context, bucketName string, opts SnowballOptions, objs <-chan SnowballObject) (err error) {
	err = opts.Opts.validate()
	if err != nil {
		return err
	}
	var tmpWriter io.Writer
	var getTmpReader func() (rc readSeekCloser, sz int64, err error)
	if opts.InMemory {
		b := bytes.NewBuffer(nil)
		tmpWriter = b
		getTmpReader = func() (readSeekCloser, int64, error) {
			return nopReadSeekCloser{bytes.NewReader(b.Bytes())}, int64(b.Len()), nil
		}
	} else {
		f, err := os.CreateTemp("", "s3-putsnowballobjects-*")
		if err != nil {
			return err
		}
		name := f.Name()
		tmpWriter = f
		var once sync.Once
		defer once.Do(func() {
			f.Close()
		})
		defer os.Remove(name)
		getTmpReader = func() (readSeekCloser, int64, error) {
			once.Do(func() {
				f.Close()
			})
			f, err := os.Open(name)
			if err != nil {
				return nil, 0, err
			}
			st, err := f.Stat()
			if err != nil {
				return nil, 0, err
			}
			return f, st.Size(), nil
		}
	}
	flush := func() error { return nil }
	if !opts.Compress {
		if !opts.InMemory {
			// Insert buffer for writes.
			buf := bufio.NewWriterSize(tmpWriter, 1<<20)
			flush = buf.Flush
			tmpWriter = buf
		}
	} else {
		s2c := s2.NewWriter(tmpWriter, s2.WriterBetterCompression())
		flush = s2c.Close
		defer s2c.Close()
		tmpWriter = s2c
	}
	t := tar.NewWriter(tmpWriter)

objectLoop:
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case obj, ok := <-objs:
			if !ok {
				break objectLoop
			}

			closeObj := func() {}
			if obj.Close != nil {
				closeObj = obj.Close
			}

			// Trim accidental slash prefix.
			obj.Key = strings.TrimPrefix(obj.Key, "/")
			header := tar.Header{
				Typeflag: tar.TypeReg,
				Name:     obj.Key,
				Size:     obj.Size,
				ModTime:  obj.ModTime,
				Format:   tar.FormatPAX,
			}
			if header.ModTime.IsZero() {
				header.ModTime = time.Now().UTC()
			}

			header.PAXRecords = make(map[string]string)
			if obj.VersionID != "" {
				header.PAXRecords["minio.versionId"] = obj.VersionID
			}
			for k, vals := range obj.Headers {
				header.PAXRecords["minio.metadata."+k] = strings.Join(vals, ",")
			}

			if err := t.WriteHeader(&header); err != nil {
				closeObj()
				return err
			}
			n, err := io.Copy(t, obj.Content)
			if err != nil {
				closeObj()
				if opts.SkipErrs {
					continue
				}
				return err
			}
			if n != obj.Size {
				closeObj()
				if opts.SkipErrs {
					continue
				}
				return io.ErrUnexpectedEOF
			}
			closeObj()
		}
	}
	// Flush tar
	err = t.Flush()
	if err != nil {
		return err
	}
	// Flush compression
	err = flush()
	if err != nil {
		return err
	}
	if opts.Opts.UserMetadata == nil {
		opts.Opts.UserMetadata = map[string]string{}
	}
	opts.Opts.UserMetadata["X-Amz-Meta-Snowball-Auto-Extract"] = "true"
	opts.Opts.DisableMultipart = true
	rc, sz, err := getTmpReader()
	if err != nil {
		return err
	}
	defer rc.Close()
	rand := c.random.Uint64()
	_, err = c.PutObject(ctx, bucketName, fmt.Sprintf("snowball-upload-%x.tar", rand), rc, sz, opts.Opts)
	return err
}