diff --git a/cmd/bucket-lifecycle-incomplete.go b/cmd/bucket-lifecycle-incomplete.go new file mode 100644 index 000000000..c50b67d54 --- /dev/null +++ b/cmd/bucket-lifecycle-incomplete.go @@ -0,0 +1,148 @@ +// Copyright (c) 2026 PGSTY +// +// This file is part of Silo Object Storage stack +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + + "github.com/minio/minio/internal/bucket/lifecycle" + "github.com/minio/minio/internal/event" +) + +// An ILM expiry attempt that ends without removing what it addressed is +// incomplete: the rule no longer applied when it was checked again under the +// object lock, Object Lock protects the version, the history needed to decide +// could not be read, or another error. It is counted by reason, traced with +// the reason, and, unless it was refused, audited and logged. It never sends +// a removal notification: what it addressed is still there. +const ( + ilmIncompleteNotApplicable = "not_applicable" + ilmIncompleteLocked = "locked" + ilmIncompleteHistoryUnavailable = "history_unavailable" + ilmIncompleteError = "error" +) + +// ilmIncompleteReasons are the reasons expiry_incomplete_total counts, a +// bounded set of label values. +var ilmIncompleteReasons = []string{ + ilmIncompleteNotApplicable, + ilmIncompleteLocked, + ilmIncompleteHistoryUnavailable, + ilmIncompleteError, +} + +// ilmIncompleteStats counts incomplete ILM expiries by reason. +type ilmIncompleteStats struct { + counts map[string]*atomic.Int64 +} + +func newILMIncompleteStats(reasons ...string) *ilmIncompleteStats { + s := &ilmIncompleteStats{counts: make(map[string]*atomic.Int64, len(reasons))} + for _, r := range reasons { + s.counts[r] = &atomic.Int64{} + } + return s +} + +func (s *ilmIncompleteStats) add(reason string) { + if c, ok := s.counts[reason]; ok { + c.Add(1) + } +} + +// snapshot returns the counts by reason. +func (s *ilmIncompleteStats) snapshot() map[string]int64 { + out := make(map[string]int64, len(s.counts)) + for r, c := range s.counts { + out[r] = c.Load() + } + return out +} + +var globalILMIncomplete = newILMIncompleteStats(ilmIncompleteReasons...) + +// ilmExpiryNotDone classifies the error of an ILM expiry that did not remove +// what it addressed. A version that is already gone is not counted. +func ilmExpiryNotDone(err error) (reason string, counted bool) { + switch { + case err == nil || isErrObjectNotFound(err) || isErrVersionNotFound(err): + return "", false + case errors.Is(err, errLifecycleNotApplicable): + return ilmIncompleteNotApplicable, true + } + var locked ObjectLocked + if errors.As(err, &locked) { + return ilmIncompleteLocked, true + } + if isErrReadQuorum(err) || isErrWriteQuorum(err) { + return ilmIncompleteHistoryUnavailable, true + } + return ilmIncompleteError, true +} + +// ilmExpiryRefused reports whether reason is a refusal, which changes +// nothing, rather than a failure. +func ilmExpiryRefused(reason string) bool { + return reason == ilmIncompleteNotApplicable || reason == ilmIncompleteLocked +} + +// reportILMExpiryNotDone reports an ILM expiry of oi that ended with err. It +// reports nothing and returns false when err says the version is already +// gone; the caller reports that as it did before. +func reportILMExpiryNotDone(ctx context.Context, oi ObjectInfo, lcEvent lifecycle.Event, src lcEventSrc, err error, logID string, traceFn func(event string, metadata map[string]string, err error)) bool { + reason, counted := ilmExpiryNotDone(err) + if !counted { + return false + } + globalILMIncomplete.add(reason) + if ilmExpiryRefused(reason) { + traceFn(ILMExpiry, map[string]string{"result": "refused", "reason": reason, "version-id": oi.VersionID}, nil) + return true + } + traceFn(ILMExpiry, map[string]string{"result": "failed", "reason": reason, "version-id": oi.VersionID}, err) + tags := newLifecycleAuditEvent(src, lcEvent).Tags() + tags["version-id"] = oi.VersionID + tags["ilm-incomplete-reason"] = reason + auditLogInternal(ctx, AuditLogOptions{ + Event: ILMExpiry, + APIName: "ILMExpiry", + Bucket: oi.Bucket, + Object: oi.Name, + VersionID: oi.VersionID, + Error: err.Error(), + Tags: tags, + }) + ilmLogOnceIf(ctx, fmt.Errorf("ILM expiry of %s/%s (%s) not done: %w", oi.Bucket, oi.Name, oi.VersionID, err), logID) + return true +} + +// ilmExpiryEventName is the notification of an expiry that completed. An +// expiry that addressed no version ID (Expiration of a latest version in a +// versioned or suspended bucket) and whose result is a delete marker created +// that marker, also when the marker takes the null version ID of the version +// it replaces. Any other expiry removed the version it addressed, also when +// that version was itself a delete marker. +func ilmExpiryEventName(requestedVersionID string, result ObjectInfo) event.Name { + if requestedVersionID == "" && result.DeleteMarker { + return event.ObjectRemovedDeleteMarkerCreated + } + return event.ObjectRemovedDelete +} diff --git a/cmd/bucket-lifecycle-incomplete_test.go b/cmd/bucket-lifecycle-incomplete_test.go new file mode 100644 index 000000000..92b400d18 --- /dev/null +++ b/cmd/bucket-lifecycle-incomplete_test.go @@ -0,0 +1,325 @@ +// Copyright (c) 2026 PGSTY +// +// This file is part of Silo Object Storage stack +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "slices" + "testing" + "time" + + "github.com/minio/madmin-go/v3" + "github.com/minio/minio/internal/bucket/lifecycle" + "github.com/minio/minio/internal/bucket/versioning" + "github.com/minio/minio/internal/event" + "github.com/minio/minio/internal/logger" + loghttp "github.com/minio/minio/internal/logger/target/http" + "github.com/minio/minio/internal/pubsub" + xnet "github.com/pgsty/silo-pkg/v3/net" +) + +func TestILMExpiryNotDoneClassification(t *testing.T) { + for _, tc := range []struct { + err error + reason string + counted bool + }{ + {nil, "", false}, + {ObjectNotFound{}, "", false}, + {VersionNotFound{}, "", false}, + {errLifecycleNotApplicable, ilmIncompleteNotApplicable, true}, + {fmt.Errorf("wrapped: %w", errLifecycleNotApplicable), ilmIncompleteNotApplicable, true}, + {ObjectLocked{}, ilmIncompleteLocked, true}, + {InsufficientWriteQuorum{}, ilmIncompleteHistoryUnavailable, true}, + {InsufficientReadQuorum{}, ilmIncompleteHistoryUnavailable, true}, + {errFaultyDisk, ilmIncompleteError, true}, + } { + reason, counted := ilmExpiryNotDone(tc.err) + if reason != tc.reason || counted != tc.counted { + t.Errorf("%v: %q %v, want %q %v", tc.err, reason, counted, tc.reason, tc.counted) + } + } +} + +// fixedDeleteResults is an object layer whose bulk deletes return errs. +type fixedDeleteResults struct { + ObjectLayer + errs []error +} + +func (o fixedDeleteResults) DeleteObjects(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions) ([]DeletedObject, []error) { + deleted := make([]DeletedObject, len(objects)) + for i, obj := range objects { + deleted[i] = DeletedObject{ObjectName: obj.ObjectName, VersionID: obj.VersionID} + } + return deleted, slices.Clone(o.errs) +} + +// ilmTestAudit captures the ILMExpiry audit entries of bucket. +func ilmTestAudit(t *testing.T, bucket string) <-chan map[string]any { + t.Helper() + if len(logger.AuditTargets()) != 0 { + t.Fatal("unexpected pre-existing test audit targets") + } + entries := make(chan map[string]any, 16) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var entry struct { + Error string `json:"error"` + Tags map[string]any `json:"tags"` + API struct{ Name, Bucket string } + } + if err := json.NewDecoder(r.Body).Decode(&entry); err == nil && entry.API.Name == "ILMExpiry" && entry.API.Bucket == bucket { + tags := entry.Tags + if tags == nil { + tags = map[string]any{} + } + tags["error"] = entry.Error + entries <- tags + } + w.WriteHeader(http.StatusOK) + })) + endpoint, err := xnet.ParseHTTPURL(server.URL) + if err != nil { + t.Fatal(err) + } + ctx := context.Background() + if errs := logger.UpdateAuditWebhooks(ctx, map[string]loghttp.Config{"ilm": {Enabled: true, Name: "ilm", Endpoint: endpoint, BatchSize: 1, QueueSize: 128, MaxRetry: 1, RetryIntvl: time.Millisecond, HTTPTimeout: time.Second}}); len(errs) > 0 { + t.Fatal(errs) + } + targets := logger.AuditTargets() + t.Cleanup(func() { + logger.UpdateAuditWebhooks(ctx, nil) + for _, target := range targets { + target.Cancel() + } + server.Close() + }) + return entries +} + +// A refused or failed noncurrent purge is counted and traced with its reason, +// a failure is audited, and only the version that was removed is announced. +func TestDeleteObjectVersionsReportsRefusalAndFailure(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + audits := ilmTestAudit(t, s.bucket) + done := make(chan struct{}) + defer close(done) + events := make(chan event.Event, 8) + if err := globalHTTPListen.Subscribe(pubsub.MaskFromMaskable(event.ObjectRemovedAll), events, done, nil); err != nil { + t.Fatal(err) + } + traces := make(chan madmin.TraceInfo, 8) + if err := globalTrace.Subscribe(madmin.TraceILM, traces, done, nil); err != nil { + t.Fatal(err) + } + before := globalILMIncomplete.snapshot() + + layer := fixedDeleteResults{ObjectLayer: s.z, errs: []error{errLifecycleNotApplicable, InsufficientWriteQuorum{}, nil}} + toDel := []ObjectToDelete{ + {ObjectV: ObjectV{ObjectName: "ncv-a", VersionID: "11111111-1111-1111-1111-111111111111"}}, + {ObjectV: ObjectV{ObjectName: "ncv-b", VersionID: "22222222-2222-2222-2222-222222222222"}}, + {ObjectV: ObjectV{ObjectName: "ncv-c", VersionID: "33333333-3333-3333-3333-333333333333"}}, + } + evs := []lifecycle.Event{{Action: lifecycle.DeleteVersionAction}, {Action: lifecycle.DeleteVersionAction}, {Action: lifecycle.DeleteVersionAction}} + deleteObjectVersions(ctx, layer, s.bucket, toDel, evs) + + after := globalILMIncomplete.snapshot() + if d := after[ilmIncompleteNotApplicable] - before[ilmIncompleteNotApplicable]; d != 1 { + t.Errorf("not_applicable moved by %d, want 1", d) + } + if d := after[ilmIncompleteHistoryUnavailable] - before[ilmIncompleteHistoryUnavailable]; d != 1 { + t.Errorf("history_unavailable moved by %d, want 1", d) + } + var removed []string + results := map[string]string{} + for drained := false; !drained; { + select { + case ev := <-events: + removed = append(removed, ev.S3.Object.Key) + case tr := <-traces: + if r := tr.Custom["result"]; r != "" { + results[tr.Custom["version-id"]] = r + } + default: + drained = true + } + } + if !slices.Equal(removed, []string{"ncv-c"}) { + t.Errorf("removal events for %v, want only ncv-c", removed) + } + if results[toDel[0].VersionID] != "refused" || results[toDel[1].VersionID] != "failed" { + t.Errorf("traces %v, want refused and failed", results) + } + select { + case tags := <-audits: + if tags["ilm-incomplete-reason"] != ilmIncompleteHistoryUnavailable || tags["error"] == "" { + t.Errorf("audit of the failure: %v", tags) + } + case <-time.After(5 * time.Second): + t.Fatal("the failure was not audited") + } + select { + case tags := <-audits: + if _, failed := tags["ilm-incomplete-reason"]; failed { + t.Errorf("a second incomplete expiry was audited: %v", tags) + } + case <-time.After(500 * time.Millisecond): + } +} + +// expiryEvent returns the removal events sent while run runs. +func expiryEvents(t *testing.T, run func()) []event.Event { + t.Helper() + done := make(chan struct{}) + defer close(done) + events := make(chan event.Event, 8) + if err := globalHTTPListen.Subscribe(pubsub.MaskFromMaskable(event.ObjectRemovedAll), events, done, nil); err != nil { + t.Fatal(err) + } + run() + var got []event.Event + for { + select { + case ev := <-events: + got = append(got, ev) + default: + return got + } + } +} + +// The notification of a completed expiry names what its delete did: the +// delete marker it created, or the version it removed. +func TestExpiryEventsNameTheResult(t *testing.T) { + type expiry struct { + name string + status versioning.State + key string + versions []splitVersion + expire string // label of the version expired + action lifecycle.Action + wantEvent event.Name + wantID func(s *splitILM, key string) []string // acceptable version IDs + } + marker := func(s *splitILM, key string) []string { + oi, err := s.z.GetObjectInfo(s.t.Context(), s.bucket, key, ObjectOptions{}) + if !oi.DeleteMarker { + s.t.Fatalf("latest of %s is not a marker: %+v %v", key, oi, err) + } + return []string{oi.VersionID} + } + label := func(l string) func(s *splitILM, key string) []string { + return func(s *splitILM, key string) []string { return []string{s.ids[key+" "+l]} } + } + null := func(*splitILM, string) []string { return []string{"", nullVersionID} } + cases := []expiry{ + {"latest-versioned", versioning.Enabled, "exp-ev-latest", []splitVersion{splitAt("v1", 5, 0)}, "v1", lifecycle.DeleteAction, event.ObjectRemovedDeleteMarkerCreated, marker}, + {"noncurrent", versioning.Enabled, "ncv-ev-noncurrent", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 3, 0)}, "v1", lifecycle.DeleteVersionAction, event.ObjectRemovedDelete, label("v1")}, + {"expired-marker", versioning.Enabled, "edm-ev-marker", []splitVersion{splitMarker("dm", 5, 0)}, "dm", lifecycle.DeleteVersionAction, event.ObjectRemovedDelete, label("dm")}, + {"latest-suspended-null", versioning.Suspended, "exp-ev-null", []splitVersion{splitNull("v1", 5, 0)}, "v1", lifecycle.DeleteAction, event.ObjectRemovedDeleteMarkerCreated, null}, + {"tiered-latest-versioned", versioning.Enabled, "exp-ev-tiered", []splitVersion{splitTiered("v1", 5, 0)}, "v1", lifecycle.DeleteAction, event.ObjectRemovedDeleteMarkerCreated, marker}, + {"tiered-noncurrent", versioning.Enabled, "ncv-ev-tiered", []splitVersion{splitTiered("v1", 5, 0), splitAt("v2", 3, 0)}, "v1", lifecycle.DeleteVersionAction, event.ObjectRemovedDelete, label("v1")}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + s := newSplitILM(t, tc.status, splitILMRules) + installExpiryTestTier(t, map[string][]byte{"remote-v1": []byte("v1")}) + for _, v := range tc.versions { + s.write(tc.key, v) + } + vid := s.ids[tc.key+" "+tc.expire] + if vid == "" { + vid = nullVersionID + } + oi, err := s.z.GetObjectInfo(t.Context(), s.bucket, tc.key, ObjectOptions{VersionID: vid}) + if err != nil && !isErrMethodNotAllowed(err) { + t.Fatal(err) + } + oi.Bucket, oi.Name = s.bucket, tc.key + lcEvent := lifecycle.Event{Action: tc.action} + got := expiryEvents(t, func() { + if oi.TransitionedObject.Status != "" { + if !applyExpiryOnTransitionedObject(t.Context(), s.z, oi, lcEvent, lcEventSrc_Scanner) { + t.Fatal("transitioned expiry did not apply") + } + return + } + if !applyExpiryOnNonTransitionedObjects(t.Context(), s.z, oi, lcEvent, lcEventSrc_Scanner) { + t.Fatal("expiry did not apply") + } + }) + if len(got) != 1 { + t.Fatalf("events %+v, want one", got) + } + if got[0].EventName != tc.wantEvent { + t.Errorf("event %s, want %s", got[0].EventName, tc.wantEvent) + } + if want := tc.wantID(s, tc.key); !slices.Contains(want, got[0].S3.Object.VersionID) { + t.Errorf("event names version %q, want one of %v", got[0].S3.Object.VersionID, want) + } + }) + } +} + +// expiry_incomplete_total is exported as a counter with a reason label by +// both metrics versions. +func TestILMExpiryIncompleteMetrics(t *testing.T) { + // Metrics v3 exports a counter once it is above zero. + for _, r := range ilmIncompleteReasons { + globalILMIncomplete.add(r) + } + if md := ilmExpiryIncompleteTotalMD; md.Type != CounterMT || !slices.Equal(md.VariableLabels, []string{"reason"}) { + t.Fatalf("v3 descriptor %+v", md) + } + d := map[MetricName]MetricDescriptor{} + for _, md := range []MetricDescriptor{ilmExpiryPendingTasksMD, ilmTransitionActiveTasksMD, ilmTransitionPendingTasksMD, ilmTransitionMissedImmediateTasksMD, ilmVersionsScannedMD, ilmExpiryIncompleteTotalMD} { + d[md.Name] = md + } + m := newMetricValues(d) + if err := loadILMMetrics(t.Context(), m, nil); err != nil { + t.Fatal(err) + } + reasons := map[string]bool{} + for _, v := range m.values[expiryIncompleteTotal] { + reasons[v.Labels["reason"]] = true + } + for _, r := range ilmIncompleteReasons { + if !reasons[r] { + t.Errorf("v3: reason %q not exported", r) + } + } + reasons = map[string]bool{} + for _, mv := range getILMNodeMetrics().Get() { + if mv.Description.Name == expiryIncompleteTotal { + if mv.Description.Type != counterMetric { + t.Errorf("v2 type %v", mv.Description.Type) + } + reasons[mv.VariableLabels["reason"]] = true + } + } + for _, r := range ilmIncompleteReasons { + if !reasons[r] { + t.Errorf("v2: reason %q not exported", r) + } + } +} diff --git a/cmd/bucket-lifecycle.go b/cmd/bucket-lifecycle.go index d43d02c9b..b1732184b 100644 --- a/cmd/bucket-lifecycle.go +++ b/cmd/bucket-lifecycle.go @@ -39,7 +39,6 @@ import ( "github.com/minio/minio/internal/bucket/lifecycle" "github.com/minio/minio/internal/bucket/object/lock" "github.com/minio/minio/internal/bucket/replication" - "github.com/minio/minio/internal/event" xhttp "github.com/minio/minio/internal/http" "github.com/minio/minio/internal/logger" "github.com/minio/minio/internal/s3select" @@ -656,7 +655,8 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob // leaves a durable free-version record until the remote delete succeeds; // that record is the retry point if the process stops between the two // operations. - if _, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts); err != nil { + dobj, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts) + if err != nil { return err } @@ -669,14 +669,16 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob // Send audit for the lifecycle delete operation defer auditLogLifecycle(ctx, *oi, ILMExpiry, tags, traceFn) - eventName := event.ObjectRemovedDelete - if oi.DeleteMarker { - eventName = event.ObjectRemovedDeleteMarkerCreated - } + // Name what the delete did: a marker it created, or the version it + // removed. + eventName := ilmExpiryEventName(opts.VersionID, dobj) objInfo := ObjectInfo{ Name: oi.Name, - VersionID: oi.VersionID, - DeleteMarker: oi.DeleteMarker, + VersionID: dobj.VersionID, + DeleteMarker: dobj.DeleteMarker, + } + if objInfo.VersionID == "" { + objInfo.VersionID = opts.VersionID } // Notify object deleted event. sendEvent(eventArgs{ diff --git a/cmd/data-scanner.go b/cmd/data-scanner.go index 448f521b5..0f7584590 100644 --- a/cmd/data-scanner.go +++ b/cmd/data-scanner.go @@ -1210,10 +1210,7 @@ func applyTransitionRule(event lifecycle.Event, src lcEventSrc, obj ObjectInfo) func applyExpiryOnTransitionedObject(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, lcEvent lifecycle.Event, src lcEventSrc) (ok bool) { timeILM := globalScannerMetrics.timeILM(lcEvent.Action) if err := expireTransitionedObject(ctx, objLayer, &obj, lcEvent, src); err != nil { - if errors.Is(err, errLifecycleNotApplicable) || isErrObjectNotFound(err) || isErrVersionNotFound(err) { - return false - } - ilmLogIf(ctx, fmt.Errorf("expireTransitionedObject(%s, %s): %w", obj.Bucket, obj.Name, err)) + reportILMExpiryNotDone(ctx, obj, lcEvent, src, err, "transition-expiry"+obj.Name, globalLifecycleSys.trace(obj)) return false } timeILM(1) @@ -1272,14 +1269,10 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay return deleteAllNotCompleted(ctx, obj, lcEvent, src, purge, err, traceFn) } if err != nil { - if errors.Is(err, errLifecycleNotApplicable) || isErrObjectNotFound(err) || isErrVersionNotFound(err) { + // Assume it is still there, unless it is gone already. + if !reportILMExpiryNotDone(ctx, obj, lcEvent, src, err, "non-transition-expiry"+obj.Name, traceFn) { traceFn(ILMExpiry, nil, nil) - return false } - // Assume it is still there. - err := fmt.Errorf("DeleteObject(%s, %s): %w", obj.Bucket, obj.Name, err) - ilmLogOnceIf(ctx, err, "non-transition-expiry"+obj.Name) - traceFn(ILMExpiry, nil, err) return false } if dobj.Name == "" { @@ -1295,10 +1288,7 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay // Send audit for the lifecycle delete operation auditLogLifecycle(ctx, dobj, ILMExpiry, tags, traceFn) - eventName := event.ObjectRemovedDelete - if obj.DeleteMarker { - eventName = event.ObjectRemovedDeleteMarkerCreated - } + eventName := ilmExpiryEventName(opts.VersionID, dobj) switch lcEvent.Action { case lifecycle.DeleteAllVersionsAction: eventName = event.ObjectRemovedDeleteAllVersions diff --git a/cmd/erasure-server-pool-lifecycle-authority_test.go b/cmd/erasure-server-pool-lifecycle-authority_test.go index d75525356..196bde257 100644 --- a/cmd/erasure-server-pool-lifecycle-authority_test.go +++ b/cmd/erasure-server-pool-lifecycle-authority_test.go @@ -1245,6 +1245,15 @@ func TestPoolsLifecycleNullGenerations(t *testing.T) { // authoritySideEffects collects the version IDs that removal notifications // and ILM expiry audit traces report. func authoritySideEffects(t *testing.T, run func()) (notified, traced []string) { + t.Helper() + notified, traced, _ = authoritySideEffectsAll(t, run) + return notified, traced +} + +// authoritySideEffectsAll also returns, by version ID, the result an ILM +// trace reports for an expiry that did not remove its version: "refused" or +// "failed". Such traces are not removal traces. +func authoritySideEffectsAll(t *testing.T, run func()) (notified, traced []string, incomplete map[string]string) { t.Helper() done := make(chan struct{}) defer close(done) @@ -1257,19 +1266,25 @@ func authoritySideEffects(t *testing.T, run func()) (notified, traced []string) t.Fatal(err) } run() + incomplete = map[string]string{} // Both publishers deliver synchronously, so everything sent is buffered. for { select { case ev := <-events: notified = append(notified, ev.S3.Object.VersionID) case tr := <-traces: - if tr.FuncName == ILMExpiry { - traced = append(traced, tr.Custom["version-id"]) + if tr.FuncName != ILMExpiry { + continue } + if r := tr.Custom["result"]; r != "" { + incomplete[tr.Custom["version-id"]] = r + continue + } + traced = append(traced, tr.Custom["version-id"]) default: slices.Sort(notified) slices.Sort(traced) - return notified, traced + return notified, traced, incomplete } } } @@ -1283,9 +1298,12 @@ func TestPoolsLifecycleBatchQuorumNotAnnounced(t *testing.T) { id := s.ids[key+" old"] lifecycleReadDisks(t, s, key, 1, errDiskNotFound) result := &lifecycleBatchResult{ObjectLayer: s.z} - notified, traced := authoritySideEffects(t, func() { + notified, traced, incomplete := authoritySideEffectsAll(t, func() { deleteObjectVersions(t.Context(), result, s.bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: key, VersionID: id}}}, []lifecycle.Event{{Action: lifecycle.DeleteVersionAction}}) }) + if incomplete[id] != "failed" { + t.Errorf("unapplied expiry not reported as failed: %v", incomplete) + } if !isErrWriteQuorum(result.errs[0]) { t.Fatalf("fixture result: %v", result.errs[0]) } @@ -1330,9 +1348,15 @@ func TestPoolsLifecycleBatchSideEffects(t *testing.T) { evts[i] = lifecycle.Event{Action: lifecycle.DeleteVersionAction} } result := &lifecycleBatchResult{ObjectLayer: s.z} - notified, traced := authoritySideEffects(t, func() { + notified, traced, incomplete := authoritySideEffectsAll(t, func() { deleteObjectVersions(t.Context(), result, s.bucket, entries, evts) }) + if incomplete[entries[1].VersionID] != "failed" { + t.Errorf("failed entry: %v", incomplete) + } + if len(entries) > 3 && incomplete[entries[3].VersionID] != "refused" { + t.Errorf("refused entry: %v", incomplete) + } if result.errs[0] != nil || !isErrWriteQuorum(result.errs[1]) || result.errs[2] != nil && !isErrVersionNotFound(result.errs[2]) || len(entries) > 3 && !errors.Is(result.errs[3], errLifecycleNotApplicable) { diff --git a/cmd/erasure-server-pool-lifecycle-batch_test.go b/cmd/erasure-server-pool-lifecycle-batch_test.go index f3ea7c446..2f2797084 100644 --- a/cmd/erasure-server-pool-lifecycle-batch_test.go +++ b/cmd/erasure-server-pool-lifecycle-batch_test.go @@ -216,12 +216,18 @@ func TestPoolsLifecycleBatchRejectionNotAnnounced(t *testing.T) { } // Both publishers deliver synchronously, so everything sent is buffered. var notified, traced []string + refusedTraces := map[string]string{} for drained := false; !drained; { select { case ev := <-events: notified = append(notified, ev.S3.Object.VersionID) case tr := <-traces: if tr.FuncName == ILMExpiry { + if r := tr.Custom["result"]; r != "" { + // Not a removal: the expiry was refused or failed. + refusedTraces[tr.Custom["version-id"]] = r + continue + } traced = append(traced, tr.Custom["version-id"]) } default: @@ -234,4 +240,7 @@ func TestPoolsLifecycleBatchRejectionNotAnnounced(t *testing.T) { if len(traced) != 1 || traced[0] != eligible { t.Errorf("ILM expiry traces for %v, want only eligible %s (retained %s)", traced, eligible, retained) } + if refusedTraces[retained] != "refused" { + t.Errorf("the retained version's refusal is not traced: %v", refusedTraces) + } } diff --git a/cmd/metrics-v2.go b/cmd/metrics-v2.go index 120db3494..6932bbf69 100644 --- a/cmd/metrics-v2.go +++ b/cmd/metrics-v2.go @@ -2127,7 +2127,7 @@ func getILMNodeMetrics() *MetricsGroupV2 { trActiveTasks.Value = float64(globalTransitionState.ActiveTasks()) trMissedTasks.Value = float64(globalTransitionState.MissedImmediateTasks()) } - return []MetricV2{ + metrics := []MetricV2{ expPendingTasks, expMissedTasks, expMissedFreeVersions, @@ -2137,6 +2137,20 @@ func getILMNodeMetrics() *MetricsGroupV2 { trActiveTasks, trMissedTasks, } + for reason, n := range globalILMIncomplete.snapshot() { + metrics = append(metrics, MetricV2{ + Description: MetricDescription{ + Namespace: nodeMetricNamespace, + Subsystem: ilmSubsystem, + Name: expiryIncompleteTotal, + Help: "Total number of ILM expiry attempts since server start that ended without removing what they addressed, by reason", + Type: counterMetric, + }, + VariableLabels: map[string]string{"reason": reason}, + Value: float64(n), + }) + } + return metrics }) return mg } diff --git a/cmd/metrics-v3-ilm.go b/cmd/metrics-v3-ilm.go index 604beaae7..0d8738e9f 100644 --- a/cmd/metrics-v3-ilm.go +++ b/cmd/metrics-v3-ilm.go @@ -27,6 +27,7 @@ const ( transitionPendingTasks = "transition_pending_tasks" transitionMissedImmediateTasks = "transition_missed_immediate_tasks" versionsScanned = "versions_scanned" + expiryIncompleteTotal = "expiry_incomplete_total" ) var ( @@ -35,6 +36,7 @@ var ( ilmTransitionPendingTasksMD = NewGaugeMD(transitionPendingTasks, "Number of pending ILM transition tasks in the queue") ilmTransitionMissedImmediateTasksMD = NewCounterMD(transitionMissedImmediateTasks, "Number of missed immediate ILM transition tasks") ilmVersionsScannedMD = NewCounterMD(versionsScanned, "Total number of object versions checked for ILM actions since server start") + ilmExpiryIncompleteTotalMD = NewCounterMD(expiryIncompleteTotal, "Total number of ILM expiry attempts since server start that ended without removing what they addressed, by reason", "reason") ) // loadILMMetrics - `MetricsLoaderFn` for ILM metrics. @@ -48,6 +50,9 @@ func loadILMMetrics(_ context.Context, m MetricValues, _ *metricsCache) error { m.Set(transitionMissedImmediateTasks, float64(globalTransitionState.MissedImmediateTasks())) } m.Set(versionsScanned, float64(globalScannerMetrics.lifetime(scannerMetricILM))) + for reason, n := range globalILMIncomplete.snapshot() { + m.Set(expiryIncompleteTotal, float64(n), "reason", reason) + } return nil } diff --git a/cmd/metrics-v3.go b/cmd/metrics-v3.go index 62e4d89fd..16264c384 100644 --- a/cmd/metrics-v3.go +++ b/cmd/metrics-v3.go @@ -397,6 +397,7 @@ func newMetricGroups(r *prometheus.Registry) *metricsV3Collection { ilmTransitionPendingTasksMD, ilmTransitionMissedImmediateTasksMD, ilmVersionsScannedMD, + ilmExpiryIncompleteTotalMD, }, loadILMMetrics, ) diff --git a/cmd/object-handlers-common.go b/cmd/object-handlers-common.go index bbce6f021..f71cb1a77 100644 --- a/cmd/object-handlers-common.go +++ b/cmd/object-handlers-common.go @@ -473,8 +473,11 @@ func deleteObjectVersions(ctx context.Context, o ObjectLayer, bucket string, toD if errs[i] != nil && !isErrObjectNotFound(errs[i]) && !isErrVersionNotFound(errs[i]) { // Only a completed delete, or one confirmed already absent, is // a removal. After a refusal, a quorum, configuration or any - // other error the version may remain; do not audit or announce - // it. The next scan re-evaluates it. + // other error the version may remain; never announce it as + // removed. Report it as not done; the next scan re-evaluates + // it. + oi := ObjectInfo{Bucket: bucket, Name: toDel[i].ObjectName, VersionID: toDel[i].VersionID} + reportILMExpiryNotDone(ctx, oi, lcEvent[i], lcEventSrc_Scanner, errs[i], "noncurrent-expiry"+oi.Name, globalLifecycleSys.trace(oi)) continue } oi := ObjectInfo{