/
repository.go
365 lines (298 loc) · 11.9 KB
/
repository.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
365
package repo
import (
"context"
"fmt"
"sync/atomic"
"time"
"github.com/pkg/errors"
"github.com/kopia/kopia/internal/clock"
"github.com/kopia/kopia/repo/blob"
"github.com/kopia/kopia/repo/blob/throttling"
"github.com/kopia/kopia/repo/content"
"github.com/kopia/kopia/repo/manifest"
"github.com/kopia/kopia/repo/object"
)
// Repository exposes public API of Kopia repository, including objects and manifests.
type Repository interface {
OpenObject(ctx context.Context, id object.ID) (object.Reader, error)
VerifyObject(ctx context.Context, id object.ID) ([]content.ID, error)
GetManifest(ctx context.Context, id manifest.ID, data interface{}) (*manifest.EntryMetadata, error)
FindManifests(ctx context.Context, labels map[string]string) ([]*manifest.EntryMetadata, error)
Time() time.Time
ClientOptions() ClientOptions
NewWriter(ctx context.Context, opt WriteSessionOptions) (context.Context, RepositoryWriter, error)
UpdateDescription(d string)
Refresh(ctx context.Context) error
Close(ctx context.Context) error
}
// RepositoryWriter provides methods to write to a repository.
type RepositoryWriter interface {
Repository
NewObjectWriter(ctx context.Context, opt object.WriterOptions) object.Writer
PutManifest(ctx context.Context, labels map[string]string, payload interface{}) (manifest.ID, error)
DeleteManifest(ctx context.Context, id manifest.ID) error
Flush(ctx context.Context) error
}
// DirectRepository provides additional low-level repository functionality.
type DirectRepository interface {
Repository
ObjectFormat() object.Format
BlobCfg() content.BlobCfgBlob
BlobReader() blob.Reader
ContentReader() content.Reader
IndexBlobs(ctx context.Context, includeInactive bool) ([]content.IndexBlobInfo, error)
Crypter() *content.Crypter
NewDirectWriter(ctx context.Context, opt WriteSessionOptions) (context.Context, DirectRepositoryWriter, error)
AlsoLogToContentLog(ctx context.Context) context.Context
UniqueID() []byte
ConfigFilename() string
DeriveKey(purpose []byte, keyLength int) []byte
Token(password string) (string, error)
Throttler() throttling.SettableThrottler
DisableIndexRefresh()
}
// DirectRepositoryWriter provides low-level write access to the repository.
type DirectRepositoryWriter interface {
RepositoryWriter
DirectRepository
BlobStorage() blob.Storage
ContentManager() *content.WriteManager
SetParameters(ctx context.Context, m content.MutableParameters, blobcfg content.BlobCfgBlob) error
ChangePassword(ctx context.Context, newPassword string) error
}
type directRepositoryParameters struct {
uniqueID []byte
configFile string
cachingOptions content.CachingOptions
cliOpts ClientOptions
timeNow func() time.Time
formatBlob *formatBlob
blobCfgBlob content.BlobCfgBlob
formatEncryptionKey []byte
nextWriterID *int32
}
// directRepository is an implementation of repository that directly manipulates underlying storage.
type directRepository struct {
directRepositoryParameters
blobs blob.Storage
cmgr *content.WriteManager
omgr *object.Manager
mmgr *manifest.Manager
sm *content.SharedManager
throttler throttling.SettableThrottler
closed chan struct{}
}
// DeriveKey derives encryption key of the provided length from the master key.
func (r *directRepository) DeriveKey(purpose []byte, keyLength int) []byte {
if r.cmgr.ContentFormat().EnablePasswordChange {
return deriveKeyFromMasterKey(r.cmgr.ContentFormat().MasterKey, r.uniqueID, purpose, keyLength)
}
// version of kopia <v0.9 had a bug where certain keys were derived directly from
// the password and not from the random master key. This made it impossible to change
// password.
return deriveKeyFromMasterKey(r.formatEncryptionKey, r.uniqueID, purpose, keyLength)
}
// ClientOptions returns client options.
func (r *directRepository) ClientOptions() ClientOptions {
return r.cliOpts
}
// BlobStorage returns the blob storage.
func (r *directRepository) BlobStorage() blob.Storage {
return r.blobs
}
// Throttler returns the blob storage throttler.
func (r *directRepository) Throttler() throttling.SettableThrottler {
return r.throttler
}
// ContentManager returns the content manager.
func (r *directRepository) ContentManager() *content.WriteManager {
return r.cmgr
}
// ConfigFilename returns the name of the configuration file.
func (r *directRepository) ConfigFilename() string {
return r.configFile
}
// Crypter returns a Crypter object.
func (r *directRepository) Crypter() *content.Crypter {
return r.sm.Crypter()
}
// NewObjectWriter creates an object writer.
func (r *directRepository) NewObjectWriter(ctx context.Context, opt object.WriterOptions) object.Writer {
return r.omgr.NewWriter(ctx, opt)
}
// DisableIndexRefresh disables index refresh for the duration of the write session.
func (r *directRepository) DisableIndexRefresh() {
r.cmgr.DisableIndexRefresh()
}
// OpenObject opens the reader for a given object, returns object.ErrNotFound.
func (r *directRepository) OpenObject(ctx context.Context, id object.ID) (object.Reader, error) {
// nolint:wrapcheck
return object.Open(ctx, r.cmgr, id)
}
// VerifyObject verifies that the given object is stored properly in a repository and returns backing content IDs.
func (r *directRepository) VerifyObject(ctx context.Context, id object.ID) ([]content.ID, error) {
// nolint:wrapcheck
return object.VerifyObject(ctx, r.cmgr, id)
}
// GetManifest returns the given manifest data and metadata.
func (r *directRepository) GetManifest(ctx context.Context, id manifest.ID, data interface{}) (*manifest.EntryMetadata, error) {
// nolint:wrapcheck
return r.mmgr.Get(ctx, id, data)
}
// PutManifest saves the given manifest payload with a set of labels.
func (r *directRepository) PutManifest(ctx context.Context, labels map[string]string, payload interface{}) (manifest.ID, error) {
// nolint:wrapcheck
return r.mmgr.Put(ctx, labels, payload)
}
// FindManifests returns metadata for manifests matching given set of labels.
func (r *directRepository) FindManifests(ctx context.Context, labels map[string]string) ([]*manifest.EntryMetadata, error) {
// nolint:wrapcheck
return r.mmgr.Find(ctx, labels)
}
// DeleteManifest deletes the manifest with a given ID.
func (r *directRepository) DeleteManifest(ctx context.Context, id manifest.ID) error {
// nolint:wrapcheck
return r.mmgr.Delete(ctx, id)
}
// ListActiveSessions returns the map of active sessions.
func (r *directRepository) ListActiveSessions(ctx context.Context) (map[content.SessionID]*content.SessionInfo, error) {
// nolint:wrapcheck
return r.cmgr.ListActiveSessions(ctx)
}
// UpdateDescription updates the description of a connected repository.
func (r *directRepository) UpdateDescription(d string) {
r.cliOpts.Description = d
}
// AlsoLogToContentLog returns a context that causes all logs to also be sent to content log.
func (r *directRepository) AlsoLogToContentLog(ctx context.Context) context.Context {
return r.sm.AlsoLogToContentLog(ctx)
}
// NewWriter returns new RepositoryWriter session for repository.
func (r *directRepository) NewWriter(ctx context.Context, opt WriteSessionOptions) (context.Context, RepositoryWriter, error) {
return r.NewDirectWriter(ctx, opt)
}
// NewDirectWriter returns new DirectRepositoryWriter session for repository.
func (r *directRepository) NewDirectWriter(ctx context.Context, opt WriteSessionOptions) (context.Context, DirectRepositoryWriter, error) {
writeManagerID := fmt.Sprintf("writer-%v:%v", atomic.AddInt32(r.nextWriterID, 1), opt.Purpose)
cmgr := content.NewWriteManager(ctx, r.sm, content.SessionOptions{
SessionUser: r.cliOpts.Username,
SessionHost: r.cliOpts.Hostname,
OnUpload: opt.OnUpload,
}, writeManagerID)
mmgr, err := manifest.NewManager(ctx, cmgr, manifest.ManagerOptions{
TimeNow: r.timeNow,
})
if err != nil {
return nil, nil, errors.Wrap(err, "error creating manifest manager")
}
omgr, err := object.NewObjectManager(ctx, cmgr, r.omgr.Format)
if err != nil {
return nil, nil, errors.Wrap(err, "error creating object manager")
}
w := &directRepository{
directRepositoryParameters: r.directRepositoryParameters,
blobs: r.blobs,
cmgr: cmgr,
omgr: omgr,
mmgr: mmgr,
sm: r.sm,
closed: make(chan struct{}),
}
return ctx, w, nil
}
// Close closes the repository and releases all resources.
func (r *directRepository) Close(ctx context.Context) error {
select {
case <-r.closed:
// already closed
return nil
default:
}
// this will release shared manager and MAY release blob.Store (on last outstanding reference).
if err := r.cmgr.Close(ctx); err != nil {
return errors.Wrap(err, "error closing content-addressable storage manager")
}
close(r.closed)
return nil
}
// Flush waits for all in-flight writes to complete.
func (r *directRepository) Flush(ctx context.Context) error {
if err := r.mmgr.Flush(ctx); err != nil {
return errors.Wrap(err, "error flushing manifests")
}
return errors.Wrap(r.cmgr.Flush(ctx), "error flushing contents")
}
// ObjectFormat returns the object format.
func (r *directRepository) ObjectFormat() object.Format {
return r.omgr.Format
}
// UniqueID returns unique repository ID from which many keys and secrets are derived.
func (r *directRepository) UniqueID() []byte {
return r.uniqueID
}
// BlobReader returns the blob reader.
func (r *directRepository) BlobReader() blob.Reader {
return r.blobs
}
// ContentReader returns the content reader.
func (r *directRepository) ContentReader() content.Reader {
return r.cmgr
}
// IndexBlobs returns the index blobs in use.
func (r *directRepository) IndexBlobs(ctx context.Context, includeInactive bool) ([]content.IndexBlobInfo, error) {
// nolint:wrapcheck
return r.cmgr.IndexBlobs(ctx, includeInactive)
}
// Refresh makes external changes visible to repository.
func (r *directRepository) Refresh(ctx context.Context) error {
return errors.Wrap(r.cmgr.Refresh(ctx), "error refreshing content index")
}
// Time returns the current local time for the repo.
func (r *directRepository) Time() time.Time {
return defaultTime(r.timeNow)()
}
func (r *directRepository) BlobCfg() content.BlobCfgBlob {
return r.directRepositoryParameters.blobCfgBlob
}
// WriteSessionOptions describes options for a write session.
type WriteSessionOptions struct {
Purpose string
FlushOnFailure bool // whether to flush regardless of write session result.
OnUpload func(int64) // function to invoke after completing each upload in the session.
}
// WriteSession executes the provided callback in a repository writer created for the purpose and flushes writes.
func WriteSession(ctx context.Context, r Repository, opt WriteSessionOptions, cb func(ctx context.Context, w RepositoryWriter) error) error {
ctx, w, err := r.NewWriter(ctx, opt)
if err != nil {
return errors.Wrap(err, "unable to create writer")
}
return handleWriteSessionResult(ctx, w, opt, cb(ctx, w))
}
// DirectWriteSession executes the provided callback in a DirectRepositoryWriter created for the purpose and flushes writes.
func DirectWriteSession(ctx context.Context, r DirectRepository, opt WriteSessionOptions, cb func(ctx context.Context, dw DirectRepositoryWriter) error) error {
ctx, w, err := r.NewDirectWriter(ctx, opt)
if err != nil {
return errors.Wrap(err, "unable to create direct writer")
}
return handleWriteSessionResult(ctx, w, opt, cb(ctx, w))
}
func handleWriteSessionResult(ctx context.Context, w RepositoryWriter, opt WriteSessionOptions, resultErr error) error {
defer func() {
if err := w.Close(ctx); err != nil {
log(ctx).Errorf("error closing writer: %v", err)
}
}()
if resultErr == nil || opt.FlushOnFailure {
if err := w.Flush(ctx); err != nil {
return errors.Wrap(err, "error flushing writer")
}
}
return resultErr
}
func defaultTime(f func() time.Time) func() time.Time {
if f != nil {
return f
}
return clock.Now
}
var _ DirectRepositoryWriter = (*directRepository)(nil)