Ensure type exists for all monitoring configuration (elastic#57399)

elastic#47711 and elastic#47246 helped to validate that monitoring settings are
rejected at time of setting the monitoring settings. Else an invalid
monitoring setting can find it's way into the cluster state and result
in an exception thrown [1] on the cluster state application (there by
causing significant issues). Some additional monitoring settings have
been identified that can result in invalid cluster state that also
result in exceptions thrown on cluster state application.

All settings require a type of either http or local to be
applicable. When a setting is changed, the exporters are automatically
updated with the new settings. However, if the old or new settings lack
of a type setting an exception will be thrown (since exporters are
always of type 'http' or 'local'). Arguably we shouldn't blindly create
and destroy new exporters on each monitoring setting update, but the
lifecycle of the exporters is abit out the scope this PR is trying to
address.

This commit introduces a similar methodology to check for validity as
elastic#47711 and elastic#47246 but this time for ALL (including non-http) settings.
Monitoring settings are not useful unless there an exporter with a type
defined. The type is used as dependent setting, such that it must
exist to set the value. This ensures that when any monitoring settings
changes that they can only get added to cluster state if the type
exists. If the type exists (and the other validations pass) then the
exporters will get re-built and the cluster state remains valid.

Tests have been included to ensure that all dynamic monitoring settings
have the type as dependent settings.

[1]
org.elasticsearch.common.settings.SettingsException: missing exporter type for [found-user-defined] exporter
at org.elasticsearch.xpack.monitoring.exporter.Exporters.initExporters(Exporters.java:126) ~[?:?]

server/src/main/java/org/elasticsearch/common/settings/Setting.java

@@ -750,6 +750,15 @@ private Stream<String> matchStream(Settings settings) {
             return settings.keySet().stream().filter(this::match).map(key::getConcreteString);
         }
 
+        /**
+         * Get the raw list of dependencies. This method is exposed for testing purposes and {@link #getSettingsDependencies(String)}
+         * should be preferred for most all cases.
+         * @return the raw list of dependencies for this setting
+         */
+        public Set<AffixSettingDependency> getDependencies() {
+            return Collections.unmodifiableSet(dependencies);
+        }
+
         @Override
         public Set<SettingDependency> getSettingsDependencies(String settingsKey) {
             if (dependencies.isEmpty()) {

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

@@ -596,6 +596,7 @@ public void enableMonitoring() throws Exception {
 
         final Settings settings = Settings.builder()
                 .put("xpack.monitoring.collection.enabled", true)
+                .put("xpack.monitoring.exporters._local.type", "local")
                 .put("xpack.monitoring.exporters._local.enabled", true)
                 .build();
 
@@ -622,6 +623,7 @@ public void enableMonitoring() throws Exception {
     public void disableMonitoring() throws Exception {
         final Settings settings = Settings.builder()
                 .putNull("xpack.monitoring.collection.enabled")
+                .putNull("xpack.monitoring.exporters._local.type")
                 .putNull("xpack.monitoring.exporters._local.enabled")
                 .putNull("cluster.metadata.display_name")
                 .build();

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

@@ -25,9 +25,11 @@
 
 public abstract class Exporter implements AutoCloseable {
 
+    public static Setting.AffixSettingDependency TYPE_DEPENDENCY = () -> Exporter.TYPE_SETTING;
+
     private static final Setting.AffixSetting<Boolean> ENABLED_SETTING =
             Setting.affixKeySetting("xpack.monitoring.exporters.","enabled",
-                    key -> Setting.boolSetting(key, true, Property.Dynamic, Property.NodeScope));
+                    key -> Setting.boolSetting(key, true, Property.Dynamic, Property.NodeScope), TYPE_DEPENDENCY);
 
     public static final Setting.AffixSetting<String> TYPE_SETTING = Setting.affixKeySetting(
         "xpack.monitoring.exporters.",
@@ -83,21 +85,22 @@ public Iterator<Setting<?>> settings() {
      */
     public static final Setting.AffixSetting<Boolean> USE_INGEST_PIPELINE_SETTING =
             Setting.affixKeySetting("xpack.monitoring.exporters.","use_ingest",
-                    key -> Setting.boolSetting(key, true, Property.Dynamic, Property.NodeScope));
+                    key -> Setting.boolSetting(key, true, Property.Dynamic, Property.NodeScope), TYPE_DEPENDENCY);
     /**
      * Every {@code Exporter} allows users to explicitly disable cluster alerts.
      */
     public static final Setting.AffixSetting<Boolean> CLUSTER_ALERTS_MANAGEMENT_SETTING =
             Setting.affixKeySetting("xpack.monitoring.exporters.", "cluster_alerts.management.enabled",
-                    key -> Setting.boolSetting(key, true, Property.Dynamic, Property.NodeScope));
+                    key -> Setting.boolSetting(key, true, Property.Dynamic, Property.NodeScope), TYPE_DEPENDENCY);
     /**
      * Every {@code Exporter} allows users to explicitly disable specific cluster alerts.
      * <p>
      * When cluster alerts management is enabled, this should delete anything blacklisted here in addition to not creating it.
      */
     public static final Setting.AffixSetting<List<String>> CLUSTER_ALERTS_BLACKLIST_SETTING = Setting
                 .affixKeySetting("xpack.monitoring.exporters.", "cluster_alerts.management.blacklist",
-                    key -> Setting.listSetting(key, Collections.emptyList(), Function.identity(), Property.Dynamic, Property.NodeScope));
+                    key -> Setting.listSetting(key, Collections.emptyList(), Function.identity(), Property.Dynamic, Property.NodeScope),
+                    TYPE_DEPENDENCY);
 
     /**
      * Every {@code Exporter} allows users to use a different index time format.
@@ -109,7 +112,7 @@ public Iterator<Setting<?>> settings() {
                                 Exporter.INDEX_FORMAT,
                                 DateFormatter::forPattern,
                                 Property.Dynamic,
-                                Property.NodeScope));
+                                Property.NodeScope), TYPE_DEPENDENCY);
 
     private static final String INDEX_FORMAT = "yyyy.MM.dd";
 

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/exporter/http/HttpExporter.java

@@ -81,7 +81,7 @@ public class HttpExporter extends Exporter {
 
     public static final String TYPE = "http";
 
-    private static Setting.AffixSettingDependency TYPE_DEPENDENCY = new Setting.AffixSettingDependency() {
+    private static Setting.AffixSettingDependency HTTP_TYPE_DEPENDENCY = new Setting.AffixSettingDependency() {
         @Override
         public Setting.AffixSetting<String> getSetting() {
             return Exporter.TYPE_SETTING;
@@ -168,30 +168,32 @@ public Iterator<Setting<?>> settings() {
                     },
                     Property.Dynamic,
                     Property.NodeScope),
-                TYPE_DEPENDENCY);
+                HTTP_TYPE_DEPENDENCY);
 
     /**
      * Master timeout associated with bulk requests.
      */
     public static final Setting.AffixSetting<TimeValue> BULK_TIMEOUT_SETTING =
             Setting.affixKeySetting("xpack.monitoring.exporters.","bulk.timeout",
-                    (key) -> Setting.timeSetting(key, TimeValue.MINUS_ONE, Property.Dynamic, Property.NodeScope), TYPE_DEPENDENCY);
+                    (key) -> Setting.timeSetting(key, TimeValue.MINUS_ONE, Property.Dynamic, Property.NodeScope), HTTP_TYPE_DEPENDENCY);
     /**
      * Timeout used for initiating a connection.
      */
     public static final Setting.AffixSetting<TimeValue> CONNECTION_TIMEOUT_SETTING =
             Setting.affixKeySetting(
                 "xpack.monitoring.exporters.",
                 "connection.timeout",
-                (key) -> Setting.timeSetting(key, TimeValue.timeValueSeconds(6), Property.Dynamic, Property.NodeScope), TYPE_DEPENDENCY);
+                (key) -> Setting.timeSetting(key, TimeValue.timeValueSeconds(6), Property.Dynamic, Property.NodeScope),
+                HTTP_TYPE_DEPENDENCY);
     /**
      * Timeout used for reading from the connection.
      */
     public static final Setting.AffixSetting<TimeValue> CONNECTION_READ_TIMEOUT_SETTING =
             Setting.affixKeySetting(
                 "xpack.monitoring.exporters.",
                 "connection.read_timeout",
-                (key) -> Setting.timeSetting(key, TimeValue.timeValueSeconds(60), Property.Dynamic, Property.NodeScope), TYPE_DEPENDENCY);
+                (key) -> Setting.timeSetting(key, TimeValue.timeValueSeconds(60), Property.Dynamic, Property.NodeScope),
+                HTTP_TYPE_DEPENDENCY);
     /**
      * Username for basic auth.
      */
@@ -242,7 +244,7 @@ public Iterator<Setting<?>> settings() {
                         Property.Dynamic,
                         Property.NodeScope,
                         Property.Filtered),
-                    TYPE_DEPENDENCY);
+                HTTP_TYPE_DEPENDENCY);
     /**
      * Password for basic auth.
      */
