feat: add metrics for merge operations with total shard aggregation

.chloggen/elasticsearch-total-merge.yaml

@@ -0,0 +1,5 @@
+change_type: enhancement
+component: elasticsearchreceiver
+note: Add metrics related to merge operations with aggregated for all shards
+issues: [14635]
+

receiver/elasticsearchreceiver/documentation.md

@@ -23,6 +23,8 @@ These are the metrics available for this scraper.
 | **elasticsearch.cluster.state_update.count** | The number of cluster state update attempts that changed the cluster state since the node started. | 1 | Sum(Int) | <ul> <li>cluster_state_update_state</li> </ul> |
 | **elasticsearch.cluster.state_update.time** | The cumulative amount of time updating the cluster state since the node started. | ms | Sum(Int) | <ul> <li>cluster_state_update_state</li> <li>cluster_state_update_type</li> </ul> |
 | **elasticsearch.index.operations.completed** | The number of operations completed for an index. | {operations} | Sum(Int) | <ul> <li>operation</li> <li>index_aggregation_type</li> </ul> |
+| elasticsearch.index.operations.merge.docs_count | The total number of documents in merge operations for an index. | {documents} | Sum(Int) | <ul> <li>index_aggregation_type</li> </ul> |
+| elasticsearch.index.operations.merge.size | The total size of merged segments for an index. | By | Sum(Int) | <ul> <li>index_aggregation_type</li> </ul> |
 | **elasticsearch.index.operations.time** | Time spent on operations for an index. | ms | Sum(Int) | <ul> <li>operation</li> <li>index_aggregation_type</li> </ul> |
 | **elasticsearch.index.shards.size** | The size of the shards assigned to this index. | By | Sum(Int) | <ul> <li>index_aggregation_type</li> </ul> |
 | **elasticsearch.indexing_pressure.memory.limit** | Configured memory limit, in bytes, for the indexing requests. | By | Gauge(Int) | <ul> </ul> |

receiver/elasticsearchreceiver/internal/model/nodestats.go

