Always extract monitoring.cluster_uuid setting from Beat config (#17420)

* Pass monitoring.* config object

* Adding CHANGELOG entry

* Adding system test

CHANGELOG.next.asciidoc

@@ -72,6 +72,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix k8s metadata issue regarding node labels not shown up on root level of metadata. {pull}16834[16834]
 - Fail to start if httpprof is used and it cannot be initialized. {pull}17028[17028]
 - Fix concurrency issues in convert processor when used in the global context. {pull}17032[17032]
+- Fix bug with `monitoring.cluster_uuid` setting not always being exposed via GET /state Beats API. {issue}16732[16732] {pull}17420[17420]
 
 *Auditbeat*
 

libbeat/cmd/instance/beat.go

@@ -874,7 +874,7 @@ func (b *Beat) setupMonitoring(settings Settings) (report.Reporter, error) {
 		return nil, err
 	}
 
-	monitoringClusterUUID, err := monitoring.GetClusterUUID(monitoringCfg)
+	monitoringClusterUUID, err := monitoring.GetClusterUUID(b.Config.MonitoringBeatConfig.Monitoring)
 	if err != nil {
 		return nil, err
 	}

libbeat/tests/system/test_monitoring.py

@@ -177,6 +177,30 @@ def test_cluster_uuid_setting(self):
 
         self.assertEqual(test_cluster_uuid, state["monitoring"]["cluster_uuid"])
 
+    @unittest.skipUnless(INTEGRATION_TESTS, "integration test")
+    @attr('integration')
+    def test_cluster_uuid_setting_monitoring_disabled(self):
+        """
+        Test that monitoring.cluster_uuid setting may be set with monitoring.enabled explicitly set to false
+        """
+        test_cluster_uuid = self.random_string(10)
+        self.render_config_template(
+            "mockbeat",
+            monitoring={
+                "enabled": False,
+                "cluster_uuid": test_cluster_uuid
+            },
+            http_enabled="true"
+        )
+
+        proc = self.start_beat(config="mockbeat.yml")
+        self.wait_until(lambda: self.log_contains("mockbeat start running."))
+
+        state = self.get_beat_state()
+        proc.check_kill_and_wait()
+
+        self.assertEqual(test_cluster_uuid, state["monitoring"]["cluster_uuid"])
+
     def search_monitoring_doc(self, monitoring_type):
         results = self.es_monitoring.search(
             index='.monitoring-beats-*',