mirror/kopia
1
0 Fork 0
mirror of https://github.com/kopia/kopia.git synced 2026-10-23 06:38:42 -04:00
Code Issues Packages Projects Releases Wiki Activity

reorganized blob storage package - renamed to storage and put gcs/filesystem/logging into subpackages

Browse Source
This commit is contained in:
Jarek Kowalski committed 2016-07-13 18:50:31 -07:00
1 parent 141dfd3c62
commit 0be422f897
31 files changed
+287 -282
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Makefile
+1 -1
View File
@@ -14,7 +14,7 @@ lint:
golint github.com/kopia/kopia/...
vet:
go vet github.com/kopia/kopia/...
go tool vet -all .
deps:
go get -u -t -v github.com/kopia/kopia/...
blob/doc.go
-2
View File
@@ -1,2 +0,0 @@
// Package blob implements simple storage of immutable, unstructured binary large objects (BLOBs).
package blob
cmd/kopia/command_create.go
+3 -3
View File
@@ -7,8 +7,8 @@ import (
"gopkg.in/alecthomas/kingpin.v2"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/repo"
"github.com/kopia/kopia/storage"
"github.com/kopia/kopia/vault"
)
@@ -59,7 +59,7 @@ func repositoryFormat() (*repo.Format, error) {
return f, nil
}
func openStorageAndEnsureEmpty(url string) (blob.Storage, error) {
func openStorageAndEnsureEmpty(url string) (storage.Storage, error) {
s, err := newStorageFromURL(url)
if err != nil {
return nil, err
@@ -83,7 +83,7 @@ func runCreateCommand(context *kingpin.ParseContext) error {
return fmt.Errorf("unable to get vault storage: %v", err)
}
var repositoryStorage blob.Storage
var repositoryStorage storage.Storage
if *createCommandRepository == "colocated" {
repositoryStorage = vaultStorage
cmd/kopia/urls.go
+10 -8
View File
@@ -7,10 +7,12 @@ import (
"strconv"
"strings"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/storage"
fsstorage "github.com/kopia/kopia/storage/filesystem"
gcsstorage "github.com/kopia/kopia/storage/gcs"
)
func newStorageFromURL(urlString string) (blob.Storage, error) {
func newStorageFromURL(urlString string) (storage.Storage, error) {
if strings.HasPrefix(urlString, "/") {
urlString = "file://" + urlString
}
@@ -22,26 +24,26 @@ func newStorageFromURL(urlString string) (blob.Storage, error) {
switch u.Scheme {
case "file":
var fso blob.FSStorageOptions
var fso fsstorage.Options
if err := parseFilesystemURL(&fso, u); err != nil {
return nil, err
}
return blob.NewFSStorage(&fso)
return fsstorage.New(&fso)
case "gs", "gcs":
var gcso blob.GCSStorageOptions
var gcso gcsstorage.Options
if err := parseGoogleCloudStorageURL(&gcso, u); err != nil {
return nil, err
}
return blob.NewGCSStorage(&gcso)
return gcsstorage.New(&gcso)
default:
return nil, fmt.Errorf("unrecognized storage type: %v", u.Scheme)
}
}
func parseFilesystemURL(fso *blob.FSStorageOptions, u *url.URL) error {
func parseFilesystemURL(fso *fsstorage.Options, u *url.URL) error {
if u.Opaque != "" {
fso.Path = u.Opaque
} else {
@@ -66,7 +68,7 @@ func parseFilesystemURL(fso *blob.FSStorageOptions, u *url.URL) error {
return nil
}
func parseGoogleCloudStorageURL(gcso *blob.GCSStorageOptions, u *url.URL) error {
func parseGoogleCloudStorageURL(gcso *gcsstorage.Options, u *url.URL) error {
gcso.BucketName = u.Host
gcso.Prefix = u.Path
return nil
fs/upload_test.go
+4 -3
View File
@@ -6,8 +6,9 @@ import (
"os"
"path/filepath"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/repo"
"github.com/kopia/kopia/storage"
"github.com/kopia/kopia/storage/filesystem"
"testing"
)
@@ -16,7 +17,7 @@ type uploadTestHarness struct {
sourceDir string
repoDir string
repo repo.Repository
storage blob.Storage
storage storage.Storage
lister *perPathLister
uploader Uploader
}
@@ -61,7 +62,7 @@ func newUploadTestHarness() *uploadTestHarness {
panic("cannot create temp directory: " + err.Error())
}
storage, err := blob.NewFSStorage(&blob.FSStorageOptions{
storage, err := filesystem.New(&filesystem.Options{
Path: repoDir,
})
repo/buffer_manager.go
+2 -2
View File
@@ -7,7 +7,7 @@ import (
"sync"
"sync/atomic"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/storage"
)
// bufferManager manages pool of reusable bytes.Buffer objects.
@@ -35,7 +35,7 @@ func (mgr *bufferManager) returnBuffer(b *bytes.Buffer) {
mgr.pool.Put(b)
}
func (mgr *bufferManager) returnBufferOnClose(b *bytes.Buffer) blob.ReaderWithLength {
func (mgr *bufferManager) returnBufferOnClose(b *bytes.Buffer) storage.ReaderWithLength {
return &returnOnCloser{
buffer: b,
mgr: mgr,
repo/counting_reader.go
+3 -3
View File
@@ -5,12 +5,12 @@ import (
"sync/atomic"
"unsafe"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/storage"
)
// countingReader wraps an io.ReadCloser and counts bytes read.
type countingReader struct {
blob.ReaderWithLength
storage.ReaderWithLength
counter *int64
}
@@ -25,7 +25,7 @@ func (cr *countingReader) String() string {
return fmt.Sprintf("countingReader(%v)", cr.ReaderWithLength)
}
func newCountingReader(source blob.ReaderWithLength, counter *int64) blob.ReaderWithLength {
func newCountingReader(source storage.ReaderWithLength, counter *int64) storage.ReaderWithLength {
if uintptr(unsafe.Pointer(counter))&7 != 0 {
panic("counter address must be 64-bit aligned")
}
repo/crypto_stream.go
+2 -2
View File
@@ -5,7 +5,7 @@ import (
"fmt"
"io"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/storage"
)
// encryptingReader wraps an io.Reader and returns data encrypted using a stream cipher
@@ -55,7 +55,7 @@ func (er *encryptingReader) String() string {
return fmt.Sprintf("encryptingReader(%v)", er.source)
}
func newEncryptingReader(source blob.ReaderWithLength, c cipher.Stream) blob.ReaderWithLength {
func newEncryptingReader(source storage.ReaderWithLength, c cipher.Stream) storage.ReaderWithLength {
return &encryptingReader{
source: source,
closer: source,
repo/crypto_stream_test.go
+2 -2
View File
@@ -7,7 +7,7 @@ import (
"fmt"
"io/ioutil"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/storage"
"testing"
)
@@ -45,7 +45,7 @@ func TestCryptoStream(t *testing.T) {
continue
}
enc := newEncryptingReader(
blob.NewReader(bytes.NewBuffer(data)),
storage.NewReader(bytes.NewBuffer(data)),
s.cipher)
v, err := ioutil.ReadAll(enc)
actual := fmt.Sprintf("%x", v)
repo/object_reader.go
+2 -2
View File
@@ -4,7 +4,7 @@ import (
"fmt"
"io"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/storage"
)
type seekTableEntry struct {
@@ -26,7 +26,7 @@ func (r *seekTableEntry) String() string {
}
type objectReader struct {
storage blob.Storage
storage storage.Storage
seekTable []seekTableEntry
repo/object_writer.go
+2 -2
View File
@@ -5,7 +5,7 @@ import (
"fmt"
"io"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/storage"
)
// ObjectWriter allows writing content to the storage and supports automatic deduplication and encryption
@@ -101,7 +101,7 @@ func (w *objectWriter) flushBuffer(force bool) error {
return err
}
if err := w.repo.storage.PutBlock(objectID.BlockID(), blockReader, blob.PutOptionsDefault); err != nil {
if err := w.repo.storage.PutBlock(objectID.BlockID(), blockReader, storage.PutOptionsDefault); err != nil {
return fmt.Errorf(
"error when flushing chunk %d of %s to %#v: %#v",
w.flushedObjectCount,
repo/repository.go
+12 -11
View File
@@ -14,7 +14,8 @@ import (
"strings"
"sync/atomic"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/storage"
"github.com/kopia/kopia/storage/logging"
)
const (
@@ -34,14 +35,14 @@ var constantIV = []byte("kopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopia
// Repository objects addressed by their content allows reading and writing them.
type Repository interface {
// NewWriter opens an ObjectWriter for writing new content to the blob.
// NewWriter opens an ObjectWriter for writing new content to the storage.
NewWriter(options ...WriterOption) ObjectWriter
// Open creates an io.ReadSeeker for reading object with a specified ID.
Open(objectID ObjectID) (ObjectReader, error)
Flush() error
Storage() blob.Storage
Storage() storage.Storage
Close()
ResetStats()
@@ -68,7 +69,7 @@ type RepositoryStats struct {
type keygenFunc func([]byte) (blockIDBytes []byte, key []byte)
type repository struct {
storage blob.Storage
storage storage.Storage
verbose bool
bufferManager *bufferManager
stats *RepositoryStats
@@ -83,7 +84,7 @@ func (repo *repository) Close() {
}
func (repo *repository) Flush() error {
if f, ok := repo.storage.(blob.Flusher); ok {
if f, ok := repo.storage.(storage.Flusher); ok {
return f.Flush()
}
@@ -98,7 +99,7 @@ func (repo *repository) Stats() RepositoryStats {
return *repo.stats
}
func (repo *repository) Storage() blob.Storage {
func (repo *repository) Storage() storage.Storage {
return repo.storage
}
@@ -145,7 +146,7 @@ type RepositoryOption func(o *repository) error
// of goroutines.
func WriteBack(workerCount int) RepositoryOption {
return func(o *repository) error {
o.storage = blob.NewWriteBackWrapper(o.storage, workerCount)
o.storage = storage.NewWriteBackWrapper(o.storage, workerCount)
return nil
}
}
@@ -153,7 +154,7 @@ func WriteBack(workerCount int) RepositoryOption {
// EnableLogging is an RepositoryOption that causes all storage access to be logged.
func EnableLogging() RepositoryOption {
return func(o *repository) error {
o.storage = blob.NewLoggingWrapper(o.storage)
o.storage = logging.NewWrapper(o.storage)
return nil
}
}
@@ -166,7 +167,7 @@ func hmacFunc(key []byte, hf func() hash.Hash) func() hash.Hash {
// NewRepository creates new Repository with the specified storage, options, and key provider.
func NewRepository(
r blob.Storage,
r storage.Storage,
f *Format,
options ...RepositoryOption,
) (Repository, error) {
@@ -206,7 +207,7 @@ func (repo *repository) hashBuffer(data []byte) ([]byte, []byte) {
return repo.idFormat.hashBuffer(data, repo.format.Secret)
}
func (repo *repository) hashBufferForWriting(buffer *bytes.Buffer, prefix string) (ObjectID, blob.ReaderWithLength, error) {
func (repo *repository) hashBufferForWriting(buffer *bytes.Buffer, prefix string) (ObjectID, storage.ReaderWithLength, error) {
var data []byte
if buffer != nil {
data = buffer.Bytes()
@@ -234,7 +235,7 @@ func (repo *repository) hashBufferForWriting(buffer *bytes.Buffer, prefix string
atomic.AddInt64(&repo.stats.HashedBytes, int64(len(data)))
if buffer == nil {
return objectID, blob.NewReader(bytes.NewBuffer(nil)), nil
return objectID, storage.NewReader(bytes.NewBuffer(nil)), nil
}
blockReader := repo.bufferManager.returnBufferOnClose(buffer)
repo/repository_test.go
+7 -6
View File
@@ -12,7 +12,8 @@ import (
"strings"
"testing"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/storage"
"github.com/kopia/kopia/storage/storagetesting"
)
func testFormat() *Format {
@@ -40,7 +41,7 @@ func getMd5LObjectID(data []byte) string {
func setupTest(t *testing.T) (data map[string][]byte, repo Repository) {
data = map[string][]byte{}
st := blob.NewMapStorage(data)
st := storagetesting.NewMapStorage(data)
repo, err := NewRepository(st, testFormat())
if err != nil {
@@ -238,7 +239,7 @@ func TestHMAC(t *testing.T) {
s.ObjectFormat = "md5"
s.Secret = []byte{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}
repo, err := NewRepository(blob.NewMapStorage(data), s)
repo, err := NewRepository(storagetesting.NewMapStorage(data), s)
if err != nil {
t.Errorf("cannot create manager: %v", err)
}
@@ -323,7 +324,7 @@ func TestReaderStoredBlockNotFound(t *testing.T) {
t.Errorf("cannot parse object ID: %v", err)
}
reader, err := repo.Open(objectID)
if err != blob.ErrBlockNotFound || reader != nil {
if err != storage.ErrBlockNotFound || reader != nil {
t.Errorf("unexpected result: reader: %v err: %v", reader, err)
}
}
@@ -478,7 +479,7 @@ func TestFormats(t *testing.T) {
for caseIndex, c := range cases {
data := map[string][]byte{}
st := blob.NewMapStorage(data)
st := storagetesting.NewMapStorage(data)
t.Logf("verifying %#v", c.format)
repo, err := NewRepository(st, c.format)
@@ -517,7 +518,7 @@ func TestFormats(t *testing.T) {
func TestInvalidEncryptionKey(t *testing.T) {
data := map[string][]byte{}
st := blob.NewMapStorage(data)
st := storagetesting.NewMapStorage(data)
format := Format{
Version: "1",
ObjectFormat: "sha512-aes256",
blob/config.go → storage/config.go
+1 -1
View File
@@ -1,4 +1,4 @@
package blob
package storage
import (
"encoding/json"
storage/doc.go
+2
View File
@@ -0,0 +1,2 @@
// Package storage implements simple storage of immutable, unstructured binary large objects (BLOBs).
package storage
blob/errors.go → storage/errors.go
+2 -2
View File
@@ -1,9 +1,9 @@
package blob
package storage
import "errors"
var (
// ErrBlockNotFound is returned when a block cannot be found in blob.
// ErrBlockNotFound is returned when a block cannot be found in storage.
ErrBlockNotFound = errors.New("block not found")
// ErrInvalidChecksum is returned when a storage block is invalid, which may indicate
blob/filesystem.go → storage/filesystem/filesystem.go
+20 -54
View File
@@ -1,4 +1,4 @@
package blob
package filesystem
import (
"fmt"
@@ -9,6 +9,8 @@ import (
"path/filepath"
"strconv"
"strings"
"github.com/kopia/kopia/storage"
)
const (
@@ -23,45 +25,9 @@ var (
)
type fsStorage struct {
FSStorageOptions
Options
}
// FSStorageOptions defines options for Filesystem-backed blob.
type FSStorageOptions struct {
Path string `json:"path"`
DirectoryShards []int `json:"dirShards,omitempty"`
FileMode os.FileMode `json:"fileMode,omitempty"`
DirectoryMode os.FileMode `json:"dirMode,omitempty"`
FileUID *int `json:"uid,omitempty"`
FileGID *int `json:"gid,omitempty"`
}
func (fso *FSStorageOptions) fileMode() os.FileMode {
if fso.FileMode == 0 {
return fsDefaultFileMode
}
return fso.FileMode
}
func (fso *FSStorageOptions) dirMode() os.FileMode {
if fso.DirectoryMode == 0 {
return fsDefaultDirMode
}
return fso.DirectoryMode
}
func (fso *FSStorageOptions) shards() []int {
if fso.DirectoryShards == nil {
return fsDefaultShards
}
return fso.DirectoryShards
}
func (fs *fsStorage) BlockExists(blockID string) (bool, error) {
_, path := fs.getShardedPathAndFilePath(blockID)
_, err := os.Stat(path)
@@ -84,7 +50,7 @@ func (fs *fsStorage) GetBlock(blockID string) ([]byte, error) {
}
if os.IsNotExist(err) {
return nil, ErrBlockNotFound
return nil, storage.ErrBlockNotFound
}
return nil, err
@@ -102,8 +68,8 @@ func makeFileName(blockID string) string {
return string(blockID) + fsStorageChunkSuffix
}
func (fs *fsStorage) ListBlocks(prefix string) chan (BlockMetadata) {
result := make(chan (BlockMetadata))
func (fs *fsStorage) ListBlocks(prefix string) chan (storage.BlockMetadata) {
result := make(chan (storage.BlockMetadata))
prefixString := string(prefix)
@@ -129,7 +95,7 @@ func (fs *fsStorage) ListBlocks(prefix string) chan (BlockMetadata) {
}
} else if fullID, ok := getstringFromFileName(currentPrefix + e.Name()); ok {
if strings.HasPrefix(string(fullID), prefixString) {
result <- BlockMetadata{
result <- storage.BlockMetadata{
BlockID: fullID,
Length: uint64(e.Size()),
TimeStamp: e.ModTime(),
@@ -149,7 +115,7 @@ func (fs *fsStorage) ListBlocks(prefix string) chan (BlockMetadata) {
return result
}
func (fs *fsStorage) PutBlock(blockID string, data ReaderWithLength, options PutOptions) error {
func (fs *fsStorage) PutBlock(blockID string, data storage.ReaderWithLength, options storage.PutOptions) error {
// Close the data reader regardless of whether we use it or not.
defer data.Close()
@@ -238,10 +204,10 @@ func parseShardString(shardString string) ([]int, error) {
return result, nil
}
func (fs *fsStorage) ConnectionInfo() ConnectionInfo {
return ConnectionInfo{
fsStorageType,
&fs.FSStorageOptions,
func (fs *fsStorage) ConnectionInfo() storage.ConnectionInfo {
return storage.ConnectionInfo{
Type: fsStorageType,
Config: &fs.Options,
}
}
@@ -249,8 +215,8 @@ func (fs *fsStorage) Close() error {
return nil
}
// NewFSStorage creates new fs-backed storage in a specified directory.
func NewFSStorage(options *FSStorageOptions) (Storage, error) {
// New creates new filesystem-backed storage in a specified directory.
func New(options *Options) (storage.Storage, error) {
var err error
if _, err = os.Stat(options.Path); err != nil {
@@ -258,17 +224,17 @@ func NewFSStorage(options *FSStorageOptions) (Storage, error) {
}
r := &fsStorage{
FSStorageOptions: *options,
Options: *options,
}
return r, nil
}
func init() {
AddSupportedStorage(
storage.AddSupportedStorage(
fsStorageType,
func() interface{} { return &FSStorageOptions{} },
func(o interface{}) (Storage, error) {
return NewFSStorage(o.(*FSStorageOptions))
func() interface{} { return &Options{} },
func(o interface{}) (storage.Storage, error) {
return New(o.(*Options))
})
}
storage/filesystem/filesystem_test.go
+36
View File
@@ -0,0 +1,36 @@
package filesystem
import (
"io/ioutil"
"os"
"testing"
"github.com/kopia/kopia/storage/storagetesting"
)
func TestFileStorage(t *testing.T) {
// Test varioush shard configurations.
for _, shardSpec := range [][]int{
[]int{0},
[]int{1},
[]int{3, 3},
[]int{2},
[]int{1, 1},
[]int{1, 2},
[]int{2, 2, 2},
} {
path, _ := ioutil.TempDir("", "r-fs")
defer os.RemoveAll(path)
r, err := New(&Options{
Path: path,
DirectoryShards: shardSpec,
})
if r == nil || err != nil {
t.Errorf("unexpected result: %v %v", r, err)
}
storagetesting.VerifyStorage(t, r)
}
}
storage/filesystem/options.go
+40
View File
@@ -0,0 +1,40 @@
package filesystem
import "os"
// Options defines options for Filesystem-backed storage.
type Options struct {
Path string `json:"path"`
DirectoryShards []int `json:"dirShards,omitempty"`
FileMode os.FileMode `json:"fileMode,omitempty"`
DirectoryMode os.FileMode `json:"dirMode,omitempty"`
FileUID *int `json:"uid,omitempty"`
FileGID *int `json:"gid,omitempty"`
}
func (fso *Options) fileMode() os.FileMode {
if fso.FileMode == 0 {
return fsDefaultFileMode
}
return fso.FileMode
}
func (fso *Options) dirMode() os.FileMode {
if fso.DirectoryMode == 0 {
return fsDefaultDirMode
}
return fso.DirectoryMode
}
func (fso *Options) shards() []int {
if fso.DirectoryShards == nil {
return fsDefaultShards
}
return fso.DirectoryShards
}
blob/gcs.go → storage/gcs/gcs.go
+23 -80
View File
@@ -1,4 +1,4 @@
package blob
package gcs
import (
"encoding/json"
@@ -7,10 +7,11 @@ import (
"io/ioutil"
"log"
"net/http"
"net/http/httptest"
"os"
"time"
"github.com/kopia/kopia/storage"
"golang.org/x/net/context"
"golang.org/x/oauth2"
"golang.org/x/oauth2/google"
@@ -22,36 +23,13 @@ import (
const (
gcsStorageType = "gcs"
// Those are not really secret, since the app is installed.
// Those are not really set, since the app is installed.
googleCloudClientID = "194841383482-nmn10h4mnllnsvou7qr55tfh5jsmtkap.apps.googleusercontent.com"
googleCloudClientSecret = "ZL52E96Q7iRCD9YXVA7U6UaI"
)
// GCSStorageOptions defines options Google Cloud Storage-backed blob.
type GCSStorageOptions struct {
// BucketName is the name of the GCS bucket where data is stored.
BucketName string `json:"bucket"`
// Prefix specifies additional string to prepend to all objects.
Prefix string `json:"prefix,omitempty"`
// TokenCacheFile is the name of the file that will persist the OAuth2 token.
// If not specified, the token will be persisted in GCSStorageOptions.
TokenCacheFile string `json:"tokenCacheFile,omitempty"`
// Token is the OAuth2 token (when TokenCacheFile is empty)
Token *oauth2.Token `json:"token,omitempty"`
// ReadOnly causes the storage to be configured without write permissions, to prevent accidental
// modifications to the data.
ReadOnly bool `json:"readonly"`
// IgnoreDefaultCredentials disables the use of credentials managed by Google Cloud SDK (gcloud).
IgnoreDefaultCredentials bool `json:"ignoreDefaultCredentials"`
}
type gcsStorage struct {
GCSStorageOptions
Options
objectsService *gcsclient.ObjectsService
}
@@ -70,7 +48,7 @@ func (gcs *gcsStorage) GetBlock(b string) ([]byte, error) {
if err != nil {
if err, ok := err.(*googleapi.Error); ok {
if err.Code == http.StatusNotFound {
return nil, ErrBlockNotFound
return nil, storage.ErrBlockNotFound
}
}
@@ -82,7 +60,7 @@ func (gcs *gcsStorage) GetBlock(b string) ([]byte, error) {
return ioutil.ReadAll(v.Body)
}
func (gcs *gcsStorage) PutBlock(b string, data ReaderWithLength, options PutOptions) error {
func (gcs *gcsStorage) PutBlock(b string, data storage.ReaderWithLength, options storage.PutOptions) error {
defer data.Close()
object := gcsclient.Object{
Name: gcs.getObjectNameString(b),
@@ -94,7 +72,7 @@ func (gcs *gcsStorage) PutBlock(b string, data ReaderWithLength, options PutOpti
// Specify exact chunk size to ensure data is uploaded in one shot or not at all.
googleapi.ChunkSize(data.Len()),
)
if options&PutOptionsOverwrite == 0 {
if options&storage.PutOptionsOverwrite == 0 {
if ok, _ := gcs.BlockExists(b); ok {
return nil
}
@@ -129,8 +107,8 @@ func (gcs *gcsStorage) getObjectNameString(b string) string {
return gcs.Prefix + string(b)
}
func (gcs *gcsStorage) ListBlocks(prefix string) chan (BlockMetadata) {
ch := make(chan BlockMetadata, 100)
func (gcs *gcsStorage) ListBlocks(prefix string) chan (storage.BlockMetadata) {
ch := make(chan storage.BlockMetadata, 100)
go func() {
ps := gcs.getObjectNameString(prefix)
@@ -143,11 +121,11 @@ func (gcs *gcsStorage) ListBlocks(prefix string) chan (BlockMetadata) {
for _, o := range page.Items {
t, e := time.Parse(time.RFC3339, o.TimeCreated)
if e != nil {
ch <- BlockMetadata{
ch <- storage.BlockMetadata{
Error: e,
}
} else {
ch <- BlockMetadata{
ch <- storage.BlockMetadata{
BlockID: string(o.Name)[len(gcs.Prefix):],
Length: o.Size,
TimeStamp: t,
@@ -173,10 +151,10 @@ func (gcs *gcsStorage) Flush() error {
return nil
}
func (gcs *gcsStorage) ConnectionInfo() ConnectionInfo {
return ConnectionInfo{
gcsStorageType,
&gcs.GCSStorageOptions,
func (gcs *gcsStorage) ConnectionInfo() storage.ConnectionInfo {
return storage.ConnectionInfo{
Type: gcsStorageType,
Config: &gcs.Options,
}
}
@@ -210,19 +188,19 @@ func saveToken(file string, token *oauth2.Token) {
json.NewEncoder(f).Encode(token)
}
// NewGCSStorage creates new Google Cloud Storage-backed storage with specified options:
// New creates new Google Cloud Storage-backed storage with specified options:
//
// - the 'BucketName' field is required and all other parameters are optional.
//
// By default the connection reuses credentials managed by (https://cloud.google.com/sdk/),
// but this can be disabled by setting IgnoreDefaultCredentials to true.
func NewGCSStorage(options *GCSStorageOptions) (Storage, error) {
func New(options *Options) (storage.Storage, error) {
ctx := context.TODO()
//ctx = httptrace.WithClientTrace(ctx, &httptrace.ClientTrace{})
gcs := &gcsStorage{
GCSStorageOptions: *options,
Options: *options,
}
if gcs.BucketName == "" {
@@ -346,48 +324,13 @@ func tokenFromWeb(ctx context.Context, config *oauth2.Config) (*oauth2.Token, er
return token, nil
}
func authPrompt(url string, state string) (authenticationCode string, err error) {
ch := make(chan string)
ts := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
if req.URL.Path == "/favicon.ico" {
http.Error(rw, "", 404)
return
}
if req.FormValue("state") != state {
log.Printf("State doesn't match: req = %#v", req)
http.Error(rw, "", 500)
return
}
if code := req.FormValue("code"); code != "" {
fmt.Fprintf(rw, "<h1>Success</h1>Authorized.")
rw.(http.Flusher).Flush()
ch <- code
return
}
log.Printf("no code")
http.Error(rw, "", 500)
}))
defer ts.Close()
log.Println("Go to", url)
var code string
n, err := fmt.Scanf("%s", &code)
if n == 1 {
return code, nil
}
return "", err
}
func init() {
AddSupportedStorage(
storage.AddSupportedStorage(
gcsStorageType,
func() interface{} {
return &GCSStorageOptions{}
return &Options{}
},
func(o interface{}) (Storage, error) {
return NewGCSStorage(o.(*GCSStorageOptions))
func(o interface{}) (storage.Storage, error) {
return New(o.(*Options))
})
}
storage/gcs/options.go
+26
View File
@@ -0,0 +1,26 @@
package gcs
import "golang.org/x/oauth2"
// Options defines options Google Cloud Storage-backed storage.
type Options struct {
// BucketName is the name of the GCS bucket where data is stored.
BucketName string `json:"bucket"`
// Prefix specifies additional string to prepend to all objects.
Prefix string `json:"prefix,omitempty"`
// TokenCacheFile is the name of the file that will persist the OAuth2 token.
// If not specified, the token will be persisted in Options.
TokenCacheFile string `json:"tokenCacheFile,omitempty"`
// Token is the OAuth2 token (when TokenCacheFile is empty)
Token *oauth2.Token `json:"token,omitempty"`
// ReadOnly causes the storage to be configured without write permissions, to prevent accidental
// modifications to the data.
ReadOnly bool `json:"readonly"`
// IgnoreDefaultCredentials disables the use of credentials managed by Google Cloud SDK (gcloud).
IgnoreDefaultCredentials bool `json:"ignoreDefaultCredentials"`
}
storage/logging/log_test.go
+16
View File
@@ -0,0 +1,16 @@
package logging
import (
"testing"
"github.com/kopia/kopia/storage/storagetesting"
)
func TestLoggingStorage(t *testing.T) {
data := map[string][]byte{}
r := NewWrapper(storagetesting.NewMapStorage(data))
if r == nil {
t.Errorf("unexpected result: %v", r)
}
storagetesting.VerifyStorage(t, r)
}
blob/logging.go → storage/logging/logging.go
+9 -7
View File
@@ -1,12 +1,14 @@
package blob
package logging
import (
"io"
"log"
"github.com/kopia/kopia/storage"
)
type loggingStorage struct {
Storage
storage.Storage
}
func (s *loggingStorage) BlockExists(id string) (bool, error) {
@@ -25,7 +27,7 @@ func (s *loggingStorage) GetBlock(id string) ([]byte, error) {
return result, err
}
func (s *loggingStorage) PutBlock(id string, data ReaderWithLength, options PutOptions) error {
func (s *loggingStorage) PutBlock(id string, data storage.ReaderWithLength, options storage.PutOptions) error {
err := s.Storage.PutBlock(id, data, options)
log.Printf("PutBlock(%#v, options=%v, len=%v)=%#v", id, options, data.Len(), err)
return err
@@ -37,13 +39,13 @@ func (s *loggingStorage) DeleteBlock(id string) error {
return err
}
func (s *loggingStorage) ListBlocks(prefix string) chan (BlockMetadata) {
func (s *loggingStorage) ListBlocks(prefix string) chan (storage.BlockMetadata) {
log.Printf("ListBlocks(%#v)", prefix)
return s.Storage.ListBlocks(prefix)
}
func (s *loggingStorage) Flush() error {
if s, ok := s.Storage.(Flusher); ok {
if s, ok := s.Storage.(storage.Flusher); ok {
log.Printf("Flush()")
return s.Flush()
}
@@ -60,7 +62,7 @@ func (s *loggingStorage) Close() error {
return nil
}
// NewLoggingWrapper returns a Storage wrapper that logs all storage commands.
func NewLoggingWrapper(wrapped Storage) Storage {
// NewWrapper returns a Storage wrapper that logs all storage commands.
func NewWrapper(wrapped storage.Storage) storage.Storage {
return &loggingStorage{wrapped}
}
blob/registry.go → storage/registry.go
+1 -1
View File
@@ -1,4 +1,4 @@
package blob
package storage
import "fmt"
blob/storage.go → storage/storage.go
+2 -2
View File
@@ -1,4 +1,4 @@
package blob
package storage
import (
"bytes"
@@ -58,7 +58,7 @@ func (bbr *bytesReaderWithLength) String() string {
return fmt.Sprintf("buffer(len=%v)", bbr.Len())
}
// BlockMetadata represents metadata about a single block in a blob.
// BlockMetadata represents metadata about a single block in a storage.
// If Error field is set, no other field values should be assumed to be correct.
type BlockMetadata struct {
BlockID string
blob/map.go → storage/storagetesting/map.go
+9 -7
View File
@@ -1,4 +1,4 @@
package blob
package storagetesting
import (
"io/ioutil"
@@ -6,6 +6,8 @@ import (
"strings"
"sync"
"time"
"github.com/kopia/kopia/storage"
)
type mapStorage struct {
@@ -29,10 +31,10 @@ func (s *mapStorage) GetBlock(id string) ([]byte, error) {
return data, nil
}
return nil, ErrBlockNotFound
return nil, storage.ErrBlockNotFound
}
func (s *mapStorage) PutBlock(id string, data ReaderWithLength, options PutOptions) error {
func (s *mapStorage) PutBlock(id string, data storage.ReaderWithLength, options storage.PutOptions) error {
s.mutex.Lock()
defer s.mutex.Unlock()
@@ -59,8 +61,8 @@ func (s *mapStorage) DeleteBlock(id string) error {
return nil
}
func (s *mapStorage) ListBlocks(prefix string) chan (BlockMetadata) {
ch := make(chan (BlockMetadata))
func (s *mapStorage) ListBlocks(prefix string) chan (storage.BlockMetadata) {
ch := make(chan (storage.BlockMetadata))
fixedTime := time.Now()
go func() {
s.mutex.RLock()
@@ -77,7 +79,7 @@ func (s *mapStorage) ListBlocks(prefix string) chan (BlockMetadata) {
for _, k := range keys {
v := s.data[k]
ch <- BlockMetadata{
ch <- storage.BlockMetadata{
BlockID: string(k),
Length: uint64(len(v)),
TimeStamp: fixedTime,
@@ -90,6 +92,6 @@ func (s *mapStorage) ListBlocks(prefix string) chan (BlockMetadata) {
// NewMapStorage returns an implementation of Storage backed by the contents of given map.
// Used primarily for testing.
func NewMapStorage(data map[string][]byte) Storage {
func NewMapStorage(data map[string][]byte) storage.Storage {
return &mapStorage{data: data}
}
storage/storagetesting/map_test.go
+12
View File
@@ -0,0 +1,12 @@
package storagetesting
import "testing"
func TestMapStorage(t *testing.T) {
data := map[string][]byte{}
r := NewMapStorage(data)
if r == nil {
t.Errorf("unexpected result: %v", r)
}
VerifyStorage(t, r)
}
blob/storage_test.go → storage/storagetesting/verify.go
+7 -51
View File
@@ -1,58 +1,14 @@
package blob
package storagetesting
import (
"bytes"
"io/ioutil"
"os"
"testing"
"github.com/kopia/kopia/storage"
)
func TestLoggingStorage(t *testing.T) {
data := map[string][]byte{}
r := NewLoggingWrapper(NewMapStorage(data))
if r == nil {
t.Errorf("unexpected result: %v", r)
}
verifyStorage(t, r)
}
func TestMapStorage(t *testing.T) {
data := map[string][]byte{}
r := NewMapStorage(data)
if r == nil {
t.Errorf("unexpected result: %v", r)
}
verifyStorage(t, r)
}
func TestFileStorage(t *testing.T) {
// Test varioush shard configurations.
for _, shardSpec := range [][]int{
[]int{0},
[]int{1},
[]int{3, 3},
[]int{2},
[]int{1, 1},
[]int{1, 2},
[]int{2, 2, 2},
} {
path, _ := ioutil.TempDir("", "r-fs")
defer os.RemoveAll(path)
r, err := NewFSStorage(&FSStorageOptions{
Path: path,
DirectoryShards: shardSpec,
})
if r == nil || err != nil {
t.Errorf("unexpected result: %v %v", r, err)
}
verifyStorage(t, r)
}
}
func verifyStorage(t *testing.T, r Storage) {
// VerifyStorage verifies the behavior of the specified storage.
func VerifyStorage(t *testing.T, r storage.Storage) {
blocks := []struct {
blk string
contents []byte
@@ -70,7 +26,7 @@ func verifyStorage(t *testing.T, r Storage) {
}
data, err := r.GetBlock(b.blk)
if err != ErrBlockNotFound {
if err != storage.ErrBlockNotFound {
t.Errorf("unexpected error when calling GetBlock(%v): %v", b.blk, err)
}
if data != nil {
@@ -80,7 +36,7 @@ func verifyStorage(t *testing.T, r Storage) {
// Now add blocks.
for _, b := range blocks {
r.PutBlock(b.blk, NewReader(bytes.NewBuffer(b.contents)), PutOptionsDefault)
r.PutBlock(b.blk, storage.NewReader(bytes.NewBuffer(b.contents)), storage.PutOptionsDefault)
if x, err := r.BlockExists(b.blk); !x || err != nil {
t.Errorf("block does not exist after adding it: %v %v", b.blk, err)
blob/writeback.go → storage/writeback.go
+1 -1
View File
@@ -1,7 +1,7 @@
// Wrapper which implements asynchronous (write-back) PutBlock and DeleteBlock operation
// useful for slower backends (cloud).
package blob
package storage
import (
"fmt"
vault/vault.go
+26 -26
View File
@@ -15,8 +15,8 @@ import (
"io"
"strings"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/repo"
"github.com/kopia/kopia/storage"
"golang.org/x/crypto/hkdf"
)
@@ -38,7 +38,7 @@ var ErrItemNotFound = errors.New("item not found")
// Vault is a secure storage for the secrets.
type Vault struct {
storage blob.Storage
storage storage.Storage
masterKey []byte
format Format
itemPrefix string
@@ -46,12 +46,12 @@ type Vault struct {
}
type repositoryConfig struct {
Connection *blob.ConnectionInfo `json:"connection"`
Format *repo.Format `json:"format"`
Connection *storage.ConnectionInfo `json:"connection"`
Format *repo.Format `json:"format"`
}
// Storage returns the underlying blob storage that stores the repository.
func (v *Vault) Storage() blob.Storage {
func (v *Vault) Storage() storage.Storage {
return v.storage
}
@@ -101,15 +101,15 @@ func (v *Vault) writeEncryptedBlock(itemID string, content []byte) error {
return v.storage.PutBlock(
v.itemPrefix+itemID,
blob.NewReader(bytes.NewBuffer(content)),
blob.PutOptionsOverwrite,
storage.NewReader(bytes.NewBuffer(content)),
storage.PutOptionsOverwrite,
)
}
func (v *Vault) readEncryptedBlock(itemID string) ([]byte, error) {
content, err := v.storage.GetBlock(v.itemPrefix + itemID)
if err != nil {
if err == blob.ErrBlockNotFound {
if err == storage.ErrBlockNotFound {
return nil, ErrItemNotFound
}
return nil, fmt.Errorf("unexpected error reading %v: %v", itemID, err)
@@ -197,19 +197,19 @@ func (v *Vault) RepositoryFormat() *repo.Format {
// OpenRepository connects to the repository the vault is associated with.
func (v *Vault) OpenRepository() (repo.Repository, error) {
var storage blob.Storage
var st storage.Storage
var err error
if v.repoConfig.Connection != nil {
storage, err = blob.NewStorage(*v.repoConfig.Connection)
st, err = storage.NewStorage(*v.repoConfig.Connection)
if err != nil {
return nil, fmt.Errorf("unable to open repository: %v", err)
}
} else {
storage = v.storage
st = v.storage
}
return repo.NewRepository(storage, v.repoConfig.Format)
return repo.NewRepository(st, v.repoConfig.Format)
}
// Get returns the contents of a specified vault item.
@@ -263,14 +263,14 @@ func (v *Vault) Close() error {
}
type vaultConfig struct {
ConnectionInfo blob.ConnectionInfo `json:"connection"`
Key []byte `json:"key,omitempty"`
ConnectionInfo storage.ConnectionInfo `json:"connection"`
Key []byte `json:"key,omitempty"`
}
// Token returns a persistent opaque string that encodes the configuration of vault storage
// and its credentials in a way that can be later used to open the vault.
func (v *Vault) Token() (string, error) {
cip, ok := v.storage.(blob.ConnectionInfoProvider)
cip, ok := v.storage.(storage.ConnectionInfoProvider)
if !ok {
return "", errors.New("repository does not support persisting configuration")
}
@@ -301,10 +301,10 @@ func (v *Vault) Remove(itemID string) error {
// Create initializes a Vault attached to the specified repository.
func Create(
vaultStorage blob.Storage,
vaultStorage storage.Storage,
vaultFormat *Format,
vaultCreds Credentials,
repoStorage blob.Storage,
repoStorage storage.Storage,
repoFormat *repo.Format,
) (*Vault, error) {
v := Vault{
@@ -317,7 +317,7 @@ func Create(
v.itemPrefix = colocatedVaultItemPrefix
}
cip, ok := repoStorage.(blob.ConnectionInfoProvider)
cip, ok := repoStorage.(storage.ConnectionInfoProvider)
if !ok {
return nil, errors.New("repository does not support persisting configuration")
}
@@ -336,8 +336,8 @@ func Create(
vaultStorage.PutBlock(
v.itemPrefix+formatBlockID,
blob.NewReader(bytes.NewBuffer(formatBytes)),
blob.PutOptionsOverwrite,
storage.NewReader(bytes.NewBuffer(formatBytes)),
storage.PutOptionsOverwrite,
)
// Write encrypted repository configuration block.
@@ -359,7 +359,7 @@ func Create(
}
// Open opens a vault.
func Open(vaultStorage blob.Storage, vaultCreds Credentials) (*Vault, error) {
func Open(vaultStorage storage.Storage, vaultCreds Credentials) (*Vault, error) {
v := Vault{
storage: vaultStorage,
}
@@ -367,7 +367,7 @@ func Open(vaultStorage blob.Storage, vaultCreds Credentials) (*Vault, error) {
var prefix string
f, err := vaultStorage.GetBlock(formatBlockID)
if err == blob.ErrBlockNotFound {
if err == storage.ErrBlockNotFound {
prefix = colocatedVaultItemPrefix
f, err = vaultStorage.GetBlock(prefix + formatBlockID)
}
@@ -407,7 +407,7 @@ func OpenWithToken(token string) (*Vault, error) {
return nil, fmt.Errorf("invalid vault json token: %v", err)
}
st, err := blob.NewStorage(vc.ConnectionInfo)
st, err := storage.NewStorage(vc.ConnectionInfo)
if err != nil {
return nil, fmt.Errorf("cannot open vault storage: %v", err)
}
@@ -438,17 +438,17 @@ func checkReservedName(itemID string) error {
return nil
}
func sameStorage(s1, s2 blob.Storage) bool {
func sameStorage(s1, s2 storage.Storage) bool {
if s1 == s2 {
return true
}
cip1, ok := s1.(blob.ConnectionInfoProvider)
cip1, ok := s1.(storage.ConnectionInfoProvider)
if !ok {
return false
}
cip2, ok := s2.(blob.ConnectionInfoProvider)
cip2, ok := s2.(storage.ConnectionInfoProvider)
if !ok {
return false
}
vault/vault_test.go
+4 -3
View File
@@ -7,8 +7,9 @@ import (
"reflect"
"strings"
"github.com/kopia/kopia/blob"
"github.com/kopia/kopia/repo"
"github.com/kopia/kopia/storage"
"github.com/kopia/kopia/storage/filesystem"
"testing"
)
@@ -304,9 +305,9 @@ func assertVaultItems(t *testing.T, v *Vault, prefix string, expected []string)
}
}
func mustCreateFileStorage(t *testing.T, path string) blob.Storage {
func mustCreateFileStorage(t *testing.T, path string) storage.Storage {
os.MkdirAll(path, 0700)
s, err := blob.NewFSStorage(&blob.FSStorageOptions{
s, err := filesystem.New(&filesystem.Options{
Path: path,
})
if err != nil {