diff --git a/docs/slurm-exporter.md b/docs/slurm-exporter.md
index 47e868df4..16a504148 100644
--- a/docs/slurm-exporter.md
+++ b/docs/slurm-exporter.md
@@ -134,7 +134,10 @@ The exporter provides self-monitoring metrics to track its own health and perfor
 | **slurm_exporter_collection_duration_seconds**<br>*Gauge* | Duration of the most recent metrics collection from SLURM APIs<br><br>**Labels:** None |
 | **slurm_exporter_collection_attempts_total**<br>*Counter* | Total number of metrics collection attempts<br><br>**Labels:** None |
 | **slurm_exporter_collection_failures_total**<br>*Counter* | Total number of failed metrics collection attempts<br><br>**Labels:** None |
+| **slurm_exporter_collector_duration_seconds**<br>*Gauge* | Duration of the most recent sub-collector run during metrics collection<br><br>**Labels:** `collector` (one of `nodes`, `jobs`, `diag`) |
 | **slurm_exporter_collector_errors_total**<br>*Counter* | Total number of errors per sub-collector during metrics collection. Sub-collectors are isolated, so a failure in one does not drop the others' metrics for that cycle; the failed collector's metrics simply gap until it recovers.<br><br>**Labels:** `collector` (one of `nodes`, `jobs`, `diag`) |
+| **slurm_exporter_collector_inflight**<br>*Gauge* | Whether a sub-collector run is currently in progress<br><br>**Labels:** `collector` (one of `nodes`, `jobs`, `diag`) |
+| **slurm_exporter_collector_skipped_total**<br>*Counter* | Total number of skipped sub-collector runs because the previous run for the same collector was still in progress<br><br>**Labels:** `collector` (one of `nodes`, `jobs`, `diag`) |
 | **slurm_exporter_metrics_requests_total**<br>*Counter* | Total number of requests to the `/metrics` endpoint<br><br>**Labels:** None |
 | **slurm_exporter_metrics_exported**<br>*Gauge* | Number of metrics exported in the last scrape<br><br>**Labels:** None |
 
