diff --git a/cmd/batch-authz.go b/cmd/batch-authz.go new file mode 100644 index 000000000..192fbb473 --- /dev/null +++ b/cmd/batch-authz.go @@ -0,0 +1,742 @@ +// Copyright (c) 2026 PGSTY +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "context" + "crypto/hmac" + "crypto/sha256" + "crypto/subtle" + "encoding/base64" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/minio/minio-go/v7/pkg/tags" + "github.com/minio/minio/internal/arn" + "github.com/minio/minio/internal/auth" + "github.com/pgsty/silo-pkg/v3/policy" + "github.com/pgsty/silo-pkg/v3/policy/condition" +) + +// A batch job acts on objects long after the request that submitted it has +// gone. It therefore records who submitted it and acts only as that +// principal: +// +// - StartBatchJob binds the authenticated credential to the job +// (batchJobSubmitter). Nothing in the submitted definition names it. +// - Every start of the job, and every object it touches, resolves the +// binding against IAM again. A credential that is gone, disabled, +// expired, revoked or edited since submission fails the job; a reused +// access key does not match the binding. +// - Each action is evaluated with the submitter's policies and session +// policy but without a request. A policy statement that would need +// request context the job does not have (source address, transport, +// headers and so on) refuses the action instead of being skipped. +// +// A job without a binding was queued by an earlier release. It is never +// authorized, whoever submitted it, and has to be submitted again. + +//go:generate msgp -file $GOFILE -unexported + +// batchJobSubmitter is the server-assigned identity of the credential that +// submitted a batch job. +type batchJobSubmitter struct { + Version int `msg:"v"` + Type string `msg:"t"` + AccessKey string `msg:"ak"` + // ParentUser is the parent of a derived credential. + ParentUser string `msg:"pu"` + // UpdatedAt and RevokedBefore are the revision of the identity record; + // any edit of the identity changes them. They are unset for root. + UpdatedAt time.Time `msg:"ua"` + RevokedBefore time.Time `msg:"rb"` + // Binding is a keyed fingerprint of the credential itself, see + // batchSubmitterBinding. + Binding []byte `msg:"bd"` +} + +const batchJobSubmitterVersion = 1 + +// Submitter types. A credential is classified in the order IsAllowed +// evaluates it. +const ( + batchSubmitterRoot = "root" + batchSubmitterUser = "user" + batchSubmitterTemporary = "temporary" + batchSubmitterService = "service-account" +) + +//msgp:ignore batchJobAuthzError + +// batchJobAuthzError refuses a batch job, or one of its actions, for its +// submitter. It ends the job: it is persisted as the job's failure and is +// never retried. +type batchJobAuthzError struct { + reason string +} + +func (e batchJobAuthzError) Error() string { + return e.reason +} + +func isBatchJobAuthzError(err error) bool { + var e batchJobAuthzError + return errors.As(err, &e) +} + +var ( + errBatchJobResubmit = batchJobAuthzError{reason: "the batch job does not record who submitted it; submit it again"} + errBatchJobChanged = batchJobAuthzError{reason: "the credential that submitted the batch job has changed since submission; submit it again"} + errBatchJobInvalid = batchJobAuthzError{reason: "the credential that submitted the batch job is no longer valid"} +) + +func batchSubmitterType(cred auth.Credentials) string { + switch { + case cred.IsTemp(): + return batchSubmitterTemporary + case cred.IsServiceAccount(): + return batchSubmitterService + default: + return batchSubmitterUser + } +} + +// batchSubmitterBinding is a fingerprint of one credential keyed with its own +// secret. It changes when the secret, the session token or the parent +// changes, and a new credential that reuses the access key does not have it. +func batchSubmitterBinding(typ string, cred auth.Credentials) []byte { + mac := hmac.New(sha256.New, []byte(cred.SecretKey)) + for _, s := range []string{"silo batch-job submitter v1", typ, cred.AccessKey, cred.ParentUser, cred.SessionToken} { + var n [8]byte + binary.BigEndian.PutUint64(n[:], uint64(len(s))) + mac.Write(n[:]) + mac.Write([]byte(s)) + } + return mac.Sum(nil) +} + +// identityRecord returns the stored identity of a non-root access key. +func (sys *IAMSys) identityRecord(accessKey string) (UserIdentity, bool, error) { + if !sys.Initialized() { + return UserIdentity{}, false, errServerNotInitialized + } + u, ok := sys.store.GetUser(accessKey) + return u, ok, nil +} + +// newBatchJobSubmitter binds the credential that authenticated a +// StartBatchJob request. +func newBatchJobSubmitter(cred auth.Credentials) (*batchJobSubmitter, error) { + root := globalActiveCred + if cred.AccessKey == "" { + return nil, errBatchJobInvalid + } + if cred.AccessKey == root.AccessKey { + if subtle.ConstantTimeCompare([]byte(cred.SecretKey), []byte(root.SecretKey)) != 1 { + return nil, errBatchJobChanged + } + return &batchJobSubmitter{ + Version: batchJobSubmitterVersion, + Type: batchSubmitterRoot, + AccessKey: root.AccessKey, + Binding: batchSubmitterBinding(batchSubmitterRoot, root), + }, nil + } + u, ok, err := globalIAMSys.identityRecord(cred.AccessKey) + if err != nil { + return nil, err + } + // The record must be the one the request was authenticated with. + if !ok || u.Credentials.AccessKey != cred.AccessKey || u.Credentials.ParentUser != cred.ParentUser || + subtle.ConstantTimeCompare([]byte(u.Credentials.SecretKey), []byte(cred.SecretKey)) != 1 || + subtle.ConstantTimeCompare([]byte(u.Credentials.SessionToken), []byte(cred.SessionToken)) != 1 { + return nil, errBatchJobChanged + } + typ := batchSubmitterType(u.Credentials) + return &batchJobSubmitter{ + Version: batchJobSubmitterVersion, + Type: typ, + AccessKey: u.Credentials.AccessKey, + ParentUser: u.Credentials.ParentUser, + UpdatedAt: u.UpdatedAt, + RevokedBefore: u.RevokedBefore, + Binding: batchSubmitterBinding(typ, u.Credentials), + }, nil +} + +//msgp:ignore batchJobPrincipal batchAuthzRequest + +// batchJobPrincipal is a submitter resolved against current IAM state. +type batchJobPrincipal struct { + typ string + accessKey string + parentUser string + groups []string + claims map[string]any + owner bool + temp bool // evaluated as a temporary credential + service bool // evaluated as a service account +} + +// principal resolves the binding against current IAM state. It fails if the +// credential no longer exists, is no longer valid, or is no longer the one +// that was bound. +func (s *batchJobSubmitter) principal(ctx context.Context) (batchJobPrincipal, error) { + if s == nil || s.AccessKey == "" { + return batchJobPrincipal{}, errBatchJobResubmit + } + if s.Version != batchJobSubmitterVersion { + return batchJobPrincipal{}, batchJobAuthzError{reason: fmt.Sprintf("unsupported batch job submitter version %d; submit the job again", s.Version)} + } + root := globalActiveCred + if s.Type == batchSubmitterRoot { + if s.AccessKey != root.AccessKey || !hmac.Equal(s.Binding, batchSubmitterBinding(batchSubmitterRoot, root)) { + return batchJobPrincipal{}, errBatchJobChanged + } + if !globalAPIConfig.permitRootAccess() { + return batchJobPrincipal{}, errBatchJobInvalid + } + return batchJobPrincipal{typ: batchSubmitterRoot, accessKey: root.AccessKey, owner: true}, nil + } + switch s.Type { + case batchSubmitterUser, batchSubmitterTemporary, batchSubmitterService: + default: + return batchJobPrincipal{}, errBatchJobResubmit + } + if s.AccessKey == root.AccessKey { + return batchJobPrincipal{}, errBatchJobChanged + } + u, ok, err := globalIAMSys.CheckKey(ctx, s.AccessKey) + if err != nil { + return batchJobPrincipal{}, err + } + if !ok { + // Deleted, disabled, expired or revoked. + return batchJobPrincipal{}, errBatchJobInvalid + } + cred := u.Credentials + typ := batchSubmitterType(cred) + if typ != s.Type || cred.AccessKey != s.AccessKey || cred.ParentUser != s.ParentUser || + !u.UpdatedAt.Equal(s.UpdatedAt) || !u.RevokedBefore.Equal(s.RevokedBefore) || + !hmac.Equal(s.Binding, batchSubmitterBinding(typ, cred)) { + return batchJobPrincipal{}, errBatchJobChanged + } + if cred.IsExpired() || cred.Status == auth.AccountOff { + return batchJobPrincipal{}, errBatchJobInvalid + } + p := batchJobPrincipal{ + typ: typ, + accessKey: cred.AccessKey, + parentUser: cred.ParentUser, + groups: cred.Groups, + claims: map[string]any{}, + temp: cred.IsTemp(), + service: !cred.IsTemp() && cred.IsServiceAccount(), + } + if typ != batchSubmitterUser { + // Verify the token as a request would, with the current signing keys. + jwtClaims, err := extractJWTClaims(u) + if err != nil { + return batchJobPrincipal{}, errBatchJobInvalid + } + p.claims = jwtClaims.Map() + if err := decodeBatchSessionPolicy(p.claims); err != nil { + return batchJobPrincipal{}, errBatchJobInvalid + } + } + p.owner = cred.ParentUser == root.AccessKey && cred.AccessKey != siteReplicatorSvcAcc + if p.owner && !globalAPIConfig.permitRootAccess() { + return batchJobPrincipal{}, errBatchJobInvalid + } + if _, ok := p.claims[policy.SessionPolicyName]; ok { + p.owner = false + } + return p, nil +} + +// decodeBatchSessionPolicy decodes a session policy claim for evaluation. +// A request decodes it while verifying its token, except when an +// authorization plugin is configured; a batch job decodes it always, so the +// session policy also limits what the plugin allows. +func decodeBatchSessionPolicy(claims map[string]any) error { + sp, ok := claims[policy.SessionPolicyName] + if !ok { + return nil + } + s, ok := sp.(string) + if !ok { + return errAuthentication + } + b, err := base64.StdEncoding.DecodeString(s) + if err != nil { + return errAuthentication + } + claims[sessionPolicyNameExtracted] = string(b) + return nil +} + +// batchAuthzRequest is one action a batch job is about to take. +type batchAuthzRequest struct { + action policy.Action + bucket string + object string // empty for a bucket action + // versionID is the version an object version action addresses. + versionID string + // tags are the current tags of the object version acted on. + tags string + // tagsUnknown is set when the job does not read the version it acts + // on, so s3:ExistingObjectTag is context it does not have. + tagsUnknown bool + // values are request values the job supplies itself, such as the + // listing prefix or the encryption and retention it writes, keyed by + // condition key name. + values map[string][]string +} + +// batchAuthzMarker marks the condition values of a batch job for an +// authorization plugin. Requests cannot set s3:authType. +const batchAuthzMarker = "batch-job" + +// Condition keys a batch job always provides, as a request would: identity, +// time, the addressed version and the stored tags. Keys a request takes from +// itself are provided only when the job sets them (see batchAuthzRequest +// values); every other key is request context the job does not have. +var batchServerConditionKeys = map[string]bool{ + condition.AWSCurrentTime.ToKey().String(): true, + condition.AWSEpochTime.ToKey().String(): true, + condition.AWSPrincipalType.ToKey().String(): true, + condition.AWSUserID.ToKey().String(): true, + condition.AWSUsername.ToKey().String(): true, + condition.AWSGroups.ToKey().String(): true, + condition.S3VersionID.ToKey().String(): true, + condition.ExistingObjectTag.ToKey().String(): true, +} + +// Request keys a job may supply, by the name they have in condition values. +var batchRequestConditionKeys = map[string]condition.KeyName{ + condition.S3Prefix.Name(): condition.S3Prefix, + condition.S3XAmzServerSideEncryption.Name(): condition.S3XAmzServerSideEncryption, + condition.S3XAmzServerSideEncryptionAwsKmsKeyID.Name(): condition.S3XAmzServerSideEncryptionAwsKmsKeyID, + condition.S3ObjectLockMode.Name(): condition.S3ObjectLockMode, + condition.S3ObjectLockRetainUntilDate.Name(): condition.S3ObjectLockRetainUntilDate, + condition.S3ObjectLockLegalHold.Name(): condition.S3ObjectLockLegalHold, + condition.RequestObjectTagKeys.Name(): condition.RequestObjectTagKeys, +} + +// suppliedConditionKeys returns the condition keys whose values req provides. +func (req batchAuthzRequest) suppliedConditionKeys() map[string]bool { + supplied := make(map[string]bool, len(batchServerConditionKeys)+len(req.values)) + for k := range batchServerConditionKeys { + supplied[k] = true + } + if req.tagsUnknown { + delete(supplied, condition.ExistingObjectTag.ToKey().String()) + } + for name := range req.values { + if key, ok := batchRequestConditionKeys[name]; ok { + supplied[key.ToKey().String()] = true + } + if strings.HasPrefix(name, condition.RequestObjectTag.Name()+"/") { + supplied[condition.RequestObjectTag.ToKey().String()] = true + } + } + return supplied +} + +func batchKeySupplied(supplied map[string]bool, name string) bool { + if i := strings.IndexByte(name, '/'); i >= 0 { + name = name[:i] + } + if strings.HasPrefix(name, "jwt:") || strings.HasPrefix(name, "ldap:") { + // Claims of the verified token, as a request has them. + return true + } + return supplied[name] +} + +// conditionValues are the values a request with the same identity would +// carry, without any value taken from the request itself. +func (p batchJobPrincipal) conditionValues(req batchAuthzRequest) map[string][]string { + now := UTCNow() + username := p.accessKey + if p.temp || p.service { + username = p.parentUser + } + principalType := "User" + if len(p.claims) > 0 { + principalType = "AssumedRole" + } + if username == globalActiveCred.AccessKey { + principalType = "Account" + } + // Identity and time, then the claims over them, as a request has them + // (an LDAP session's short username replaces its parent DN). + values := map[string][]string{ + "CurrentTime": {now.Format(time.RFC3339)}, + "EpochTime": {strconv.FormatInt(now.Unix(), 10)}, + "principaltype": {principalType}, + "userid": {username}, + "username": {username}, + } + for k, v := range p.claims { + if s, ok := v.(string); ok { + values[strings.ToLower(strings.TrimPrefix(k, "ldap"))] = []string{s} + } + } + // No claim stands for a value of the object or of the job, which + // follow and always win. + for k := range values { + if k == "versionid" || k == "prefix" || k == "authtype" || + strings.HasPrefix(k, "existingobjecttag/") || strings.HasPrefix(k, "requestobjecttag") || + strings.HasPrefix(k, "x-amz-") || strings.HasPrefix(k, "object-lock-") { + delete(values, k) + } + } + // For an authorization plugin: this is a batch job, not a request. + values["authType"] = []string{batchAuthzMarker} + if req.versionID != "" { + values["versionid"] = []string{req.versionID} + } + if req.tags != "" { + if t, err := tags.ParseObjectTags(req.tags); err == nil { + for k, v := range t.ToMap() { + values[pathJoin("ExistingObjectTag", k)] = []string{v} + } + } + } + for k, v := range req.values { + values[k] = v + } + if grpsVal, ok := p.claims["groups"]; ok { + if grpsIs, ok := grpsVal.([]any); ok { + grps := []string{} + for _, gI := range grpsIs { + if g, ok := gI.(string); ok { + grps = append(grps, g) + } + } + if len(grps) > 0 { + values["groups"] = grps + } + } + } + if _, ok := values["groups"]; !ok && len(p.groups) > 0 { + values["groups"] = p.groups + } + return values +} + +// authorize evaluates one action of the job as its submitter. +func (p batchJobPrincipal) authorize(req batchAuthzRequest) error { + args := policy.Args{ + AccountName: p.accessKey, + Groups: p.groups, + Action: req.action, + BucketName: req.bucket, + ObjectName: req.object, + ConditionValues: p.conditionValues(req), + IsOwner: p.owner, + Claims: p.claims, + } + denied := batchJobAuthzError{reason: batchDeniedReason(req)} + plugin := newGlobalAuthZPluginFn() + policies, err := globalIAMSys.batchLocalPolicies(p, args, plugin == nil) + if err != nil { + return denied + } + supplied := req.suppliedConditionKeys() + for _, pol := range policies { + for _, st := range pol.Statements { + if batchStatementNeedsRequest(st, args, supplied) { + return batchJobAuthzError{reason: denied.reason + ": a policy statement depends on request context that a batch job does not have"} + } + } + } + if plugin != nil { + ok, err := plugin.IsAllowed(args) + if err != nil { + authZLogIf(GlobalContext, err) + } + if !ok { + return denied + } + // The plugin does not see a session policy; it only limits further. + if has, allowed := p.sessionPolicyDecision(args); has && !allowed { + return denied + } + return nil + } + if !globalIAMSys.isAllowedLocally(args) { + return denied + } + return nil +} + +func batchDeniedReason(req batchAuthzRequest) string { + resource := req.bucket + if req.object != "" { + resource = pathJoin(req.bucket, req.object) + } + if req.versionID != "" { + resource += " (versionId " + req.versionID + ")" + } + return fmt.Sprintf("the batch job submitter is not allowed %s on %s", req.action, resource) +} + +// sessionPolicyDecision evaluates the session policy the way IsAllowed would +// for this credential. +func (p batchJobPrincipal) sessionPolicyDecision(args policy.Args) (bool, bool) { + switch { + case p.temp: + return isAllowedBySessionPolicy(args) + case p.service: + if p.claims[iamPolicyClaimNameSA()] == inheritedPolicyType { + return false, false + } + return isAllowedBySessionPolicyForServiceAccount(args) + } + return false, false +} + +// batchLocalPolicies returns every local policy that can take part in the +// decision on args: the mapped policies that IsAllowed would merge when +// withMapped is set (no authorization plugin), and the session policy. +func (sys *IAMSys) batchLocalPolicies(p batchJobPrincipal, args policy.Args, withMapped bool) ([]policy.Policy, error) { + var out []policy.Policy + if withMapped && args.IsOwner { + // IsAllowed does not evaluate any policy for the owner. + return nil, nil + } + if withMapped { + var names []string + ownerDerived := (p.temp || p.service) && p.parentUser == globalActiveCred.AccessKey + switch { + case p.typ == batchSubmitterRoot || ownerDerived: + case p.temp || p.service: + if roleArn := args.GetRoleArn(); roleArn != "" { + a, err := arn.Parse(roleArn) + if err != nil { + return nil, err + } + names = append(names, newMappedPolicy(sys.rolesMap[a]).toSlice()...) + } else { + mapped, err := sys.PolicyDBGet(p.parentUser, args.Groups...) + if err != nil { + return nil, err + } + names = append(names, mapped...) + // The claimed policies apply only when nothing is mapped. + if len(mapped) == 0 { + if claimed, ok := policy.GetPoliciesFromClaims(args.Claims, iamPolicyClaimNameOpenID()); ok { + names = append(names, claimed.ToSlice()...) + } + } + } + default: + mapped, err := sys.PolicyDBGet(p.accessKey, args.Groups...) + if err != nil { + return nil, err + } + names = append(names, mapped...) + } + if len(names) > 0 { + out = append(out, sys.GetCombinedPolicy(names...)) + } + } + if p.temp || p.service { + if p.service && p.claims[iamPolicyClaimNameSA()] == inheritedPolicyType { + return out, nil + } + if sp, ok := p.claims[sessionPolicyNameExtracted]; ok { + s, ok := sp.(string) + if !ok { + return nil, errAuthentication + } + parsed, err := policy.ParseConfig(strings.NewReader(s)) + if err != nil { + return nil, err + } + out = append(out, *parsed) + } + } + return out, nil +} + +// batchStatementNeedsRequest reports whether a statement that applies to +// args depends on a condition key that the job does not supply. Applying +// such a statement without its value could grant (Allow) or drop (Deny) +// what a request with that context would not, so the job refuses the action. +// Resources are matched without conditions; a statement is taken to apply +// whenever it might. +func batchStatementNeedsRequest(st policy.Statement, args policy.Args, supplied map[string]bool) bool { + if (!st.Actions.Match(args.Action) && !st.Actions.IsEmpty()) || st.NotActions.Match(args.Action) { + return false + } + // Policy variables in resources or condition values. + if data, err := json.Marshal(st); err == nil { + s := string(data) + for { + i := strings.Index(s, "${") + if i < 0 { + break + } + s = s[i+2:] + j := strings.IndexByte(s, '}') + if j < 0 { + break + } + if !batchKeySupplied(supplied, s[:j]) { + return true + } + s = s[j+1:] + } + } + if !batchStatementMayApply(st, args) { + return false + } + for key := range st.Conditions.Keys() { + if !batchKeySupplied(supplied, key.String()) { + return true + } + } + return false +} + +// batchStatementMayApply matches the resources of a statement, in both forms +// a bucket request can take. A statement whose resources the evaluator +// translates (S3 Tables and vectors actions cover S3 data actions) is taken +// to apply. +func batchStatementMayApply(st policy.Statement, args policy.Args) bool { + for _, actions := range []policy.ActionSet{st.Actions, st.NotActions} { + for action := range actions { + if policy.TableAction(action).IsValid() || policy.VectorsAction(action).IsValid() { + return true + } + } + } + for _, resources := range []policy.ResourceSet{st.Resources, st.NotResources} { + for r := range resources { + if r.Type != policy.ResourceARNS3 && r.Type != policy.ResourceARNAll { + return true + } + } + } + resources := []string{args.BucketName + "/" + strings.TrimPrefix(args.ObjectName, "/")} + if args.ObjectName == "" { + resources = []string{args.BucketName, args.BucketName + "/"} + } + for _, resource := range resources { + if len(st.Resources) > 0 && !st.Resources.Match(resource, args.ConditionValues) { + continue + } + if len(st.NotResources) > 0 && st.NotResources.Match(resource, args.ConditionValues) { + continue + } + return true + } + return false +} + +// authorize resolves the submitter and evaluates each request. +func (s *batchJobSubmitter) authorize(ctx context.Context, reqs ...batchAuthzRequest) error { + p, err := s.principal(ctx) + if err != nil { + return err + } + for _, req := range reqs { + if err := p.authorize(req); err != nil { + return err + } + } + return nil +} + +// startAuthzRequests are the bucket actions a job needs before it acts on any +// object: the listings it walks. Object actions are authorized per object. +func (j BatchJobRequest) startAuthzRequests() []batchAuthzRequest { + list := func(bucket string, prefixes []string) []batchAuthzRequest { + if len(prefixes) == 0 { + prefixes = []string{""} + } + var reqs []batchAuthzRequest + for _, prefix := range prefixes { + values := map[string][]string{condition.S3Prefix.Name(): {prefix}} + reqs = append(reqs, + batchAuthzRequest{action: policy.ListBucketAction, bucket: bucket, values: values}, + batchAuthzRequest{action: policy.ListBucketVersionsAction, bucket: bucket, values: values}, + ) + } + return reqs + } + // The same order as Type() and the job pool. + switch { + case j.Replicate != nil: + if !j.Replicate.RemoteToLocal() { + return list(j.Replicate.Source.Bucket, j.Replicate.Source.Prefix.F()) + } + case j.KeyRotate != nil: + return list(j.KeyRotate.Bucket, []string{j.KeyRotate.Prefix}) + case j.Expire != nil: + return list(j.Expire.Bucket, batchCanonicalPrefixes(j.Expire.Prefix.F())) + } + return nil +} + +// jobSections counts the job types a definition holds. A job is one of them. +func (j BatchJobRequest) jobSections() int { + n := 0 + for _, set := range []bool{j.Replicate != nil, j.KeyRotate != nil, j.Expire != nil} { + if set { + n++ + } + } + return n +} + +var errBatchJobSections = batchJobAuthzError{reason: "a batch job must define exactly one job type"} + +// refuse ends the job for its submitter. +func (ri *batchJobInfo) refuse(err error) { + ri.mu.Lock() + defer ri.mu.Unlock() + if ri.FailureReason == "" { + ri.FailureReason = err.Error() + } + ri.Complete = false + ri.Failed = true +} + +func (ri *batchJobInfo) refused() bool { + ri.mu.RLock() + defer ri.mu.RUnlock() + return ri.FailureReason != "" +} + +// startAuthorized checks the submitter of a job that is about to run. It +// returns done for a job that must not run now: one refused before, which +// stays refused, or one refused now, which is persisted as failed first. +// An error that is not a decision leaves the job to a later start. +func (ri *batchJobInfo) startAuthorized(ctx context.Context, api ObjectLayer, job BatchJobRequest) (done bool, err error) { + if !ri.refused() { + err := job.Submitter.authorize(ctx, job.startAuthzRequests()...) + if err == nil && job.jobSections() != 1 { + err = errBatchJobSections + } + if err == nil { + return false, nil + } + if !isBatchJobAuthzError(err) { + return true, err + } + ri.refuse(err) + batchLogIf(ctx, fmt.Errorf("batch job %s refused: %w", job.ID, err)) + } + globalBatchJobsMetrics.save(job.ID, ri) + return true, ri.updateAfter(ctx, api, 0, job) +} diff --git a/cmd/batch-authz_gen.go b/cmd/batch-authz_gen.go new file mode 100644 index 000000000..b5ebe3f64 --- /dev/null +++ b/cmd/batch-authz_gen.go @@ -0,0 +1,260 @@ +// Code generated by github.com/tinylib/msgp DO NOT EDIT. + +package cmd + +import ( + "github.com/tinylib/msgp/msgp" +) + +// DecodeMsg implements msgp.Decodable +func (z *batchJobSubmitter) DecodeMsg(dc *msgp.Reader) (err error) { + var field []byte + _ = field + var zb0001 uint32 + zb0001, err = dc.ReadMapHeader() + if err != nil { + err = msgp.WrapError(err) + return + } + for zb0001 > 0 { + zb0001-- + field, err = dc.ReadMapKeyPtr() + if err != nil { + err = msgp.WrapError(err) + return + } + switch msgp.UnsafeString(field) { + case "v": + z.Version, err = dc.ReadInt() + if err != nil { + err = msgp.WrapError(err, "Version") + return + } + case "t": + z.Type, err = dc.ReadString() + if err != nil { + err = msgp.WrapError(err, "Type") + return + } + case "ak": + z.AccessKey, err = dc.ReadString() + if err != nil { + err = msgp.WrapError(err, "AccessKey") + return + } + case "pu": + z.ParentUser, err = dc.ReadString() + if err != nil { + err = msgp.WrapError(err, "ParentUser") + return + } + case "ua": + z.UpdatedAt, err = dc.ReadTime() + if err != nil { + err = msgp.WrapError(err, "UpdatedAt") + return + } + case "rb": + z.RevokedBefore, err = dc.ReadTime() + if err != nil { + err = msgp.WrapError(err, "RevokedBefore") + return + } + case "bd": + z.Binding, err = dc.ReadBytes(z.Binding) + if err != nil { + err = msgp.WrapError(err, "Binding") + return + } + default: + err = dc.Skip() + if err != nil { + err = msgp.WrapError(err) + return + } + } + } + return +} + +// EncodeMsg implements msgp.Encodable +func (z *batchJobSubmitter) EncodeMsg(en *msgp.Writer) (err error) { + // map header, size 7 + // write "v" + err = en.Append(0x87, 0xa1, 0x76) + if err != nil { + return + } + err = en.WriteInt(z.Version) + if err != nil { + err = msgp.WrapError(err, "Version") + return + } + // write "t" + err = en.Append(0xa1, 0x74) + if err != nil { + return + } + err = en.WriteString(z.Type) + if err != nil { + err = msgp.WrapError(err, "Type") + return + } + // write "ak" + err = en.Append(0xa2, 0x61, 0x6b) + if err != nil { + return + } + err = en.WriteString(z.AccessKey) + if err != nil { + err = msgp.WrapError(err, "AccessKey") + return + } + // write "pu" + err = en.Append(0xa2, 0x70, 0x75) + if err != nil { + return + } + err = en.WriteString(z.ParentUser) + if err != nil { + err = msgp.WrapError(err, "ParentUser") + return + } + // write "ua" + err = en.Append(0xa2, 0x75, 0x61) + if err != nil { + return + } + err = en.WriteTime(z.UpdatedAt) + if err != nil { + err = msgp.WrapError(err, "UpdatedAt") + return + } + // write "rb" + err = en.Append(0xa2, 0x72, 0x62) + if err != nil { + return + } + err = en.WriteTime(z.RevokedBefore) + if err != nil { + err = msgp.WrapError(err, "RevokedBefore") + return + } + // write "bd" + err = en.Append(0xa2, 0x62, 0x64) + if err != nil { + return + } + err = en.WriteBytes(z.Binding) + if err != nil { + err = msgp.WrapError(err, "Binding") + return + } + return +} + +// MarshalMsg implements msgp.Marshaler +func (z *batchJobSubmitter) MarshalMsg(b []byte) (o []byte, err error) { + o = msgp.Require(b, z.Msgsize()) + // map header, size 7 + // string "v" + o = append(o, 0x87, 0xa1, 0x76) + o = msgp.AppendInt(o, z.Version) + // string "t" + o = append(o, 0xa1, 0x74) + o = msgp.AppendString(o, z.Type) + // string "ak" + o = append(o, 0xa2, 0x61, 0x6b) + o = msgp.AppendString(o, z.AccessKey) + // string "pu" + o = append(o, 0xa2, 0x70, 0x75) + o = msgp.AppendString(o, z.ParentUser) + // string "ua" + o = append(o, 0xa2, 0x75, 0x61) + o = msgp.AppendTime(o, z.UpdatedAt) + // string "rb" + o = append(o, 0xa2, 0x72, 0x62) + o = msgp.AppendTime(o, z.RevokedBefore) + // string "bd" + o = append(o, 0xa2, 0x62, 0x64) + o = msgp.AppendBytes(o, z.Binding) + return +} + +// UnmarshalMsg implements msgp.Unmarshaler +func (z *batchJobSubmitter) UnmarshalMsg(bts []byte) (o []byte, err error) { + var field []byte + _ = field + var zb0001 uint32 + zb0001, bts, err = msgp.ReadMapHeaderBytes(bts) + if err != nil { + err = msgp.WrapError(err) + return + } + for zb0001 > 0 { + zb0001-- + field, bts, err = msgp.ReadMapKeyZC(bts) + if err != nil { + err = msgp.WrapError(err) + return + } + switch msgp.UnsafeString(field) { + case "v": + z.Version, bts, err = msgp.ReadIntBytes(bts) + if err != nil { + err = msgp.WrapError(err, "Version") + return + } + case "t": + z.Type, bts, err = msgp.ReadStringBytes(bts) + if err != nil { + err = msgp.WrapError(err, "Type") + return + } + case "ak": + z.AccessKey, bts, err = msgp.ReadStringBytes(bts) + if err != nil { + err = msgp.WrapError(err, "AccessKey") + return + } + case "pu": + z.ParentUser, bts, err = msgp.ReadStringBytes(bts) + if err != nil { + err = msgp.WrapError(err, "ParentUser") + return + } + case "ua": + z.UpdatedAt, bts, err = msgp.ReadTimeBytes(bts) + if err != nil { + err = msgp.WrapError(err, "UpdatedAt") + return + } + case "rb": + z.RevokedBefore, bts, err = msgp.ReadTimeBytes(bts) + if err != nil { + err = msgp.WrapError(err, "RevokedBefore") + return + } + case "bd": + z.Binding, bts, err = msgp.ReadBytesBytes(bts, z.Binding) + if err != nil { + err = msgp.WrapError(err, "Binding") + return + } + default: + bts, err = msgp.Skip(bts) + if err != nil { + err = msgp.WrapError(err) + return + } + } + } + o = bts + return +} + +// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message +func (z *batchJobSubmitter) Msgsize() (s int) { + s = 1 + 2 + msgp.IntSize + 2 + msgp.StringPrefixSize + len(z.Type) + 3 + msgp.StringPrefixSize + len(z.AccessKey) + 3 + msgp.StringPrefixSize + len(z.ParentUser) + 3 + msgp.TimeSize + 3 + msgp.TimeSize + 3 + msgp.BytesPrefixSize + len(z.Binding) + return +} diff --git a/cmd/batch-authz_gen_test.go b/cmd/batch-authz_gen_test.go new file mode 100644 index 000000000..15f9cdfa1 --- /dev/null +++ b/cmd/batch-authz_gen_test.go @@ -0,0 +1,123 @@ +// Code generated by github.com/tinylib/msgp DO NOT EDIT. + +package cmd + +import ( + "bytes" + "testing" + + "github.com/tinylib/msgp/msgp" +) + +func TestMarshalUnmarshalbatchJobSubmitter(t *testing.T) { + v := batchJobSubmitter{} + bts, err := v.MarshalMsg(nil) + if err != nil { + t.Fatal(err) + } + left, err := v.UnmarshalMsg(bts) + if err != nil { + t.Fatal(err) + } + if len(left) > 0 { + t.Errorf("%d bytes left over after UnmarshalMsg(): %q", len(left), left) + } + + left, err = msgp.Skip(bts) + if err != nil { + t.Fatal(err) + } + if len(left) > 0 { + t.Errorf("%d bytes left over after Skip(): %q", len(left), left) + } +} + +func BenchmarkMarshalMsgbatchJobSubmitter(b *testing.B) { + v := batchJobSubmitter{} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + v.MarshalMsg(nil) + } +} + +func BenchmarkAppendMsgbatchJobSubmitter(b *testing.B) { + v := batchJobSubmitter{} + bts := make([]byte, 0, v.Msgsize()) + bts, _ = v.MarshalMsg(bts[0:0]) + b.SetBytes(int64(len(bts))) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + bts, _ = v.MarshalMsg(bts[0:0]) + } +} + +func BenchmarkUnmarshalbatchJobSubmitter(b *testing.B) { + v := batchJobSubmitter{} + bts, _ := v.MarshalMsg(nil) + b.ReportAllocs() + b.SetBytes(int64(len(bts))) + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, err := v.UnmarshalMsg(bts) + if err != nil { + b.Fatal(err) + } + } +} + +func TestEncodeDecodebatchJobSubmitter(t *testing.T) { + v := batchJobSubmitter{} + var buf bytes.Buffer + msgp.Encode(&buf, &v) + + m := v.Msgsize() + if buf.Len() > m { + t.Log("WARNING: TestEncodeDecodebatchJobSubmitter Msgsize() is inaccurate") + } + + vn := batchJobSubmitter{} + err := msgp.Decode(&buf, &vn) + if err != nil { + t.Error(err) + } + + buf.Reset() + msgp.Encode(&buf, &v) + err = msgp.NewReader(&buf).Skip() + if err != nil { + t.Error(err) + } +} + +func BenchmarkEncodebatchJobSubmitter(b *testing.B) { + v := batchJobSubmitter{} + var buf bytes.Buffer + msgp.Encode(&buf, &v) + b.SetBytes(int64(buf.Len())) + en := msgp.NewWriter(msgp.Nowhere) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + v.EncodeMsg(en) + } + en.Flush() +} + +func BenchmarkDecodebatchJobSubmitter(b *testing.B) { + v := batchJobSubmitter{} + var buf bytes.Buffer + msgp.Encode(&buf, &v) + b.SetBytes(int64(buf.Len())) + rd := msgp.NewEndlessReader(buf.Bytes(), b) + dc := msgp.NewReader(rd) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + err := v.DecodeMsg(dc) + if err != nil { + b.Fatal(err) + } + } +} diff --git a/cmd/batch-expire.go b/cmd/batch-expire.go index 17dcf556f..3f8b1e15a 100644 --- a/cmd/batch-expire.go +++ b/cmd/batch-expire.go @@ -27,6 +27,7 @@ import ( "net/http" "runtime" "strconv" + "sync/atomic" "time" "github.com/minio/minio-go/v7/pkg/tags" @@ -34,6 +35,7 @@ import ( xhttp "github.com/minio/minio/internal/http" xioutil "github.com/minio/minio/internal/ioutil" "github.com/pgsty/silo-pkg/v3/env" + "github.com/pgsty/silo-pkg/v3/policy" "github.com/pgsty/silo-pkg/v3/wildcard" "github.com/pgsty/silo-pkg/v3/workers" "github.com/pgsty/silo-pkg/v3/xtime" @@ -289,16 +291,6 @@ type BatchJobExpire struct { var _ yaml.Unmarshaler = &BatchJobExpire{} -// RedactSensitive will redact any sensitive information in b. -func (r *BatchJobExpire) RedactSensitive() { - if r == nil { - return - } - if r.NotificationCfg.Token != "" { - r.NotificationCfg.Token = redactedText - } -} - // UnmarshalYAML - BatchJobExpire extends default unmarshal to extract line, col information. func (r *BatchJobExpire) UnmarshalYAML(val *yaml.Node) error { type expireJob BatchJobExpire @@ -345,32 +337,6 @@ func (r BatchJobExpire) Notify(ctx context.Context, body io.Reader) error { return nil } -// Expire expires object versions which have already matched supplied filter conditions -func (r *BatchJobExpire) Expire(ctx context.Context, api ObjectLayer, vc *versioning.Versioning, objsToDel []ObjectToDelete) []error { - opts := ObjectOptions{ - PrefixEnabledFn: vc.PrefixEnabled, - VersionSuspended: vc.Suspended(), - } - - allErrs := make([]error, 0, len(objsToDel)) - - for { - count := len(objsToDel) - if count == 0 { - break - } - if count > maxDeleteList { - count = maxDeleteList - } - _, errs := api.DeleteObjects(ctx, r.Bucket, objsToDel[:count], opts) - allErrs = append(allErrs, errs...) - // Next batch of deletion - objsToDel = objsToDel[count:] - } - - return allErrs -} - const ( batchExpireName = "batch-expire.bin" batchExpireFormat = 1 @@ -381,22 +347,238 @@ const ( batchExpireJobDefaultRetryDelay = 250 * time.Millisecond ) -type objInfoCache map[string]*ObjectInfo - -func newObjInfoCache() objInfoCache { - return objInfoCache(make(map[string]*ObjectInfo)) +// expireVersionRef is a version of a key whose versions are all purged, as +// the walk saw it. +type expireVersionRef struct { + VersionID string + UserTags string + DeleteMarker bool + Locked bool } -func (oiCache objInfoCache) Add(toDel ObjectToDelete, oi *ObjectInfo) { - oiCache[fmt.Sprintf("%s-%s", toDel.ObjectName, toDel.VersionID)] = oi +type expireObjInfo struct { + ObjectInfo + ExpireAll bool + DeleteMarkerCount int64 + // versions are the versions of an ExpireAll key the walk saw, the + // latest first. + versions []expireVersionRef } -func (oiCache objInfoCache) Get(toDel ObjectToDelete) (*ObjectInfo, bool) { - oi, ok := oiCache[fmt.Sprintf("%s-%s", toDel.ObjectName, toDel.VersionID)] - return oi, ok +// isErrObjectLocked reports a version that Object Lock protects. +func isErrObjectLocked(err error) bool { + var locked ObjectLocked + return errors.As(err, &locked) } -func batchObjsForDelete(ctx context.Context, r *BatchJobExpire, ri *batchJobInfo, job BatchJobRequest, api ObjectLayer, wk *workers.Workers, expireCh <-chan []expireObjInfo) { +func (r *BatchJobExpire) tagsFiltered() bool { + for _, rule := range r.Rules { + if len(rule.Tags) > 0 { + return true + } + } + return false +} + +// expireAuthzRequests are the actions the job takes on one version: it reads +// the version, and its tags when a rule filters by them, and deletes it. +func (r *BatchJobExpire) expireAuthzRequests(name, versionID, userTags string) []batchAuthzRequest { + if versionID == "" { + versionID = nullVersionID + } + reqs := []batchAuthzRequest{ + {action: policy.GetObjectAction, bucket: r.Bucket, object: name, versionID: versionID, tags: userTags}, + {action: policy.GetObjectVersionAction, bucket: r.Bucket, object: name, versionID: versionID, tags: userTags}, + {action: policy.DeleteObjectAction, bucket: r.Bucket, object: name, versionID: versionID, tags: userTags}, + {action: policy.DeleteObjectVersionAction, bucket: r.Bucket, object: name, versionID: versionID, tags: userTags}, + } + if r.tagsFiltered() { + reqs = append(reqs, + batchAuthzRequest{action: policy.GetObjectTaggingAction, bucket: r.Bucket, object: name, versionID: versionID, tags: userTags}, + batchAuthzRequest{action: policy.GetObjectVersionTaggingAction, bucket: r.Bucket, object: name, versionID: versionID, tags: userTags}, + ) + } + return reqs +} + +// expireVersion deletes one version that is older than the versions the rule +// retains. Under the object's write lock, with the version as it is then, it +// leaves a version that Object Lock protects (ObjectLocked) and refuses one +// that the submitter may not delete (batchJobAuthzError). An unreadable +// version is not deleted. +func (r *BatchJobExpire) expireVersion(ctx context.Context, api ObjectLayer, vc *versioning.Versioning, job BatchJobRequest, exp expireObjInfo) error { + p, err := job.Submitter.principal(ctx) + if err != nil { + return err + } + versionID := exp.VersionID + if versionID == "" { + // An empty version ID would add a delete marker instead. + versionID = nullVersionID + } + opts := ObjectOptions{ + VersionID: versionID, + Versioned: vc.PrefixEnabled(exp.Name), + VersionSuspended: vc.Suspended(), + } + opts.SetEvalRetentionBypassFn(func(goi ObjectInfo, gerr error) error { + // A delete marker reads as MethodNotAllowed; any other error, + // including a missing version, stops the delete. + marker := isErrMethodNotAllowed(gerr) && (goi.DeleteMarker || !goi.VersionPurgeStatus.Empty()) + if gerr != nil && !marker { + return gerr + } + if enforceRetentionForDeletion(ctx, goi) { + return ObjectLocked{Bucket: r.Bucket, Object: exp.Name, VersionID: versionID} + } + for _, req := range r.expireAuthzRequests(exp.Name, versionID, goi.UserTags) { + if err := p.authorize(req); err != nil { + return err + } + } + return nil + }) + _, err = api.DeleteObject(ctx, r.Bucket, exp.Name, opts) + return err +} + +// expireVersions deletes the versions of one batch entry and records the +// outcome in res. It returns an error that ends the job. +func (r *BatchJobExpire) expireVersions(ctx context.Context, api ObjectLayer, vc *versioning.Versioning, job BatchJobRequest, ri *batchJobInfo, exp expireObjInfo, retryAttempts int, delay time.Duration, res *batchCounts) error { + for attempts := 1; attempts <= retryAttempts; attempts++ { + if err := ctx.Err(); err != nil { + return err + } + stopFn := globalBatchJobsMetrics.trace(batchJobMetricExpire, ri.JobID, attempts) + err := r.expireVersion(ctx, api, vc, job, exp) + stopFn(exp, err) + switch { + case err == nil: + if exp.DeleteMarker { + ri.countRemoved(r.Bucket, 0, 1) + } else { + ri.countRemoved(r.Bucket, 1, 0) + } + return nil + case isBatchJobAuthzError(err): + return err + case isErrObjectLocked(err): + // Left in place, not retried and not a failure. + res.versionsSkipped++ + return nil + case isErrVersionNotFound(err) || isErrObjectNotFound(err): + // Removed concurrently. + return nil + } + batchLogIf(ctx, fmt.Errorf("Failed to expire %s/%s versionID=%s due to %v (attempts=%d)", r.Bucket, exp.Name, exp.VersionID, err, attempts)) + if attempts < retryAttempts { + time.Sleep(delay) + } + } + if exp.DeleteMarker { + res.deleteMarkersFailed++ + } else { + res.objectsFailed++ + } + return nil +} + +// expireAllVersions purges every version of a key whose rule retains none, +// with the purge of exactly that key (erasureServerPools.deleteAllVersions), +// while the version the walk matched is still its latest. What the purge +// removed is counted at once; a key it refused because Object Lock protects +// a version is counted in res as refused, and the versions a failed purge +// left in res as failures. A failed purge is retried up to retryAttempts +// times, each time with the version the walk matched; a purge that stopped +// at its budget after removing versions continues without spending an +// attempt. A key whose latest version changed since the walk is not +// expired. It returns an error that ends the job. +// +// TODO(rebase onto the delete-all purge, claude/fix-delete-all): before the +// purge removes anything, authorize every version of the history it +// approved under its lock, with version-specific conditions +// (expireAuthzRequests). Until then the checks here use the versions the +// walk saw, and do not cover a version added after the walk; the purge +// itself refuses a key that holds a locked version. +func (r *BatchJobExpire) expireAllVersions(ctx context.Context, api ObjectLayer, job BatchJobRequest, ri *batchJobInfo, exp expireObjInfo, retryAttempts int, res *batchCounts) error { + for _, v := range exp.versions { + if v.Locked { + res.keysRefused++ + return nil + } + } + var reqs []batchAuthzRequest + for _, v := range exp.versions { + reqs = append(reqs, r.expireAuthzRequests(exp.Name, v.VersionID, v.UserTags)...) + } + if err := job.Submitter.authorize(ctx, reqs...); err != nil { + return err + } + var objects, markers int64 + matched := deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime} + for attempts := 1; attempts <= retryAttempts; attempts++ { + if err := ctx.Err(); err != nil { + return err + } + stopFn := globalBatchJobsMetrics.trace(batchJobMetricExpire, ri.JobID, attempts) + latest := matched + var purge deleteAllResult + _, err := api.DeleteObject(ctx, exp.Bucket, encodeDirObject(exp.Name), ObjectOptions{ + DeleteAllVersions: true, + DeleteAllLatest: &latest, + DeleteAllResult: &purge, + }) + // Removed versions are gone for any later run: count them now. + if purge.Removed > 0 { + ri.countRemoved(r.Bucket, int64(purge.Removed-purge.Markers), int64(purge.Markers)) + objects += int64(purge.Removed - purge.Markers) + markers += int64(purge.Markers) + } + stopFn(exp, err) + switch { + case err == nil: + return nil + case errors.Is(err, errDeleteAllChanged): + // Changed since the walk matched it: not expired. + return nil + case isErrObjectLocked(err): + // Refused as a whole: nothing removed, not retried and not a + // failure. + res.keysRefused++ + return nil + case errors.Is(err, errDeleteAllBudget) && purge.Removed > 0: + attempts-- // progress + continue + } + if err := ctx.Err(); err != nil { + return err + } + batchLogIf(ctx, fmt.Errorf("Failed to expire %s/%s due to %v (attempts=%d)", exp.Bucket, exp.Name, err, attempts)) + if errors.Is(err, errDeleteAllTiered) { + break // refused; a later run of the job checks again + } + } + res.expireAllFailed(exp, objects, markers) + return nil +} + +// expireAllFailed counts the versions of info, as the walk saw them, that a +// failed purge of all its versions did not remove: those beyond the objects +// and markers it removed, at least one. +func (c *batchCounts) expireAllFailed(info expireObjInfo, objects, markers int64) { + failedObjects := max(int64(info.NumVersions)-info.DeleteMarkerCount-objects, 0) + failedMarkers := max(info.DeleteMarkerCount-markers, 0) + if failedObjects+failedMarkers == 0 { + failedObjects = 1 + } + c.objectsFailed += failedObjects + c.deleteMarkersFailed += failedMarkers +} + +// batchObjsForDelete expires the batches of expireCh. walkDone reports, once +// expireCh is closed, whether the walk reached its end: only then is its last +// key known to be complete. +func batchObjsForDelete(ctx context.Context, refuse func(error), walkDone func() bool, r *BatchJobExpire, ri *batchJobInfo, job BatchJobRequest, api ObjectLayer, wk *workers.Workers, expireCh <-chan []expireObjInfo) { vc, _ := globalBucketVersioningSys.Get(r.Bucket) retryAttempts := job.Expire.Retry.Attempts if retryAttempts <= 0 { @@ -407,162 +589,53 @@ func batchObjsForDelete(ctx context.Context, r *BatchJobExpire, ri *batchJobInfo delay = batchExpireJobDefaultRetryDelay } - var i int + progress := newBatchProgress(ri, r.Bucket) + defer func() { + wk.Wait() + if walkDone() { + progress.finish() + } + }() for toExpire := range expireCh { select { case <-ctx.Done(): return default: } - if i > 0 { + seq := progress.begin() + if seq > 0 { if wait := globalBatchConfig.ExpirationWait(); wait > 0 { time.Sleep(wait) } } - i++ wk.Take() - go func(toExpire []expireObjInfo) { + go func(seq uint64, toExpire []expireObjInfo) { defer wk.Give() - - toExpireAll := make([]expireObjInfo, 0, len(toExpire)) - toDel := make([]ObjectToDelete, 0, len(toExpire)) - oiCache := newObjInfoCache() + var res batchCounts for _, exp := range toExpire { + var err error if exp.ExpireAll { - toExpireAll = append(toExpireAll, exp) - continue + err = r.expireAllVersions(ctx, api, job, ri, exp, retryAttempts, &res) + } else { + err = r.expireVersions(ctx, api, vc, job, ri, exp, retryAttempts, delay, &res) } - // Cache ObjectInfo value via pointers for - // subsequent use to track objects which - // couldn't be deleted. - od := ObjectToDelete{ - ObjectV: ObjectV{ - ObjectName: exp.Name, - VersionID: exp.VersionID, - }, - } - toDel = append(toDel, od) - oiCache.Add(od, &exp.ObjectInfo) - } - - // Purge all versions of exactly each key while the version the - // walk matched is still its latest; see - // erasureServerPools.deleteAllVersions. - for _, exp := range toExpireAll { - if !expireAllVersions(ctx, ri, api, exp, retryAttempts) { - return - } - } - - // DeleteMultiple objects - toDelCopy := make([]ObjectToDelete, len(toDel)) - for attempts := 1; attempts <= retryAttempts; attempts++ { - select { - case <-ctx.Done(): - return - default: - } - - stopFn := globalBatchJobsMetrics.trace(batchJobMetricExpire, ri.JobID, attempts) - // Copying toDel to select from objects whose - // deletion failed - copy(toDelCopy, toDel) - var failed int - errs := r.Expire(ctx, api, vc, toDel) - // reslice toDel in preparation for next retry attempt - toDel = toDel[:0] - for i, err := range errs { - if err != nil { - stopFn(toDelCopy[i], err) - batchLogIf(ctx, fmt.Errorf("Failed to expire %s/%s versionID=%s due to %v (attempts=%d)", ri.Bucket, toDelCopy[i].ObjectName, toDelCopy[i].VersionID, - err, attempts)) - failed++ - if oi, ok := oiCache.Get(toDelCopy[i]); ok { - ri.trackCurrentBucketObject(r.Bucket, *oi, false, attempts) - } - if attempts != retryAttempts { - // retry - toDel = append(toDel, toDelCopy[i]) - } - } else { - stopFn(toDelCopy[i], nil) - if oi, ok := oiCache.Get(toDelCopy[i]); ok { - ri.trackCurrentBucketObject(r.Bucket, *oi, true, attempts) - } + if err != nil { + // The batch is not done; a later run does it again. + if isBatchJobAuthzError(err) { + refuse(err) } - } - - globalBatchJobsMetrics.save(ri.JobID, ri) - - if failed == 0 { - break - } - - // Add a delay between retry attempts - if attempts < retryAttempts { - time.Sleep(delay) + globalBatchJobsMetrics.save(ri.JobID, ri) + return } } - }(toExpire) + if n := len(toExpire); n > 0 { + progress.done(seq, toExpire[n-1].Name, res) + } + globalBatchJobsMetrics.save(ri.JobID, ri) + }(seq, toExpire) } } -// expireAllVersions purges all versions of the key exp, which the walk -// matched by its latest version, and counts what the purge removed. It -// retries a failed purge up to attempts times, each time with the version -// the walk matched; a purge that stopped at its budget continues without -// spending an attempt. It does not retry once the key no longer has that -// latest version, which is not an expiry, or once the purge refused the -// key. It returns false once ctx ends. -func expireAllVersions(ctx context.Context, ri *batchJobInfo, api ObjectLayer, exp expireObjInfo, attempts int) bool { - var objects, markers int64 - matched := deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime} - for attempt := 1; attempt <= attempts; attempt++ { - select { - case <-ctx.Done(): - ri.trackExpireAll(exp, objects, markers, true) - return false - default: - } - stopFn := globalBatchJobsMetrics.trace(batchJobMetricExpire, ri.JobID, attempt) - latest := matched - var purge deleteAllResult - _, err := api.DeleteObject(ctx, exp.Bucket, encodeDirObject(exp.Name), ObjectOptions{ - DeleteAllVersions: true, - DeleteAllLatest: &latest, - DeleteAllResult: &purge, - }) - objects += int64(purge.Removed - purge.Markers) - markers += int64(purge.Markers) - stopFn(exp, err) - switch { - case err == nil: - ri.trackExpireAll(exp, objects, markers, false) - return true - case errors.Is(err, errDeleteAllChanged): - // Changed since the walk matched it: not expired. - ri.trackExpireAll(exp, objects, markers, false) - return true - case errors.Is(err, errDeleteAllBudget) && purge.Removed > 0: - attempt-- // progress - continue - } - batchLogIf(ctx, fmt.Errorf("Failed to expire %s/%s due to %v (attempts=%d)", exp.Bucket, exp.Name, err, attempt)) - var locked ObjectLocked - if errors.Is(err, errDeleteAllTiered) || errors.As(err, &locked) { - break // refused; a later run of the job checks again - } - } - ri.trackExpireAll(exp, objects, markers, true) - return true -} - -type expireObjInfo struct { - ObjectInfo - ExpireAll bool - DeleteMarkerCount int64 -} - // Start the batch expiration job, resumes if there was a pending job via "job.ID" func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJobRequest) error { ri := &batchJobInfo{ @@ -573,6 +646,9 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo if err := ri.loadOrInit(ctx, api, job); err != nil { return err } + if done, err := ri.startAuthorized(ctx, api, job); done || err != nil { + return err + } globalBatchJobsMetrics.save(job.ID, ri) lastObject := ri.Object @@ -592,27 +668,43 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo ctx, cancelCause := context.WithCancelCause(ctx) defer cancelCause(nil) + // refuse ends the job for its submitter: record why before stopping. + refuse := func(err error) { + ri.refuse(err) + cancelCause(err) + } + // The walk stops at a listing failure: work after it is left to a later + // run, so the cursor cannot move past what was not listed. + walkCtx, stopWalk := context.WithCancel(ctx) + defer stopWalk() results := make(chan itemOrErr[ObjectInfo], workerSize) go func() { - prefixes := r.Prefix.F() - if len(prefixes) == 0 { - prefixes = []string{""} - } - for _, prefix := range prefixes { + // One ascending walk over disjoint prefixes, which the cursor + // describes. + for _, prefix := range batchCanonicalPrefixes(r.Prefix.F()) { + if walkCtx.Err() != nil { + break + } prefixResultCh := make(chan itemOrErr[ObjectInfo], workerSize) - err := api.Walk(ctx, r.Bucket, prefix, prefixResultCh, WalkOptions{ + err := api.Walk(walkCtx, r.Bucket, prefix, prefixResultCh, WalkOptions{ Marker: lastObject, LatestOnly: false, // we need to visit all versions of the object to implement purge: retainVersions VersionsSort: WalkVersionsSortDesc, }) if err != nil { - cancelCause(err) + if walkCtx.Err() == nil { + cancelCause(err) + } xioutil.SafeClose(results) return } for result := range prefixResultCh { - results <- result + select { + case results <- result: + case <-walkCtx.Done(): + // Stopped: let the walk end without a reader. + } } } xioutil.SafeClose(results) @@ -620,7 +712,9 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo // Goroutine to periodically save batch-expire job's in-memory state saverQuitCh := make(chan struct{}) + saverDoneCh := make(chan struct{}) go func() { + defer close(saverDoneCh) saveTicker := time.NewTicker(10 * time.Second) defer saveTicker.Stop() quit := false @@ -647,9 +741,10 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo expireCh := make(chan []expireObjInfo, workerSize) expireDoneCh := make(chan struct{}) + var walkDone atomic.Bool go func() { defer close(expireDoneCh) - batchObjsForDelete(ctx, r, ri, job, api, wk, expireCh) + batchObjsForDelete(ctx, refuse, walkDone.Load, r, ri, job, api, wk, expireCh) }() var ( @@ -661,6 +756,8 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo done bool ) deleteMarkerCountMap := map[string]int64{} + // pushToExpire is called between keys only, so that a batch holds all + // the versions of its keys and its last key is done with it. pushToExpire := func() { // set preObject deleteMarkerCount if len(toDel) > 0 { @@ -695,6 +792,12 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo if result.Err != nil { failed = true batchLogIf(ctx, result.Err) + stopWalk() + done = true + break + } + if !batchResumeAfter(lastObject, result.Item.Name) { + // Completed by the run that set the cursor. continue } if result.Item.DeleteMarker { @@ -730,12 +833,17 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo toDel = append(toDel, expireObjInfo{ ObjectInfo: result.Item, ExpireAll: true, + versions: []expireVersionRef{newExpireVersionRef(ctx, result.Item)}, }) continue } } else if prevObj.Name == result.Item.Name { if matchedFilter.Purge.RetainVersions == 0 { - continue // including latest version in toDel suffices, skipping other versions + // The latest version in toDel purges them all. + if n := len(toDel); n > 0 && toDel[n-1].Name == result.Item.Name { + toDel[n-1].versions = append(toDel[n-1].versions, newExpireVersionRef(ctx, result.Item)) + } + continue } versionsCount++ } else { @@ -752,7 +860,6 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo toDel = append(toDel, expireObjInfo{ ObjectInfo: result.Item, }) - pushToExpire() case <-ctx.Done(): done = true } @@ -761,16 +868,15 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo } } - if context.Cause(ctx) != nil { - xioutil.SafeClose(expireCh) - return context.Cause(ctx) - } - pushToExpire() - // Send any remaining objects downstream - if len(toDel) > 0 { - select { - case <-ctx.Done(): - case expireCh <- toDel: + if context.Cause(ctx) == nil { + walkDone.Store(!failed) + pushToExpire() + // Send any remaining objects downstream + if len(toDel) > 0 { + select { + case <-ctx.Done(): + case expireCh <- toDel: + } } } xioutil.SafeClose(expireCh) @@ -778,13 +884,29 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo <-expireDoneCh // waits for the expire goroutine to complete wk.Wait() // waits for all expire workers to retire - ri.Complete = !failed && !ri.expireFailures() - ri.Failed = failed || ri.expireFailures() - globalBatchJobsMetrics.save(job.ID, ri) - - // Close the saverQuitCh - this also triggers saving in-memory state - // immediately one last time before we exit this method. + // Persist the final state once the saver has stopped, so that its last + // save cannot overwrite it. xioutil.SafeClose(saverQuitCh) + <-saverDoneCh + refused := ri.refused() + cause := context.Cause(ctx) + if cause != nil && !refused { + saveCtx, cancel := context.WithTimeout(GlobalContext, 30*time.Second) + batchLogIf(saveCtx, ri.updateAfter(saveCtx, api, 0, job)) + cancel() + globalBatchJobsMetrics.save(job.ID, ri) + return cause + } + + ri.mu.Lock() + itemsFailed := ri.expireFailures() + ri.Complete = !refused && !failed && !itemsFailed + ri.Failed = refused || failed || itemsFailed + ri.mu.Unlock() + globalBatchJobsMetrics.save(job.ID, ri) + saveCtx, cancel := context.WithTimeout(GlobalContext, 30*time.Second) + batchLogIf(saveCtx, ri.updateAfter(saveCtx, api, 0, job)) + cancel() // Notify expire jobs final status to the configured endpoint buf, _ := json.Marshal(ri) @@ -795,6 +917,15 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo return nil } +func newExpireVersionRef(ctx context.Context, oi ObjectInfo) expireVersionRef { + return expireVersionRef{ + VersionID: oi.VersionID, + UserTags: oi.UserTags, + DeleteMarker: oi.DeleteMarker, + Locked: enforceRetentionForDeletion(ctx, oi), + } +} + //msgp:ignore batchExpireJobError type batchExpireJobError struct { Code string diff --git a/cmd/batch-handlers.go b/cmd/batch-handlers.go index dba3c6a20..28985f728 100644 --- a/cmd/batch-handlers.go +++ b/cmd/batch-handlers.go @@ -43,6 +43,8 @@ import ( "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/amztime" + objectlock "github.com/minio/minio/internal/bucket/object/lock" "github.com/minio/minio/internal/config/batch" "github.com/minio/minio/internal/crypto" "github.com/minio/minio/internal/hash" @@ -72,32 +74,47 @@ type BatchJobRequest struct { Replicate *BatchJobReplicateV1 `yaml:"replicate" json:"replicate"` KeyRotate *BatchJobKeyRotateV1 `yaml:"keyrotate" json:"keyrotate"` Expire *BatchJobExpire `yaml:"expire" json:"expire"` - ctx context.Context `msg:"-"` + // Submitter is the credential the job acts as. The server assigns it + // at submission; a job definition cannot name it. + Submitter *batchJobSubmitter `yaml:"-" json:"-"` + ctx context.Context `msg:"-"` } -// RedactSensitive will redact any sensitive information in b. -func (j *BatchJobRequest) RedactSensitive() { - j.Replicate.RedactSensitive() - j.Expire.RedactSensitive() - j.KeyRotate.RedactSensitive() +// redacted returns a copy of the job for output outside the server: the +// describe response, logs and errors. The copy has no secret keys, session +// tokens or notification tokens, and no submitter binding; the job itself, +// which a run needs as it is, is not changed. +func (j BatchJobRequest) redacted() (BatchJobRequest, error) { + buf, err := j.MarshalMsg(nil) + if err != nil { + return BatchJobRequest{}, err + } + var c BatchJobRequest + if _, err = c.UnmarshalMsg(buf); err != nil { + return BatchJobRequest{}, err + } + redact := func(s *string) { + if *s != "" { + *s = redactedText + } + } + c.Submitter = nil + if r := c.Replicate; r != nil { + redact(&r.Source.Creds.SecretKey) + redact(&r.Source.Creds.SessionToken) + redact(&r.Target.Creds.SecretKey) + redact(&r.Target.Creds.SessionToken) + redact(&r.Flags.Notify.Token) + } + if r := c.KeyRotate; r != nil { + redact(&r.Flags.Notify.Token) + } + if r := c.Expire; r != nil { + redact(&r.NotificationCfg.Token) + } + return c, nil } -// RedactSensitive will redact any sensitive information in b. -func (r *BatchJobReplicateV1) RedactSensitive() { - if r == nil { - return - } - if r.Target.Creds.SecretKey != "" { - r.Target.Creds.SecretKey = redactedText - } - if r.Target.Creds.SessionToken != "" { - r.Target.Creds.SessionToken = redactedText - } -} - -// RedactSensitive will redact any sensitive information in b. -func (r *BatchJobKeyRotateV1) RedactSensitive() {} - func notifyEndpoint(ctx context.Context, ri *batchJobInfo, endpoint, token string) error { if endpoint == "" { return nil @@ -140,8 +157,114 @@ 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 *minio.Core, srcObjInfo ObjectInfo, retry bool) error { +// batchPullObjectLock prepares the Object Lock metadata that a pull writes +// with a version, as a PUT would on the destination: Object Lock metadata +// copied from the source must be valid, a version without retention takes the +// destination bucket's default retention, and both are dated with the +// destination's clock. Outside a lock-enabled bucket the metadata has no +// meaning and is not written. It reports the retention and legal hold that +// the pull copies from the source. +func batchPullObjectLock(bucket string, meta map[string]string) (retention, legalHold bool, err error) { + cfg, err := globalBucketObjectLockSys.Get(bucket) + if err != nil { + return false, false, err + } + h := make(http.Header) + for k, v := range meta { + for _, name := range []string{xhttp.AmzObjectLockMode, xhttp.AmzObjectLockRetainUntilDate, xhttp.AmzObjectLockLegalHold} { + if strings.EqualFold(k, name) { + h.Set(name, v) + delete(meta, k) + } + } + } + if !cfg.LockEnabled { + return false, false, nil + } + if h.Get(xhttp.AmzObjectLockMode) != "" || h.Get(xhttp.AmzObjectLockRetainUntilDate) != "" { + // A copied retention may have expired already. + mode, until, err := objectlock.ParseObjectLockRetentionHeaders(h, true) + if err != nil { + return false, false, err + } + meta[strings.ToLower(xhttp.AmzObjectLockMode)] = string(mode) + meta[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(until.UTC()) + retention = true + } else if cfg.Validity > 0 { + now, err := objectlock.UTCNowNTP() + if err != nil { + return false, false, err + } + meta[strings.ToLower(xhttp.AmzObjectLockMode)] = string(cfg.Mode) + meta[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(now.Add(cfg.Validity).UTC()) + } + if _, ok := h[xhttp.AmzObjectLockLegalHold]; ok { + hold, err := objectlock.ParseObjectLockLegalHoldHeaders(h) + if err != nil { + return false, false, err + } + meta[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(hold.Status) + legalHold = true + } + return retention, legalHold, nil +} + +// pullAuthzRequests are the writes of one version that a pull makes to the +// destination key: the version, with the tags, retention and legal hold it +// copies, and the replication permission for a version ID it keeps. +// +// The pull does not encrypt what it writes, so it supplies no encryption +// values, whatever the source reports; and it does not read the destination +// version it may replace, so the tags of that version are unknown. +func pullAuthzRequests(bucket, object, versionID string, keepsVersion bool, opts ObjectOptions, retention, legalHold bool) []batchAuthzRequest { + values := map[string][]string{} + meta := caseInsensitiveMap(opts.UserDefined) + if retention { + mode, _ := meta.Lookup(xhttp.AmzObjectLockMode) + until, _ := meta.Lookup(xhttp.AmzObjectLockRetainUntilDate) + values[strings.TrimPrefix(strings.ToLower(xhttp.AmzObjectLockMode), "x-amz-")] = []string{mode} + values[strings.TrimPrefix(strings.ToLower(xhttp.AmzObjectLockRetainUntilDate), "x-amz-")] = []string{until} + } + if legalHold { + hold, _ := meta.Lookup(xhttp.AmzObjectLockLegalHold) + values[strings.TrimPrefix(strings.ToLower(xhttp.AmzObjectLockLegalHold), "x-amz-")] = []string{hold} + } + tagStr, hasTags := meta.Lookup(xhttp.AmzObjectTagging) + if hasTags && tagStr != "" { + if t, err := tags.ParseObjectTags(tagStr); err == nil { + m := t.ToMap() + keys := make([]string, 0, len(m)) + for k, v := range m { + values[pathJoin("RequestObjectTag", k)] = []string{v} + keys = append(keys, k) + } + values["RequestObjectTagKeys"] = keys + } + } + actions := []policy.Action{policy.PutObjectAction} + if keepsVersion { + actions = append(actions, policy.ReplicateObjectAction) + } + if hasTags && tagStr != "" { + actions = append(actions, policy.PutObjectTaggingAction) + } + if retention { + actions = append(actions, policy.PutObjectRetentionAction) + } + if legalHold { + actions = append(actions, policy.PutObjectLegalHoldAction) + } + reqs := make([]batchAuthzRequest, 0, len(actions)) + for _, action := range actions { + reqs = append(reqs, batchAuthzRequest{action: action, bucket: bucket, object: object, versionID: versionID, tagsUnknown: true, values: values}) + } + return reqs +} + +// ReplicateFromSource copies one version from the remote source to the local +// target. A write that would replace a version protected by Object Lock +// returns ObjectLocked and leaves it. +func (r *BatchJobReplicateV1) ReplicateFromSource(ctx context.Context, api ObjectLayer, core *minio.Core, srcObjInfo ObjectInfo, retry bool, sub *batchJobSubmitter) error { srcBucket := r.Source.Bucket tgtBucket := r.Target.Bucket srcObject := srcObjInfo.Name @@ -150,11 +273,28 @@ func (r *BatchJobReplicateV1) ReplicateFromSource(ctx context.Context, api Objec tgtObject = pathJoin(r.Target.Prefix, srcObjInfo.Name) } + p, err := sub.principal(ctx) + if err != nil { + return err + } versionID := srcObjInfo.VersionID if r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3 { versionID = "" } if srcObjInfo.DeleteMarker { + markerVersion := versionID + if markerVersion == "" { + markerVersion = nullVersionID + } + for _, req := range []batchAuthzRequest{ + {action: policy.DeleteObjectAction, bucket: tgtBucket, object: tgtObject, versionID: markerVersion, tagsUnknown: true}, + {action: policy.DeleteObjectVersionAction, bucket: tgtBucket, object: tgtObject, versionID: markerVersion, tagsUnknown: true}, + {action: policy.ReplicateDeleteAction, bucket: tgtBucket, object: tgtObject, versionID: markerVersion, tagsUnknown: true}, + } { + if err := p.authorize(req); err != nil { + return err + } + } _, 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. @@ -164,22 +304,43 @@ func (r *BatchJobReplicateV1) ReplicateFromSource(ctx context.Context, api Objec MTime: srcObjInfo.ModTime, DeleteMarker: srcObjInfo.DeleteMarker, ReplicationRequest: true, + // A new marker may hide a locked version; one that would replace + // it is refused. + RefuseLockedOverwrite: true, }) return err } - opts := ObjectOptions{ - VersionID: srcObjInfo.VersionID, - MTime: srcObjInfo.ModTime, - PreserveETag: srcObjInfo.ETag, - UserDefined: srcObjInfo.UserDefined, + userDefined := cloneMSS(srcObjInfo.UserDefined) + retention, legalHold, err := batchPullObjectLock(tgtBucket, userDefined) + if err != nil { + return err } - if r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3 { - opts.VersionID = "" + opts := ObjectOptions{ + VersionID: versionID, + Versioned: globalBucketVersioningSys.PrefixEnabled(tgtBucket, tgtObject), + VersionSuspended: globalBucketVersioningSys.PrefixSuspended(tgtBucket, tgtObject), + MTime: srcObjInfo.ModTime, + PreserveETag: srcObjInfo.ETag, + UserDefined: userDefined, + // Checked with the commit, under the destination's write lock. + RefuseLockedOverwrite: true, } if crypto.S3.IsEncrypted(srcObjInfo.UserDefined) { opts.ServerSideEncryption = encrypt.NewSSE() } + // The version the write replaces or creates: the source's version ID it + // keeps, or the null version of a destination that does not version + // the key. A versioned destination gives a write without an ID a new one. + writeVersion := versionID + if writeVersion == "" && !opts.Versioned { + writeVersion = nullVersionID + } + for _, req := range pullAuthzRequests(tgtBucket, tgtObject, writeVersion, versionID != "", opts, retention, legalHold) { + if err := p.authorize(req); err != nil { + return err + } + } slc := strings.Split(srcObjInfo.ETag, "-") if len(slc) == 2 { partsCount, err := strconv.Atoi(slc[1]) @@ -298,6 +459,9 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay if ri.Complete { return nil } + if done, err := ri.startAuthorized(ctx, api, job); done || err != nil { + return err + } globalBatchJobsMetrics.save(job.ID, ri) retryAttempts := job.Replicate.Flags.Retry.Attempts @@ -411,7 +575,17 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay // 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) + ctx, cancelCause := context.WithCancelCause(ctx) + // refuse ends the job for its submitter: record why before stopping. + refuse := func(err error) { + ri.refuse(err) + cancelCause(err) + } + // Each run lists the source from the start and counts every locked + // collision again. + ri.mu.Lock() + ri.VersionsSkippedLocked = 0 + ri.mu.Unlock() objInfoCh := make(chan minio.ObjectInfo, 1) go func() { @@ -437,6 +611,10 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay skipReplicate := false for obj := range objInfoCh { + if context.Cause(ctx) != nil { + // Refused or canceled: schedule nothing more. + continue + } oi := toObjectInfo(r.Source.Bucket, obj.Key, obj) if !minioSrc { // Check if metadata filter was requested and it is expected to have @@ -484,7 +662,19 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay defer wk.Give() stopFn := globalBatchJobsMetrics.trace(batchJobMetricReplication, job.ID, attempts) success := true - if err := r.ReplicateFromSource(ctx, api, core, oi, retry); err != nil { + if err := r.ReplicateFromSource(ctx, api, core, oi, retry, job.Submitter); err != nil { + if isBatchJobAuthzError(err) { + refuse(err) + return + } + // A locked destination version is left as it is: the + // job completes with it skipped, not replicated. + if isErrObjectLocked(err) { + stopFn(oi, err) + ri.countSkipped() + globalBatchJobsMetrics.save(job.ID, ri) + return + } // object must be deleted concurrently, allow these failures but do not count them if isErrVersionNotFound(err) || isErrObjectNotFound(err) { return @@ -495,7 +685,7 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay } else { stopFn(oi, nil) } - ri.trackCurrentBucketObject(r.Target.Bucket, oi, success, attempts) + ri.trackCurrentBucketObject(r.Target.Bucket, oi, success) globalBatchJobsMetrics.save(job.ID, ri) // persist in-memory state to disk after every 10secs. batchLogIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job)) @@ -507,9 +697,19 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay } wk.Wait() + if ri.refused() { + cancelCause(nil) + globalBatchJobsMetrics.save(job.ID, ri) + batchLogIf(ctx, ri.updateAfter(GlobalContext, api, 0, job)) + if err := r.Notify(GlobalContext, ri); err != nil { + batchLogIf(ctx, fmt.Errorf("unable to notify %v", err)) + } + return nil + } + ri.RetryAttempts = attempts - ri.Complete = ri.ObjectsFailed == 0 - ri.Failed = ri.ObjectsFailed > 0 + ri.Complete = ri.ObjectsFailed == 0 && ri.DeleteMarkersFailed == 0 + ri.Failed = !ri.Complete globalBatchJobsMetrics.save(job.ID, ri) // persist in-memory state to disk. @@ -519,12 +719,14 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay batchLogIf(ctx, fmt.Errorf("unable to notify %v", err)) } - cancel() + cancelCause(nil) if ri.Failed { ri.ObjectsFailed = 0 + ri.DeleteMarkersFailed = 0 ri.Bucket = "" ri.Object = "" ri.Objects = 0 + ri.DeleteMarkers = 0 ri.BytesFailed = 0 ri.BytesTransferred = 0 retry = true // indicate we are retrying.. @@ -580,7 +782,10 @@ func toObjectInfo(bucket, object string, objInfo minio.ObjectInfo) ObjectInfo { return oi } -func (r BatchJobReplicateV1) writeAsArchive(ctx context.Context, objAPI ObjectLayer, remoteClnt *minio.Client, entries []ObjectInfo, prefix string) error { +func (r BatchJobReplicateV1) writeAsArchive(ctx context.Context, objAPI ObjectLayer, remoteClnt *minio.Client, entries []ObjectInfo, prefix string, sub *batchJobSubmitter) error { + // An entry the submitter may not read ends the archive and the job; an + // entry whose metadata cannot be read ends the archive. + var refused, readErr error input := make(chan minio.SnowballObject, 1) opts := minio.SnowballOptions{ Opts: minio.PutObjectOptions{ @@ -593,17 +798,41 @@ func (r BatchJobReplicateV1) writeAsArchive(ctx context.Context, objAPI ObjectLa SkipErrs: *r.Source.Snowball.SkipErrs, } + uploadCtx, cancelUpload := context.WithCancel(ctx) + defer cancelUpload() + producerDone := make(chan struct{}) go func() { + defer close(producerDone) defer xioutil.SafeClose(input) for _, entry := range entries { - gr, err := objAPI.GetObjectNInfo(ctx, r.Source.Bucket, + gr, err := objAPI.GetObjectNInfo(uploadCtx, r.Source.Bucket, entry.Name, nil, nil, ObjectOptions{ VersionID: entry.VersionID, }) if err != nil { batchLogIf(ctx, err) - continue + if isErrObjectNotFound(err) || isErrVersionNotFound(err) || *r.Source.Snowball.SkipErrs { + continue + } + // Not sent and not skippable: the batch goes to the + // per-object path. + readErr = err + return + } + // Each entry is a decision of its own, for the submitter as it is, + // on the metadata of all copies of the version. + logical, lerr := batchAuthoritativeRead(uploadCtx, objAPI, r.Source.Bucket, entry.Name, gr.ObjInfo) + if lerr != nil { + // Not sent: the batch goes to the per-object path. + gr.Close() + readErr = lerr + return + } + refused = sub.authorize(uploadCtx, append(r.pushAuthzRequests(gr.ObjInfo), r.pushAuthzRequests(logical)...)...) + if refused != nil { + gr.Close() + return } if prefix != "" { @@ -625,6 +854,7 @@ func (r BatchJobReplicateV1) writeAsArchive(ctx context.Context, objAPI ObjectLa opts, _, err := batchReplicationOpts(ctx, "", gr.ObjInfo) if err != nil { + gr.Close() batchLogIf(ctx, err) continue } @@ -635,23 +865,75 @@ func (r BatchJobReplicateV1) writeAsArchive(ctx context.Context, objAPI ObjectLa } } - input <- snowballObj + select { + case input <- snowballObj: + case <-uploadCtx.Done(): + gr.Close() + return + } } }() // Collect and upload all entries. - return remoteClnt.PutObjectsSnowball(ctx, r.Target.Bucket, opts, input) + err := remoteClnt.PutObjectsSnowball(uploadCtx, r.Target.Bucket, opts, input) + cancelUpload() + <-producerDone + if refused != nil { + return refused + } + if readErr != nil { + return readErr + } + return err +} + +// pushAuthzRequests are the reads of one source version that a push copies to +// the target: the version itself, and its tags, retention and legal hold. +func (r *BatchJobReplicateV1) pushAuthzRequests(oi ObjectInfo) []batchAuthzRequest { + bucket, object, versionID := r.Source.Bucket, oi.Name, oi.VersionID + if versionID == "" { + versionID = nullVersionID + } + reqs := []batchAuthzRequest{ + {action: policy.GetObjectAction, bucket: bucket, object: object, versionID: versionID, tags: oi.UserTags}, + {action: policy.GetObjectVersionAction, bucket: bucket, object: object, versionID: versionID, tags: oi.UserTags}, + } + if oi.UserTags != "" { + reqs = append(reqs, + batchAuthzRequest{action: policy.GetObjectTaggingAction, bucket: bucket, object: object, versionID: versionID, tags: oi.UserTags}, + batchAuthzRequest{action: policy.GetObjectVersionTaggingAction, bucket: bucket, object: object, versionID: versionID, tags: oi.UserTags}, + ) + } + meta := caseInsensitiveMap(oi.UserDefined) + _, mode := meta.Lookup(xhttp.AmzObjectLockMode) + _, until := meta.Lookup(xhttp.AmzObjectLockRetainUntilDate) + if mode || until { + reqs = append(reqs, batchAuthzRequest{action: policy.GetObjectRetentionAction, bucket: bucket, object: object, versionID: versionID, tags: oi.UserTags}) + } + if _, hold := meta.Lookup(xhttp.AmzObjectLockLegalHold); hold { + reqs = append(reqs, batchAuthzRequest{action: policy.GetObjectLegalHoldAction, bucket: bucket, object: object, versionID: versionID, tags: oi.UserTags}) + } + return reqs } // ReplicateToTarget read from source and replicate to configured target -func (r *BatchJobReplicateV1) ReplicateToTarget(ctx context.Context, api ObjectLayer, c *minio.Core, srcObjInfo ObjectInfo, retry bool) error { +func (r *BatchJobReplicateV1) ReplicateToTarget(ctx context.Context, api ObjectLayer, c *minio.Core, srcObjInfo ObjectInfo, retry bool, sub *batchJobSubmitter) 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 + p, err := sub.principal(ctx) + if err != nil { + return err + } if srcObjInfo.DeleteMarker || !srcObjInfo.VersionPurgeStatus.Empty() { + for _, req := range r.pushAuthzRequests(srcObjInfo) { + if err := p.authorize(req); err != nil { + return err + } + } if retry && !s3Type { if _, err := c.StatObject(ctx, tgtBucket, pathJoin(tgtPrefix, srcObject), minio.StatObjectOptions{ VersionID: srcObjInfo.VersionID, @@ -707,6 +989,19 @@ func (r *BatchJobReplicateV1) ReplicateToTarget(ctx context.Context, api ObjectL } defer rd.Close() objInfo := rd.ObjInfo + // Authorize the version as it is read, with the metadata of all its + // copies. + logical, err := batchAuthoritativeRead(ctx, api, srcBucket, srcObject, objInfo) + if err != nil { + return err + } + // What is sent comes from the opened copy; both it and the version as + // all copies describe it must be authorized. + for _, req := range append(r.pushAuthzRequests(objInfo), r.pushAuthzRequests(logical)...) { + if err := p.authorize(req); err != nil { + return err + } + } size, err := objInfo.GetActualSize() if err != nil { @@ -760,6 +1055,16 @@ type batchJobInfo struct { DeleteMarkersFailed int64 `json:"deleteMarkersFailed" msg:"dmf"` BytesTransferred int64 `json:"bytesTransferred" msg:"bt"` BytesFailed int64 `json:"bytesFailed" msg:"bf"` + + // Versions the job left in place because Object Lock protects them, + // and keys it refused as a whole for the same reason. A job that + // completes with them has not acted on those versions. + VersionsSkippedLocked int64 `json:"versionsSkippedLocked" msg:"vsl"` + KeysRefused int64 `json:"keysRefused" msg:"kr"` + + // FailureReason is set when the job was refused for its submitter. It + // ends the job: a job that has one does not run again. + FailureReason string `json:"failureReason,omitempty" msg:"fr"` } const ( @@ -901,40 +1206,31 @@ func (ri *batchJobInfo) clone() *batchJobInfo { BytesTransferred: ri.BytesTransferred, BytesFailed: ri.BytesFailed, Attempts: ri.Attempts, + + VersionsSkippedLocked: ri.VersionsSkippedLocked, + KeysRefused: ri.KeysRefused, + FailureReason: ri.FailureReason, } } -func (ri *batchJobInfo) countItem(size int64, dmarker, success bool, attempt int) { +// countItem counts the outcome of one item in the current attempt of a +// replicate job, which resets its counters before it retries. +func (ri *batchJobInfo) countItem(size int64, dmarker, success bool) { if ri == nil { return } ri.Attempts++ - if success { - if dmarker { - ri.DeleteMarkers++ - } else { - ri.Objects++ - ri.BytesTransferred += size - } - if attempt > 1 { - if dmarker { - ri.DeleteMarkersFailed-- - } else { - ri.ObjectsFailed-- - ri.BytesFailed += size - } - } - } else { - if attempt > 1 { - // Only count first attempt - return - } - if dmarker { - ri.DeleteMarkersFailed++ - } else { - ri.ObjectsFailed++ - ri.BytesFailed += size - } + switch { + case success && dmarker: + ri.DeleteMarkers++ + case success: + ri.Objects++ + ri.BytesTransferred += size + case dmarker: + ri.DeleteMarkersFailed++ + default: + ri.ObjectsFailed++ + ri.BytesFailed += size } } @@ -1002,35 +1298,7 @@ func (ri *batchJobInfo) expireFailures() bool { return ri.ObjectsFailed > 0 || ri.DeleteMarkersFailed > 0 } -// trackExpireAll counts the objects and delete markers that an expiry of all -// versions of info removed and, when it failed, those the walk saw that it -// did not remove, at least one. Only batch-expire uses it, for a rule that -// expires all versions of an object. -func (ri *batchJobInfo) trackExpireAll(info expireObjInfo, objects, markers int64, failed bool) { - if ri == nil { - return - } - - ri.mu.Lock() - defer ri.mu.Unlock() - - ri.Objects += objects - ri.DeleteMarkers += markers - if !failed { - ri.Bucket = info.Bucket - ri.Object = info.Name - return - } - failedObjects := max(int64(info.NumVersions)-info.DeleteMarkerCount-objects, 0) - failedMarkers := max(info.DeleteMarkerCount-markers, 0) - if failedObjects+failedMarkers == 0 { - failedObjects = 1 - } - ri.ObjectsFailed += failedObjects - ri.DeleteMarkersFailed += failedMarkers -} - -func (ri *batchJobInfo) trackCurrentBucketObject(bucket string, info ObjectInfo, success bool, attempt int) { +func (ri *batchJobInfo) trackCurrentBucketObject(bucket string, info ObjectInfo, success bool) { if ri == nil { return } @@ -1042,7 +1310,7 @@ func (ri *batchJobInfo) trackCurrentBucketObject(bucket string, info ObjectInfo, ri.Bucket = bucket ri.Object = info.Name } - ri.countItem(info.Size, info.DeleteMarker, success, attempt) + ri.countItem(info.Size, info.DeleteMarker, success) } func (ri *batchJobInfo) trackCurrentBucketBatch(bucket string, batch []ObjectInfo) { @@ -1056,7 +1324,7 @@ func (ri *batchJobInfo) trackCurrentBucketBatch(bucket string, batch []ObjectInf ri.Bucket = bucket for i := range batch { ri.Object = batch[i].Name - ri.countItem(batch[i].Size, batch[i].DeleteMarker, true, 1) + ri.countItem(batch[i].Size, batch[i].DeleteMarker, true) } } @@ -1073,6 +1341,9 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba if ri.Complete { return nil } + if done, err := ri.startAuthorized(ctx, api, job); done || err != nil { + return err + } globalBatchJobsMetrics.save(job.ID, ri) lastObject := ri.Object @@ -1180,6 +1451,12 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba walkCh = make(chan itemOrErr[ObjectInfo], 100) slowCh = make(chan itemOrErr[ObjectInfo], 100) ) + ctx, cancelCause := context.WithCancelCause(ctx) + // refuse ends the job for its submitter: record why before stopping. + refuse := func(err error) { + ri.refuse(err) + cancelCause(err) + } if r.Source.Snowball.Disable != nil && !*r.Source.Snowball.Disable && r.Source.Type.isMinio() && r.Target.Type.isMinio() { go func() { @@ -1201,7 +1478,11 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba batch := make([]ObjectInfo, 0, *r.Source.Snowball.Batch) writeFn := func(batch []ObjectInfo) { if len(batch) > 0 { - if err := r.writeAsArchive(ctx, api, cl, batch, r.Target.Prefix); err != nil { + if err := r.writeAsArchive(ctx, api, cl, batch, r.Target.Prefix, job.Submitter); err != nil { + if isBatchJobAuthzError(err) { + refuse(err) + return + } batchLogOnceIf(ctx, err, job.ID+"writeAsArchive") for _, b := range batch { slowCh <- itemOrErr[ObjectInfo]{Item: b} @@ -1250,7 +1531,6 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba if walkQuorum == "" { walkQuorum = "strict" } - ctx, cancelCause := context.WithCancelCause(ctx) // 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 @@ -1281,6 +1561,10 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba skipReplicate := false for res := range slowCh { + if context.Cause(ctx) != nil { + // Refused or canceled: schedule nothing more. + continue + } if res.Err != nil { ri.Failed = true batchLogOnceIf(ctx, res.Err, job.ID+"res.Err") @@ -1300,7 +1584,11 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba stopFn := globalBatchJobsMetrics.trace(batchJobMetricReplication, job.ID, attempts) success := true - if err := r.ReplicateToTarget(ctx, api, c, result, retry); err != nil { + if err := r.ReplicateToTarget(ctx, api, c, result, retry, job.Submitter); err != nil { + if isBatchJobAuthzError(err) { + refuse(err) + return + } if minio.ToErrorResponse(err).Code == "PreconditionFailed" { // pre-condition failed means we already have the object copied over. return @@ -1315,7 +1603,7 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba } else { stopFn(result, nil) } - ri.trackCurrentBucketObject(r.Source.Bucket, result, success, attempts) + ri.trackCurrentBucketObject(r.Source.Bucket, result, success) globalBatchJobsMetrics.save(job.ID, ri) // persist in-memory state to disk after every 10secs. batchLogOnceIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job), job.ID+"updateAfter2") @@ -1326,13 +1614,22 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba }() } wk.Wait() + if ri.refused() { + cancelCause(nil) + globalBatchJobsMetrics.save(job.ID, ri) + batchLogOnceIf(ctx, ri.updateAfter(GlobalContext, api, 0, job), job.ID+"updateAfter3") + if err := r.Notify(GlobalContext, ri); err != nil { + batchLogOnceIf(ctx, fmt.Errorf("unable to notify %v", err), job.ID+"notify") + } + return nil + } // Do not need to retry if we can't list objects on source. if context.Cause(ctx) != nil { return context.Cause(ctx) } ri.RetryAttempts = attempts - ri.Complete = ri.ObjectsFailed == 0 - ri.Failed = ri.ObjectsFailed > 0 + ri.Complete = ri.ObjectsFailed == 0 && ri.DeleteMarkersFailed == 0 + ri.Failed = !ri.Complete globalBatchJobsMetrics.save(job.ID, ri) // persist in-memory state to disk. @@ -1345,9 +1642,11 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba cancelCause(nil) if ri.Failed { ri.ObjectsFailed = 0 + ri.DeleteMarkersFailed = 0 ri.Bucket = "" ri.Object = "" ri.Objects = 0 + ri.DeleteMarkers = 0 ri.BytesFailed = 0 ri.BytesTransferred = 0 retry = true // indicate we are retrying.. @@ -1549,8 +1848,12 @@ func (j BatchJobRequest) Validate(ctx context.Context, o ObjectLayer) error { return errInvalidArgument } -func (j BatchJobRequest) delete(ctx context.Context, api ObjectLayer) { - deleteConfig(ctx, api, getJobPath(j)) +func (j BatchJobRequest) delete(ctx context.Context, api ObjectLayer) error { + err := deleteConfig(ctx, api, getJobPath(j)) + if errors.Is(err, errConfigNotFound) { + return nil + } + return err } func (j BatchJobRequest) getJobReportPath() (string, error) { @@ -1709,7 +2012,7 @@ func (a adminAPIHandlers) BatchJobStatus(w http.ResponseWriter, r *http.Request) return } - buf, err := json.Marshal(madmin.BatchJobStatus{LastMetric: ri.metric()}) + buf, err := json.Marshal(batchJobStatus{LastMetric: ri.statusMetric()}) if err != nil { batchLogIf(ctx, err) writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL) @@ -1747,8 +2050,13 @@ func (a adminAPIHandlers) DescribeBatchJob(w http.ResponseWriter, r *http.Reques } // Remove sensitive fields. - req.RedactSensitive() - buf, err := yaml.Marshal(req) + out, err := req.redacted() + if err != nil { + batchLogIf(ctx, err) + writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL) + return + } + buf, err := yaml.Marshal(out) if err != nil { batchLogIf(ctx, err) writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL) @@ -1806,6 +2114,27 @@ func (a adminAPIHandlers) StartBatchJob(w http.ResponseWriter, r *http.Request) } } + if job.jobSections() != 1 { + writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrInvalidRequest, errBatchJobSections), r.URL) + return + } + + // The job acts as the credential that submits it, and only as far as + // that credential may act on the job's buckets. + submitter, err := newBatchJobSubmitter(creds) + if err == nil { + job.Submitter = submitter + err = submitter.authorize(ctx, job.startAuthzRequests()...) + } + if err != nil { + apiErr := toAPIError(ctx, err) + if isBatchJobAuthzError(err) { + apiErr = errorCodes.ToAPIErrWithErr(ErrAccessDenied, err) + } + writeErrorResponseJSON(ctx, w, apiErr, r.URL) + return + } + // Validate the incoming job request if err := job.Validate(ctx, objectAPI); err != nil { writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL) @@ -1942,6 +2271,14 @@ func (j *BatchJobPool) cleanupReports(randomWait func() time.Duration) { continue } if (ri.Complete || ri.Failed) && time.Since(ri.LastUpdate) > oldJobsExpiration { + // Remove the job before its report, so that a finished + // job is never resumed without its outcome. + if ri.JobID != "" { + if err := (BatchJobRequest{ID: ri.JobID}).delete(ctx, j.objLayer); err != nil { + batchLogIf(ctx, err) + continue + } + } deleteConfig(ctx, j.objLayer, result.Item.Name) } } @@ -2082,7 +2419,7 @@ func (j *BatchJobPool) queueJob(req *BatchJobRequest) error { }) case j.jobCh <- req: default: - return fmt.Errorf("batch job queue is currently full please try again later %#v", req) + return fmt.Errorf("batch job queue is currently full please try again later (job %s)", req.ID) } return nil } diff --git a/cmd/batch-handlers_gen.go b/cmd/batch-handlers_gen.go index ad29932b4..d0e1d7086 100644 --- a/cmd/batch-handlers_gen.go +++ b/cmd/batch-handlers_gen.go @@ -179,6 +179,24 @@ func (z *BatchJobRequest) DecodeMsg(dc *msgp.Reader) (err error) { return } } + case "Submitter": + if dc.IsNil() { + err = dc.ReadNil() + if err != nil { + err = msgp.WrapError(err, "Submitter") + return + } + z.Submitter = nil + } else { + if z.Submitter == nil { + z.Submitter = new(batchJobSubmitter) + } + err = z.Submitter.DecodeMsg(dc) + if err != nil { + err = msgp.WrapError(err, "Submitter") + return + } + } default: err = dc.Skip() if err != nil { @@ -192,9 +210,9 @@ func (z *BatchJobRequest) DecodeMsg(dc *msgp.Reader) (err error) { // EncodeMsg implements msgp.Encodable func (z *BatchJobRequest) EncodeMsg(en *msgp.Writer) (err error) { - // map header, size 6 + // map header, size 7 // write "ID" - err = en.Append(0x86, 0xa2, 0x49, 0x44) + err = en.Append(0x87, 0xa2, 0x49, 0x44) if err != nil { return } @@ -274,15 +292,32 @@ func (z *BatchJobRequest) EncodeMsg(en *msgp.Writer) (err error) { return } } + // write "Submitter" + err = en.Append(0xa9, 0x53, 0x75, 0x62, 0x6d, 0x69, 0x74, 0x74, 0x65, 0x72) + if err != nil { + return + } + if z.Submitter == nil { + err = en.WriteNil() + if err != nil { + return + } + } else { + err = z.Submitter.EncodeMsg(en) + if err != nil { + err = msgp.WrapError(err, "Submitter") + return + } + } return } // MarshalMsg implements msgp.Marshaler func (z *BatchJobRequest) MarshalMsg(b []byte) (o []byte, err error) { o = msgp.Require(b, z.Msgsize()) - // map header, size 6 + // map header, size 7 // string "ID" - o = append(o, 0x86, 0xa2, 0x49, 0x44) + o = append(o, 0x87, 0xa2, 0x49, 0x44) o = msgp.AppendString(o, z.ID) // string "User" o = append(o, 0xa4, 0x55, 0x73, 0x65, 0x72) @@ -323,6 +358,17 @@ func (z *BatchJobRequest) MarshalMsg(b []byte) (o []byte, err error) { return } } + // string "Submitter" + o = append(o, 0xa9, 0x53, 0x75, 0x62, 0x6d, 0x69, 0x74, 0x74, 0x65, 0x72) + if z.Submitter == nil { + o = msgp.AppendNil(o) + } else { + o, err = z.Submitter.MarshalMsg(o) + if err != nil { + err = msgp.WrapError(err, "Submitter") + return + } + } return } @@ -413,6 +459,23 @@ func (z *BatchJobRequest) UnmarshalMsg(bts []byte) (o []byte, err error) { return } } + case "Submitter": + if msgp.IsNil(bts) { + bts, err = msgp.ReadNilBytes(bts) + if err != nil { + return + } + z.Submitter = nil + } else { + if z.Submitter == nil { + z.Submitter = new(batchJobSubmitter) + } + bts, err = z.Submitter.UnmarshalMsg(bts) + if err != nil { + err = msgp.WrapError(err, "Submitter") + return + } + } default: bts, err = msgp.Skip(bts) if err != nil { @@ -445,6 +508,12 @@ func (z *BatchJobRequest) Msgsize() (s int) { } else { s += z.Expire.Msgsize() } + s += 10 + if z.Submitter == nil { + s += msgp.NilSize + } else { + s += z.Submitter.Msgsize() + } return } @@ -568,6 +637,24 @@ func (z *batchJobInfo) DecodeMsg(dc *msgp.Reader) (err error) { err = msgp.WrapError(err, "BytesFailed") return } + case "vsl": + z.VersionsSkippedLocked, err = dc.ReadInt64() + if err != nil { + err = msgp.WrapError(err, "VersionsSkippedLocked") + return + } + case "kr": + z.KeysRefused, err = dc.ReadInt64() + if err != nil { + err = msgp.WrapError(err, "KeysRefused") + return + } + case "fr": + z.FailureReason, err = dc.ReadString() + if err != nil { + err = msgp.WrapError(err, "FailureReason") + return + } default: err = dc.Skip() if err != nil { @@ -581,9 +668,9 @@ func (z *batchJobInfo) DecodeMsg(dc *msgp.Reader) (err error) { // EncodeMsg implements msgp.Encodable func (z *batchJobInfo) EncodeMsg(en *msgp.Writer) (err error) { - // map header, size 17 + // map header, size 20 // write "v" - err = en.Append(0xde, 0x0, 0x11, 0xa1, 0x76) + err = en.Append(0xde, 0x0, 0x14, 0xa1, 0x76) if err != nil { return } @@ -752,15 +839,45 @@ func (z *batchJobInfo) EncodeMsg(en *msgp.Writer) (err error) { err = msgp.WrapError(err, "BytesFailed") return } + // write "vsl" + err = en.Append(0xa3, 0x76, 0x73, 0x6c) + if err != nil { + return + } + err = en.WriteInt64(z.VersionsSkippedLocked) + if err != nil { + err = msgp.WrapError(err, "VersionsSkippedLocked") + return + } + // write "kr" + err = en.Append(0xa2, 0x6b, 0x72) + if err != nil { + return + } + err = en.WriteInt64(z.KeysRefused) + if err != nil { + err = msgp.WrapError(err, "KeysRefused") + return + } + // write "fr" + err = en.Append(0xa2, 0x66, 0x72) + if err != nil { + return + } + err = en.WriteString(z.FailureReason) + if err != nil { + err = msgp.WrapError(err, "FailureReason") + return + } return } // MarshalMsg implements msgp.Marshaler func (z *batchJobInfo) MarshalMsg(b []byte) (o []byte, err error) { o = msgp.Require(b, z.Msgsize()) - // map header, size 17 + // map header, size 20 // string "v" - o = append(o, 0xde, 0x0, 0x11, 0xa1, 0x76) + o = append(o, 0xde, 0x0, 0x14, 0xa1, 0x76) o = msgp.AppendInt(o, z.Version) // string "jid" o = append(o, 0xa3, 0x6a, 0x69, 0x64) @@ -810,6 +927,15 @@ func (z *batchJobInfo) MarshalMsg(b []byte) (o []byte, err error) { // string "bf" o = append(o, 0xa2, 0x62, 0x66) o = msgp.AppendInt64(o, z.BytesFailed) + // string "vsl" + o = append(o, 0xa3, 0x76, 0x73, 0x6c) + o = msgp.AppendInt64(o, z.VersionsSkippedLocked) + // string "kr" + o = append(o, 0xa2, 0x6b, 0x72) + o = msgp.AppendInt64(o, z.KeysRefused) + // string "fr" + o = append(o, 0xa2, 0x66, 0x72) + o = msgp.AppendString(o, z.FailureReason) return } @@ -933,6 +1059,24 @@ func (z *batchJobInfo) UnmarshalMsg(bts []byte) (o []byte, err error) { err = msgp.WrapError(err, "BytesFailed") return } + case "vsl": + z.VersionsSkippedLocked, bts, err = msgp.ReadInt64Bytes(bts) + if err != nil { + err = msgp.WrapError(err, "VersionsSkippedLocked") + return + } + case "kr": + z.KeysRefused, bts, err = msgp.ReadInt64Bytes(bts) + if err != nil { + err = msgp.WrapError(err, "KeysRefused") + return + } + case "fr": + z.FailureReason, bts, err = msgp.ReadStringBytes(bts) + if err != nil { + err = msgp.WrapError(err, "FailureReason") + return + } default: bts, err = msgp.Skip(bts) if err != nil { @@ -947,6 +1091,6 @@ func (z *batchJobInfo) UnmarshalMsg(bts []byte) (o []byte, err error) { // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message func (z *batchJobInfo) Msgsize() (s int) { - s = 3 + 2 + msgp.IntSize + 4 + msgp.StringPrefixSize + len(z.JobID) + 3 + msgp.StringPrefixSize + len(z.JobType) + 3 + msgp.TimeSize + 3 + msgp.TimeSize + 3 + msgp.IntSize + 3 + msgp.IntSize + 4 + msgp.BoolSize + 4 + msgp.BoolSize + 5 + msgp.StringPrefixSize + len(z.Bucket) + 5 + msgp.StringPrefixSize + len(z.Object) + 3 + msgp.Int64Size + 3 + msgp.Int64Size + 4 + msgp.Int64Size + 4 + msgp.Int64Size + 3 + msgp.Int64Size + 3 + msgp.Int64Size + s = 3 + 2 + msgp.IntSize + 4 + msgp.StringPrefixSize + len(z.JobID) + 3 + msgp.StringPrefixSize + len(z.JobType) + 3 + msgp.TimeSize + 3 + msgp.TimeSize + 3 + msgp.IntSize + 3 + msgp.IntSize + 4 + msgp.BoolSize + 4 + msgp.BoolSize + 5 + msgp.StringPrefixSize + len(z.Bucket) + 5 + msgp.StringPrefixSize + len(z.Object) + 3 + msgp.Int64Size + 3 + msgp.Int64Size + 4 + msgp.Int64Size + 4 + msgp.Int64Size + 3 + msgp.Int64Size + 3 + msgp.Int64Size + 4 + msgp.Int64Size + 3 + msgp.Int64Size + 3 + msgp.StringPrefixSize + len(z.FailureReason) return } diff --git a/cmd/batch-progress.go b/cmd/batch-progress.go new file mode 100644 index 000000000..968e8e0ef --- /dev/null +++ b/cmd/batch-progress.go @@ -0,0 +1,187 @@ +// Copyright (c) 2026 PGSTY +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "slices" + "strings" + "sync" + + "github.com/minio/madmin-go/v3" +) + +// batchCounts are outcomes of a batch job that a later run of the job would +// see again, and so must be persisted together with the cursor of the walk. +type batchCounts struct { + objects int64 + objectsFailed int64 + deleteMarkersFailed int64 + versionsSkipped int64 + keysRefused int64 +} + +func (c *batchCounts) add(o batchCounts) { + c.objects += o.objects + c.objectsFailed += o.objectsFailed + c.deleteMarkersFailed += o.deleteMarkersFailed + c.versionsSkipped += o.versionsSkipped + c.keysRefused += o.keysRefused +} + +//msgp:ignore batchProgress batchProgressItem batchCounts + +// batchProgress is the cursor of a walk whose items are done concurrently. +// +// The items of a walk are numbered in walk order. A key is done when every +// item of it, and every item before it, is done; only then does the cursor +// move past it and its counts become part of the job's counts. The cursor and +// the counts are persisted together, so a job resumed after the cursor +// neither skips an item that was not done nor counts an item twice. +type batchProgress struct { + mu sync.Mutex + ri *batchJobInfo + bucket string + issued uint64 + next uint64 + pending map[uint64]batchProgressItem + // key is the key of the item last committed, and held are the counts of + // its items so far. + key string + held batchCounts +} + +type batchProgressItem struct { + key string + counts batchCounts +} + +func newBatchProgress(ri *batchJobInfo, bucket string) *batchProgress { + return &batchProgress{ri: ri, bucket: bucket, pending: make(map[uint64]batchProgressItem)} +} + +// begin numbers the next item of the walk. +func (p *batchProgress) begin() uint64 { + p.mu.Lock() + defer p.mu.Unlock() + seq := p.issued + p.issued++ + return seq +} + +// done records item seq, of key, as done with counts. +func (p *batchProgress) done(seq uint64, key string, counts batchCounts) { + p.mu.Lock() + defer p.mu.Unlock() + p.pending[seq] = batchProgressItem{key: key, counts: counts} + for { + item, ok := p.pending[p.next] + if !ok { + return + } + delete(p.pending, p.next) + p.next++ + if item.key != p.key { + p.commitLocked() + p.key = item.key + } + p.held.add(item.counts) + } +} + +// finish commits the last key if every item of the walk is done. Call it +// once no item is in progress. +func (p *batchProgress) finish() { + p.mu.Lock() + defer p.mu.Unlock() + if p.next == p.issued { + p.commitLocked() + } +} + +func (p *batchProgress) commitLocked() { + if p.key == "" { + return + } + ri := p.ri + ri.mu.Lock() + ri.Bucket = p.bucket + ri.Object = p.key + ri.Objects += p.held.objects + ri.ObjectsFailed += p.held.objectsFailed + ri.DeleteMarkersFailed += p.held.deleteMarkersFailed + ri.VersionsSkippedLocked += p.held.versionsSkipped + ri.KeysRefused += p.held.keysRefused + ri.mu.Unlock() + p.held = batchCounts{} +} + +// countRemoved counts versions a job removed. A removed version is gone for +// any later run, so it is counted at once, not at the cursor. +func (ri *batchJobInfo) countRemoved(bucket string, objects, markers int64) { + ri.mu.Lock() + defer ri.mu.Unlock() + ri.Bucket = bucket + ri.Objects += objects + ri.DeleteMarkers += markers +} + +// countSkipped counts a version a job left in place because Object Lock +// protects it, for a job that is counted per run rather than at a cursor. +func (ri *batchJobInfo) countSkipped() { + ri.mu.Lock() + defer ri.mu.Unlock() + ri.VersionsSkippedLocked++ +} + +// batchCanonicalPrefixes orders the prefixes of a walk and drops a prefix +// that another one covers. The keys of the walk are then visited once each, +// in one ascending order, which a single cursor can describe. +func batchCanonicalPrefixes(prefixes []string) []string { + if len(prefixes) == 0 { + return []string{""} + } + sorted := slices.Clone(prefixes) + slices.Sort(sorted) + out := sorted[:0] + for _, p := range sorted { + if len(out) > 0 && strings.HasPrefix(p, out[len(out)-1]) { + continue + } + out = append(out, p) + } + return out +} + +// batchResumeAfter reports whether a resumed walk returned a key after the +// cursor. The cursor key is done; the walk forwards conservatively and +// returns it again. Keys compare as the walk orders them, by name, so a +// directory object comes before the keys under it. +func batchResumeAfter(cursor, name string) bool { + return cursor == "" || name > cursor +} + +// batchJobStatus is the status response: madmin.BatchJobStatus with the +// fields this server adds. The added fields are additive; a client that does +// not know them ignores them. +type batchJobStatus struct { + LastMetric batchJobStatusMetric +} + +type batchJobStatusMetric struct { + madmin.JobMetric + VersionsSkippedLocked int64 `json:"versionsSkippedLocked,omitempty"` + KeysRefused int64 `json:"keysRefused,omitempty"` + FailureReason string `json:"failureReason,omitempty"` +} + +func (ri *batchJobInfo) statusMetric() batchJobStatusMetric { + ri.mu.RLock() + defer ri.mu.RUnlock() + return batchJobStatusMetric{ + JobMetric: ri.metric(), + VersionsSkippedLocked: ri.VersionsSkippedLocked, + KeysRefused: ri.KeysRefused, + FailureReason: ri.FailureReason, + } +} diff --git a/cmd/batch-rotate.go b/cmd/batch-rotate.go index 3ebbc844d..1f7306ff1 100644 --- a/cmd/batch-rotate.go +++ b/cmd/batch-rotate.go @@ -35,6 +35,7 @@ import ( xhttp "github.com/minio/minio/internal/http" "github.com/minio/minio/internal/kms" "github.com/pgsty/silo-pkg/v3/env" + "github.com/pgsty/silo-pkg/v3/policy" "github.com/pgsty/silo-pkg/v3/workers" ) @@ -161,8 +162,68 @@ func (r BatchJobKeyRotateV1) Notify(ctx context.Context, ri *batchJobInfo) error return notifyEndpoint(ctx, ri, r.Flags.Notify.Endpoint, r.Flags.Notify.Token) } -// KeyRotate rotates encryption key of an object -func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, objInfo ObjectInfo) error { +// batchAuthoritativeVersion returns the version obj as all its copies +// describe it. With several pools a version can have copies in more than one +// pool, and a read returns one of them; the merged copy carries the newest +// Object Lock and tag state. The caller holds the object's write lock. +func batchAuthoritativeVersion(ctx context.Context, api ObjectLayer, bucket, object string, obj ObjectInfo) (ObjectInfo, error) { + z, ok := api.(*erasureServerPools) + if !ok || z.SinglePool() { + return obj, nil + } + versionID := obj.VersionID + if versionID == "" { + versionID = nullVersionID + } + return z.replicaObjectInfo(ctx, bucket, encodeDirObject(object), ObjectOptions{VersionID: versionID, NoLock: true, NoAuditLog: true}) +} + +// batchAuthoritativeRead is batchAuthoritativeVersion for a version that has +// been opened for reading. It takes the object's read lock itself, since an +// opened reader may hold none. +func batchAuthoritativeRead(ctx context.Context, api ObjectLayer, bucket, object string, obj ObjectInfo) (ObjectInfo, error) { + z, ok := api.(*erasureServerPools) + if !ok || z.SinglePool() { + return obj, nil + } + lk := z.NewNSLock(bucket, encodeDirObject(object)) + lkctx, err := lk.GetRLock(ctx, globalOperationTimeout) + if err != nil { + return ObjectInfo{}, err + } + defer lk.RUnlock(lkctx) + return batchAuthoritativeVersion(lkctx.Context(), api, bucket, object, obj) +} + +// rotateAuthzRequests are the actions of rotating one version: it is read and +// written again in place with the new key. +func (r *BatchJobKeyRotateV1) rotateAuthzRequests(oi ObjectInfo) []batchAuthzRequest { + versionID := oi.VersionID + if versionID == "" { + versionID = nullVersionID + } + // The encryption the rewrite leaves: rotation keeps the kind of the + // version (rotateKey), and SSE-KMS takes the requested key. + values := map[string][]string{} + switch { + case crypto.S3KMS.IsEncrypted(oi.UserDefined): + values[strings.ToLower(xhttp.AmzServerSideEncryption)] = []string{xhttp.AmzEncryptionKMS} + if r.Encryption.Type == ssekms && r.Encryption.Key != "" { + values[strings.ToLower(xhttp.AmzServerSideEncryptionKmsID)] = []string{r.Encryption.Key} + } + case crypto.S3.IsEncrypted(oi.UserDefined): + values[strings.ToLower(xhttp.AmzServerSideEncryption)] = []string{xhttp.AmzEncryptionAES} + } + return []batchAuthzRequest{ + {action: policy.GetObjectAction, bucket: r.Bucket, object: oi.Name, versionID: versionID, tags: oi.UserTags}, + {action: policy.GetObjectVersionAction, bucket: r.Bucket, object: oi.Name, versionID: versionID, tags: oi.UserTags}, + {action: policy.PutObjectAction, bucket: r.Bucket, object: oi.Name, versionID: versionID, tags: oi.UserTags, values: values}, + } +} + +// KeyRotate rotates encryption key of an object. A version that Object Lock +// protects is not rewritten: it returns ObjectLocked. +func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, objInfo ObjectInfo, sub *batchJobSubmitter) error { srcBucket := r.Bucket srcObject := objInfo.Name @@ -180,7 +241,8 @@ func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, ob versioned := globalBucketVersioningSys.PrefixEnabled(srcBucket, srcObject) versionSuspended := globalBucketVersioningSys.PrefixSuspended(srcBucket, srcObject) - lock := api.NewObjectNSLock(r.Bucket, objInfo.Name) + // Lock the name writers lock: the stored name of a directory object. + lock := api.NewObjectNSLock(r.Bucket, encodeDirObject(objInfo.Name)) lkctx, err := lock.GetLock(ctx, globalOperationTimeout) if err != nil { return err @@ -198,6 +260,24 @@ func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, ob if err != nil { return err } + // Decide on the version as it is under the lock that the rewrite holds, + // before any key is requested, with the metadata of all its copies. + logical, err := batchAuthoritativeVersion(ctx, api, r.Bucket, objInfo.Name, obj) + if err != nil { + return err + } + if enforceRetentionForDeletion(ctx, logical) { + return ObjectLocked{Bucket: r.Bucket, Object: obj.Name, VersionID: obj.VersionID} + } + p, err := sub.principal(ctx) + if err != nil { + return err + } + for _, req := range r.rotateAuthzRequests(logical) { + if err := p.authorize(req); err != nil { + return err + } + } oi := obj.Clone() var ( newKeyID string @@ -260,6 +340,9 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba if ri.Complete { return nil } + if done, err := ri.startAuthorized(ctx, api, job); done || err != nil { + return err + } globalBatchJobsMetrics.save(job.ID, ri) lastObject := ri.Object @@ -273,8 +356,6 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba delay = batchKeyRotateJobDefaultRetryDelay } - 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 @@ -355,65 +436,92 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba return err } - ctx, cancel := context.WithCancel(ctx) + ctx, cancelCause := context.WithCancelCause(ctx) + defer cancelCause(nil) + // refuse ends the job for its submitter: record why before stopping. + refuse := func(err error) { + ri.refuse(err) + cancelCause(err) + } + progress := newBatchProgress(ri, r.Bucket) results := make(chan itemOrErr[ObjectInfo], 100) if err := api.Walk(ctx, r.Bucket, r.Prefix, results, WalkOptions{ Marker: lastObject, Filter: selectObj, }); err != nil { - cancel() // Do not need to retry if we can't list objects on source. return err } failed := false for res := range results { + if context.Cause(ctx) != nil { + // Refused or canceled: schedule nothing more. + continue + } if res.Err != nil { failed = true batchLogIf(ctx, res.Err) break } result := res.Item + if !batchResumeAfter(lastObject, result.Name) { + // Completed by the run that set the cursor. + continue + } sseKMS := crypto.S3KMS.IsEncrypted(result.UserDefined) sseS3 := crypto.S3.IsEncrypted(result.UserDefined) if !sseKMS && !sseS3 { // neither sse-s3 nor sse-kms disallowed continue } + seq := progress.begin() wk.Take() go func() { defer wk.Give() + var counts batchCounts for attempts := 1; attempts <= retryAttempts; attempts++ { stopFn := globalBatchJobsMetrics.trace(batchJobMetricKeyRotation, job.ID, attempts) - success := true - if err := r.KeyRotate(ctx, api, result); err != nil { - stopFn(result, err) - batchLogIf(ctx, err) - success = false - if attempts >= retryAttempts { - auditOptions := AuditLogOptions{ - Event: "KeyRotate", - APIName: "StartBatchJob", - Bucket: result.Bucket, - Object: result.Name, - VersionID: result.VersionID, - Error: err.Error(), - } - auditLogInternal(ctx, auditOptions) - } - } else { - stopFn(result, nil) + err := r.KeyRotate(ctx, api, result, job.Submitter) + stopFn(result, err) + if err == nil { + counts.objects = 1 + break } - ri.trackCurrentBucketObject(r.Bucket, result, success, attempts) - globalBatchJobsMetrics.save(job.ID, ri) - // persist in-memory state to disk after every 10secs. - batchLogIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job)) - if success { + if isBatchJobAuthzError(err) { + refuse(err) + return + } + if isErrObjectLocked(err) { + // Left as it is, not retried and not a failure. + counts.versionsSkipped = 1 + break + } + if ctx.Err() != nil { + // Not done; a later run rotates it. + return + } + batchLogIf(ctx, err) + if attempts >= retryAttempts { + auditOptions := AuditLogOptions{ + Event: "KeyRotate", + APIName: "StartBatchJob", + Bucket: result.Bucket, + Object: result.Name, + VersionID: result.VersionID, + Error: err.Error(), + } + auditLogInternal(ctx, auditOptions) + counts.objectsFailed = 1 break } if delay > 0 { - time.Sleep(delay + time.Duration(rnd.Float64()*float64(delay))) + time.Sleep(delay + time.Duration(rand.Float64()*float64(delay))) } } + progress.done(seq, result.Name, counts) + globalBatchJobsMetrics.save(job.ID, ri) + // persist in-memory state to disk after every 10secs. + batchLogIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job)) if wait := globalBatchConfig.KeyRotationWait(); wait > 0 { time.Sleep(wait) @@ -421,18 +529,31 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba }() } wk.Wait() - - ri.Complete = !failed && ri.ObjectsFailed == 0 - ri.Failed = failed || ri.ObjectsFailed > 0 - globalBatchJobsMetrics.save(job.ID, ri) - // persist in-memory state to disk. - batchLogIf(ctx, ri.updateAfter(ctx, api, 0, job)) - - if err := r.Notify(ctx, ri); err != nil { - batchLogIf(ctx, fmt.Errorf("unable to notify %v", err)) + cause := context.Cause(ctx) + if !failed && cause == nil { + // The walk ended: its last key is done. + progress.finish() + } + if cause != nil && !ri.refused() { + // Stopped before the end: keep the work after the cursor pending. + globalBatchJobsMetrics.save(job.ID, ri) + batchLogIf(ctx, ri.updateAfter(GlobalContext, api, 0, job)) + return cause } - cancel() + if !ri.refused() { + ri.mu.Lock() + ri.Complete = !failed && ri.ObjectsFailed == 0 + ri.Failed = failed || ri.ObjectsFailed > 0 + ri.mu.Unlock() + } + globalBatchJobsMetrics.save(job.ID, ri) + // persist in-memory state to disk. + batchLogIf(ctx, ri.updateAfter(GlobalContext, api, 0, job)) + + if err := r.Notify(GlobalContext, ri); err != nil { + batchLogIf(ctx, fmt.Errorf("unable to notify %v", err)) + } return nil } diff --git a/cmd/erasure-server-pool-delete-all-options_test.go b/cmd/erasure-server-pool-delete-all-options_test.go index 96039ffae..4b64eeaf3 100644 --- a/cmd/erasure-server-pool-delete-all-options_test.go +++ b/cmd/erasure-server-pool-delete-all-options_test.go @@ -149,7 +149,11 @@ func TestPurgeAllVersionsBudgetCoversReads(t *testing.T) { // A marker that an expiration job failed to remove fails the job. func TestPurgeAllVersionsMarkerFailureFailsJob(t *testing.T) { ri := &batchJobInfo{} - ri.trackExpireAll(expireObjInfo{ObjectInfo: ObjectInfo{NumVersions: 2}, DeleteMarkerCount: 1}, 1, 0, true) + var res batchCounts + res.expireAllFailed(expireObjInfo{ObjectInfo: ObjectInfo{Name: "key", NumVersions: 2}, DeleteMarkerCount: 1}, 1, 0) + progress := newBatchProgress(ri, "bucket") + progress.done(progress.begin(), "key", res) + progress.finish() if ri.ObjectsFailed != 0 || ri.DeleteMarkersFailed != 1 || !ri.expireFailures() { t.Errorf("failed=%d/%d failures=%t", ri.ObjectsFailed, ri.DeleteMarkersFailed, ri.expireFailures()) } diff --git a/cmd/erasure-server-pool-delete-all_test.go b/cmd/erasure-server-pool-delete-all_test.go index b3c861cb1..eb6ef2ecf 100644 --- a/cmd/erasure-server-pool-delete-all_test.go +++ b/cmd/erasure-server-pool-delete-all_test.go @@ -332,8 +332,9 @@ func (p *purgeTest) matched(name string) expireObjInfo { return expireObjInfo{ObjectInfo: oi, ExpireAll: true, DeleteMarkerCount: markers} } -// batch runs the deletes of a batch expiration job for walked, the keys its -// walk matched, and returns the counts of the job. +// batch runs the deletes of a batch expiration job of the root user for +// walked, the keys its walk matched, as one batch of a walk that reached its +// end, and returns the counts of the job. func (p *purgeTest) batch(walked ...expireObjInfo) *batchJobInfo { p.t.Helper() globalBatchJobsMetrics.Lock() @@ -341,7 +342,11 @@ func (p *purgeTest) batch(walked ...expireObjInfo) *batchJobInfo { globalBatchJobsMetrics.metrics = make(map[string]*batchJobInfo) } globalBatchJobsMetrics.Unlock() - job := BatchJobRequest{ID: "purge-" + mustGetUUID(), Started: time.Now(), Expire: &BatchJobExpire{Bucket: p.bucket}} + sub, err := newBatchJobSubmitter(globalActiveCred) + if err != nil { + p.t.Fatal(err) + } + job := BatchJobRequest{ID: "purge-" + mustGetUUID(), Started: time.Now(), Submitter: sub, Expire: &BatchJobExpire{Bucket: p.bucket}} ri := &batchJobInfo{JobID: job.ID} wk, err := workers.New(1) if err != nil { @@ -350,7 +355,8 @@ func (p *purgeTest) batch(walked ...expireObjInfo) *batchJobInfo { ch := make(chan []expireObjInfo, 1) ch <- walked close(ch) - batchObjsForDelete(p.t.Context(), job.Expire, ri, job, p.z, wk, ch) + refuse := func(err error) { p.t.Errorf("batch job refused: %v", err) } + batchObjsForDelete(p.t.Context(), refuse, func() bool { return true }, job.Expire, ri, job, p.z, wk, ch) wk.Wait() return ri } @@ -837,8 +843,8 @@ func TestPurgeAllVersionsKeepsProtectedWrites(t *testing.T) { t.Errorf("%s holds %v, want both versions", name, got) } } - if ri.Objects != 2 || ri.ObjectsFailed == 0 { - t.Errorf("job counts objects=%d failed=%d, want 2 and the refusals", ri.Objects, ri.ObjectsFailed) + if ri.Objects != 2 || ri.KeysRefused != 3 || ri.ObjectsFailed != 0 { + t.Errorf("job counts objects=%d refused=%d failed=%d, want 2, 3 and 0", ri.Objects, ri.KeysRefused, ri.ObjectsFailed) } } diff --git a/cmd/erasure-server-pool-lock-domain_test.go b/cmd/erasure-server-pool-lock-domain_test.go index 13249b213..2097f731d 100644 --- a/cmd/erasure-server-pool-lock-domain_test.go +++ b/cmd/erasure-server-pool-lock-domain_test.go @@ -1013,9 +1013,13 @@ func TestLockDomainKeyRotationWaitsForWriter(t *testing.T) { // The rotation's listing found an SSE-S3 object; it reads it again // under the lock. listed := ObjectInfo{Bucket: bucket, Name: key, UserDefined: map[string]string{crypto.MetaSealedKeyS3: "listed"}} + sub, err := newBatchJobSubmitter(globalActiveCred) + if err != nil { + t.Fatal(err) + } waited, _, _ := ldOverlapRun(t, func(ctx context.Context) error { return ldPut(ctx, z, bucket, key, "written") }, - func(ctx context.Context) error { return job.KeyRotate(ctx, z, listed) }) + func(ctx context.Context) error { return job.KeyRotate(ctx, z, listed, sub) }) if !waited { t.Errorf("key rotation ran while a PUT held its commit") }