@@ -297,7 +299,7 @@ public Iterator<Setting<?>> settings() {
             "xpack.monitoring.exporters.",
             "auth.secure_password",
             key -> SecureSetting.secureString(key, null),
-            TYPE_DEPENDENCY);
+            HTTP_TYPE_DEPENDENCY);
     /**
      * The SSL settings.
      *
@@ -308,7 +310,7 @@ public Iterator<Setting<?>> settings() {
                 "xpack.monitoring.exporters.",
                 "ssl",
                 (key) -> Setting.groupSetting(key + ".", Property.Dynamic, Property.NodeScope, Property.Filtered),
-                TYPE_DEPENDENCY);
+                HTTP_TYPE_DEPENDENCY);
 
     /**
      * Proxy setting to allow users to send requests to a remote cluster that requires a proxy base path.
@@ -329,13 +331,13 @@ public Iterator<Setting<?>> settings() {
                         },
                         Property.Dynamic,
                         Property.NodeScope),
-                    TYPE_DEPENDENCY);
+                HTTP_TYPE_DEPENDENCY);
     /**
      * A boolean setting to enable or disable sniffing for extra connections.
      */
     public static final Setting.AffixSetting<Boolean> SNIFF_ENABLED_SETTING =
             Setting.affixKeySetting("xpack.monitoring.exporters.","sniff.enabled",
-                    (key) -> Setting.boolSetting(key, false, Property.Dynamic, Property.NodeScope), TYPE_DEPENDENCY);
+                    (key) -> Setting.boolSetting(key, false, Property.Dynamic, Property.NodeScope), HTTP_TYPE_DEPENDENCY);
     /**
      * A parent setting to header key/value pairs, whose names are user defined.
      */