diff --git a/internal/exporter/collector.go b/internal/exporter/collector.go
index 90bb28379..4fd280086 100644
--- a/internal/exporter/collector.go
+++ b/internal/exporter/collector.go
@@ -7,6 +7,7 @@ import (
 	"iter"
 	"maps"
 	"strconv"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -51,7 +52,8 @@ type MetricsCollector struct {
 	controllerServerThreadCount *prometheus.Desc
 
 	// Atomic pointer to the current state for lock-free reads
-	state atomic.Pointer[metricsCollectorState]
+	stateMu sync.Mutex
+	state   atomic.Pointer[metricsCollectorState]
 
 	// Monitoring contains self-monitoring metrics
 	Monitoring *MonitoringMetrics
@@ -333,6 +335,8 @@ func (c *MetricsCollector) Describe(ch chan<- *prometheus.Desc) {
 func (c *MetricsCollector) updateState(ctx context.Context) (err error) {
 	logger := log.FromContext(ctx).WithName(ControllerName)
 	startTime := time.Now()
+	c.stateMu.Lock()
+	defer c.stateMu.Unlock()
 
 	defer func() {
 		duration := time.Since(startTime).Seconds()
@@ -369,11 +373,13 @@ func (c *MetricsCollector) updateState(ctx context.Context) (err error) {
 
 	var errs []error
 	for _, col := range collectors {
+		colStart := time.Now()
 		if cerr := col.run(); cerr != nil {
 			logger.Error(cerr, "Sub-collector failed", "collector", col.name)
 			c.Monitoring.RecordCollectorError(col.name)
 			errs = append(errs, fmt.Errorf("%s: %w", col.name, cerr))
 		}
+		c.Monitoring.RecordCollectorDuration(col.name, time.Since(colStart).Seconds())
 	}
 
 	logger.Info("Collected metrics", "elapsed_seconds", time.Since(startTime).Seconds())
@@ -383,54 +389,144 @@ func (c *MetricsCollector) updateState(ctx context.Context) (err error) {
 
 // collectNodes fetches nodes and updates node-derived metrics on the new state.
 func (c *MetricsCollector) collectNodes(ctx context.Context, previousState, newState *metricsCollectorState) error {
+	nodes, err := c.listNodes(ctx)
+	if err != nil {
+		return err
+	}
+	c.applyNodes(ctx, nodes, previousState, newState)
+	return nil
+}
+
+func (c *MetricsCollector) listNodes(ctx context.Context) ([]slurmapi.Node, error) {
 	logger := log.FromContext(ctx).WithName(ControllerName)
 
 	nodesStart := time.Now()
 	nodes, err := c.slurmAPIClient.ListNodes(ctx)
 	if err != nil {
-		return fmt.Errorf("get nodes from SLURM API: %w", err)
+		return nil, fmt.Errorf("get nodes from SLURM API: %w", err)
 	}
 	logger.Info("Fetched nodes", "count", len(nodes), "elapsed_seconds", time.Since(nodesStart).Seconds())
-	newState.nodes = nodes
 
+	return nodes, nil
+}
+
+func (c *MetricsCollector) applyNodes(ctx context.Context, nodes []slurmapi.Node, previousState, newState *metricsCollectorState) {
+	newState.nodes = nodes
 	c.updateNodeFailureMetrics(nodes, previousState.nodes)
 	c.updateNodeStateMetrics(nodes, previousState, newState, time.Now())
 	newState.lastGPUSecondsUpdate = c.updateGPUSecondsMetrics(ctx, nodes, previousState.lastGPUSecondsUpdate, time.Now())
+}
+
+func cloneMetricsCollectorState(previousState *metricsCollectorState) *metricsCollectorState {
+	if previousState == nil {
+		previousState = newMetricsCollectorState()
+	}
+	newState := newMetricsCollectorState()
+	newState.lastGPUSecondsUpdate = previousState.lastGPUSecondsUpdate
+	newState.nodes = previousState.nodes
+	newState.jobs = previousState.jobs
+	newState.diag = previousState.diag
+	maps.Copy(newState.nodeUnavailabilityStartTimes, previousState.nodeUnavailabilityStartTimes)
+	maps.Copy(newState.nodeDrainingStartTimes, previousState.nodeDrainingStartTimes)
+	return newState
+}
+
+func (c *MetricsCollector) refreshNodes(ctx context.Context) error {
+	nodes, err := c.listNodes(ctx)
+	if err != nil {
+		return err
+	}
+
+	c.stateMu.Lock()
+	defer c.stateMu.Unlock()
 
+	previousState := c.state.Load()
+	if previousState == nil {
+		previousState = newMetricsCollectorState()
+	}
+	newState := cloneMetricsCollectorState(previousState)
+	c.applyNodes(ctx, nodes, previousState, newState)
+	c.state.Store(newState)
 	return nil
 }
 
 // collectJobs fetches jobs and stores them on the new state.
 func (c *MetricsCollector) collectJobs(ctx context.Context, newState *metricsCollectorState) error {
+	jobs, err := c.listJobs(ctx)
+	if err != nil {
+		return err
+	}
+	newState.jobs = jobs
+	return nil
+}
+
+func (c *MetricsCollector) listJobs(ctx context.Context) ([]slurmapi.Job, error) {
 	logger := log.FromContext(ctx).WithName(ControllerName)
 
 	jobsStart := time.Now()
 	jobs, err := c.slurmAPIClient.ListJobsWithParams(ctx, c.jobListParams)
 	if err != nil {
-		return fmt.Errorf("get jobs from SLURM API: %w", err)
+		return nil, fmt.Errorf("get jobs from SLURM API: %w", err)
 	}
 	logger.Info("Fetched jobs",
 		"source", string(c.jobListParams.Source),
 		"count", len(jobs),
 		"elapsed_seconds", time.Since(jobsStart).Seconds(),
 	)
-	newState.jobs = jobs
 
+	return jobs, nil
+}
+
+func (c *MetricsCollector) refreshJobs(ctx context.Context) error {
+	jobs, err := c.listJobs(ctx)
+	if err != nil {
+		return err
+	}
+
+	c.stateMu.Lock()
+	defer c.stateMu.Unlock()
+
+	newState := cloneMetricsCollectorState(c.state.Load())
+	newState.jobs = jobs
+	c.state.Store(newState)
 	return nil
 }
 
 // collectDiag fetches controller diagnostics and stores them on the new state.
 func (c *MetricsCollector) collectDiag(ctx context.Context, newState *metricsCollectorState) error {
+	diag, err := c.getDiag(ctx)
+	if err != nil {
+		return err
+	}
+	newState.diag = diag
+	return nil
+}
+
+func (c *MetricsCollector) getDiag(ctx context.Context) (*api.V0041OpenapiDiagResp, error) {
 	logger := log.FromContext(ctx).WithName(ControllerName)
 
 	diagStart := time.Now()
 	diag, err := c.slurmAPIClient.GetDiag(ctx)
 	if err != nil {
-		return fmt.Errorf("get diag from SLURM API: %w", err)
+		return nil, fmt.Errorf("get diag from SLURM API: %w", err)
 	}
 	logger.Info("Fetched controller diag", "elapsed_seconds", time.Since(diagStart).Seconds())
-	newState.diag = diag
 
+	return diag, nil
+}
+
+func (c *MetricsCollector) refreshDiag(ctx context.Context) error {
+	diag, err := c.getDiag(ctx)
+	if err != nil {
+		return err
+	}
+
+	c.stateMu.Lock()
+	defer c.stateMu.Unlock()
+
+	newState := cloneMetricsCollectorState(c.state.Load())
+	newState.diag = diag
+	c.state.Store(newState)
 	return nil
 }
 
diff --git a/internal/exporter/exporter.go b/internal/exporter/exporter.go
index fca0a0248..5ced0aec6 100644
--- a/internal/exporter/exporter.go
+++ b/internal/exporter/exporter.go
@@ -5,6 +5,8 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -51,6 +53,12 @@ type Exporter struct {
 	monitoringServer *http.Server
 }
 
+type asyncSubCollector struct {
+	name    string
+	running atomic.Bool
+	run     func(context.Context) error
+}
+
 // NewClusterExporter creates a new SLURM cluster exporter
 func NewClusterExporter(slurmAPIClient slurmapi.Client, params Params) *Exporter {
 	registry := prometheus.NewRegistry()
@@ -102,13 +110,67 @@ func (e *Exporter) collectionLoop(ctx context.Context) {
 	logger := log.FromContext(ctx).WithName(ControllerName)
 	logger.Info("Starting metrics collection loop", "interval", e.params.CollectionInterval)
 
+	loopCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
 	ticker := time.NewTicker(e.params.CollectionInterval)
 	defer ticker.Stop()
 
-	if err := e.collector.updateState(ctx); err != nil {
-		logger.Error(err, "Initial metrics collection failed")
+	collectors := []*asyncSubCollector{
+		{name: "nodes", run: e.collector.refreshNodes},
+		{name: "jobs", run: e.collector.refreshJobs},
+		{name: "diag", run: e.collector.refreshDiag},
 	}
 
+	startCollectors := func() {
+		start := time.Now()
+		var wg sync.WaitGroup
+		var failed atomic.Bool
+		started := false
+
+		for _, col := range collectors {
+			if !col.running.CompareAndSwap(false, true) {
+				e.monitoringMetrics.RecordCollectorSkipped(col.name)
+				logger.Info("Skipping sub-collector; previous run still in progress", "collector", col.name)
+				continue
+			}
+
+			started = true
+			wg.Add(1)
+			e.monitoringMetrics.SetCollectorInflight(col.name, true)
+			go func(col *asyncSubCollector) {
+				defer wg.Done()
+				defer col.running.Store(false)
+				defer e.monitoringMetrics.SetCollectorInflight(col.name, false)
+
+				start := time.Now()
+				if err := col.run(loopCtx); err != nil {
+					if loopCtx.Err() == nil {
+						logger.Error(err, "Sub-collector failed", "collector", col.name)
+						e.monitoringMetrics.RecordCollectorError(col.name)
+						failed.Store(true)
+					}
+				}
+				e.monitoringMetrics.RecordCollectorDuration(col.name, time.Since(start).Seconds())
+			}(col)
+		}
+
+		if !started {
+			e.monitoringMetrics.RecordCollection(0, nil)
+			return
+		}
+		go func() {
+			wg.Wait()
+			var err error
+			if failed.Load() {
+				err = errors.New("sub-collector failed")
+			}
+			e.monitoringMetrics.RecordCollection(time.Since(start).Seconds(), err)
+		}()
+	}
+
+	startCollectors()
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -118,9 +180,7 @@ func (e *Exporter) collectionLoop(ctx context.Context) {
 			logger.Info("Stopping metrics collection loop via stop channel")
 			return
 		case <-ticker.C:
-			if err := e.collector.updateState(ctx); err != nil {
-				logger.Error(err, "Metrics collection failed")
-			}
+			startCollectors()
 		}
 	}
 }
diff --git a/internal/exporter/exporter_test.go b/internal/exporter/exporter_test.go
new file mode 100644
index 000000000..a3832da8a
--- /dev/null
+++ b/internal/exporter/exporter_test.go
@@ -0,0 +1,86 @@
+package exporter
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	api "github.com/SlinkyProject/slurm-client/api/v0041"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+
+	"nebius.ai/slurm-operator/internal/slurmapi"
+	"nebius.ai/slurm-operator/internal/slurmapi/fake"
+)
+
+func TestExporterCollectionLoopSkipsRunningSubCollector(t *testing.T) {
+	log.SetLogger(zap.New(zap.UseDevMode(true)))
+
+	mockClient := &fake.MockClient{}
+	var jobsCalls atomic.Int32
+	jobsStarted := make(chan struct{})
+	jobsDone := make(chan struct{})
+	releaseJobs := make(chan struct{})
+	var closeStarted, closeDone sync.Once
+
+	mockClient.EXPECT().ListNodes(mock.Anything).Return([]slurmapi.Node{}, nil).Maybe()
+	mockClient.EXPECT().GetDiag(mock.Anything).Return(&api.V0041OpenapiDiagResp{}, nil).Maybe()
+	mockClient.EXPECT().ListJobsWithParams(mock.Anything, mock.Anything).RunAndReturn(
+		func(ctx context.Context, _ slurmapi.ListJobsParams) ([]slurmapi.Job, error) {
+			jobsCalls.Add(1)
+			closeStarted.Do(func() { close(jobsStarted) })
+			defer closeDone.Do(func() { close(jobsDone) })
+
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			case <-releaseJobs:
+				return nil, nil
+			}
+		},
+	)
+
+	exporter := NewClusterExporter(mockClient, Params{CollectionInterval: 5 * time.Millisecond})
+	registry := prometheus.NewRegistry()
+	require.NoError(t, exporter.monitoringMetrics.Register(registry))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	loopDone := make(chan struct{})
+	go func() {
+		exporter.collectionLoop(ctx)
+		close(loopDone)
+	}()
+
+	select {
+	case <-jobsStarted:
+	case <-time.After(time.Second):
+		t.Fatal("jobs collector did not start")
+	}
+
+	require.Eventually(t, func() bool {
+		families, err := registry.Gather()
+		require.NoError(t, err)
+		return collectorCounterValue(families, "slurm_exporter_collector_skipped_total", "jobs") > 0
+	}, time.Second, 10*time.Millisecond)
+	assert.Equal(t, int32(1), jobsCalls.Load())
+
+	cancel()
+	close(releaseJobs)
+
+	select {
+	case <-jobsDone:
+	case <-time.After(time.Second):
+		t.Fatal("jobs collector did not stop")
+	}
+	select {
+	case <-loopDone:
+	case <-time.After(time.Second):
+		t.Fatal("collection loop did not stop")
+	}
+}
diff --git a/internal/exporter/monitoring.go b/internal/exporter/monitoring.go
index 3a8432e66..13a0e2c4c 100644
--- a/internal/exporter/monitoring.go
+++ b/internal/exporter/monitoring.go
@@ -9,7 +9,10 @@ type MonitoringMetrics struct {
 	collectionDuration prometheus.Gauge
 	collectionAttempts prometheus.Counter
 	collectionFailures prometheus.Counter
+	collectorDuration  *prometheus.GaugeVec
 	collectorErrors    *prometheus.CounterVec
+	collectorInflight  *prometheus.GaugeVec
+	collectorSkipped   *prometheus.CounterVec
 	metricsRequests    prometheus.Counter
 	metricsExported    prometheus.Gauge
 }
@@ -29,10 +32,22 @@ func NewMonitoringMetrics() *MonitoringMetrics {
 			Name: "slurm_exporter_collection_failures_total",
 			Help: "Total number of failed metrics collection attempts",
 		}),
+		collectorDuration: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "slurm_exporter_collector_duration_seconds",
+			Help: "Duration of the most recent sub-collector run, labeled by collector",
+		}, []string{"collector"}),
 		collectorErrors: prometheus.NewCounterVec(prometheus.CounterOpts{
 			Name: "slurm_exporter_collector_errors_total",
 			Help: "Total number of errors per sub-collector during metrics collection, labeled by collector",
 		}, []string{"collector"}),
+		collectorInflight: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "slurm_exporter_collector_inflight",
+			Help: "Whether a sub-collector run is currently in progress, labeled by collector",
+		}, []string{"collector"}),
+		collectorSkipped: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "slurm_exporter_collector_skipped_total",
+			Help: "Total number of skipped sub-collector runs because a previous run was still in progress, labeled by collector",
+		}, []string{"collector"}),
 		metricsRequests: prometheus.NewCounter(prometheus.CounterOpts{
 			Name: "slurm_exporter_metrics_requests_total",
 			Help: "Total number of requests to /metrics endpoint",
@@ -50,7 +65,10 @@ func (m *MonitoringMetrics) Register(registry *prometheus.Registry) error {
 		m.collectionDuration,
 		m.collectionAttempts,
 		m.collectionFailures,
+		m.collectorDuration,
 		m.collectorErrors,
+		m.collectorInflight,
+		m.collectorSkipped,
 		m.metricsRequests,
 		m.metricsExported,
 	}
@@ -73,11 +91,29 @@ func (m *MonitoringMetrics) RecordCollection(duration float64, err error) {
 	}
 }
 
+// RecordCollectorDuration records the latest duration for the named sub-collector.
+func (m *MonitoringMetrics) RecordCollectorDuration(collector string, duration float64) {
+	m.collectorDuration.WithLabelValues(collector).Set(duration)
+}
+
+func (m *MonitoringMetrics) SetCollectorInflight(collector string, inflight bool) {
+	value := 0.0
+	if inflight {
+		value = 1.0
+	}
+	m.collectorInflight.WithLabelValues(collector).Set(value)
+}
+
 // RecordCollectorError increments the error counter for the named sub-collector
 func (m *MonitoringMetrics) RecordCollectorError(collector string) {
 	m.collectorErrors.WithLabelValues(collector).Inc()
 }
 
+// RecordCollectorSkipped increments the skip counter for the named sub-collector.
+func (m *MonitoringMetrics) RecordCollectorSkipped(collector string) {
+	m.collectorSkipped.WithLabelValues(collector).Inc()
+}
+
 // RecordMetricsRequest increments the metrics request counter
 func (m *MonitoringMetrics) RecordMetricsRequest() {
 	m.metricsRequests.Inc()
diff --git a/internal/exporter/monitoring_test.go b/internal/exporter/monitoring_test.go
index 221e87147..5e02cfd26 100644
--- a/internal/exporter/monitoring_test.go
+++ b/internal/exporter/monitoring_test.go
@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -89,6 +90,72 @@ func TestMonitoringMetrics_RecordCollectorError(t *testing.T) {
 	assert.Equal(t, float64(0), collectorErrorValue(metricFamilies, "jobs"), "Expected no jobs collector errors")
 }
 
+func TestMonitoringMetrics_RecordCollectorDuration(t *testing.T) {
+	metrics := NewMonitoringMetrics()
+	registry := prometheus.NewRegistry()
+	require.NoError(t, metrics.Register(registry))
+
+	metrics.RecordCollectorDuration("nodes", 1.5)
+	metrics.RecordCollectorDuration("jobs", 2.5)
+	metrics.RecordCollectorDuration("nodes", 0.5)
+
+	metricFamilies, err := registry.Gather()
+	require.NoError(t, err)
+
+	assert.Equal(t, float64(0.5), collectorGaugeValue(metricFamilies, "slurm_exporter_collector_duration_seconds", "nodes"))
+	assert.Equal(t, float64(2.5), collectorGaugeValue(metricFamilies, "slurm_exporter_collector_duration_seconds", "jobs"))
+	assert.Equal(t, float64(0), collectorGaugeValue(metricFamilies, "slurm_exporter_collector_duration_seconds", "diag"))
+}
+
+func TestMonitoringMetrics_RecordCollectorRuntimeState(t *testing.T) {
+	metrics := NewMonitoringMetrics()
+	registry := prometheus.NewRegistry()
+	require.NoError(t, metrics.Register(registry))
+
+	metrics.SetCollectorInflight("jobs", true)
+	metrics.SetCollectorInflight("jobs", false)
+	metrics.RecordCollectorSkipped("jobs")
+	metrics.RecordCollectorSkipped("jobs")
+
+	metricFamilies, err := registry.Gather()
+	require.NoError(t, err)
+
+	assert.Equal(t, float64(0), collectorGaugeValue(metricFamilies, "slurm_exporter_collector_inflight", "jobs"))
+	assert.Equal(t, float64(2), collectorCounterValue(metricFamilies, "slurm_exporter_collector_skipped_total", "jobs"))
+}
+
+func collectorGaugeValue(families []*dto.MetricFamily, metricName, collector string) float64 {
+	for _, mf := range families {
+		if mf.GetName() != metricName {
+			continue
+		}
+		for _, m := range mf.GetMetric() {
+			for _, lp := range m.GetLabel() {
+				if lp.GetName() == "collector" && lp.GetValue() == collector {
+					return m.GetGauge().GetValue()
+				}
+			}
+		}
+	}
+	return 0
+}
+
+func collectorCounterValue(families []*dto.MetricFamily, metricName, collector string) float64 {
+	for _, mf := range families {
+		if mf.GetName() != metricName {
+			continue
+		}
+		for _, m := range mf.GetMetric() {
+			for _, lp := range m.GetLabel() {
+				if lp.GetName() == "collector" && lp.GetValue() == collector {
+					return m.GetCounter().GetValue()
+				}
+			}
+		}
+	}
+	return 0
+}
+
 func TestMonitoringMetrics_RecordMetricsRequest(t *testing.T) {
 	metrics := NewMonitoringMetrics()
 	registry := prometheus.NewRegistry()