minio/cmd/batch-handlers.go at minkms · bdx0/minio

Repository navigation

11

942

943

944

945

946

947

948

949

950

951

952

953

954

955

956

957

958

959

960

961

962

963

964

965

966

967

968

969

970

971

972

973

974

975

976

977

978

979

980

981

982

983

984

985

986

987

988

989

990

991

992

993

994

995

996

997

998

999

1000

// 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"

"encoding/binary"

"encoding/json"

"errors"

"fmt"

"io"

"math/rand"

"net/http"

"net/url"

"runtime"

"strconv"

"strings"

"sync"

"time"

"github.com/dustin/go-humanize"

"github.com/lithammer/shortuuid/v4"

"github.com/minio/madmin-go/v3"

"github.com/minio/minio-go/v7"

miniogo "github.com/minio/minio-go/v7"

"github.com/minio/minio-go/v7/pkg/credentials"

"github.com/minio/minio-go/v7/pkg/encrypt"

"github.com/minio/minio-go/v7/pkg/tags"

"github.com/minio/minio/internal/config/batch"

"github.com/minio/minio/internal/crypto"

"github.com/minio/minio/internal/hash"

xhttp "github.com/minio/minio/internal/http"

"github.com/minio/minio/internal/ioutil"

xioutil "github.com/minio/minio/internal/ioutil"

"github.com/minio/minio/internal/logger"

"github.com/minio/pkg/v2/console"

"github.com/minio/pkg/v2/env"

"github.com/minio/pkg/v2/policy"

"github.com/minio/pkg/v2/workers"

"gopkg.in/yaml.v3"

)

var globalBatchConfig batch.Config

// BatchJobRequest this is an internal data structure not for external consumption.

type BatchJobRequest struct {

ID string `yaml:"-" json:"name"`

User string `yaml:"-" json:"user"`

Started time.Time `yaml:"-" json:"started"`

Replicate *BatchJobReplicateV1 `yaml:"replicate" json:"replicate"`

KeyRotate *BatchJobKeyRotateV1 `yaml:"keyrotate" json:"keyrotate"`

Expire *BatchJobExpire `yaml:"expire" json:"expire"`

ctx context.Context `msg:"-"`

}

func notifyEndpoint(ctx context.Context, ri *batchJobInfo, endpoint, token string) error {

if endpoint == "" {

return nil

}

buf, err := json.Marshal(ri)

if err != nil {

return err

}

ctx, cancel := context.WithTimeout(ctx, 10*time.Second)

defer cancel()

req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(buf))

if err != nil {

return err

}

if token != "" {

req.Header.Set("Authorization", token)

}

req.Header.Set("Content-Type", "application/json")

clnt := http.Client{Transport: getRemoteInstanceTransport}

resp, err := clnt.Do(req)

if err != nil {

return err

}

xhttp.DrainBody(resp.Body)

if resp.StatusCode != http.StatusOK {

return errors.New(resp.Status)

}

return nil

}

// Notify notifies notification endpoint if configured regarding job failure or success.

func (r BatchJobReplicateV1) Notify(ctx context.Context, ri *batchJobInfo) error {

return notifyEndpoint(ctx, ri, r.Flags.Notify.Endpoint, r.Flags.Notify.Token)

}

// ReplicateFromSource - this is not implemented yet where source is 'remote' and target is local.

