-
Notifications
You must be signed in to change notification settings - Fork 1.4k
/
object_store.go
364 lines (297 loc) · 10.3 KB
/
object_store.go
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
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
/*
Copyright 2017 the Heptio Ark contributors.
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 plugin
import (
"io"
"time"
"github.com/hashicorp/go-plugin"
"github.com/pkg/errors"
"golang.org/x/net/context"
"google.golang.org/grpc"
"github.com/heptio/velero/pkg/cloudprovider"
proto "github.com/heptio/velero/pkg/plugin/generated"
)
const byteChunkSize = 16384
// ObjectStorePlugin is an implementation of go-plugin's Plugin
// interface with support for gRPC for the cloudprovider/ObjectStore
// interface.
type ObjectStorePlugin struct {
plugin.NetRPCUnsupportedPlugin
*pluginBase
}
// NewObjectStorePlugin construct an ObjectStorePlugin.
func NewObjectStorePlugin(options ...pluginOption) *ObjectStorePlugin {
return &ObjectStorePlugin{
pluginBase: newPluginBase(options...),
}
}
//////////////////////////////////////////////////////////////////////////////
// client code
//////////////////////////////////////////////////////////////////////////////
// GRPCClient returns an ObjectStore gRPC client.
func (p *ObjectStorePlugin) GRPCClient(c *grpc.ClientConn) (interface{}, error) {
return newClientDispenser(p.clientLogger, c, newObjectStoreGRPCClient), nil
}
// ObjectStoreGRPCClient implements the cloudprovider.ObjectStore interface and uses a
// gRPC client to make calls to the plugin server.
type ObjectStoreGRPCClient struct {
*clientBase
grpcClient proto.ObjectStoreClient
}
func newObjectStoreGRPCClient(base *clientBase, clientConn *grpc.ClientConn) interface{} {
return &ObjectStoreGRPCClient{
clientBase: base,
grpcClient: proto.NewObjectStoreClient(clientConn),
}
}
// Init prepares the ObjectStore for usage using the provided map of
// configuration key-value pairs. It returns an error if the ObjectStore
// cannot be initialized from the provided config.
func (c *ObjectStoreGRPCClient) Init(config map[string]string) error {
_, err := c.grpcClient.Init(context.Background(), &proto.InitRequest{Plugin: c.plugin, Config: config})
return err
}
// PutObject creates a new object using the data in body within the specified
// object storage bucket with the given key.
func (c *ObjectStoreGRPCClient) PutObject(bucket, key string, body io.Reader) error {
stream, err := c.grpcClient.PutObject(context.Background())
if err != nil {
return err
}
// read from the provider io.Reader into chunks, and send each one over
// the gRPC stream
chunk := make([]byte, byteChunkSize)
for {
n, err := body.Read(chunk)
if err == io.EOF {
_, resErr := stream.CloseAndRecv()
return resErr
}
if err != nil {
stream.CloseSend()
return err
}
if err := stream.Send(&proto.PutObjectRequest{Plugin: c.plugin, Bucket: bucket, Key: key, Body: chunk[0:n]}); err != nil {
return err
}
}
}
// GetObject retrieves the object with the given key from the specified
// bucket in object storage.
func (c *ObjectStoreGRPCClient) GetObject(bucket, key string) (io.ReadCloser, error) {
stream, err := c.grpcClient.GetObject(context.Background(), &proto.GetObjectRequest{Plugin: c.plugin, Bucket: bucket, Key: key})
if err != nil {
return nil, err
}
receive := func() ([]byte, error) {
data, err := stream.Recv()
if err != nil {
return nil, err
}
return data.Data, nil
}
close := func() error {
return stream.CloseSend()
}
return &StreamReadCloser{receive: receive, close: close}, nil
}
// ListCommonPrefixes gets a list of all object key prefixes that come
// after the provided prefix and before the provided delimiter (this is
// often used to simulate a directory hierarchy in object storage).
func (c *ObjectStoreGRPCClient) ListCommonPrefixes(bucket, prefix, delimiter string) ([]string, error) {
req := &proto.ListCommonPrefixesRequest{
Plugin: c.plugin,
Bucket: bucket,
Prefix: prefix,
Delimiter: delimiter,
}
res, err := c.grpcClient.ListCommonPrefixes(context.Background(), req)
if err != nil {
return nil, err
}
return res.Prefixes, nil
}
// ListObjects gets a list of all objects in bucket that have the same prefix.
func (c *ObjectStoreGRPCClient) ListObjects(bucket, prefix string) ([]string, error) {
res, err := c.grpcClient.ListObjects(context.Background(), &proto.ListObjectsRequest{Plugin: c.plugin, Bucket: bucket, Prefix: prefix})
if err != nil {
return nil, err
}
return res.Keys, nil
}
// DeleteObject removes object with the specified key from the given
// bucket.
func (c *ObjectStoreGRPCClient) DeleteObject(bucket, key string) error {
_, err := c.grpcClient.DeleteObject(context.Background(), &proto.DeleteObjectRequest{Plugin: c.plugin, Bucket: bucket, Key: key})
return err
}
// CreateSignedURL creates a pre-signed URL for the given bucket and key that expires after ttl.
func (c *ObjectStoreGRPCClient) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
res, err := c.grpcClient.CreateSignedURL(context.Background(), &proto.CreateSignedURLRequest{
Plugin: c.plugin,
Bucket: bucket,
Key: key,
Ttl: int64(ttl),
})
if err != nil {
return "", nil
}
return res.Url, nil
}
//////////////////////////////////////////////////////////////////////////////
// server code
//////////////////////////////////////////////////////////////////////////////
// GRPCServer registers an ObjectStore gRPC server.
func (p *ObjectStorePlugin) GRPCServer(s *grpc.Server) error {
proto.RegisterObjectStoreServer(s, &ObjectStoreGRPCServer{mux: p.serverMux})
return nil
}
// ObjectStoreGRPCServer implements the proto-generated ObjectStoreServer interface, and accepts
// gRPC calls and forwards them to an implementation of the pluggable interface.
type ObjectStoreGRPCServer struct {
mux *serverMux
}
func (s *ObjectStoreGRPCServer) getImpl(name string) (cloudprovider.ObjectStore, error) {
impl, err := s.mux.getHandler(name)
if err != nil {
return nil, err
}
itemAction, ok := impl.(cloudprovider.ObjectStore)
if !ok {
return nil, errors.Errorf("%T is not an object store", impl)
}
return itemAction, nil
}
// Init prepares the ObjectStore for usage using the provided map of
// configuration key-value pairs. It returns an error if the ObjectStore
// cannot be initialized from the provided config.
func (s *ObjectStoreGRPCServer) Init(ctx context.Context, req *proto.InitRequest) (*proto.Empty, error) {
impl, err := s.getImpl(req.Plugin)
if err != nil {
return nil, err
}
if err := impl.Init(req.Config); err != nil {
return nil, err
}
return &proto.Empty{}, nil
}
// PutObject creates a new object using the data in body within the specified
// object storage bucket with the given key.
func (s *ObjectStoreGRPCServer) PutObject(stream proto.ObjectStore_PutObjectServer) error {
// we need to read the first chunk ahead of time to get the bucket and key;
// in our receive method, we'll use `first` on the first call
firstChunk, err := stream.Recv()
if err != nil {
return err
}
impl, err := s.getImpl(firstChunk.Plugin)
if err != nil {
return err
}
bucket := firstChunk.Bucket
key := firstChunk.Key
receive := func() ([]byte, error) {
if firstChunk != nil {
res := firstChunk.Body
firstChunk = nil
return res, nil
}
data, err := stream.Recv()
if err != nil {
return nil, err
}
return data.Body, nil
}
close := func() error {
return nil
}
if err := impl.PutObject(bucket, key, &StreamReadCloser{receive: receive, close: close}); err != nil {
return err
}
return stream.SendAndClose(&proto.Empty{})
}
// GetObject retrieves the object with the given key from the specified
// bucket in object storage.
func (s *ObjectStoreGRPCServer) GetObject(req *proto.GetObjectRequest, stream proto.ObjectStore_GetObjectServer) error {
impl, err := s.getImpl(req.Plugin)
if err != nil {
return err
}
rdr, err := impl.GetObject(req.Bucket, req.Key)
if err != nil {
return err
}
chunk := make([]byte, byteChunkSize)
for {
n, err := rdr.Read(chunk)
if err != nil && err != io.EOF {
return err
}
if n == 0 {
return nil
}
if err := stream.Send(&proto.Bytes{Data: chunk[0:n]}); err != nil {
return err
}
}
}
// ListCommonPrefixes gets a list of all object key prefixes that start with
// the specified prefix and stop at the next instance of the provided delimiter
// (this is often used to simulate a directory hierarchy in object storage).
func (s *ObjectStoreGRPCServer) ListCommonPrefixes(ctx context.Context, req *proto.ListCommonPrefixesRequest) (*proto.ListCommonPrefixesResponse, error) {
impl, err := s.getImpl(req.Plugin)
if err != nil {
return nil, err
}
prefixes, err := impl.ListCommonPrefixes(req.Bucket, req.Prefix, req.Delimiter)
if err != nil {
return nil, err
}
return &proto.ListCommonPrefixesResponse{Prefixes: prefixes}, nil
}
// ListObjects gets a list of all objects in bucket that have the same prefix.
func (s *ObjectStoreGRPCServer) ListObjects(ctx context.Context, req *proto.ListObjectsRequest) (*proto.ListObjectsResponse, error) {
impl, err := s.getImpl(req.Plugin)
if err != nil {
return nil, err
}
keys, err := impl.ListObjects(req.Bucket, req.Prefix)
if err != nil {
return nil, err
}
return &proto.ListObjectsResponse{Keys: keys}, nil
}
// DeleteObject removes object with the specified key from the given
// bucket.
func (s *ObjectStoreGRPCServer) DeleteObject(ctx context.Context, req *proto.DeleteObjectRequest) (*proto.Empty, error) {
impl, err := s.getImpl(req.Plugin)
if err != nil {
return nil, err
}
if err := impl.DeleteObject(req.Bucket, req.Key); err != nil {
return nil, err
}
return &proto.Empty{}, nil
}
// CreateSignedURL creates a pre-signed URL for the given bucket and key that expires after ttl.
func (s *ObjectStoreGRPCServer) CreateSignedURL(ctx context.Context, req *proto.CreateSignedURLRequest) (*proto.CreateSignedURLResponse, error) {
impl, err := s.getImpl(req.Plugin)
if err != nil {
return nil, err
}
url, err := impl.CreateSignedURL(req.Bucket, req.Key, time.Duration(req.Ttl))
if err != nil {
return nil, err
}
return &proto.CreateSignedURLResponse{Url: url}, nil
}