Add elasticsearch/ingest_pipeline metricset

CHANGELOG.next.asciidoc

@@ -197,6 +197,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 - Add GCP Redis regions support {pull}33728[33728]
 - Add namespace metadata to all namespaced kubernetes resources. {pull}33763[33763]
 - Changed cloudwatch module to call ListMetrics API only once per region, instead of per AWS namespace {pull}34055[34055]
+- Add beta ingest_pipeline metricset to Elasticsearch module for ingest pipeline monitoring {pull}34012[34012]
 - Handle duplicated TYPE line for prometheus metrics {issue}18813[18813] {pull}33865[33865]
 
 *Packetbeat*

metricbeat/docs/fields.asciidoc

@@ -29995,6 +29995,16 @@ type: keyword
 Node name.
 
 
+type: keyword
+
+--
+
+*`elasticsearch.node.roles`*::
++
+--
+Node roles.
+
+
 type: keyword
 
 --
@@ -31677,6 +31687,118 @@ type: long
 
 --
 
+[float]
+=== ingest_pipeline
+
+Runtime metrics on ingest pipeline execution
+
+
+*`elasticsearch.ingest_pipeline.name`*::
++
+--
+Name / id of the ingest pipeline
+
+type: wildcard
+
+--
+
+[float]
+=== total
+
+Metrics on the total ingest pipeline execution, including all processors.
+
+
+*`elasticsearch.ingest_pipeline.total.count`*::
++
+--
+Number of documents processed by this pipeline
+
+type: long
+
+--
+
+*`elasticsearch.ingest_pipeline.total.failed`*::
++
+--
+Number of documented failed to process by this pipeline
+
+type: long
+
+--
+
+*`elasticsearch.ingest_pipeline.total.time.total.ms`*::
++
+--
+Total time spent processing documents through this pipeline, inclusive of other pipelines called
+
+type: long
+
+--
+
+*`elasticsearch.ingest_pipeline.total.time.self.ms`*::
++
+--
+Time spent processing documents through this pipeline, exclusive of other pipelines called
+
+type: long
+
+--
+
+
+*`elasticsearch.ingest_pipeline.processor.type`*::
++
+--
+The type of ingest processor
+
+type: keyword
+
+--
+
+*`elasticsearch.ingest_pipeline.processor.type_tag`*::
++
+--
+The type and the tag for this processor in the format "<type>:<tag>"
+
+type: keyword
+
+--
+
+*`elasticsearch.ingest_pipeline.processor.order_index`*::
++
+--
+The order this processor appears in the pipeline definition
+
+type: long
+
+--
+
+*`elasticsearch.ingest_pipeline.processor.count`*::
++
+--
+Number of documents processed by this processor
+
+type: long
+
+--
+
+*`elasticsearch.ingest_pipeline.processor.failed`*::
++
+--
+Number of documented failed to process by this processor
+
+type: long
+
+--
+
+*`elasticsearch.ingest_pipeline.processor.time.total.ms`*::
++
+--
+Total time spent processing documents through this processor
+
+type: long
+
+--
+
 [float]
 === ml.job
 

metricbeat/docs/modules/elasticsearch.asciidoc

@@ -64,6 +64,7 @@ metricbeat.modules:
     #- index
     #- index_recovery
     #- index_summary
+    #- ingest_pipeline
     #- shard
     #- ml_job
   period: 10s
@@ -73,6 +74,7 @@ metricbeat.modules:
   #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
   #index_recovery.active_only: true
+  #ingest_pipeline.processor_sample_rate: 0.25
   #xpack.enabled: false
   #scope: node
 ----
@@ -97,6 +99,8 @@ The following metricsets are available:
 
 * <<metricbeat-metricset-elasticsearch-index_summary,index_summary>>
 
+* <<metricbeat-metricset-elasticsearch-ingest_pipeline,ingest_pipeline>>
+
 * <<metricbeat-metricset-elasticsearch-ml_job,ml_job>>
 
 * <<metricbeat-metricset-elasticsearch-node,node>>
@@ -119,6 +123,8 @@ include::elasticsearch/index_recovery.asciidoc[]
 
 include::elasticsearch/index_summary.asciidoc[]
 
+include::elasticsearch/ingest_pipeline.asciidoc[]
+
 include::elasticsearch/ml_job.asciidoc[]
 
 include::elasticsearch/node.asciidoc[]

metricbeat/docs/modules/elasticsearch/ingest_pipeline.asciidoc

@@ -0,0 +1,29 @@
+////
+This file is generated! See scripts/mage/docs_collector.go
+////
+:edit_url: https://github.com/elastic/beats/edit/main/metricbeat/module/elasticsearch/ingest_pipeline/_meta/docs.asciidoc
+
+
+[[metricbeat-metricset-elasticsearch-ingest_pipeline]]
+=== Elasticsearch ingest_pipeline metricset
+
+beta[]
+
+include::../../../module/elasticsearch/ingest_pipeline/_meta/docs.asciidoc[]
+
+This is a default metricset. If the host module is unconfigured, this metricset is enabled by default.
+
+:edit_url:
+
+==== Fields
+
+For a description of each field in the metricset, see the
+<<exported-fields-elasticsearch,exported fields>> section.
+
+Here is an example document generated by this metricset:
+
+[source,json]
+----
+include::../../../module/elasticsearch/ingest_pipeline/_meta/data.json[]
+----
+:edit_url!:

metricbeat/docs/modules_list.asciidoc

