[CCR] Update es monitoring mapping and (#33635)

* [CCR] Update es monitoring mapping and
change qa tests to query based on leader index.


Co-authored-by: Jason Tedor <jason@tedor.me>

x-pack/plugin/ccr/qa/multi-cluster-with-security/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexSecurityIT.java

@@ -29,6 +29,7 @@ import java.util.Map;
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken.basicAuthHeaderValue;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.is;
@@ -209,22 +210,19 @@ public class FollowIndexSecurityIT extends ESRestTestCase {
         ensureYellow(".monitoring-*");
 
         Request request = new Request("GET", "/.monitoring-*/_search");
-        request.setJsonEntity("{\"query\": {\"term\": {\"type\": \"ccr_stats\"}}}");
+        request.setJsonEntity("{\"query\": {\"term\": {\"ccr_stats.leader_index\": \"leader_cluster:" + expectedLeaderIndex + "\"}}}");
         Map<String, ?> response = toMap(adminClient().performRequest(request));
 
-        int numDocs = (int) XContentMapValues.extractValue("hits.total", response);
-        assertThat(numDocs, greaterThanOrEqualTo(1));
-
         int numberOfOperationsReceived = 0;
         int numberOfOperationsIndexed = 0;
 
         List<?> hits = (List<?>) XContentMapValues.extractValue("hits.hits", response);
-        for (int i = 0; i < numDocs; i++) {
+        assertThat(hits.size(), greaterThanOrEqualTo(1));
+
+        for (int i = 0; i < hits.size(); i++) {
             Map<?, ?> hit = (Map<?, ?>) hits.get(i);
             String leaderIndex = (String) XContentMapValues.extractValue("_source.ccr_stats.leader_index", hit);
-            if (leaderIndex.endsWith(expectedLeaderIndex) == false) {
+            assertThat(leaderIndex, endsWith(leaderIndex));
-                continue;
-            }
 
             int foundNumberOfOperationsReceived =
                 (int) XContentMapValues.extractValue("_source.ccr_stats.operations_received", hit);

x-pack/plugin/ccr/qa/multi-cluster/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexIT.java

@@ -24,6 +24,7 @@ import java.util.List;
 import java.util.Map;
 
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 
@@ -162,22 +163,19 @@ public class FollowIndexIT extends ESRestTestCase {
         ensureYellow(".monitoring-*");
 
         Request request = new Request("GET", "/.monitoring-*/_search");
-        request.setJsonEntity("{\"query\": {\"term\": {\"type\": \"ccr_stats\"}}}");
+        request.setJsonEntity("{\"query\": {\"term\": {\"ccr_stats.leader_index\": \"leader_cluster:" + expectedLeaderIndex + "\"}}}");
         Map<String, ?> response = toMap(client().performRequest(request));
 
-        int numDocs = (int) XContentMapValues.extractValue("hits.total", response);
-        assertThat(numDocs, greaterThanOrEqualTo(1));
-
         int numberOfOperationsReceived = 0;
         int numberOfOperationsIndexed = 0;
 
         List<?> hits = (List<?>) XContentMapValues.extractValue("hits.hits", response);
-        for (int i = 0; i < numDocs; i++) {
+        assertThat(hits.size(), greaterThanOrEqualTo(1));
+
+        for (int i = 0; i < hits.size(); i++) {
             Map<?, ?> hit = (Map<?, ?>) hits.get(i);
             String leaderIndex = (String) XContentMapValues.extractValue("_source.ccr_stats.leader_index", hit);
-            if (leaderIndex.endsWith(expectedLeaderIndex) == false) {
+            assertThat(leaderIndex, endsWith(leaderIndex));
-                continue;
-            }
 
             int foundNumberOfOperationsReceived =
                 (int) XContentMapValues.extractValue("_source.ccr_stats.operations_received", hit);

x-pack/plugin/core/src/main/resources/monitoring-es.json

@@ -916,6 +916,83 @@
               }
             }
           }
+        },
+        "ccr_stats": {
+          "properties": {
+            "leader_index": {
+              "type": "keyword"
+            },
+            "shard_id": {
+              "type": "integer"
+            },
+            "leader_global_checkpoint": {
+              "type": "long"
+            },
+            "leader_max_seq_no": {
+              "type": "long"
+            },
+            "follower_global_checkpoint": {
+              "type": "long"
+            },
+            "follower_max_seq_no": {
+              "type": "long"
+            },
+            "last_requested_seq_no": {
+              "type": "long"
+            },
+            "number_of_concurrent_reads": {
+              "type": "long"
+            },
+            "number_of_concurrent_writes": {
+              "type": "long"
+            },
+            "number_of_queued_writes": {
+              "type": "long"
+            },
+            "mapping_version": {
+              "type": "long"
+            },
+            "total_fetch_time_millis": {
+              "type": "long"
+            },
+            "number_of_successful_fetches": {
+              "type": "long"
+            },
+            "number_of_failed_fetches": {
+              "type": "long"
+            },
+            "operations_received": {
+              "type": "long"
+            },
+            "total_transferred_bytes": {
+              "type": "long"
+            },
+            "total_index_time_millis": {
+              "type": "long"
+            },
+            "number_of_successful_bulk_operations": {
+              "type": "long"
+            },
+            "number_of_failed_bulk_operations": {
+              "type": "long"
+            },
+            "number_of_operations_indexed": {
+              "type": "long"
+            },
+            "fetch_exceptions": {
+              "properties": {
+                "from_seq_no": {
+                  "type": "long"
+                },
+                "exception": {
+                  "type": "text"
+                }
+              }
+            },
+            "time_since_last_fetch_millis": {
+              "type": "long"
+            }
+          }
         }
       }
     }