add prirep and index tags to elasticsearch.shards metric

Author: VJean

elastic/datadog_checks/elastic/data/conf.yaml.example

@@ -112,7 +112,12 @@ instances:
 
     ## @param cat_allocation_stats - boolean - optional - default: false
     ## Enable to collect Elastic Cat Allocation metrics. Available only for Elasticsearch 5.0 or higher.
-    ## Ref: https://www.elastic.co/guide/en/elasticsearch/reference/current/cat-allocation.html.
+    ## Collects disk usage metrics per node and detailed shard placement metrics.
+    ## The elasticsearch.shards metric includes index and prirep tags for monitoring
+    ## shard placement skew and imbalance between nodes.
+    ## Tags: node_name:<node>, index:<index_name>, prirep:primary or prirep:replica
+    ## Ref: https://www.elastic.co/guide/en/elasticsearch/reference/current/cat-allocation.html
+    ##      https://www.elastic.co/guide/en/elasticsearch/reference/current/cat-shards.html
     #
     # cat_allocation_stats: false
 

elastic/datadog_checks/elastic/elastic.py

@@ -99,6 +99,7 @@ class ESCheck(AgentCheck):
     SERVICE_CHECK_CONNECT_NAME = 'elasticsearch.can_connect'
     SERVICE_CHECK_CLUSTER_STATUS = 'elasticsearch.cluster_health'
     CAT_ALLOC_PATH = '/_cat/allocation?v=true&format=json&bytes=b'
+    CAT_SHARDS_PATH = '/_cat/shards?format=json&bytes=b'
     SOURCE_TYPE_NAME = 'elasticsearch'
 
     def __init__(self, name, init_config, instances):
@@ -490,16 +491,18 @@ def _process_cat_allocation_data(self, admin_forwarder, version, base_tags):
             return
 
         self.log.debug("Collecting cat allocation metrics")
+
+        # Collect disk metrics from /_cat/allocation
         cat_allocation_url = self._join_url(self.CAT_ALLOC_PATH, admin_forwarder)
         try:
             cat_allocation_data = self._get_data(cat_allocation_url)
         except requests.ReadTimeout as e:
             self.log.error("Timed out reading cat allocation stats from servers (%s) - stats will be missing", e)
             return
 
