// +build !plan9,go1.7 package cache import (	"fmt"	"io"	"path"	"path/filepath"	"strings"	"sync"	"time" 	"os" 	"os/signal"	"syscall" 	"github.com/ncw/rclone/crypt"	"github.com/ncw/rclone/fs"	"github.com/pkg/errors"	"golang.org/x/net/context"	"golang.org/x/time/rate") const (	// DefCacheChunkSize is the default value for chunk size	DefCacheChunkSize = "5M"	// DefCacheTotalChunkSize is the default value for the maximum size of stored chunks	DefCacheTotalChunkSize = "10G"	// DefCacheChunkCleanInterval is the interval at which chunks are cleaned	DefCacheChunkCleanInterval = "1m"	// DefCacheInfoAge is the default value for object info age	DefCacheInfoAge = "6h"	// DefCacheReadRetries is the default value for read retries	DefCacheReadRetries = 10	// DefCacheTotalWorkers is how many workers run in parallel to download chunks	DefCacheTotalWorkers = 4	// DefCacheChunkNoMemory will enable or disable in-memory storage for chunks	DefCacheChunkNoMemory = false	// DefCacheRps limits the number of requests per second to the source FS	DefCacheRps = -1	// DefCacheWrites will cache file data on writes through the cache	DefCacheWrites = false) // Globalsvar (	// Flags	cacheDbPath             = fs.StringP("cache-db-path", "", filepath.Join(fs.CacheDir, "cache-backend"), "Directory to cache DB")	cacheChunkPath          = fs.StringP("cache-chunk-path", "", filepath.Join(fs.CacheDir, "cache-backend"), "Directory to cached chunk files")	cacheDbPurge            = fs.BoolP("cache-db-purge", "", false, "Purge the cache DB before")	cacheChunkSize          = fs.StringP("cache-chunk-size", "", DefCacheChunkSize, "The size of a chunk")	cacheTotalChunkSize     = fs.StringP("cache-total-chunk-size", "", DefCacheTotalChunkSize, "The total size which the chunks can take up from the disk")	cacheChunkCleanInterval = fs.StringP("cache-chunk-clean-interval", "", DefCacheChunkCleanInterval, "Interval at which chunk cleanup runs")	cacheInfoAge            = fs.StringP("cache-info-age", "", DefCacheInfoAge, "How much time should object info be stored in cache")	cacheReadRetries        = fs.IntP("cache-read-retries", "", DefCacheReadRetries, "How many times to retry a read from a cache storage")	cacheTotalWorkers       = fs.IntP("cache-workers", "", DefCacheTotalWorkers, "How many workers should run in parallel to download chunks")	cacheChunkNoMemory      = fs.BoolP("cache-chunk-no-memory", "", DefCacheChunkNoMemory, "Disable the in-memory cache for storing chunks during streaming")	cacheRps                = fs.IntP("cache-rps", "", int(DefCacheRps), "Limits the number of requests per second to the source FS. -1 disables the rate limiter")	cacheStoreWrites        = fs.BoolP("cache-writes", "", DefCacheWrites, "Will cache file data on writes through the FS")) // Register with Fsfunc init() {	fs.Register(&fs.RegInfo{		Name:        "cache",		Description: "Cache a remote",		NewFs:       NewFs,		Options: []fs.Option{{			Name: "remote",			Help: "Remote to cache.\nNormally should contain a ':' and a path, eg \"myremote:path/to/dir\",\n\"myremote:bucket\" or maybe \"myremote:\" (not recommended).",		}, {			Name:     "plex_url",			Help:     "Optional: The URL of the Plex server",			Optional: true,		}, {			Name:     "plex_username",			Help:     "Optional: The username of the Plex user",			Optional: true,		}, {			Name:       "plex_password",			Help:       "Optional: The password of the Plex user",			IsPassword: true,			Optional:   true,		}, {			Name: "chunk_size",			Help: "The size of a chunk. Lower value good for slow connections but can affect seamless reading. \nDefault: " + DefCacheChunkSize,			Examples: []fs.OptionExample{				{					Value: "1m",					Help:  "1MB",				}, {					Value: "5M",					Help:  "5 MB",				}, {					Value: "10M",					Help:  "10 MB",				},			},			Optional: true,		}, {			Name: "info_age",			Help: "How much time should object info (file size, file hashes etc) be stored in cache. Use a very high value if you don't plan on changing the source FS from outside the cache. \nAccepted units are: \"s\", \"m\", \"h\".\nDefault: " + DefCacheInfoAge,			Examples: []fs.OptionExample{				{					Value: "1h",					Help:  "1 hour",				}, {					Value: "24h",					Help:  "24 hours",				}, {					Value: "48h",					Help:  "48 hours",				},			},			Optional: true,		}, {			Name: "chunk_total_size",			Help: "The maximum size of stored chunks. When the storage grows beyond this size, the oldest chunks will be deleted. \nDefault: " + DefCacheTotalChunkSize,			Examples: []fs.OptionExample{				{					Value: "500M",					Help:  "500 MB",				}, {					Value: "1G",					Help:  "1 GB",				}, {					Value: "10G",					Help:  "10 GB",				},			},			Optional: true,		}},	})} // ChunkStorage is a storage type that supports only chunk operations (i.e in RAM)type ChunkStorage interface {	// will check if the chunk is in storage. should be fast and not read the chunk itself if possible	HasChunk(cachedObject *Object, offset int64) bool 	// returns the chunk in storage. return an error if it's not	GetChunk(cachedObject *Object, offset int64) ([]byte, error) 	// add a new chunk	AddChunk(fp string, data []byte, offset int64) error 	// if the storage can cleanup on a cron basis	// otherwise it can do a noop operation	CleanChunksByAge(chunkAge time.Duration) 	// if the storage can cleanup chunks after we no longer need them	// otherwise it can do a noop operation	CleanChunksByNeed(offset int64) 	// if the storage can cleanup chunks after the total size passes a certain point	// otherwise it can do a noop operation	CleanChunksBySize(maxSize int64)} // Storage is a storage type (Bolt) which needs to support both chunk and file based operationstype Storage interface {	ChunkStorage 	// will update/create a directory or an error if it's not found	AddDir(cachedDir *Directory) error 	// will return a directory with all the entries in it or an error if it's not found	GetDirEntries(cachedDir *Directory) (fs.DirEntries, error) 	// remove a directory and all the objects and chunks in it	RemoveDir(fp string) error 	// remove a directory and all the objects and chunks in it	ExpireDir(cd *Directory) error 	// will return an object (file) or error if it doesn't find it	GetObject(cachedObject *Object) (err error) 	// add a new object to its parent directory	// the directory structure (all the parents of this object) is created if its not found	AddObject(cachedObject *Object) error 	// remove an object and all its chunks	RemoveObject(fp string) error 	// Stats returns stats about the cache storage	Stats() (map[string]map[string]interface{}, error) 	// Purge will flush the entire cache	Purge() 	// Close should be called when the program ends gracefully	Close()} // Fs represents a wrapped fs.Fstype Fs struct {	fs.Fs	wrapper fs.Fs 	name     string	root     string	features *fs.Features // optional features	cache    Storage 	fileAge            time.Duration	chunkSize          int64	chunkTotalSize     int64	chunkCleanInterval time.Duration	readRetries        int	totalWorkers       int	totalMaxWorkers    int	chunkMemory        bool	cacheWrites        bool 	lastChunkCleanup time.Time	cleanupMu        sync.Mutex	rateLimiter      *rate.Limiter	plexConnector    *plexConnector} // NewFs contstructs an Fs from the path, container:pathfunc NewFs(name, rpath string) (fs.Fs, error) {	remote := fs.ConfigFileGet(name, "remote")	if strings.HasPrefix(remote, name+":") {		return nil, errors.New("can't point cache remote at itself - check the value of the remote setting")	}	// Look for a file first	remotePath := path.Join(remote, rpath)	wrappedFs, wrapErr := fs.NewFs(remotePath)	if wrapErr != fs.ErrorIsFile && wrapErr != nil {		return nil, errors.Wrapf(wrapErr, "failed to make remote %q to wrap", remotePath)	}	fs.Debugf(name, "wrapped %v:%v at root %v", wrappedFs.Name(), wrappedFs.Root(), rpath) 	plexURL := fs.ConfigFileGet(name, "plex_url")	plexToken := fs.ConfigFileGet(name, "plex_token")	var chunkSize fs.SizeSuffix	chunkSizeString := fs.ConfigFileGet(name, "chunk_size", DefCacheChunkSize)	if *cacheChunkSize != DefCacheChunkSize {		chunkSizeString = *cacheChunkSize	}	err := chunkSize.Set(chunkSizeString)	if err != nil {		return nil, errors.Wrapf(err, "failed to understand chunk size", chunkSizeString)	}	var chunkTotalSize fs.SizeSuffix	chunkTotalSizeString := fs.ConfigFileGet(name, "chunk_total_size", DefCacheTotalChunkSize)	if *cacheTotalChunkSize != DefCacheTotalChunkSize {		chunkTotalSizeString = *cacheTotalChunkSize	}	err = chunkTotalSize.Set(chunkTotalSizeString)	if err != nil {		return nil, errors.Wrapf(err, "failed to understand chunk total size", chunkTotalSizeString)	}	chunkCleanIntervalStr := *cacheChunkCleanInterval	chunkCleanInterval, err := time.ParseDuration(chunkCleanIntervalStr)	if err != nil {		return nil, errors.Wrapf(err, "failed to understand duration %v", chunkCleanIntervalStr)	}	infoAge := fs.ConfigFileGet(name, "info_age", DefCacheInfoAge)	if *cacheInfoAge != DefCacheInfoAge {		infoAge = *cacheInfoAge	}	infoDuration, err := time.ParseDuration(infoAge)	if err != nil {		return nil, errors.Wrapf(err, "failed to understand duration", infoAge)	}	// configure cache backend	if *cacheDbPurge {		fs.Debugf(name, "Purging the DB")	}	f := &Fs{		Fs:                 wrappedFs,		name:               name,		root:               rpath,		fileAge:            infoDuration,		chunkSize:          int64(chunkSize),		chunkTotalSize:     int64(chunkTotalSize),		chunkCleanInterval: chunkCleanInterval,		readRetries:        *cacheReadRetries,		totalWorkers:       *cacheTotalWorkers,		totalMaxWorkers:    *cacheTotalWorkers,		chunkMemory:        !*cacheChunkNoMemory,		cacheWrites:        *cacheStoreWrites,		lastChunkCleanup:   time.Now().Truncate(time.Hour * 24 * 30),	}	if f.chunkTotalSize < (f.chunkSize * int64(f.totalWorkers)) {		return nil, errors.Errorf("don't set cache-total-chunk-size(%v) less than cache-chunk-size(%v) * cache-workers(%v)",			f.chunkTotalSize, f.chunkSize, f.totalWorkers)	}	f.rateLimiter = rate.NewLimiter(rate.Limit(float64(*cacheRps)), f.totalWorkers) 	f.plexConnector = &plexConnector{}	if plexURL != "" {		if plexToken != "" {			f.plexConnector, err = newPlexConnectorWithToken(f, plexURL, plexToken)			if err != nil {				return nil, errors.Wrapf(err, "failed to connect to the Plex API %v", plexURL)			}		} else {			plexUsername := fs.ConfigFileGet(name, "plex_username")			plexPassword := fs.ConfigFileGet(name, "plex_password")			if plexPassword != "" && plexUsername != "" {				decPass, err := fs.Reveal(plexPassword)				if err != nil {					decPass = plexPassword				}				f.plexConnector, err = newPlexConnector(f, plexURL, plexUsername, decPass)				if err != nil {					return nil, errors.Wrapf(err, "failed to connect to the Plex API %v", plexURL)				}			}		}	} 	dbPath := *cacheDbPath	chunkPath := *cacheChunkPath	// if the dbPath is non default but the chunk path is default, we overwrite the last to follow the same one as dbPath	if dbPath != filepath.Join(fs.CacheDir, "cache-backend") &&		chunkPath == filepath.Join(fs.CacheDir, "cache-backend") {		chunkPath = dbPath	}	if filepath.Ext(dbPath) != "" {		dbPath = filepath.Dir(dbPath)	}	if filepath.Ext(chunkPath) != "" {		chunkPath = filepath.Dir(chunkPath)	}	err = os.MkdirAll(dbPath, os.ModePerm)	if err != nil {		return nil, errors.Wrapf(err, "failed to create cache directory %v", dbPath)	}	err = os.MkdirAll(chunkPath, os.ModePerm)	if err != nil {		return nil, errors.Wrapf(err, "failed to create cache directory %v", chunkPath)	} 	dbPath = filepath.Join(dbPath, name+".db")	chunkPath = filepath.Join(chunkPath, name)	fs.Infof(name, "Cache DB path: %v", dbPath)	fs.Infof(name, "Cache chunk path: %v", chunkPath)	f.cache, err = GetPersistent(dbPath, chunkPath, &Features{		PurgeDb: *cacheDbPurge,	})	if err != nil {		return nil, errors.Wrapf(err, "failed to start cache db")	}	// Trap SIGINT and SIGTERM to close the DB handle gracefully	c := make(chan os.Signal, 1)	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP)	go func() {		for {			s := <-c			if s == syscall.SIGINT || s == syscall.SIGTERM {				fs.Debugf(f, "Got signal: %v", s)				f.cache.Close()			} else if s == syscall.SIGHUP {				fs.Infof(f, "Clearing cache from signal")				f.DirCacheFlush()			}		}	}() 	fs.Infof(name, "Chunk Memory: %v", f.chunkMemory)	fs.Infof(name, "Chunk Size: %v", fs.SizeSuffix(f.chunkSize))	fs.Infof(name, "Chunk Total Size: %v", fs.SizeSuffix(f.chunkTotalSize))	fs.Infof(name, "Chunk Clean Interval: %v", f.chunkCleanInterval.String())	fs.Infof(name, "Workers: %v", f.totalWorkers)	fs.Infof(name, "File Age: %v", f.fileAge.String())	fs.Infof(name, "Cache Writes: %v", f.cacheWrites) 	go f.CleanUpCache(false) 	// TODO: Explore something here but now it's not something we want	// when writing from cache, source FS will send a notification and clear it out immediately	//setup dir notification	//doDirChangeNotify := wrappedFs.Features().DirChangeNotify	//if doDirChangeNotify != nil {	//	doDirChangeNotify(func(dir string) {	//		d := NewAbsDirectory(f, dir)	//		d.Flush()	//		fs.Infof(dir, "updated from notification")	//	}, time.Second * 10)	//} 	f.features = (&fs.Features{		CanHaveEmptyDirectories: true,		DuplicateFiles:          false, // storage doesn't permit this		Purge:                   f.Purge,		Copy:                    f.Copy,		Move:                    f.Move,		DirMove:                 f.DirMove,		DirChangeNotify:         nil,		PutUnchecked:            f.PutUnchecked,		PutStream:               f.PutStream,		CleanUp:                 f.CleanUp,		UnWrap:                  f.UnWrap,		WrapFs:                  f.WrapFs,		SetWrapper:              f.SetWrapper,	}).Fill(f).Mask(wrappedFs).WrapsFs(f, wrappedFs)	f.features.DirCacheFlush = f.DirCacheFlush 	return f, wrapErr} // Name of the remote (as passed into NewFs)func (f *Fs) Name() string {	return f.name} // Root of the remote (as passed into NewFs)func (f *Fs) Root() string {	return f.root} // Features returns the optional features of this Fsfunc (f *Fs) Features() *fs.Features {	return f.features} // String returns a description of the FSfunc (f *Fs) String() string {	return fmt.Sprintf("%s:%s", f.name, f.root)} // ChunkSize returns the configured chunk sizefunc (f *Fs) ChunkSize() int64 {	return f.chunkSize} // NewObject finds the Object at remote.func (f *Fs) NewObject(remote string) (fs.Object, error) {	co := NewObject(f, remote)	err := f.cache.GetObject(co)	if err != nil {		fs.Debugf(remote, "find: error: %v", err)	} else if time.Now().After(co.CacheTs.Add(f.fileAge)) {		fs.Debugf(remote, "find: cold object ts: %v", co.CacheTs)	} else {		fs.Debugf(remote, "find: warm object ts: %v", co.CacheTs)		return co, nil	}	obj, err := f.Fs.NewObject(remote)	if err != nil {		return nil, err	}	co = ObjectFromOriginal(f, obj)	co.persist()	return co, nil} // List the objects and directories in dir into entriesfunc (f *Fs) List(dir string) (entries fs.DirEntries, err error) {	// clean cache	go f.CleanUpCache(false) 	cd := ShallowDirectory(f, dir)	entries, err = f.cache.GetDirEntries(cd)	if err != nil {		fs.Debugf(dir, "list: error: %v", err)	} else if time.Now().After(cd.CacheTs.Add(f.fileAge)) {		fs.Debugf(dir, "list: cold listing: %v", cd.CacheTs)	} else if len(entries) == 0 {		// TODO: read empty dirs from source?		fs.Debugf(dir, "list: empty listing")	} else {		fs.Debugf(dir, "list: warm %v from cache for: %v, ts: %v", len(entries), cd.abs(), cd.CacheTs)		return entries, nil	} 	entries, err = f.Fs.List(dir)	if err != nil {		return nil, err	}	fs.Debugf(dir, "list: read %v from source", len(entries)) 	var cachedEntries fs.DirEntries	for _, entry := range entries {		switch o := entry.(type) {		case fs.Object:			co := ObjectFromOriginal(f, o)			co.persist()			cachedEntries = append(cachedEntries, co)		case fs.Directory:			cd := DirectoryFromOriginal(f, o)			err = f.cache.AddDir(cd)			cachedEntries = append(cachedEntries, cd)		default:			err = errors.Errorf("Unknown object type %T", entry)		}	}	if err != nil {		fs.Errorf(dir, "err caching listing: %v", err)	} else {		t := time.Now()		cd.CacheTs = &t		err := f.cache.AddDir(cd)

Lines 1–500 of 1017. Load more