[Monitoring] Add cluster metadata to cluster_stats docs (#33860) (#34023

)

Backport of #33860 and #34040.

This PR teaches Monitoring to collect cluster metadata, if any is set, and index it into `cluster_stats` docs in `.monitoring-es-*`.

After this PR, `cluster_stats` docs in `.monitoring-es-*` will contain an additional top-level `cluster_settings` field like so:

```
{
   ...
   "cluster_settings": {
     "cluster": {
       "metadata": {
         ...
       }
     }
   }
}
```

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/collector/cluster/ClusterStatsCollector.java

@@ -113,7 +113,8 @@ protected Collection<MonitoringDoc> doCollect(final MonitoringDoc.Node node,
         // Adds a cluster stats document
         return Collections.singleton(
                 new ClusterStatsMonitoringDoc(clusterUuid, timestamp(), interval, node, clusterName, version,  clusterStats.getStatus(),
-                                              license, apmIndicesExist, xpackUsage, clusterStats, clusterState, clusterNeedsTLSEnabled));
+                                              license, apmIndicesExist, xpackUsage, clusterStats, clusterState,
+                                              clusterNeedsTLSEnabled));
     }
 
     boolean doAPMIndicesExist(final ClusterState clusterState) {

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/collector/cluster/ClusterStatsMonitoringDoc.java

@@ -8,6 +8,7 @@
 import org.elasticsearch.action.admin.cluster.stats.ClusterStatsResponse;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.Nullable;
@@ -45,6 +46,7 @@ public class ClusterStatsMonitoringDoc extends MonitoringDoc {
                                 ClusterState.Metric.NODES));
 
     public static final String TYPE = "cluster_stats";
+    protected static final String SETTING_DISPLAY_NAME = "cluster.metadata.display_name";
 
     private final String clusterName;
     private final String version;
@@ -118,6 +120,14 @@ boolean getClusterNeedsTLSEnabled() {
         return clusterNeedsTLSEnabled;
     }
 
+    String getClusterDisplayName() {
+        MetaData metaData = this.clusterState.getMetaData();
+        if (metaData == null) {
+            return null;
+        }
+        return metaData.settings().get(SETTING_DISPLAY_NAME);
+    }
+
     @Override
     protected void innerToXContent(XContentBuilder builder, Params params) throws IOException {
         builder.field("cluster_name", clusterName);
@@ -156,6 +166,23 @@ protected void innerToXContent(XContentBuilder builder, Params params) throws IO
             builder.endObject();
         }
 
+        String displayName = getClusterDisplayName();
+        if (displayName != null) {
+            builder.startObject("cluster_settings");
+            {
+                    builder.startObject("cluster");
+                    {
+                        builder.startObject("metadata");
+                        {
+                            builder.field("display_name", displayName);
+                        }
+                        builder.endObject();
+                    }
+                    builder.endObject();
+            }
+            builder.endObject();
+        }
+
         builder.startObject("stack_stats");
         {
             // in the future, it may be useful to pass in an object that represents APM (and others), but for now this

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/collector/cluster/ClusterStatsMonitoringDocTests.java

@@ -203,7 +203,12 @@ public void testToXContent() throws IOException {
                                                                 Version.V_6_0_0_beta1);
 
         final ClusterState clusterState = ClusterState.builder(clusterName)
-                                                        .metaData(MetaData.builder().clusterUUID(clusterUuid).build())
+                                                        .metaData(MetaData.builder()
+                                                            .clusterUUID(clusterUuid)
+                                                            .transientSettings(Settings.builder()
+                                                                .put("cluster.metadata.display_name", "my_prod_cluster")
+                                                                .build())
+                                                            .build())
                                                         .stateUUID("_state_uuid")
                                                         .version(12L)
                                                         .nodes(DiscoveryNodes.builder()
@@ -521,6 +526,13 @@ public void testToXContent() throws IOException {
                       + "}"
                     + "}"
                   + "},"
+                  + "\"cluster_settings\":{"
+                    + "\"cluster\":{"
+                      + "\"metadata\":{"
+                        + "\"display_name\":\"my_prod_cluster\""
+                      + "}"
+                    + "}"
+                  + "},"
                   + "\"stack_stats\":{"
                     + "\"apm\":{"
                       + "\"found\":" + apmIndicesExist

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/integration/MonitoringIT.java

@@ -207,6 +207,12 @@ public void testMonitoringService() throws Exception {
                            .status(),
                    is(RestStatus.CREATED));
 
+        final Settings settings = Settings.builder()
+            .put("cluster.metadata.display_name", "my cluster")
+            .build();
+
+        assertAcked(client().admin().cluster().prepareUpdateSettings().setTransientSettings(settings));
+
         whenExportersAreReady(() -> {
             final AtomicReference<SearchResponse> searchResponse = new AtomicReference<>();
 
@@ -316,7 +322,7 @@ private void assertMonitoringDocSourceNode(final Map<String, Object> sourceNode)
     private void assertClusterStatsMonitoringDoc(final Map<String, Object> document,
                                                  final boolean apmIndicesExist) {
         final Map<String, Object> source = (Map<String, Object>) document.get("_source");
-        assertEquals(11, source.size());
+        assertEquals(12, source.size());
 
         assertThat((String) source.get("cluster_name"), not(isEmptyOrNullString()));
         assertThat(source.get("version"), equalTo(Version.CURRENT.toString()));
@@ -377,6 +383,12 @@ private void assertClusterStatsMonitoringDoc(final Map<String, Object> document,
         assertThat(clusterState.remove("master_node"), notNullValue());
         assertThat(clusterState.remove("nodes"), notNullValue());
         assertThat(clusterState.keySet(), empty());
+
+
+        final Map<String, Object> clusterSettings = (Map<String, Object>) source.get("cluster_settings");
+        assertThat(clusterSettings, notNullValue());
+        assertThat(clusterSettings.remove("cluster"), notNullValue());
+        assertThat(clusterSettings.keySet(), empty());
     }
 
     /**
@@ -617,6 +629,7 @@ public void disableMonitoring() throws Exception {
         final Settings settings = Settings.builder()
                 .putNull("xpack.monitoring.collection.enabled")
                 .putNull("xpack.monitoring.exporters._local.enabled")
+                .putNull("cluster.metadata.display_name")
                 .build();
 
         assertAcked(client().admin().cluster().prepareUpdateSettings().setTransientSettings(settings));