Remove a data node from the cluster (#17248)

* service, rest & permissions for data node management * add tests * add reset data node event * add states and transitions for node removal add node removal functionality * changelog * add datanode permissions to permission list * add audit event types to list * add custom thread factory * add node removal integration test refactor rest calls in ITs * wait for primary shard node availability * refactor duplicate code * simplify node retrieval * add `@OnlyInSecuredNode` annotation and request filter move delete endpoint to management controller and secure it adjust IT change javadoc * stabilize tests
2026-03-13 09:32:21 +08:00 · 2023-11-20 13:33:26 +01:00
parent 51b4f4c226
commit c6101f395c
44 changed files with 1519 additions and 274 deletions
--- a/data-node/src/test/java/org/graylog/datanode/integration/DatanodeClusterIT.java
+++ b/data-node/src/test/java/org/graylog/datanode/integration/DatanodeClusterIT.java
@@ -19,13 +19,19 @@ package org.graylog.datanode.integration;
 import com.github.joschi.jadconfig.util.Duration;
 import com.github.rholder.retry.RetryException;
 import org.apache.commons.lang3.RandomStringUtils;
-import org.graylog.datanode.DatanodeOpensearchWait;
 import org.graylog.datanode.configuration.variants.KeystoreInformation;
+import org.graylog.datanode.restoperations.DatanodeOpensearchWait;
+import org.graylog.datanode.restoperations.DatanodeRestApiWait;
+import org.graylog.datanode.restoperations.DatanodeStatusChangeOperation;
+import org.graylog.datanode.restoperations.OpensearchTestIndexCreation;
+import org.graylog.datanode.restoperations.RestOperationParameters;
 import org.graylog.datanode.testinfra.DatanodeContainerizedBackend;
+import org.graylog.testing.completebackend.ContainerizedGraylogBackend;
 import org.graylog.testing.containermatrix.MongodbServer;
 import org.graylog.testing.mongodb.MongoDBTestService;
 import org.graylog2.security.IndexerJwtAuthTokenProvider;
 import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
@@ -39,10 +45,13 @@ import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.security.GeneralSecurityException;
 import java.security.KeyStore;
+import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Stream;

 import static org.graylog.datanode.testinfra.DatanodeContainerizedBackend.IMAGE_WORKING_DIR;
+import static org.junit.jupiter.api.Assertions.assertTrue;

 public class DatanodeClusterIT {
    private static final Logger LOG = LoggerFactory.getLogger(DatanodeClusterIT.class);
@@ -57,10 +66,13 @@ public class DatanodeClusterIT {
    private KeystoreInformation ca;
    private Network network;
    private MongoDBTestService mongoDBTestService;
+    private String restAdminUsername;

    @BeforeEach
    void setUp() throws GeneralSecurityException, IOException {

+        restAdminUsername = RandomStringUtils.randomAlphanumeric(10);
+
        // first generate a self-signed CA
        ca = DatanodeSecurityTestUtils.generateCa(tempDir);

@@ -100,8 +112,12 @@ public class DatanodeClusterIT {

    @AfterEach
    void tearDown() {
-        nodeB.stop();
-        nodeA.stop();
+        if (nodeB != null) {
+            nodeB.stop();
+        }
+        if (nodeA != null) {
+            nodeA.stop();
+        }
        mongoDBTestService.close();
        network.close();
    }
@@ -131,6 +147,67 @@ public class DatanodeClusterIT {
        waitForNodesCount(2);
    }

+    @Test
+    void testRemovingNodeReallocatesShards() throws ExecutionException, RetryException {
+
+        final String hostnameNodeC = "graylog-datanode-host-" + RandomStringUtils.random(8, "0123456789abcdef");
+        final KeystoreInformation transportNodeC = DatanodeSecurityTestUtils.generateTransportCert(tempDir, ca, hostnameNodeC);
+        final KeystoreInformation httpNodeC = DatanodeSecurityTestUtils.generateHttpCert(tempDir, ca, hostnameNodeC);
+
+        final DatanodeContainerizedBackend nodeC = createDatanodeContainer(
+                network, mongoDBTestService,
+                hostnameNodeC,
+                transportNodeC,
+                httpNodeC
+        );
+
+        nodeC.start();
+        waitForNodesCount(3);
+
+        String jwtToken = IndexerJwtAuthTokenProvider.createToken(DatanodeContainerizedBackend.SIGNING_SECRET.getBytes(StandardCharsets.UTF_8), Duration.seconds(600));
+        OpensearchTestIndexCreation osIndexClient = new OpensearchTestIndexCreation(RestOperationParameters.builder()
+                .port(nodeA.getOpensearchRestPort())
+                .truststore(trustStore)
+                .jwtToken(jwtToken)
+                .build());
+
+        // create index and get primary and replica shard node
+        osIndexClient.createIndex();
+        List<String> shardNodes = osIndexClient.getShardNodes();
+        Assertions.assertEquals(shardNodes.size(), 2);
+
+        List<DatanodeContainerizedBackend> nodes = List.of(nodeA, nodeB, nodeC);
+        final Optional<DatanodeContainerizedBackend> primary = nodes.stream().filter(n -> shardNodes.get(0).equals(n.getNodeName())).findFirst();
+        assertTrue(primary.isPresent());
+        final Optional<DatanodeContainerizedBackend> replica = nodes.stream().filter(n -> shardNodes.get(1).equals(n.getNodeName())).findFirst();
+        assertTrue(replica.isPresent());
+
+        // remove node for primary shard, waiting for it to be in AVAILABLE state first
+        final RestOperationParameters datanodeRestParameters = RestOperationParameters.builder()
+                .port(primary.get().getDatanodeRestPort())
+                .truststore(trustStore)
+                .username(restAdminUsername)
+                .password(ContainerizedGraylogBackend.ROOT_PASSWORD_PLAINTEXT)
+                .build();
+        new DatanodeRestApiWait(datanodeRestParameters)
+                .waitForAvailableStatus();
+        new DatanodeStatusChangeOperation(datanodeRestParameters)
+                .triggerNodeRemoval();
+
+
+        // check that primary shard node is gone and there are still a primary and a secondary
+        waitForNodesCount(replica.get(), 2);
+        jwtToken = IndexerJwtAuthTokenProvider.createToken(DatanodeContainerizedBackend.SIGNING_SECRET.getBytes(StandardCharsets.UTF_8), Duration.seconds(600));
+        osIndexClient = new OpensearchTestIndexCreation(RestOperationParameters.builder()
+                .port(replica.get().getOpensearchRestPort())
+                .truststore(trustStore)
+                .jwtToken(jwtToken)
+                .build());
+        List<String> newShardNodes = osIndexClient.getShardNodes();
+        Assertions.assertEquals(newShardNodes.size(), 2);
+        Assertions.assertFalse(newShardNodes.contains(primary.get().getNodeName()));
+    }
+
    @NotNull
    private DatanodeContainerizedBackend createDatanodeContainer(Network network,
                                                                 MongoDBTestService mongodb,
@@ -168,17 +245,28 @@ public class DatanodeClusterIT {
                    datanodeContainer.withCreateContainerCmdModifier(createContainerCmd -> createContainerCmd.withName(hostname));
                    datanodeContainer.withEnv("GRAYLOG_DATANODE_HOSTNAME", hostname);
                    datanodeContainer.withEnv("GRAYLOG_DATANODE_MONGODB_URI", mongodb.internalUri());
+
+                    datanodeContainer.withEnv("GRAYLOG_DATANODE_ROOT_USERNAME", restAdminUsername);
+
                });
    }

+
    private void waitForNodesCount(final int countOfNodes) throws ExecutionException, RetryException {
+        waitForNodesCount(nodeA, countOfNodes);
+    }
+
+    private void waitForNodesCount(DatanodeContainerizedBackend node, final int countOfNodes) throws ExecutionException, RetryException {
        final String jwtToken = IndexerJwtAuthTokenProvider.createToken(DatanodeContainerizedBackend.SIGNING_SECRET.getBytes(StandardCharsets.UTF_8), Duration.seconds(120));

        try {
-            DatanodeOpensearchWait.onPort(nodeA.getOpensearchRestPort())
-                    .withTruststore(trustStore)
-                    .withJwtAuth(jwtToken)
+            new DatanodeOpensearchWait(RestOperationParameters.builder()
+                    .port(node.getOpensearchRestPort())
+                    .truststore(trustStore)
+                    .jwtToken(jwtToken)
+                    .build())
                    .waitForNodesCount(countOfNodes);
+
        } catch (Exception retryException) {
            LOG.error("DataNode Container logs from nodeA follow:\n" + nodeA.getLogs());
            LOG.error("DataNode Container logs from node B follow:\n" + nodeB.getLogs());
--- a/data-node/src/test/java/org/graylog/datanode/integration/DatanodeSecuritySetupIT.java
+++ b/data-node/src/test/java/org/graylog/datanode/integration/DatanodeSecuritySetupIT.java
@@ -20,8 +20,9 @@ import com.github.joschi.jadconfig.util.Duration;
 import com.github.rholder.retry.RetryException;
 import io.restassured.response.ValidatableResponse;
 import org.apache.commons.lang3.RandomStringUtils;
-import org.graylog.datanode.DatanodeRestApiWait;
 import org.graylog.datanode.configuration.variants.KeystoreInformation;
+import org.graylog.datanode.restoperations.DatanodeRestApiWait;
+import org.graylog.datanode.restoperations.RestOperationParameters;
 import org.graylog.datanode.testinfra.DatanodeContainerizedBackend;
 import org.graylog2.plugin.Tools;
 import org.graylog2.security.IndexerJwtAuthTokenProvider;
@@ -125,9 +126,13 @@ public class DatanodeSecuritySetupIT {
    private ValidatableResponse waitForOpensearchAvailableStatus(final Integer datanodeRestPort, final KeyStore trustStore, String authUsername, String authPassword) throws ExecutionException, RetryException {

        try {
-            return DatanodeRestApiWait.onPort(datanodeRestPort)
-                    .withTruststore(trustStore)
-                    .withBasicAuth(authUsername, authPassword)
+            return new DatanodeRestApiWait(
+                    RestOperationParameters.builder()
+                            .port(datanodeRestPort)
+                            .truststore(trustStore)
+                            .username(authUsername)
+                            .password(authPassword)
+                            .build())
                    .waitForAvailableStatus();
        } catch (Exception ex) {
            LOG.error("Error starting the DataNode, showing logs:\n" + backend.getLogs());
--- a/data-node/src/test/java/org/graylog/datanode/integration/DatanodeStartupIT.java
+++ b/data-node/src/test/java/org/graylog/datanode/integration/DatanodeStartupIT.java
@@ -63,12 +63,21 @@ public class DatanodeStartupIT {
                    .assertThat()
                    .body("opensearch.node.node_name", Matchers.equalTo("node1"))
                    .body("opensearch.node.process.pid", Matchers.notNullValue());
+
+            assertInsecuredMethodSecurity(datanodeRestApiPort);
        } catch (RetryException retryException) {
            LOG.error("DataNode Container logs follow:\n" + backend.getLogs());
            throw retryException;
        }
    }

+    private void assertInsecuredMethodSecurity(Integer mappedPort) {
+        RestAssured.given()
+                .delete("http://localhost:" + mappedPort + "/management")
+                .then()
+                .statusCode(401);
+    }
+
    private ValidatableResponse getStatus(Integer mappedPort) {
        return RestAssured.given()
                .get("http://localhost:" + mappedPort)