[receiver/elasticsearch]: add metrics for merge operations with total…

… shard aggregation (open-telemetry#14870)

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"
  }
  ]
  },