Add guard for matching graylog/datanode version in migration

changelog/unreleased/pr-19828.toml

@@ -0,0 +1,4 @@
+type = "a"
+message = "Add guard for matching graylog/datanode version in data node migration."
+
+pulls = ["19828"]

graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/state/actions/MigrationActions.java

@@ -41,6 +41,8 @@ public interface MigrationActions {
     boolean renewalPolicyDoesNotExist();
     boolean caAndRenewalPolicyExist();
 
+    boolean compatibleDatanodesRunning();
+
     void provisionDataNodes();
 
     void provisionAndStartDataNodes();

graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/state/actions/MigrationActionsImpl.java

@@ -37,6 +37,7 @@
 import org.graylog2.indexer.datanode.RemoteReindexRequest;
 import org.graylog2.indexer.datanode.RemoteReindexingMigrationAdapter;
 import org.graylog2.plugin.GlobalMetricNames;
+import org.graylog2.plugin.Version;
 import org.graylog2.plugin.certificates.RenewalPolicy;
 import org.graylog2.plugin.cluster.ClusterConfigService;
 import org.graylog2.rest.resources.datanodes.DatanodeRestApiProxy;
@@ -77,6 +78,8 @@ public class MigrationActionsImpl implements MigrationActions {
     private final List<URI> elasticsearchHosts;
     private final ObjectMapper objectMapper;
 
+    private final Version graylogVersion = Version.CURRENT_CLASSPATH;
+
     @Inject
     public MigrationActionsImpl(final ClusterConfigService clusterConfigService, NodeService<DataNodeDto> nodeService,
                                 final CaKeystore caKeystore, DataNodeCommandService dataNodeCommandService,
@@ -177,6 +180,14 @@ public boolean caAndRenewalPolicyExist() {
         return !caDoesNotExist() && !renewalPolicyDoesNotExist();
     }
 
+    @Override
+    public boolean compatibleDatanodesRunning() {
+        Map<String, DataNodeDto> nodes = nodeService.allActive();
+        return !nodes.isEmpty() && nodes.values().stream()
+                .allMatch(node -> node.getDatanodeVersion() != null &&
+                        graylogVersion.compareTo(new Version(com.github.zafarkhaja.semver.Version.valueOf(node.getDatanodeVersion()))) == 0);
+    }
+
     @Override
     public void provisionDataNodes() {
         // if we start provisioning DataNodes via the migration, Preflight is definitely done/no option anymore

graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/state/machine/MigrationStateMachineBuilder.java

@@ -67,7 +67,7 @@ private static StateMachineConfig<MigrationState, MigrationStep> configureStates
         // remote reindexing branch of the migration
         config.configure(MigrationState.REMOTE_REINDEX_WELCOME_PAGE)
                 .onEntry(migrationActions::reindexUpgradeSelected)
-                .permitIf(MigrationStep.PROVISION_DATANODE_CERTIFICATES, MigrationState.PROVISION_DATANODE_CERTIFICATES_RUNNING, () -> !migrationActions.dataNodeStartupFinished(), migrationActions::provisionAndStartDataNodes)
+                .permitIf(MigrationStep.PROVISION_DATANODE_CERTIFICATES, MigrationState.PROVISION_DATANODE_CERTIFICATES_RUNNING, () -> !migrationActions.dataNodeStartupFinished() && migrationActions.compatibleDatanodesRunning(), migrationActions::provisionAndStartDataNodes)
                 .permitIf(MigrationStep.SHOW_DATA_MIGRATION_QUESTION, MigrationState.EXISTING_DATA_MIGRATION_QUESTION_PAGE, migrationActions::dataNodeStartupFinished);
 
         // This page should contain the "Please restart Graylog to continue with data migration"
@@ -98,7 +98,7 @@ private static StateMachineConfig<MigrationState, MigrationStep> configureStates
         // in place / rolling upgrade branch of the migration
         config.configure(MigrationState.ROLLING_UPGRADE_MIGRATION_WELCOME_PAGE)
                 .onEntry(migrationActions::rollingUpgradeSelected)
-                .permit(MigrationStep.RUN_DIRECTORY_COMPATIBILITY_CHECK, MigrationState.DIRECTORY_COMPATIBILITY_CHECK_PAGE, migrationActions::runDirectoryCompatibilityCheck);
+                .permitIf(MigrationStep.RUN_DIRECTORY_COMPATIBILITY_CHECK, MigrationState.DIRECTORY_COMPATIBILITY_CHECK_PAGE, migrationActions::compatibleDatanodesRunning, migrationActions::runDirectoryCompatibilityCheck);
 
         config.configure(MigrationState.DIRECTORY_COMPATIBILITY_CHECK_PAGE)
                 .permitReentryIf(MigrationStep.RUN_DIRECTORY_COMPATIBILITY_CHECK, () -> !migrationActions.directoryCompatibilityCheckOk(), migrationActions::runDirectoryCompatibilityCheck)

graylog2-server/src/test/java/org/graylog/plugins/views/storage/migration/state/machine/MigrationActionsAdapter.java

@@ -124,6 +124,11 @@ public boolean caAndRenewalPolicyExist() {
         return false;
     }
 
+    @Override
+    public boolean compatibleDatanodesRunning() {
+        return false;
+    }
+
     @Override
     public void provisionDataNodes() {
     }

graylog2-server/src/test/java/org/graylog/plugins/views/storage/migration/state/machine/MigrationStateMachineBuilderTest.java

@@ -146,6 +146,7 @@ public void testMigrationSelectionPage() {
     @Test
     public void testRemoteReindexWelcomePage() {
         StateMachine<MigrationState, MigrationStep> stateMachine = getStateMachine(MigrationState.MIGRATION_SELECTION_PAGE);
+        when(migrationActions.compatibleDatanodesRunning()).thenReturn(true);
         stateMachine.fire(MigrationStep.SELECT_REMOTE_REINDEX_MIGRATION);
         assertThat(stateMachine.getState()).isEqualTo(MigrationState.REMOTE_REINDEX_WELCOME_PAGE);
         verify(migrationActions).reindexUpgradeSelected();
@@ -161,6 +162,7 @@ public void testRemoteReindexWelcomePage() {
     @Test
     public void testProvisionDatanodeCertificatesRunning() {
         StateMachine<MigrationState, MigrationStep> stateMachine = getStateMachine(MigrationState.REMOTE_REINDEX_WELCOME_PAGE);
+        when(migrationActions.compatibleDatanodesRunning()).thenReturn(true);
         stateMachine.fire(MigrationStep.PROVISION_DATANODE_CERTIFICATES);
         verify(migrationActions, times(1)).dataNodeStartupFinished();
         verify(migrationActions, times(1)).provisionAndStartDataNodes();
@@ -202,6 +204,7 @@ public void testMigrateExistingData() {
     @Test
     public void testRollingUpgradeMigrationWelcomePage() {
         StateMachine<MigrationState, MigrationStep> stateMachine = getStateMachine(MigrationState.MIGRATION_SELECTION_PAGE);
+        when(migrationActions.compatibleDatanodesRunning()).thenReturn(true);
         when(migrationActions.isCompatibleInPlaceMigrationVersion()).thenReturn(true);
         stateMachine.fire(MigrationStep.SELECT_ROLLING_UPGRADE_MIGRATION);
         assertThat(stateMachine.getState()).isEqualTo(MigrationState.ROLLING_UPGRADE_MIGRATION_WELCOME_PAGE);
@@ -213,6 +216,7 @@ public void testRollingUpgradeMigrationWelcomePage() {
     @Test
     public void testDirectoryCompatibilityCheckPage() {
         StateMachine<MigrationState, MigrationStep> stateMachine = getStateMachine(MigrationState.ROLLING_UPGRADE_MIGRATION_WELCOME_PAGE);
+        when(migrationActions.compatibleDatanodesRunning()).thenReturn(true);
         stateMachine.fire(MigrationStep.RUN_DIRECTORY_COMPATIBILITY_CHECK);
         assertThat(stateMachine.getState()).isEqualTo(MigrationState.DIRECTORY_COMPATIBILITY_CHECK_PAGE);
         assertThat(stateMachine.getPermittedTriggers()).containsOnly(MigrationStep.RUN_DIRECTORY_COMPATIBILITY_CHECK);