// Copyright (c) 2015-2021 MinIO, Inc.//// This file is part of MinIO Object Storage stack//// This program is free software: you can redistribute it and/or modify// it under the terms of the GNU Affero General Public License as published by// the Free Software Foundation, either version 3 of the License, or// (at your option) any later version.//// This program is distributed in the hope that it will be useful// but WITHOUT ANY WARRANTY; without even the implied warranty of// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the// GNU Affero General Public License for more details.//// You should have received a copy of the GNU Affero General Public License// along with this program.  If not, see <http://www.gnu.org/licenses/>. package cmd import (	"bytes"	"context"	"errors"	"fmt"	"io"	"math/rand"	"net/http"	"sort"	"strconv"	"strings"	"sync"	"time" 	"github.com/minio/madmin-go"	"github.com/minio/minio-go/v7/pkg/set"	"github.com/minio/minio-go/v7/pkg/tags"	"github.com/minio/minio/internal/bucket/lifecycle"	"github.com/minio/minio/internal/config/storageclass"	"github.com/minio/minio/internal/logger"	"github.com/minio/minio/internal/sync/errgroup"	"github.com/minio/pkg/wildcard") type erasureServerPools struct {	GatewayUnsupported 	poolMetaMutex sync.RWMutex	poolMeta      poolMeta	serverPools   []*erasureSets 	// Shut down async operations	shutdown context.CancelFunc 	// Active decommission canceler	decommissionCancelers []context.CancelFunc} func (z *erasureServerPools) SinglePool() bool {	return len(z.serverPools) == 1} // Initialize new pool of erasure sets.func newErasureServerPools(ctx context.Context, endpointServerPools EndpointServerPools) (ObjectLayer, error) {	if endpointServerPools.NEndpoints() == 1 {		ep := endpointServerPools[0]		storageDisks, format, err := waitForFormatErasure(true, ep.Endpoints, 1, ep.SetCount, ep.DrivesPerSet, "", "")		if err != nil {			return nil, err		} 		objLayer, err := newErasureSingle(ctx, storageDisks[0], format)		if err != nil {			return nil, err		} 		globalLocalDrives = storageDisks		return objLayer, nil	} 	var (		deploymentID       string		distributionAlgo   string		commonParityDrives int		err                error 		formats      = make([]*formatErasureV3, len(endpointServerPools))		storageDisks = make([][]StorageAPI, len(endpointServerPools))		z            = &erasureServerPools{			serverPools: make([]*erasureSets, len(endpointServerPools)),		}	) 	var localDrives []StorageAPI	local := endpointServerPools.FirstLocal()	for i, ep := range endpointServerPools {		// If storage class is not set during startup, default values are used		// -- Default for Reduced Redundancy Storage class is, parity = 2		// -- Default for Standard Storage class is, parity = 2 - disks 4, 5		// -- Default for Standard Storage class is, parity = 3 - disks 6, 7		// -- Default for Standard Storage class is, parity = 4 - disks 8 to 16		if commonParityDrives == 0 {			commonParityDrives = ecDrivesNoConfig(ep.DrivesPerSet)		} 		if err = storageclass.ValidateParity(commonParityDrives, ep.DrivesPerSet); err != nil {			return nil, fmt.Errorf("All current serverPools should have same parity ratio - expected %d, got %d", commonParityDrives, ecDrivesNoConfig(ep.DrivesPerSet))		} 		storageDisks[i], formats[i], err = waitForFormatErasure(local, ep.Endpoints, i+1,			ep.SetCount, ep.DrivesPerSet, deploymentID, distributionAlgo)		if err != nil {			return nil, err		} 		for _, storageDisk := range storageDisks[i] {			if storageDisk != nil && storageDisk.IsLocal() {				localDrives = append(localDrives, storageDisk)			}		} 		if deploymentID == "" {			// all pools should have same deployment ID			deploymentID = formats[i].ID		} 		if distributionAlgo == "" {			distributionAlgo = formats[i].Erasure.DistributionAlgo		} 		// Validate if users brought different DeploymentID pools.		if deploymentID != formats[i].ID {			return nil, fmt.Errorf("All serverPools should have same deployment ID expected %s, got %s", deploymentID, formats[i].ID)		} 		z.serverPools[i], err = newErasureSets(ctx, ep, storageDisks[i], formats[i], commonParityDrives, i)		if err != nil {			return nil, err		}	} 	z.decommissionCancelers = make([]context.CancelFunc, len(z.serverPools))	r := rand.New(rand.NewSource(time.Now().UnixNano()))	for {		err := z.Init(ctx) // Initializes all pools.		if err != nil {			if !configRetriableErrors(err) {				logger.Fatal(err, "Unable to initialize backend")			}			retry := time.Duration(r.Float64() * float64(5*time.Second))			logger.LogIf(ctx, fmt.Errorf("Unable to initialize backend: %w, retrying in %s", err, retry))			time.Sleep(retry)			continue		}		break	} 	drives := make([]string, 0, len(localDrives))	for _, localDrive := range localDrives {		drives = append(drives, localDrive.Endpoint().Path)	} 	globalLocalDrives = localDrives	ctx, z.shutdown = context.WithCancel(ctx)	go intDataUpdateTracker.start(ctx, drives...)	return z, nil} func (z *erasureServerPools) NewNSLock(bucket string, objects ...string) RWLocker {	return z.serverPools[0].NewNSLock(bucket, objects...)} // GetDisksID will return disks by their ID.func (z *erasureServerPools) GetDisksID(ids ...string) []StorageAPI {	idMap := make(map[string]struct{})	for _, id := range ids {		idMap[id] = struct{}{}	}	res := make([]StorageAPI, 0, len(idMap))	for _, s := range z.serverPools {		s.erasureDisksMu.RLock()		defer s.erasureDisksMu.RUnlock()		for _, disks := range s.erasureDisks {			for _, disk := range disks {				if disk == OfflineDisk {					continue				}				if id, _ := disk.GetDiskID(); id != "" {					if _, ok := idMap[id]; ok {						res = append(res, disk)					}				}			}		}	}	return res} // GetRawData will return all files with a given raw path to the callback.// Errors are ignored, only errors from the callback are returned.// For now only direct file paths are supported.func (z *erasureServerPools) GetRawData(ctx context.Context, volume, file string, fn func(r io.Reader, host string, disk string, filename string, info StatInfo) error) error {	found := 0	for _, s := range z.serverPools {		for _, disks := range s.erasureDisks {			for _, disk := range disks {				if disk == OfflineDisk {					continue				}				stats, err := disk.StatInfoFile(ctx, volume, file, true)				if err != nil {					continue				}				for _, si := range stats {					found++					var r io.ReadCloser					if !si.Dir {						r, err = disk.ReadFileStream(ctx, volume, si.Name, 0, si.Size)						if err != nil {							continue						}					} else {						r = io.NopCloser(bytes.NewBuffer([]byte{}))					}					// Keep disk path instead of ID, to ensure that the downloaded zip file can be					// easily automated with `minio server hostname{1...n}/disk{1...m}`.					err = fn(r, disk.Hostname(), disk.Endpoint().Path, pathJoin(volume, si.Name), si)					r.Close()					if err != nil {						return err					}				}			}		}	}	if found == 0 {		return errFileNotFound	}	return nil} // Return the count of disks in each poolfunc (z *erasureServerPools) SetDriveCounts() []int {	setDriveCounts := make([]int, len(z.serverPools))	for i := range z.serverPools {		setDriveCounts[i] = z.serverPools[i].SetDriveCount()	}	return setDriveCounts} type serverPoolsAvailableSpace []poolAvailableSpace type poolAvailableSpace struct {	Index      int	Available  uint64	MaxUsedPct int // Used disk percentage of most filled disk, rounded down.} // TotalAvailable - total available spacefunc (p serverPoolsAvailableSpace) TotalAvailable() uint64 {	total := uint64(0)	for _, z := range p {		total += z.Available	}	return total} // FilterMaxUsed will filter out any pools that has used percent bigger than max,// unless all have that, in which case all are preserved.func (p serverPoolsAvailableSpace) FilterMaxUsed(max int) {	// We aren't modifying p, only entries in it, so we don't need to receive a pointer.	if len(p) <= 1 {		// Nothing to do.		return	}	var ok bool	for _, z := range p {		if z.MaxUsedPct < max {			ok = true			break		}	}	if !ok {		// All above limit.		// Do not modify		return	} 	// Remove entries that are above.	for i, z := range p {		if z.MaxUsedPct < max {			continue		}		p[i].Available = 0	}} // getAvailablePoolIdx will return an index that can hold size bytes.// -1 is returned if no serverPools have available space for the size given.func (z *erasureServerPools) getAvailablePoolIdx(ctx context.Context, bucket, object string, size int64) int {	serverPools := z.getServerPoolsAvailableSpace(ctx, bucket, object, size)	serverPools.FilterMaxUsed(100 - (100 * diskReserveFraction))	total := serverPools.TotalAvailable()	if total == 0 {		return -1	}	// choose when we reach this many	choose := rand.Uint64() % total	atTotal := uint64(0)	for _, pool := range serverPools {		atTotal += pool.Available		if atTotal > choose && pool.Available > 0 {			return pool.Index		}	}	// Should not happen, but print values just in case.	logger.LogIf(ctx, fmt.Errorf("reached end of serverPools (total: %v, atTotal: %v, choose: %v)", total, atTotal, choose))	return -1} // getServerPoolsAvailableSpace will return the available space of each pool after storing the content.// If there is not enough space the pool will return 0 bytes available.// The size of each will be multiplied by the number of sets.// Negative sizes are seen as 0 bytes.func (z *erasureServerPools) getServerPoolsAvailableSpace(ctx context.Context, bucket, object string, size int64) serverPoolsAvailableSpace {	serverPools := make(serverPoolsAvailableSpace, len(z.serverPools)) 	storageInfos := make([][]*DiskInfo, len(z.serverPools))	nSets := make([]int, len(z.serverPools))	g := errgroup.WithNErrs(len(z.serverPools))	for index := range z.serverPools {		index := index		// skip suspended pools for any new I/O.		if z.IsSuspended(index) {			continue		}		pool := z.serverPools[index]		nSets[index] = pool.setCount		g.Go(func() error {			// Get the set where it would be placed.			storageInfos[index] = getDiskInfos(ctx, pool.getHashedSet(object).getDisks()...)			return nil		}, index)	} 	// Wait for the go routines.	g.Wait() 	for i, zinfo := range storageInfos {		var available uint64		if !isMinioMetaBucketName(bucket) && !hasSpaceFor(zinfo, size) {			serverPools[i] = poolAvailableSpace{Index: i}			continue		}		var maxUsedPct int		for _, disk := range zinfo {			if disk == nil || disk.Total == 0 {				continue			}			available += disk.Total - disk.Used 			// set maxUsedPct to the value from the disk with the least space percentage.			if pctUsed := int(disk.Used * 100 / disk.Total); pctUsed > maxUsedPct {				maxUsedPct = pctUsed			}		} 		// Since we are comparing pools that may have a different number of sets		// we multiply by the number of sets in the pool.		// This will compensate for differences in set sizes		// when choosing destination pool.		// Different set sizes are already compensated by less disks.		available *= uint64(nSets[i]) 		serverPools[i] = poolAvailableSpace{			Index:      i,			Available:  available,			MaxUsedPct: maxUsedPct,		}	}	return serverPools} // poolObjInfo represents the state of an object per pooltype poolObjInfo struct {	PoolIndex int	ObjInfo   ObjectInfo	Err       error} func (z *erasureServerPools) getPoolIdxExistingWithOpts(ctx context.Context, bucket, object string, opts ObjectOptions) (idx int, err error) {	if z.SinglePool() {		return 0, nil	} 	poolObjInfos := make([]poolObjInfo, len(z.serverPools))	poolOpts := make([]ObjectOptions, len(z.serverPools))	for i := range z.serverPools {		poolOpts[i] = opts	} 	var wg sync.WaitGroup	for i, pool := range z.serverPools {		wg.Add(1)		go func(i int, pool *erasureSets, opts ObjectOptions) {			defer wg.Done()			// remember the pool index, we may sort the slice original index might be lost.			pinfo := poolObjInfo{				PoolIndex: i,			}			// do not remove this check as it can lead to inconsistencies			// for all callers of bucket replication.			opts.VersionID = ""			pinfo.ObjInfo, pinfo.Err = pool.GetObjectInfo(ctx, bucket, object, opts)			poolObjInfos[i] = pinfo		}(i, pool, poolOpts[i])	}	wg.Wait() 	// Sort the objInfos such that we always serve latest	// this is a defensive change to handle any duplicate	// content that may have been created, we always serve	// the latest object.	sort.Slice(poolObjInfos, func(i, j int) bool {		mtime1 := poolObjInfos[i].ObjInfo.ModTime		mtime2 := poolObjInfos[j].ObjInfo.ModTime		return mtime1.After(mtime2)	}) 	for _, pinfo := range poolObjInfos {		// skip all objects from suspended pools if asked by the		// caller.		if z.IsSuspended(pinfo.PoolIndex) && opts.SkipDecommissioned {			continue		} 		if pinfo.Err != nil && !isErrObjectNotFound(pinfo.Err) {			return -1, pinfo.Err		} 		if isErrObjectNotFound(pinfo.Err) {			// No object exists or its a delete marker,			// check objInfo to confirm.			if pinfo.ObjInfo.DeleteMarker && pinfo.ObjInfo.Name != "" {				return pinfo.PoolIndex, nil			} 			// objInfo is not valid, truly the object doesn't			// exist proceed to next pool.			continue		} 		return pinfo.PoolIndex, nil	} 	return -1, toObjectErr(errFileNotFound, bucket, object)} // getPoolIdxExistingNoLock returns the (first) found object pool index containing an object.// If the object exists, but the latest version is a delete marker, the index with it is still returned.// If the object does not exist ObjectNotFound error is returned.// If any other error is found, it is returned.// The check is skipped if there is only one pool, and 0, nil is always returned in that case.func (z *erasureServerPools) getPoolIdxExistingNoLock(ctx context.Context, bucket, object string) (idx int, err error) {	return z.getPoolIdxExistingWithOpts(ctx, bucket, object, ObjectOptions{		NoLock:             true,		SkipDecommissioned: true,	})} func (z *erasureServerPools) getPoolIdxNoLock(ctx context.Context, bucket, object string, size int64) (idx int, err error) {	idx, err = z.getPoolIdxExistingNoLock(ctx, bucket, object)	if err != nil && !isErrObjectNotFound(err) {		return idx, err	} 	if isErrObjectNotFound(err) {		idx = z.getAvailablePoolIdx(ctx, bucket, object, size)		if idx < 0 {			return -1, toObjectErr(errDiskFull)		}	} 	return idx, nil} // getPoolIdx returns the found previous object and its corresponding pool idx,// if none are found falls back to most available space pool, this function is// designed to be only used by PutObject, CopyObject (newObject creation) and NewMultipartUpload.func (z *erasureServerPools) getPoolIdx(ctx context.Context, bucket, object string, size int64) (idx int, err error) {	idx, err = z.getPoolIdxExistingWithOpts(ctx, bucket, object, ObjectOptions{SkipDecommissioned: true})	if err != nil && !isErrObjectNotFound(err) {		return idx, err	} 	if isErrObjectNotFound(err) {		idx = z.getAvailablePoolIdx(ctx, bucket, object, size)		if idx < 0 {			return -1, toObjectErr(errDiskFull)		}	}

Lines 1–500 of 2347. Load more