From 30c11dc926a447785c038bbff2ce9efee96a3ee4 Mon Sep 17 00:00:00 2001 From: Jarek Kowalski Date: Mon, 27 Nov 2017 18:07:16 -0800 Subject: [PATCH] refactored block manager to support on-disk caching --- block/block_cache.go | 41 +++++ block/block_formatting_options.go | 1 + block/block_manager.go | 76 +++++---- block/block_manager_test.go | 11 +- block/disk_block_cache.go | 260 ++++++++++++++++++++++++++++++ block/null_block_cache.go | 35 ++++ cli/command_repository_connect.go | 11 +- cli/command_repository_status.go | 7 +- internal/config/local_config.go | 4 +- manifest/manifest_manager_test.go | 16 +- repo/connect.go | 38 ++++- repo/format_block.go | 10 +- repo/initialize.go | 3 +- repo/open.go | 74 ++++++--- repo/repository_test.go | 4 +- 15 files changed, 511 insertions(+), 80 deletions(-) create mode 100644 block/block_cache.go create mode 100644 block/disk_block_cache.go create mode 100644 block/null_block_cache.go diff --git a/block/block_cache.go b/block/block_cache.go new file mode 100644 index 000000000..e95d15c7e --- /dev/null +++ b/block/block_cache.go @@ -0,0 +1,41 @@ +package block + +import ( + "time" + + "github.com/kopia/kopia/storage" +) + +type blockCache interface { + getBlock(blockID string, offset, length int64) ([]byte, error) + putBlock(blockID string, data []byte) error + listIndexBlocks() ([]Info, error) +} + +// CachingOptions specifies configuration of local cache. +type CachingOptions struct { + CacheDirectory string `json:"cacheDirectory,omitempty"` + MaxCacheSizeBytes int64 `json:"maxCacheSize,omitempty"` + MaxListCacheDurationSec int `json:"maxListCacheDuration,omitempty"` + HMACSecret []byte `json:"-"` +} + +func newBlockCache(st storage.Storage, caching CachingOptions) blockCache { + if caching.MaxCacheSizeBytes == 0 || caching.CacheDirectory == "" { + return nullBlockCache{st} + } + + c := &diskBlockCache{ + st: st, + directory: caching.CacheDirectory, + maxSizeBytes: caching.MaxCacheSizeBytes, + hmacSecret: append([]byte(nil), caching.HMACSecret...), + listCacheDuration: time.Duration(caching.MaxListCacheDurationSec) * time.Second, + closed: make(chan struct{}), + } + + c.sweepDirectory() + go c.sweepDirectoryPeriodically() + + return c +} diff --git a/block/block_formatting_options.go b/block/block_formatting_options.go index 872b06b9c..857e390cf 100644 --- a/block/block_formatting_options.go +++ b/block/block_formatting_options.go @@ -7,4 +7,5 @@ type FormattingOptions struct { HMACSecret []byte `json:"secret,omitempty"` // HMAC secret used to generate encryption keys MasterKey []byte `json:"masterKey,omitempty"` // master encryption key (SIV-mode encryption only) MaxPackedContentLength int `json:"maxPackedContentLength,omitempty"` // maximum size of object to be considered for storage in a pack + MaxPackSize int `json:"maxPackSize,omitempty"` // maximum size of a pack object } diff --git a/block/block_manager.go b/block/block_manager.go index d49c7602f..0d2b87fe4 100644 --- a/block/block_manager.go +++ b/block/block_manager.go @@ -43,18 +43,21 @@ type blockLocation struct { // Info is an information about a single block managed by Manager. type Info struct { - BlockID string - Length int64 - Timestamp time.Time - PackGroup string - PackBlockID string - PackOffset int64 + BlockID string `json:"blockID"` + Length int64 `json:"length"` + Timestamp time.Time `json:"time"` + PackGroup string `json:"packGroup,omitempty"` + PackBlockID string `json:"packBlockID,omitempty"` + PackOffset int64 `json:"packOffset,omitempty"` } // Manager manages storage blocks at a low level with encryption, deduplication and packaging. type Manager struct { - storage storage.Storage - stats Stats + Format FormattingOptions + + stats Stats + + cache blockCache mu sync.Mutex locked bool @@ -215,6 +218,14 @@ func (bm *Manager) ResetStats() { bm.stats = Stats{} } +// Close closes block manager. +func (bm *Manager) Close() error { + if c, ok := bm.cache.(io.Closer); ok { + return c.Close() + } + return nil +} + func (bm *Manager) ensurePackGroupLocked(packGroup string, unpacked bool) *packInfo { var suffix string @@ -323,20 +334,9 @@ func (bm *Manager) finishPackLocked(g *packInfo) error { // ListIndexBlocks returns the list of all index blocks, including inactive, sorted by time. func (bm *Manager) ListIndexBlocks() ([]Info, error) { - ch, cancel := bm.storage.ListBlocks(packBlockPrefix) - defer cancel() - - var blocks []Info - for b := range ch { - if b.Error != nil { - return nil, fmt.Errorf("error listing index blocks: %v", b.Error) - } - - blocks = append(blocks, Info{ - BlockID: b.BlockID, - Length: b.Length, - Timestamp: b.TimeStamp, - }) + blocks, err := bm.cache.listIndexBlocks() + if err != nil { + return nil, fmt.Errorf("error listing index blocks: %v", err) } sortBlocksByTime(blocks) @@ -865,7 +865,7 @@ func (bm *Manager) writeUnpackedBlockNotLocked(data []byte, prefix string, suffi atomic.AddInt32(&bm.stats.WrittenBlocks, 1) atomic.AddInt64(&bm.stats.WrittenBytes, int64(len(data))) - if err := bm.storage.PutBlock(blockID, data2); err != nil { + if err := bm.cache.putBlock(blockID, data2); err != nil { return "", err } @@ -977,10 +977,10 @@ func (bm *Manager) getBlockInternalLocked(blockID string) ([]byte, error) { if s.PackBlockID != "" { underlyingBlockID = s.PackBlockID - payload, err = bm.storage.GetBlock(underlyingBlockID, s.PackOffset, s.Length) + payload, err = bm.cache.getBlock(underlyingBlockID, s.PackOffset, s.Length) decryptSkip = int(s.PackOffset) } else { - payload, err = bm.storage.GetBlock(blockID, 0, -1) + payload, err = bm.cache.getBlock(blockID, 0, -1) } if err != nil { @@ -1035,20 +1035,28 @@ func (bm *Manager) assertLocked() { } } -// SetCacheDir enables local caching of data in the provided directory up to maximum length. -func (bm *Manager) SetCacheDir(dir string) { -} - // NewManager creates new block manager with given packing options and a formatter. -func NewManager(st storage.Storage, maxPackedContentLength, maxPackSize int, formatter Formatter) *Manager { +func NewManager(st storage.Storage, f FormattingOptions, caching CachingOptions) (*Manager, error) { + sf := FormatterFactories[f.BlockFormat] + if sf == nil { + return nil, fmt.Errorf("unsupported block format: %v", f.BlockFormat) + } + + formatter, err := sf(f) + if err != nil { + return nil, err + } + return &Manager{ - storage: st, + Format: f, + openPackGroups: make(map[string]*packInfo), timeNow: time.Now, flushPackIndexesAfter: time.Now().Add(flushPackIndexTimeout), pendingPackIndexes: nil, - maxPackedContentLength: maxPackedContentLength, - maxPackSize: maxPackSize, + maxPackedContentLength: f.MaxPackedContentLength, + maxPackSize: f.MaxPackSize, formatter: formatter, - } + cache: newBlockCache(st, caching), + }, nil } diff --git a/block/block_manager_test.go b/block/block_manager_test.go index 06e18148e..71536e82e 100644 --- a/block/block_manager_test.go +++ b/block/block_manager_test.go @@ -614,10 +614,15 @@ func TestDeleteAndRecreate(t *testing.T) { func newTestBlockManager(data map[string][]byte, keyTime map[string]time.Time) *Manager { st := storagetesting.NewMapStorage(data, keyTime) - - f := &unencryptedFormat{computeHash(md5.New, md5.Size)} //st = logging.NewWrapper(st) - bm := NewManager(st, maxPackedContentLength, maxPackSize, f) + bm, err := NewManager(st, FormattingOptions{ + BlockFormat: "TESTONLY_MD5", + MaxPackedContentLength: maxPackedContentLength, + MaxPackSize: maxPackSize, + }, CachingOptions{}) + if err != nil { + panic("can't create block manager: " + err.Error()) + } setFakeTime(bm, fakeTime) return bm diff --git a/block/disk_block_cache.go b/block/disk_block_cache.go new file mode 100644 index 000000000..56da7ff54 --- /dev/null +++ b/block/disk_block_cache.go @@ -0,0 +1,260 @@ +package block + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "io/ioutil" + "log" + "math/rand" + "os" + "path/filepath" + "sort" + "strings" + "time" + + "github.com/kopia/kopia/storage" +) + +const ( + sweepCacheFrequency = 1 * time.Minute + cachedSuffix = ".cached" +) + +type diskBlockCache struct { + st storage.Storage + directory string + maxSizeBytes int64 + listCacheDuration time.Duration + hmacSecret []byte + + closed chan struct{} +} + +func (c *diskBlockCache) getBlock(blockID string, offset, length int64) ([]byte, error) { + fn := filepath.Join(c.directory, blockID) + cachedSuffix + + b, err := ioutil.ReadFile(fn) + if err == nil { + b, err := c.verifyHMAC(b) + if err == nil { + // retrieved from blockCache and HMAC valid + return applyOffsetAndLength(b, offset, length) + } + + // ignore malformed blocks + log.Printf("warning: malformed block %v: %v", blockID, err) + } else if !os.IsNotExist(err) { + log.Printf("warning: unable to read blockCache file %v: %v", fn, err) + } + + b, err = c.st.GetBlock(blockID, 0, -1) + if err == storage.ErrBlockNotFound { + // not found in underlying storage + return nil, err + } + + if err == nil { + //log.Printf("adding %v to blockCache", blockID) + if err := c.writeFileAtomic(fn, c.appendHMAC(b)); err != nil { + log.Printf("warning: unable to write file %v: %v", fn, err) + } + } + + return applyOffsetAndLength(b, offset, length) +} + +func applyOffsetAndLength(b []byte, offset, length int64) ([]byte, error) { + if offset > int64(len(b)) { + return nil, fmt.Errorf("offset of bounds (offset=%v, length=%v, actual length=%v)", offset, length, len(b)) + } + + if length < 0 { + return b[offset:], nil + } + + if offset+length > int64(len(b)) { + return nil, fmt.Errorf("length of bounds (offset=%v, length=%v, actual length=%v)", offset, length, len(b)) + } + + return b[offset : offset+length], nil +} + +func (c *diskBlockCache) putBlock(blockID string, data []byte) error { + err := c.st.PutBlock(blockID, data) + if err != nil { + return err + } + + c.writeFileAtomic(filepath.Join(c.directory, blockID)+cachedSuffix, c.appendHMAC(data)) + + return nil +} + +func (c *diskBlockCache) listIndexBlocks() ([]Info, error) { + cachedListFile := filepath.Join(c.directory, "list"+cachedSuffix) + f, err := os.Open(cachedListFile) + if err == nil { + defer f.Close() + + st, err := f.Stat() + if err == nil { + expirationTime := st.ModTime().UTC().Add(c.listCacheDuration) + if time.Now().UTC().Before(expirationTime) { + return c.readBlocksFromCacheFile(f) + } + } + } + + blocks, err := c.readBlocksFromSource() + if err == nil { + // save to blockCache + if data, err := json.Marshal(blocks); err == nil { + if err := c.writeFileAtomic(cachedListFile, c.appendHMAC(data)); err != nil { + log.Printf("warning: can't save list: %v", err) + } + } + } + + return blocks, err +} + +func (c *diskBlockCache) readBlocksFromCacheFile(f *os.File) ([]Info, error) { + var blocks []Info + data, err := ioutil.ReadAll(f) + if err != nil { + return nil, err + } + + data, err = c.verifyHMAC(data) + if err != nil { + return nil, err + } + + if err := json.Unmarshal(data, &blocks); err != nil { + return nil, fmt.Errorf("can't unmarshal cached list results: %v", err) + } + + return blocks, nil + +} + +func (c *diskBlockCache) readBlocksFromSource() ([]Info, error) { + var blocks []Info + ch, cancel := c.st.ListBlocks(packBlockPrefix) + defer cancel() + + for e := range ch { + if e.Error != nil { + return nil, e.Error + } + + blocks = append(blocks, Info{ + BlockID: e.BlockID, + Length: e.Length, + Timestamp: e.TimeStamp, + }) + } + return blocks, nil +} + +func (c *diskBlockCache) appendHMAC(data []byte) []byte { + h := hmac.New(sha256.New, c.hmacSecret) + h.Write(data) + validSignature := h.Sum(nil) + return append(append([]byte(nil), data...), validSignature...) +} + +func (c *diskBlockCache) verifyHMAC(b []byte) ([]byte, error) { + if len(b) < sha256.Size { + return nil, errors.New("invalid data - too short") + } + + p := len(b) - sha256.Size + data := b[0:p] + signature := b[p:] + h := hmac.New(sha256.New, c.hmacSecret) + h.Write(data) + validSignature := h.Sum(nil) + if len(signature) != len(validSignature) { + return nil, errors.New("invalid signature length") + } + if hmac.Equal(validSignature, signature) { + return data, nil + } + + return nil, errors.New("invalid data - corrupted") +} + +func (c *diskBlockCache) writeFileAtomic(fname string, contents []byte) error { + tn := filepath.Join(c.directory, fmt.Sprintf("tmp-%v.%v"+cachedSuffix, time.Now().UnixNano(), rand.Int63())) + if err := ioutil.WriteFile(tn, contents, 0600); err != nil { + if !os.IsNotExist(err) { + return err + } + + // create blockCache directory, and retry write + os.MkdirAll(c.directory, 0700) + if err := ioutil.WriteFile(tn, contents, 0600); err != nil { + return err + } + } + + if err := os.Rename(tn, fname); err != nil { + os.Remove(tn) + return err + } + + return nil +} + +func (c *diskBlockCache) lose() error { + close(c.closed) + return nil +} + +func (c *diskBlockCache) sweepDirectoryPeriodically() { + for { + select { + case <-c.closed: + return + + case <-time.After(sweepCacheFrequency): + if err := c.sweepDirectory(); err != nil { + log.Printf("warning: blockCache sweep failed: %v", err) + } + } + } +} + +func (c *diskBlockCache) sweepDirectory() error { + items, err := ioutil.ReadDir(c.directory) + if os.IsNotExist(err) { + // blockCache not found, that's ok + return nil + } + if err != nil { + return err + } + + sort.Slice(items, func(i, j int) bool { + return items[i].ModTime().Before(items[j].ModTime()) + }) + + var totalSize int64 + for _, it := range items { + if !strings.HasSuffix(it.Name(), cachedSuffix) { + continue + } + if totalSize > c.maxSizeBytes { + fn := filepath.Join(c.directory, it.Name()) + if err := os.Remove(fn); err != nil { + log.Printf("warning: unable to remove %v: %v", fn, err) + } + } + totalSize += it.Size() + } + return nil +} diff --git a/block/null_block_cache.go b/block/null_block_cache.go new file mode 100644 index 000000000..9b5e57715 --- /dev/null +++ b/block/null_block_cache.go @@ -0,0 +1,35 @@ +package block + +import "github.com/kopia/kopia/storage" + +type nullBlockCache struct { + st storage.Storage +} + +func (c nullBlockCache) getBlock(blockID string, offset, length int64) ([]byte, error) { + return c.st.GetBlock(blockID, offset, length) +} + +func (c nullBlockCache) putBlock(blockID string, data []byte) error { + return c.st.PutBlock(blockID, data) +} + +func (c nullBlockCache) listIndexBlocks() ([]Info, error) { + ch, cancel := c.st.ListBlocks(packBlockPrefix) + defer cancel() + + var results []Info + for it := range ch { + if it.Error != nil { + return nil, it.Error + } + + results = append(results, Info{ + BlockID: it.BlockID, + Timestamp: it.TimeStamp, + Length: it.Length, + }) + } + + return results, nil +} diff --git a/cli/command_repository_connect.go b/cli/command_repository_connect.go index 8e1e499e2..c2fa14252 100644 --- a/cli/command_repository_connect.go +++ b/cli/command_repository_connect.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "os" + "time" "github.com/kopia/kopia/repo" @@ -15,6 +16,8 @@ var ( connectRepositoryLocation = connectCommand.Arg("location", "Repository address").Required().String() connectDontPersistCredentials bool connectCacheDirectory string + connectMaxCacheSizeMB int64 + connectMaxListCacheDuration time.Duration // options shared by various providers connectCredentialsFile string @@ -44,12 +47,18 @@ func setupConnectOptions(cmd *kingpin.CmdClause) { cmd.Flag("max-download-speed", "Limit the download speed.").PlaceHolder("BYTES_PER_SEC").IntVar(&connectMaxDownloadSpeedBytesPerSecond) cmd.Flag("max-upload-speed", "Limit the upload speed.").PlaceHolder("BYTES_PER_SEC").IntVar(&connectMaxUploadSpeedBytesPerSecond) + + cmd.Flag("cache-size", "Size of local cache").PlaceHolder("MB").Int64Var(&connectMaxCacheSizeMB) + cmd.Flag("max-list-cache-duration", "Duration of index cache").Default("600s").DurationVar(&connectMaxListCacheDuration) } func connectOptions() repo.ConnectOptions { return repo.ConnectOptions{ PersistCredentials: !connectDontPersistCredentials, - CacheDirectory: connectCacheDirectory, + + CacheDirectory: connectCacheDirectory, + MaxCacheSizeBytes: connectMaxCacheSizeMB << 20, + MaxListDuration: connectMaxListCacheDuration, } } diff --git a/cli/command_repository_status.go b/cli/command_repository_status.go index e5572c23e..29c7bff29 100644 --- a/cli/command_repository_status.go +++ b/cli/command_repository_status.go @@ -45,15 +45,14 @@ func runStatusCommand(context *kingpin.ParseContext) error { } fmt.Println() - fmt.Printf("Metadata manager: v%v\n", rep.Metadata.Format.Version) - fmt.Printf("Metadata Encryption: %v\n", rep.Metadata.Format.EncryptionAlgorithm) fmt.Printf("Key Derivation: %v\n", rep.Security.KeyDerivationAlgorithm) fmt.Printf("Unique ID: %x\n", rep.Security.UniqueID) fmt.Println() fmt.Printf("Object manager: v%v\n", rep.Objects.Format.Version) - fmt.Printf("Block format: %v\n", rep.Objects.Format.BlockFormat) + fmt.Printf("Block format: %v\n", rep.Blocks.Format.BlockFormat) + fmt.Printf("Max packed len: %v\n", units.BytesStringBase2(int64(rep.Blocks.Format.MaxPackedContentLength))) + fmt.Printf("Max pack length: %v\n", units.BytesStringBase2(int64(rep.Blocks.Format.MaxPackSize))) fmt.Printf("Splitter: %v%v\n", rep.Objects.Format.Splitter, splitterExtraInfo) - fmt.Printf("Max packed len: %v\n", units.BytesStringBase2(int64(rep.Objects.Format.MaxPackedContentLength))) return nil } diff --git a/internal/config/local_config.go b/internal/config/local_config.go index 9983a923f..08868ab81 100644 --- a/internal/config/local_config.go +++ b/internal/config/local_config.go @@ -11,8 +11,8 @@ import ( // LocalConfig is a configuration of Kopia. type LocalConfig struct { - Connection *RepositoryConnectionInfo `json:"connection,omitempty"` - CacheDirectory string `json:"cacheDirectory,omitempty"` + Connection *RepositoryConnectionInfo `json:"connection,omitempty"` + Caching block.CachingOptions `json:"caching"` } // RepositoryObjectFormat describes the format of objects in a repository. diff --git a/manifest/manifest_manager_test.go b/manifest/manifest_manager_test.go index f98123e89..75d11daf8 100644 --- a/manifest/manifest_manager_test.go +++ b/manifest/manifest_manager_test.go @@ -1,6 +1,7 @@ package manifest import ( + "fmt" "reflect" "sort" "testing" @@ -165,11 +166,16 @@ func verifyMatches(t *testing.T, mgr *Manager, labels map[string]string, expecte } func newManagerForTesting(t *testing.T, data map[string][]byte, keyTime map[string]time.Time) (*Manager, error) { - formatter, err := block.FormatterFactories["TESTONLY_MD5"](block.FormattingOptions{}) - if err != nil { - panic("can't create formatter") - } st := storagetesting.NewMapStorage(data, keyTime) - return NewManager(block.NewManager(st, 10000, 100000, formatter)) + bm, err := block.NewManager(st, block.FormattingOptions{ + BlockFormat: "TESTONLY_MD5", + MaxPackedContentLength: 10000, + MaxPackSize: 100000, + }, block.CachingOptions{}) + if err != nil { + return nil, fmt.Errorf("can't create block manager: %v", err) + } + + return NewManager(bm) } diff --git a/repo/connect.go b/repo/connect.go index 6afaf0324..33b68d605 100644 --- a/repo/connect.go +++ b/repo/connect.go @@ -6,11 +6,14 @@ import ( "errors" "fmt" "io/ioutil" + "log" "os" "path/filepath" + "time" "github.com/kopia/kopia/auth" "github.com/kopia/kopia/internal/config" + "github.com/kopia/kopia/internal/units" "github.com/kopia/kopia/storage" // Register well-known blob storage providers @@ -21,7 +24,10 @@ import ( // ConnectOptions specifies options when persisting configuration to connect to a repository. type ConnectOptions struct { PersistCredentials bool - CacheDirectory string + + CacheDirectory string + MaxCacheSizeBytes int64 + MaxListDuration time.Duration } // Connect connects to the repository in the specified storage and persists the configuration and credentials in the file provided. @@ -31,7 +37,12 @@ func Connect(ctx context.Context, configFile string, st storage.Storage, creds a return errors.New("repository does not support persisting configuration") } - f, err := readFormatBlock(st) + formatBytes, err := st.GetBlock(formatBlockID, 0, -1) + if err != nil { + return fmt.Errorf("unable to read format block: %v", err) + } + + f, err := parseFormatBlock(formatBytes) if err != nil { return err } @@ -53,6 +64,29 @@ func Connect(ctx context.Context, configFile string, st storage.Storage, creds a lc.Connection.Key = nil } + if opt.MaxCacheSizeBytes > 0 { + if opt.CacheDirectory == "" { + // derive cache directory from config + absConfig, err := filepath.Abs(configFile) + if err != nil { + return err + } + lc.Caching.CacheDirectory = filepath.Join(filepath.Dir(absConfig), fmt.Sprintf("cache-%x", f.UniqueID)) + } else { + absCacheDir, err := filepath.Abs(opt.CacheDirectory) + if err != nil { + return err + } + + lc.Caching.CacheDirectory = absCacheDir + } + lc.Caching.MaxCacheSizeBytes = opt.MaxCacheSizeBytes + lc.Caching.MaxListCacheDurationSec = int(opt.MaxListDuration.Seconds()) + + log.Printf("Creating cache directory '%v' with max size %v", lc.Caching.CacheDirectory, units.BytesStringBase2(lc.Caching.MaxCacheSizeBytes)) + os.MkdirAll(lc.Caching.CacheDirectory, 0700) + } + d, err := json.MarshalIndent(&lc, "", " ") if err != nil { return err diff --git a/repo/format_block.go b/repo/format_block.go index 5bf50430c..54012a673 100644 --- a/repo/format_block.go +++ b/repo/format_block.go @@ -41,18 +41,14 @@ type encryptedRepositoryConfig struct { Format config.RepositoryObjectFormat `json:"format"` } -func readFormatBlock(st storage.Storage) (*formatBlock, error) { +func parseFormatBlock(b []byte) (*formatBlock, error) { f := &formatBlock{} - b, err := st.GetBlock(formatBlockID, 0, -1) - if err != nil { - return nil, fmt.Errorf("unable to read format block: %v", err) - } - if err := json.Unmarshal(b, &f); err != nil { return nil, fmt.Errorf("invalid format block: %v", err) } - return f, err + + return f, nil } func writeFormatBlock(st storage.Storage, f *formatBlock) error { diff --git a/repo/initialize.go b/repo/initialize.go index 49766fde9..333c0f995 100644 --- a/repo/initialize.go +++ b/repo/initialize.go @@ -83,7 +83,8 @@ func repositoryObjectFormatFromOptions(opt *NewRepositoryOptions) *config.Reposi BlockFormat: applyDefaultString(opt.BlockFormat, block.DefaultFormat), HMACSecret: applyDefaultRandomBytes(opt.ObjectHMACSecret, 32), MasterKey: applyDefaultRandomBytes(opt.ObjectEncryptionKey, 32), - MaxPackedContentLength: applyDefaultInt(opt.MaxPackedContentLength, 4<<20), // 3 MB + MaxPackedContentLength: applyDefaultInt(opt.MaxPackedContentLength, 4<<20), // 4 MB + MaxPackSize: applyDefaultInt(opt.MaxBlockSize, 20<<20), // 20 MB }, Splitter: applyDefaultString(opt.Splitter, object.DefaultSplitter), MaxBlockSize: applyDefaultInt(opt.MaxBlockSize, 20<<20), // 20MiB diff --git a/repo/open.go b/repo/open.go index 8c5d73121..296dfe013 100644 --- a/repo/open.go +++ b/repo/open.go @@ -4,11 +4,13 @@ import ( "context" "errors" "fmt" + "io/ioutil" + "log" "path/filepath" "github.com/kopia/kopia/manifest" - "github.com/kopia/kopia/metadata" + "github.com/kopia/kopia/object" "github.com/kopia/kopia/auth" @@ -65,19 +67,49 @@ func Open(ctx context.Context, configFile string, options *Options) (*Repository return nil, fmt.Errorf("cannot open storage: %v", err) } - r, err := connect(ctx, st, creds, options) + caching := lc.Caching + r, err := connect(ctx, st, creds, options, caching) if err != nil { st.Close() return nil, err } r.ConfigFile = configFile - r.CacheDirectory = applyDefaultString(lc.CacheDirectory, filepath.Join(filepath.Dir(configFile), "cache")) return r, nil } -func connect(ctx context.Context, st storage.Storage, creds auth.Credentials, options *Options) (*Repository, error) { +func readAndCacheFormatBlock(st storage.Storage, cacheDirectory string) (*formatBlock, error) { + cachedFile := filepath.Join(cacheDirectory, "kopia.repository") + if cacheDirectory != "" { + b, err := ioutil.ReadFile(cachedFile) + if err == nil { + // read from cache. + return parseFormatBlock(b) + } + } + + b, err := st.GetBlock(formatBlockID, 0, -1) + if err != nil { + return nil, err + } + + // block successfully read from storage. + f, err := parseFormatBlock(b) + if err != nil { + return nil, err + } + + if cacheDirectory != "" { + if err := ioutil.WriteFile(cachedFile, b, 0600); err != nil { + log.Printf("warning: unable to write cache: %v", err) + } + } + + return f, nil +} + +func connect(ctx context.Context, st storage.Storage, creds auth.Credentials, options *Options, caching block.CachingOptions) (*Repository, error) { if options == nil { options = &Options{} } @@ -85,9 +117,10 @@ func connect(ctx context.Context, st storage.Storage, creds auth.Credentials, op st = logging.NewWrapper(st, logging.Prefix("[STORAGE] "), logging.Output(options.TraceStorage)) } - f, err := readFormatBlock(st) + // Read cache block, potentially from cache. + f, err := readAndCacheFormatBlock(st, caching.CacheDirectory) if err != nil { - return nil, err + return nil, fmt.Errorf("unable to read format block: %v", err) } km, err := auth.NewKeyManager(creds, f.SecurityOptions) @@ -108,18 +141,18 @@ func connect(ctx context.Context, st storage.Storage, creds auth.Credentials, op return nil, fmt.Errorf("unable to decrypt repository config: %v", err) } - sf := block.FormatterFactories[repoConfig.BlockFormat] - if sf == nil { - return nil, fmt.Errorf("unsupported block format: %v", repoConfig.BlockFormat) + fo := repoConfig.FormattingOptions + if fo.MaxPackSize == 0 { + fo.MaxPackSize = repoConfig.MaxBlockSize } - formatter, err := sf(repoConfig.FormattingOptions) + caching.HMACSecret = km.DeriveKey([]byte("local-cache-integrity"), 16) + + bm, err := block.NewManager(st, fo, caching) if err != nil { - return nil, err + return nil, fmt.Errorf("unable to open block manager: %v", err) } - bm := block.NewManager(st, repoConfig.MaxPackedContentLength, repoConfig.MaxBlockSize, formatter) - om, err := object.NewObjectManager(bm, *repoConfig, options.ObjectManagerOptions) if err != nil { return nil, fmt.Errorf("unable to open object manager: %v", err) @@ -131,12 +164,13 @@ func connect(ctx context.Context, st storage.Storage, creds auth.Credentials, op } return &Repository{ - Blocks: bm, - Objects: om, - Metadata: mm, - Storage: st, - KeyManager: km, - Manifests: manifests, - Security: f.SecurityOptions, + Blocks: bm, + Objects: om, + Metadata: mm, + Storage: st, + KeyManager: km, + Manifests: manifests, + Security: f.SecurityOptions, + CacheDirectory: caching.CacheDirectory, }, nil } diff --git a/repo/repository_test.go b/repo/repository_test.go index eac19de72..2cbcc2fca 100644 --- a/repo/repository_test.go +++ b/repo/repository_test.go @@ -15,6 +15,8 @@ import ( "testing" "time" + "github.com/kopia/kopia/block" + "github.com/kopia/kopia/auth" "github.com/kopia/kopia/object" @@ -49,7 +51,7 @@ func setupTestWithData(t *testing.T, data map[string][]byte, keyTime map[string] r, err := connect(ctx, st, creds, &Options{ //TraceStorage: log.Printf, - }) + }, block.CachingOptions{}) if err != nil { t.Fatalf("can't connect: %v", err) }