func (r *BatchJobReplicateV1) ReplicateFromSource(ctx context.Context, api ObjectLayer, core *miniogo.Core, srcObjInfo ObjectInfo, retry bool) error {

srcBucket := r.Source.Bucket

tgtBucket := r.Target.Bucket

srcObject := srcObjInfo.Name

tgtObject := srcObjInfo.Name

if r.Target.Prefix != "" {

tgtObject = pathJoin(r.Target.Prefix, srcObjInfo.Name)

}

versionID := srcObjInfo.VersionID

if r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3 {

versionID = ""

}

if srcObjInfo.DeleteMarker {

_, err := api.DeleteObject(ctx, tgtBucket, tgtObject, ObjectOptions{

VersionID: versionID,

// Since we are preserving a delete marker, we have to make sure this is always true.

// regardless of the current configuration of the bucket we must preserve all versions

// on the pool being batch replicated from source.

Versioned: true,

MTime: srcObjInfo.ModTime,

DeleteMarker: srcObjInfo.DeleteMarker,

ReplicationRequest: true,

})

return err

}

opts := ObjectOptions{

VersionID: srcObjInfo.VersionID,

MTime: srcObjInfo.ModTime,

PreserveETag: srcObjInfo.ETag,

UserDefined: srcObjInfo.UserDefined,

}

if r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3 {

opts.VersionID = ""

}

if crypto.S3.IsEncrypted(srcObjInfo.UserDefined) {

opts.ServerSideEncryption = encrypt.NewSSE()

}

slc := strings.Split(srcObjInfo.ETag, "-")

if len(slc) == 2 {

partsCount, err := strconv.Atoi(slc[1])

if err != nil {

return err

}

return r.copyWithMultipartfromSource(ctx, api, core, srcObjInfo, opts, partsCount)

}

gopts := miniogo.GetObjectOptions{

VersionID: srcObjInfo.VersionID,

}

if err := gopts.SetMatchETag(srcObjInfo.ETag); err != nil {

return err

}

rd, objInfo, _, err := core.GetObject(ctx, srcBucket, srcObject, gopts)

if err != nil {

return ErrorRespToObjectError(err, srcBucket, srcObject, srcObjInfo.VersionID)

}

defer rd.Close()

hr, err := hash.NewReader(ctx, rd, objInfo.Size, "", "", objInfo.Size)

if err != nil {

return err

}

pReader := NewPutObjReader(hr)

_, err = api.PutObject(ctx, tgtBucket, tgtObject, pReader, opts)

return err

}