@@ -98,12 +98,13 @@ This file is generated! See scripts/mage/docs_collector.go
 |<<metricbeat-module-dropwizard,Dropwizard>>     |image:./images/icon-no.png[No prebuilt dashboards]    |  
 .1+| .1+|  |<<metricbeat-metricset-dropwizard-collector,collector>>   
 |<<metricbeat-module-elasticsearch,Elasticsearch>>     |image:./images/icon-no.png[No prebuilt dashboards]    |  
-.11+| .11+|  |<<metricbeat-metricset-elasticsearch-ccr,ccr>>   
+.12+| .12+|  |<<metricbeat-metricset-elasticsearch-ccr,ccr>>   
 |<<metricbeat-metricset-elasticsearch-cluster_stats,cluster_stats>>   
 |<<metricbeat-metricset-elasticsearch-enrich,enrich>>   
 |<<metricbeat-metricset-elasticsearch-index,index>>   
 |<<metricbeat-metricset-elasticsearch-index_recovery,index_recovery>>   
 |<<metricbeat-metricset-elasticsearch-index_summary,index_summary>>   
+|<<metricbeat-metricset-elasticsearch-ingest_pipeline,ingest_pipeline>> beta[]  
 |<<metricbeat-metricset-elasticsearch-ml_job,ml_job>>   
 |<<metricbeat-metricset-elasticsearch-node,node>>   
 |<<metricbeat-metricset-elasticsearch-node_stats,node_stats>>   

metricbeat/metricbeat.reference.yml

@@ -275,6 +275,7 @@ metricbeat.modules:
     #- index
     #- index_recovery
     #- index_summary
+    #- ingest_pipeline
     #- shard
     #- ml_job
   period: 10s
@@ -284,6 +285,7 @@ metricbeat.modules:
   #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
   #index_recovery.active_only: true
+  #ingest_pipeline.processor_sample_rate: 0.25
   #xpack.enabled: false
   #scope: node
 

metricbeat/module/elasticsearch/_meta/config.reference.yml

@@ -5,6 +5,7 @@
     #- index
     #- index_recovery
     #- index_summary
+    #- ingest_pipeline
     #- shard
     #- ml_job
   period: 10s
@@ -14,5 +15,6 @@
   #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
   #index_recovery.active_only: true
+  #ingest_pipeline.processor_sample_rate: 0.25
   #xpack.enabled: false
   #scope: node

metricbeat/module/elasticsearch/_meta/fields.yml

@@ -676,6 +676,10 @@
               type: keyword
               description: >
                 Node name.
+            - name: roles
+              type: keyword
+              description: >
+                Node roles.
             - name: master
               type: boolean
               description: >

metricbeat/module/elasticsearch/elasticsearch.go

@@ -56,7 +56,8 @@ func NewModule(base mb.BaseModule) (mb.Module, error) {
 		"node_stats",
 		"shard",
 	}
-	return elastic.NewModule(&base, xpackEnabledMetricSets, []string{}, logp.NewLogger(ModuleName))
+	optionalXpackMetricsets := []string{"ingest_pipeline"}
+	return elastic.NewModule(&base, xpackEnabledMetricSets, optionalXpackMetricsets, logp.NewLogger(ModuleName))
 }
 
 var (

metricbeat/module/elasticsearch/ingest_pipeline/_meta/data.json

@@ -0,0 +1,58 @@
+{
+  "@timestamp": "2023-01-19T19:08:28.770Z",
+  "elasticsearch": {
+    "cluster": {
+      "id": "XXXX",
+      "name": "my-cluster"
+    },
+    "node": {
+      "name": "instance-0000000022",
+      "roles": [
+        "ingest",
+        "remote_cluster_client"
+      ],
+      "id": "ZZZZZ"
+    },
+    "ingest_pipeline": {
+      "name": "my-pipeline",
+      "total": {
+        "count": 64,
+        "failed": 0,
+        "time": {
+          "total": {
+            "ms": 39
+          },
+          "self": {
+            "ms": 39
+          }
+        }
+      }
+    }
+  },
+  "ecs": {
+    "version": "8.0.0"
+  },
+  "host": {
+    "name": "myhost.local"
+  },
+  "agent": {
+    "type": "metricbeat",
+    "version": "8.7.0",
+    "ephemeral_id": "80fe51b6-57df-46d1-9240-20424df6d675",
+    "id": "faab8154-7ceb-470f-b2ca-0aee636951c3",
+    "name": "myhost.local"
+  },
+  "event": {
+    "dataset": "elasticsearch.ingest_pipeline",
+    "module": "elasticsearch",
+    "duration": 558169708
+  },
+  "metricset": {
+    "period": 10000,
+    "name": "ingest_pipeline"
+  },
+  "service": {
+    "address": "https://localhost:9200",
+    "type": "elasticsearch"
+  }
+}

metricbeat/module/elasticsearch/ingest_pipeline/_meta/docs.asciidoc

@@ -0,0 +1,22 @@
+This is the ingest_pipeline metricset of the module elasticsearch.
+
+Collects metrics on ingest pipeline executions, with processor-level granularity.
+
+[float]
+=== Processor-level metrics sampling
+
+Processor-level metrics can produce a high volume of data, so the default behavior is to collect those metrics less
+frequently than the `period` for pipeline-level metrics, by applying a sampling strategy. By default, the
+processor-level metrics will be collected during 25% of the time. This can be configured with the
+`ingest.processor_sample_rate` setting:
+
+[float]
+=== Configuration example
+[source,yaml]
+----
+- module: elasticsearch
+  period: 10s
+  metricsets:
+    - ingest
+  ingest.processor_sample_rate: 0.1 # decrease to 10% of fetches
+----