diff --git a/cmd/admin-handlers.go b/cmd/admin-handlers.go
index e4f06bc9f..511e9bc71 100644
--- a/cmd/admin-handlers.go
+++ b/cmd/admin-handlers.go
@@ -791,7 +791,7 @@ func (a adminAPIHandlers) MetricsHandler(w http.ResponseWriter, r *http.Request)
enc := json.NewEncoder(w)
for n > 0 {
- var m madmin.RealtimeMetrics
+ var m realtimeMetrics
mLocal := collectLocalMetrics(types, collectMetricsOpts{
hosts: hostMap,
disks: diskMap,
@@ -2675,7 +2675,7 @@ func fetchHealthInfo(healthCtx context.Context, objectAPI ObjectLayer, query *ur
// collect all realtime metrics except disk
// disk metrics are already included under drive info of each server
getRealtimeMetrics := func() *madmin.RealtimeMetrics {
- var m madmin.RealtimeMetrics
+ var m realtimeMetrics
types := madmin.MetricsAll &^ madmin.MetricsDisk
mLocal := collectLocalMetrics(types, collectMetricsOpts{})
m.Merge(&mLocal)
@@ -2690,7 +2690,8 @@ func fetchHealthInfo(healthCtx context.Context, objectAPI ObjectLayer, query *ur
m.ByHost[anonAddr(host)] = metrics
delete(m.ByHost, host)
}
- return &m
+ legacy := m.legacy()
+ return &legacy
}
anonymizeCmdLine := func(cmdLine string) string {
diff --git a/cmd/batch-handlers.go b/cmd/batch-handlers.go
index e86af95a8..5e3d4e48a 100644
--- a/cmd/batch-handlers.go
+++ b/cmd/batch-handlers.go
@@ -462,6 +462,7 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
if done, err := ri.startAuthorized(ctx, api, job); done || err != nil {
return err
}
+ ri.resetReplicationAttempt()
globalBatchJobsMetrics.save(job.ID, ri)
retryAttempts := job.Replicate.Flags.Retry.Attempts
@@ -576,9 +577,10 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
s3Type := r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3
minioSrc := r.Source.Type == BatchJobReplicateResourceMinIO
ctx, cancelCause := context.WithCancelCause(ctx)
- // refuse ends the job for its submitter: record why before stopping.
- refuse := func(err error) {
- ri.refuse(err)
+ stop := func(err error) {
+ if isBatchJobAuthzError(err) {
+ ri.refuse(err)
+ }
cancelCause(err)
}
// Each run lists the source from the start and counts every locked
@@ -615,6 +617,15 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
// Refused or canceled: schedule nothing more.
continue
}
+ if obj.Err != nil {
+ switch minio.ToErrorResponse(obj.Err).StatusCode {
+ case http.StatusUnauthorized, http.StatusForbidden:
+ stop(batchJobAuthzError{reason: "source listing denied: " + obj.Err.Error()})
+ default:
+ stop(batchListingError(obj.Err))
+ }
+ continue
+ }
oi := toObjectInfo(r.Source.Bucket, obj.Key, obj)
if !minioSrc {
// Check if metadata filter was requested and it is expected to have
@@ -663,8 +674,11 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
stopFn := globalBatchJobsMetrics.trace(batchJobMetricReplication, job.ID, attempts)
success := true
if err := r.ReplicateFromSource(ctx, api, core, oi, retry, job.Submitter); err != nil {
- if isBatchJobAuthzError(err) {
- refuse(err)
+ if isBatchJobStopped(err) {
+ stop(err)
+ return
+ }
+ if ctx.Err() != nil {
return
}
// A locked destination version is left as it is: the
@@ -706,6 +720,14 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
}
return nil
}
+ if cause := context.Cause(ctx); cause != nil {
+ cancelCause(nil)
+ if isBatchJobUnavailable(cause) {
+ ri.resetReplicationAttempt()
+ return ri.savePaused(ctx, api, job, cause)
+ }
+ return cause
+ }
ri.RetryAttempts = attempts
ri.Complete = ri.ObjectsFailed == 0 && ri.DeleteMarkersFailed == 0
@@ -721,14 +743,7 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
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
+ ri.resetReplicationAttempt()
retry = true // indicate we are retrying..
time.Sleep(delay + time.Duration(rnd.Float64()*float64(delay)))
continue
@@ -1337,6 +1352,21 @@ func (ri *batchJobInfo) trackCurrentBucketBatch(bucket string, batch []ObjectInf
}
}
+func (ri *batchJobInfo) resetReplicationAttempt() {
+ ri.mu.Lock()
+ defer ri.mu.Unlock()
+ ri.Complete, ri.Failed = false, false
+ ri.ObjectsFailed = 0
+ ri.DeleteMarkersFailed = 0
+ ri.Bucket = ""
+ ri.Object = ""
+ ri.Objects = 0
+ ri.DeleteMarkers = 0
+ ri.BytesFailed = 0
+ ri.BytesTransferred = 0
+ ri.VersionsSkippedLocked = 0
+}
+
// Start start the batch replication job, resumes if there was a pending job via "job.ID"
func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job BatchJobRequest) error {
ri := &batchJobInfo{
@@ -1353,8 +1383,8 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
if done, err := ri.startAuthorized(ctx, api, job); done || err != nil {
return err
}
+ ri.resetReplicationAttempt()
globalBatchJobsMetrics.save(job.ID, ri)
- lastObject := ri.Object
retryAttempts := job.Replicate.Flags.Retry.Attempts
if retryAttempts <= 0 {
@@ -1461,9 +1491,10 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
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)
+ stop := func(err error) {
+ if isBatchJobAuthzError(err) {
+ ri.refuse(err)
+ }
cancelCause(err)
}
@@ -1488,8 +1519,11 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
writeFn := func(batch []ObjectInfo) {
if len(batch) > 0 {
if err := r.writeAsArchive(ctx, api, cl, batch, r.Target.Prefix, job.Submitter); err != nil {
- if isBatchJobAuthzError(err) {
- refuse(err)
+ if isBatchJobStopped(err) {
+ stop(err)
+ return
+ }
+ if ctx.Err() != nil {
return
}
batchLogOnceIf(ctx, err, job.ID+"writeAsArchive")
@@ -1505,7 +1539,10 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
}
}
for obj := range walkCh {
- if obj.Item.DeleteMarker || !obj.Item.VersionPurgeStatus.Empty() || obj.Item.Size >= int64(smallerThan) {
+ if context.Cause(ctx) != nil {
+ continue
+ }
+ if obj.Err != nil || obj.Item.DeleteMarker || !obj.Item.VersionPurgeStatus.Empty() || obj.Item.Size >= int64(smallerThan) {
slowCh <- obj
continue
}
@@ -1518,7 +1555,9 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
writeFn(batch)
batch = batch[:0]
}
- writeFn(batch)
+ if context.Cause(ctx) == nil {
+ writeFn(batch)
+ }
xioutil.SafeClose(slowCh)
}()
} else {
@@ -1551,11 +1590,10 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
for _, prefix := range prefixes {
prefixWalkCh := make(chan itemOrErr[ObjectInfo], 100)
if err := api.Walk(ctx, r.Source.Bucket, prefix, prefixWalkCh, WalkOptions{
- Marker: lastObject,
Filter: selectObj,
AskDisks: walkQuorum,
}); err != nil {
- cancelCause(err)
+ cancelCause(batchListingError(err))
xioutil.SafeClose(walkCh)
return
}
@@ -1575,7 +1613,7 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
continue
}
if res.Err != nil {
- ri.Failed = true
+ cancelCause(batchListingError(res.Err))
batchLogOnceIf(ctx, res.Err, job.ID+"res.Err")
continue
}
@@ -1594,8 +1632,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, job.Submitter); err != nil {
- if isBatchJobAuthzError(err) {
- refuse(err)
+ if isBatchJobStopped(err) {
+ stop(err)
+ return
+ }
+ if ctx.Err() != nil {
return
}
if minio.ToErrorResponse(err).Code == "PreconditionFailed" {
@@ -1632,9 +1673,13 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
}
return nil
}
- // Do not need to retry if we can't list objects on source.
- if context.Cause(ctx) != nil {
- return context.Cause(ctx)
+ if cause := context.Cause(ctx); cause != nil {
+ cancelCause(nil)
+ if isBatchJobUnavailable(cause) {
+ ri.resetReplicationAttempt()
+ return ri.savePaused(ctx, api, job, cause)
+ }
+ return cause
}
ri.RetryAttempts = attempts
ri.Complete = ri.ObjectsFailed == 0 && ri.DeleteMarkersFailed == 0
@@ -1650,14 +1695,7 @@ 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
+ ri.resetReplicationAttempt()
retry = true // indicate we are retrying..
time.Sleep(delay + time.Duration(rnd.Float64()*float64(delay)))
continue
@@ -2627,20 +2665,20 @@ func (ri *batchJobInfo) metric() madmin.JobMetric {
return m
}
-func (m *batchJobMetrics) report(jobID string) (metrics *madmin.BatchJobMetrics) {
- metrics = &madmin.BatchJobMetrics{CollectedAt: time.Now(), Jobs: make(map[string]madmin.JobMetric)}
+func (m *batchJobMetrics) report(jobID string) (metrics *batchJobMetricsReport) {
+ metrics = &batchJobMetricsReport{CollectedAt: time.Now(), Jobs: make(map[string]batchJobStatusMetric)}
m.RLock()
defer m.RUnlock()
if jobID != "" {
if job, ok := m.metrics[jobID]; ok {
- metrics.Jobs[jobID] = job.metric()
+ metrics.Jobs[jobID] = job.statusMetric()
}
return metrics
}
for id, job := range m.metrics {
- metrics.Jobs[id] = job.metric()
+ metrics.Jobs[id] = job.statusMetric()
}
return metrics
}
diff --git a/cmd/batch-progress.go b/cmd/batch-progress.go
index a8c4eb824..79e316909 100644
--- a/cmd/batch-progress.go
+++ b/cmd/batch-progress.go
@@ -170,7 +170,7 @@ type batchJobStatus struct {
type batchJobStatusMetric struct {
madmin.JobMetric
- VersionsSkippedLocked int64 `json:"versionsSkippedLocked,omitempty"`
+ VersionsSkippedLocked *int64 `json:"versionsSkippedLocked,omitempty"`
KeysRefused int64 `json:"keysRefused,omitempty"`
FailureReason string `json:"failureReason,omitempty"`
PausedReason string `json:"pausedReason,omitempty"`
@@ -179,9 +179,10 @@ type batchJobStatusMetric struct {
func (ri *batchJobInfo) statusMetric() batchJobStatusMetric {
ri.mu.RLock()
defer ri.mu.RUnlock()
+ versionsSkipped := ri.VersionsSkippedLocked
return batchJobStatusMetric{
JobMetric: ri.metric(),
- VersionsSkippedLocked: ri.VersionsSkippedLocked,
+ VersionsSkippedLocked: &versionsSkipped,
KeysRefused: ri.KeysRefused,
FailureReason: ri.FailureReason,
PausedReason: ri.PausedReason,
diff --git a/cmd/batch-replicate-retry_test.go b/cmd/batch-replicate-retry_test.go
new file mode 100644
index 000000000..2a88a3f9f
--- /dev/null
+++ b/cmd/batch-replicate-retry_test.go
@@ -0,0 +1,343 @@
+package cmd
+
+import (
+ "archive/tar"
+ "context"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "strconv"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ polplugin "github.com/minio/minio/internal/config/policy/plugin"
+ xnet "github.com/pgsty/silo-pkg/v3/net"
+ "github.com/pgsty/silo-pkg/v3/policy"
+)
+
+type batchReplicationWriteObserver struct {
+ ObjectLayer
+ written func(string, string)
+}
+
+type batchReplicationWalkFailure struct {
+ ObjectLayer
+ streamed bool
+}
+
+func (failure batchReplicationWalkFailure) Walk(ctx context.Context, _, _ string, results chan<- itemOrErr[ObjectInfo], _ WalkOptions) error {
+ if !failure.streamed {
+ return errErasureReadQuorum
+ }
+ go func() {
+ defer close(results)
+ select {
+ case results <- itemOrErr[ObjectInfo]{Err: errErasureReadQuorum}:
+ case <-ctx.Done():
+ }
+ }()
+ return nil
+}
+
+func TestBatchReplicationListingUnavailable(t *testing.T) {
+ for _, snowball := range []bool{false, true} {
+ for _, streamed := range []bool{false, true} {
+ t.Run(fmt.Sprintf("snowball=%t/streamed=%t", snowball, streamed), func(t *testing.T) {
+ fixture := newBatchAuthzFixture(t)
+ job := BatchJobRequest{
+ ID: "replicate-listing" + getKeySeparator() + "0", Started: UTCNow(), Submitter: batchRootSubmitter(t),
+ Replicate: &BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{
+ Type: BatchJobReplicateResourceMinIO, Bucket: fixture.bucket,
+ Snowball: BatchJobSnowball{
+ Disable: ptr(!snowball), InMemory: ptr(true), Compress: ptr(false), SkipErrs: ptr(false),
+ Batch: ptr(1), SmallerThan: ptr("1MiB"),
+ },
+ },
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: "remote", Endpoint: "https://target.invalid"},
+ },
+ }
+ api := batchReplicationWalkFailure{ObjectLayer: fixture.obj, streamed: streamed}
+ if err := job.Replicate.Start(fixture.ctx, api, job); !isBatchJobUnavailable(err) {
+ t.Fatalf("listing failure did not pause: %v", err)
+ }
+ report := fixture.jobInfo(t, &job)
+ if report.PausedReason == "" || report.Failed || report.Complete || report.Object != "" || report.ObjectsFailed != 0 {
+ t.Fatalf("listing failure saved an invalid outcome: %+v", report)
+ }
+ })
+ }
+ }
+}
+
+func (observer batchReplicationWriteObserver) PutObject(ctx context.Context, bucket, object string, reader *PutObjReader, opts ObjectOptions) (ObjectInfo, error) {
+ info, err := observer.ObjectLayer.PutObject(ctx, bucket, object, reader, opts)
+ if err == nil {
+ observer.written(bucket, object)
+ }
+ return info, err
+}
+
+func TestBatchReplicationRemoteListingDenied(t *testing.T) {
+ fixture := newBatchAuthzFixture(t)
+ previousTransport := remoteInstanceTransport
+ remoteInstanceTransport = atomic.Value{}
+ setRemoteInstanceTransport(batchLockTransport(func(request *http.Request) (*http.Response, error) {
+ status := http.StatusForbidden
+ body := `AccessDeniedsource listing denied`
+ if request.URL.Query().Has("location") {
+ status = http.StatusOK
+ body = `us-east-1`
+ }
+ return &http.Response{
+ StatusCode: status, Header: make(http.Header),
+ Body: io.NopCloser(strings.NewReader(body)), ContentLength: int64(len(body)), Request: request,
+ }, nil
+ }))
+ defer func() { remoteInstanceTransport = previousTransport }()
+ job := BatchJobRequest{
+ ID: "replicate-listing-denied" + getKeySeparator() + "0", Started: UTCNow(), Submitter: batchRootSubmitter(t),
+ Replicate: &BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{
+ Type: BatchJobReplicateResourceMinIO, Bucket: "remote", Endpoint: "https://source.invalid",
+ Creds: BatchJobReplicateCredentials{AccessKey: "test-access", SecretKey: "test-secret"},
+ },
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: fixture.bucket},
+ },
+ }
+ if err := job.Replicate.StartFromSource(fixture.ctx, fixture.obj, job); err != nil {
+ t.Fatalf("explicit remote denial was not terminal: %v", err)
+ }
+ report := fixture.jobInfo(t, &job)
+ if !report.Failed || report.Complete || report.FailureReason == "" || report.PausedReason != "" {
+ t.Fatalf("remote denial saved an invalid outcome: %+v", report)
+ }
+}
+
+func TestBatchReplicationDependencyRecovery(t *testing.T) {
+ for _, mode := range []string{"push", "pull", "snowball"} {
+ for _, denied := range []bool{false, true} {
+ t.Run(fmt.Sprintf("%s/denied=%t", mode, denied), func(t *testing.T) {
+ t.Setenv("_MINIO_BATCH_REPLICATION_WORKERS", "2")
+ fixture := newBatchAuthzFixture(t)
+ keys := []string{"a-done", "b-pending", "c-later"}
+ versions := make(map[string]string)
+ for _, key := range keys {
+ versions[key] = mustGetUUID()
+ if mode != "pull" {
+ versions[key] = batchPutVersion(t, fixture.obj, fixture.bucket, key, nil).VersionID
+ }
+ }
+ laterDone := make(chan struct{})
+ signalLater := sync.OnceFunc(func() { close(laterDone) })
+ var writtenMu sync.Mutex
+ written := make(map[string]bool)
+ recordWrite := func(bucket, key string) {
+ if bucket != fixture.bucket && bucket != "remote" {
+ return
+ }
+ writtenMu.Lock()
+ written[key] = true
+ writtenMu.Unlock()
+ if key == "c-later" {
+ signalLater()
+ }
+ }
+ var outage atomic.Bool
+ outage.Store(true)
+ var decisions atomic.Int32
+ pluginServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+ var input struct {
+ Input policy.Args `json:"input"`
+ }
+ if err := json.NewDecoder(request.Body).Decode(&input); err != nil {
+ t.Error(err)
+ writer.WriteHeader(http.StatusBadRequest)
+ return
+ }
+ action := policy.Action(policy.GetObjectAction)
+ if mode == "pull" {
+ action = policy.PutObjectAction
+ }
+ if outage.Load() && input.Input.Action == action && input.Input.ObjectName == "b-pending" {
+ decisions.Add(1)
+ if mode != "snowball" {
+ select {
+ case <-laterDone:
+ case <-request.Context().Done():
+ return
+ }
+ }
+ if denied {
+ io.WriteString(writer, `{"result":false}`)
+ return
+ }
+ writer.WriteHeader(http.StatusServiceUnavailable)
+ }
+ io.WriteString(writer, `{"result":true}`)
+ }))
+ defer pluginServer.Close()
+ pluginURL, err := xnet.ParseHTTPURL(pluginServer.URL)
+ mustIAM(t, err)
+ previousPlugin := newGlobalAuthZPluginFn()
+ setGlobalAuthZPlugin(polplugin.New(polplugin.Args{
+ URL: pluginURL, CloseRespFn: func(reader io.ReadCloser) { reader.Close() },
+ }))
+ defer setGlobalAuthZPlugin(previousPlugin)
+ previousTransport := remoteInstanceTransport
+ remoteInstanceTransport = atomic.Value{}
+ setRemoteInstanceTransport(batchLockTransport(func(request *http.Request) (*http.Response, error) {
+ if err := request.Context().Err(); err != nil {
+ return nil, err
+ }
+ body := ""
+ header := make(http.Header)
+ query := request.URL.Query()
+ switch {
+ case query.Has("location"):
+ body = `us-east-1`
+ case query.Has("versions"):
+ body = `remotefalse`
+ for _, key := range keys {
+ body += fmt.Sprintf(`%s%strue2026-11-01T00:00:00.000Z"etag"7STANDARD`, key, versions[key])
+ }
+ body += ``
+ case request.Method == http.MethodGet:
+ body = "payload"
+ header.Set("Etag", `"etag"`)
+ header.Set("Last-Modified", UTCNow().Format(http.TimeFormat))
+ header.Set("X-Amz-Version-Id", query.Get("versionId"))
+ case request.Method == http.MethodPut:
+ if mode == "snowball" && request.Body != nil {
+ archive := tar.NewReader(request.Body)
+ for {
+ entry, readErr := archive.Next()
+ if readErr == io.EOF {
+ break
+ }
+ if readErr != nil {
+ return nil, readErr
+ }
+ recordWrite("remote", entry.Name)
+ }
+ } else if mode != "snowball" {
+ if _, err := io.Copy(io.Discard, request.Body); err != nil {
+ return nil, err
+ }
+ recordWrite("remote", strings.TrimPrefix(request.URL.Path, "/remote/"))
+ }
+ header.Set("Etag", `"etag"`)
+ default:
+ return nil, fmt.Errorf("unexpected replication request: %s %s", request.Method, request.URL)
+ }
+ header.Set("Content-Length", strconv.Itoa(len(body)))
+ return &http.Response{
+ StatusCode: http.StatusOK, Header: header,
+ Body: io.NopCloser(strings.NewReader(body)), ContentLength: int64(len(body)), Request: request,
+ }, nil
+ }))
+ defer func() { remoteInstanceTransport = previousTransport }()
+ job := &BatchJobRequest{
+ ID: "replicate-retry" + getKeySeparator() + "0", Started: UTCNow(), Submitter: batchRootSubmitter(t),
+ Replicate: &BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: fixture.bucket},
+ Target: BatchJobReplicateTarget{
+ Type: BatchJobReplicateResourceMinIO, Bucket: "remote", Endpoint: "https://target.invalid",
+ Creds: BatchJobReplicateCredentials{AccessKey: "test-access", SecretKey: "test-secret"},
+ },
+ },
+ }
+ if mode == "pull" {
+ job.Replicate.Source.Endpoint = job.Replicate.Target.Endpoint
+ job.Replicate.Source.Creds = job.Replicate.Target.Creds
+ job.Replicate.Source.Bucket = "remote"
+ job.Replicate.Target.Endpoint = ""
+ job.Replicate.Target.Bucket = fixture.bucket
+ }
+ if mode == "snowball" {
+ job.Replicate.Source.Snowball = BatchJobSnowball{
+ Disable: ptr(false), InMemory: ptr(true), Compress: ptr(false), SkipErrs: ptr(false),
+ Batch: ptr(1), SmallerThan: ptr("1MiB"),
+ }
+ }
+ job.Replicate.Flags.Retry.Attempts = 1
+ job.Replicate.Flags.Retry.Delay = time.Millisecond
+ interrupted := &batchJobInfo{
+ JobID: job.ID, JobType: string(job.Type()), StartTime: job.Started,
+ Object: "c-later", Objects: 2,
+ }
+ mustIAM(t, interrupted.updateAfter(fixture.ctx, fixture.obj, 0, *job))
+ pool := globalBatchJobPool
+ pool.objLayer = batchReplicationWriteObserver{ObjectLayer: fixture.obj, written: recordWrite}
+ ctx, cancel := context.WithCancel(fixture.ctx)
+ pool.ctx = ctx
+ mustIAM(t, pool.queueJob(job))
+ done := make(chan struct{})
+ go func() { defer close(done); pool.AddWorker() }()
+ defer func() {
+ cancel()
+ <-done
+ if t.Failed() {
+ t.Logf("last report: %+v", fixture.jobInfo(t, job))
+ }
+ }()
+ state := "paused"
+ if denied {
+ state = "terminal"
+ }
+ waitBatchState(t, pool, job.ID, state)
+ report := fixture.jobInfo(t, job)
+ if decisions.Load() == 0 {
+ t.Fatal("test did not reach per-object authorization")
+ }
+ if denied {
+ if !report.Failed || report.Complete || report.FailureReason == "" || report.PausedReason != "" {
+ t.Fatalf("denial was not terminal: %+v", report)
+ }
+ pool.retryDue(time.Now().Add(time.Hour))
+ if len(pool.jobCh) != 0 {
+ t.Fatal("denied job was retried")
+ }
+ return
+ }
+ if report.Failed || report.Complete || report.FailureReason != "" || report.PausedReason == "" ||
+ report.Object != "" || report.Objects != 0 || report.ObjectsFailed != 0 {
+ t.Fatalf("outage was not saved with a safe restart point: %+v", report)
+ }
+ pool.jmu.Lock()
+ owner := pool.jobOwners[job.ID]
+ sameOwner := owner.req == job && owner.retries == 1
+ pool.jmu.Unlock()
+ if !sameOwner {
+ t.Fatal("pause lost the job owner")
+ }
+ outage.Store(false)
+ pool.retryDue(time.Now().Add(time.Hour))
+ waitBatchState(t, pool, job.ID, "terminal")
+ report = fixture.jobInfo(t, job)
+ if !report.Complete || report.Failed || report.PausedReason != "" || report.Objects != int64(len(keys)) || report.ObjectsFailed != 0 {
+ t.Fatalf("retry did not finish with consistent counts: %+v", report)
+ }
+ writtenMu.Lock()
+ for _, key := range keys {
+ if !written[key] {
+ t.Errorf("retry skipped %s", key)
+ }
+ }
+ writtenMu.Unlock()
+ if mode == "pull" {
+ for _, key := range keys {
+ if !batchVersionExists(t, fixture.obj, fixture.bucket, key, versions[key]) {
+ t.Errorf("retry lost remote version of %s", key)
+ }
+ }
+ }
+ })
+ }
+ }
+}
diff --git a/cmd/iam-load-user-race_test.go b/cmd/iam-load-user-race_test.go
index b798a9d6b..1b684f78a 100644
--- a/cmd/iam-load-user-race_test.go
+++ b/cmd/iam-load-user-race_test.go
@@ -5,22 +5,44 @@ package cmd
import (
"context"
+ "errors"
+ "slices"
"sync"
+ "sync/atomic"
"testing"
"time"
+
+ "github.com/pgsty/silo-pkg/v3/policy"
)
// Hold a lazy user load after it has read storage. This makes the ordering
// between an in-flight read and a completed credential edit deterministic.
type iamLoadUserBlocker struct {
IAMStorageAPI
- key string
- waiting chan struct{}
- release chan struct{}
- once sync.Once
+ key string
+ waiting chan struct{}
+ release chan struct{}
+ once sync.Once
+ loads atomic.Int32
+ retryWaiting chan struct{}
+ retryRelease chan struct{}
+ retryErr error
}
func (s *iamLoadUserBlocker) loadUser(ctx context.Context, key string, userType IAMUserType, out map[string]UserIdentity) error {
+ if key == s.key && userType == regUser && s.loads.Add(1) == 2 {
+ if s.retryWaiting != nil {
+ close(s.retryWaiting)
+ select {
+ case <-s.retryRelease:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ }
+ if s.retryErr != nil {
+ return s.retryErr
+ }
+ }
err := s.IAMStorageAPI.loadUser(ctx, key, userType, out)
if err == nil && key == s.key && userType == regUser {
s.once.Do(func() { close(s.waiting) })
@@ -33,6 +55,230 @@ func (s *iamLoadUserBlocker) loadUser(ctx context.Context, key string, userType
return err
}
+func TestIAMLoadUserConcurrentUpdates(t *testing.T) {
+ for _, change := range []string{"unrelated-edit", "unrelated-load", "delete", "mapping", "policy"} {
+ t.Run(change, func(t *testing.T) {
+ ctx, sys, _ := prepareIAMRevisionFixture(t)
+ const policyName = "lazy-policy"
+ batchAuthzPolicy(ctx, t, sys, policyName, `{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::bucket/*"]}`)
+ cold := iamAdminUser(ctx, t, sys, "lazy-cold", policyName)
+ other := iamAdminUser(ctx, t, sys, "lazy-other", "readwrite")
+ cache := sys.store.lock()
+ delete(cache.iamUsersMap, cold.AccessKey)
+ sys.store.unlock()
+ original := sys.store.IAMStorageAPI
+ blocker := &iamLoadUserBlocker{
+ IAMStorageAPI: original, key: cold.AccessKey,
+ waiting: make(chan struct{}), release: make(chan struct{}),
+ }
+ sys.store.IAMStorageAPI = blocker
+ unblock := sync.OnceFunc(func() { close(blocker.release) })
+ done := make(chan struct{})
+ t.Cleanup(func() {
+ unblock()
+ <-done
+ sys.store.IAMStorageAPI = original
+ })
+ var valid bool
+ var loadErr error
+ go func() {
+ defer close(done)
+ _, valid, loadErr = sys.CheckKey(ctx, cold.AccessKey)
+ }()
+ select {
+ case <-blocker.waiting:
+ case <-time.After(5 * time.Second):
+ t.Fatal("cold load did not read the identity")
+ }
+ switch change {
+ case "unrelated-edit":
+ mustIAM(t, sys.SetUserSecretKey(ctx, other.AccessKey, "unrelated-rotated-secret"))
+ case "unrelated-load":
+ mustIAM(t, sys.store.LoadUser(ctx, other.AccessKey))
+ case "delete":
+ mustIAM(t, sys.DeleteUser(ctx, cold.AccessKey, false))
+ case "mapping":
+ _, err := sys.PolicyDBSet(ctx, cold.AccessKey, "readonly", regUser, false)
+ mustIAM(t, err)
+ case "policy":
+ batchAuthzPolicy(ctx, t, sys, policyName, `{"Effect":"Deny","Action":["s3:*"],"Resource":["arn:aws:s3:::bucket/*"]}`)
+ }
+ unblock()
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("cold load did not finish")
+ }
+ if loadErr != nil || valid != (change != "delete") {
+ t.Fatalf("valid=%v err=%v after %s", valid, loadErr, change)
+ }
+ if blocker.loads.Load() != 2 {
+ t.Fatalf("conflicting load read the identity %d times, want 2", blocker.loads.Load())
+ }
+ if change == "mapping" {
+ names, err := sys.store.PolicyDBGet(cold.AccessKey)
+ mustIAM(t, err)
+ if !slices.Equal(names, []string{"readonly"}) {
+ t.Fatalf("restored stale mapping: %v", names)
+ }
+ }
+ if change == "policy" {
+ current, err := sys.store.GetPolicy(policyName)
+ mustIAM(t, err)
+ if current.IsAllowed(policy.Args{Action: policy.PutObjectAction, BucketName: "bucket", ObjectName: "object"}) {
+ t.Fatal("restored stale policy")
+ }
+ }
+ })
+ }
+}
+
+func TestIAMLoadUserRetryKeepsCachedReadsAvailable(t *testing.T) {
+ for _, unavailable := range []bool{false, true} {
+ t.Run(map[bool]string{false: "loaded", true: "unavailable"}[unavailable], func(t *testing.T) {
+ ctx, sys, _ := prepareIAMRevisionFixture(t)
+ cold := iamAdminUser(ctx, t, sys, "retry-cold", "readwrite")
+ other := iamAdminUser(ctx, t, sys, "retry-cached", "readwrite")
+ cache := sys.store.lock()
+ delete(cache.iamUsersMap, cold.AccessKey)
+ sys.store.unlock()
+ original := sys.store.IAMStorageAPI
+ blocker := &iamLoadUserBlocker{
+ IAMStorageAPI: original, key: cold.AccessKey,
+ waiting: make(chan struct{}), release: make(chan struct{}),
+ retryWaiting: make(chan struct{}), retryRelease: make(chan struct{}),
+ }
+ if unavailable {
+ blocker.retryErr = errErasureReadQuorum
+ }
+ sys.store.IAMStorageAPI = blocker
+ unblock := sync.OnceFunc(func() { close(blocker.release) })
+ unblockRetry := sync.OnceFunc(func() { close(blocker.retryRelease) })
+ done := make(chan struct{})
+ t.Cleanup(func() {
+ unblock()
+ unblockRetry()
+ <-done
+ sys.store.IAMStorageAPI = original
+ })
+ var valid bool
+ var loadErr error
+ go func() {
+ defer close(done)
+ _, valid, loadErr = sys.CheckKey(ctx, cold.AccessKey)
+ }()
+ awaitIAMPolicyReload(t, blocker.waiting)
+ mustIAM(t, sys.SetUserSecretKey(ctx, other.AccessKey, "retry-rotated-secret"))
+ unblock()
+ awaitIAMPolicyReload(t, blocker.retryWaiting)
+ readDone := make(chan struct{})
+ go func() {
+ defer close(readDone)
+ user, ok, err := sys.CheckKey(ctx, other.AccessKey)
+ if err != nil || !ok || user.Credentials.SecretKey != "retry-rotated-secret" {
+ t.Errorf("cached read failed: valid=%v err=%v", ok, err)
+ }
+ }()
+ select {
+ case <-readDone:
+ case <-time.After(time.Second):
+ unblockRetry()
+ <-readDone
+ t.Fatal("cold-load retry blocked cached authentication")
+ }
+ unblockRetry()
+ <-done
+ if unavailable {
+ if valid || !errors.Is(loadErr, errIAMNotInitialized) {
+ t.Fatalf("retry outage was not retryable: valid=%v err=%v", valid, loadErr)
+ }
+ } else if !valid || loadErr != nil {
+ t.Fatalf("retry lost the cold identity: valid=%v err=%v", valid, loadErr)
+ }
+ })
+ }
+}
+
+func TestIAMLoadUserRefreshesAssociatedRecords(t *testing.T) {
+ t.Setenv("_MINIO_IAM_GROUP_REFRESH", "on")
+ for _, change := range []string{"policy", "mapping", "group", "parent"} {
+ t.Run(change, func(t *testing.T) {
+ ctx, sys, obj := prepareIAMRevisionFixture(t)
+ const policyName = "associated-policy"
+ const groupName = "associated-group"
+ batchAuthzPolicy(ctx, t, sys, policyName, `{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::bucket/*"]}`)
+ parent := iamAdminUser(ctx, t, sys, "associated-parent", policyName)
+ _, err := sys.AddUsersToGroup(ctx, groupName, []string{parent.AccessKey})
+ mustIAM(t, err)
+ key := parent.AccessKey
+ if change == "parent" {
+ key = iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "associated-service", "").AccessKey
+ }
+ gate := &iamPolicyRefreshReadGate{
+ ObjectLayer: obj, userPath: getGroupInfoPath(groupName),
+ refreshEntered: make(chan struct{}), refreshRelease: make(chan struct{}),
+ }
+ sys.store.IAMStorageAPI.(*IAMObjectStore).objAPI = gate
+ unblock := sync.OnceFunc(func() { close(gate.refreshRelease) })
+ done := make(chan struct{})
+ var loadErr error
+ t.Cleanup(func() {
+ unblock()
+ <-done
+ })
+ go func() {
+ defer close(done)
+ loadErr = sys.store.LoadUser(context.WithValue(ctx, iamPolicyRefreshContext{}, "refresh"), key)
+ }()
+ awaitIAMPolicyReload(t, gate.refreshEntered)
+ switch change {
+ case "policy":
+ batchAuthzPolicy(ctx, t, sys, policyName, `{"Effect":"Deny","Action":["s3:*"],"Resource":["arn:aws:s3:::bucket/*"]}`)
+ case "mapping":
+ _, err = sys.PolicyDBSet(ctx, parent.AccessKey, "readonly", regUser, false)
+ mustIAM(t, err)
+ case "group":
+ _, err = sys.RemoveUsersFromGroup(ctx, groupName, []string{parent.AccessKey})
+ mustIAM(t, err)
+ case "parent":
+ mustIAM(t, sys.SetUserSecretKey(ctx, parent.AccessKey, "associated-rotated-secret"))
+ }
+ unblock()
+ select {
+ case <-done:
+ mustIAM(t, loadErr)
+ case <-time.After(5 * time.Second):
+ t.Fatal("associated-record reload did not finish")
+ }
+ switch change {
+ case "policy":
+ current, err := sys.store.GetPolicy(policyName)
+ mustIAM(t, err)
+ if current.IsAllowed(policy.Args{Action: policy.PutObjectAction, BucketName: "bucket", ObjectName: "object"}) {
+ t.Fatal("partial snapshot restored old policy")
+ }
+ case "mapping":
+ mapping, ok := sys.store.GetMappedPolicy(parent.AccessKey, false)
+ if !ok || mapping.Policies != "readonly" {
+ t.Fatalf("partial snapshot restored old mapping: %+v", mapping)
+ }
+ case "group":
+ cache := sys.store.rlock()
+ member := cache.iamUserGroupMemberships[parent.AccessKey].Contains(groupName)
+ sys.store.runlock()
+ if member {
+ t.Fatal("partial snapshot restored old group membership")
+ }
+ case "parent":
+ current, valid, err := sys.CheckKey(ctx, parent.AccessKey)
+ if err != nil || !valid || current.Credentials.SecretKey != "associated-rotated-secret" {
+ t.Fatal("child reload restored old parent credential")
+ }
+ }
+ })
+ }
+}
+
func TestIAMLoadUserDoesNotRestoreRotatedCredential(t *testing.T) {
ctx, sys, _ := prepareIAMRevisionFixture(t)
user := iamAdminUser(ctx, t, sys, "load-user-race", "readwrite")
diff --git a/cmd/iam-store.go b/cmd/iam-store.go
index 788f73a29..3707e883f 100644
--- a/cmd/iam-store.go
+++ b/cmd/iam-store.go
@@ -3225,15 +3225,17 @@ func (store *IAMStoreSys) forgetCachedIdentity(cache *iamCache, accessKey string
// LoadUser - attempts to load user info from storage and updates cache.
func (store *IAMStoreSys) LoadUser(ctx context.Context, accessKey string) error {
- loadedAt := time.Now()
groupLoad := env.Get("_MINIO_IAM_GROUP_REFRESH", config.EnableOff) == config.EnableOn
- newCachePopulate := func() (val any, err error) {
- newCache := newIamCache()
+ newCachePopulate := func(ctx context.Context) (newCache *iamCache, err error) {
+ newCache = newIamCache()
newCache.revisions = store.revisionIndex()
// Check for service account first
- store.loadUser(ctx, accessKey, svcUser, newCache.iamUsersMap)
+ err = store.loadUser(ctx, accessKey, svcUser, newCache.iamUsersMap)
+ if err != nil && !errors.Is(err, errNoSuchUser) && !errors.Is(err, errConfigNotFound) {
+ return nil, err
+ }
svc, found := newCache.iamUsersMap[accessKey]
if found {
@@ -3251,6 +3253,9 @@ func (store *IAMStoreSys) LoadUser(ctx context.Context, accessKey string) error
if !found {
err = store.loadUser(ctx, accessKey, regUser, newCache.iamUsersMap)
+ if err != nil && !errors.Is(err, errNoSuchUser) && !errors.Is(err, errConfigNotFound) {
+ return nil, err
+ }
if _, found = newCache.iamUsersMap[accessKey]; found {
// NOTE: we are not worried about loading errors from policies.
store.loadMappedPolicyWithRetry(ctx, accessKey, regUser, false, newCache.iamUserPolicyMap, 3)
@@ -3294,69 +3299,67 @@ func (store *IAMStoreSys) LoadUser(ctx context.Context, accessKey string) error
return newCache, err
}
- var (
- val any
- err error
- )
- if store.group != nil {
- val, err, _ = store.group.Do(accessKey, newCachePopulate)
- } else {
- val, err = newCachePopulate()
- }
-
- // Return error right away if any.
- if err != nil {
- if errors.Is(err, errNoSuchUser) || errors.Is(err, errConfigNotFound) {
- return nil
+ load := func() (any, error) {
+ loadedAt := time.Now()
+ newCache, err := newCachePopulate(ctx)
+ if err != nil {
+ return nil, err
}
- return err
+
+ cache := store.lock()
+ defer store.unlock()
+
+ if cache.updatedAt.After(loadedAt) {
+ err = store.withIAMStorage(ctx, func(ctx context.Context) error {
+ newCache, err = newCachePopulate(ctx)
+ return err
+ })
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ // We need to merge the new cache with the existing cache because the
+ // periodic IAM reload is partial. The periodic load here is to account.
+ newCache.iamGroupPolicyMap.Range(func(k string, v MappedPolicy) bool {
+ cache.iamGroupPolicyMap.Store(k, v)
+ return true
+ })
+
+ maps.Copy(cache.iamGroupsMap, newCache.iamGroupsMap)
+
+ maps.Copy(cache.iamPolicyDocsMap, newCache.iamPolicyDocsMap)
+
+ maps.Copy(cache.iamUserGroupMemberships, newCache.iamUserGroupMemberships)
+
+ newCache.iamUserPolicyMap.Range(func(k string, v MappedPolicy) bool {
+ cache.iamUserPolicyMap.Store(k, v)
+ return true
+ })
+
+ maps.Copy(cache.iamUsersMap, newCache.iamUsersMap)
+
+ maps.Copy(cache.iamSTSAccountsMap, newCache.iamSTSAccountsMap)
+
+ newCache.iamSTSPolicyMap.Range(func(k string, v MappedPolicy) bool {
+ cache.iamSTSPolicyMap.Store(k, v)
+ return true
+ })
+
+ cache.updatedAt = time.Now()
+ return nil, nil
}
- newCache, ok := val.(*iamCache)
- if !ok {
+ var err error
+ if store.group != nil {
+ _, err, _ = store.group.Do(accessKey, load)
+ } else {
+ _, err = load()
+ }
+ if errors.Is(err, errNoSuchUser) || errors.Is(err, errConfigNotFound) {
return nil
}
-
- cache := store.lock()
- defer store.unlock()
-
- // A partial load may have started before a credential edit or deletion and
- // finished after it. Do not merge that snapshot over the newer cache state;
- // the next notification or load will supply the missing entry.
- if cache.updatedAt.After(loadedAt) {
- return nil
- }
-
- // We need to merge the new cache with the existing cache because the
- // periodic IAM reload is partial. The periodic load here is to account.
- newCache.iamGroupPolicyMap.Range(func(k string, v MappedPolicy) bool {
- cache.iamGroupPolicyMap.Store(k, v)
- return true
- })
-
- maps.Copy(cache.iamGroupsMap, newCache.iamGroupsMap)
-
- maps.Copy(cache.iamPolicyDocsMap, newCache.iamPolicyDocsMap)
-
- maps.Copy(cache.iamUserGroupMemberships, newCache.iamUserGroupMemberships)
-
- newCache.iamUserPolicyMap.Range(func(k string, v MappedPolicy) bool {
- cache.iamUserPolicyMap.Store(k, v)
- return true
- })
-
- maps.Copy(cache.iamUsersMap, newCache.iamUsersMap)
-
- maps.Copy(cache.iamSTSAccountsMap, newCache.iamSTSAccountsMap)
-
- newCache.iamSTSPolicyMap.Range(func(k string, v MappedPolicy) bool {
- cache.iamSTSPolicyMap.Store(k, v)
- return true
- })
-
- cache.updatedAt = time.Now()
-
- return nil
+ return err
}
var errTokenSigningKeyUnavailable = errors.New("token signing key unavailable")
diff --git a/cmd/metrics-batch.go b/cmd/metrics-batch.go
new file mode 100644
index 000000000..c0ffc1177
--- /dev/null
+++ b/cmd/metrics-batch.go
@@ -0,0 +1,78 @@
+package cmd
+
+import (
+ "maps"
+ "time"
+
+ "github.com/minio/madmin-go/v3"
+)
+
+type realtimeMetrics struct {
+ madmin.RealtimeMetrics
+ Aggregated realtimeMetricGroups `json:"aggregated"`
+ ByHost map[string]realtimeMetricGroups `json:"by_host,omitempty"`
+}
+
+type realtimeMetricGroups struct {
+ madmin.Metrics
+ BatchJobs *batchJobMetricsReport `json:"batchJobs,omitempty"`
+}
+
+type batchJobMetricsReport struct {
+ CollectedAt time.Time `json:"collected"`
+ Jobs map[string]batchJobStatusMetric
+}
+
+func (metrics *realtimeMetrics) Merge(other *realtimeMetrics) {
+ if other == nil {
+ return
+ }
+ metrics.RealtimeMetrics.Merge(&other.RealtimeMetrics)
+ metrics.Aggregated.Merge(&other.Aggregated.Metrics)
+ if other.Aggregated.BatchJobs != nil {
+ if metrics.Aggregated.BatchJobs == nil {
+ metrics.Aggregated.BatchJobs = &batchJobMetricsReport{Jobs: make(map[string]batchJobStatusMetric)}
+ }
+ jobs := metrics.Aggregated.BatchJobs
+ if other.Aggregated.BatchJobs.CollectedAt.After(jobs.CollectedAt) {
+ jobs.CollectedAt = other.Aggregated.BatchJobs.CollectedAt
+ }
+ for id, job := range other.Aggregated.BatchJobs.Jobs {
+ if current, ok := jobs.Jobs[id]; !ok || !job.LastUpdate.Before(current.LastUpdate) {
+ jobs.Jobs[id] = job
+ }
+ }
+ }
+ if len(other.ByHost) != 0 {
+ if metrics.ByHost == nil {
+ metrics.ByHost = make(map[string]realtimeMetricGroups, len(other.ByHost))
+ }
+ maps.Copy(metrics.ByHost, other.ByHost)
+ }
+}
+
+func (metrics realtimeMetricGroups) legacy() madmin.Metrics {
+ out := metrics.Metrics
+ if metrics.BatchJobs != nil {
+ out.BatchJobs = &madmin.BatchJobMetrics{
+ CollectedAt: metrics.BatchJobs.CollectedAt,
+ Jobs: make(map[string]madmin.JobMetric, len(metrics.BatchJobs.Jobs)),
+ }
+ for id, job := range metrics.BatchJobs.Jobs {
+ out.BatchJobs.Jobs[id] = job.JobMetric
+ }
+ }
+ return out
+}
+
+func (metrics realtimeMetrics) legacy() madmin.RealtimeMetrics {
+ out := metrics.RealtimeMetrics
+ out.Aggregated = metrics.Aggregated.legacy()
+ if metrics.ByHost != nil {
+ out.ByHost = make(map[string]madmin.Metrics, len(metrics.ByHost))
+ for host, groups := range metrics.ByHost {
+ out.ByHost[host] = groups.legacy()
+ }
+ }
+ return out
+}
diff --git a/cmd/metrics-batch_test.go b/cmd/metrics-batch_test.go
new file mode 100644
index 000000000..fd8158181
--- /dev/null
+++ b/cmd/metrics-batch_test.go
@@ -0,0 +1,153 @@
+package cmd
+
+import (
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "reflect"
+ "testing"
+ "time"
+
+ "github.com/minio/madmin-go/v3"
+ "github.com/minio/minio/internal/grid"
+)
+
+func TestBatchMetricsStatusSnapshots(t *testing.T) {
+ fixture := newBatchAuthzFixture(t)
+ job := BatchJobRequest{
+ ID: "expire-metrics-snapshot" + getKeySeparator() + "0", Started: UTCNow(),
+ Submitter: batchRootSubmitter(t), Expire: &BatchJobExpire{},
+ }
+ for _, state := range []string{"skipped", "paused", "failed", "running"} {
+ t.Run(state, func(t *testing.T) {
+ report := &batchJobInfo{JobID: job.ID, JobType: string(job.Type()), StartTime: job.Started}
+ switch state {
+ case "skipped":
+ report.Complete = true
+ report.VersionsSkippedLocked, report.KeysRefused = 2, 1
+ case "paused":
+ report.PausedReason = "authorization plugin unavailable"
+ case "failed":
+ report.Failed = true
+ report.FailureReason = "authorization denied"
+ }
+ mustIAM(t, report.updateAfter(fixture.ctx, fixture.obj, 0, job))
+ globalBatchJobsMetrics.save(job.ID, report)
+ saved := batchJobStatusJSON(t, job.ID)
+ for _, field := range []string{"startTime", "lastUpdate"} {
+ stamp, err := time.Parse(time.RFC3339Nano, saved[field].(string))
+ mustIAM(t, err)
+ saved[field] = stamp.UTC().Format(time.RFC3339Nano)
+ }
+ target := fmt.Sprintf("http://localhost/minio/admin/v3/metrics?types=%d&by-jobID=%s&n=1&by-host=true", madmin.MetricsBatchJobs, job.ID)
+ request, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+ mustIAM(t, err)
+ mustIAM(t, request.ParseForm())
+ response := httptest.NewRecorder()
+ request = request.WithContext(newContext(request, response, "Metrics"))
+ adminAPIHandlers{}.MetricsHandler(response, request)
+ if response.Code != http.StatusOK {
+ t.Fatalf("metrics returned %d: %s", response.Code, response.Body.String())
+ }
+ var wire struct {
+ Aggregated struct {
+ BatchJobs struct{ Jobs map[string]map[string]any }
+ }
+ ByHost map[string]struct {
+ BatchJobs struct{ Jobs map[string]map[string]any }
+ } `json:"by_host"`
+ }
+ mustIAM(t, json.Unmarshal(response.Body.Bytes(), &wire))
+ if !reflect.DeepEqual(wire.Aggregated.BatchJobs.Jobs[job.ID], saved) {
+ t.Fatalf("live snapshot differs from saved status: live=%v saved=%v", wire.Aggregated.BatchJobs.Jobs[job.ID], saved)
+ }
+ if len(wire.ByHost) != 1 {
+ t.Fatalf("missing per-host snapshot: %s", response.Body.String())
+ }
+ for _, host := range wire.ByHost {
+ if !reflect.DeepEqual(host.BatchJobs.Jobs[job.ID], saved) {
+ t.Fatal("per-host metrics lost details")
+ }
+ }
+ var legacy madmin.RealtimeMetrics
+ mustIAM(t, json.Unmarshal(response.Body.Bytes(), &legacy))
+ if legacy.Aggregated.BatchJobs.Jobs[job.ID].JobID != job.ID || legacy.Aggregated.BatchJobs.Jobs[job.ID].Complete != report.Complete {
+ t.Fatal("additive details changed legacy metrics")
+ }
+ })
+ }
+}
+
+func TestBatchMetricsPeerRoundTrip(t *testing.T) {
+ report := &batchJobInfo{JobID: "remote-job", LastUpdate: UTCNow(), PausedReason: "IAM unavailable", VersionsSkippedLocked: 3, KeysRefused: 2}
+ remote := realtimeMetrics{}
+ remote.Hosts = []string{"peer"}
+ remote.Aggregated.BatchJobs = &batchJobMetricsReport{
+ CollectedAt: UTCNow(), Jobs: map[string]batchJobStatusMetric{report.JobID: report.statusMetric()},
+ }
+ remote.ByHost = map[string]realtimeMetricGroups{"peer": remote.Aggregated}
+ encoded := madminRealtimeMetrics.NewJSONWith(&remote)
+ data, err := encoded.MarshalMsg(nil)
+ mustIAM(t, err)
+ decoded := madminRealtimeMetrics.NewJSON()
+ defer decoded.Recycle()
+ _, err = decoded.UnmarshalMsg(data)
+ mustIAM(t, err)
+ var merged realtimeMetrics
+ received := decoded.ValueOrZero()
+ merged.Merge(&received)
+ got := merged.Aggregated.BatchJobs.Jobs[report.JobID]
+ if got.PausedReason != report.PausedReason || got.KeysRefused != 2 || got.VersionsSkippedLocked == nil || *got.VersionsSkippedLocked != 3 {
+ t.Fatalf("peer aggregation lost details: %+v", got)
+ }
+ older := received
+ older.Aggregated.BatchJobs = &batchJobMetricsReport{Jobs: map[string]batchJobStatusMetric{
+ report.JobID: {JobMetric: madmin.JobMetric{JobID: report.JobID, LastUpdate: report.LastUpdate.Add(-time.Second), Complete: true}},
+ }}
+ merged.Merge(&older)
+ if got := merged.Aggregated.BatchJobs.Jobs[report.JobID]; got.Complete || got.PausedReason != report.PausedReason {
+ t.Fatal("older peer snapshot replaced current progress")
+ }
+ legacyPool := grid.NewJSONPool[madmin.RealtimeMetrics]()
+ legacy := legacyPool.NewJSON()
+ defer legacy.Recycle()
+ _, err = legacy.UnmarshalMsg(data)
+ mustIAM(t, err)
+ if got := legacy.ValueOrZero().Aggregated.BatchJobs.Jobs[report.JobID]; got.JobID != report.JobID || !got.LastUpdate.Equal(report.LastUpdate) {
+ t.Fatal("legacy peer could not read base metrics")
+ }
+ legacyData, err := legacy.MarshalMsg(nil)
+ mustIAM(t, err)
+ _, err = decoded.UnmarshalMsg(legacyData)
+ mustIAM(t, err)
+ received = decoded.ValueOrZero()
+ if received.Aggregated.BatchJobs.Jobs[report.JobID].VersionsSkippedLocked != nil {
+ t.Fatal("legacy peer falsely advertised complete status details")
+ }
+}
+
+func TestBatchMetricsPreservesOtherMetrics(t *testing.T) {
+ var legacy madmin.RealtimeMetrics
+ var extended realtimeMetrics
+ for _, host := range []string{"first", "second"} {
+ groups := madmin.Metrics{
+ Disk: &madmin.DiskMetric{CollectedAt: time.Unix(1, 0).UTC(), NDisks: 2, Offline: 1},
+ CPU: &madmin.CPUMetrics{CPUCount: 4},
+ }
+ sample := madmin.RealtimeMetrics{
+ Hosts: []string{host}, Errors: []string{host + ": test"},
+ Aggregated: groups, ByHost: map[string]madmin.Metrics{host: groups},
+ ByDisk: map[string]madmin.DiskMetric{host: *groups.Disk},
+ }
+ data, err := json.Marshal(sample)
+ mustIAM(t, err)
+ var decoded realtimeMetrics
+ mustIAM(t, json.Unmarshal(data, &decoded))
+ extended.Merge(&decoded)
+ legacy.Merge(&sample)
+ }
+ if got := extended.legacy(); !reflect.DeepEqual(got, legacy) {
+ t.Fatalf("batch extension changed other metrics: got=%+v want=%+v", got, legacy)
+ }
+}
diff --git a/cmd/metrics-realtime.go b/cmd/metrics-realtime.go
index 9eb10990f..f7b73ddff 100644
--- a/cmd/metrics-realtime.go
+++ b/cmd/metrics-realtime.go
@@ -38,7 +38,7 @@ type collectMetricsOpts struct {
depID string
}
-func collectLocalMetrics(types madmin.MetricType, opts collectMetricsOpts) (m madmin.RealtimeMetrics) {
+func collectLocalMetrics(types madmin.MetricType, opts collectMetricsOpts) (m realtimeMetrics) {
if types == madmin.MetricsNone {
return m
}
@@ -144,7 +144,7 @@ func collectLocalMetrics(types madmin.MetricType, opts collectMetricsOpts) (m ma
// Add types...
// ByHost is a shallow reference, so careful about sharing.
- m.ByHost = map[string]madmin.Metrics{byHostName: m.Aggregated}
+ m.ByHost = map[string]realtimeMetricGroups{byHostName: m.Aggregated}
m.Hosts = append(m.Hosts, byHostName)
return m
@@ -221,11 +221,11 @@ func collectLocalDisksMetrics(disks map[string]struct{}) map[string]madmin.DiskM
return metrics
}
-func collectRemoteMetrics(ctx context.Context, types madmin.MetricType, opts collectMetricsOpts) (m madmin.RealtimeMetrics) {
+func collectRemoteMetrics(ctx context.Context, types madmin.MetricType, opts collectMetricsOpts) (m realtimeMetrics) {
if !globalIsDistErasure {
return m
}
- all := globalNotificationSys.GetMetrics(ctx, types, opts)
+ all := globalNotificationSys.getMetrics(ctx, types, opts)
for _, remote := range all {
m.Merge(&remote)
}
diff --git a/cmd/metrics-resource.go b/cmd/metrics-resource.go
index 94c918733..3ce465edd 100644
--- a/cmd/metrics-resource.go
+++ b/cmd/metrics-resource.go
@@ -246,7 +246,7 @@ func updateDriveIOStats(currentStats madmin.DiskIOStats, latestStats madmin.Disk
updateResourceMetrics(driveSubsystem, percUtil, float64(diffStats.TotalTicks)/(diffInSeconds*10), labels, false)
}
-func collectDriveMetrics(m madmin.RealtimeMetrics) {
+func collectDriveMetrics(m realtimeMetrics) {
latestDriveStatsMu.Lock()
for d, dm := range m.ByDisk {
labels := map[string]string{"drive": d}
diff --git a/cmd/metrics-v2.go b/cmd/metrics-v2.go
index 186047e76..70624d040 100644
--- a/cmd/metrics-v2.go
+++ b/cmd/metrics-v2.go
@@ -3804,7 +3804,7 @@ func getBatchJobsMetrics(opts MetricsGroupOpts) *MetricsGroupV2 {
}
mg.RegisterRead(func(ctx context.Context) (metrics []MetricV2) {
- var m madmin.RealtimeMetrics
+ var m realtimeMetrics
mLocal := collectLocalMetrics(madmin.MetricsBatchJobs, collectMetricsOpts{})
m.Merge(&mLocal)
diff --git a/cmd/notification.go b/cmd/notification.go
index 4b9ad3ffb..bc48d2da7 100644
--- a/cmd/notification.go
+++ b/cmd/notification.go
@@ -873,9 +873,9 @@ func (sys *NotificationSys) GetOSInfo(ctx context.Context) []madmin.OSInfo {
return reply
}
-// GetMetrics - Get metrics from all peers.
-func (sys *NotificationSys) GetMetrics(ctx context.Context, t madmin.MetricType, opts collectMetricsOpts) []madmin.RealtimeMetrics {
- reply := make([]madmin.RealtimeMetrics, len(sys.peerClients))
+// getMetrics - Get metrics from all peers.
+func (sys *NotificationSys) getMetrics(ctx context.Context, t madmin.MetricType, opts collectMetricsOpts) []realtimeMetrics {
+ reply := make([]realtimeMetrics, len(sys.peerClients))
g := errgroup.WithNErrs(len(sys.peerClients))
for index, client := range sys.peerClients {
diff --git a/cmd/peer-rest-client.go b/cmd/peer-rest-client.go
index 082b0a36c..760525d6c 100644
--- a/cmd/peer-rest-client.go
+++ b/cmd/peer-rest-client.go
@@ -228,7 +228,7 @@ func (client *peerRESTClient) GetMemInfo(ctx context.Context) (info madmin.MemIn
}
// GetMetrics - fetch metrics from a remote node.
-func (client *peerRESTClient) GetMetrics(ctx context.Context, t madmin.MetricType, opts collectMetricsOpts) (info madmin.RealtimeMetrics, err error) {
+func (client *peerRESTClient) GetMetrics(ctx context.Context, t madmin.MetricType, opts collectMetricsOpts) (info realtimeMetrics, err error) {
values := make(url.Values)
values.Set(peerRESTMetricsTypes, strconv.FormatUint(uint64(t), 10))
for disk := range opts.disks {
diff --git a/cmd/peer-rest-server.go b/cmd/peer-rest-server.go
index ed13661e3..80107150b 100644
--- a/cmd/peer-rest-server.go
+++ b/cmd/peer-rest-server.go
@@ -62,7 +62,7 @@ var (
madminOSInfo = grid.NewJSONPool[madmin.OSInfo]()
madminPartitions = grid.NewJSONPool[madmin.Partitions]()
madminProcInfo = grid.NewJSONPool[madmin.ProcInfo]()
- madminRealtimeMetrics = grid.NewJSONPool[madmin.RealtimeMetrics]()
+ madminRealtimeMetrics = grid.NewJSONPool[realtimeMetrics]()
madminServerProperties = grid.NewJSONPool[madmin.ServerProperties]()
madminStorageInfo = grid.NewJSONPool[madmin.StorageInfo]()
madminSysConfig = grid.NewJSONPool[madmin.SysConfig]()
@@ -84,7 +84,7 @@ var (
getLocksRPC = grid.NewSingleHandler[*grid.MSS, *localLockMap](grid.HandlerGetLocks, grid.NewMSS, func() *localLockMap { return &localLockMap{} })
getMemInfoRPC = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.MemInfo]](grid.HandlerGetMemInfo, grid.NewMSS, madminMemInfo.NewJSON)
getMetacacheListingRPC = grid.NewSingleHandler[*listPathOptions, *metacache](grid.HandlerGetMetacacheListing, func() *listPathOptions { return &listPathOptions{} }, func() *metacache { return &metacache{} })
- getMetricsRPC = grid.NewSingleHandler[*grid.URLValues, *grid.JSON[madmin.RealtimeMetrics]](grid.HandlerGetMetrics, grid.NewURLValues, madminRealtimeMetrics.NewJSON)
+ getMetricsRPC = grid.NewSingleHandler[*grid.URLValues, *grid.JSON[realtimeMetrics]](grid.HandlerGetMetrics, grid.NewURLValues, madminRealtimeMetrics.NewJSON)
getNetInfoRPC = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.NetInfo]](grid.HandlerGetNetInfo, grid.NewMSS, madminNetInfo.NewJSON)
getOSInfoRPC = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.OSInfo]](grid.HandlerGetOSInfo, grid.NewMSS, madminOSInfo.NewJSON)
getPartitionsRPC = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.Partitions]](grid.HandlerGetPartitions, grid.NewMSS, madminPartitions.NewJSON)
@@ -423,7 +423,7 @@ func (s *peerRESTServer) GetMemInfoHandler(_ *grid.MSS) (*grid.JSON[madmin.MemIn
}
// GetMetricsHandler - returns server metrics.
-func (s *peerRESTServer) GetMetricsHandler(v *grid.URLValues) (*grid.JSON[madmin.RealtimeMetrics], *grid.RemoteErr) {
+func (s *peerRESTServer) GetMetricsHandler(v *grid.URLValues) (*grid.JSON[realtimeMetrics], *grid.RemoteErr) {
values := v.Values()
var types madmin.MetricType
if t, _ := strconv.ParseUint(values.Get(peerRESTMetricsTypes), 10, 64); t != 0 {