blob: 8897fe16d89bfad96183a5fdb071b726223a7dab [file] [log] [blame]
package fscache // import "github.com/docker/docker/builder/fscache"
import (
"archive/tar"
"context"
"crypto/sha256"
"encoding/json"
"hash"
"os"
"path/filepath"
"sort"
"sync"
"time"
"github.com/docker/docker/builder"
"github.com/docker/docker/builder/remotecontext"
"github.com/docker/docker/pkg/archive"
"github.com/docker/docker/pkg/directory"
"github.com/docker/docker/pkg/stringid"
"github.com/docker/docker/pkg/tarsum"
"github.com/moby/buildkit/session/filesync"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/tonistiigi/fsutil"
fsutiltypes "github.com/tonistiigi/fsutil/types"
bolt "go.etcd.io/bbolt"
"golang.org/x/sync/singleflight"
)
const dbFile = "fscache.db"
const cacheKey = "cache"
const metaKey = "meta"
// Backend is a backing implementation for FSCache
type Backend interface {
Get(id string) (string, error)
Remove(id string) error
}
// FSCache allows syncing remote resources to cached snapshots
type FSCache struct {
opt Opt
transports map[string]Transport
mu sync.Mutex
g singleflight.Group
store *fsCacheStore
}
// Opt defines options for initializing FSCache
type Opt struct {
Backend Backend
Root string // for storing local metadata
GCPolicy GCPolicy
}
// GCPolicy defines policy for garbage collection
type GCPolicy struct {
MaxSize uint64
MaxKeepDuration time.Duration
}
// NewFSCache returns new FSCache object
func NewFSCache(opt Opt) (*FSCache, error) {
store, err := newFSCacheStore(opt)
if err != nil {
return nil, err
}
return &FSCache{
store: store,
opt: opt,
transports: make(map[string]Transport),
}, nil
}
// Transport defines a method for syncing remote data to FSCache
type Transport interface {
Copy(ctx context.Context, id RemoteIdentifier, dest string, cs filesync.CacheUpdater) error
}
// RemoteIdentifier identifies a transfer request
type RemoteIdentifier interface {
Key() string
SharedKey() string
Transport() string
}
// RegisterTransport registers a new transport method
func (fsc *FSCache) RegisterTransport(id string, transport Transport) error {
fsc.mu.Lock()
defer fsc.mu.Unlock()
if _, ok := fsc.transports[id]; ok {
return errors.Errorf("transport %v already exists", id)
}
fsc.transports[id] = transport
return nil
}
// SyncFrom returns a source based on a remote identifier
func (fsc *FSCache) SyncFrom(ctx context.Context, id RemoteIdentifier) (builder.Source, error) { // cacheOpt
trasportID := id.Transport()
fsc.mu.Lock()
transport, ok := fsc.transports[id.Transport()]
if !ok {
fsc.mu.Unlock()
return nil, errors.Errorf("invalid transport %s", trasportID)
}
logrus.Debugf("SyncFrom %s %s", id.Key(), id.SharedKey())
fsc.mu.Unlock()
sourceRef, err, _ := fsc.g.Do(id.Key(), func() (interface{}, error) {
var sourceRef *cachedSourceRef
sourceRef, err := fsc.store.Get(id.Key())
if err == nil {
return sourceRef, nil
}
// check for unused shared cache
sharedKey := id.SharedKey()
if sharedKey != "" {
r, err := fsc.store.Rebase(sharedKey, id.Key())
if err == nil {
sourceRef = r
}
}
if sourceRef == nil {
var err error
sourceRef, err = fsc.store.New(id.Key(), sharedKey)
if err != nil {
return nil, errors.Wrap(err, "failed to create remote context")
}
}
if err := syncFrom(ctx, sourceRef, transport, id); err != nil {
sourceRef.Release()
return nil, err
}
if err := sourceRef.resetSize(-1); err != nil {
return nil, err
}
return sourceRef, nil
})
if err != nil {
return nil, err
}
ref := sourceRef.(*cachedSourceRef)
if ref.src == nil { // failsafe
return nil, errors.Errorf("invalid empty pull")
}
wc := &wrappedContext{Source: ref.src, closer: func() error {
ref.Release()
return nil
}}
return wc, nil
}
// DiskUsage reports how much data is allocated by the cache
func (fsc *FSCache) DiskUsage(ctx context.Context) (int64, error) {
return fsc.store.DiskUsage(ctx)
}
// Prune allows manually cleaning up the cache
func (fsc *FSCache) Prune(ctx context.Context) (uint64, error) {
return fsc.store.Prune(ctx)
}
// Close stops the gc and closes the persistent db
func (fsc *FSCache) Close() error {
return fsc.store.Close()
}
func syncFrom(ctx context.Context, cs *cachedSourceRef, transport Transport, id RemoteIdentifier) (retErr error) {
src := cs.src
if src == nil {
src = remotecontext.NewCachableSource(cs.Dir())
}
if !cs.cached {
if err := cs.storage.db.View(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte(id.Key()))
dt := b.Get([]byte(cacheKey))
if dt != nil {
if err := src.UnmarshalBinary(dt); err != nil {
return err
}
} else {
return errors.Wrap(src.Scan(), "failed to scan cache records")
}
return nil
}); err != nil {
return err
}
}
dc := &detectChanges{f: src.HandleChange}
// todo: probably send a bucket to `Copy` and let it return source
// but need to make sure that tx is safe
if err := transport.Copy(ctx, id, cs.Dir(), dc); err != nil {
return errors.Wrapf(err, "failed to copy to %s", cs.Dir())
}
if !dc.supported {
if err := src.Scan(); err != nil {
return errors.Wrap(err, "failed to scan cache records after transfer")
}
}
cs.cached = true
cs.src = src
return cs.storage.db.Update(func(tx *bolt.Tx) error {
dt, err := src.MarshalBinary()
if err != nil {
return err
}
b := tx.Bucket([]byte(id.Key()))
return b.Put([]byte(cacheKey), dt)
})
}
type fsCacheStore struct {
mu sync.Mutex
sources map[string]*cachedSource
db *bolt.DB
fs Backend
gcTimer *time.Timer
gcPolicy GCPolicy
}
// CachePolicy defines policy for keeping a resource in cache
type CachePolicy struct {
Priority int
LastUsed time.Time
}
func defaultCachePolicy() CachePolicy {
return CachePolicy{Priority: 10, LastUsed: time.Now()}
}
func newFSCacheStore(opt Opt) (*fsCacheStore, error) {
if err := os.MkdirAll(opt.Root, 0700); err != nil {
return nil, err
}
p := filepath.Join(opt.Root, dbFile)
db, err := bolt.Open(p, 0600, nil)
if err != nil {
return nil, errors.Wrap(err, "failed to open database file %s")
}
s := &fsCacheStore{db: db, sources: make(map[string]*cachedSource), fs: opt.Backend, gcPolicy: opt.GCPolicy}
db.View(func(tx *bolt.Tx) error {
return tx.ForEach(func(name []byte, b *bolt.Bucket) error {
dt := b.Get([]byte(metaKey))
if dt == nil {
return nil
}
var sm sourceMeta
if err := json.Unmarshal(dt, &sm); err != nil {
return err
}
dir, err := s.fs.Get(sm.BackendID)
if err != nil {
return err // TODO: handle gracefully
}
source := &cachedSource{
refs: make(map[*cachedSourceRef]struct{}),
id: string(name),
dir: dir,
sourceMeta: sm,
storage: s,
}
s.sources[string(name)] = source
return nil
})
})
s.gcTimer = s.startPeriodicGC(5 * time.Minute)
return s, nil
}
func (s *fsCacheStore) startPeriodicGC(interval time.Duration) *time.Timer {
var t *time.Timer
t = time.AfterFunc(interval, func() {
if err := s.GC(); err != nil {
logrus.Errorf("build gc error: %v", err)
}
t.Reset(interval)
})
return t
}
func (s *fsCacheStore) Close() error {
s.gcTimer.Stop()
return s.db.Close()
}
func (s *fsCacheStore) New(id, sharedKey string) (*cachedSourceRef, error) {
s.mu.Lock()
defer s.mu.Unlock()
var ret *cachedSource
if err := s.db.Update(func(tx *bolt.Tx) error {
b, err := tx.CreateBucket([]byte(id))
if err != nil {
return err
}
backendID := stringid.GenerateRandomID()
dir, err := s.fs.Get(backendID)
if err != nil {
return err
}
source := &cachedSource{
refs: make(map[*cachedSourceRef]struct{}),
id: id,
dir: dir,
sourceMeta: sourceMeta{
BackendID: backendID,
SharedKey: sharedKey,
CachePolicy: defaultCachePolicy(),
},
storage: s,
}
dt, err := json.Marshal(source.sourceMeta)
if err != nil {
return err
}
if err := b.Put([]byte(metaKey), dt); err != nil {
return err
}
s.sources[id] = source
ret = source
return nil
}); err != nil {
return nil, err
}
return ret.getRef(), nil
}
func (s *fsCacheStore) Rebase(sharedKey, newid string) (*cachedSourceRef, error) {
s.mu.Lock()
defer s.mu.Unlock()
var ret *cachedSource
for id, snap := range s.sources {
if snap.SharedKey == sharedKey && len(snap.refs) == 0 {
if err := s.db.Update(func(tx *bolt.Tx) error {
if err := tx.DeleteBucket([]byte(id)); err != nil {
return err
}
b, err := tx.CreateBucket([]byte(newid))
if err != nil {
return err
}
snap.id = newid
snap.CachePolicy = defaultCachePolicy()
dt, err := json.Marshal(snap.sourceMeta)
if err != nil {
return err
}
if err := b.Put([]byte(metaKey), dt); err != nil {
return err
}
delete(s.sources, id)
s.sources[newid] = snap
return nil
}); err != nil {
return nil, err
}
ret = snap
break
}
}
if ret == nil {
return nil, errors.Errorf("no candidate for rebase")
}
return ret.getRef(), nil
}
func (s *fsCacheStore) Get(id string) (*cachedSourceRef, error) {
s.mu.Lock()
defer s.mu.Unlock()
src, ok := s.sources[id]
if !ok {
return nil, errors.Errorf("not found")
}
return src.getRef(), nil
}
// DiskUsage reports how much data is allocated by the cache
func (s *fsCacheStore) DiskUsage(ctx context.Context) (int64, error) {
s.mu.Lock()
defer s.mu.Unlock()
var size int64
for _, snap := range s.sources {
if len(snap.refs) == 0 {
ss, err := snap.getSize(ctx)
if err != nil {
return 0, err
}
size += ss
}
}
return size, nil
}
// Prune allows manually cleaning up the cache
func (s *fsCacheStore) Prune(ctx context.Context) (uint64, error) {
s.mu.Lock()
defer s.mu.Unlock()
var size uint64
for id, snap := range s.sources {
select {
case <-ctx.Done():
logrus.Debugf("Cache prune operation cancelled, pruned size: %d", size)
// when the context is cancelled, only return current size and nil
return size, nil
default:
}
if len(snap.refs) == 0 {
ss, err := snap.getSize(ctx)
if err != nil {
return size, err
}
if err := s.delete(id); err != nil {
return size, errors.Wrapf(err, "failed to delete %s", id)
}
size += uint64(ss)
}
}
return size, nil
}
// GC runs a garbage collector on FSCache
func (s *fsCacheStore) GC() error {
s.mu.Lock()
defer s.mu.Unlock()
var size uint64
ctx := context.Background()
cutoff := time.Now().Add(-s.gcPolicy.MaxKeepDuration)
var blacklist []*cachedSource
for id, snap := range s.sources {
if len(snap.refs) == 0 {
if cutoff.After(snap.CachePolicy.LastUsed) {
if err := s.delete(id); err != nil {
return errors.Wrapf(err, "failed to delete %s", id)
}
} else {
ss, err := snap.getSize(ctx)
if err != nil {
return err
}
size += uint64(ss)
blacklist = append(blacklist, snap)
}
}
}
sort.Sort(sortableCacheSources(blacklist))
for _, snap := range blacklist {
if size <= s.gcPolicy.MaxSize {
break
}
ss, err := snap.getSize(ctx)
if err != nil {
return err
}
if err := s.delete(snap.id); err != nil {
return errors.Wrapf(err, "failed to delete %s", snap.id)
}
size -= uint64(ss)
}
return nil
}
// keep mu while calling this
func (s *fsCacheStore) delete(id string) error {
src, ok := s.sources[id]
if !ok {
return nil
}
if len(src.refs) > 0 {
return errors.Errorf("can't delete %s because it has active references", id)
}
delete(s.sources, id)
if err := s.db.Update(func(tx *bolt.Tx) error {
return tx.DeleteBucket([]byte(id))
}); err != nil {
return err
}
return s.fs.Remove(src.BackendID)
}
type sourceMeta struct {
SharedKey string
BackendID string
CachePolicy CachePolicy
Size int64
}
type cachedSource struct {
sourceMeta
refs map[*cachedSourceRef]struct{}
id string
dir string
src *remotecontext.CachableSource
storage *fsCacheStore
cached bool // keep track if cache is up to date
}
type cachedSourceRef struct {
*cachedSource
}
func (cs *cachedSource) Dir() string {
return cs.dir
}
// hold storage lock before calling
func (cs *cachedSource) getRef() *cachedSourceRef {
ref := &cachedSourceRef{cachedSource: cs}
cs.refs[ref] = struct{}{}
return ref
}
// hold storage lock before calling
func (cs *cachedSource) getSize(ctx context.Context) (int64, error) {
if cs.sourceMeta.Size < 0 {
ss, err := directory.Size(ctx, cs.dir)
if err != nil {
return 0, err
}
if err := cs.resetSize(ss); err != nil {
return 0, err
}
return ss, nil
}
return cs.sourceMeta.Size, nil
}
func (cs *cachedSource) resetSize(val int64) error {
cs.sourceMeta.Size = val
return cs.saveMeta()
}
func (cs *cachedSource) saveMeta() error {
return cs.storage.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte(cs.id))
dt, err := json.Marshal(cs.sourceMeta)
if err != nil {
return err
}
return b.Put([]byte(metaKey), dt)
})
}
func (csr *cachedSourceRef) Release() error {
csr.cachedSource.storage.mu.Lock()
defer csr.cachedSource.storage.mu.Unlock()
delete(csr.cachedSource.refs, csr)
if len(csr.cachedSource.refs) == 0 {
go csr.cachedSource.storage.GC()
}
return nil
}
type detectChanges struct {
f fsutil.ChangeFunc
supported bool
}
func (dc *detectChanges) HandleChange(kind fsutil.ChangeKind, path string, fi os.FileInfo, err error) error {
if dc == nil {
return nil
}
return dc.f(kind, path, fi, err)
}
func (dc *detectChanges) MarkSupported(v bool) {
if dc == nil {
return
}
dc.supported = v
}
func (dc *detectChanges) ContentHasher() fsutil.ContentHasher {
return newTarsumHash
}
type wrappedContext struct {
builder.Source
closer func() error
}
func (wc *wrappedContext) Close() error {
if err := wc.Source.Close(); err != nil {
return err
}
return wc.closer()
}
type sortableCacheSources []*cachedSource
// Len is the number of elements in the collection.
func (s sortableCacheSources) Len() int {
return len(s)
}
// Less reports whether the element with
// index i should sort before the element with index j.
func (s sortableCacheSources) Less(i, j int) bool {
return s[i].CachePolicy.LastUsed.Before(s[j].CachePolicy.LastUsed)
}
// Swap swaps the elements with indexes i and j.
func (s sortableCacheSources) Swap(i, j int) {
s[i], s[j] = s[j], s[i]
}
func newTarsumHash(stat *fsutiltypes.Stat) (hash.Hash, error) {
fi := &fsutil.StatInfo{Stat: stat}
p := stat.Path
if fi.IsDir() {
p += string(os.PathSeparator)
}
h, err := archive.FileInfoHeader(p, fi, stat.Linkname)
if err != nil {
return nil, err
}
h.Name = p
h.Uid = int(stat.Uid)
h.Gid = int(stat.Gid)
h.Linkname = stat.Linkname
if stat.Xattrs != nil {
h.Xattrs = make(map[string]string)
for k, v := range stat.Xattrs {
h.Xattrs[k] = string(v)
}
}
tsh := &tarsumHash{h: h, Hash: sha256.New()}
tsh.Reset()
return tsh, nil
}
// Reset resets the Hash to its initial state.
func (tsh *tarsumHash) Reset() {
tsh.Hash.Reset()
tarsum.WriteV1Header(tsh.h, tsh.Hash)
}
type tarsumHash struct {
hash.Hash
h *tar.Header
}