diff --git a/cmd/bucket-handlers-listobjects.go b/cmd/bucket-handlers-listobjects.go index 017c10cc5..13b86ae83 100644 --- a/cmd/bucket-handlers-listobjects.go +++ b/cmd/bucket-handlers-listobjects.go @@ -100,9 +100,7 @@ func (api objectAPIHandlers) ListObjectsV2Handler(w http.ResponseWriter, r *http } listObjectsV2 := objectAPI.ListObjectsV2 - if api.CacheAPI() != nil { - listObjectsV2 = api.CacheAPI().ListObjectsV2 - } + // Inititate a list objects operation based on the input params. // On success would return back ListObjectsInfo object to be // marshaled into S3 compatible XML header. @@ -179,9 +177,6 @@ func (api objectAPIHandlers) ListObjectsV1Handler(w http.ResponseWriter, r *http } listObjects := objectAPI.ListObjects - if api.CacheAPI() != nil { - listObjects = api.CacheAPI().ListObjects - } // Inititate a list objects operation based on the input params. // On success would return back ListObjectsInfo object to be diff --git a/cmd/bucket-handlers.go b/cmd/bucket-handlers.go index 84e43b54f..ea9a6241e 100644 --- a/cmd/bucket-handlers.go +++ b/cmd/bucket-handlers.go @@ -105,9 +105,7 @@ func (api objectAPIHandlers) GetBucketLocationHandler(w http.ResponseWriter, r * } getBucketInfo := objectAPI.GetBucketInfo - if api.CacheAPI() != nil { - getBucketInfo = api.CacheAPI().GetBucketInfo - } + if _, err := getBucketInfo(ctx, bucket); err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r)) return @@ -202,9 +200,6 @@ func (api objectAPIHandlers) ListBucketsHandler(w http.ResponseWriter, r *http.R } listBuckets := objectAPI.ListBuckets - if api.CacheAPI() != nil { - listBuckets = api.CacheAPI().ListBuckets - } if s3Error := checkRequestAuthType(ctx, r, policy.ListAllMyBucketsAction, "", ""); s3Error != ErrNone { writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL, guessIsBrowserReq(r)) @@ -742,9 +737,7 @@ func (api objectAPIHandlers) HeadBucketHandler(w http.ResponseWriter, r *http.Re } getBucketInfo := objectAPI.GetBucketInfo - if api.CacheAPI() != nil { - getBucketInfo = api.CacheAPI().GetBucketInfo - } + if _, err := getBucketInfo(ctx, bucket); err != nil { writeErrorResponseHeadersOnly(w, toAPIError(ctx, err)) return @@ -774,9 +767,7 @@ func (api objectAPIHandlers) DeleteBucketHandler(w http.ResponseWriter, r *http. } deleteBucket := objectAPI.DeleteBucket - if api.CacheAPI() != nil { - deleteBucket = api.CacheAPI().DeleteBucket - } + // Attempt to delete bucket. if err := deleteBucket(ctx, bucket); err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r)) diff --git a/cmd/disk-cache-backend.go b/cmd/disk-cache-backend.go new file mode 100644 index 000000000..c617ba615 --- /dev/null +++ b/cmd/disk-cache-backend.go @@ -0,0 +1,417 @@ +/* + * MinIO Cloud Storage, (C) 2019 MinIO, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package cmd + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "io/ioutil" + "log" + "net/http" + "os" + "path" + "sync" + "time" + + "github.com/djherbis/atime" + "github.com/minio/minio/cmd/logger" + "github.com/minio/minio/pkg/disk" +) + +const ( + // cache.json object metadata for cached objects. + cacheMetaJSONFile = "cache.json" + cacheDataFile = "data" + + cacheEnvDelimiter = ";" +) + +// Represents the cache metadata struct +type cacheMeta struct { + // Metadata map for current object. + Meta map[string]string `json:"meta,omitempty"` +} + +func (m *cacheMeta) ToObjectInfo(bucket, object string, fi os.FileInfo) (o ObjectInfo) { + if len(m.Meta) == 0 { + m.Meta = make(map[string]string) + } + + o = ObjectInfo{ + Bucket: bucket, + Name: object, + } + + // We set file info only if its valid. + o.ModTime = timeSentinel + if fi != nil { + o.ModTime = fi.ModTime() + o.Size = fi.Size() + if fi.IsDir() { + // Directory is always 0 bytes in S3 API, treat it as such. + o.Size = 0 + o.IsDir = fi.IsDir() + } + } + + o.ETag = extractETag(m.Meta) + o.ContentType = m.Meta["content-type"] + o.ContentEncoding = m.Meta["content-encoding"] + if storageClass, ok := m.Meta[amzStorageClass]; ok { + o.StorageClass = storageClass + } else { + o.StorageClass = globalMinioDefaultStorageClass + } + var ( + t time.Time + e error + ) + if exp, ok := m.Meta["expires"]; ok { + if t, e = time.Parse(http.TimeFormat, exp); e == nil { + o.Expires = t.UTC() + } + } + // etag/md5Sum has already been extracted. We need to + // remove to avoid it from appearing as part of + o.UserDefined = cleanMetadata(m.Meta) + return o +} + +// disk cache +type diskCache struct { + dir string // caching directory + maxDiskUsagePct int // max usage in % + expiry int // cache expiry in days + // to manage cache operations + nsMutex *nsLockMap + // mark false if drive is offline + online bool + // mutex to protect updates to online variable + onlineMutex *sync.RWMutex + // purge() listens on this channel to start the cache-purge process + purgeChan chan struct{} +} + +// Inits the disk cache dir if it is not init'ed already. +func newdiskCache(dir string, expiry int, maxDiskUsagePct int) (*diskCache, error) { + if err := os.MkdirAll(dir, 0777); err != nil { + return nil, fmt.Errorf("Unable to initialize '%s' dir, %s", dir, err) + } + + if expiry == 0 { + expiry = globalCacheExpiry + } + + cache := diskCache{ + dir: dir, + expiry: expiry, + maxDiskUsagePct: maxDiskUsagePct, + purgeChan: make(chan struct{}), + online: true, + onlineMutex: &sync.RWMutex{}, + nsMutex: newNSLock(false), + } + return &cache, nil +} + +// Returns if the disk usage is low. +// Disk usage is low if usage is < 80% of cacheMaxDiskUsagePct +// Ex. for a 100GB disk, if maxUsage is configured as 70% then cacheMaxDiskUsagePct is 70G +// hence disk usage is low if the disk usage is less than 56G (because 80% of 70G is 56G) +func (c *diskCache) diskUsageLow() bool { + minUsage := c.maxDiskUsagePct * 80 / 100 + di, err := disk.GetInfo(c.dir) + if err != nil { + reqInfo := (&logger.ReqInfo{}).AppendTags("cachePath", c.dir) + ctx := logger.SetReqInfo(context.Background(), reqInfo) + logger.LogIf(ctx, err) + return false + } + usedPercent := (di.Total - di.Free) * 100 / di.Total + return int(usedPercent) < minUsage +} + +// Return if the disk usage is high. +// Disk usage is high if disk used is > cacheMaxDiskUsagePct +func (c *diskCache) diskUsageHigh() bool { + di, err := disk.GetInfo(c.dir) + if err != nil { + reqInfo := (&logger.ReqInfo{}).AppendTags("cachePath", c.dir) + ctx := logger.SetReqInfo(context.Background(), reqInfo) + logger.LogIf(ctx, err) + return true + } + usedPercent := (di.Total - di.Free) * 100 / di.Total + return int(usedPercent) > c.maxDiskUsagePct +} + +// Returns if size space can be allocated without exceeding +// max disk usable for caching +func (c *diskCache) diskAvailable(size int64) bool { + di, err := disk.GetInfo(c.dir) + if err != nil { + reqInfo := (&logger.ReqInfo{}).AppendTags("cachePath", c.dir) + ctx := logger.SetReqInfo(context.Background(), reqInfo) + logger.LogIf(ctx, err) + return false + } + usedPercent := (di.Total - (di.Free - uint64(size))) * 100 / di.Total + return int(usedPercent) < c.maxDiskUsagePct +} + +// Purge cache entries that were not accessed. +func (c *diskCache) purge() { + ctx := context.Background() + for { + olderThan := c.expiry + for !c.diskUsageLow() { + // delete unaccessed objects older than expiry duration + expiry := UTCNow().AddDate(0, 0, -1*olderThan) + olderThan /= 2 + if olderThan < 1 { + break + } + deletedCount := 0 + + objDirs, err := ioutil.ReadDir(c.dir) + if err != nil { + log.Fatal(err) + } + + for _, obj := range objDirs { + var fi os.FileInfo + fi, err := os.Stat(pathJoin(c.dir, obj.Name())) + if err != nil { + continue + } + objInfo, err := c.statCache(ctx, pathJoin(c.dir, obj.Name())) + if err != nil { + continue + } + if !filterFromCache(objInfo.UserDefined) || + !isStaleCache(objInfo) || + atime.Get(fi).After(expiry) { + continue + } + if err = os.RemoveAll(pathJoin(c.dir, obj.Name())); err != nil { + logger.LogIf(ctx, err) + continue + } + deletedCount++ + } + if deletedCount == 0 { + // to avoid a busy loop + time.Sleep(time.Minute * 30) + } + } + <-c.purgeChan + } +} + +// sets cache drive status +func (c *diskCache) setOnline(status bool) { + c.onlineMutex.Lock() + c.online = status + c.onlineMutex.Unlock() +} + +// returns true if cache drive is online +func (c *diskCache) IsOnline() bool { + c.onlineMutex.RLock() + defer c.onlineMutex.RUnlock() + return c.online +} + +// Stat returns ObjectInfo from disk cache +func (c *diskCache) Stat(ctx context.Context, bucket, object string) (oi ObjectInfo, err error) { + cachePath := c.getCacheSHADir(bucket, object) + cLock := c.nsMutex.NewNSLock(cachePath, "") + if err := cLock.GetRLock(globalObjectTimeout); err != nil { + return oi, err + } + defer cLock.RUnlock() + + oi, err = c.stat(ctx, bucket, object) + return +} + +// lockless helper to Stat +func (c *diskCache) stat(ctx context.Context, bucket, object string) (oi ObjectInfo, e error) { + cacheObjPath := c.getCacheSHADir(bucket, object) + oi, e = c.statCache(ctx, cacheObjPath) + if e != nil { + return + } + oi.Bucket = bucket + oi.Name = object + return +} + +// statCache is a convenience function for purge() to get ObjectInfo for cached object +func (c *diskCache) statCache(ctx context.Context, cacheObjPath string) (oi ObjectInfo, e error) { + // Stat the file to get file size. + fi, err := fsStatFile(ctx, pathJoin(cacheObjPath, cacheDataFile)) + if err != nil { + return oi, err + } + metaPath := pathJoin(cacheObjPath, cacheMetaJSONFile) + + f, err := os.Open(metaPath) + if err != nil { + return oi, err + } + defer f.Close() + meta := &cacheMeta{} + if err := jsonLoad(f, meta); err != nil { + return oi, err + } + return meta.ToObjectInfo("", "", fi), nil +} + +// caches object metadata to disk cache +func (c *diskCache) saveMetadata(ctx context.Context, bucket, object string, meta map[string]string) error { + fileName := c.getCacheSHADir(bucket, object) + metaPath := pathJoin(fileName, cacheMetaJSONFile) + + f, err := os.Create(metaPath) + if err != nil { + return err + } + defer f.Close() + m := cacheMeta{Meta: meta} + jsonData, err := json.Marshal(m) + if err != nil { + return err + } + f.Write(jsonData) + return nil +} + +func (c *diskCache) getCacheSHADir(bucket, object string) string { + return path.Join(c.dir, getSHA256Hash([]byte(path.Join(bucket, object)))) +} + +// Caches the object to disk +func (c *diskCache) Put(ctx context.Context, bucket, object string, data io.Reader, size int64, opts ObjectOptions) error { + if c.diskUsageHigh() { + select { + case c.purgeChan <- struct{}{}: + default: + } + return errDiskFull + } + if !c.diskAvailable(size) { + return errDiskFull + } + cachePath := c.getCacheSHADir(bucket, object) + cLock := c.nsMutex.NewNSLock(cachePath, "") + if err := cLock.GetLock(globalObjectTimeout); err != nil { + return err + } + defer cLock.Unlock() + if err := os.MkdirAll(cachePath, 0777); err != nil { + return err + } + bufSize := int64(readSizeV1) + if size > 0 && bufSize > size { + bufSize = size + } + filePath := path.Join(cachePath, cacheDataFile) + buf := make([]byte, int(bufSize)) + _, err := fsCreateFile(ctx, filePath, data, buf, size) + if IsErr(err, baseErrs...) { + c.setOnline(false) + } + if err != nil { + return err + } + return c.saveMetadata(ctx, bucket, object, opts.UserDefined) +} + +// Get returns ObjectInfo and reader for object from disk cache +func (c *diskCache) Get(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, opts ObjectOptions) (gr *GetObjectReader, err error) { + cachePath := c.getCacheSHADir(bucket, object) + fileName := path.Join(cachePath, cacheDataFile) + + cLock := c.nsMutex.NewNSLock(cachePath, "") + if err := cLock.GetRLock(globalObjectTimeout); err != nil { + return nil, err + } + + defer cLock.RUnlock() + var objInfo ObjectInfo + + if objInfo, err = c.stat(ctx, bucket, object); err != nil { + return nil, toObjectErr(err, bucket, object) + } + var nsUnlocker = func() {} + // For a directory, we need to send an reader that returns no bytes. + if hasSuffix(object, slashSeparator) { + // The lock taken above is released when + // objReader.Close() is called by the caller. + return NewGetObjectReaderFromReader(bytes.NewBuffer(nil), objInfo, opts.CheckCopyPrecondFn, nsUnlocker) + } + objReaderFn, off, length, rErr := NewGetObjectReader(rs, objInfo, opts.CheckCopyPrecondFn, nsUnlocker) + if rErr != nil { + return nil, rErr + } + + // Read the object, doesn't exist returns an s3 compatible error. + readCloser, size, err := fsOpenFile(ctx, fileName, off) + if err != nil { + return nil, toObjectErr(err, bucket, object) + } + reader := io.LimitReader(readCloser, length) + closeFn := func() { + readCloser.Close() + } + + // Check if range is valid + if off > size || off+length > size { + err = InvalidRange{off, length, size} + logger.LogIf(ctx, err) + closeFn() + nsUnlocker() + return nil, err + } + + return objReaderFn(reader, h, opts.CheckCopyPrecondFn, closeFn) +} + +// Deletes the cached object +func (c *diskCache) Delete(ctx context.Context, bucket, object string) (err error) { + cachePath := c.getCacheSHADir(bucket, object) + cLock := c.nsMutex.NewNSLock(cachePath, "") + if err := cLock.GetLock(globalObjectTimeout); err != nil { + return err + } + defer cLock.Unlock() + return os.RemoveAll(cachePath) + +} + +// convenience function to check if object is cached on this diskCache +func (c *diskCache) Exists(ctx context.Context, bucket, object string) bool { + fileName := getSHA256Hash([]byte(path.Join(bucket, object))) + if _, err := os.Stat(path.Join(c.dir, fileName)); err != nil { + return false + } + return true +} diff --git a/cmd/disk-cache-fs.go b/cmd/disk-cache-fs.go deleted file mode 100644 index 46ec572d0..000000000 --- a/cmd/disk-cache-fs.go +++ /dev/null @@ -1,537 +0,0 @@ -/* - * MinIO Cloud Storage, (C) 2018 MinIO, Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package cmd - -import ( - "context" - "encoding/json" - "fmt" - "io" - "io/ioutil" - "os" - "path" - "sync" - "time" - - "github.com/minio/minio/cmd/logger" - "github.com/minio/minio/pkg/disk" - "github.com/minio/minio/pkg/lock" -) - -const ( - // cache.json object metadata for cached objects. - cacheMetaJSONFile = "cache.json" - - cacheEnvDelimiter = ";" -) - -// cacheFSObjects implements the cache backend operations. -type cacheFSObjects struct { - *FSObjects - // caching drive path (from cache "drives" in config.json) - dir string - // expiry in days specified in config.json - expiry int - // max disk usage pct - maxDiskUsagePct int - // purge() listens on this channel to start the cache-purge process - purgeChan chan struct{} - // mark false if drive is offline - online bool - // mutex to protect updates to online variable - onlineMutex *sync.RWMutex -} - -// Inits the cache directory if it is not init'ed already. -// Initializing implies creation of new FS Object layer. -func newCacheFSObjects(dir string, expiry int, maxDiskUsagePct int) (*cacheFSObjects, error) { - // Assign a new UUID for FS minio mode. Each server instance - // gets its own UUID for temporary file transaction. - fsUUID := mustGetUUID() - - // Initialize meta volume, if volume already exists ignores it. - if err := initMetaVolumeFS(dir, fsUUID); err != nil { - return nil, fmt.Errorf("Unable to initialize '.minio.sys' meta volume, %s", err) - } - - trashPath := pathJoin(dir, minioMetaBucket, cacheTrashDir) - if err := os.MkdirAll(trashPath, 0777); err != nil { - return nil, err - } - - if expiry == 0 { - expiry = globalCacheExpiry - } - - // Initialize fs objects. - fsObjects := &FSObjects{ - fsPath: dir, - metaJSONFile: cacheMetaJSONFile, - fsUUID: fsUUID, - rwPool: &fsIOPool{ - readersMap: make(map[string]*lock.RLockedFile), - }, - nsMutex: newNSLock(false), - listPool: NewTreeWalkPool(globalLookupTimeout), - appendFileMap: make(map[string]*fsAppendFile), - } - - go fsObjects.cleanupStaleMultipartUploads(context.Background(), GlobalMultipartCleanupInterval, GlobalMultipartExpiry, GlobalServiceDoneCh) - - cacheFS := cacheFSObjects{ - FSObjects: fsObjects, - dir: dir, - expiry: expiry, - maxDiskUsagePct: maxDiskUsagePct, - purgeChan: make(chan struct{}), - online: true, - onlineMutex: &sync.RWMutex{}, - } - return &cacheFS, nil -} - -// Returns if the disk usage is low. -// Disk usage is low if usage is < 80% of cacheMaxDiskUsagePct -// Ex. for a 100GB disk, if maxUsage is configured as 70% then cacheMaxDiskUsagePct is 70G -// hence disk usage is low if the disk usage is less than 56G (because 80% of 70G is 56G) -func (cfs *cacheFSObjects) diskUsageLow() bool { - - minUsage := cfs.maxDiskUsagePct * 80 / 100 - di, err := disk.GetInfo(cfs.dir) - if err != nil { - reqInfo := (&logger.ReqInfo{}).AppendTags("cachePath", cfs.dir) - ctx := logger.SetReqInfo(context.Background(), reqInfo) - logger.LogIf(ctx, err) - return false - } - usedPercent := (di.Total - di.Free) * 100 / di.Total - return int(usedPercent) < minUsage -} - -// Return if the disk usage is high. -// Disk usage is high if disk used is > cacheMaxDiskUsagePct -func (cfs *cacheFSObjects) diskUsageHigh() bool { - di, err := disk.GetInfo(cfs.dir) - if err != nil { - reqInfo := (&logger.ReqInfo{}).AppendTags("cachePath", cfs.dir) - ctx := logger.SetReqInfo(context.Background(), reqInfo) - logger.LogIf(ctx, err) - return true - } - usedPercent := (di.Total - di.Free) * 100 / di.Total - return int(usedPercent) > cfs.maxDiskUsagePct -} - -// Returns if size space can be allocated without exceeding -// max disk usable for caching -func (cfs *cacheFSObjects) diskAvailable(size int64) bool { - di, err := disk.GetInfo(cfs.dir) - if err != nil { - reqInfo := (&logger.ReqInfo{}).AppendTags("cachePath", cfs.dir) - ctx := logger.SetReqInfo(context.Background(), reqInfo) - logger.LogIf(ctx, err) - return false - } - usedPercent := (di.Total - (di.Free - uint64(size))) * 100 / di.Total - return int(usedPercent) < cfs.maxDiskUsagePct -} - -// purges all content marked trash from the cache. -func (cfs *cacheFSObjects) purgeTrash() { - ticker := time.NewTicker(time.Minute * cacheCleanupInterval) - defer ticker.Stop() - - for { - select { - case <-GlobalServiceDoneCh: - return - case <-ticker.C: - trashPath := path.Join(cfs.fsPath, minioMetaBucket, cacheTrashDir) - entries, err := readDir(trashPath) - if err != nil { - return - } - for _, entry := range entries { - ctx := logger.SetReqInfo(context.Background(), &logger.ReqInfo{}) - fi, err := fsStatVolume(ctx, pathJoin(trashPath, entry)) - if err != nil { - continue - } - dir := path.Join(trashPath, fi.Name()) - - // Delete all expired cache content. - fsRemoveAll(ctx, dir) - } - } - } -} - -// Purge cache entries that were not accessed. -func (cfs *cacheFSObjects) purge() { - delimiter := slashSeparator - maxKeys := 1000 - ctx := context.Background() - for { - olderThan := cfs.expiry - for !cfs.diskUsageLow() { - // delete unaccessed objects older than expiry duration - expiry := UTCNow().AddDate(0, 0, -1*olderThan) - olderThan /= 2 - if olderThan < 1 { - break - } - deletedCount := 0 - buckets, err := cfs.ListBuckets(ctx) - if err != nil { - logger.LogIf(ctx, err) - } - // Reset cache online status if drive was offline earlier. - if !cfs.IsOnline() { - cfs.setOnline(true) - } - for _, bucket := range buckets { - var continuationToken string - var marker string - for { - objects, err := cfs.ListObjects(ctx, bucket.Name, marker, continuationToken, delimiter, maxKeys) - if err != nil { - break - } - - if !objects.IsTruncated { - break - } - marker = objects.NextMarker - for _, object := range objects.Objects { - // purge objects that qualify because of cache-control directives or - // past cache expiry duration. - if !filterFromCache(object.UserDefined) || - !isStaleCache(object) || - object.AccTime.After(expiry) { - continue - } - if err = cfs.DeleteObject(ctx, bucket.Name, object.Name); err != nil { - logger.LogIf(ctx, err) - continue - } - deletedCount++ - } - } - } - if deletedCount == 0 { - // to avoid a busy loop - time.Sleep(time.Minute * 30) - } - } - <-cfs.purgeChan - } -} - -// sets cache drive status -func (cfs *cacheFSObjects) setOnline(status bool) { - cfs.onlineMutex.Lock() - cfs.online = status - cfs.onlineMutex.Unlock() -} - -// returns true if cache drive is online -func (cfs *cacheFSObjects) IsOnline() bool { - cfs.onlineMutex.RLock() - defer cfs.onlineMutex.RUnlock() - return cfs.online -} - -// Caches the object to disk -func (cfs *cacheFSObjects) Put(ctx context.Context, bucket, object string, data *PutObjReader, opts ObjectOptions) error { - if cfs.diskUsageHigh() { - select { - case cfs.purgeChan <- struct{}{}: - default: - } - return errDiskFull - } - if !cfs.diskAvailable(data.Size()) { - return errDiskFull - } - if _, err := cfs.GetBucketInfo(ctx, bucket); err != nil { - pErr := cfs.MakeBucketWithLocation(ctx, bucket, "") - if pErr != nil { - return pErr - } - } - _, err := cfs.PutObject(ctx, bucket, object, data, opts) - // if err is due to disk being offline , mark cache drive as offline - if IsErr(err, baseErrs...) { - cfs.setOnline(false) - } - return err -} - -// Returns the handle for the cached object -func (cfs *cacheFSObjects) Get(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) (err error) { - return cfs.GetObject(ctx, bucket, object, startOffset, length, writer, etag, opts) -} - -// Deletes the cached object -func (cfs *cacheFSObjects) Delete(ctx context.Context, bucket, object string) (err error) { - return cfs.DeleteObject(ctx, bucket, object) -} - -// convenience function to check if object is cached on this cacheFSObjects -func (cfs *cacheFSObjects) Exists(ctx context.Context, bucket, object string) bool { - _, err := cfs.GetObjectInfo(ctx, bucket, object, ObjectOptions{}) - return err == nil -} - -// Identical to fs PutObject operation except that it uses ETag in metadata -// headers. -func (cfs *cacheFSObjects) PutObject(ctx context.Context, bucket string, object string, r *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, retErr error) { - data := r.Reader - fs := cfs.FSObjects - // Lock the object. - objectLock := fs.nsMutex.NewNSLock(bucket, object) - if err := objectLock.GetLock(globalObjectTimeout); err != nil { - return objInfo, err - } - defer objectLock.Unlock() - - // No metadata is set, allocate a new one. - meta := make(map[string]string) - for k, v := range opts.UserDefined { - meta[k] = v - } - - var err error - - // Validate if bucket name is valid and exists. - if _, err = fs.statBucketDir(ctx, bucket); err != nil { - return ObjectInfo{}, toObjectErr(err, bucket) - } - - fsMeta := newFSMetaV1() - fsMeta.Meta = meta - - // This is a special case with size as '0' and object ends - // with a slash separator, we treat it like a valid operation - // and return success. - if isObjectDir(object, data.Size()) { - // Check if an object is present as one of the parent dir. - if fs.parentDirIsObject(ctx, bucket, path.Dir(object)) { - return ObjectInfo{}, toObjectErr(errFileParentIsFile, bucket, object) - } - if err = mkdirAll(pathJoin(fs.fsPath, bucket, object), 0777); err != nil { - return ObjectInfo{}, toObjectErr(err, bucket, object) - } - var fi os.FileInfo - if fi, err = fsStatDir(ctx, pathJoin(fs.fsPath, bucket, object)); err != nil { - return ObjectInfo{}, toObjectErr(err, bucket, object) - } - return fsMeta.ToObjectInfo(bucket, object, fi), nil - } - - if err = checkPutObjectArgs(ctx, bucket, object, fs, data.Size()); err != nil { - return ObjectInfo{}, err - } - - // Check if an object is present as one of the parent dir. - if fs.parentDirIsObject(ctx, bucket, path.Dir(object)) { - return ObjectInfo{}, toObjectErr(errFileParentIsFile, bucket, object) - } - - // Validate input data size and it can never be less than zero. - if data.Size() < -1 { - logger.LogIf(ctx, errInvalidArgument) - return ObjectInfo{}, errInvalidArgument - } - - var wlk *lock.LockedFile - if bucket != minioMetaBucket { - bucketMetaDir := pathJoin(fs.fsPath, minioMetaBucket, bucketMetaPrefix) - fsMetaPath := pathJoin(bucketMetaDir, bucket, object, fs.metaJSONFile) - - wlk, err = fs.rwPool.Create(fsMetaPath) - if err != nil { - logger.LogIf(ctx, err) - return ObjectInfo{}, toObjectErr(err, bucket, object) - } - // This close will allow for locks to be synchronized on `fs.json`. - defer wlk.Close() - defer func() { - // Remove meta file when PutObject encounters any error - if retErr != nil { - tmpDir := pathJoin(fs.fsPath, minioMetaTmpBucket, fs.fsUUID) - fsRemoveMeta(ctx, bucketMetaDir, fsMetaPath, tmpDir) - } - }() - } - - // Uploaded object will first be written to the temporary location which will eventually - // be renamed to the actual location. It is first written to the temporary location - // so that cleaning it up will be easy if the server goes down. - tempObj := mustGetUUID() - - // Allocate a buffer to Read() from request body - bufSize := int64(readSizeV1) - if size := data.Size(); size > 0 && bufSize > size { - bufSize = size - } - - buf := make([]byte, int(bufSize)) - fsTmpObjPath := pathJoin(fs.fsPath, minioMetaTmpBucket, fs.fsUUID, tempObj) - bytesWritten, err := fsCreateFile(ctx, fsTmpObjPath, data, buf, data.Size()) - if err != nil { - fsRemoveFile(ctx, fsTmpObjPath) - return ObjectInfo{}, toObjectErr(err, bucket, object) - } - if fsMeta.Meta["etag"] == "" { - fsMeta.Meta["etag"] = r.MD5CurrentHexString() - } - // Should return IncompleteBody{} error when reader has fewer - // bytes than specified in request header. - if bytesWritten < data.Size() { - fsRemoveFile(ctx, fsTmpObjPath) - return ObjectInfo{}, IncompleteBody{} - } - - // Delete the temporary object in the case of a - // failure. If PutObject succeeds, then there would be - // nothing to delete. - defer fsRemoveFile(ctx, fsTmpObjPath) - - // Entire object was written to the temp location, now it's safe to rename it to the actual location. - fsNSObjPath := pathJoin(fs.fsPath, bucket, object) - if err = fsRenameFile(ctx, fsTmpObjPath, fsNSObjPath); err != nil { - return ObjectInfo{}, toObjectErr(err, bucket, object) - } - - if bucket != minioMetaBucket { - // Write FS metadata after a successful namespace operation. - if _, err = fsMeta.WriteTo(wlk); err != nil { - return ObjectInfo{}, toObjectErr(err, bucket, object) - } - } - - // Stat the file to fetch timestamp, size. - fi, err := fsStatFile(ctx, pathJoin(fs.fsPath, bucket, object)) - if err != nil { - return ObjectInfo{}, toObjectErr(err, bucket, object) - } - // Success. - return fsMeta.ToObjectInfo(bucket, object, fi), nil -} - -// Implements S3 compatible initiate multipart API. Operation here is identical -// to fs backend implementation - with the exception that cache FS uses the uploadID -// generated on the backend -func (cfs *cacheFSObjects) NewMultipartUpload(ctx context.Context, bucket, object string, uploadID string, opts ObjectOptions) (string, error) { - if cfs.diskUsageHigh() { - select { - case cfs.purgeChan <- struct{}{}: - default: - } - return "", errDiskFull - } - - if _, err := cfs.GetBucketInfo(ctx, bucket); err != nil { - pErr := cfs.MakeBucketWithLocation(ctx, bucket, "") - if pErr != nil { - return "", pErr - } - } - fs := cfs.FSObjects - if err := checkNewMultipartArgs(ctx, bucket, object, fs); err != nil { - return "", toObjectErr(err, bucket) - } - - if _, err := fs.statBucketDir(ctx, bucket); err != nil { - return "", toObjectErr(err, bucket) - } - - uploadIDDir := fs.getUploadIDDir(bucket, object, uploadID) - - err := mkdirAll(uploadIDDir, 0755) - if err != nil { - logger.LogIf(ctx, err) - return "", err - } - - // Initialize fs.json values. - fsMeta := newFSMetaV1() - fsMeta.Meta = opts.UserDefined - - fsMetaBytes, err := json.Marshal(fsMeta) - if err != nil { - logger.LogIf(ctx, err) - return "", err - } - - if err = ioutil.WriteFile(pathJoin(uploadIDDir, fs.metaJSONFile), fsMetaBytes, 0644); err != nil { - logger.LogIf(ctx, err) - return "", err - } - return uploadID, nil -} - -// moveBucketToTrash clears cacheFSObjects of bucket contents and moves it to trash folder. -func (cfs *cacheFSObjects) moveBucketToTrash(ctx context.Context, bucket string) (err error) { - fs := cfs.FSObjects - bucketLock := fs.nsMutex.NewNSLock(bucket, "") - if err = bucketLock.GetLock(globalObjectTimeout); err != nil { - return err - } - defer bucketLock.Unlock() - bucketDir, err := fs.getBucketDir(ctx, bucket) - if err != nil { - return toObjectErr(err, bucket) - } - trashPath := pathJoin(cfs.fsPath, minioMetaBucket, cacheTrashDir) - expiredDir := path.Join(trashPath, bucket) - // Attempt to move regular bucket to expired directory. - if err = fsRenameDir(bucketDir, expiredDir); err != nil { - logger.LogIf(ctx, err) - return toObjectErr(err, bucket) - } - // Cleanup all the bucket metadata. - ominioMetadataBucketDir := pathJoin(fs.fsPath, minioMetaBucket, bucketMetaPrefix, bucket) - nminioMetadataBucketDir := pathJoin(trashPath, MustGetUUID()) - logger.LogIf(ctx, fsRenameDir(ominioMetadataBucketDir, nminioMetadataBucketDir)) - return nil -} - -// Removes a directory only if its empty, handles long -// paths for windows automatically. -func fsRenameDir(dirPath, newPath string) (err error) { - if dirPath == "" || newPath == "" { - return errInvalidArgument - } - - if err = checkPathLength(dirPath); err != nil { - return err - } - if err = checkPathLength(newPath); err != nil { - return err - } - if err = os.Rename(dirPath, newPath); err != nil { - if os.IsNotExist(err) { - return errVolumeNotFound - } else if isSysErrNotEmpty(err) { - return errVolumeNotEmpty - } - return err - } - return nil -} diff --git a/cmd/disk-cache-utils.go b/cmd/disk-cache-utils.go new file mode 100644 index 000000000..f8d9f8bf6 --- /dev/null +++ b/cmd/disk-cache-utils.go @@ -0,0 +1,146 @@ +/* + * MinIO Cloud Storage, (C) 2019 MinIO, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package cmd + +import ( + "net/http" + "strconv" + "strings" + "time" + + "github.com/minio/minio/cmd/crypto" +) + +type cacheControl struct { + exclude bool + expiry time.Time + maxAge int + sMaxAge int + minFresh int +} + +// cache exclude directives in cache-control header +var cacheExcludeDirectives = []string{ + "no-cache", + "no-store", + "must-revalidate", +} + +// returns true if cache exclude directives are set. +func isCacheExcludeDirective(s string) bool { + for _, directive := range cacheExcludeDirectives { + if s == directive { + return true + } + } + return false +} + +// returns struct with cache-control settings from user metadata. +func getCacheControlOpts(m map[string]string) (c cacheControl, err error) { + var headerVal string + for k, v := range m { + if k == "cache-control" { + headerVal = v + } + if k == "expires" { + if e, err := http.ParseTime(v); err == nil { + c.expiry = e + } + } + } + if headerVal == "" { + return + } + headerVal = strings.ToLower(headerVal) + headerVal = strings.TrimSpace(headerVal) + + vals := strings.Split(headerVal, ",") + for _, val := range vals { + val = strings.TrimSpace(val) + p := strings.Split(val, "=") + if isCacheExcludeDirective(p[0]) { + c.exclude = true + continue + } + + if len(p) != 2 { + continue + } + if p[0] == "max-age" || + p[0] == "s-maxage" || + p[0] == "min-fresh" { + i, err := strconv.Atoi(p[1]) + if err != nil { + return c, err + } + if p[0] == "max-age" { + c.maxAge = i + } + if p[0] == "s-maxage" { + c.sMaxAge = i + } + if p[0] == "min-fresh" { + c.minFresh = i + } + } + } + return c, nil +} + +// return true if metadata has a cache-control header +// directive to exclude object from cache. +func filterFromCache(m map[string]string) bool { + c, err := getCacheControlOpts(m) + if err != nil { + return false + } + return c.exclude +} + +// returns true if cache expiry conditions met in cache-control/expiry metadata. +func isStaleCache(objInfo ObjectInfo) bool { + c, err := getCacheControlOpts(objInfo.UserDefined) + if err != nil { + return false + } + now := time.Now() + if c.sMaxAge > 0 && c.sMaxAge > int(now.Sub(objInfo.ModTime).Seconds()) { + return true + } + if c.maxAge > 0 && c.maxAge > int(now.Sub(objInfo.ModTime).Seconds()) { + return true + } + if !c.expiry.Equal(time.Time{}) && c.expiry.Before(time.Now()) { + return true + } + if c.minFresh > 0 && c.minFresh <= int(now.Sub(objInfo.ModTime).Seconds()) { + return true + } + return false +} + +// backendDownError returns true if err is due to backend failure or faulty disk if in server mode +func backendDownError(err error) bool { + _, backendDown := err.(BackendDown) + return backendDown || IsErr(err, baseErrs...) +} + +// IsCacheable returns if the object should be saved in the cache. +func (o ObjectInfo) IsCacheable() bool { + return !crypto.IsEncrypted(o.UserDefined) +} diff --git a/cmd/disk-cache.go b/cmd/disk-cache.go index ed84bf6cb..b470deb55 100644 --- a/cmd/disk-cache.go +++ b/cmd/disk-cache.go @@ -1,19 +1,3 @@ -/* - * MinIO Cloud Storage, (C) 2018 MinIO, Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - package cmd import ( @@ -24,56 +8,13 @@ import ( "io/ioutil" "net/http" "os" - "sort" - "strconv" - "strings" - "time" "github.com/djherbis/atime" - - "github.com/minio/minio/cmd/crypto" "github.com/minio/minio/cmd/logger" "github.com/minio/minio/pkg/hash" "github.com/minio/minio/pkg/wildcard" ) -const ( - // disk cache needs to have object size space free for a cache entry to be created. - cacheTrashDir = "trash" - cacheCleanupInterval = 10 // in minutes -) - -// abstract slice of cache drives backed by FS. -type diskCache struct { - cfs []*cacheFSObjects -} - -// Abstracts disk caching - used by the S3 layer -type cacheObjects struct { - // pointer to disk cache - cache *diskCache - // ListObjects pool management. - listPool *TreeWalkPool - // file path patterns to exclude from cache - exclude []string - // Object functions pointing to the corresponding functions of backend implementation. - GetObjectNInfoFn func(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, lockType LockType, opts ObjectOptions) (gr *GetObjectReader, err error) - GetObjectFn func(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) (err error) - GetObjectInfoFn func(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error) - PutObjectFn func(ctx context.Context, bucket, object string, data *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error) - DeleteObjectFn func(ctx context.Context, bucket, object string) error - DeleteObjectsFn func(ctx context.Context, bucket string, objects []string) ([]error, error) - ListObjectsFn func(ctx context.Context, bucket, prefix, marker, delimiter string, maxKeys int) (result ListObjectsInfo, err error) - ListObjectsV2Fn func(ctx context.Context, bucket, prefix, continuationToken, delimiter string, maxKeys int, fetchOwner bool, startAfter string) (result ListObjectsV2Info, err error) - ListBucketsFn func(ctx context.Context) (buckets []BucketInfo, err error) - GetBucketInfoFn func(ctx context.Context, bucket string) (bucketInfo BucketInfo, err error) - NewMultipartUploadFn func(ctx context.Context, bucket, object string, opts ObjectOptions) (uploadID string, err error) - PutObjectPartFn func(ctx context.Context, bucket, object, uploadID string, partID int, data *PutObjReader, opts ObjectOptions) (info PartInfo, err error) - AbortMultipartUploadFn func(ctx context.Context, bucket, object, uploadID string) error - CompleteMultipartUploadFn func(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart, opts ObjectOptions) (objInfo ObjectInfo, err error) - DeleteBucketFn func(ctx context.Context, bucket string) error -} - // CacheStorageInfo - represents total, free capacity of // underlying cache storage. type CacheStorageInfo struct { @@ -83,92 +24,49 @@ type CacheStorageInfo struct { // CacheObjectLayer implements primitives for cache object API layer. type CacheObjectLayer interface { - // Bucket operations. - ListObjects(ctx context.Context, bucket, prefix, marker, delimiter string, maxKeys int) (result ListObjectsInfo, err error) - ListObjectsV2(ctx context.Context, bucket, prefix, continuationToken, delimiter string, maxKeys int, fetchOwner bool, startAfter string) (result ListObjectsV2Info, err error) - GetBucketInfo(ctx context.Context, bucket string) (bucketInfo BucketInfo, err error) - ListBuckets(ctx context.Context) (buckets []BucketInfo, err error) - DeleteBucket(ctx context.Context, bucket string) error // Object operations. GetObjectNInfo(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, lockType LockType, opts ObjectOptions) (gr *GetObjectReader, err error) - GetObject(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) (err error) GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error) - PutObject(ctx context.Context, bucket, object string, data *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error) DeleteObject(ctx context.Context, bucket, object string) error DeleteObjects(ctx context.Context, bucket string, objects []string) ([]error, error) - - // Multipart operations. - NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (uploadID string, err error) - PutObjectPart(ctx context.Context, bucket, object, uploadID string, partID int, data *PutObjReader, opts ObjectOptions) (info PartInfo, err error) - AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string) error - CompleteMultipartUpload(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart, opts ObjectOptions) (objInfo ObjectInfo, err error) - // Storage operations. StorageInfo(ctx context.Context) CacheStorageInfo } -// IsCacheable returns if the object should be saved in the cache. -func (o ObjectInfo) IsCacheable() bool { - return !crypto.IsEncrypted(o.UserDefined) +// Abstracts disk caching - used by the S3 layer +type cacheObjects struct { + // slice of cache drives + cache []*diskCache + // file path patterns to exclude from cache + exclude []string + // Object functions pointing to the corresponding functions of backend implementation. + GetObjectNInfoFn func(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, lockType LockType, opts ObjectOptions) (gr *GetObjectReader, err error) + GetObjectInfoFn func(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error) + DeleteObjectFn func(ctx context.Context, bucket, object string) error + DeleteObjectsFn func(ctx context.Context, bucket string, objects []string) ([]error, error) } -// backendDownError returns true if err is due to backend failure or faulty disk if in server mode -func backendDownError(err error) bool { - _, backendDown := err.(BackendDown) - return backendDown || IsErr(err, baseErrs...) -} - -// get cache disk where object is currently cached for a GET operation. If object does not exist at that location, -// treat the list of cache drives as a circular buffer and walk through them starting at hash index -// until an online drive is found.If object is not found, fall back to the first online cache drive -// closest to the hash index, so that object can be recached. -func (c diskCache) getCachedFSLoc(ctx context.Context, bucket, object string) (*cacheFSObjects, error) { - index := c.hashIndex(bucket, object) - numDisks := len(c.cfs) - // save first online cache disk closest to the hint index - var firstOnlineDisk *cacheFSObjects - for k := 0; k < numDisks; k++ { - i := (index + k) % numDisks - if c.cfs[i] == nil { - continue - } - if c.cfs[i].IsOnline() { - if firstOnlineDisk == nil { - firstOnlineDisk = c.cfs[i] - } - if c.cfs[i].Exists(ctx, bucket, object) { - return c.cfs[i], nil - } - } +// DeleteObject clears cache entry if backend delete operation succeeds +func (c cacheObjects) DeleteObject(ctx context.Context, bucket, object string) (err error) { + if err = c.DeleteObjectFn(ctx, bucket, object); err != nil { + return } - - if firstOnlineDisk != nil { - return firstOnlineDisk, nil + if c.isCacheExclude(bucket, object) { + return } - return nil, errDiskNotFound + if dcache, cerr := c.getCachedLoc(ctx, bucket, object); cerr == nil { + dcache.Delete(ctx, bucket, object) + } + return } -// choose a cache deterministically based on hash of bucket,object. The hash index is treated as -// a hint. In the event that the cache drive at hash index is offline, treat the list of cache drives -// as a circular buffer and walk through them starting at hash index until an online drive is found. -func (c diskCache) getCacheFS(ctx context.Context, bucket, object string) (*cacheFSObjects, error) { - index := c.hashIndex(bucket, object) - numDisks := len(c.cfs) - for k := 0; k < numDisks; k++ { - i := (index + k) % numDisks - if c.cfs[i] == nil { - continue - } - if c.cfs[i].IsOnline() { - return c.cfs[i], nil - } +// DeleteObjects batch deletes objects in slice, and clears any cached entries +func (c cacheObjects) DeleteObjects(ctx context.Context, bucket string, objects []string) ([]error, error) { + errs := make([]error, len(objects)) + for idx, object := range objects { + errs[idx] = c.DeleteObject(ctx, bucket, object) } - return nil, errDiskNotFound -} - -// Compute a unique hash sum for bucket and object -func (c diskCache) hashIndex(bucket, object string) int { - return crcHashMod(pathJoin(bucket, object), len(c.cfs)) + return errs, nil } // construct a metadata k-v map @@ -189,15 +87,15 @@ func (c cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string, return c.GetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts) } - // fetch cacheFSObjects if object is currently cached or nearest available cache drive - dcache, err := c.cache.getCachedFSLoc(ctx, bucket, object) + // fetch diskCache if object is currently cached or nearest available cache drive + dcache, err := c.getCachedLoc(ctx, bucket, object) if err != nil { return c.GetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts) } - cacheReader, cacheErr := dcache.GetObjectNInfo(ctx, bucket, object, rs, h, lockType, opts) - + cacheReader, cacheErr := dcache.Get(ctx, bucket, object, rs, h, opts) objInfo, err := c.GetObjectInfoFn(ctx, bucket, object, opts) + if backendDownError(err) && cacheErr == nil { return cacheReader, nil } else if err != nil { @@ -229,11 +127,24 @@ func (c cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string, // Since we got here, we are serving the request from backend, // and also adding the object to the cache. - if rs != nil { - // We don't cache partial objects. + if !dcache.diskAvailable(objInfo.Size) { return c.GetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts) } - if !dcache.diskAvailable(objInfo.Size) { + + if rs != nil { + go func() { + // fill cache in the background for range GET requests + bReader, bErr := c.GetObjectNInfoFn(ctx, bucket, object, nil, h, lockType, opts) + if bErr != nil { + return + } + defer bReader.Close() + oi, err := dcache.Stat(ctx, bucket, object) + // avoid cache overwrite if another background routine filled cache + if err != nil || oi.ETag != bReader.ObjInfo.ETag { + dcache.Put(ctx, bucket, object, bReader, bReader.ObjInfo.Size, ObjectOptions{UserDefined: c.getMetadata(bReader.ObjInfo)}) + } + }() return c.GetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts) } @@ -252,96 +163,26 @@ func (c cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string, } go func() { - putErr := dcache.Put(ctx, bucket, object, NewPutObjReader(hashReader, nil, nil), ObjectOptions{UserDefined: c.getMetadata(bkReader.ObjInfo)}) + putErr := dcache.Put(ctx, bucket, object, hashReader, hashReader.Size(), ObjectOptions{UserDefined: c.getMetadata(bkReader.ObjInfo)}) // close the write end of the pipe, so the error gets // propagated to getObjReader pipeWriter.CloseWithError(putErr) }() - cleanupBackend := func() { bkReader.Close() } cleanupPipe := func() { pipeReader.Close() } return NewGetObjectReaderFromReader(teeReader, bkReader.ObjInfo, opts.CheckCopyPrecondFn, cleanupBackend, cleanupPipe) } -// Uses cached-object to serve the request. If object is not cached it serves the request from the backend and also -// stores it in the cache for serving subsequent requests. -func (c cacheObjects) GetObject(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) (err error) { - GetObjectFn := c.GetObjectFn - GetObjectInfoFn := c.GetObjectInfoFn - - if c.isCacheExclude(bucket, object) { - return GetObjectFn(ctx, bucket, object, startOffset, length, writer, etag, opts) - } - // fetch cacheFSObjects if object is currently cached or nearest available cache drive - dcache, err := c.cache.getCachedFSLoc(ctx, bucket, object) - if err != nil { - return GetObjectFn(ctx, bucket, object, startOffset, length, writer, etag, opts) - } - // stat object on backend - objInfo, err := GetObjectInfoFn(ctx, bucket, object, opts) - backendDown := backendDownError(err) - if err != nil && !backendDown { - if _, ok := err.(ObjectNotFound); ok { - // Delete the cached entry if backend object was deleted. - dcache.Delete(ctx, bucket, object) - } - return err - } - - if !backendDown && !objInfo.IsCacheable() { - return GetObjectFn(ctx, bucket, object, startOffset, length, writer, etag, opts) - } - - if !backendDown && filterFromCache(objInfo.UserDefined) { - return GetObjectFn(ctx, bucket, object, startOffset, length, writer, etag, opts) - } - - cachedObjInfo, err := dcache.GetObjectInfo(ctx, bucket, object, opts) - if err == nil { - if backendDown { - // If the backend is down, serve the request from cache. - return dcache.Get(ctx, bucket, object, startOffset, length, writer, etag, opts) - } - if cachedObjInfo.ETag == objInfo.ETag && !isStaleCache(objInfo) { - return dcache.Get(ctx, bucket, object, startOffset, length, writer, etag, opts) - } - dcache.Delete(ctx, bucket, object) - } - if startOffset != 0 || (length > 0 && length != objInfo.Size) { - // We don't cache partial objects. - return GetObjectFn(ctx, bucket, object, startOffset, length, writer, etag, opts) - } - if !dcache.diskAvailable(objInfo.Size) { - return GetObjectFn(ctx, bucket, object, startOffset, length, writer, etag, opts) - } - // Initialize pipe. - pipeReader, pipeWriter := io.Pipe() - hashReader, err := hash.NewReader(pipeReader, objInfo.Size, "", "", objInfo.Size, globalCLIContext.StrictS3Compat) - if err != nil { - return err - } - go func() { - gerr := GetObjectFn(ctx, bucket, object, 0, objInfo.Size, io.MultiWriter(writer, pipeWriter), etag, opts) - pipeWriter.CloseWithError(gerr) // Close writer explicitly signaling we wrote all data. - }() - - opts.UserDefined = c.getMetadata(objInfo) - err = dcache.Put(ctx, bucket, object, NewPutObjReader(hashReader, nil, nil), opts) - if err != nil { - return err - } - pipeReader.Close() - return -} - // Returns ObjectInfo from cache if available. func (c cacheObjects) GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) { getObjectInfoFn := c.GetObjectInfoFn + if c.isCacheExclude(bucket, object) { return getObjectInfoFn(ctx, bucket, object, opts) } - // fetch cacheFSObjects if object is currently cached or nearest available cache drive - dcache, err := c.cache.getCachedFSLoc(ctx, bucket, object) + + // fetch diskCache if object is currently cached or nearest available cache drive + dcache, err := c.getCachedLoc(ctx, bucket, object) if err != nil { return getObjectInfoFn(ctx, bucket, object, opts) } @@ -356,14 +197,14 @@ func (c cacheObjects) GetObjectInfo(ctx context.Context, bucket, object string, return ObjectInfo{}, err } // when backend is down, serve from cache. - cachedObjInfo, cerr := dcache.GetObjectInfo(ctx, bucket, object, opts) + cachedObjInfo, cerr := dcache.Stat(ctx, bucket, object) if cerr == nil { return cachedObjInfo, nil } return ObjectInfo{}, BackendDown{} } // when backend is up, do a sanity check on cached object - cachedObjInfo, err := dcache.GetObjectInfo(ctx, bucket, object, opts) + cachedObjInfo, err := dcache.Stat(ctx, bucket, object) if err != nil { return objInfo, nil } @@ -374,269 +215,23 @@ func (c cacheObjects) GetObjectInfo(ctx context.Context, bucket, object string, return objInfo, nil } -// Returns function "listDir" of the type listDirFunc. -// isLeaf - is used by listDir function to check if an entry is a leaf or non-leaf entry. -// disks - list of fsObjects -func listDirCacheFactory(isLeaf func(string, string) bool, disks []*cacheFSObjects) ListDirFunc { - listCacheDirs := func(bucket, prefixDir, prefixEntry string) (dirs []string) { - var entries []string - for _, disk := range disks { - // ignore disk-caches that might be missing/offline - if disk == nil { - continue - } - - fs := disk.FSObjects - var err error - entries, err = readDir(pathJoin(fs.fsPath, bucket, prefixDir)) - if err != nil { - continue - } - - for i := range entries { - if isLeaf(bucket, entries[i]) { - entries[i] = strings.TrimSuffix(entries[i], slashSeparator) - } - } - - // Filter entries that have the prefix prefixEntry. - entries = filterMatchingPrefix(entries, prefixEntry) - dirs = append(dirs, entries...) - } - return dirs - } - - // listDir - lists all the entries at a given prefix and given entry in the prefix. - listDir := func(bucket, prefixDir, prefixEntry string) (mergedEntries []string) { - cacheEntries := listCacheDirs(bucket, prefixDir, prefixEntry) - for _, entry := range cacheEntries { - // Find elements in entries which are not in mergedEntries - idx := sort.SearchStrings(mergedEntries, entry) - // if entry is already present in mergedEntries don't add. - if idx < len(mergedEntries) && mergedEntries[idx] == entry { - continue - } - mergedEntries = append(mergedEntries, entry) - sort.Strings(mergedEntries) - } - return mergedEntries - } - return listDir -} - -// List all objects at prefix upto maxKeys, optionally delimited by '/' from the cache. Maintains the list pool -// state for future re-entrant list requests. -func (c cacheObjects) listCacheObjects(ctx context.Context, bucket, prefix, marker, delimiter string, maxKeys int) (result ListObjectsInfo, err error) { - var objInfos []ObjectInfo - var eof bool - var nextMarker string - - recursive := true - if delimiter == slashSeparator { - recursive = false - } - walkResultCh, endWalkCh := c.listPool.Release(listParams{bucket, recursive, marker, prefix}) - if walkResultCh == nil { - endWalkCh = make(chan struct{}) - - listDir := listDirCacheFactory(func(bucket, object string) bool { - fs, err := c.cache.getCacheFS(ctx, bucket, object) - if err != nil { - return false - } - _, err = fs.getObjectInfo(ctx, bucket, object) - return err == nil - }, c.cache.cfs) - walkResultCh = startTreeWalk(ctx, bucket, prefix, marker, recursive, listDir, endWalkCh) - } - - for i := 0; i < maxKeys; { - walkResult, ok := <-walkResultCh - if !ok { - // Closed channel. - eof = true - break - } - - entry := walkResult.entry - var objInfo ObjectInfo - if hasSuffix(entry, slashSeparator) { - // Object name needs to be full path. - objInfo.Bucket = bucket - objInfo.Name = entry - objInfo.IsDir = true - } else { - // Set the Mode to a "regular" file. - var err error - fs, err := c.cache.getCacheFS(ctx, bucket, entry) - if err != nil { - // Ignore errDiskNotFound - if err == errDiskNotFound { - continue - } - return result, toObjectErr(err, bucket, prefix) - } - objInfo, err = fs.getObjectInfo(ctx, bucket, entry) - if err != nil { - // Ignore ObjectNotFound error - if _, ok := err.(ObjectNotFound); ok { - continue - } - return result, toObjectErr(err, bucket, prefix) - } - } - nextMarker = objInfo.Name - objInfos = append(objInfos, objInfo) - i++ - if walkResult.end { - eof = true - break - } - } - - params := listParams{bucket, recursive, nextMarker, prefix} - if !eof { - c.listPool.Set(params, walkResultCh, endWalkCh) - } - - result = ListObjectsInfo{IsTruncated: !eof} - for _, objInfo := range objInfos { - result.NextMarker = objInfo.Name - if objInfo.IsDir && delimiter == slashSeparator { - result.Prefixes = append(result.Prefixes, objInfo.Name) - continue - } - result.Objects = append(result.Objects, objInfo) - } - return result, nil -} - -// listCacheV2Objects lists all blobs in bucket filtered by prefix from the cache -func (c cacheObjects) listCacheV2Objects(ctx context.Context, bucket, prefix, continuationToken, delimiter string, maxKeys int, fetchOwner bool, startAfter string) (result ListObjectsV2Info, err error) { - loi, err := c.listCacheObjects(ctx, bucket, prefix, continuationToken, delimiter, maxKeys) - if err != nil { - return result, err - } - - listObjectsV2Info := ListObjectsV2Info{ - IsTruncated: loi.IsTruncated, - ContinuationToken: continuationToken, - NextContinuationToken: loi.NextMarker, - Objects: loi.Objects, - Prefixes: loi.Prefixes, - } - return listObjectsV2Info, err -} - -// List all objects at prefix upto maxKeys., optionally delimited by '/'. Maintains the list pool -// state for future re-entrant list requests. Retrieve from cache if backend is down -func (c cacheObjects) ListObjects(ctx context.Context, bucket, prefix, marker, delimiter string, maxKeys int) (result ListObjectsInfo, err error) { - - listObjectsFn := c.ListObjectsFn - - result, err = listObjectsFn(ctx, bucket, prefix, marker, delimiter, maxKeys) - if err != nil { - if backendDownError(err) { - return c.listCacheObjects(ctx, bucket, prefix, marker, delimiter, maxKeys) - } - return - } - return -} - -// ListObjectsV2 lists all blobs in bucket filtered by prefix -func (c cacheObjects) ListObjectsV2(ctx context.Context, bucket, prefix, continuationToken, delimiter string, maxKeys int, fetchOwner bool, startAfter string) (result ListObjectsV2Info, err error) { - listObjectsV2Fn := c.ListObjectsV2Fn - - result, err = listObjectsV2Fn(ctx, bucket, prefix, continuationToken, delimiter, maxKeys, fetchOwner, startAfter) - if err != nil { - if backendDownError(err) { - return c.listCacheV2Objects(ctx, bucket, prefix, continuationToken, delimiter, maxKeys, fetchOwner, startAfter) - } - return - } - return -} - -// Lists all the buckets in the cache -func (c cacheObjects) listBuckets(ctx context.Context) (buckets []BucketInfo, err error) { - m := make(map[string]string) - for _, cache := range c.cache.cfs { - // ignore disk-caches that might be missing/offline +// StorageInfo - returns underlying storage statistics. +func (c cacheObjects) StorageInfo(ctx context.Context) (cInfo CacheStorageInfo) { + var total, free uint64 + for _, cache := range c.cache { if cache == nil { continue } - entries, err := cache.ListBuckets(ctx) - - if err != nil { - return nil, err - } - for _, entry := range entries { - _, ok := m[entry.Name] - if !ok { - m[entry.Name] = entry.Name - buckets = append(buckets, entry) - } - } + info, err := getDiskInfo(cache.dir) + logger.GetReqInfo(ctx).AppendTags("cachePath", cache.dir) + logger.LogIf(ctx, err) + total += info.Total + free += info.Free } - // Sort bucket infos by bucket name. - sort.Sort(byBucketName(buckets)) - return -} - -// Returns list of buckets from cache or the backend. If the backend is down, buckets -// available on cache are served. -func (c cacheObjects) ListBuckets(ctx context.Context) (buckets []BucketInfo, err error) { - listBucketsFn := c.ListBucketsFn - buckets, err = listBucketsFn(ctx) - if err != nil { - if backendDownError(err) { - return c.listBuckets(ctx) - } - return []BucketInfo{}, err + return CacheStorageInfo{ + Total: total, + Free: free, } - return -} - -// Returns bucket info from cache if backend is down. -func (c cacheObjects) GetBucketInfo(ctx context.Context, bucket string) (bucketInfo BucketInfo, err error) { - getBucketInfoFn := c.GetBucketInfoFn - bucketInfo, err = getBucketInfoFn(ctx, bucket) - if backendDownError(err) { - for _, cache := range c.cache.cfs { - // ignore disk-caches that might be missing/offline - if cache == nil { - continue - } - if bucketInfo, err = cache.GetBucketInfo(ctx, bucket); err == nil { - return - } - } - } - return -} - -// Delete Object deletes from cache as well if backend operation succeeds -func (c cacheObjects) DeleteObject(ctx context.Context, bucket, object string) (err error) { - if err = c.DeleteObjectFn(ctx, bucket, object); err != nil { - return - } - if c.isCacheExclude(bucket, object) { - return - } - dcache, cerr := c.cache.getCachedFSLoc(ctx, bucket, object) - if cerr == nil { - _ = dcache.DeleteObject(ctx, bucket, object) - } - return -} - -func (c cacheObjects) DeleteObjects(ctx context.Context, bucket string, objects []string) ([]error, error) { - errs := make([]error, len(objects)) - for idx, object := range objects { - errs[idx] = c.DeleteObject(ctx, bucket, object) - } - return errs, nil } // Returns true if object should be excluded from cache @@ -650,288 +245,70 @@ func (c cacheObjects) isCacheExclude(bucket, object string) bool { return false } -// PutObject - caches the uploaded object for single Put operations -func (c cacheObjects) PutObject(ctx context.Context, bucket, object string, r *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error) { - putObjectFn := c.PutObjectFn - data := r.Reader - dcache, err := c.cache.getCacheFS(ctx, bucket, object) - if err != nil { - // disk cache could not be located,execute backend call. - return putObjectFn(ctx, bucket, object, r, opts) - } - size := r.Size() +// get cache disk where object is currently cached for a GET operation. If object does not exist at that location, +// treat the list of cache drives as a circular buffer and walk through them starting at hash index +// until an online drive is found.If object is not found, fall back to the first online cache drive +// closest to the hash index, so that object can be re-cached. +func (c cacheObjects) getCachedLoc(ctx context.Context, bucket, object string) (*diskCache, error) { + index := c.hashIndex(bucket, object) - // fetch from backend if there is no space on cache drive - if !dcache.diskAvailable(size) { - return putObjectFn(ctx, bucket, object, r, opts) - } - // fetch from backend if cache exclude pattern or cache-control - // directive set to exclude - if c.isCacheExclude(bucket, object) || filterFromCache(opts.UserDefined) { - dcache.Delete(ctx, bucket, object) - return putObjectFn(ctx, bucket, object, r, opts) - } - objInfo = ObjectInfo{} - // Initialize pipe to stream data to backend - pipeReader, pipeWriter := io.Pipe() - hashReader, err := hash.NewReader(pipeReader, size, data.MD5HexString(), data.SHA256HexString(), data.ActualSize(), globalCLIContext.StrictS3Compat) - if err != nil { - return ObjectInfo{}, err - } - // Initialize pipe to stream data to cache - rPipe, wPipe := io.Pipe() - cHashReader, err := hash.NewReader(rPipe, size, data.MD5HexString(), data.SHA256HexString(), data.ActualSize(), globalCLIContext.StrictS3Compat) - if err != nil { - return ObjectInfo{}, err - } - oinfoCh := make(chan ObjectInfo) - errCh := make(chan error) - go func() { - oinfo, perr := putObjectFn(ctx, bucket, object, NewPutObjReader(hashReader, nil, nil), opts) - if perr != nil { - pipeWriter.CloseWithError(perr) - wPipe.CloseWithError(perr) - close(oinfoCh) - errCh <- perr - return - } - close(errCh) - oinfoCh <- oinfo - }() - - go func() { - if err = dcache.Put(ctx, bucket, object, NewPutObjReader(cHashReader, nil, nil), opts); err != nil { - wPipe.CloseWithError(err) - return - } - }() - - mwriter := io.MultiWriter(pipeWriter, wPipe) - _, err = io.Copy(mwriter, data) - if err != nil { - err = <-errCh - return objInfo, err - } - pipeWriter.Close() - wPipe.Close() - objInfo = <-oinfoCh - return objInfo, err -} - -// NewMultipartUpload - Starts a new multipart upload operation to backend and cache. -func (c cacheObjects) NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (uploadID string, err error) { - newMultipartUploadFn := c.NewMultipartUploadFn - - if c.isCacheExclude(bucket, object) || filterFromCache(opts.UserDefined) { - return newMultipartUploadFn(ctx, bucket, object, opts) - } - - dcache, err := c.cache.getCacheFS(ctx, bucket, object) - if err != nil { - // disk cache could not be located,execute backend call. - return newMultipartUploadFn(ctx, bucket, object, opts) - } - - uploadID, err = newMultipartUploadFn(ctx, bucket, object, opts) - if err != nil { - return - } - // create new multipart upload in cache with same uploadID - dcache.NewMultipartUpload(ctx, bucket, object, uploadID, opts) - return uploadID, err -} - -// PutObjectPart - uploads part to backend and cache simultaneously. -func (c cacheObjects) PutObjectPart(ctx context.Context, bucket, object, uploadID string, partID int, r *PutObjReader, opts ObjectOptions) (info PartInfo, err error) { - data := r.Reader - putObjectPartFn := c.PutObjectPartFn - dcache, err := c.cache.getCacheFS(ctx, bucket, object) - if err != nil { - // disk cache could not be located,execute backend call. - return putObjectPartFn(ctx, bucket, object, uploadID, partID, r, opts) - } - - if c.isCacheExclude(bucket, object) { - return putObjectPartFn(ctx, bucket, object, uploadID, partID, r, opts) - } - - // make sure cache has at least size space available - size := data.Size() - if !dcache.diskAvailable(size) { - select { - case dcache.purgeChan <- struct{}{}: - default: - } - return putObjectPartFn(ctx, bucket, object, uploadID, partID, r, opts) - } - - info = PartInfo{} - // Initialize pipe to stream data to backend - pipeReader, pipeWriter := io.Pipe() - hashReader, err := hash.NewReader(pipeReader, size, data.MD5HexString(), data.SHA256HexString(), data.ActualSize(), globalCLIContext.StrictS3Compat) - if err != nil { - return - } - // Initialize pipe to stream data to cache - rPipe, wPipe := io.Pipe() - cHashReader, err := hash.NewReader(rPipe, size, data.MD5HexString(), data.SHA256HexString(), data.ActualSize(), globalCLIContext.StrictS3Compat) - if err != nil { - return - } - pinfoCh := make(chan PartInfo) - errorCh := make(chan error) - go func() { - info, err = putObjectPartFn(ctx, bucket, object, uploadID, partID, NewPutObjReader(hashReader, nil, nil), opts) - if err != nil { - close(pinfoCh) - pipeWriter.CloseWithError(err) - wPipe.CloseWithError(err) - errorCh <- err - return - } - close(errorCh) - pinfoCh <- info - }() - go func() { - if _, perr := dcache.PutObjectPart(ctx, bucket, object, uploadID, partID, NewPutObjReader(cHashReader, nil, nil), opts); perr != nil { - wPipe.CloseWithError(perr) - return - } - }() - - mwriter := io.MultiWriter(pipeWriter, wPipe) - _, err = io.Copy(mwriter, data) - if err != nil { - err = <-errorCh - return PartInfo{}, err - } - pipeWriter.Close() - wPipe.Close() - info = <-pinfoCh - return info, err -} - -// AbortMultipartUpload - aborts multipart upload on backend and cache. -func (c cacheObjects) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string) error { - abortMultipartUploadFn := c.AbortMultipartUploadFn - - if c.isCacheExclude(bucket, object) { - return abortMultipartUploadFn(ctx, bucket, object, uploadID) - } - - dcache, err := c.cache.getCacheFS(ctx, bucket, object) - if err != nil { - // disk cache could not be located,execute backend call. - return abortMultipartUploadFn(ctx, bucket, object, uploadID) - } - // execute backend operation - err = abortMultipartUploadFn(ctx, bucket, object, uploadID) - if err != nil { - return err - } - // abort multipart upload on cache - dcache.AbortMultipartUpload(ctx, bucket, object, uploadID) - return nil -} - -// CompleteMultipartUpload - completes multipart upload operation on backend and cache. -func (c cacheObjects) CompleteMultipartUpload(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart, opts ObjectOptions) (objInfo ObjectInfo, err error) { - completeMultipartUploadFn := c.CompleteMultipartUploadFn - - if c.isCacheExclude(bucket, object) { - return completeMultipartUploadFn(ctx, bucket, object, uploadID, uploadedParts, opts) - } - - dcache, err := c.cache.getCacheFS(ctx, bucket, object) - if err != nil { - // disk cache could not be located,execute backend call. - return completeMultipartUploadFn(ctx, bucket, object, uploadID, uploadedParts, opts) - } - // perform backend operation - objInfo, err = completeMultipartUploadFn(ctx, bucket, object, uploadID, uploadedParts, opts) - if err != nil { - return - } - // create new multipart upload in cache with same uploadID - dcache.CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts) - return -} - -// StorageInfo - returns underlying storage statistics. -func (c cacheObjects) StorageInfo(ctx context.Context) (cInfo CacheStorageInfo) { - var total, free uint64 - for _, cfs := range c.cache.cfs { - if cfs == nil { + numDisks := len(c.cache) + // save first online cache disk closest to the hint index + var firstOnlineDisk *diskCache + for k := 0; k < numDisks; k++ { + i := (index + k) % numDisks + if c.cache[i] == nil { continue } - info, err := getDiskInfo(cfs.fsPath) - logger.GetReqInfo(ctx).AppendTags("cachePath", cfs.fsPath) - logger.LogIf(ctx, err) - total += info.Total - free += info.Free + if c.cache[i].IsOnline() { + if firstOnlineDisk == nil { + firstOnlineDisk = c.cache[i] + } + if c.cache[i].Exists(ctx, bucket, object) { + return c.cache[i], nil + } + } } - return CacheStorageInfo{ - Total: total, - Free: free, + + if firstOnlineDisk != nil { + return firstOnlineDisk, nil } + return nil, errDiskNotFound } -// DeleteBucket - marks bucket to be deleted from cache if bucket is deleted from backend. -func (c cacheObjects) DeleteBucket(ctx context.Context, bucket string) (err error) { - deleteBucketFn := c.DeleteBucketFn - var toDel []*cacheFSObjects - for _, cfs := range c.cache.cfs { - // ignore disk-caches that might be missing/offline - if cfs == nil { - continue - } - if _, cerr := cfs.GetBucketInfo(ctx, bucket); cerr == nil { - toDel = append(toDel, cfs) - } - } - // perform backend operation - err = deleteBucketFn(ctx, bucket) - if err != nil { - return - } - // move bucket metadata and content to cache's trash dir - for _, d := range toDel { - d.moveBucketToTrash(ctx, bucket) - } - return +// Compute a unique hash sum for bucket and object +func (c cacheObjects) hashIndex(bucket, object string) int { + return crcHashMod(pathJoin(bucket, object), len(c.cache)) } // newCache initializes the cacheFSObjects for the "drives" specified in config.json // or the global env overrides. -func newCache(config CacheConfig) (*diskCache, error) { - var cfsObjects []*cacheFSObjects +func newCache(config CacheConfig) ([]*diskCache, error) { + var caches []*diskCache ctx := logger.SetReqInfo(context.Background(), &logger.ReqInfo{}) formats, err := loadAndValidateCacheFormat(ctx, config.Drives) if err != nil { return nil, err } for i, dir := range config.Drives { - // skip cacheFSObjects creation for cache drives missing a format.json + // skip diskCache creation for cache drives missing a format.json if formats[i] == nil { - cfsObjects = append(cfsObjects, nil) + caches = append(caches, nil) continue } if err := checkAtimeSupport(dir); err != nil { return nil, errors.New("Atime support required for disk caching") } - cache, err := newCacheFSObjects(dir, config.Expiry, config.MaxUse) + cache, err := newdiskCache(dir, config.Expiry, config.MaxUse) if err != nil { return nil, err } // Start the purging go-routine for entries that have expired go cache.purge() - // Start trash purge routine for deleted buckets. - go cache.purgeTrash() - - cfsObjects = append(cfsObjects, cache) + caches = append(caches, cache) } - return &diskCache{cfs: cfsObjects}, nil + return caches, nil } // Return error if Atime is disabled on the O/S @@ -960,27 +337,19 @@ func checkAtimeSupport(dir string) (err error) { // Returns cacheObjects for use by Server. func newServerCacheObjects(config CacheConfig) (CacheObjectLayer, error) { // list of disk caches for cache "drives" specified in config.json or MINIO_CACHE_DRIVES env var. - dcache, err := newCache(config) + cache, err := newCache(config) if err != nil { return nil, err } - return &cacheObjects{ - cache: dcache, - exclude: config.Exclude, - listPool: NewTreeWalkPool(globalLookupTimeout), - GetObjectFn: func(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) error { - return newObjectLayerFn().GetObject(ctx, bucket, object, startOffset, length, writer, etag, opts) - }, + cache: cache, + exclude: config.Exclude, GetObjectInfoFn: func(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) { return newObjectLayerFn().GetObjectInfo(ctx, bucket, object, opts) }, GetObjectNInfoFn: func(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, lockType LockType, opts ObjectOptions) (gr *GetObjectReader, err error) { return newObjectLayerFn().GetObjectNInfo(ctx, bucket, object, rs, h, lockType, opts) }, - PutObjectFn: func(ctx context.Context, bucket, object string, data *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error) { - return newObjectLayerFn().PutObject(ctx, bucket, object, data, opts) - }, DeleteObjectFn: func(ctx context.Context, bucket, object string) error { return newObjectLayerFn().DeleteObject(ctx, bucket, object) }, @@ -991,142 +360,5 @@ func newServerCacheObjects(config CacheConfig) (CacheObjectLayer, error) { } return errs, nil }, - - ListObjectsFn: func(ctx context.Context, bucket, prefix, marker, delimiter string, maxKeys int) (result ListObjectsInfo, err error) { - return newObjectLayerFn().ListObjects(ctx, bucket, prefix, marker, delimiter, maxKeys) - }, - ListObjectsV2Fn: func(ctx context.Context, bucket, prefix, continuationToken, delimiter string, maxKeys int, fetchOwner bool, startAfter string) (result ListObjectsV2Info, err error) { - return newObjectLayerFn().ListObjectsV2(ctx, bucket, prefix, continuationToken, delimiter, maxKeys, fetchOwner, startAfter) - }, - ListBucketsFn: func(ctx context.Context) (buckets []BucketInfo, err error) { - return newObjectLayerFn().ListBuckets(ctx) - }, - GetBucketInfoFn: func(ctx context.Context, bucket string) (bucketInfo BucketInfo, err error) { - return newObjectLayerFn().GetBucketInfo(ctx, bucket) - }, - NewMultipartUploadFn: func(ctx context.Context, bucket, object string, opts ObjectOptions) (uploadID string, err error) { - return newObjectLayerFn().NewMultipartUpload(ctx, bucket, object, opts) - }, - PutObjectPartFn: func(ctx context.Context, bucket, object, uploadID string, partID int, data *PutObjReader, opts ObjectOptions) (info PartInfo, err error) { - return newObjectLayerFn().PutObjectPart(ctx, bucket, object, uploadID, partID, data, opts) - }, - AbortMultipartUploadFn: func(ctx context.Context, bucket, object, uploadID string) error { - return newObjectLayerFn().AbortMultipartUpload(ctx, bucket, object, uploadID) - }, - CompleteMultipartUploadFn: func(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart, opts ObjectOptions) (objInfo ObjectInfo, err error) { - return newObjectLayerFn().CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts) - }, - DeleteBucketFn: func(ctx context.Context, bucket string) error { - return newObjectLayerFn().DeleteBucket(ctx, bucket) - }, }, nil } - -type cacheControl struct { - exclude bool - expiry time.Time - maxAge int - sMaxAge int - minFresh int -} - -// cache exclude directives in cache-control header -var cacheExcludeDirectives = []string{ - "no-cache", - "no-store", - "must-revalidate", -} - -// returns true if cache exclude directives are set. -func isCacheExcludeDirective(s string) bool { - for _, directive := range cacheExcludeDirectives { - if s == directive { - return true - } - } - return false -} - -// returns struct with cache-control settings from user metadata. -func getCacheControlOpts(m map[string]string) (c cacheControl, err error) { - var headerVal string - for k, v := range m { - if k == "cache-control" { - headerVal = v - } - if k == "expires" { - if e, err := http.ParseTime(v); err == nil { - c.expiry = e - } - } - } - if headerVal == "" { - return - } - headerVal = strings.ToLower(headerVal) - headerVal = strings.TrimSpace(headerVal) - - vals := strings.Split(headerVal, ",") - for _, val := range vals { - val = strings.TrimSpace(val) - p := strings.Split(val, "=") - if isCacheExcludeDirective(p[0]) { - c.exclude = true - continue - } - - if len(p) != 2 { - continue - } - if p[0] == "max-age" || - p[0] == "s-maxage" || - p[0] == "min-fresh" { - i, err := strconv.Atoi(p[1]) - if err != nil { - return c, err - } - if p[0] == "max-age" { - c.maxAge = i - } - if p[0] == "s-maxage" { - c.sMaxAge = i - } - if p[0] == "min-fresh" { - c.minFresh = i - } - } - } - return c, nil -} - -// return true if metadata has a cache-control header -// directive to exclude object from cache. -func filterFromCache(m map[string]string) bool { - c, err := getCacheControlOpts(m) - if err != nil { - return false - } - return c.exclude -} - -// returns true if cache expiry conditions met in cache-control/expiry metadata. -func isStaleCache(objInfo ObjectInfo) bool { - c, err := getCacheControlOpts(objInfo.UserDefined) - if err != nil { - return false - } - now := time.Now() - if c.sMaxAge > 0 && c.sMaxAge > int(now.Sub(objInfo.ModTime).Seconds()) { - return true - } - if c.maxAge > 0 && c.maxAge > int(now.Sub(objInfo.ModTime).Seconds()) { - return true - } - if !c.expiry.Equal(time.Time{}) && c.expiry.Before(time.Now()) { - return true - } - if c.minFresh > 0 && c.minFresh <= int(now.Sub(objInfo.ModTime).Seconds()) { - return true - } - return false -} diff --git a/cmd/disk-cache_test.go b/cmd/disk-cache_test.go index 0313390d8..a6fe71908 100644 --- a/cmd/disk-cache_test.go +++ b/cmd/disk-cache_test.go @@ -1,5 +1,5 @@ /* - * MinIO Cloud Storage, (C) 2018 MinIO, Inc. + * MinIO Cloud Storage, (C) 2018,2019 MinIO, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,6 +19,8 @@ package cmd import ( "bytes" "context" + "io" + "net/http" "reflect" "testing" "time" @@ -26,27 +28,45 @@ import ( "github.com/minio/minio/pkg/hash" ) -// Initialize cache FS objects. -func initCacheFSObjects(disk string, cacheMaxUse int) (*cacheFSObjects, error) { - return newCacheFSObjects(disk, globalCacheExpiry, cacheMaxUse) +// Initialize cache objects. +func initCacheObjects(disk string, cacheMaxUse int) (*diskCache, error) { + return newdiskCache(disk, globalCacheExpiry, cacheMaxUse) } // inits diskCache struct for nDisks -func initDiskCaches(drives []string, cacheMaxUse int, t *testing.T) (*diskCache, error) { - var cfs []*cacheFSObjects +func initDiskCaches(drives []string, cacheMaxUse int, t *testing.T) ([]*diskCache, error) { + var cb []*diskCache for _, d := range drives { - obj, err := initCacheFSObjects(d, cacheMaxUse) + obj, err := initCacheObjects(d, cacheMaxUse) if err != nil { return nil, err } - cfs = append(cfs, obj) + cb = append(cb, obj) + } + return cb, nil +} + +// Tests ToObjectInfo function. +func TestCacheMetadataObjInfo(t *testing.T) { + m := cacheMeta{Meta: nil} + objInfo := m.ToObjectInfo("testbucket", "testobject", nil) + if objInfo.Size != 0 { + t.Fatal("Unexpected object info value for Size", objInfo.Size) + } + if objInfo.ModTime != timeSentinel { + t.Fatal("Unexpected object info value for ModTime ", objInfo.ModTime) + } + if objInfo.IsDir { + t.Fatal("Unexpected object info value for IsDir", objInfo.IsDir) + } + if !objInfo.Expires.IsZero() { + t.Fatal("Unexpected object info value for Expires ", objInfo.Expires) } - return &diskCache{cfs: cfs}, nil } // test whether a drive being offline causes -// getCacheFS to fetch next online drive -func TestGetCacheFS(t *testing.T) { +// getCachedLoc to fetch next online drive +func TestGetCachedLoc(t *testing.T) { for n := 1; n < 10; n++ { fsDirs, err := getRandomDisks(n) if err != nil { @@ -56,14 +76,15 @@ func TestGetCacheFS(t *testing.T) { if err != nil { t.Fatal(err) } + c := cacheObjects{cache: d} bucketName := "testbucket" objectName := "testobject" ctx := context.Background() // find cache drive where object would be hashed - index := d.hashIndex(bucketName, objectName) + index := c.hashIndex(bucketName, objectName) // turn off drive by setting online status to false - d.cfs[index].online = false - cfs, err := d.getCacheFS(ctx, bucketName, objectName) + c.cache[index].online = false + cfs, err := c.getCachedLoc(ctx, bucketName, objectName) if n == 1 && err == errDiskNotFound { continue } @@ -71,7 +92,7 @@ func TestGetCacheFS(t *testing.T) { t.Fatal(err) } i := -1 - for j, f := range d.cfs { + for j, f := range c.cache { if f == cfs { i = j break @@ -84,8 +105,8 @@ func TestGetCacheFS(t *testing.T) { } // test whether a drive being offline causes -// getCacheFS to fetch next online drive -func TestGetCacheFSMaxUse(t *testing.T) { +// getCachedLoc to fetch next online drive +func TestGetCacheMaxUse(t *testing.T) { for n := 1; n < 10; n++ { fsDirs, err := getRandomDisks(n) if err != nil { @@ -95,14 +116,16 @@ func TestGetCacheFSMaxUse(t *testing.T) { if err != nil { t.Fatal(err) } + c := cacheObjects{cache: d} + bucketName := "testbucket" objectName := "testobject" ctx := context.Background() // find cache drive where object would be hashed - index := d.hashIndex(bucketName, objectName) + index := c.hashIndex(bucketName, objectName) // turn off drive by setting online status to false - d.cfs[index].online = false - cfs, err := d.getCacheFS(ctx, bucketName, objectName) + c.cache[index].online = false + cb, err := c.getCachedLoc(ctx, bucketName, objectName) if n == 1 && err == errDiskNotFound { continue } @@ -110,8 +133,8 @@ func TestGetCacheFSMaxUse(t *testing.T) { t.Fatal(err) } i := -1 - for j, f := range d.cfs { - if f == cfs { + for j, f := range d { + if f == cb { i = j break } @@ -172,7 +195,9 @@ func TestDiskCache(t *testing.T) { if err != nil { t.Fatal(err) } - cache := d.cfs[0] + c := cacheObjects{cache: d} + + cache := c.cache[0] ctx := context.Background() bucketName := "testbucket" objectName := "testobject" @@ -198,14 +223,17 @@ func TestDiskCache(t *testing.T) { if err != nil { t.Fatal(err) } - err = cache.Put(ctx, bucketName, objectName, NewPutObjReader(hashReader, nil, nil), ObjectOptions{UserDefined: httpMeta}) + err = cache.Put(ctx, bucketName, objectName, hashReader, hashReader.Size(), ObjectOptions{UserDefined: httpMeta}) if err != nil { t.Fatal(err) } - cachedObjInfo, err := cache.GetObjectInfo(ctx, bucketName, objectName, opts) + cReader, err := cache.Get(ctx, bucketName, objectName, nil, http.Header{ + "Content-Type": []string{"application/json"}, + }, opts) if err != nil { t.Fatal(err) } + cachedObjInfo := cReader.ObjInfo if !cache.Exists(ctx, bucketName, objectName) { t.Fatal("Expected object to exist on cache") } @@ -219,7 +247,7 @@ func TestDiskCache(t *testing.T) { t.Fatal("Cached content-type does not match") } writer := bytes.NewBuffer(nil) - err = cache.Get(ctx, bucketName, objectName, 0, int64(size), writer, "", opts) + _, err = io.Copy(writer, cReader) if err != nil { t.Fatal(err) } @@ -246,7 +274,7 @@ func TestDiskCacheMaxUse(t *testing.T) { if err != nil { t.Fatal(err) } - cache := d.cfs[0] + cache := d[0] ctx := context.Background() bucketName := "testbucket" objectName := "testobject" @@ -274,19 +302,20 @@ func TestDiskCacheMaxUse(t *testing.T) { t.Fatal(err) } if !cache.diskAvailable(int64(size)) { - err = cache.Put(ctx, bucketName, objectName, NewPutObjReader(hashReader, nil, nil), ObjectOptions{UserDefined: httpMeta}) + err = cache.Put(ctx, bucketName, objectName, hashReader, hashReader.Size(), ObjectOptions{UserDefined: httpMeta}) if err != errDiskFull { t.Fatal("Cache max-use limit violated.") } } else { - err = cache.Put(ctx, bucketName, objectName, NewPutObjReader(hashReader, nil, nil), ObjectOptions{UserDefined: httpMeta}) + err = cache.Put(ctx, bucketName, objectName, hashReader, hashReader.Size(), ObjectOptions{UserDefined: httpMeta}) if err != nil { t.Fatal(err) } - cachedObjInfo, err := cache.GetObjectInfo(ctx, bucketName, objectName, opts) + cReader, err := cache.Get(ctx, bucketName, objectName, nil, nil, opts) if err != nil { t.Fatal(err) } + cachedObjInfo := cReader.ObjInfo if !cache.Exists(ctx, bucketName, objectName) { t.Fatal("Expected object to exist on cache") } @@ -300,7 +329,7 @@ func TestDiskCacheMaxUse(t *testing.T) { t.Fatal("Cached content-type does not match") } writer := bytes.NewBuffer(nil) - err = cache.Get(ctx, bucketName, objectName, 0, int64(size), writer, "", opts) + _, err = io.Copy(writer, cReader) if err != nil { t.Fatal(err) } diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go index 9768a2a8c..e8c020f67 100644 --- a/cmd/object-handlers.go +++ b/cmd/object-handlers.go @@ -1249,10 +1249,6 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req // Ensure that metadata does not contain sensitive information crypto.RemoveSensitiveEntries(metadata) - if api.CacheAPI() != nil && !hasServerSideEncryptionHeader(r.Header) { - putObject = api.CacheAPI().PutObject - } - // Create the object.. objInfo, err := putObject(ctx, bucket, object, pReader, opts) if err != nil { @@ -1404,9 +1400,7 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r return } newMultipartUpload := objectAPI.NewMultipartUpload - if api.CacheAPI() != nil && !hasServerSideEncryptionHeader(r.Header) { - newMultipartUpload = api.CacheAPI().NewMultipartUpload - } + uploadID, err := newMultipartUpload(ctx, bucket, object, opts) if err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r)) @@ -1935,9 +1929,7 @@ func (api objectAPIHandlers) PutObjectPartHandler(w http.ResponseWriter, r *http } putObjectPart := objectAPI.PutObjectPart - if api.CacheAPI() != nil && !isEncrypted { - putObjectPart = api.CacheAPI().PutObjectPart - } + partInfo, err := putObjectPart(ctx, bucket, object, uploadID, partID, pReader, opts) if err != nil { // Verify if the underlying error is signature mismatch. @@ -1970,9 +1962,6 @@ func (api objectAPIHandlers) AbortMultipartUploadHandler(w http.ResponseWriter, return } abortMultipartUpload := objectAPI.AbortMultipartUpload - if api.CacheAPI() != nil { - abortMultipartUpload = api.CacheAPI().AbortMultipartUpload - } if s3Error := checkRequestAuthType(ctx, r, policy.AbortMultipartUploadAction, bucket, object); s3Error != ErrNone { writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL, guessIsBrowserReq(r)) @@ -2253,9 +2242,6 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite } completeMultiPartUpload := objectAPI.CompleteMultipartUpload - if api.CacheAPI() != nil { - completeMultiPartUpload = api.CacheAPI().CompleteMultipartUpload - } // This code is specifically to handle the requirements for slow // complete multipart upload operations on FS mode. diff --git a/cmd/web-handlers.go b/cmd/web-handlers.go index 181ec3c16..bdde213f2 100644 --- a/cmd/web-handlers.go +++ b/cmd/web-handlers.go @@ -253,9 +253,6 @@ func (web *webAPIHandlers) DeleteBucket(r *http.Request, args *RemoveBucketArgs, ctx := context.Background() deleteBucket := objectAPI.DeleteBucket - if web.CacheAPI() != nil { - deleteBucket = web.CacheAPI().DeleteBucket - } if err := deleteBucket(ctx, args.BucketName); err != nil { return toJSONError(err, args.BucketName) @@ -297,9 +294,6 @@ func (web *webAPIHandlers) ListBuckets(r *http.Request, args *WebGenericArgs, re return toJSONError(errServerNotInitialized) } listBuckets := objectAPI.ListBuckets - if web.CacheAPI() != nil { - listBuckets = web.CacheAPI().ListBuckets - } claims, owner, authErr := webRequestAuthenticate(r) if authErr != nil { @@ -403,9 +397,6 @@ func (web *webAPIHandlers) ListObjects(r *http.Request, args *ListObjectsArgs, r } listObjects := objectAPI.ListObjects - if web.CacheAPI() != nil { - listObjects = web.CacheAPI().ListObjects - } if isRemoteCallRequired(context.Background(), args.BucketName, objectAPI) { sr, err := globalDNSConfig.Get(args.BucketName) @@ -577,9 +568,6 @@ func (web *webAPIHandlers) RemoveObject(r *http.Request, args *RemoveObjectArgs, return toJSONError(errServerNotInitialized) } listObjects := objectAPI.ListObjects - if web.CacheAPI() != nil { - listObjects = web.CacheAPI().ListObjects - } claims, owner, authErr := webRequestAuthenticate(r) if authErr != nil { @@ -1018,9 +1006,7 @@ func (web *webAPIHandlers) Upload(w http.ResponseWriter, r *http.Request) { } putObject := objectAPI.PutObject - if !hasServerSideEncryptionHeader(r.Header) && web.CacheAPI() != nil { - putObject = web.CacheAPI().PutObject - } + objInfo, err := putObject(context.Background(), bucket, object, pReader, opts) if err != nil { writeWebErrorResponse(w, err) @@ -1260,32 +1246,29 @@ func (web *webAPIHandlers) DownloadZip(w http.ResponseWriter, r *http.Request) { writeWebErrorResponse(w, errInvalidBucketName) return } + getObjectNInfo := objectAPI.GetObjectNInfo + if web.CacheAPI() != nil { + getObjectNInfo = web.CacheAPI().GetObjectNInfo + } - getObject := objectAPI.GetObject - if web.CacheAPI() != nil { - getObject = web.CacheAPI().GetObject - } listObjects := objectAPI.ListObjects - if web.CacheAPI() != nil { - listObjects = web.CacheAPI().ListObjects - } archive := zip.NewWriter(w) defer archive.Close() - getObjectInfo := objectAPI.GetObjectInfo - if web.CacheAPI() != nil { - getObjectInfo = web.CacheAPI().GetObjectInfo - } - opts := ObjectOptions{} var length int64 for _, object := range args.Objects { // Writes compressed object file to the response. zipit := func(objectName string) error { - info, err := getObjectInfo(ctx, args.BucketName, objectName, opts) + var opts ObjectOptions + gr, err := getObjectNInfo(ctx, args.BucketName, objectName, nil, r.Header, readLock, opts) if err != nil { return err } + defer gr.Close() + + info := gr.ObjInfo + length = info.Size if objectAPI.IsEncryptionSupported() { if _, err = DecryptObjectInfo(&info, r.Header); err != nil { @@ -1348,7 +1331,7 @@ func (web *webAPIHandlers) DownloadZip(w http.ResponseWriter, r *http.Request) { // Response writer should be limited early on for decryption upto required length, // additionally also skipping mod(offset)64KiB boundaries. writer = ioutil.LimitedWriter(writer, startOffset%(64*1024), length) - writer, startOffset, length, err = DecryptBlocksRequest(writer, r, + writer, _, length, err = DecryptBlocksRequest(writer, r, args.BucketName, objectName, startOffset, length, info, false) if err != nil { writeWebErrorResponse(w, err) @@ -1356,14 +1339,20 @@ func (web *webAPIHandlers) DownloadZip(w http.ResponseWriter, r *http.Request) { } } httpWriter := ioutil.WriteOnClose(writer) - if err = getObject(ctx, args.BucketName, objectName, startOffset, length, httpWriter, "", opts); err != nil { + + // Write object content to response body + if _, err = io.Copy(httpWriter, gr); err != nil { httpWriter.Close() if info.IsCompressed() { // Wait for decompression go-routine to retire. wg.Wait() } + if !httpWriter.HasWritten() { // write error response only if no data or headers has been written to client yet + writeWebErrorResponse(w, err) + } return err } + if err = httpWriter.Close(); err != nil { if !httpWriter.HasWritten() { // write error response only if no data has been written to client yet writeWebErrorResponse(w, err) diff --git a/docs/disk-caching/DESIGN.md b/docs/disk-caching/DESIGN.md index 27a9fa39d..6d90c8b47 100644 --- a/docs/disk-caching/DESIGN.md +++ b/docs/disk-caching/DESIGN.md @@ -34,13 +34,12 @@ minio server -h - An object is only cached when drive has sufficient disk space. ## Behavior -Disk caching caches objects for both **uploaded** and **downloaded** objects i.e +Disk caching caches objects for **downloaded** objects i.e - Caches new objects for entries not found in cache while downloading. Otherwise serves from the cache. -- Caches all successfully uploaded objects. Replaces existing cached entry of the same object if needed. - When an object is deleted, corresponding entry in cache if any is deleted as well. - Cache continues to work for read-only operations such as GET, HEAD when backend is offline. -- Cache disallows write operations when backend is offline. +- Cache-Control and Expires headers can be used to control how long objects stay in the cache > NOTE: Expiration happens automatically based on the configured interval as explained above, frequently accessed objects stay alive in cache for a significantly longer time. @@ -49,5 +48,6 @@ Upon restart of minio server after a running minio process is killed or crashes, ## Limits - Bucket policies are not cached, so anonymous operations are not supported when backend is offline. +- To ensure security guarantees, encrypted objects are never cached. - Objects are distributed using deterministic hashing among the list of configured cache drives. If one or more drives go offline, or cache drive configuration is altered in any way, performance may degrade to a linear lookup time depending on the number of disks in cache.