diff --git a/docs/changelog/135652.yaml b/docs/changelog/135652.yaml
new file mode 100644
index 0000000000000..93a7dd66c034b
--- /dev/null
+++ b/docs/changelog/135652.yaml
@@ -0,0 +1,5 @@
+pr: 135652
+summary: DFS search phase per shard duration APM metric
+area: Search
+type: enhancement
+issues: []
diff --git a/server/src/main/java/org/elasticsearch/index/search/stats/ShardSearchPhaseAPMMetrics.java b/server/src/main/java/org/elasticsearch/index/search/stats/ShardSearchPhaseAPMMetrics.java
index ffd18d9712e3a..fc823f4c51f7c 100644
--- a/server/src/main/java/org/elasticsearch/index/search/stats/ShardSearchPhaseAPMMetrics.java
+++ b/server/src/main/java/org/elasticsearch/index/search/stats/ShardSearchPhaseAPMMetrics.java
@@ -22,13 +22,20 @@
 
 public final class ShardSearchPhaseAPMMetrics implements SearchOperationListener {
 
+    public static final String DFS_SEARCH_PHASE_METRIC = "es.search.shards.phases.dfs.duration.histogram";
     public static final String QUERY_SEARCH_PHASE_METRIC = "es.search.shards.phases.query.duration.histogram";
     public static final String FETCH_SEARCH_PHASE_METRIC = "es.search.shards.phases.fetch.duration.histogram";
 
+    private final LongHistogram dfsPhaseMetric;
     private final LongHistogram queryPhaseMetric;
     private final LongHistogram fetchPhaseMetric;
 
     public ShardSearchPhaseAPMMetrics(MeterRegistry meterRegistry) {
+        this.dfsPhaseMetric = meterRegistry.registerLongHistogram(
+            DFS_SEARCH_PHASE_METRIC,
+            "DFS search phase execution times at the shard level, expressed as a histogram",
+            "ms"
+        );
         this.queryPhaseMetric = meterRegistry.registerLongHistogram(
             QUERY_SEARCH_PHASE_METRIC,
             "Query search phase execution times at the shard level, expressed as a histogram",
@@ -41,6 +48,11 @@ public ShardSearchPhaseAPMMetrics(MeterRegistry meterRegistry) {
         );
     }
 
+    @Override
+    public void onDfsPhase(SearchContext searchContext, long tookInNanos) {
+        recordPhaseLatency(dfsPhaseMetric, tookInNanos);
+    }
+
     @Override
     public void onQueryPhase(SearchContext searchContext, long tookInNanos) {
         SearchExecutionContext searchExecutionContext = searchContext.getSearchExecutionContext();
@@ -55,6 +67,10 @@ public void onFetchPhase(SearchContext searchContext, long tookInNanos) {
         recordPhaseLatency(fetchPhaseMetric, tookInNanos, searchContext.request(), rangeTimestampFrom);
     }
 
+    private static void recordPhaseLatency(LongHistogram histogramMetric, long tookInNanos) {
+        histogramMetric.record(TimeUnit.NANOSECONDS.toMillis(tookInNanos));
+    }
+
     private static void recordPhaseLatency(
         LongHistogram histogramMetric,
         long tookInNanos,
diff --git a/server/src/test/java/org/elasticsearch/search/TelemetryMetrics/ShardSearchPhaseAPMMetricsTests.java b/server/src/test/java/org/elasticsearch/search/TelemetryMetrics/ShardSearchPhaseAPMMetricsTests.java
index 25fa7d1f36b41..583a2e3fa5803 100644
--- a/server/src/test/java/org/elasticsearch/search/TelemetryMetrics/ShardSearchPhaseAPMMetricsTests.java
+++ b/server/src/test/java/org/elasticsearch/search/TelemetryMetrics/ShardSearchPhaseAPMMetricsTests.java
@@ -34,6 +34,7 @@
 
 import static org.elasticsearch.action.support.WriteRequest.RefreshPolicy.IMMEDIATE;
 import static org.elasticsearch.index.query.QueryBuilders.simpleQueryStringQuery;
+import static org.elasticsearch.index.search.stats.ShardSearchPhaseAPMMetrics.DFS_SEARCH_PHASE_METRIC;
 import static org.elasticsearch.index.search.stats.ShardSearchPhaseAPMMetrics.FETCH_SEARCH_PHASE_METRIC;
 import static org.elasticsearch.index.search.stats.ShardSearchPhaseAPMMetrics.QUERY_SEARCH_PHASE_METRIC;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
@@ -92,6 +93,8 @@ public void testMetricsDfsQueryThenFetch() {
             client().prepareSearch(indexName).setSearchType(SearchType.DFS_QUERY_THEN_FETCH).setQuery(simpleQueryStringQuery("doc1")),
             "1"
         );
+        final List<Measurement> dfsMeasurements = getTestTelemetryPlugin().getLongHistogramMeasurement(DFS_SEARCH_PHASE_METRIC);
+        assertEquals(num_primaries, dfsMeasurements.size());
         final List<Measurement> queryMeasurements = getTestTelemetryPlugin().getLongHistogramMeasurement(QUERY_SEARCH_PHASE_METRIC);
         assertEquals(num_primaries, queryMeasurements.size());
         final List<Measurement> fetchMeasurements = getTestTelemetryPlugin().getLongHistogramMeasurement(FETCH_SEARCH_PHASE_METRIC);
@@ -106,6 +109,8 @@ public void testMetricsDfsQueryThenFetchSystem() {
                 .setQuery(simpleQueryStringQuery("doc1")),
             "1"
         );
+        final List<Measurement> dfsMeasurements = getTestTelemetryPlugin().getLongHistogramMeasurement(DFS_SEARCH_PHASE_METRIC);
+        assertEquals(0, dfsMeasurements.size()); // DFS phase not done for index with single shard
         final List<Measurement> queryMeasurements = getTestTelemetryPlugin().getLongHistogramMeasurement(QUERY_SEARCH_PHASE_METRIC);
         assertEquals(1, queryMeasurements.size());
         final List<Measurement> fetchMeasurements = getTestTelemetryPlugin().getLongHistogramMeasurement(FETCH_SEARCH_PHASE_METRIC);