// +build !plan9 package cache import (	"context"	"fmt"	"io"	"os"	"os/signal"	"path"	"path/filepath"	"strings"	"sync"	"syscall"	"time" 	"github.com/ncw/rclone/backend/crypt"	"github.com/ncw/rclone/fs"	"github.com/ncw/rclone/fs/config"	"github.com/ncw/rclone/fs/config/flags"	"github.com/ncw/rclone/fs/config/obscure"	"github.com/ncw/rclone/fs/hash"	"github.com/ncw/rclone/fs/rc"	"github.com/ncw/rclone/fs/walk"	"github.com/ncw/rclone/lib/atexit"	"github.com/pkg/errors"	"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	// DefCacheTmpWaitTime says how long should files be stored in local cache before being uploaded	DefCacheTmpWaitTime = "15m"	// DefCacheDbWaitTime defines how long the cache backend should wait for the DB to be available	DefCacheDbWaitTime = 1 * time.Second) // Globalsvar (	// Flags	cacheDbPath             = flags.StringP("cache-db-path", "", filepath.Join(config.CacheDir, "cache-backend"), "Directory to cache DB")	cacheChunkPath          = flags.StringP("cache-chunk-path", "", filepath.Join(config.CacheDir, "cache-backend"), "Directory to cached chunk files")	cacheDbPurge            = flags.BoolP("cache-db-purge", "", false, "Purge the cache DB before")	cacheChunkSize          = flags.StringP("cache-chunk-size", "", DefCacheChunkSize, "The size of a chunk")	cacheTotalChunkSize     = flags.StringP("cache-total-chunk-size", "", DefCacheTotalChunkSize, "The total size which the chunks can take up from the disk")	cacheChunkCleanInterval = flags.StringP("cache-chunk-clean-interval", "", DefCacheChunkCleanInterval, "Interval at which chunk cleanup runs")	cacheInfoAge            = flags.StringP("cache-info-age", "", DefCacheInfoAge, "How much time should object info be stored in cache")	cacheReadRetries        = flags.IntP("cache-read-retries", "", DefCacheReadRetries, "How many times to retry a read from a cache storage")	cacheTotalWorkers       = flags.IntP("cache-workers", "", DefCacheTotalWorkers, "How many workers should run in parallel to download chunks")	cacheChunkNoMemory      = flags.BoolP("cache-chunk-no-memory", "", DefCacheChunkNoMemory, "Disable the in-memory cache for storing chunks during streaming")	cacheRps                = flags.IntP("cache-rps", "", int(DefCacheRps), "Limits the number of requests per second to the source FS. -1 disables the rate limiter")	cacheStoreWrites        = flags.BoolP("cache-writes", "", DefCacheWrites, "Will cache file data on writes through the FS")	cacheTempWritePath      = flags.StringP("cache-tmp-upload-path", "", "", "Directory to keep temporary files until they are uploaded to the cloud storage")	cacheTempWaitTime       = flags.StringP("cache-tmp-wait-time", "", DefCacheTmpWaitTime, "How long should files be stored in local cache before being uploaded")	cacheDbWaitTime         = flags.DurationP("cache-db-wait-time", "", DefCacheDbWaitTime, "How long to wait for the DB to be available - 0 is unlimited")) // 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,		}},	})} // Fs represents a wrapped fs.Fstype Fs struct {	fs.Fs	wrapper fs.Fs 	name     string	root     string	features *fs.Features // optional features	cache    *Persistent 	fileAge            time.Duration	chunkSize          int64	chunkTotalSize     int64	chunkCleanInterval time.Duration	readRetries        int	totalWorkers       int	totalMaxWorkers    int	chunkMemory        bool	cacheWrites        bool	tempWritePath      string	tempWriteWait      time.Duration	tempFs             fs.Fs 	lastChunkCleanup time.Time	cleanupMu        sync.Mutex	rateLimiter      *rate.Limiter	plexConnector    *plexConnector	backgroundRunner *backgroundWriter	cleanupChan      chan bool	parentsForgetFn  []func(string, fs.EntryType)	notifiedRemotes  map[string]bool	notifiedMu       sync.Mutex	parentsForgetMu  sync.Mutex} // parseRootPath returns a cleaned root path and a nil error or "" and an error when the path is invalidfunc parseRootPath(path string) (string, error) {	return strings.Trim(path, "/"), nil} // NewFs constructs a Fs from the path, container:pathfunc NewFs(name, rootPath string) (fs.Fs, error) {	remote := config.FileGet(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")	} 	rpath, err := parseRootPath(rootPath)	if err != nil {		return nil, errors.Wrapf(err, "failed to clean root path %q", rootPath)	} 	remotePath := path.Join(remote, rpath)	wrappedFs, wrapErr := fs.NewFs(remotePath)	if wrapErr != nil && wrapErr != fs.ErrorIsFile {		return nil, errors.Wrapf(wrapErr, "failed to make remote %q to wrap", remotePath)	}	var fsErr error	fs.Debugf(name, "wrapped %v:%v at root %v", wrappedFs.Name(), wrappedFs.Root(), rpath)	if wrapErr == fs.ErrorIsFile {		fsErr = fs.ErrorIsFile		rpath = cleanPath(path.Dir(rpath))	}	plexURL := config.FileGet(name, "plex_url")	plexToken := config.FileGet(name, "plex_token")	var chunkSize fs.SizeSuffix	chunkSizeString := config.FileGet(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 %v", chunkSizeString)	}	var chunkTotalSize fs.SizeSuffix	chunkTotalSizeString := config.FileGet(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 %v", chunkTotalSizeString)	}	chunkCleanIntervalStr := *cacheChunkCleanInterval	chunkCleanInterval, err := time.ParseDuration(chunkCleanIntervalStr)	if err != nil {		return nil, errors.Wrapf(err, "failed to understand duration %v", chunkCleanIntervalStr)	}	infoAge := config.FileGet(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 %v", infoAge)	}	waitTime, err := time.ParseDuration(*cacheTempWaitTime)	if err != nil {		return nil, errors.Wrapf(err, "failed to understand duration %v", *cacheTempWaitTime)	}	// 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),		tempWritePath:      *cacheTempWritePath,		tempWriteWait:      waitTime,		cleanupChan:        make(chan bool, 1),		notifiedRemotes:    make(map[string]bool),	}	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 := config.FileGet(name, "plex_username")			plexPassword := config.FileGet(name, "plex_password")			if plexPassword != "" && plexUsername != "" {				decPass, err := obscure.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(config.CacheDir, "cache-backend") &&		chunkPath == filepath.Join(config.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.SIGHUP)	atexit.Register(func() {		if plexURL != "" {			f.plexConnector.closeWebsocket()		}		f.StopBackgroundRunners()	})	go func() {		for {			s := <-c			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())	if f.cacheWrites {		fs.Infof(name, "Cache Writes: enabled")	} 	if f.tempWritePath != "" {		err = os.MkdirAll(f.tempWritePath, os.ModePerm)		if err != nil {			return nil, errors.Wrapf(err, "failed to create cache directory %v", f.tempWritePath)		}		f.tempWritePath = filepath.ToSlash(f.tempWritePath)		f.tempFs, err = fs.NewFs(f.tempWritePath)		if err != nil {			return nil, errors.Wrapf(err, "failed to create temp fs: %v", err)		}		fs.Infof(name, "Upload Temp Rest Time: %v", f.tempWriteWait.String())		fs.Infof(name, "Upload Temp FS: %v", f.tempWritePath)		f.backgroundRunner, _ = initBackgroundUploader(f)		go f.backgroundRunner.run()	} 	go func() {		for {			time.Sleep(f.chunkCleanInterval)			select {			case <-f.cleanupChan:				fs.Infof(f, "stopping cleanup")				return			default:				fs.Debugf(f, "starting cleanup")				f.CleanUpCache(false)			}		}	}() 	if doChangeNotify := wrappedFs.Features().ChangeNotify; doChangeNotify != nil {		doChangeNotify(f.receiveChangeNotify, f.chunkCleanInterval)	} 	f.features = (&fs.Features{		CanHaveEmptyDirectories: true,		DuplicateFiles:          false, // storage doesn't permit this	}).Fill(f).Mask(wrappedFs).WrapsFs(f, wrappedFs)	// override only those features that use a temp fs and it doesn't support them	//f.features.ChangeNotify = f.ChangeNotify	if f.tempWritePath != "" {		if f.tempFs.Features().Copy == nil {			f.features.Copy = nil		}		if f.tempFs.Features().Move == nil {			f.features.Move = nil		}		if f.tempFs.Features().Move == nil {			f.features.Move = nil		}		if f.tempFs.Features().DirMove == nil {			f.features.DirMove = nil		}		if f.tempFs.Features().MergeDirs == nil {			f.features.MergeDirs = nil		}	}	// even if the wrapped fs doesn't support it, we still want it	f.features.DirCacheFlush = f.DirCacheFlush 	rc.Add(rc.Call{		Path:  "cache/expire",		Fn:    f.httpExpireRemote,		Title: "Purge a remote from cache",		Help: `Purge a remote from the cache backend. Supports either a directory or a file.Params:  - remote = path to remote (required)  - withData = true/false to delete cached data (chunks) as well (optional) Eg     rclone rc cache/expire remote=path/to/sub/folder/    rclone rc cache/expire remote=/ withData=true `,	}) 	rc.Add(rc.Call{		Path:  "cache/stats",		Fn:    f.httpStats,		Title: "Get cache stats",		Help: `Show statistics for the cache remote.`,	}) 	return f, fsErr} func (f *Fs) httpStats(in rc.Params) (out rc.Params, err error) {	out = make(rc.Params)	m, err := f.Stats()	if err != nil {		return out, errors.Errorf("error while getting cache stats")	}	out["status"] = "ok"	out["stats"] = m	return out, nil} func (f *Fs) httpExpireRemote(in rc.Params) (out rc.Params, err error) {	out = make(rc.Params)	remoteInt, ok := in["remote"]	if !ok {		return out, errors.Errorf("remote is needed")	}	remote := remoteInt.(string)	withData := false	_, ok = in["withData"]	if ok {		withData = true	} 	if cleanPath(remote) != "" {		// if it's wrapped by crypt we need to check what format we got		if cryptFs, yes := f.isWrappedByCrypt(); yes {			_, err := cryptFs.DecryptFileName(remote)			// if it failed to decrypt then it is a decrypted format and we need to encrypt it			if err != nil {				remote = cryptFs.EncryptFileName(remote)			}			// else it's an encrypted format and we can use it as it is		} 		if !f.cache.HasEntry(path.Join(f.Root(), remote)) {			return out, errors.Errorf("%s doesn't exist in cache", remote)		}	} 	co := NewObject(f, remote)	err = f.cache.GetObject(co)	if err != nil { // it could be a dir		cd := NewDirectory(f, remote)		err := f.cache.ExpireDir(cd)		if err != nil {			return out, errors.WithMessage(err, "error expiring directory")		}		// notify vfs too

Lines 1–500 of 1574. Load more