mirror of
https://github.com/kopia/kopia.git
synced 2026-10-20 21:24:43 -04:00
refactored block manager to support on-disk caching
This commit is contained in:
1 parent
a0cc09b673
commit
30c11dc926
15 files changed
+511
-80
No files matched your search
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
+42
-34
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
+36
-2
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
+2
-1
@@ -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
|
||||
|
||||
+54
-20
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user