From d8b2ba4c247a19f4bdcaadce59f457ca826dea60 Mon Sep 17 00:00:00 2001 From: Jarek Kowalski Date: Tue, 18 Jul 2017 12:37:44 +0200 Subject: [PATCH] moved most methods from Repository to casManager --- cli/command_backups.go | 2 +- client/connection.go | 2 +- repo/blocks.go | 4 ++-- repo/cas_manager.go | 45 ++++++++++++++++------------------------ repo/cas_manager_test.go | 16 +++++++------- repo/object_reader.go | 2 +- repo/object_writer.go | 2 +- repo/repository.go | 40 +++++++++++++++++++++++++++++++++++ snapshot/manifest.go | 1 - snapshot/upload_test.go | 2 +- 10 files changed, 73 insertions(+), 43 deletions(-) create mode 100644 repo/repository.go diff --git a/cli/command_backups.go b/cli/command_backups.go index 759203e08..aa93057b9 100644 --- a/cli/command_backups.go +++ b/cli/command_backups.go @@ -95,7 +95,7 @@ func runBackupsCommand(context *kingpin.ParseContext) error { if count < *maxResultsPerPath { fmt.Printf( " %v%v %v %10v %v\n", - m.Handle, + m.RootObjectID, relPath, m.StartTime.Format("2006-01-02 15:04:05 MST"), units.BytesString(m.Stats.TotalFileSize), diff --git a/client/connection.go b/client/connection.go index 35ba7c127..c572a04b7 100644 --- a/client/connection.go +++ b/client/connection.go @@ -106,7 +106,7 @@ func Open(ctx context.Context, configFile string, options *Options) (*Connection } } - conn.Repository, err = repo.New(repositoryStorage, conn.Vault.RepoConfig.Format) + conn.Repository, err = repo.NewRepository(repositoryStorage, conn.Vault.RepoConfig.Format) if err != nil { vaultStorage.Close() repositoryStorage.Close() diff --git a/repo/blocks.go b/repo/blocks.go index 477667404..2ae5e66d3 100644 --- a/repo/blocks.go +++ b/repo/blocks.go @@ -1,7 +1,7 @@ package repo // GetStorageBlocks returns the list of storage blocks used by the specified object ID. -func (r *Repository) GetStorageBlocks(oid ObjectID) ([]string, error) { +func (r *casManager) GetStorageBlocks(oid ObjectID) ([]string, error) { result := map[string]bool{} if err := r.addStorageBlocks(result, oid); err != nil { return nil, err @@ -14,7 +14,7 @@ func (r *Repository) GetStorageBlocks(oid ObjectID) ([]string, error) { return b, nil } -func (r *Repository) addStorageBlocks(result map[string]bool, oid ObjectID) error { +func (r *casManager) addStorageBlocks(result map[string]bool, oid ObjectID) error { if oid.Section != nil { return r.addStorageBlocks(result, oid.Section.Base) } diff --git a/repo/cas_manager.go b/repo/cas_manager.go index 1197baa0f..836e5a6ef 100644 --- a/repo/cas_manager.go +++ b/repo/cas_manager.go @@ -22,8 +22,8 @@ type ObjectReader interface { Length() int64 } -// Repository implements a content-addressable storage on top of blob storage. -type Repository struct { +// casManager implements a content-addressable storage on top of blob storage. +type casManager struct { Stats Stats // vital statistics Storage blob.Storage // underlying blob storage @@ -38,24 +38,15 @@ type Repository struct { } // Close closes the connection to the underlying blob storage and releases any resources. -func (r *Repository) Close() error { - r.Flush() - if err := r.Storage.Close(); err != nil { - return err - } +func (r *casManager) Close() error { + r.writeBack.flush() r.bufferManager.close() return nil } -// Flush waits for all in-flight writes to complete. -func (r *Repository) Flush() error { - r.writeBack.flush() - return nil -} - // NewWriter creates an ObjectWriter for writing to the repository. -func (r *Repository) NewWriter(options ...WriterOption) ObjectWriter { +func (r *casManager) NewWriter(options ...WriterOption) ObjectWriter { result := &objectWriter{ repo: r, blockTracker: &blockTracker{}, @@ -70,12 +61,12 @@ func (r *Repository) NewWriter(options ...WriterOption) ObjectWriter { } // Open creates new ObjectReader for reading given object from a repository. -func (r *Repository) Open(objectID ObjectID) (ObjectReader, error) { +func (r *casManager) Open(objectID ObjectID) (ObjectReader, error) { // log.Printf("Repository::Open %v", objectID.String()) // defer log.Printf("finished Repository::Open() %v", objectID.String()) // Flush any pending writes. - r.Flush() + r.writeBack.flush() if objectID.Section != nil { baseReader, err := r.Open(objectID.Section.Base) @@ -111,32 +102,32 @@ func (r *Repository) Open(objectID ObjectID) (ObjectReader, error) { } // RepositoryOption controls the behavior of Repository. -type RepositoryOption func(o *Repository) +type RepositoryOption func(o *casManager) // WriteBack is an RepositoryOption that enables asynchronous writes to the storage using the pool // of goroutines. func WriteBack(writeBackWorkers int) RepositoryOption { - return func(o *Repository) { + return func(o *casManager) { o.writeBack.workers = writeBackWorkers } } // EnableLogging is an RepositoryOption that causes all storage access to be logged. func EnableLogging(options ...logging.Option) RepositoryOption { - return func(o *Repository) { + return func(o *casManager) { o.Storage = logging.NewWrapper(o.Storage, options...) } } -// New creates a Repository with the specified storage, format and options. -func New(s blob.Storage, f *Format, options ...RepositoryOption) (*Repository, error) { +// newCASManager creates a casManager with the specified storage, format and options. +func newCASManager(s blob.Storage, f *Format, options ...RepositoryOption) (*casManager, error) { if err := f.Validate(); err != nil { return nil, err } sf := SupportedFormats[f.ObjectFormat] - r := &Repository{ + r := &casManager{ Storage: s, format: *f, } @@ -177,7 +168,7 @@ func New(s blob.Storage, f *Format, options ...RepositoryOption) (*Repository, e // hashEncryptAndWriteMaybeAsync computes hash of a given buffer, optionally encrypts and writes it to storage. // The write is not guaranteed to complete synchronously in case write-back is used, but by the time // Repository.Close() returns all writes are guaranteed be over. -func (r *Repository) hashEncryptAndWriteMaybeAsync(buffer *bytes.Buffer, prefix string) (ObjectID, error) { +func (r *casManager) hashEncryptAndWriteMaybeAsync(buffer *bytes.Buffer, prefix string) (ObjectID, error) { var data []byte if buffer != nil { data = buffer.Bytes() @@ -211,7 +202,7 @@ func (r *Repository) hashEncryptAndWriteMaybeAsync(buffer *bytes.Buffer, prefix return r.encryptAndMaybeWrite(objectID, buffer, prefix) } -func (r *Repository) encryptAndMaybeWrite(objectID ObjectID, buffer *bytes.Buffer, prefix string) (ObjectID, error) { +func (r *casManager) encryptAndMaybeWrite(objectID ObjectID, buffer *bytes.Buffer, prefix string) (ObjectID, error) { defer r.bufferManager.returnBuffer(buffer) var data []byte @@ -250,7 +241,7 @@ func (r *Repository) encryptAndMaybeWrite(objectID ObjectID, buffer *bytes.Buffe return objectID, nil } -func (r *Repository) flattenListChunk(rawReader io.Reader) ([]indirectObjectEntry, error) { +func (r *casManager) flattenListChunk(rawReader io.Reader) ([]indirectObjectEntry, error) { pr, err := jsonstream.NewReader(bufio.NewReader(rawReader), indirectStreamType) if err != nil { return nil, err @@ -284,7 +275,7 @@ func removeIndirection(o ObjectID) ObjectID { return o } -func (r *Repository) newRawReader(objectID ObjectID) (ObjectReader, error) { +func (r *casManager) newRawReader(objectID ObjectID) (ObjectReader, error) { if objectID.BinaryContent != nil { return newObjectReaderWithData(objectID.BinaryContent), nil } @@ -317,7 +308,7 @@ func (r *Repository) newRawReader(objectID ObjectID) (ObjectReader, error) { return newObjectReaderWithData(payload), nil } -func (r *Repository) verifyChecksum(data []byte, blockID string) error { +func (r *casManager) verifyChecksum(data []byte, blockID string) error { expected := r.formatter.ComputeObjectID(data) if !strings.HasSuffix(blockID, expected.StorageBlock) { atomic.AddInt32(&r.Stats.InvalidBlocks, 1) diff --git a/repo/cas_manager_test.go b/repo/cas_manager_test.go index 9c939768d..ed1a8737b 100644 --- a/repo/cas_manager_test.go +++ b/repo/cas_manager_test.go @@ -44,11 +44,11 @@ func getMd5LObjectID(data []byte) string { return fmt.Sprintf("L%s", getMd5Digest(data)) } -func setupTest(t *testing.T) (data map[string][]byte, repo *Repository) { +func setupTest(t *testing.T) (data map[string][]byte, repo *casManager) { data = map[string][]byte{} st := storagetesting.NewMapStorage(data) - repo, err := New(st, testFormat(), WriteBack(5)) + repo, err := newCASManager(st, testFormat(), WriteBack(5)) if err != nil { t.Errorf("cannot create manager: %v", err) } @@ -88,7 +88,7 @@ func TestWriters(t *testing.T) { continue } - repo.Flush() + repo.writeBack.flush() if !objectIDsEqual(result, c.objectID) { t.Errorf("incorrect result for %v, expected: %v got: %v %#v", c.data, c.objectID.String(), result.String(), result.BinaryContent) @@ -125,7 +125,7 @@ func TestWriterCompleteChunkInTwoWrites(t *testing.T) { } } -func verifyIndirectBlock(t *testing.T, r *Repository, oid ObjectID) { +func verifyIndirectBlock(t *testing.T, r *casManager, oid ObjectID) { for level := int32(0); level < oid.Indirect; level++ { direct := oid direct.Indirect = level @@ -178,7 +178,7 @@ func TestIndirection(t *testing.T) { writer := repo.NewWriter() writer.Write(contentBytes) result, err := writer.Result(false) - repo.Flush() + repo.writeBack.flush() if err != nil { t.Errorf("error getting writer results: %v", err) } @@ -211,7 +211,7 @@ func TestHMAC(t *testing.T) { s := testFormat() s.ObjectFormat = "TESTONLY_MD5" - repo, err := New(storagetesting.NewMapStorage(data), s) + repo, err := newCASManager(storagetesting.NewMapStorage(data), s) if err != nil { t.Errorf("cannot create manager: %v", err) } @@ -342,7 +342,7 @@ func TestEndToEndReadAndSeek(t *testing.T) { } } -func verify(t *testing.T, repo *Repository, objectID ObjectID, expectedData []byte, testCaseID string) { +func verify(t *testing.T, repo *casManager, objectID ObjectID, expectedData []byte, testCaseID string) { reader, err := repo.Open(objectID) if err != nil { t.Errorf("cannot get reader for %v: %v", testCaseID, err) @@ -422,7 +422,7 @@ func TestFormats(t *testing.T) { data := map[string][]byte{} st := storagetesting.NewMapStorage(data) - repo, err := New(st, c.format) + repo, err := newCASManager(st, c.format) if err != nil { t.Errorf("cannot create manager: %v", err) continue diff --git a/repo/object_reader.go b/repo/object_reader.go index e98d18517..6f87f6160 100644 --- a/repo/object_reader.go +++ b/repo/object_reader.go @@ -10,7 +10,7 @@ func (i *indirectObjectEntry) endOffset() int64 { } type objectReader struct { - repo *Repository + repo *casManager seekTable []indirectObjectEntry diff --git a/repo/object_writer.go b/repo/object_writer.go index 64f295320..c40159f9b 100644 --- a/repo/object_writer.go +++ b/repo/object_writer.go @@ -37,7 +37,7 @@ func (t *blockTracker) blockIDs() []string { } type objectWriter struct { - repo *Repository + repo *casManager buffer *bytes.Buffer totalLength int64 diff --git a/repo/repository.go b/repo/repository.go new file mode 100644 index 000000000..04662634c --- /dev/null +++ b/repo/repository.go @@ -0,0 +1,40 @@ +package repo + +import ( + "github.com/kopia/kopia/blob" +) + +// Repository represents storage where both content-addressable and user-addressable data is kept. +type Repository struct { + *casManager + Storage blob.Storage +} + +// Close closes the repository and releases all resources. +func (r *Repository) Close() error { + if err := r.casManager.Close(); err != nil { + return err + } + if err := r.Storage.Close(); err != nil { + return err + } + return nil +} + +// Flush waits for all in-flight writes to complete. +func (r *Repository) Flush() error { + r.casManager.writeBack.flush() + return nil +} + +// NewRepository initializes a new repository with a given format. +func NewRepository(s blob.Storage, f *Format, options ...RepositoryOption) (*Repository, error) { + cm, err := newCASManager(s, f, options...) + if err != nil { + return nil, err + } + return &Repository{ + Storage: s, + casManager: cm, + }, nil +} diff --git a/snapshot/manifest.go b/snapshot/manifest.go index ee7bebcee..56edf5fcf 100644 --- a/snapshot/manifest.go +++ b/snapshot/manifest.go @@ -14,7 +14,6 @@ type Manifest struct { StartTime time.Time `json:"startTime"` EndTime time.Time `json:"endTime"` - Handle string `json:"handle"` RootObjectID repo.ObjectID `json:"root"` HashCacheID repo.ObjectID `json:"hashCache"` diff --git a/snapshot/upload_test.go b/snapshot/upload_test.go index 9130e2afe..c99b40dfa 100644 --- a/snapshot/upload_test.go +++ b/snapshot/upload_test.go @@ -51,7 +51,7 @@ func newUploadTestHarness() *uploadTestHarness { MaxInlineContentLength: 0, } - repo, err := repo.New(storage, &format) + repo, err := repo.NewRepository(storage, &format) if err != nil { panic("unable to create repository: " + err.Error()) }