func (r *BatchJobReplicateV1) copyWithMultipartfromSource(ctx context.Context, api ObjectLayer, c *miniogo.Core, srcObjInfo ObjectInfo, opts ObjectOptions, partsCount int) (err error) {

srcBucket := r.Source.Bucket

tgtBucket := r.Target.Bucket

srcObject := srcObjInfo.Name

tgtObject := srcObjInfo.Name

if r.Target.Prefix != "" {

tgtObject = pathJoin(r.Target.Prefix, srcObjInfo.Name)

}

if r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3 {

opts.VersionID = ""

}

var uploadedParts []CompletePart

res, err := api.NewMultipartUpload(context.Background(), tgtBucket, tgtObject, opts)

if err != nil {

return err

}

defer func() {

if err != nil {

// block and abort remote upload upon failure.

attempts := 1

for attempts <= 3 {

aerr := api.AbortMultipartUpload(ctx, tgtBucket, tgtObject, res.UploadID, ObjectOptions{})

if aerr == nil {

return

}

logger.LogIf(ctx,

fmt.Errorf("trying %s: Unable to cleanup failed multipart replication %s on remote %s/%s: %w - this may consume space on remote cluster",

humanize.Ordinal(attempts), res.UploadID, tgtBucket, tgtObject, aerr))

attempts++

time.Sleep(time.Second)

}

}()

var (

hr *hash.Reader

pInfo PartInfo

)

for i := 0; i < partsCount; i++ {

gopts := miniogo.GetObjectOptions{

VersionID: srcObjInfo.VersionID,

PartNumber: i + 1,

}

if err := gopts.SetMatchETag(srcObjInfo.ETag); err != nil {

return err

}

rd, objInfo, _, err := c.GetObject(ctx, srcBucket, srcObject, gopts)

if err != nil {

return ErrorRespToObjectError(err, srcBucket, srcObject, srcObjInfo.VersionID)

}

defer rd.Close()

hr, err = hash.NewReader(ctx, io.LimitReader(rd, objInfo.Size), objInfo.Size, "", "", objInfo.Size)

if err != nil {

return err

}

pReader := NewPutObjReader(hr)

opts.PreserveETag = ""

pInfo, err = api.PutObjectPart(ctx, tgtBucket, tgtObject, res.UploadID, i+1, pReader, opts)

if err != nil {

return err

}

if pInfo.Size != objInfo.Size {

return fmt.Errorf("Part size mismatch: got %d, want %d", pInfo.Size, objInfo.Size)

}

uploadedParts = append(uploadedParts, CompletePart{

PartNumber: pInfo.PartNumber,

ETag: pInfo.ETag,

})

}

_, err = api.CompleteMultipartUpload(ctx, tgtBucket, tgtObject, res.UploadID, uploadedParts, opts)

return err

}

// StartFromSource starts the batch replication job from remote source, resumes if there was a pending job via "job.ID"

func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLayer, job BatchJobRequest) error {

ri := &batchJobInfo{

JobID: job.ID,

JobType: string(job.Type()),

StartTime: job.Started,

}

if err := ri.load(ctx, api, job); err != nil {

return err

}

if ri.Complete {

return nil

}

globalBatchJobsMetrics.save(job.ID, ri)

delay := job.Replicate.Flags.Retry.Delay

if delay == 0 {

delay = batchReplJobDefaultRetryDelay

}

rnd := rand.New(rand.NewSource(time.Now().UnixNano()))

isTags := len(r.Flags.Filter.Tags) != 0

isMetadata := len(r.Flags.Filter.Metadata) != 0

isStorageClassOnly := len(r.Flags.Filter.Metadata) == 1 && strings.EqualFold(r.Flags.Filter.Metadata[0].Key, xhttp.AmzStorageClass)

skip := func(oi ObjectInfo) (ok bool) {

if r.Flags.Filter.OlderThan > 0 && time.Since(oi.ModTime) < r.Flags.Filter.OlderThan {

// skip all objects that are newer than specified older duration

return true

}

if r.Flags.Filter.NewerThan > 0 && time.Since(oi.ModTime) >= r.Flags.Filter.NewerThan {

// skip all objects that are older than specified newer duration

return true

}

if !r.Flags.Filter.CreatedAfter.IsZero() && r.Flags.Filter.CreatedAfter.Before(oi.ModTime) {

// skip all objects that are created before the specified time.

return true

}

if !r.Flags.Filter.CreatedBefore.IsZero() && r.Flags.Filter.CreatedBefore.After(oi.ModTime) {

// skip all objects that are created after the specified time.

return true

}

if isTags {

// Only parse object tags if tags filter is specified.

tagMap := map[string]string{}

tagStr := oi.UserTags

if len(tagStr) != 0 {

t, err := tags.ParseObjectTags(tagStr)

if err != nil {

return false

}

tagMap = t.ToMap()

}

for _, kv := range r.Flags.Filter.Tags {

for t, v := range tagMap {

if kv.Match(BatchJobKV{Key: t, Value: v}) {

return true

}

// None of the provided tags filter match skip the object

return false

}

for _, kv := range r.Flags.Filter.Metadata {

for k, v := range oi.UserDefined {

if !stringsHasPrefixFold(k, "x-amz-meta-") && !isStandardHeader(k) {

continue

}

// We only need to match x-amz-meta or standardHeaders

if kv.Match(BatchJobKV{Key: k, Value: v}) {

return true

}

// None of the provided filters match

return false

}

u, err := url.Parse(r.Source.Endpoint)

if err != nil {

return err

}

cred := r.Source.Creds

c, err := miniogo.New(u.Host, &miniogo.Options{

Creds: credentials.NewStaticV4(cred.AccessKey, cred.SecretKey, cred.SessionToken),

Secure: u.Scheme == "https",

Transport: getRemoteInstanceTransport,

BucketLookup: lookupStyle(r.Source.Path),

})

if err != nil {

return err

}

c.SetAppInfo("minio-"+batchJobPrefix, r.APIVersion+" "+job.ID)

core := &miniogo.Core{Client: c}

workerSize, err := strconv.Atoi(env.Get("_MINIO_BATCH_REPLICATION_WORKERS", strconv.Itoa(runtime.GOMAXPROCS(0)/2)))

if err != nil {

return err

}

wk, err := workers.New(workerSize)

if err != nil {

// invalid worker size.

return err

}

retryAttempts := ri.RetryAttempts

retry := false

for attempts := 1; attempts <= retryAttempts; attempts++ {

attempts := attempts

// one of source/target is s3, skip delete marker and all versions under the same object name.

s3Type := r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3

minioSrc := r.Source.Type == BatchJobReplicateResourceMinIO

ctx, cancel := context.WithCancel(ctx)

objInfoCh := c.ListObjects(ctx, r.Source.Bucket, miniogo.ListObjectsOptions{

Prefix: r.Source.Prefix,

WithVersions: minioSrc,

Recursive: true,

WithMetadata: true,

})

prevObj := ""

skipReplicate := false

for obj := range objInfoCh {

oi := toObjectInfo(r.Source.Bucket, obj.Key, obj)

if !minioSrc {

// Check if metadata filter was requested and it is expected to have

// all user metadata or just storageClass. If its only storageClass

// List() already returns relevant information for filter to be applied.

if isMetadata && !isStorageClassOnly {

oi2, err := c.StatObject(ctx, r.Source.Bucket, obj.Key, miniogo.StatObjectOptions{})

if err == nil {

oi = toObjectInfo(r.Source.Bucket, obj.Key, oi2)

} else {

if !isErrMethodNotAllowed(ErrorRespToObjectError(err, r.Source.Bucket, obj.Key)) &&

!isErrObjectNotFound(ErrorRespToObjectError(err, r.Source.Bucket, obj.Key)) {

logger.LogIf(ctx, err)

}

continue

}

if isTags {

tags, err := c.GetObjectTagging(ctx, r.Source.Bucket, obj.Key, minio.GetObjectTaggingOptions{})

if err == nil {

oi.UserTags = tags.String()

} else {

if !isErrMethodNotAllowed(ErrorRespToObjectError(err, r.Source.Bucket, obj.Key)) &&

!isErrObjectNotFound(ErrorRespToObjectError(err, r.Source.Bucket, obj.Key)) {

logger.LogIf(ctx, err)

}

continue

}

if skip(oi) {

continue

}

if obj.Key != prevObj {

prevObj = obj.Key

// skip replication of delete marker and all versions under the same object name if one of source or target is s3.

skipReplicate = obj.IsDeleteMarker && s3Type

}

if skipReplicate {

continue

}

wk.Take()

go func() {

defer wk.Give()

stopFn := globalBatchJobsMetrics.trace(batchJobMetricReplication, job.ID, attempts)

success := true

if err := r.ReplicateFromSource(ctx, api, core, oi, retry); err != nil {

// object must be deleted concurrently, allow these failures but do not count them

if isErrVersionNotFound(err) || isErrObjectNotFound(err) {

return

}

stopFn(oi, err)

logger.LogIf(ctx, err)

success = false

} else {

stopFn(oi, nil)

}

ri.trackCurrentBucketObject(r.Target.Bucket, oi, success)

globalBatchJobsMetrics.save(job.ID, ri)

// persist in-memory state to disk after every 10secs.

logger.LogIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job))

if wait := globalBatchConfig.ReplicationWait(); wait > 0 {

time.Sleep(wait)

}

}()

}

wk.Wait()

ri.RetryAttempts = attempts

ri.Complete = ri.ObjectsFailed == 0

ri.Failed = ri.ObjectsFailed > 0

globalBatchJobsMetrics.save(job.ID, ri)

// persist in-memory state to disk.

logger.LogIf(ctx, ri.updateAfter(ctx, api, 0, job))

if err := r.Notify(ctx, ri); err != nil {

logger.LogIf(ctx, fmt.Errorf("unable to notify %v", err))

}

cancel()

if ri.Failed {

ri.ObjectsFailed = 0

ri.Bucket = ""

ri.Object = ""

ri.Objects = 0

ri.BytesFailed = 0

ri.BytesTransferred = 0

retry = true // indicate we are retrying..

time.Sleep(delay + time.Duration(rnd.Float64()*float64(delay)))

continue

}

break

}

return nil

}

// toObjectInfo converts minio.ObjectInfo to ObjectInfo

func toObjectInfo(bucket, object string, objInfo miniogo.ObjectInfo) ObjectInfo {

tags, _ := tags.MapToObjectTags(objInfo.UserTags)

oi := ObjectInfo{

Bucket: bucket,

Name: object,

ModTime: objInfo.LastModified,

Size: objInfo.Size,

ETag: objInfo.ETag,

VersionID: objInfo.VersionID,

IsLatest: objInfo.IsLatest,

DeleteMarker: objInfo.IsDeleteMarker,

ContentType: objInfo.ContentType,

Expires: objInfo.Expires,

StorageClass: objInfo.StorageClass,

ReplicationStatusInternal: objInfo.ReplicationStatus,

UserTags: tags.String(),

}

oi.UserDefined = make(map[string]string, len(objInfo.Metadata))

for k, v := range objInfo.Metadata {

oi.UserDefined[k] = v[0]

}

ce, ok := oi.UserDefined[xhttp.ContentEncoding]

if !ok {

ce, ok = oi.UserDefined[strings.ToLower(xhttp.ContentEncoding)]

}

if ok {

oi.ContentEncoding = ce

}

_, ok = oi.UserDefined[xhttp.AmzStorageClass]

if !ok {

oi.UserDefined[xhttp.AmzStorageClass] = objInfo.StorageClass

}

for k, v := range objInfo.UserMetadata {

oi.UserDefined[k] = v

}

return oi

}

func (r BatchJobReplicateV1) writeAsArchive(ctx context.Context, objAPI ObjectLayer, remoteClnt *minio.Client, entries []ObjectInfo) error {

input := make(chan minio.SnowballObject, 1)

opts := minio.SnowballOptions{

Opts: minio.PutObjectOptions{},

InMemory: *r.Source.Snowball.InMemory,

Compress: *r.Source.Snowball.Compress,

SkipErrs: *r.Source.Snowball.SkipErrs,

}

go func() {

defer xioutil.SafeClose(input)

for _, entry := range entries {

gr, err := objAPI.GetObjectNInfo(ctx, r.Source.Bucket,

entry.Name, nil, nil, ObjectOptions{

VersionID: entry.VersionID,

})

if err != nil {

logger.LogIf(ctx, err)

continue

}

snowballObj := minio.SnowballObject{

// Create path to store objects within the bucket.

Key: entry.Name,

Size: entry.Size,

ModTime: entry.ModTime,

VersionID: entry.VersionID,

Content: gr,

Headers: make(http.Header),

Close: func() {

gr.Close()

}

opts, err := batchReplicationOpts(ctx, "", gr.ObjInfo)

if err != nil {

logger.LogIf(ctx, err)

continue

}

for k, vals := range opts.Header() {

for _, v := range vals {

snowballObj.Headers.Add(k, v)

}

input <- snowballObj

}

}()

// Collect and upload all entries.

return remoteClnt.PutObjectsSnowball(ctx, r.Target.Bucket, opts, input)

}

// ReplicateToTarget read from source and replicate to configured target

func (r *BatchJobReplicateV1) ReplicateToTarget(ctx context.Context, api ObjectLayer, c *miniogo.Core, srcObjInfo ObjectInfo, retry bool) error {

srcBucket := r.Source.Bucket

tgtBucket := r.Target.Bucket

tgtPrefix := r.Target.Prefix

srcObject := srcObjInfo.Name

s3Type := r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3

if srcObjInfo.DeleteMarker || !srcObjInfo.VersionPurgeStatus.Empty() {

if retry && !s3Type {

if _, err := c.StatObject(ctx, tgtBucket, pathJoin(tgtPrefix, srcObject), miniogo.StatObjectOptions{

VersionID: srcObjInfo.VersionID,

Internal: miniogo.AdvancedGetOptions{

ReplicationProxyRequest: "false",

}); isErrMethodNotAllowed(ErrorRespToObjectError(err, tgtBucket, pathJoin(tgtPrefix, srcObject))) {

return nil

}

versionID := srcObjInfo.VersionID

dmVersionID := ""

if srcObjInfo.VersionPurgeStatus.Empty() {

dmVersionID = srcObjInfo.VersionID

}

if r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3 {

dmVersionID = ""

versionID = ""

}

return c.RemoveObject(ctx, tgtBucket, pathJoin(tgtPrefix, srcObject), miniogo.RemoveObjectOptions{

VersionID: versionID,

Internal: miniogo.AdvancedRemoveOptions{

ReplicationDeleteMarker: dmVersionID != "",

ReplicationMTime: srcObjInfo.ModTime,

ReplicationStatus: miniogo.ReplicationStatusReplica,

ReplicationRequest: true, // always set this to distinguish between `mc mirror` replication and serverside

})

}

if retry && !s3Type { // when we are retrying avoid copying if necessary.

gopts := miniogo.GetObjectOptions{}

if err := gopts.SetMatchETag(srcObjInfo.ETag); err != nil {

return err

}

if _, err := c.StatObject(ctx, tgtBucket, pathJoin(tgtPrefix, srcObject), gopts); err == nil {

return nil

}

versioned := globalBucketVersioningSys.PrefixEnabled(srcBucket, srcObject)

versionSuspended := globalBucketVersioningSys.PrefixSuspended(srcBucket, srcObject)

opts := ObjectOptions{

VersionID: srcObjInfo.VersionID,

Versioned: versioned,

VersionSuspended: versionSuspended,

}

rd, err := api.GetObjectNInfo(ctx, srcBucket, srcObject, nil, http.Header{}, opts)

if err != nil {

return err

}

defer rd.Close()

objInfo := rd.ObjInfo

size, err := objInfo.GetActualSize()

if err != nil {

return err

}

putOpts, err := batchReplicationOpts(ctx, "", objInfo)

if err != nil {

return err

}

if r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3 {

putOpts.Internal = miniogo.AdvancedPutOptions{}

}

if objInfo.isMultipart() {

if err := replicateObjectWithMultipart(ctx, c, tgtBucket, pathJoin(tgtPrefix, objInfo.Name), rd, objInfo, putOpts); err != nil {

return err

}

} else {

if _, err = c.PutObject(ctx, tgtBucket, pathJoin(tgtPrefix, objInfo.Name), rd, size, "", "", putOpts); err != nil {

return err

}

return nil

}

//go:generate msgp -file $GOFILE -unexported

// batchJobInfo current batch replication information

type batchJobInfo struct {

mu sync.RWMutex `json:"-" msg:"-"`

Version int `json:"-" msg:"v"`

JobID string `json:"jobID" msg:"jid"`

JobType string `json:"jobType" msg:"jt"`

StartTime time.Time `json:"startTime" msg:"st"`

LastUpdate time.Time `json:"lastUpdate" msg:"lu"`

RetryAttempts int `json:"retryAttempts" msg:"ra"`

Complete bool `json:"complete" msg:"cmp"`

Failed bool `json:"failed" msg:"fld"`

// Last bucket/object batch replicated

Bucket string `json:"-" msg:"lbkt"`

Object string `json:"-" msg:"lobj"`

// Verbose information

Objects int64 `json:"objects" msg:"ob"`

DeleteMarkers int64 `json:"deleteMarkers" msg:"dm"`

ObjectsFailed int64 `json:"objectsFailed" msg:"obf"`

DeleteMarkersFailed int64 `json:"deleteMarkersFailed" msg:"dmf"`

BytesTransferred int64 `json:"bytesTransferred" msg:"bt"`

BytesFailed int64 `json:"bytesFailed" msg:"bf"`

}

const (

batchReplName = "batch-replicate.bin"

batchReplFormat = 1

batchReplVersionV1 = 1

batchReplVersion = batchReplVersionV1

batchJobName = "job.bin"

batchJobPrefix = "batch-jobs"

batchJobReportsPrefix = batchJobPrefix + "/reports"

batchReplJobAPIVersion = "v1"

batchReplJobDefaultRetries = 3

batchReplJobDefaultRetryDelay = 250 * time.Millisecond

)

func getJobReportPath(job BatchJobRequest) string {

var fileName string

switch {

case job.Replicate != nil:

fileName = batchReplName

case job.KeyRotate != nil:

fileName = batchKeyRotationName

case job.Expire != nil:

fileName = batchExpireName

}

return pathJoin(batchJobReportsPrefix, job.ID, fileName)

}

func getJobPath(job BatchJobRequest) string {

return pathJoin(batchJobPrefix, job.ID)

}

func (ri *batchJobInfo) load(ctx context.Context, api ObjectLayer, job BatchJobRequest) error {

var format, version uint16

switch {

case job.Replicate != nil:

version = batchReplVersionV1

format = batchReplFormat

case job.KeyRotate != nil:

version = batchKeyRotateVersionV1

format = batchKeyRotationFormat

case job.Expire != nil:

version = batchExpireVersionV1

format = batchExpireFormat

default:

return errors.New("no supported batch job request specified")

}

data, err := readConfig(ctx, api, getJobReportPath(job))

if err != nil {

if errors.Is(err, errConfigNotFound) || isErrObjectNotFound(err) {

ri.Version = int(version)

switch {

case job.Replicate != nil:

ri.RetryAttempts = batchReplJobDefaultRetries

if job.Replicate.Flags.Retry.Attempts > 0 {

ri.RetryAttempts = job.Replicate.Flags.Retry.Attempts

}

case job.KeyRotate != nil:

ri.RetryAttempts = batchKeyRotateJobDefaultRetries

if job.KeyRotate.Flags.Retry.Attempts > 0 {

ri.RetryAttempts = job.KeyRotate.Flags.Retry.Attempts

}

case job.Expire != nil:

ri.RetryAttempts = batchExpireJobDefaultRetries

if job.Expire.Retry.Attempts > 0 {

ri.RetryAttempts = job.Expire.Retry.Attempts

}

return nil

}

return err

}

if len(data) == 0 {

// Seems to be empty create a new batchRepl object.

return nil

}

if len(data) <= 4 {

return fmt.Errorf("%s: no data", ri.JobType)

}

// Read header

switch binary.LittleEndian.Uint16(data[0:2]) {

case format:

default:

return fmt.Errorf("%s: unknown format: %d", ri.JobType, binary.LittleEndian.Uint16(data[0:2]))

}

switch binary.LittleEndian.Uint16(data[2:4]) {

case version:

default:

return fmt.Errorf("%s: unknown version: %d", ri.JobType, binary.LittleEndian.Uint16(data[2:4]))

}

ri.mu.Lock()

defer ri.mu.Unlock()

// OK, parse data.

if _, err = ri.UnmarshalMsg(data[4:]); err != nil {

return err

}

switch ri.Version {

case batchReplVersionV1:

default:

return fmt.Errorf("unexpected batch %s meta version: %d", ri.JobType, ri.Version)

}

return nil

}

func (ri *batchJobInfo) clone() *batchJobInfo {

ri.mu.RLock()

defer ri.mu.RUnlock()

return &batchJobInfo{

Version: ri.Version,

JobID: ri.JobID,

JobType: ri.JobType,

RetryAttempts: ri.RetryAttempts,

Complete: ri.Complete,

Failed: ri.Failed,

StartTime: ri.StartTime,

LastUpdate: ri.LastUpdate,

Bucket: ri.Bucket,

Object: ri.Object,

Objects: ri.Objects,

ObjectsFailed: ri.ObjectsFailed,

BytesTransferred: ri.BytesTransferred,

BytesFailed: ri.BytesFailed,

}

func (ri *batchJobInfo) countItem(size int64, dmarker, success bool) {

if ri == nil {

return

}

if success {

if dmarker {

ri.DeleteMarkers++

} else {

ri.Objects++

ri.BytesTransferred += size

}

} else {

if dmarker {

ri.DeleteMarkersFailed++

} else {

ri.ObjectsFailed++

ri.BytesFailed += size

}

func (ri *batchJobInfo) updateAfter(ctx context.Context, api ObjectLayer, duration time.Duration, job BatchJobRequest) error {

if ri == nil {

return errInvalidArgument

}

now := UTCNow()

ri.mu.Lock()

var (

format, version uint16

jobTyp string

)

if now.Sub(ri.LastUpdate) >= duration {

switch job.Type() {

case madmin.BatchJobReplicate:

format = batchReplFormat

version = batchReplVersion

jobTyp = string(job.Type())

ri.Version = batchReplVersionV1

case madmin.BatchJobKeyRotate:

format = batchKeyRotationFormat

version = batchKeyRotateVersion

jobTyp = string(job.Type())

ri.Version = batchKeyRotateVersionV1

case madmin.BatchJobExpire:

format = batchExpireFormat

version = batchExpireVersion

jobTyp = string(job.Type())

ri.Version = batchExpireVersionV1

default:

return errInvalidArgument

}

if serverDebugLog {

console.Debugf("%s: persisting info on drive: threshold:%s, %s:%#v\n", jobTyp, now.Sub(ri.LastUpdate), jobTyp, ri)

}

ri.LastUpdate = now

data := make([]byte, 4, ri.Msgsize()+4)

// Initialize the header.

binary.LittleEndian.PutUint16(data[0:2], format)

binary.LittleEndian.PutUint16(data[2:4], version)

buf, err := ri.MarshalMsg(data)

ri.mu.Unlock()

if err != nil {

return err

}

return saveConfig(ctx, api, getJobReportPath(job), buf)

}

ri.mu.Unlock()

return nil

}

// Note: to be used only with batch jobs that affect multiple versions through

// a single action. e.g batch-expire has an option to expire all versions of an

// object which matches the given filters.

func (ri *batchJobInfo) trackMultipleObjectVersions(bucket string, info ObjectInfo, success bool) {

if success {

ri.Objects += int64(info.NumVersions)

} else {

ri.ObjectsFailed += int64(info.NumVersions)

}

func (ri *batchJobInfo) trackCurrentBucketObject(bucket string, info ObjectInfo, success bool) {

if ri == nil {

return

}

ri.mu.Lock()

defer ri.mu.Unlock()

ri.Bucket = bucket

ri.Object = info.Name

ri.countItem(info.Size, info.DeleteMarker, success)

}

func (ri *batchJobInfo) trackCurrentBucketBatch(bucket string, batch []ObjectInfo) {

if ri == nil {

return

}

ri.mu.Lock()

defer ri.mu.Unlock()

ri.Bucket = bucket

for i := range batch {

ri.Object = batch[i].Name

ri.countItem(batch[i].Size, batch[i].DeleteMarker, true)

}

// Start start the batch replication job, resumes if there was a pending job via "job.ID"

func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job BatchJobRequest) error {

ri := &batchJobInfo{

JobID: job.ID,

JobType: string(job.Type()),

StartTime: job.Started,

}

if err := ri.load(ctx, api, job); err != nil {

return err

}

if ri.Complete {

return nil

}

globalBatchJobsMetrics.save(job.ID, ri)

lastObject := ri.Object

delay := job.Replicate.Flags.Retry.Delay

if delay == 0 {

delay = batchReplJobDefaultRetryDelay

}

rnd := rand.New(rand.NewSource(time.Now().UnixNano()))

selectObj := func(info FileInfo) (ok bool) {

if r.Flags.Filter.OlderThan > 0 && time.Since(info.ModTime) < r.Flags.Filter.OlderThan {

// skip all objects that are newer than specified older duration

return false

}

if r.Flags.Filter.NewerThan > 0 && time.Since(info.ModTime) >= r.Flags.Filter.NewerThan {

// skip all objects that are older than specified newer duration

return false

}

if !r.Flags.Filter.CreatedAfter.IsZero() && r.Flags.Filter.CreatedAfter.Before(info.ModTime) {

// skip all objects that are created before the specified time.

return false

}

if !r.Flags.Filter.CreatedBefore.IsZero() && r.Flags.Filter.CreatedBefore.After(info.ModTime) {

// skip all objects that are created after the specified time.

return false

}

if len(r.Flags.Filter.Tags) > 0 {

// Only parse object tags if tags filter is specified.

tagMap := map[string]string{}

tagStr := info.Metadata[xhttp.AmzObjectTagging]

if len(tagStr) != 0 {

View remainder of file in raw view

FilesExpand file tree

batch-handlers.go

Latest commit

History

batch-handlers.go

File metadata and controls