-        # we need to remap metric names because the ones from elastic
-        # contain dots and that would confuse `_process_metric()` (sic)
-        data_to_collect = {'disk.indices', 'disk.used', 'disk.avail', 'disk.total', 'disk.percent', 'shards'}
+        # Process disk metrics (we need to remap metric names because the ones from elastic
+        # contain dots and that would confuse `_process_metric()`)
+        data_to_collect = {'disk.indices', 'disk.used', 'disk.avail', 'disk.total', 'disk.percent'}
         for dic in cat_allocation_data:
             cat_allocation_dic = {
                 k.replace('.', '_'): v for k, v in dic.items() if k in data_to_collect and v is not None
@@ -509,6 +512,43 @@ def _process_cat_allocation_data(self, admin_forwarder, version, base_tags):
                 desc = CAT_ALLOCATION_METRICS[metric]
                 self._process_metric(cat_allocation_dic, metric, *desc, tags=tags)
 
+        # Collect detailed shard placement from /_cat/shards
+        self.log.debug("Collecting detailed shard placement metrics")
+        cat_shards_url = self._join_url(self.CAT_SHARDS_PATH, admin_forwarder)
+        try:
+            cat_shards_data = self._get_data(cat_shards_url)
+        except requests.ReadTimeout as e:
+            self.log.error("Timed out reading cat shards stats from servers (%s) - stats will be missing", e)
+            return
+
+        # Group shards by node, index, and prirep to count them
+        from collections import Counter
+        shard_counts = Counter()
+
+        for shard in cat_shards_data:
+            node = shard.get('node')
+            index = shard.get('index')
+            prirep_raw = shard.get('prirep')
+
+            # Skip unassigned shards (they have no node)
+            if node is None or node == 'UNASSIGNED':
+                continue
+
+            # Map p/r to primary/replica for better readability
+            prirep = 'primary' if prirep_raw == 'p' else 'replica'
+
+            key = (node, index, prirep)
+            shard_counts[key] += 1
+
+        # Submit metrics with tags
+        for (node, index, prirep), count in shard_counts.items():
+            tags = base_tags + [
+                'node_name:{}'.format(node.lower()),
+                'index:{}'.format(index),
+                'prirep:{}'.format(prirep)
+            ]
+            self.gauge('elasticsearch.shards', count, tags=tags)
+
     def _process_custom_metric(
         self,
         value,

elastic/datadog_checks/elastic/metrics.py

@@ -766,7 +766,6 @@
 }
 
 CAT_ALLOCATION_METRICS = {
-    'elasticsearch.shards': ('gauge', 'shards'),
     'elasticsearch.disk.indices': ('gauge', 'disk_indices'),
     'elasticsearch.disk.used': ('gauge', 'disk_used'),
     'elasticsearch.disk.avail': ('gauge', 'disk_avail'),

elastic/metadata.csv

@@ -187,7 +187,7 @@ elasticsearch.search.scroll.time,gauge,,second,,The total time spent on scroll q
 elasticsearch.search.scroll.time.count,count,,second,,The total time spent on scroll queries submitted as a count [v5+].,0,elasticsearch,total scroll query time,
 elasticsearch.search.scroll.total,gauge,,query,,The total number of scroll queries [v5+].,0,elasticsearch,total scroll queries,
 elasticsearch.search.scroll.total.count,count,,query,,The total number of scroll queries submitted as a count [v5+].,0,elasticsearch,total scroll queries,
-elasticsearch.shards,gauge,,shard,,Number of primary and replica shards assigned to the node.,0,elasticsearch,shards,
+elasticsearch.shards,gauge,,shard,,Number of shards per node-index-prirep combination. Collected when cat_allocation_stats is enabled. Includes tags: node_name index and prirep (primary or replica) for detailed shard placement analysis.,0,elasticsearch,shards,
 elasticsearch.slm.snapshot_deletion_failures,gauge,,error,,The total number of snapshot deletion failures.,0,elasticsearch,slm del fail,
 elasticsearch.slm.snapshots_deleted,gauge,,,,The total number of deleted snapshots.,0,elasticsearch,slm del,
 elasticsearch.slm.snapshots_failed,gauge,,error,,The total number of failed snapshots.,0,elasticsearch,slm fail,

elastic/tests/test_integration.py

@@ -360,9 +360,31 @@ def test_cat_allocation_metrics(dd_environment, aggregator, instance, cluster_ta
     elastic_check = ESCheck('elastic', {}, instances=[instance])
 
     elastic_check.check(None)
+
+    # Check disk metrics
     for m_name in CAT_ALLOCATION_METRICS:
         aggregator.assert_metric(m_name)
 
+    # Check detailed shard placement metrics
+    aggregator.assert_metric('elasticsearch.shards')
+
+    # Verify tags are present - we should have index and prirep tags
+    metrics = aggregator.metrics('elasticsearch.shards')
+    assert len(metrics) > 0, "Expected at least one elasticsearch.shards metric"
+
+    # Check that at least one metric has both index and prirep tags
+    has_index_tag = False
+    has_prirep_tag = False
+    for metric in metrics:
+        tags = metric.tags
+        if any(tag.startswith('index:') for tag in tags):
+            has_index_tag = True
+        if any(tag.startswith('prirep:') for tag in tags):
+            has_prirep_tag = True
+
+    assert has_index_tag, "Expected index tag on elasticsearch.shards metric"
+    assert has_prirep_tag, "Expected prirep tag on elasticsearch.shards metric"
+
 
 def test_health_event(dd_environment, aggregator):
     dummy_tags = ['elastique:recherche']