// 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 (	"context"	"encoding/xml"	"errors"	"fmt"	"io"	"net/http"	"strings"	"sync"	"sync/atomic"	"time" 	"github.com/google/uuid"	"github.com/minio/minio-go/v7/pkg/tags"	sse "github.com/minio/minio/internal/bucket/encryption"	"github.com/minio/minio/internal/bucket/lifecycle"	"github.com/minio/minio/internal/event"	xhttp "github.com/minio/minio/internal/http"	"github.com/minio/minio/internal/logger"	"github.com/minio/minio/internal/s3select") const (	// Disabled means the lifecycle rule is inactive	Disabled = "Disabled"	// TransitionStatus status of transition	TransitionStatus = "transition-status"	// TransitionedObjectName name of transitioned object	TransitionedObjectName = "transitioned-object"	// TransitionedVersionID is version of remote object	TransitionedVersionID = "transitioned-versionID"	// TransitionTier name of transition storage class	TransitionTier = "transition-tier") // LifecycleSys - Bucket lifecycle subsystem.type LifecycleSys struct{} // Get - gets lifecycle config associated to a given bucket name.func (sys *LifecycleSys) Get(bucketName string) (lc *lifecycle.Lifecycle, err error) {	if globalIsGateway {		objAPI := newObjectLayerFn()		if objAPI == nil {			return nil, errServerNotInitialized		} 		return nil, BucketLifecycleNotFound{Bucket: bucketName}	} 	return globalBucketMetadataSys.GetLifecycleConfig(bucketName)} // NewLifecycleSys - creates new lifecycle system.func NewLifecycleSys() *LifecycleSys {	return &LifecycleSys{}} type expiryTask struct {	objInfo        ObjectInfo	versionExpiry  bool	restoredObject bool} type expiryState struct {	once                sync.Once	byDaysCh            chan expiryTask	byNewerNoncurrentCh chan newerNoncurrentTask} // PendingTasks returns the number of pending ILM expiry tasks.func (es *expiryState) PendingTasks() int {	return len(es.byDaysCh) + len(es.byNewerNoncurrentCh)} // close closes work channels exactly once.func (es *expiryState) close() {	es.once.Do(func() {		close(es.byDaysCh)		close(es.byNewerNoncurrentCh)	})} // enqueueByDays enqueues object versions expired by days for expiry.func (es *expiryState) enqueueByDays(oi ObjectInfo, restoredObject bool, rmVersion bool) {	select {	case <-GlobalContext.Done():		es.close()	case es.byDaysCh <- expiryTask{objInfo: oi, versionExpiry: rmVersion, restoredObject: restoredObject}:	default:	}} // enqueueByNewerNoncurrent enqueues object versions expired by// NewerNoncurrentVersions limit for expiry.func (es *expiryState) enqueueByNewerNoncurrent(bucket string, versions []ObjectToDelete) {	select {	case <-GlobalContext.Done():		es.close()	case es.byNewerNoncurrentCh <- newerNoncurrentTask{bucket: bucket, versions: versions}:	default:	}} var globalExpiryState *expiryState func newExpiryState() *expiryState {	return &expiryState{		byDaysCh:            make(chan expiryTask, 10000),		byNewerNoncurrentCh: make(chan newerNoncurrentTask, 10000),	}} func initBackgroundExpiry(ctx context.Context, objectAPI ObjectLayer) {	globalExpiryState = newExpiryState()	go func() {		for t := range globalExpiryState.byDaysCh {			if t.objInfo.TransitionedObject.Status != "" {				applyExpiryOnTransitionedObject(ctx, objectAPI, t.objInfo, t.restoredObject)			} else {				applyExpiryOnNonTransitionedObjects(ctx, objectAPI, t.objInfo, t.versionExpiry)			}		}	}()	go func() {		for t := range globalExpiryState.byNewerNoncurrentCh {			deleteObjectVersions(ctx, objectAPI, t.bucket, t.versions)		}	}()} // newerNoncurrentTask encapsulates arguments required by worker to expire objects// by NewerNoncurrentVersionstype newerNoncurrentTask struct {	bucket   string	versions []ObjectToDelete} type transitionState struct {	once         sync.Once	transitionCh chan ObjectInfo 	ctx        context.Context	objAPI     ObjectLayer	mu         sync.Mutex	numWorkers int	killCh     chan struct{} 	activeTasks int32 	lastDayMu    sync.RWMutex	lastDayStats map[string]*lastDayTierStats} func (t *transitionState) queueTransitionTask(oi ObjectInfo) {	select {	case <-GlobalContext.Done():		t.once.Do(func() {			close(t.transitionCh)		})	case t.transitionCh <- oi:	default:	}} var globalTransitionState *transitionState func newTransitionState(ctx context.Context, objAPI ObjectLayer) *transitionState {	return &transitionState{		transitionCh: make(chan ObjectInfo, 10000),		ctx:          ctx,		objAPI:       objAPI,		killCh:       make(chan struct{}),		lastDayStats: make(map[string]*lastDayTierStats),	}} // PendingTasks returns the number of ILM transition tasks waiting for a worker// goroutine.func (t *transitionState) PendingTasks() int {	return len(globalTransitionState.transitionCh)} // ActiveTasks returns the number of active (ongoing) ILM transition tasks.func (t *transitionState) ActiveTasks() int {	return int(atomic.LoadInt32(&t.activeTasks))} // worker waits for transition tasksfunc (t *transitionState) worker(ctx context.Context, objectAPI ObjectLayer) {	for {		select {		case <-t.killCh:			return		case <-ctx.Done():			return		case oi, ok := <-t.transitionCh:			if !ok {				return			}			atomic.AddInt32(&t.activeTasks, 1)			var tier string			var err error			if tier, err = transitionObject(ctx, objectAPI, oi); err != nil {				logger.LogIf(ctx, fmt.Errorf("Transition failed for %s/%s version:%s with %w", oi.Bucket, oi.Name, oi.VersionID, err))			} else {				ts := tierStats{					TotalSize:   uint64(oi.Size),					NumVersions: 1,				}				if oi.IsLatest {					ts.NumObjects = 1				}				t.addLastDayStats(tier, ts)			}			atomic.AddInt32(&t.activeTasks, -1) 		}	}} func (t *transitionState) addLastDayStats(tier string, ts tierStats) {	t.lastDayMu.Lock()	defer t.lastDayMu.Unlock() 	if _, ok := t.lastDayStats[tier]; !ok {		t.lastDayStats[tier] = &lastDayTierStats{}	}	t.lastDayStats[tier].addStats(ts)} func (t *transitionState) getDailyAllTierStats() DailyAllTierStats {	t.lastDayMu.RLock()	defer t.lastDayMu.RUnlock() 	res := make(DailyAllTierStats, len(t.lastDayStats))	for tier, st := range t.lastDayStats {		res[tier] = st.clone()	}	return res} // UpdateWorkers at the end of this function leaves n goroutines waiting for// transition tasksfunc (t *transitionState) UpdateWorkers(n int) {	t.mu.Lock()	defer t.mu.Unlock() 	for t.numWorkers < n {		go t.worker(t.ctx, t.objAPI)		t.numWorkers++	} 	for t.numWorkers > n {		go func() { t.killCh <- struct{}{} }()		t.numWorkers--	}} func initBackgroundTransition(ctx context.Context, objectAPI ObjectLayer) {	globalTransitionState = newTransitionState(ctx, objectAPI)	n := globalAPIConfig.getTransitionWorkers()	globalTransitionState.UpdateWorkers(n)} var errInvalidStorageClass = errors.New("invalid storage class") func validateTransitionTier(lc *lifecycle.Lifecycle) error {	for _, rule := range lc.Rules {		if rule.Transition.StorageClass != "" {			if valid := globalTierConfigMgr.IsTierValid(rule.Transition.StorageClass); !valid {				return errInvalidStorageClass			}		}		if rule.NoncurrentVersionTransition.StorageClass != "" {			if valid := globalTierConfigMgr.IsTierValid(rule.NoncurrentVersionTransition.StorageClass); !valid {				return errInvalidStorageClass			}		}	}	return nil} // enqueueTransitionImmediate enqueues obj for transition if eligible.// This is to be called after a successful upload of an object (version).func enqueueTransitionImmediate(obj ObjectInfo) {	if lc, err := globalLifecycleSys.Get(obj.Bucket); err == nil {		switch lc.ComputeAction(obj.ToLifecycleOpts()) {		case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:			globalTransitionState.queueTransitionTask(obj)		}	}} // expireAction represents different actions to be performed on expiry of a// restored/transitioned objecttype expireAction int const (	// ignore the zero value	_ expireAction = iota	// expireObj indicates expiry of 'regular' transitioned objects.	expireObj	// expireRestoredObj indicates expiry of restored objects.	expireRestoredObj) // expireTransitionedObject handles expiry of transitioned/restored objects// (versions) in one of the following situations://// 1. when a restored (via PostRestoreObject API) object expires.// 2. when a transitioned object expires (based on an ILM rule).func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *ObjectInfo, lcOpts lifecycle.ObjectOpts, action expireAction) error {	var opts ObjectOptions	opts.Versioned = globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name)	opts.VersionID = lcOpts.VersionID	opts.Expiration = ExpirationOptions{Expire: true}	switch action {	case expireObj:		// When an object is past expiry or when a transitioned object is being		// deleted, 'mark' the data in the remote tier for delete.		entry := jentry{			ObjName:   oi.TransitionedObject.Name,			VersionID: oi.TransitionedObject.VersionID,			TierName:  oi.TransitionedObject.Tier,		}		if err := globalTierJournal.AddEntry(entry); err != nil {			logger.LogIf(ctx, err)			return err		}		// Delete metadata on source, now that data in remote tier has been		// marked for deletion.		if _, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts); err != nil {			logger.LogIf(ctx, err)			return err		} 		// Send audit for the lifecycle delete operation		auditLogLifecycle(ctx, *oi, ILMExpiry) 		eventName := event.ObjectRemovedDelete		if lcOpts.DeleteMarker {			eventName = event.ObjectRemovedDeleteMarkerCreated		}		objInfo := ObjectInfo{			Name:         oi.Name,			VersionID:    lcOpts.VersionID,			DeleteMarker: lcOpts.DeleteMarker,		}		// Notify object deleted event.		sendEvent(eventArgs{			EventName:  eventName,			BucketName: oi.Bucket,			Object:     objInfo,			Host:       "Internal: [ILM-EXPIRY]",		}) 	case expireRestoredObj:		// delete locally restored copy of object or object version		// from the source, while leaving metadata behind. The data on		// transitioned tier lies untouched and still accessible		opts.Transition.ExpireRestored = true		_, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts)		return err	default:		return fmt.Errorf("Unknown expire action %v", action)	} 	return nil} // generate an object name for transitioned objectfunc genTransitionObjName(bucket string) (string, error) {	u, err := uuid.NewRandom()	if err != nil {		return "", err	}	us := u.String()	obj := fmt.Sprintf("%s/%s/%s/%s/%s", globalDeploymentID, bucket, us[0:2], us[2:4], us)	return obj, nil} // transition object to target specified by the transition ARN. When an object is transitioned to another// storage specified by the transition ARN, the metadata is left behind on source cluster and original content// is moved to the transition tier. Note that in the case of encrypted objects, entire encrypted stream is moved// to the transition tier without decrypting or re-encrypting.func transitionObject(ctx context.Context, objectAPI ObjectLayer, oi ObjectInfo) (string, error) {	lc, err := globalLifecycleSys.Get(oi.Bucket)	if err != nil {		return "", err	}	tier := lc.TransitionTier(oi.ToLifecycleOpts())	opts := ObjectOptions{		Transition: TransitionOptions{			Status: lifecycle.TransitionPending,			Tier:   tier,			ETag:   oi.ETag,		},		VersionID:        oi.VersionID,		Versioned:        globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name),		VersionSuspended: globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name),		MTime:            oi.ModTime,	}	return tier, objectAPI.TransitionObject(ctx, oi.Bucket, oi.Name, opts)} // getTransitionedObjectReader returns a reader from the transitioned tier.func getTransitionedObjectReader(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, oi ObjectInfo, opts ObjectOptions) (gr *GetObjectReader, err error) {	tgtClient, err := globalTierConfigMgr.getDriver(oi.TransitionedObject.Tier)	if err != nil {		return nil, fmt.Errorf("transition storage class not configured")	} 	fn, off, length, err := NewGetObjectReader(rs, oi, opts)	if err != nil {		return nil, ErrorRespToObjectError(err, bucket, object)	}	gopts := WarmBackendGetOpts{} 	// get correct offsets for object	if off >= 0 && length >= 0 {		gopts.startOffset = off		gopts.length = length	} 	reader, err := tgtClient.Get(ctx, oi.TransitionedObject.Name, remoteVersionID(oi.TransitionedObject.VersionID), gopts)	if err != nil {		return nil, err	}	closer := func() {		reader.Close()	}	return fn(reader, h, closer)} // RestoreRequestType represents type of restore.type RestoreRequestType string const (	// SelectRestoreRequest specifies select request. This is the only valid value	SelectRestoreRequest RestoreRequestType = "SELECT") // Encryption specifies encryption setting on restored buckettype Encryption struct {	EncryptionType sse.Algorithm `xml:"EncryptionType"`	KMSContext     string        `xml:"KMSContext,omitempty"`	KMSKeyID       string        `xml:"KMSKeyId,omitempty"`} // MetadataEntry denotes name and value.type MetadataEntry struct {	Name  string `xml:"Name"`	Value string `xml:"Value"`} // S3Location specifies s3 location that receives result of a restore object requesttype S3Location struct {	BucketName   string          `xml:"BucketName,omitempty"`	Encryption   Encryption      `xml:"Encryption,omitempty"`	Prefix       string          `xml:"Prefix,omitempty"`	StorageClass string          `xml:"StorageClass,omitempty"`	Tagging      *tags.Tags      `xml:"Tagging,omitempty"`	UserMetadata []MetadataEntry `xml:"UserMetadata"`} // OutputLocation specifies bucket where object needs to be restoredtype OutputLocation struct {	S3 S3Location `xml:"S3,omitempty"`} // IsEmpty returns true if output location not specified.func (o *OutputLocation) IsEmpty() bool {	return o.S3.BucketName == ""} // SelectParameters specifies sql select parameterstype SelectParameters struct {	s3select.S3Select} // IsEmpty returns true if no select parameters set

Lines 1–500 of 794. Load more