@@ -358,7 +360,7 @@ public Iterator<Setting<?>> settings() {
                         },
                         Property.Dynamic,
                         Property.NodeScope),
-                    TYPE_DEPENDENCY);
+                HTTP_TYPE_DEPENDENCY);
     /**
      * Blacklist of headers that the user is not allowed to set.
      * <p>
@@ -370,19 +372,19 @@ public Iterator<Setting<?>> settings() {
      */
     public static final Setting.AffixSetting<TimeValue> TEMPLATE_CHECK_TIMEOUT_SETTING =
             Setting.affixKeySetting("xpack.monitoring.exporters.","index.template.master_timeout",
-                    (key) -> Setting.timeSetting(key, TimeValue.MINUS_ONE, Property.Dynamic, Property.NodeScope), TYPE_DEPENDENCY);
+                    (key) -> Setting.timeSetting(key, TimeValue.MINUS_ONE, Property.Dynamic, Property.NodeScope), HTTP_TYPE_DEPENDENCY);
     /**
      * A boolean setting to enable or disable whether to create placeholders for the old templates.
      */
     public static final Setting.AffixSetting<Boolean> TEMPLATE_CREATE_LEGACY_VERSIONS_SETTING =
             Setting.affixKeySetting("xpack.monitoring.exporters.","index.template.create_legacy_templates",
-                    (key) -> Setting.boolSetting(key, true, Property.Dynamic, Property.NodeScope), TYPE_DEPENDENCY);
+                    (key) -> Setting.boolSetting(key, true, Property.Dynamic, Property.NodeScope), HTTP_TYPE_DEPENDENCY);
     /**
      * ES level timeout used when checking and writing pipelines (used to speed up tests)
      */
     public static final Setting.AffixSetting<TimeValue> PIPELINE_CHECK_TIMEOUT_SETTING =
             Setting.affixKeySetting("xpack.monitoring.exporters.","index.pipeline.master_timeout",
-                    (key) -> Setting.timeSetting(key, TimeValue.MINUS_ONE, Property.Dynamic, Property.NodeScope), TYPE_DEPENDENCY);
+                    (key) -> Setting.timeSetting(key, TimeValue.MINUS_ONE, Property.Dynamic, Property.NodeScope), HTTP_TYPE_DEPENDENCY);
 
     /**
      * Minimum supported version of the remote monitoring cluster (same major).

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/exporter/local/LocalExporter.java

@@ -96,7 +96,7 @@ public class LocalExporter extends Exporter implements ClusterStateListener, Cle
     public static final Setting.AffixSetting<TimeValue> WAIT_MASTER_TIMEOUT_SETTING = Setting.affixKeySetting(
         "xpack.monitoring.exporters.",
         "wait_master.timeout",
-        (key) -> Setting.timeSetting(key, TimeValue.timeValueSeconds(30), Property.Dynamic, Property.NodeScope)
+        (key) -> Setting.timeSetting(key, TimeValue.timeValueSeconds(30), Property.Dynamic, Property.NodeScope), TYPE_DEPENDENCY
     );
 
     private final Client client;

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/exporter/ExportersTests.java

@@ -49,7 +49,9 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 
+import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
@@ -337,6 +339,49 @@ public void testConcurrentExports() throws Exception {
         exporters.close();
     }
 
+    /**
+     * All dynamic monitoring settings should have dependency on type.
+     */
+    public void testSettingsDependency() {
+        List<Setting.AffixSetting<?>> settings = Exporters.getSettings().stream().filter(Setting::isDynamic).collect(Collectors.toList());
+        settings.stream().filter(s -> s.getKey().equals("xpack.monitoring.exporters.*.type") == false)
+            .forEach(setting -> assertThat(setting.getKey() + " does not have a dependency on type",
+                setting.getDependencies().stream().map(Setting.AffixSettingDependency::getSetting).distinct().collect(Collectors.toList()),
+                contains(Exporter.TYPE_SETTING)));
+    }
+
+    /**
+     * This is a variation of testing that all settings validated at cluster state update ensure that the type is set. This workflow
+     * emulates adding valid settings, then attempting to remove the type. This should never be allowed since type if type is null
+     * then any associated settings are extraneous and thus invalid (and can cause validation issues on cluster state application).
+     */
+    public void testRemoveType() {
+        //run the update for all dynamic settings and ensure that they correctly throw an exception
+        List<Setting.AffixSetting<?>> settings = Exporters.getSettings().stream().filter(Setting::isDynamic).collect(Collectors.toList());
+        settings.stream().filter(s -> s.getKey().equals("xpack.monitoring.exporters.*.type") == false)
+            .forEach(setting -> {
+                String fullSettingName = setting.getKey().replace("*", "foobar");
+                Settings nodeSettings = Settings.builder()
+                    .put("xpack.monitoring.exporters.foobar.type", randomFrom("local, http")) //actual type should not matter
+                    .put(fullSettingName, "")
+                    .build();
+
+                clusterSettings = new ClusterSettings(nodeSettings, new HashSet<>(Exporters.getSettings()));
+                when(clusterService.getClusterSettings()).thenReturn(clusterSettings);
+
+                Settings update = Settings.builder()
+                    .put("xpack.monitoring.exporters.foobar.type", (String) null)
+                    .build();
+
+                Settings.Builder target = Settings.builder().put(nodeSettings);
+                clusterSettings.updateDynamicSettings(update, target, Settings.builder(), "persistent");
+                IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
+                    () -> clusterSettings.validate(target.build(), true));
+                assertThat(e.getMessage(),
+                    containsString("missing required setting [xpack.monitoring.exporters.foobar.type] for setting [" + fullSettingName));
+            });
+    }
+
     /**
      * Attempt to export a random number of documents via {@code exporters} from multiple threads.
      *

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/exporter/local/LocalExporterIntegTests.java

@@ -66,6 +66,7 @@ private void stopMonitoring() {
         // Now disabling the monitoring service, so that no more collection are started
         assertAcked(client().admin().cluster().prepareUpdateSettings().setTransientSettings(
                 Settings.builder().putNull(MonitoringService.ENABLED.getKey())
+                                  .putNull("xpack.monitoring.exporters._local.type")
                                   .putNull("xpack.monitoring.exporters._local.enabled")
                                   .putNull("xpack.monitoring.exporters._local.cluster_alerts.management.enabled")
                                   .putNull("xpack.monitoring.exporters._local.index.name.time_format")));
@@ -86,6 +87,7 @@ public void testExport() throws Exception {
             // start the monitoring service so that /_monitoring/bulk is not ignored
             final Settings.Builder exporterSettings = Settings.builder()
                     .put(MonitoringService.ENABLED.getKey(), true)
+                    .put("xpack.monitoring.exporters._local.type", LocalExporter.TYPE)
                     .put("xpack.monitoring.exporters._local.enabled", true)
                     .put("xpack.monitoring.exporters._local.cluster_alerts.management.enabled", false);
 

x-pack/plugin/src/test/java/org/elasticsearch/xpack/test/rest/XPackRestIT.java

@@ -123,6 +123,7 @@ private void enableMonitoring() throws Exception {
             final Map<String, Object> settings = new HashMap<>();
             settings.put("xpack.monitoring.collection.enabled", true);
             settings.put("xpack.monitoring.collection.interval", "1s");
+            settings.put("xpack.monitoring.exporters._local.type", "local");
             settings.put("xpack.monitoring.exporters._local.enabled", true);
 
             awaitCallApi("cluster.put_settings", emptyMap(),