diff --git a/cmd/bucket-lifecycle-incomplete.go b/cmd/bucket-lifecycle-incomplete.go index 5f8bb30aa..c12ee23b2 100644 --- a/cmd/bucket-lifecycle-incomplete.go +++ b/cmd/bucket-lifecycle-incomplete.go @@ -140,7 +140,7 @@ func reportILMExpiryNotDone(ctx context.Context, oi ObjectInfo, lcEvent lifecycl } globalILMIncomplete.add(reason) if ilmExpiryRefused(reason) { - traceFn(ILMExpiry, map[string]string{"result": "refused", "reason": reason, "version-id": oi.VersionID}, nil) + traceFn(ILMExpiry, map[string]string{"result": "refused", "reason": reason, "version-id": oi.VersionID}, err) return true } traceFn(ILMExpiry, map[string]string{"result": "failed", "reason": reason, "version-id": oi.VersionID}, err) diff --git a/cmd/bucket-lifecycle-incomplete_test.go b/cmd/bucket-lifecycle-incomplete_test.go index 92b400d18..e6c570b7b 100644 --- a/cmd/bucket-lifecycle-incomplete_test.go +++ b/cmd/bucket-lifecycle-incomplete_test.go @@ -35,6 +35,7 @@ import ( loghttp "github.com/minio/minio/internal/logger/target/http" "github.com/minio/minio/internal/pubsub" xnet "github.com/pgsty/silo-pkg/v3/net" + "github.com/prometheus/client_golang/prometheus" ) func TestILMExpiryNotDoneClassification(t *testing.T) { @@ -85,7 +86,12 @@ func ilmTestAudit(t *testing.T, bucket string) <-chan map[string]any { var entry struct { Error string `json:"error"` Tags map[string]any `json:"tags"` - API struct{ Name, Bucket string } + API struct { + Name, Bucket string + Objects []struct { + ObjectName string `json:"objectName"` + } `json:"objects"` + } } if err := json.NewDecoder(r.Body).Decode(&entry); err == nil && entry.API.Name == "ILMExpiry" && entry.API.Bucket == bucket { tags := entry.Tags @@ -93,6 +99,9 @@ func ilmTestAudit(t *testing.T, bucket string) <-chan map[string]any { tags = map[string]any{} } tags["error"] = entry.Error + if len(entry.API.Objects) > 0 { + tags["object"] = entry.API.Objects[0].ObjectName + } entries <- tags } w.WriteHeader(http.StatusOK) @@ -159,6 +168,9 @@ func TestDeleteObjectVersionsReportsRefusalAndFailure(t *testing.T) { case tr := <-traces: if r := tr.Custom["result"]; r != "" { results[tr.Custom["version-id"]] = r + if tr.Error == "" { + t.Errorf("%s expiry of %s traced without its error", r, tr.Custom["version-id"]) + } } default: drained = true @@ -170,20 +182,40 @@ func TestDeleteObjectVersionsReportsRefusalAndFailure(t *testing.T) { 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) + // Drain every audit entry of the bucket: until the removal of ncv-c, + // audited after the refusal and the failure were reported, has arrived, + // and then while the target empties its queue. + var got []map[string]any + deadline := time.After(10 * time.Second) + for removal := false; !removal; { + select { + case tags := <-audits: + got = append(got, tags) + removal = removal || tags["object"] == "ncv-c" + case <-deadline: + t.Fatalf("audits %v: the removal of ncv-c was not audited", got) } - 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) + for idle := time.After(time.Second); idle != nil; { + select { + case tags := <-audits: + got = append(got, tags) + case <-idle: + idle = nil } - case <-time.After(500 * time.Millisecond): + } + perObject := map[any][]map[string]any{} + for _, tags := range got { + perObject[tags["object"]] = append(perObject[tags["object"]], tags) + } + if a := perObject["ncv-a"]; len(a) != 0 { + t.Errorf("the refusal was audited: %v", a) + } + if b := perObject["ncv-b"]; len(b) != 1 || b[0]["ilm-incomplete-reason"] != ilmIncompleteHistoryUnavailable || b[0]["error"] == "" { + t.Errorf("audits of the failure: %v, want one with its reason and error", b) + } + if c := perObject["ncv-c"]; len(c) != 1 || c[0]["ilm-incomplete-reason"] != nil { + t.Errorf("audits of the removal: %v, want one", c) } } @@ -288,24 +320,31 @@ func TestILMExpiryIncompleteMetrics(t *testing.T) { 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 { + // v3, through the collector the metrics handler registers. + families, err := newMetricGroups(prometheus.NewRegistry()).mgGatherers[ilmCollectorPath].Gather() + if err != nil { t.Fatal(err) } + want := ilmCollectorPath.metricPrefix() + "_" + string(expiryIncompleteTotal) reasons := map[string]bool{} - for _, v := range m.values[expiryIncompleteTotal] { - reasons[v.Labels["reason"]] = true + for _, family := range families { + if family.GetName() != want { + continue + } + for _, m := range family.Metric { + if m.Counter == nil || m.GetCounter().GetValue() <= 0 { + t.Errorf("v3 %s: %v, want a counter above zero", want, m) + } + for _, l := range m.Label { + if l.GetName() == "reason" { + reasons[l.GetValue()] = true + } + } + } } for _, r := range ilmIncompleteReasons { if !reasons[r] { - t.Errorf("v3: reason %q not exported", r) + t.Errorf("v3: %s{reason=%q} not exported", want, r) } } reasons = map[string]bool{} diff --git a/cmd/global-heal-lifecycle_test.go b/cmd/global-heal-lifecycle_test.go index 05fca264f..5d53a0093 100644 --- a/cmd/global-heal-lifecycle_test.go +++ b/cmd/global-heal-lifecycle_test.go @@ -30,15 +30,29 @@ import ( // is decided again under the object lock and can be refused, and a version // left unhealed would stay without the replaced drive for as long as it is. +// setTestStandardParity sets the standard storage class parity for the test +// and restores the whole previous configuration after it, also when it was +// never initialized and so meant the default parity. +func setTestStandardParity(t *testing.T, parity int) { + t.Helper() + storageclass.ConfigLock.RLock() + saved := globalStorageClass + storageclass.ConfigLock.RUnlock() + globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: parity}}) + t.Cleanup(func() { + storageclass.ConfigLock.Lock() + globalStorageClass = saved + storageclass.ConfigLock.Unlock() + }) +} + // startTestBackgroundHeal registers the background heal sequence and a heal // worker, which drive healing needs for its bucket heals. func startTestBackgroundHeal(t *testing.T, z *erasureServerPools) { t.Helper() // The healed drive is rebuilt from parity. Earlier tests can leave a // storage class of parity 0 behind (newTestConfig on a single drive). - previous := globalStorageClass - globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: z.serverPools[0].setDriveCount / 2}}) - t.Cleanup(func() { globalStorageClass.Update(previous) }) + setTestStandardParity(t, z.serverPools[0].setDriveCount/2) if globalBackgroundHealState == nil { globalBackgroundHealState = newHealState(GlobalContext, false) } diff --git a/cmd/mrf.go b/cmd/mrf.go index 31f99c45b..e23884081 100644 --- a/cmd/mrf.go +++ b/cmd/mrf.go @@ -74,11 +74,20 @@ func newMRFState() mrfState { } } +// mrfPartialOpHook, when set, receives each partial operation in place of +// the queue, so that a test observes what is queued without racing the heal +// routines that consume it. Production never sets it. +var mrfPartialOpHook atomic.Pointer[func(PartialOperation)] + // Add a partial S3 operation (put/delete) when one or more disks are offline. func (m *mrfState) addPartialOp(op PartialOperation) { if m == nil { return } + if f := mrfPartialOpHook.Load(); f != nil { + (*f)(op) + return + } if atomic.LoadInt32(&m.closed) == 1 { return diff --git a/cmd/restore-two-phase_test.go b/cmd/restore-two-phase_test.go index 3c2e1b887..f7c9b2ad2 100644 --- a/cmd/restore-two-phase_test.go +++ b/cmd/restore-two-phase_test.go @@ -27,6 +27,7 @@ import ( "os" "path/filepath" "strings" + "sync" "sync/atomic" "testing" "time" @@ -862,9 +863,7 @@ func TestRestorePublicationBelowWriteQuorumOnWideSet(t *testing.T) { // the convergence path queues too. That heal repairs the null version on the // drive that missed the commit, also where the null version is noncurrent. func TestRestorePartialCommitQueuesNullVersionHeal(t *testing.T) { - saved := globalStorageClass - globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: 2}}) - t.Cleanup(func() { globalStorageClass.Update(saved) }) + setTestStandardParity(t, 2) r := newRestoreSafetyTest(t, 1, versioning.Suspended) oi, body := nativeRestoreSeed(t, r, 0, true, false) // A version written while versioning was enabled makes the null version @@ -878,22 +877,26 @@ func TestRestorePartialCommitQueuesNullVersionHeal(t *testing.T) { failing := &restoreTwoPhaseFaults{} failing.rename.Store("fail") disks := restoreTwoPhaseFaulty(t, r, nil, nil, nil, failing) - // The test owns the MRF queue while the commit runs: the heal routines of - // the test servers would otherwise take the entry first. - queue := make(chan PartialOperation, mrfOpsQueueSize) - previous := globalMRFState.opCh - globalMRFState.opCh = queue + // Observe the MRF entries in place of the queue, which the heal routines + // of the test servers consume. + var mu sync.Mutex + var queued []PartialOperation + hook := func(op PartialOperation) { + if op.Bucket == r.bucket && op.Object == r.object { + mu.Lock() + queued = append(queued, op) + mu.Unlock() + } + } + mrfPartialOpHook.Store(&hook) + t.Cleanup(func() { mrfPartialOpHook.Store(nil) }) err := r.z.RestoreTransitionedObject(t.Context(), r.bucket, r.object, nativeRestoreOptions(r, oi.VersionID)) - globalMRFState.opCh = previous + mrfPartialOpHook.Store(nil) if err != nil { t.Fatalf("restore with one drive failing its commit: %v", err) } - var queued []PartialOperation - for len(queue) > 0 { - if op := <-queue; op.Bucket == r.bucket && op.Object == r.object { - queued = append(queued, op) - } - } + mu.Lock() + defer mu.Unlock() if len(queued) != 1 || queued[0].VersionID != nullVersionID { t.Fatalf("queued heals %+v, want one of version %q", queued, nullVersionID) }