Updating elasticsearch/ccr metricset with ES ccr_stats API changes

CHANGELOG.asciidoc

@@ -45,6 +45,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha1...master[Check the HEAD d
 
 - Fix issue preventing diskio metrics collection for idle disks. {issue}9124[9124] {pull}9125[9125]
 - Fix panic on docker healthcheck collection on dockers without healthchecks. {pull}9171[9171]
+- Fix issue with not collecting Elasticsearch cross-cluster replication stats correctly. {pull}9179[9179]
 
 *Packetbeat*
 

metricbeat/docs/fields.asciidoc

@@ -5440,7 +5440,7 @@ Number of the shard within the index
 
 --
 
-*`elasticsearch.ccr.follower.operations_indexed`*::
+*`elasticsearch.ccr.follower.operations_written`*::
 +
 --
 type: long
@@ -5450,7 +5450,7 @@ Number of operations indexed (replicated) into the follower shard from the leade
 
 --
 
-*`elasticsearch.ccr.follower.time_since_last_fetch.ms`*::
+*`elasticsearch.ccr.follower.time_since_last_read.ms`*::
 +
 --
 type: long

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

@@ -1,30 +1,30 @@
 {
     "@timestamp": "2017-10-12T08:05:34.853Z",
-    "beat": {
+    "agent": {
         "hostname": "host.example.com",
         "name": "host.example.com"
     },
     "elasticsearch": {
         "ccr": {
             "follower": {
-                "global_checkpoint": 1,
-                "index": "my_follower",
-                "operations_indexed": 2,
+                "global_checkpoint": -1,
+                "index": "my_index_f",
+                "operations_written": 0,
                 "shard": {
                     "number": 0
                 },
-                "time_since_last_fetch": {
-                    "ms": 4926
+                "time_since_last_read": {
+                    "ms": 42294
                 }
             },
             "leader": {
-                "index": "my_leader",
-                "max_seq_no": 1
+                "index": "my_index",
+                "max_seq_no": -1
             }
         },
         "cluster": {
-            "id": "KSGkOjOuSg6whAgtpPyhQw",
-            "name": "elasticsearch"
+            "id": "3LbUkLkURz--FR-YO0wLNA",
+            "name": "es1"
         }
     },
     "metricset": {

metricbeat/module/elasticsearch/ccr/_meta/fields.yml

@@ -26,11 +26,11 @@
           type: long
           description: >
             Number of the shard within the index
-        - name: operations_indexed
+        - name: operations_written
           type: long
           description: >
             Number of operations indexed (replicated) into the follower shard from the leader shard
-        - name: time_since_last_fetch.ms
+        - name: time_since_last_read.ms
           type: long
           description: >
             Time, in ms, since the follower last fetched from the leader

metricbeat/module/elasticsearch/ccr/_meta/test/ccr_stats.700.json

@@ -1,98 +1,46 @@
 {
-    "my_follower": [
-        {
-            "leader_index": "my_leader",
-            "follower_index": "my_follower",
-            "shard_id": 0,
-            "leader_global_checkpoint": 1,
-            "leader_max_seq_no": 1,
-            "follower_global_checkpoint": 1,
-            "follower_max_seq_no": 1,
-            "last_requested_seq_no": 1,
-            "number_of_concurrent_reads": 1,
-            "number_of_concurrent_writes": 0,
-            "number_of_queued_writes": 0,
-            "mapping_version": 2,
-            "total_fetch_time_millis": 21,
-            "number_of_successful_fetches": 142,
-            "number_of_failed_fetches": 6743,
-            "operations_received": 2,
-            "total_transferred_bytes": 166,
-            "total_index_time_millis": 48,
-            "number_of_successful_bulk_operations": 2,
-            "number_of_failed_bulk_operations": 0,
-            "number_of_operations_indexed": 2,
-            "fetch_exceptions": [
-                {
-                    "from_seq_no": 2,
-                    "retries": 6743,
-                    "exception": {
-                        "type": "exception",
-                        "reason": "NoShardAvailableActionException[No shard available for [Request{fromSeqNo=2, maxOperationCount=1024, shardId=[leader][0], expectedHistoryUUID=ki9Do9c-QQKVydB-6Txkdg, maxOperationSizeInBytes=9223372036854775807}]]; nested: RemoteTransportException[[d0XH9XU][127.0.0.1:9300][indices:data/read/xpack/ccr/shard_changes[s]]]; nested: IndexNotFoundException[no such index];",
-                        "caused_by": {
-                            "type": "no_shard_available_action_exception",
-                            "reason": "No shard available for [Request{fromSeqNo=2, maxOperationCount=1024, shardId=[leader][0], expectedHistoryUUID=ki9Do9c-QQKVydB-6Txkdg, maxOperationSizeInBytes=9223372036854775807}]",
-                            "caused_by": {
-                                "type": "index_not_found_exception",
-                                "reason": "no such index",
-                                "index_uuid": "RH8_j_w0Q0GGmpY0HMVQ_A",
-                                "index": "my_leader"
-                            }
-                        }
+    "auto_follow_stats": {
+        "number_of_failed_follow_indices": 0,
+        "number_of_failed_remote_cluster_state_requests": 0,
+        "number_of_successful_follow_indices": 1,
+        "recent_auto_follow_errors": []
+    },
+    "follow_stats": {
+        "indices": [
+            {
+                "index": "follower_index",
+                "shards": [
+                    {
+                        "remote_cluster": "remote_cluster",
+                        "leader_index": "leader_index",
+                        "follower_index": "follower_index",
+                        "shard_id": 0,
+                        "leader_global_checkpoint": 1024,
+                        "leader_max_seq_no": 1536,
+                        "follower_global_checkpoint": 768,
+                        "follower_max_seq_no": 896,
+                        "last_requested_seq_no": 897,
+                        "outstanding_read_requests": 8,
+                        "outstanding_write_requests": 2,
+                        "write_buffer_operation_count": 64,
+                        "follower_mapping_version": 4,
+                        "follower_settings_version": 2,
+                        "total_read_time_millis": 32768,
+                        "total_read_remote_exec_time_millis": 16384,
+                        "successful_read_requests": 32,
+                        "failed_read_requests": 0,
+                        "operations_read": 896,
+                        "bytes_read": 32768,
+                        "total_write_time_millis": 16384,
+                        "write_buffer_size_in_bytes": 1536,
+                        "successful_write_requests": 16,
+                        "failed_write_requests": 0,
+                        "operations_written": 832,
+                        "read_exceptions": [],
+                        "time_since_last_read_millis": 8
                     }
-                }
-            ],
-            "time_since_last_fetch_millis": 470
-        },
-        {
-            "leader_index": "my_leader",
-            "follower_index": "my_follower",
-            "shard_id": 1,
-            "leader_global_checkpoint": -1,
-            "leader_max_seq_no": -1,
-            "follower_global_checkpoint": -1,
-            "follower_max_seq_no": -1,
-            "last_requested_seq_no": -1,
-            "number_of_concurrent_reads": 0,
-            "number_of_concurrent_writes": 0,
-            "number_of_queued_writes": 0,
-            "mapping_version": 2,
-            "total_fetch_time_millis": 0,
-            "number_of_successful_fetches": 336,
-            "number_of_failed_fetches": 0,
-            "operations_received": 0,
-            "total_transferred_bytes": 0,
-            "total_index_time_millis": 0,
-            "number_of_successful_bulk_operations": 0,
-            "number_of_failed_bulk_operations": 0,
-            "number_of_operations_indexed": 0,
-            "fetch_exceptions": [],
-            "time_since_last_fetch_millis": 4323
-        },
-        {
-            "leader_index": "my_leader",
-            "follower_index": "my_follower",
-            "shard_id": 2,
-            "leader_global_checkpoint": 1,
-            "leader_max_seq_no": 1,
-            "follower_global_checkpoint": 1,
-            "follower_max_seq_no": 1,
-            "last_requested_seq_no": 1,
-            "number_of_concurrent_reads": 0,
-            "number_of_concurrent_writes": 0,
-            "number_of_queued_writes": 0,
-            "mapping_version": 2,
-            "total_fetch_time_millis": 0,
-            "number_of_successful_fetches": 372,
-            "number_of_failed_fetches": 0,
-            "operations_received": 2,
-            "total_transferred_bytes": 166,
-            "total_index_time_millis": 32,
-            "number_of_successful_bulk_operations": 2,
-            "number_of_failed_bulk_operations": 0,
-            "number_of_operations_indexed": 2,
-            "fetch_exceptions": [],
-            "time_since_last_fetch_millis": 4323
-        }
-    ]
+                ]
+            }
+        ]
+    }
 }

metricbeat/module/elasticsearch/ccr/data.go

@@ -19,7 +19,6 @@ package ccr
 
 import (
 	"encoding/json"
-	"fmt"
 
 	"github.com/joeshaw/multierror"
 	"github.com/pkg/errors"
@@ -42,17 +41,25 @@ var (
 			"shard": s.Object{
 				"number": c.Int("shard_id"),
 			},
-			"operations_indexed": c.Int("number_of_operations_indexed"),
-			"time_since_last_fetch": s.Object{
-				"ms": c.Int("time_since_last_fetch_millis"),
+			"operations_written": c.Int("operations_written"),
+			"time_since_last_read": s.Object{
+				"ms": c.Int("time_since_last_read_millis"),
 			},
 			"global_checkpoint": c.Int("follower_global_checkpoint"),
 		},
 	}
 )
 
+type response struct {
+	FollowStats struct {
+		Indices []struct {
+			Shards []map[string]interface{} `json:"shards"`
+		} `json:"indices"`
+	} `json:"follow_stats"`
+}
+
 func eventsMapping(r mb.ReporterV2, info elasticsearch.Info, content []byte) error {
-	var data map[string]interface{}
+	var data response
 	err := json.Unmarshal(content, &data)
 	if err != nil {
 		err = errors.Wrap(err, "failure parsing Elasticsearch CCR Stats API response")
@@ -61,17 +68,8 @@ func eventsMapping(r mb.ReporterV2, info elasticsearch.Info, content []byte) err
 	}
 
 	var errs multierror.Errors
-	for _, followerShards := range data {
-
-		shards, ok := followerShards.([]interface{})
-		if !ok {
-			err := fmt.Errorf("shards is not an array")
-			errs = append(errs, err)
-			r.Error(err)
-			continue
-		}
-
-		for _, s := range shards {
+	for _, followerIndex := range data.FollowStats.Indices {
+		for _, followerShard := range followerIndex.Shards {
 			event := mb.Event{}
 			event.RootFields = common.MapStr{}
 			event.RootFields.Put("service.name", elasticsearch.ModuleName)
@@ -80,15 +78,7 @@ func eventsMapping(r mb.ReporterV2, info elasticsearch.Info, content []byte) err
 			event.ModuleFields.Put("cluster.name", info.ClusterName)
 			event.ModuleFields.Put("cluster.id", info.ClusterID)
 
-			shard, ok := s.(map[string]interface{})
-			if !ok {
-				event.Error = fmt.Errorf("shard is not an object")
-				r.Event(event)
-				errs = append(errs, event.Error)
-				continue
-			}
-
-			event.MetricSetFields, err = schema.Apply(shard)
+			event.MetricSetFields, err = schema.Apply(followerShard)
 			if err != nil {
 				event.Error = errors.Wrap(err, "failure applying shard schema")
 				r.Event(event)

metricbeat/module/elasticsearch/ccr/data_xpack.go

@@ -19,7 +19,6 @@ package ccr
 
 import (
 	"encoding/json"
-	"fmt"
 	"time"
 
 	"github.com/joeshaw/multierror"
@@ -32,36 +31,24 @@ import (
 )
 
 func eventsMappingXPack(r mb.ReporterV2, m *MetricSet, info elasticsearch.Info, content []byte) error {
-	var data map[string]interface{}
+	var data response
 	err := json.Unmarshal(content, &data)
 	if err != nil {
-		return errors.Wrap(err, "failure parsing Elasticsearch CCR Stats API response")
+		err = errors.Wrap(err, "failure parsing Elasticsearch CCR Stats API response")
+		r.Error(err)
+		return err
 	}
 
 	var errors multierror.Errors
-	for _, followerShards := range data {
-
-		shards, ok := followerShards.([]interface{})
-		if !ok {
-			err := fmt.Errorf("shards is not an array")
-			errors = append(errors, err)
-			continue
-		}
-
-		for _, s := range shards {
-			shard, ok := s.(map[string]interface{})
-			if !ok {
-				err := fmt.Errorf("shard is not an object")
-				errors = append(errors, err)
-				continue
-			}
+	for _, followerIndex := range data.FollowStats.Indices {
+		for _, followerShard := range followerIndex.Shards {
 			event := mb.Event{}
 			event.RootFields = common.MapStr{
 				"cluster_uuid": info.ClusterID,
 				"timestamp":    common.Time(time.Now()),
 				"interval_ms":  m.Module().Config().Period / time.Millisecond,
 				"type":         "ccr_stats",
-				"ccr_stats":    shard,
+				"ccr_stats":    followerShard,
 			}
 
 			event.Index = elastic.MakeXPackMonitoringIndexName(elastic.Elasticsearch)