@@ -169,7 +169,7 @@ type NodeStatsNodesInfoIndices struct {
 	IndexingOperations IndexingOperations  `json:"indexing"`
 	GetOperation       GetOperation        `json:"get"`
 	SearchOperations   SearchOperations    `json:"search"`
-	MergeOperations    BasicIndexOperation `json:"merges"`
+	MergeOperations    MergeOperations     `json:"merges"`
 	RefreshOperations  BasicIndexOperation `json:"refresh"`
 	FlushOperations    BasicIndexOperation `json:"flush"`
 	WarmerOperations   BasicIndexOperation `json:"warmer"`
@@ -195,6 +195,12 @@ type BasicIndexOperation struct {
 	TotalTimeInMs int64 `json:"total_time_in_millis"`
 }
 
+type MergeOperations struct {
+	BasicIndexOperation
+	TotalSizeInBytes int64 `json:"total_size_in_bytes"`
+	TotalDocs        int64 `json:"total_docs"`
+}
+
 type IndexingOperations struct {
 	IndexTotal     int64 `json:"index_total"`
 	IndexTimeInMs  int64 `json:"index_time_in_millis"`

receiver/elasticsearchreceiver/metadata.yaml

@@ -767,3 +767,21 @@ metrics:
       value_type: int
     attributes: [index_aggregation_type]
     enabled: true
+  elasticsearch.index.operations.merge.size:
+    description: The total size of merged segments for an index.
+    unit: By
+    sum:
+      monotonic: true
+      aggregation: cumulative
+      value_type: int
+    attributes: [index_aggregation_type]
+    enabled: false
+  elasticsearch.index.operations.merge.docs_count:
+    description: The total number of documents in merge operations for an index.
+    unit: "{documents}"
+    sum:
+      monotonic: true
+      aggregation: cumulative
+      value_type: int
+    attributes: [index_aggregation_type]
+    enabled: false

receiver/elasticsearchreceiver/scraper.go

@@ -333,7 +333,7 @@ func (r *elasticsearchScraper) scrapeIndicesMetrics(ctx context.Context, now pco
 	indexStats, err := r.client.IndexStats(ctx, r.cfg.Indices)
 
 	if err != nil {
-		errs.AddPartial(4, err)
+		errs.AddPartial(8, err)
 		return
 	}
 
@@ -359,6 +359,16 @@ func (r *elasticsearchScraper) scrapeOneIndexMetrics(now pcommon.Timestamp, name
 	r.mb.RecordElasticsearchIndexOperationsTimeDataPoint(
 		now, stats.Total.SearchOperations.QueryTimeInMs, metadata.AttributeOperationQuery, metadata.AttributeIndexAggregationTypeTotal,
 	)
+	r.mb.RecordElasticsearchIndexOperationsTimeDataPoint(
+		now, stats.Total.MergeOperations.TotalTimeInMs, metadata.AttributeOperationMerge, metadata.AttributeIndexAggregationTypeTotal,
+	)
+
+	r.mb.RecordElasticsearchIndexOperationsMergeSizeDataPoint(
+		now, stats.Total.MergeOperations.TotalSizeInBytes, metadata.AttributeIndexAggregationTypeTotal,
+	)
+	r.mb.RecordElasticsearchIndexOperationsMergeDocsCountDataPoint(
+		now, stats.Total.MergeOperations.TotalDocs, metadata.AttributeIndexAggregationTypeTotal,
+	)
 
 	r.mb.RecordElasticsearchIndexShardsSizeDataPoint(
 		now, stats.Total.StoreInfo.SizeInBy, metadata.AttributeIndexAggregationTypeTotal,

receiver/elasticsearchreceiver/scraper_test.go

@@ -41,7 +41,11 @@ const noNodesExpectedMetricsPath = "./testdata/expected_metrics/noNodes.json"
 func TestScraper(t *testing.T) {
 	t.Parallel()
 
-	sc := newElasticSearchScraper(componenttest.NewNopReceiverCreateSettings(), createDefaultConfig().(*Config))
+	config := createDefaultConfig().(*Config)
+	config.Metrics.ElasticsearchIndexOperationsMergeSize.Enabled = true
+	config.Metrics.ElasticsearchIndexOperationsMergeDocsCount.Enabled = true
+
+	sc := newElasticSearchScraper(componenttest.NewNopReceiverCreateSettings(), config)
 
 	err := sc.start(context.Background(), componenttest.NewNopHost())
 	require.NoError(t, err)

receiver/elasticsearchreceiver/testdata/expected_metrics/clusterSkip.json

@@ -2297,6 +2297,25 @@
                               ],
                               "startTimeUnixNano": "1661811689941624000",
                               "timeUnixNano": "1661811689943245000"
+                           },
+                           {
+                              "asInt": "12",
+                              "attributes": [
+                                 {
+                                    "key": "operation",
+                                    "value": {
+                                       "stringValue": "merge"
+                                    }
+                                 },
+                                 {
+                                    "key": "aggregation",
+                                    "value": {
+                                       "stringValue": "total"
+                                    }
+                                 }
+                              ],
+                              "startTimeUnixNano": "1661811689941624000",
+                              "timeUnixNano": "1661811689943245000"
                            }
                         ]
                      },
@@ -2447,6 +2466,25 @@
                               ],
                               "startTimeUnixNano": "1661811689941624000",
                               "timeUnixNano": "1661811689943245000"
+                           },
+                           {
+                              "asInt": "12",
+                              "attributes": [
+                                 {
+                                    "key": "operation",
+                                    "value": {
+                                       "stringValue": "merge"
+                                    }
+                                 },
+                                 {
+                                    "key": "aggregation",
+                                    "value": {
+                                       "stringValue": "total"
+                                    }
+                                 }
+                              ],
+                              "startTimeUnixNano": "1661811689941624000",
+                              "timeUnixNano": "1661811689943245000"
                            }
                         ]
                      },