Only allow x-pack metadata if all nodes are ready

x-pack/plugin/core/src/main/java/org/elasticsearch/license/LicenseService.java

@@ -223,6 +223,7 @@ protected PutLicenseResponse newResponse(boolean acknowledged) {
 
                             @Override
                             public ClusterState execute(ClusterState currentState) throws Exception {
+                                XPackPlugin.checkReadyForXPackCustomMetadata(currentState);
                                 MetaData currentMetadata = currentState.metaData();
                                 LicensesMetaData licensesMetaData = currentMetadata.custom(LicensesMetaData.TYPE);
                                 Version trialVersion = null;
@@ -341,7 +342,7 @@ protected void doStart() throws ElasticsearchException {
         if (clusterService.lifecycleState() == Lifecycle.State.STARTED) {
             final ClusterState clusterState = clusterService.state();
             if (clusterState.blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK) == false &&
-                    clusterState.nodes().getMasterNode() != null) {
+                    clusterState.nodes().getMasterNode() != null && XPackPlugin.isReadyForXPackCustomMetadata(clusterState)) {
                 final LicensesMetaData currentMetaData = clusterState.metaData().custom(LicensesMetaData.TYPE);
                 boolean noLicense = currentMetaData == null || currentMetaData.getLicense() == null;
                 if (clusterState.getNodes().isLocalNodeElectedMaster() &&
@@ -374,6 +375,12 @@ public void clusterChanged(ClusterChangedEvent event) {
         final ClusterState previousClusterState = event.previousState();
         final ClusterState currentClusterState = event.state();
         if (!currentClusterState.blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK)) {
+            if (XPackPlugin.isReadyForXPackCustomMetadata(currentClusterState) == false) {
+                logger.debug("cannot add license to cluster as the following nodes might not understand the license metadata: {}",
+                    () -> XPackPlugin.nodesNotReadyForXPackCustomMetadata(currentClusterState));
+                return;
+            }
+
             final LicensesMetaData prevLicensesMetaData = previousClusterState.getMetaData().custom(LicensesMetaData.TYPE);
             final LicensesMetaData currentLicensesMetaData = currentClusterState.getMetaData().custom(LicensesMetaData.TYPE);
             if (logger.isDebugEnabled()) {

x-pack/plugin/core/src/main/java/org/elasticsearch/license/StartBasicClusterTask.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.xpack.core.XPackPlugin;
 
 import java.time.Clock;
 import java.util.Collections;
@@ -59,6 +60,7 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS
 
     @Override
     public ClusterState execute(ClusterState currentState) throws Exception {
+        XPackPlugin.checkReadyForXPackCustomMetadata(currentState);
         LicensesMetaData licensesMetaData = currentState.metaData().custom(LicensesMetaData.TYPE);
         License currentLicense = LicensesMetaData.extractLicense(licensesMetaData);
         if (currentLicense == null || currentLicense.type().equals("basic") == false) {

x-pack/plugin/core/src/main/java/org/elasticsearch/license/StartTrialClusterTask.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.xpack.core.XPackPlugin;
 
 import java.time.Clock;
 import java.util.Collections;
@@ -64,6 +65,7 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS
 
     @Override
     public ClusterState execute(ClusterState currentState) throws Exception {
+        XPackPlugin.checkReadyForXPackCustomMetadata(currentState);
         LicensesMetaData currentLicensesMetaData = currentState.metaData().custom(LicensesMetaData.TYPE);
 
         if (request.isAcknowledged() == false) {

x-pack/plugin/core/src/main/java/org/elasticsearch/license/StartupSelfGeneratedLicenseTask.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.xpack.core.XPackPlugin;
 
 import java.time.Clock;
 import java.util.UUID;
@@ -49,6 +50,7 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS
 
     @Override
     public ClusterState execute(ClusterState currentState) throws Exception {
+        XPackPlugin.checkReadyForXPackCustomMetadata(currentState);
         final MetaData metaData = currentState.metaData();
         final LicensesMetaData currentLicensesMetaData = metaData.custom(LicensesMetaData.TYPE);
         // do not generate a license if any license is present

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackPlugin.java

@@ -9,15 +9,20 @@
 import org.apache.lucene.util.SetOnce;
 import org.bouncycastle.operator.OperatorCreationException;
 import org.elasticsearch.SpecialPermission;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.GenericAction;
 import org.elasticsearch.action.support.ActionFilter;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.client.transport.TransportClient;
+import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.inject.Binder;
 import org.elasticsearch.common.inject.Module;
 import org.elasticsearch.common.inject.multibindings.Multibinder;
@@ -33,6 +38,7 @@
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.license.LicenseService;
+import org.elasticsearch.license.LicensesMetaData;
 import org.elasticsearch.license.Licensing;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.plugins.ExtensiblePlugin;
@@ -46,10 +52,13 @@
 import org.elasticsearch.xpack.core.action.TransportXPackUsageAction;
 import org.elasticsearch.xpack.core.action.XPackInfoAction;
 import org.elasticsearch.xpack.core.action.XPackUsageAction;
+import org.elasticsearch.xpack.core.ml.MLMetadataField;
 import org.elasticsearch.xpack.core.rest.action.RestXPackInfoAction;
 import org.elasticsearch.xpack.core.rest.action.RestXPackUsageAction;
+import org.elasticsearch.xpack.core.security.authc.TokenMetaData;
 import org.elasticsearch.xpack.core.ssl.SSLConfigurationReloader;
 import org.elasticsearch.xpack.core.ssl.SSLService;
+import org.elasticsearch.xpack.core.watcher.WatcherMetaData;
 
 import javax.security.auth.DestroyFailedException;
 
@@ -62,14 +71,19 @@
 import java.time.Clock;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 
 public class XPackPlugin extends XPackClientPlugin implements ScriptPlugin, ExtensiblePlugin {
 
     private static Logger logger = ESLoggerFactory.getLogger(XPackPlugin.class);
     private static DeprecationLogger deprecationLogger = new DeprecationLogger(logger);
 
+    public static final String XPACK_INSTALLED_NODE_ATTR = "xpack.installed";
+
     // TODO: clean up this library to not ask for write access to all system properties!
     static {
         // invoke this clinit in unbound with permissions to access all system properties
@@ -138,6 +152,78 @@ protected Clock getClock() {
     public static LicenseService getSharedLicenseService() { return licenseService.get(); }
     public static XPackLicenseState getSharedLicenseState() { return licenseState.get(); }
 
+    /**
+     * Checks if the cluster state allows this node to add x-pack metadata to the cluster state,
+     * and throws an exception otherwise.
+     * This check should be called before installing any x-pack metadata to the cluster state,
+     * to ensure that the other nodes that are part of the cluster will be able to deserialize
+     * that metadata.
+     * Having this check properly in place everywhere allows to install x-pack into a cluster
+     * using a rolling restart.
+     */
+    public static void checkReadyForXPackCustomMetadata(ClusterState clusterState) {
+        List<DiscoveryNode> notReadyNodes = nodesNotReadyForXPackCustomMetadata(clusterState);
+        if (notReadyNodes.isEmpty() == false) {
+            throw new IllegalStateException("The following nodes are not ready yet for enabling x-pack custom metadata: " + notReadyNodes);
+        }
+    }
+
+    /**
+     * Checks if the cluster state allows this node to add x-pack metadata to the cluster state.
+     * See {@link #checkReadyForXPackCustomMetadata} for more details.
+     */
+    public static boolean isReadyForXPackCustomMetadata(ClusterState clusterState) {
+        return nodesNotReadyForXPackCustomMetadata(clusterState).isEmpty();
+    }
+
+    /**
+     * Returns the list of nodes that won't allow this node from adding x-pack metadata to the cluster state.
+     * See {@link #checkReadyForXPackCustomMetadata} for more details.
+     */
+    public static List<DiscoveryNode> nodesNotReadyForXPackCustomMetadata(ClusterState clusterState) {
+        // check if there's already x-pack metadata in the cluster state; if so, any further metadata won't hurt
+        final MetaData metaData = clusterState.metaData();
+        if (metaData.custom(LicensesMetaData.TYPE) != null ||
+            metaData.custom(MLMetadataField.TYPE) != null ||
+            metaData.custom(WatcherMetaData.TYPE) != null ||
+            clusterState.custom(TokenMetaData.TYPE) != null) {
+            return Collections.emptyList();
+        }
+
+        // if there's no x-pack metadata yet in the cluster state, check that all nodes would be capable
+        // of deserializing newly added x-pack metadata
+        final List<DiscoveryNode> notReadyNodes = StreamSupport.stream(clusterState.nodes().spliterator(), false).filter(node -> {
+            final String xpackInstalledAttr = node.getAttributes().getOrDefault(XPACK_INSTALLED_NODE_ATTR, "false");
+
+            // The node attribute XPACK_INSTALLED_NODE_ATTR was only introduced in 6.3.0, so when
+            // we have an older node in this mixed-version cluster without any x-pack metadata,
+            // we want to prevent x-pack from adding custom metadata
+            return node.getVersion().before(Version.V_6_3_0) || Booleans.parseBoolean(xpackInstalledAttr) == false;
+        }).collect(Collectors.toList());
+
+        return notReadyNodes;
+    }
+
+    @Override
+    public Settings additionalSettings() {
+        final String xpackInstalledNodeAttrSetting = "node.attr." + XPACK_INSTALLED_NODE_ATTR;
+
+        if (transportClientMode) {
+            if (settings.get(xpackInstalledNodeAttrSetting) != null) {
+                throw new IllegalArgumentException("Directly setting [" + xpackInstalledNodeAttrSetting + "] is not permitted");
+            }
+
+            return super.additionalSettings();
+        } else {
+            if (settings.get(xpackInstalledNodeAttrSetting) != null &&
+                settings.get(xpackInstalledNodeAttrSetting).equals("true") == false) {
+                throw new IllegalArgumentException("Conflicting setting [" + xpackInstalledNodeAttrSetting + "]");
+            }
+
+            return Settings.builder().put(super.additionalSettings()).put(xpackInstalledNodeAttrSetting, "true").build();
+        }
+    }
+
     @Override
     public Collection<Module> createGuiceModules() {
         ArrayList<Module> modules = new ArrayList<>();

x-pack/plugin/core/src/test/java/org/elasticsearch/license/AbstractLicenseServiceTestCase.java

@@ -18,14 +18,16 @@
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.watcher.ResourceWatcherService;
+import org.elasticsearch.xpack.core.XPackPlugin;
 import org.elasticsearch.xpack.core.watcher.watch.ClockMock;
 import org.junit.After;
 import org.junit.Before;
 
 import java.nio.file.Path;
+import java.util.Arrays;
 
-import static java.util.Collections.emptyMap;
 import static java.util.Collections.emptySet;
+import static java.util.Collections.singletonMap;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -66,6 +68,7 @@ protected void setInitialState(License license, XPackLicenseState licenseState,
         when(state.metaData()).thenReturn(metaData);
         final DiscoveryNode mockNode = getLocalNode();
         when(discoveryNodes.getMasterNode()).thenReturn(mockNode);
+        when(discoveryNodes.spliterator()).thenReturn(Arrays.asList(mockNode).spliterator());
         when(discoveryNodes.isLocalNodeElectedMaster()).thenReturn(false);
         when(state.nodes()).thenReturn(discoveryNodes);
         when(state.getNodes()).thenReturn(discoveryNodes); // it is really ridiculous we have nodes() and getNodes()...
@@ -76,7 +79,8 @@ protected void setInitialState(License license, XPackLicenseState licenseState,
     }
 
     protected DiscoveryNode getLocalNode() {
-        return new DiscoveryNode("b", buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT);
+        return new DiscoveryNode("b", buildNewFakeTransportAddress(), singletonMap(XPackPlugin.XPACK_INSTALLED_NODE_ATTR, "true"),
+            emptySet(), Version.CURRENT);
     }
 
     @After

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/XPackPluginTests.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.core;
+
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.ssl.SSLService;
+
+import static org.hamcrest.Matchers.containsString;
+
+public class XPackPluginTests extends ESTestCase {
+
+    public void testXPackInstalledAttrClashOnTransport() throws Exception {
+        Settings.Builder builder = Settings.builder();
+        builder.put("node.attr." + XPackPlugin.XPACK_INSTALLED_NODE_ATTR, "true");
+        builder.put(Client.CLIENT_TYPE_SETTING_S.getKey(), "transport");
+        XPackPlugin xpackPlugin = createXPackPlugin(builder.put("path.home", createTempDir()).build());
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, xpackPlugin::additionalSettings);
+        assertThat(e.getMessage(),
+            containsString("Directly setting [node.attr." + XPackPlugin.XPACK_INSTALLED_NODE_ATTR + "] is not permitted"));
+    }
+
+    public void testXPackInstalledAttrClashOnNode() throws Exception {
+        Settings.Builder builder = Settings.builder();
+        builder.put("node.attr." + XPackPlugin.XPACK_INSTALLED_NODE_ATTR, "false");
+        XPackPlugin xpackPlugin = createXPackPlugin(builder.put("path.home", createTempDir()).build());
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, xpackPlugin::additionalSettings);
+        assertThat(e.getMessage(),
+            containsString("Conflicting setting [node.attr." + XPackPlugin.XPACK_INSTALLED_NODE_ATTR + "]"));
+    }
+
+    public void testXPackInstalledAttrExists() throws Exception {
+        XPackPlugin xpackPlugin = createXPackPlugin(Settings.builder().put("path.home", createTempDir()).build());
+        assertEquals("true", xpackPlugin.additionalSettings().get("node.attr." + XPackPlugin.XPACK_INSTALLED_NODE_ATTR));
+    }
+
+    private XPackPlugin createXPackPlugin(Settings settings) throws Exception {
+        return new XPackPlugin(settings, null){
+
+            @Override
+            protected void setSslService(SSLService sslService) {
+                // disable
+            }
+
+            @Override
+            protected void setLicenseState(XPackLicenseState licenseState) {
+                // disable
+            }
+        };
+    }
+
+}

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MlInitializationService.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.gateway.GatewayService;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.xpack.core.XPackPlugin;
 import org.elasticsearch.xpack.core.ml.MLMetadataField;
 import org.elasticsearch.xpack.core.ml.MlMetadata;
 
@@ -48,6 +49,11 @@ public void clusterChanged(ClusterChangedEvent event) {
         }
 
         if (event.localNodeMaster()) {
+            if (XPackPlugin.isReadyForXPackCustomMetadata(event.state()) == false) {
+                logger.debug("cannot add ML metadata to cluster as the following nodes might not understand the ML metadata: {}",
+                    () -> XPackPlugin.nodesNotReadyForXPackCustomMetadata(event.state()));
+                return;
+            }
             MetaData metaData = event.state().metaData();
             installMlMetadata(metaData);
             installDailyMaintenanceService();
@@ -63,6 +69,7 @@ private void installMlMetadata(MetaData metaData) {
                     clusterService.submitStateUpdateTask("install-ml-metadata", new ClusterStateUpdateTask() {
                         @Override
                         public ClusterState execute(ClusterState currentState) throws Exception {
+                            XPackPlugin.checkReadyForXPackCustomMetadata(currentState);
                             // If the metadata has been added already don't try to update
                             if (currentState.metaData().custom(MLMetadataField.TYPE) != null) {
                                 return currentState;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteDatafeedAction.java

@@ -21,6 +21,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.core.XPackPlugin;
 import org.elasticsearch.xpack.core.ml.MLMetadataField;
 import org.elasticsearch.xpack.core.ml.MlMetadata;
 import org.elasticsearch.xpack.core.ml.action.DeleteDatafeedAction;
@@ -120,6 +121,7 @@ protected DeleteDatafeedAction.Response newResponse(boolean acknowledged) {
 
                     @Override
                     public ClusterState execute(ClusterState currentState) throws Exception {
+                        XPackPlugin.checkReadyForXPackCustomMetadata(currentState);
                         MlMetadata currentMetadata = currentState.getMetaData().custom(MLMetadataField.TYPE);
                         PersistentTasksCustomMetaData persistentTasks =
                                 currentState.getMetaData().custom(PersistentTasksCustomMetaData.TYPE);

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportFinalizeJobExecutionAction.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.core.XPackPlugin;
 import org.elasticsearch.xpack.core.ml.action.FinalizeJobExecutionAction;
 import org.elasticsearch.xpack.core.ml.MLMetadataField;
 import org.elasticsearch.xpack.core.ml.MlMetadata;
@@ -57,6 +58,7 @@ protected void masterOperation(FinalizeJobExecutionAction.Request request, Clust
         clusterService.submitStateUpdateTask(source, new ClusterStateUpdateTask() {
             @Override
             public ClusterState execute(ClusterState currentState) throws Exception {
+                XPackPlugin.checkReadyForXPackCustomMetadata(currentState);
                 MlMetadata mlMetadata = currentState.metaData().custom(MLMetadataField.TYPE);
                 MlMetadata.Builder mlMetadataBuilder = new MlMetadata.Builder(mlMetadata);
                 Date finishedTime = new Date();