Fix backend format for disk-cache - not to use FS format.json (#5732)
This commit is contained in:
Krishna Srinivas
Dee Koder
parent
328076f773
commit
804a4f9c15
|
|
@ -20,6 +20,7 @@ import (
|
|||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
|
|
@ -125,21 +126,22 @@ func handleCommonEnvVars() {
|
|||
}
|
||||
|
||||
if drives := os.Getenv("MINIO_CACHE_DRIVES"); drives != "" {
|
||||
driveList, err := parseCacheDrives(drives)
|
||||
fatalIf(err, "Invalid value set in environment variable MINIO_CACHE_DRIVES")
|
||||
driveList, err := parseCacheDrives(strings.Split(drives, cacheEnvDelimiter))
|
||||
fatalIf(err, "Invalid value set in environment variable MINIO_CACHE_DRIVES %s.", drives)
|
||||
globalCacheDrives = driveList
|
||||
globalIsDiskCacheEnabled = true
|
||||
}
|
||||
if excludes := os.Getenv("MINIO_CACHE_EXCLUDE"); excludes != "" {
|
||||
excludeList, err := parseCacheExcludes(excludes)
|
||||
fatalIf(err, "Invalid value set in environment variable MINIO_CACHE_EXCLUDE")
|
||||
excludeList, err := parseCacheExcludes(strings.Split(excludes, cacheEnvDelimiter))
|
||||
fatalIf(err, "Invalid value set in environment variable MINIO_CACHE_EXCLUDE %s.", excludes)
|
||||
globalCacheExcludes = excludeList
|
||||
}
|
||||
if expiryStr := os.Getenv("MINIO_CACHE_EXPIRY"); expiryStr != "" {
|
||||
expiry, err := parseCacheExpiry(expiryStr)
|
||||
fatalIf(err, "Invalid value set in environment variable MINIO_CACHE_EXPIRY")
|
||||
expiry, err := strconv.Atoi(expiryStr)
|
||||
fatalIf(err, "Invalid value set in environment variable MINIO_CACHE_EXPIRY %s.", expiryStr)
|
||||
globalCacheExpiry = expiry
|
||||
}
|
||||
|
||||
// In place update is true by default if the MINIO_UPDATE is not set
|
||||
// or is not set to 'off', if MINIO_UPDATE is set to 'off' then
|
||||
// in-place update is off.
|
||||
|
|
|
|||
|
|
@ -113,14 +113,7 @@ func (s *serverConfig) SetCacheConfig(drives, exclude []string, expiry int) {
|
|||
|
||||
// GetCacheConfig gets the current cache config
|
||||
func (s *serverConfig) GetCacheConfig() CacheConfig {
|
||||
if s.Cache.Drives != nil {
|
||||
return CacheConfig{
|
||||
Drives: s.Cache.Drives,
|
||||
Exclude: s.Cache.Exclude,
|
||||
Expiry: s.Cache.Expiry,
|
||||
}
|
||||
}
|
||||
return CacheConfig{}
|
||||
return s.Cache
|
||||
}
|
||||
|
||||
// Save config.
|
||||
|
|
@ -247,6 +240,7 @@ func newConfig() error {
|
|||
if globalIsDiskCacheEnabled {
|
||||
srvCfg.SetCacheConfig(globalCacheDrives, globalCacheExcludes, globalCacheExpiry)
|
||||
}
|
||||
|
||||
// hold the mutex lock before a new config is assigned.
|
||||
// Save the new config globally.
|
||||
// unlock the mutex.
|
||||
|
|
@ -377,6 +371,7 @@ func loadConfig() error {
|
|||
if globalIsDiskCacheEnabled {
|
||||
srvCfg.SetCacheConfig(globalCacheDrives, globalCacheExcludes, globalCacheExpiry)
|
||||
}
|
||||
|
||||
// hold the mutex lock before a new config is assigned.
|
||||
globalServerConfigMu.Lock()
|
||||
globalServerConfig = srvCfg
|
||||
|
|
|
|||
|
|
@ -1752,7 +1752,7 @@ func migrateV21ToV22() error {
|
|||
srvConfig := &serverConfigV22{
|
||||
Notify: notifier{},
|
||||
}
|
||||
srvConfig.Version = serverConfigVersion
|
||||
srvConfig.Version = "22"
|
||||
srvConfig.Credential = cv21.Credential
|
||||
srvConfig.Region = cv21.Region
|
||||
if srvConfig.Region == "" {
|
||||
|
|
@ -1856,7 +1856,7 @@ func migrateV22ToV23() error {
|
|||
srvConfig := &serverConfigV23{
|
||||
Notify: notifier{},
|
||||
}
|
||||
srvConfig.Version = serverConfigVersion
|
||||
srvConfig.Version = "23"
|
||||
srvConfig.Credential = cv22.Credential
|
||||
srvConfig.Region = cv22.Region
|
||||
if srvConfig.Region == "" {
|
||||
|
|
|
|||
|
|
@ -17,48 +17,58 @@
|
|||
package cmd
|
||||
|
||||
import (
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"errors"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
)
|
||||
|
||||
// CacheConfig represents cache config settings
|
||||
type CacheConfig struct {
|
||||
Drives []string
|
||||
Expiry int
|
||||
Exclude []string
|
||||
Drives []string `json:"drives"`
|
||||
Expiry int `json:"expiry"`
|
||||
Exclude []string `json:"exclude"`
|
||||
}
|
||||
|
||||
// UnmarshalJSON - implements JSON unmarshal interface for unmarshalling
|
||||
// json entries for CacheConfig.
|
||||
func (cfg *CacheConfig) UnmarshalJSON(data []byte) (err error) {
|
||||
type Alias CacheConfig
|
||||
var _cfg = &struct {
|
||||
*Alias
|
||||
}{
|
||||
Alias: (*Alias)(cfg),
|
||||
}
|
||||
if err = json.Unmarshal(data, _cfg); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = parseCacheDrives(_cfg.Drives); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = parseCacheExcludes(_cfg.Exclude); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Parses given cacheDrivesEnv and returns a list of cache drives.
|
||||
func parseCacheDrives(cacheDrivesEnv string) ([]string, error) {
|
||||
cacheDrivesEnv = strings.ToLower(cacheDrivesEnv)
|
||||
s := strings.Split(cacheDrivesEnv, ";")
|
||||
c2 := make([]string, 0)
|
||||
for _, d := range s {
|
||||
if len(d) > 0 {
|
||||
c2 = append(c2, d)
|
||||
func parseCacheDrives(drives []string) ([]string, error) {
|
||||
for _, d := range drives {
|
||||
if !filepath.IsAbs(d) {
|
||||
return nil, fmt.Errorf("cache dir should be absolute path: %s", d)
|
||||
}
|
||||
}
|
||||
return c2, nil
|
||||
return drives, nil
|
||||
}
|
||||
|
||||
// Parses given cacheExcludesEnv and returns a list of cache exclude patterns.
|
||||
func parseCacheExcludes(cacheExcludesEnv string) ([]string, error) {
|
||||
s := strings.Split(cacheExcludesEnv, ";")
|
||||
c2 := make([]string, 0)
|
||||
for _, e := range s {
|
||||
if len(e) > 0 {
|
||||
if strings.HasPrefix(e, "/") {
|
||||
return c2, errors.New("cache exclude patterns cannot start with / as prefix " + e)
|
||||
}
|
||||
c2 = append(c2, e)
|
||||
func parseCacheExcludes(excludes []string) ([]string, error) {
|
||||
for _, e := range excludes {
|
||||
if len(e) == 0 {
|
||||
return nil, fmt.Errorf("cache exclude path (%s) cannot be empty", e)
|
||||
}
|
||||
if hasPrefix(e, slashSeparator) {
|
||||
return nil, fmt.Errorf("cache exclude pattern (%s) cannot start with / as prefix", e)
|
||||
}
|
||||
}
|
||||
return c2, nil
|
||||
}
|
||||
|
||||
// Parses given cacheExpiryEnv and returns cache expiry in days.
|
||||
func parseCacheExpiry(cacheExpiryEnv string) (int, error) {
|
||||
return strconv.Atoi(cacheExpiryEnv)
|
||||
return excludes, nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,9 +18,52 @@ package cmd
|
|||
|
||||
import (
|
||||
"reflect"
|
||||
"runtime"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// Tests cache drive parsing.
|
||||
func TestParseCacheDrives(t *testing.T) {
|
||||
testCases := []struct {
|
||||
driveStr string
|
||||
expectedPatterns []string
|
||||
success bool
|
||||
}{
|
||||
// valid input
|
||||
|
||||
{"bucket1/*;*.png;images/trip/barcelona/*", []string{}, false},
|
||||
{"bucket1", []string{}, false},
|
||||
}
|
||||
if runtime.GOOS == globalWindowsOSName {
|
||||
testCases = append(testCases, struct {
|
||||
driveStr string
|
||||
expectedPatterns []string
|
||||
success bool
|
||||
}{"C:/home/drive1;C:/home/drive2;C:/home/drive3", []string{"C:/home/drive1", "C:/home/drive2", "C:/home/drive3"}, true})
|
||||
} else {
|
||||
testCases = append(testCases, struct {
|
||||
driveStr string
|
||||
expectedPatterns []string
|
||||
success bool
|
||||
}{"/home/drive1;/home/drive2;/home/drive3", []string{"/home/drive1", "/home/drive2", "/home/drive3"}, true})
|
||||
}
|
||||
for i, testCase := range testCases {
|
||||
drives, err := parseCacheDrives(strings.Split(testCase.driveStr, cacheEnvDelimiter))
|
||||
if err != nil && testCase.success {
|
||||
t.Errorf("Test %d: Expected success but failed instead %s", i+1, err)
|
||||
}
|
||||
if err == nil && !testCase.success {
|
||||
t.Errorf("Test %d: Expected failure but passed instead", i+1)
|
||||
}
|
||||
if err == nil {
|
||||
if !reflect.DeepEqual(drives, testCase.expectedPatterns) {
|
||||
t.Errorf("Test %d: Expected %v, got %v", i+1, testCase.expectedPatterns, drives)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Tests cache exclude parsing.
|
||||
func TestParseCacheExclude(t *testing.T) {
|
||||
testCases := []struct {
|
||||
|
|
@ -28,8 +71,6 @@ func TestParseCacheExclude(t *testing.T) {
|
|||
expectedPatterns []string
|
||||
success bool
|
||||
}{
|
||||
// Empty input.
|
||||
{"", []string{}, true},
|
||||
// valid input
|
||||
{"/home/drive1;/home/drive2;/home/drive3", []string{}, false},
|
||||
{"bucket1/*;*.png;images/trip/barcelona/*", []string{"bucket1/*", "*.png", "images/trip/barcelona/*"}, true},
|
||||
|
|
@ -37,15 +78,17 @@ func TestParseCacheExclude(t *testing.T) {
|
|||
}
|
||||
|
||||
for i, testCase := range testCases {
|
||||
excludes, err := parseCacheExcludes(testCase.excludeStr)
|
||||
excludes, err := parseCacheExcludes(strings.Split(testCase.excludeStr, cacheEnvDelimiter))
|
||||
if err != nil && testCase.success {
|
||||
t.Errorf("Test %d: Expected success but failed instead %s", i+1, err)
|
||||
}
|
||||
if err == nil && !testCase.success {
|
||||
t.Errorf("Test %d: Expected failure but passed instead", i+1)
|
||||
}
|
||||
if !reflect.DeepEqual(excludes, testCase.expectedPatterns) {
|
||||
t.Errorf("Expected %v, got %v", testCase.expectedPatterns, excludes)
|
||||
if err == nil {
|
||||
if !reflect.DeepEqual(excludes, testCase.expectedPatterns) {
|
||||
t.Errorf("Test %d: Expected %v, got %v", i+1, testCase.expectedPatterns, excludes)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ import (
|
|||
"context"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"os"
|
||||
|
|
@ -37,6 +38,8 @@ const (
|
|||
// cache.json object metadata for cached objects.
|
||||
cacheMetaJSONFile = "cache.json"
|
||||
cacheMetaFormat = "cache"
|
||||
|
||||
cacheEnvDelimiter = ";"
|
||||
)
|
||||
|
||||
// cacheFSObjects implements the cache backend operations.
|
||||
|
|
@ -59,9 +62,13 @@ type cacheFSObjects struct {
|
|||
// 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) {
|
||||
obj, err := newFSObjects(dir, cacheMetaJSONFile)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
// 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)
|
||||
|
|
@ -72,9 +79,23 @@ func newCacheFSObjects(dir string, expiry int, maxDiskUsagePct int) (*cacheFSObj
|
|||
if expiry == 0 {
|
||||
expiry = globalCacheExpiry
|
||||
}
|
||||
var cacheFS cacheFSObjects
|
||||
fsObjects := obj.(*FSObjects)
|
||||
cacheFS = cacheFSObjects{
|
||||
|
||||
// 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(globalMultipartCleanupInterval, globalMultipartExpiry, globalServiceDoneCh)
|
||||
|
||||
cacheFS := cacheFSObjects{
|
||||
FSObjects: fsObjects,
|
||||
dir: dir,
|
||||
expiry: expiry,
|
||||
|
|
|
|||
|
|
@ -18,15 +18,12 @@ package cmd
|
|||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"errors"
|
||||
"fmt"
|
||||
"hash/crc32"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"os"
|
||||
"path"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
|
@ -115,7 +112,6 @@ func backendDownError(err error) bool {
|
|||
// 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
|
||||
|
|
@ -145,7 +141,6 @@ func (c diskCache) getCachedFSLoc(ctx context.Context, bucket, object string) (*
|
|||
// 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++ {
|
||||
|
|
@ -162,8 +157,7 @@ func (c diskCache) getCacheFS(ctx context.Context, bucket, object string) (*cach
|
|||
|
||||
// Compute a unique hash sum for bucket and object
|
||||
func (c diskCache) hashIndex(bucket, object string) int {
|
||||
key := fmt.Sprintf("%x", sha256.Sum256([]byte(path.Join(bucket, object))))
|
||||
return int(crc32.Checksum([]byte(key), crc32.IEEETable)) % len(c.cfs)
|
||||
return crcHashMod(pathJoin(bucket, object), len(c.cfs))
|
||||
}
|
||||
|
||||
// construct a metadata k-v map
|
||||
|
|
@ -496,7 +490,6 @@ func (c cacheObjects) listBuckets(ctx context.Context) (buckets []BucketInfo, er
|
|||
// 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 {
|
||||
|
|
@ -510,7 +503,6 @@ func (c cacheObjects) ListBuckets(ctx context.Context) (buckets []BucketInfo, er
|
|||
|
||||
// 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) {
|
||||
|
|
@ -525,7 +517,6 @@ func (c cacheObjects) GetBucketInfo(ctx context.Context, bucket string) (bucketI
|
|||
|
||||
// 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
|
||||
}
|
||||
|
|
@ -619,7 +610,6 @@ func (c cacheObjects) PutObject(ctx context.Context, bucket, object string, r *h
|
|||
|
||||
// NewMultipartUpload - Starts a new multipart upload operation to backend and cache.
|
||||
func (c cacheObjects) NewMultipartUpload(ctx context.Context, bucket, object string, metadata map[string]string) (uploadID string, err error) {
|
||||
|
||||
newMultipartUploadFn := c.NewMultipartUploadFn
|
||||
|
||||
if c.isCacheExclude(bucket, object) || filterFromCache(metadata) {
|
||||
|
|
@ -643,7 +633,6 @@ func (c cacheObjects) NewMultipartUpload(ctx context.Context, bucket, object str
|
|||
|
||||
// PutObjectPart - uploads part to backend and cache simultaneously.
|
||||
func (c cacheObjects) PutObjectPart(ctx context.Context, bucket, object, uploadID string, partID int, data *hash.Reader) (info PartInfo, err error) {
|
||||
|
||||
putObjectPartFn := c.PutObjectPartFn
|
||||
dcache, err := c.cache.getCacheFS(ctx, bucket, object)
|
||||
if err != nil {
|
||||
|
|
@ -713,7 +702,6 @@ func (c cacheObjects) PutObjectPart(ctx context.Context, bucket, object, uploadI
|
|||
|
||||
// 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) {
|
||||
|
|
@ -737,7 +725,6 @@ func (c cacheObjects) AbortMultipartUpload(ctx context.Context, bucket, object,
|
|||
|
||||
// CompleteMultipartUpload - completes multipart upload operation on backend and cache.
|
||||
func (c cacheObjects) CompleteMultipartUpload(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart) (objInfo ObjectInfo, err error) {
|
||||
|
||||
completeMultipartUploadFn := c.CompleteMultipartUploadFn
|
||||
|
||||
if c.isCacheExclude(bucket, object) {
|
||||
|
|
@ -802,33 +789,32 @@ func (c cacheObjects) DeleteBucket(ctx context.Context, bucket string) (err erro
|
|||
|
||||
// newCache initializes the cacheFSObjects for the "drives" specified in config.json
|
||||
// or the global env overrides.
|
||||
func newCache(c CacheConfig) (*diskCache, error) {
|
||||
func newCache(config CacheConfig) (*diskCache, error) {
|
||||
var cfsObjects []*cacheFSObjects
|
||||
formats, err := loadAndValidateCacheFormat(c.Drives)
|
||||
formats, err := loadAndValidateCacheFormat(config.Drives)
|
||||
if err != nil {
|
||||
errorIf(err, "Cache drives validation error")
|
||||
return nil, err
|
||||
}
|
||||
if len(formats) == 0 {
|
||||
return nil, errors.New("Cache drives validation error")
|
||||
}
|
||||
for i, dir := range c.Drives {
|
||||
for i, dir := range config.Drives {
|
||||
// skip cacheFSObjects creation for cache drives missing a format.json
|
||||
if formats[i] == nil {
|
||||
cfsObjects = append(cfsObjects, nil)
|
||||
continue
|
||||
}
|
||||
c, err := newCacheFSObjects(dir, c.Expiry, cacheMaxDiskUsagePct)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := checkAtimeSupport(dir); err != nil {
|
||||
return nil, errors.New("Atime support required for disk caching")
|
||||
}
|
||||
cache, err := newCacheFSObjects(dir, config.Expiry, cacheMaxDiskUsagePct)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Start the purging go-routine for entries that have expired
|
||||
go c.purge()
|
||||
go cache.purge()
|
||||
|
||||
// Start trash purge routine for deleted buckets.
|
||||
go c.purgeTrash()
|
||||
cfsObjects = append(cfsObjects, c)
|
||||
go cache.purgeTrash()
|
||||
|
||||
cfsObjects = append(cfsObjects, cache)
|
||||
}
|
||||
return &diskCache{cfs: cfsObjects}, nil
|
||||
}
|
||||
|
|
@ -857,17 +843,16 @@ func checkAtimeSupport(dir string) (err error) {
|
|||
}
|
||||
|
||||
// Returns cacheObjects for use by Server.
|
||||
func newServerCacheObjects(c CacheConfig) (CacheObjectLayer, error) {
|
||||
|
||||
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(c)
|
||||
dcache, err := newCache(config)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &cacheObjects{
|
||||
cache: dcache,
|
||||
exclude: c.Exclude,
|
||||
exclude: config.Exclude,
|
||||
listPool: newTreeWalkPool(globalLookupTimeout),
|
||||
GetObjectFn: func(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string) error {
|
||||
return newObjectLayerFn().GetObject(ctx, bucket, object, startOffset, length, writer, etag)
|
||||
|
|
|
|||
|
|
@ -21,9 +21,7 @@ import (
|
|||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"syscall"
|
||||
|
||||
errors2 "github.com/minio/minio/pkg/errors"
|
||||
)
|
||||
|
|
@ -36,6 +34,8 @@ const (
|
|||
formatCacheVersionV1 = "1"
|
||||
|
||||
formatMetaVersion1 = "1"
|
||||
|
||||
formatCacheV1DistributionAlgo = "CRCMOD"
|
||||
)
|
||||
|
||||
// Represents the current cache structure with list of
|
||||
|
|
@ -49,6 +49,9 @@ type formatCacheV1 struct {
|
|||
// Disks field carries the input disk order generated the first
|
||||
// time when fresh disks were supplied.
|
||||
Disks []string `json:"disks"`
|
||||
// Distribution algorithm represents the hashing algorithm
|
||||
// to pick the right set index for an object.
|
||||
DistributionAlgo string `json:"distributionAlgo"`
|
||||
} `json:"cache"` // Cache field holds cache format.
|
||||
}
|
||||
|
||||
|
|
@ -71,6 +74,7 @@ func newFormatCacheV1(drives []string) []*formatCacheV1 {
|
|||
format.Version = formatMetaVersion1
|
||||
format.Format = formatCache
|
||||
format.Cache.Version = formatCacheVersionV1
|
||||
format.Cache.DistributionAlgo = formatCacheV1DistributionAlgo
|
||||
format.Cache.This = mustGetUUID()
|
||||
formats[i] = format
|
||||
disks[i] = formats[i].Cache.This
|
||||
|
|
@ -116,28 +120,25 @@ func createFormatCache(fsFormatPath string, format *formatCacheV1) error {
|
|||
// of format cache config
|
||||
func initFormatCache(drives []string) (formats []*formatCacheV1, err error) {
|
||||
nformats := newFormatCacheV1(drives)
|
||||
for i, drive := range drives {
|
||||
// Disallow relative paths, figure out absolute paths.
|
||||
cfsPath, err := filepath.Abs(drive)
|
||||
if err != nil {
|
||||
for _, drive := range drives {
|
||||
_, err = os.Stat(drive)
|
||||
if err == nil {
|
||||
continue
|
||||
}
|
||||
if !os.IsNotExist(err) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
fi, err := os.Stat(cfsPath)
|
||||
if err == nil {
|
||||
if !fi.IsDir() {
|
||||
return nil, syscall.ENOTDIR
|
||||
}
|
||||
if err = os.Mkdir(drive, 0777); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if os.IsNotExist(err) {
|
||||
// Disk not found create it.
|
||||
err = os.MkdirAll(cfsPath, 0777)
|
||||
if err != nil {
|
||||
}
|
||||
for i, drive := range drives {
|
||||
if err = os.Mkdir(pathJoin(drive, minioMetaBucket), 0777); err != nil {
|
||||
if !os.IsExist(err) {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
cacheFormatPath := pathJoin(drive, formatConfigFile)
|
||||
cacheFormatPath := pathJoin(drive, minioMetaBucket, formatConfigFile)
|
||||
// Fresh disk - create format.json for this cfs
|
||||
if err = createFormatCache(cacheFormatPath, nformats[i]); err != nil {
|
||||
return nil, err
|
||||
|
|
@ -146,32 +147,25 @@ func initFormatCache(drives []string) (formats []*formatCacheV1, err error) {
|
|||
return nformats, nil
|
||||
}
|
||||
|
||||
func loadFormatCache(drives []string) (formats []*formatCacheV1, err error) {
|
||||
var errs []error
|
||||
for _, drive := range drives {
|
||||
cacheFormatPath := pathJoin(drive, formatConfigFile)
|
||||
f, perr := os.Open(cacheFormatPath)
|
||||
if perr != nil {
|
||||
formats = append(formats, nil)
|
||||
errs = append(errs, perr)
|
||||
continue
|
||||
func loadFormatCache(drives []string) ([]*formatCacheV1, error) {
|
||||
formats := make([]*formatCacheV1, len(drives))
|
||||
for i, drive := range drives {
|
||||
cacheFormatPath := pathJoin(drive, minioMetaBucket, formatConfigFile)
|
||||
f, err := os.Open(cacheFormatPath)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
continue
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
defer f.Close()
|
||||
format, perr := formatMetaCacheV1(f)
|
||||
if perr != nil {
|
||||
// format could not be unmarshalled.
|
||||
formats = append(formats, nil)
|
||||
errs = append(errs, perr)
|
||||
format, err := formatMetaCacheV1(f)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
formats = append(formats, format)
|
||||
formats[i] = format
|
||||
}
|
||||
for _, perr := range errs {
|
||||
if perr != nil {
|
||||
err = perr
|
||||
}
|
||||
}
|
||||
return formats, err
|
||||
return formats, nil
|
||||
}
|
||||
|
||||
// unmarshalls the cache format.json into formatCacheV1
|
||||
|
|
@ -198,7 +192,6 @@ func checkFormatCacheValue(format *formatCacheV1) error {
|
|||
}
|
||||
|
||||
func checkFormatCacheValues(formats []*formatCacheV1) (int, error) {
|
||||
|
||||
for i, formatCache := range formats {
|
||||
if formatCache == nil {
|
||||
continue
|
||||
|
|
@ -276,6 +269,15 @@ func findCacheDiskIndex(disk string, disks []string) int {
|
|||
|
||||
// validate whether cache drives order has changed
|
||||
func validateCacheFormats(formats []*formatCacheV1) error {
|
||||
count := 0
|
||||
for _, format := range formats {
|
||||
if format == nil {
|
||||
count++
|
||||
}
|
||||
}
|
||||
if count == len(formats) {
|
||||
return errors.New("Cache format files missing on all drives")
|
||||
}
|
||||
if _, err := checkFormatCacheValues(formats); err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -290,23 +292,9 @@ func validateCacheFormats(formats []*formatCacheV1) error {
|
|||
func cacheDrivesUnformatted(drives []string) bool {
|
||||
count := 0
|
||||
for _, drive := range drives {
|
||||
cacheFormatPath := pathJoin(drive, formatConfigFile)
|
||||
|
||||
// // Disallow relative paths, figure out absolute paths.
|
||||
cfsPath, err := filepath.Abs(cacheFormatPath)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
fi, err := os.Stat(cfsPath)
|
||||
if err == nil {
|
||||
if !fi.IsDir() {
|
||||
continue
|
||||
}
|
||||
}
|
||||
if os.IsNotExist(err) {
|
||||
cacheFormatPath := pathJoin(drive, minioMetaBucket, formatConfigFile)
|
||||
if _, err := os.Stat(cacheFormatPath); os.IsNotExist(err) {
|
||||
count++
|
||||
continue
|
||||
}
|
||||
}
|
||||
return count == len(drives)
|
||||
|
|
@ -322,7 +310,10 @@ func loadAndValidateCacheFormat(drives []string) (formats []*formatCacheV1, err
|
|||
formats, err = loadFormatCache(drives)
|
||||
}
|
||||
if err != nil {
|
||||
return formats, err
|
||||
return nil, err
|
||||
}
|
||||
return formats, validateCacheFormats(formats)
|
||||
if err = validateCacheFormats(formats); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return formats, nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -27,17 +27,13 @@ func TestDiskCacheFormat(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err = initDiskCaches(fsDirs, t)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// cformat := newFormatCacheV1([]string{cacheDataDir + "/format.json"})
|
||||
|
||||
_, err = initFormatCache(fsDirs)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Do the basic sanity checks to check if initFormatCache() did its job.
|
||||
cacheFormatPath := pathJoin(fsDirs[0], formatConfigFile)
|
||||
cacheFormatPath := pathJoin(fsDirs[0], minioMetaBucket, formatConfigFile)
|
||||
f, err := os.OpenFile(cacheFormatPath, os.O_RDWR, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
|
|||
11
cmd/fs-v1.go
11
cmd/fs-v1.go
|
|
@ -95,8 +95,8 @@ func initMetaVolumeFS(fsPath, fsUUID string) error {
|
|||
|
||||
}
|
||||
|
||||
// newFSObjects - initialize new fs object layer.
|
||||
func newFSObjects(fsPath, metaJSONFile string) (ObjectLayer, error) {
|
||||
// NewFSObjectLayer - initialize new fs object layer.
|
||||
func NewFSObjectLayer(fsPath string) (ObjectLayer, error) {
|
||||
if fsPath == "" {
|
||||
return nil, errInvalidArgument
|
||||
}
|
||||
|
|
@ -150,7 +150,7 @@ func newFSObjects(fsPath, metaJSONFile string) (ObjectLayer, error) {
|
|||
// Initialize fs objects.
|
||||
fs := &FSObjects{
|
||||
fsPath: fsPath,
|
||||
metaJSONFile: metaJSONFile,
|
||||
metaJSONFile: fsMetaJSONFile,
|
||||
fsUUID: fsUUID,
|
||||
rwPool: &fsIOPool{
|
||||
readersMap: make(map[string]*lock.RLockedFile),
|
||||
|
|
@ -183,11 +183,6 @@ func newFSObjects(fsPath, metaJSONFile string) (ObjectLayer, error) {
|
|||
return fs, nil
|
||||
}
|
||||
|
||||
// NewFSObjectLayer - initialize new fs object layer.
|
||||
func NewFSObjectLayer(fsPath string) (ObjectLayer, error) {
|
||||
return newFSObjects(fsPath, fsMetaJSONFile)
|
||||
}
|
||||
|
||||
// Shutdown - should be called when process shuts down.
|
||||
func (fs *FSObjects) Shutdown(ctx context.Context) error {
|
||||
fs.fsFormatRlk.Close()
|
||||
|
|
|
|||
|
|
@ -1282,9 +1282,9 @@ func TestPosixAppendFile(t *testing.T) {
|
|||
expectedErr error
|
||||
}{"level0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001/level0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000002/level0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000003/object000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001", err})
|
||||
|
||||
for _, testCase := range testCases {
|
||||
for i, testCase := range testCases {
|
||||
if err = posixStorage.AppendFile("success-vol", testCase.fileName, []byte("hello, world")); err != testCase.expectedErr {
|
||||
t.Errorf("Case: %s, expected: %s, got: %s", testCase, testCase.expectedErr, err)
|
||||
t.Errorf("Case: %d, expected: %s, got: %s", i+1, testCase.expectedErr, err)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1373,9 +1373,9 @@ func TestPosixPrepareFile(t *testing.T) {
|
|||
expectedErr error
|
||||
}{"level0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001/level0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000002/level0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000003/object000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001", err})
|
||||
|
||||
for _, testCase := range testCases {
|
||||
for i, testCase := range testCases {
|
||||
if err = posixStorage.PrepareFile("success-vol", testCase.fileName, 16); err != testCase.expectedErr {
|
||||
t.Errorf("Case: %s, expected: %s, got: %s", testCase, testCase.expectedErr, err)
|
||||
t.Errorf("Case: %d, expected: %s, got: %s", i, testCase.expectedErr, err)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -244,7 +244,7 @@ func getUserAgent(mode string) string {
|
|||
if mode != "" {
|
||||
uaAppend("; ", mode)
|
||||
}
|
||||
if globalIsDiskCacheEnabled {
|
||||
if len(globalCacheDrives) > 0 {
|
||||
uaAppend("; ", "feature-cache")
|
||||
}
|
||||
if globalWORMEnabled {
|
||||
|
|
|
|||
Loading…
Reference in a new issue