Implement xpack.monitoring.elasticsearch.collection.enabled setting (#33474)

* Implement xpack.monitoring.elasticsearch.collection.enabled setting

* Fixing line lengths

* Updating constructor calls in test

* Removing unused import

* Fixing line lengths in test classes

* Make monitoringService.isElasticsearchCollectionEnabled() return true for tests

* Remove wrong expectation

* Adding unit tests for new flag to be false

* Fixing line wrapping/indentation for better readability

* Adding docs

* Fixing logic in ClusterStatsCollector::shouldCollect

* Rebasing with master and resolving conflicts

* Simplifying implementation by gating scheduling

* Doc fixes / improvements

* Making methods package private

* Fixing wording

* Fixing method access

docs/reference/monitoring/configuring-monitoring.asciidoc

@@ -13,10 +13,13 @@ indices. You can also adjust how monitoring data is displayed.
 
 . To collect monitoring data about your {es} cluster:
 
-.. Verify that the `xpack.monitoring.enabled` and 
+.. Verify that the `xpack.monitoring.enabled`, 
-`xpack.monitoring.collection.enabled` settings are `true` on each node in the 
+`xpack.monitoring.collection.enabled`, and 
-cluster. By default, data collection is disabled. For more information, see 
+`xpack.monitoring.elasticsearch.collection.enabled` settings are `true` on each
-<<monitoring-settings>>.
+node in the  cluster. By default xpack.monitoring.collection.enabled is disabled
+(`false`), and that overrides xpack.monitoring.elasticsearch.collection.enabled,
+which defaults to being enabled (`true`). Both settings can be set dynamically
+at runtime. For more information, see <<monitoring-settings>>.
 
 .. Optional: Specify which indices you want to monitor. 
 +

docs/reference/monitoring/pause-export.asciidoc

@@ -16,6 +16,16 @@ monitoring data from other sources such as {kib}, Beats, and Logstash is ignored
 You can update this setting by using the 
 {ref}/cluster-update-settings.html[Cluster Update Settings API].
 
+If you want to collect data from sources such as {kib}, Beats, and Logstash but
+not collect data about your {es} cluster, you can disable data collection
+just for {es}:
+
+[source,yaml]
+---------------------------------------------------
+xpack.monitoring.collection.enabled: true
+xpack.monitoring.elasticsearch.collection.enabled: false
+---------------------------------------------------
+
 If you want to separately disable a specific exporter, you can specify the 
 `enabled` setting (which defaults to `true`) per exporter. For example:
 

docs/reference/settings/monitoring-settings.asciidoc

@@ -66,6 +66,17 @@ option in `kibana.yml` to the same value.
 You can update this setting through the
 <<cluster-update-settings,Cluster Update Settings API>>.
 
+`xpack.monitoring.elasticsearch.collection.enabled`::
+
+Controls whether statistics about your {es} cluster should be collected. Defaults to `true`.
+This is different from xpack.monitoring.collection.enabled, which allows you to enable or disable
+all monitoring collection. However, this setting simply disables the collection of Elasticsearch
+data while still allowing other data (e.g., Kibana, Logstash, Beats, or APM Server monitoring data)
+to pass through this cluster.
++
+You can update this setting through the
+<<cluster-update-settings,Cluster Update Settings API>>.
+
 `xpack.monitoring.collection.cluster.stats.timeout`::
 
 Sets the timeout for collecting the cluster statistics. Defaults to `10s`.

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/Monitoring.java

@@ -174,6 +174,7 @@ public class Monitoring extends Plugin implements ActionPlugin {
         settings.add(MonitoringField.HISTORY_DURATION);
         settings.add(CLEAN_WATCHER_HISTORY);
         settings.add(MonitoringService.ENABLED);
+        settings.add(MonitoringService.ELASTICSEARCH_COLLECTION_ENABLED);
         settings.add(MonitoringService.INTERVAL);
         settings.add(Collector.INDICES);
         settings.add(ClusterStatsCollector.CLUSTER_STATS_TIMEOUT);

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/MonitoringService.java

@@ -43,8 +43,21 @@ public class MonitoringService extends AbstractLifecycleComponent {
      */
     public static final TimeValue MIN_INTERVAL = TimeValue.timeValueSeconds(1L);
 
+    /*
+     * Dynamically controls enabling or disabling the collection of Monitoring data only from Elasticsearch.
+     * <p>
+      * This should only be used while transitioning to Metricbeat-based data collection for Elasticsearch with
+      * {@linkplain #ENABLED} set to {@code true}. By setting this to {@code false} and that value to {@code true},
+      * Kibana, Logstash, Beats, and APM Server can all continue to report their stats through this cluster until they
+      * are transitioned to being monitored by Metricbeat as well.
+      */
+    public static final Setting<Boolean> ELASTICSEARCH_COLLECTION_ENABLED =
+            Setting.boolSetting("xpack.monitoring.elasticsearch.collection.enabled", true,
+                                Setting.Property.Dynamic, Setting.Property.NodeScope);
+
     /**
-     * Dynamically controls enabling or disabling the collection of Monitoring data.
+     * Dynamically controls enabling or disabling the collection of Monitoring data from Elasticsearch as well as other products
+     * in the stack.
      */
     public static final Setting<Boolean> ENABLED =
             Setting.boolSetting("xpack.monitoring.collection.enabled", false,
@@ -68,6 +81,7 @@ public class MonitoringService extends AbstractLifecycleComponent {
     private final Set<Collector> collectors;
     private final Exporters exporters;
 
+    private volatile boolean elasticsearchCollectionEnabled;
     private volatile boolean enabled;
     private volatile TimeValue interval;
     private volatile ThreadPool.Cancellable scheduler;
@@ -79,13 +93,21 @@ public class MonitoringService extends AbstractLifecycleComponent {
         this.threadPool = Objects.requireNonNull(threadPool);
         this.collectors = Objects.requireNonNull(collectors);
         this.exporters = Objects.requireNonNull(exporters);
+        this.elasticsearchCollectionEnabled = ELASTICSEARCH_COLLECTION_ENABLED.get(settings);
         this.enabled = ENABLED.get(settings);
         this.interval = INTERVAL.get(settings);
 
+        clusterService.getClusterSettings()
+            .addSettingsUpdateConsumer(ELASTICSEARCH_COLLECTION_ENABLED, this::setElasticsearchCollectionEnabled);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(ENABLED, this::setMonitoringActive);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(INTERVAL, this::setInterval);
     }
 
+    void setElasticsearchCollectionEnabled(final boolean enabled) {
+        this.elasticsearchCollectionEnabled = enabled;
+        scheduleExecution();
+    }
+
     void setMonitoringActive(final boolean enabled) {
         this.enabled = enabled;
         scheduleExecution();
@@ -104,6 +126,14 @@ public class MonitoringService extends AbstractLifecycleComponent {
         return isStarted() && enabled;
     }
 
+    boolean isElasticsearchCollectionEnabled() {
+        return this.elasticsearchCollectionEnabled;
+    }
+
+    boolean shouldScheduleExecution() {
+        return isElasticsearchCollectionEnabled() && isMonitoringActive();
+    }
+
     private String threadPoolName() {
         return ThreadPool.Names.GENERIC;
     }
@@ -155,7 +185,7 @@ public class MonitoringService extends AbstractLifecycleComponent {
         if (scheduler != null) {
             cancelExecution();
         }
-        if (isMonitoringActive()) {
+        if (shouldScheduleExecution()) {
             scheduler = threadPool.scheduleWithFixedDelay(monitor, interval, threadPoolName());
         }
     }
@@ -188,7 +218,7 @@ public class MonitoringService extends AbstractLifecycleComponent {
 
         @Override
         public void doRun() {
-            if (isMonitoringActive() == false) {
+            if (shouldScheduleExecution() == false) {
                 logger.debug("monitoring execution is skipped");
                 return;
             }
@@ -223,7 +253,7 @@ public class MonitoringService extends AbstractLifecycleComponent {
                                     new ParameterizedMessage("monitoring collector [{}] failed to collect data", collector.name()), e);
                         }
                     }
-                    if (isMonitoringActive()) {
+                    if (shouldScheduleExecution()) {
                         exporters.export(results, ActionListener.wrap(r -> semaphore.release(), this::onFailure));
                     } else {
                         semaphore.release();