Make elasticsearch/enrich metricset work for Stack Monitoring without…

… xpack.enabled flag (#20821)
elastic · Nov 19, 2020 · 1b5d63f · 1b5d63f
1 parent d440d2d
commit 1b5d63f
Show file tree

Hide file tree

Showing 10 changed files with 227 additions and 176 deletions.
diff --git a/metricbeat/docs/fields.asciidoc b/metricbeat/docs/fields.asciidoc
@@ -18457,31 +18457,69 @@ Enrich stats
 
 
 
-*`elasticsearch.enrich.queue.size`*::
+
+*`elasticsearch.enrich.executing_policy.name`*::
 +
 --
-Number of search requests in the queue.
+type: keyword
+
+--
 
 
+*`elasticsearch.enrich.executing_policy.task.id`*::
++
+--
 type: long
 
 --
 
+*`elasticsearch.enrich.executing_policy.task.task`*::
++
+--
+type: keyword
 
-*`elasticsearch.enrich.remote_requests.current`*::
+--
+
+*`elasticsearch.enrich.executing_policy.task.action`*::
 +
 --
-Current number of outstanding remote requests.
+type: keyword
 
+--
 
+*`elasticsearch.enrich.executing_policy.task.cancellable`*::
++
+--
+type: boolean
+
+--
+
+*`elasticsearch.enrich.executing_policy.task.parent_task_id`*::
++
+--
+type: keyword
+
+--
+
+
+*`elasticsearch.enrich.executing_policy.task.time.start.ms`*::
++
+--
 type: long
 
 --
 
-*`elasticsearch.enrich.remote_requests.total`*::
+*`elasticsearch.enrich.executing_policy.task.time.running.nano`*::
 +
 --
-Number of outstanding remote requests executed since node startup.
+type: long
+
+--
+
+*`elasticsearch.enrich.queue.size`*::
++
+--
+Number of search requests in the queue.
 
 
 type: long
@@ -18494,6 +18532,27 @@ type: long
 Number of search requests that enrich processors have executed since node startup.
 
 
+type: long
+
+--
+
+
+*`elasticsearch.enrich.remote_requests.current`*::
++
+--
+Current number of outstanding remote requests.
+
+
+type: long
+
+--
+
+*`elasticsearch.enrich.remote_requests.total`*::
++
+--
+Number of outstanding remote requests executed since node startup.
+
+
 type: long
 
 --
@@ -19148,12 +19207,6 @@ Statistics about each node in a Elasticsearch cluster
 
 
 
-[float]
-=== indices
-
-Node indices stats
-
-
 
 *`elasticsearch.node.stats.indices.docs.count`*::
 +
@@ -19385,18 +19438,6 @@ format: percent
 
 --
 
-[float]
-=== jvm.mem.pools
-
-JVM memory pool stats
-
-
-
-[float]
-=== old
-
-Old memory pool stats.
-
 
 
 *`elasticsearch.node.stats.jvm.mem.pools.old.max.bytes`*::
@@ -19447,12 +19488,6 @@ format: bytes
 
 --
 
-[float]
-=== young
-
-Young memory pool stats.
-
-
 
 *`elasticsearch.node.stats.jvm.mem.pools.young.max.bytes`*::
 +
@@ -19502,12 +19537,6 @@ format: bytes
 
 --
 
-[float]
-=== survivor
-
-Survivor memory pool stats.
-
-
 
 *`elasticsearch.node.stats.jvm.mem.pools.survivor.max.bytes`*::
 +
@@ -19557,18 +19586,6 @@ format: bytes
 
 --
 
-[float]
-=== jvm.gc.collectors
-
-GC collector stats.
-
-
-
-[float]
-=== old.collection
-
-Old collection gc.
-
 
 
 *`elasticsearch.node.stats.jvm.gc.collectors.old.collection.count`*::
@@ -19585,12 +19602,6 @@ type: long
 
 --
 
-[float]
-=== young.collection
-
-Young collection gc.
-
-
 
 *`elasticsearch.node.stats.jvm.gc.collectors.young.collection.count`*::
 +
@@ -19617,6 +19628,8 @@ File system summary
 *`elasticsearch.node.stats.fs.summary.total.bytes`*::
 +
 --
+
+
 type: long
 
 format: bytes
@@ -19626,6 +19639,8 @@ format: bytes
 *`elasticsearch.node.stats.fs.summary.free.bytes`*::
 +
 --
+
+
 type: long
 
 format: bytes
@@ -19635,6 +19650,8 @@ format: bytes
 *`elasticsearch.node.stats.fs.summary.available.bytes`*::
 +
 --
+
+
 type: long
 
 format: bytes

diff --git a/metricbeat/module/elasticsearch/elasticsearch_integration_test.go b/metricbeat/module/elasticsearch/elasticsearch_integration_test.go
@@ -129,7 +129,7 @@ func TestXPackEnabled(t *testing.T) {
 	metricSetToTypesMap := map[string][]string{
 		"ccr":            []string{"ccr_stats", "ccr_auto_follow_stats"},
 		"cluster_stats":  []string{"cluster_stats"},
-		"enrich":         []string{"enrich_coordinator_stats"},
+		"enrich":         []string{}, // no longer indexed into .monitoring-es-*
 		"index_recovery": []string{"index_recovery"},
 		"index_summary":  []string{}, // no longer indexed into .monitoring-es-*
 		"ml_job":         []string{"job_stats"},

diff --git a/metricbeat/module/elasticsearch/enrich/_meta/data.json b/metricbeat/module/elasticsearch/enrich/_meta/data.json
@@ -2,23 +2,23 @@
     "@timestamp": "2017-10-12T08:05:34.853Z",
     "elasticsearch": {
         "cluster": {
-            "id": "et6blfihSoytMUvkpYtEKQ",
+            "id": "8l_zoGznQRmtoX9iSC-goA",
             "name": "docker-cluster"
         },
         "enrich": {
             "executed_searches": {
-                "total": 1
+                "total": 0
             },
             "queue": {
                 "size": 0
             },
             "remote_requests": {
                 "current": 0,
-                "total": 1
+                "total": 0
             }
         },
         "node": {
-            "id": "l_XOyQ65Teyn4kW4PUFjVg"
+            "id": "1sFM8cmSROZYhPxVsiWew"
         }
     },
     "event": {
@@ -31,8 +31,7 @@
         "period": 10000
     },
     "service": {
-        "address": "localhost:32780",
-        "name": "elasticsearch",
+        "address": "127.0.0.1:51380",
         "type": "elasticsearch"
     }
 }
diff --git a/metricbeat/module/elasticsearch/enrich/_meta/fields.yml b/metricbeat/module/elasticsearch/enrich/_meta/fields.yml
@@ -4,10 +4,39 @@
     Enrich stats
   release: ga
   fields:
+    - name: executing_policy
+      type: group
+      fields:
+        - name: name
+          type: keyword
+        - name: task
+          type: group
+          fields:
+          - name: id
+            type: long
+          - name: task
+            type: keyword
+          - name: action
+            type: keyword
+          - name: cancellable
+            type: boolean
+          - name: parent_task_id
+            type: keyword
+          - name: time
+            type: group
+            fields:
+              - name: start.ms
+                type: long
+              - name: running.nano
+                type: long
     - name: queue.size
       type: long
       description: >
         Number of search requests in the queue.
+    - name: executed_searches.total
+      type: long
+      description: >
+        Number of search requests that enrich processors have executed since node startup.
     - name: remote_requests
       type: group
       fields:
@@ -19,7 +48,3 @@
           type: long
           description: >
             Number of outstanding remote requests executed since node startup.
-    - name: executed_searches.total
-      type: long
-      description: >
-        Number of search requests that enrich processors have executed since node startup.
diff --git a/metricbeat/module/elasticsearch/enrich/data.go b/metricbeat/module/elasticsearch/enrich/data.go
@@ -44,6 +44,22 @@ var (
 			"total": c.Int("executed_searches_total"),
 		},
 	}
+
+	task = s.Schema{
+		"id":     c.Int("id"),
+		"type":   c.Str("type"),
+		"action": c.Str("action"),
+		"time": s.Object{
+			"start": s.Object{
+				"ms": c.Int("start_time_in_millis"),
+			},
+			"running": s.Object{
+				"nano": c.Int("running_time_in_nanos"),
+			},
+		},
+		"cancellable":    c.Bool("cancellable"),
+		"parent_task_id": c.Str("parent_task_id"),
+	}
 )
 
 type response struct {
@@ -62,8 +78,6 @@ func eventsMapping(r mb.ReporterV2, info elasticsearch.Info, content []byte) err
 	for _, stat := range data.CoordinatorStats {
 
 		event := mb.Event{}
-		event.RootFields = common.MapStr{}
-		event.RootFields.Put("service.name", elasticsearch.ModuleName)
 
 		event.ModuleFields = common.MapStr{}
 		event.ModuleFields.Put("cluster.name", info.ClusterName)
@@ -88,5 +102,45 @@ func eventsMapping(r mb.ReporterV2, info elasticsearch.Info, content []byte) err
 		r.Event(event)
 	}
 
+	for _, policy := range data.ExecutingPolicies {
+		event := mb.Event{}
+
+		event.ModuleFields = common.MapStr{}
+		event.ModuleFields.Put("cluster.name", info.ClusterName)
+		event.ModuleFields.Put("cluster.id", info.ClusterID)
+		event.MetricSetFields = common.MapStr{}
+
+		policyName, ok := policy["name"]
+		if !ok {
+			// No name found for policy. Ignore because all policies require a name
+			errs = append(errs, errors.New("found an 'executing policy' without a name. Omitting."))
+			continue
+		}
+
+		taskData, ok := policy["task"]
+		if !ok {
+			// No task found for policy. Ignore because all policies must contain a task
+			errs = append(errs, errors.New("found an 'executing policy' without a task. Omitting."))
+			continue
+		}
+
+		taskMapstr, ok := taskData.(map[string]interface{})
+		if !ok {
+			errs = append(errs, errors.New("error trying to convert interface of task data into a map"))
+			continue
+		}
+
+		fields, err := task.Apply(taskMapstr)
+		if err != nil {
+			errs = append(errs, errors.Wrap(err, "failure applying enrich coordinator stats schema"))
+			continue
+		}
+
+		event.MetricSetFields.Put("executing_policy.name", policyName)
+		event.MetricSetFields.Put("executing_policy.task", fields)
+
+		r.Event(event)
+	}
+
 	return errs.Err()
 }