diff --git a/changelog/unreleased/pr-17248.toml b/changelog/unreleased/pr-17248.toml new file mode 100644 index 0000000000..f1eba925c4 --- /dev/null +++ b/changelog/unreleased/pr-17248.toml @@ -0,0 +1,5 @@ +type = "a" +message = "data node life cycle management: removal of nodes" + +issues = ["16820"] +pulls = ["17248"] diff --git a/data-node/pom.xml b/data-node/pom.xml index 20453df4c0..8d13f46fde 100644 --- a/data-node/pom.xml +++ b/data-node/pom.xml @@ -68,6 +68,16 @@ com.github.rvesse airline + + com.google.auto.value + auto-value-annotations + provided + + + org.graylog.autovalue + auto-value-javabean + provided + com.google.guava guava diff --git a/data-node/src/main/java/org/graylog/datanode/initializers/JerseyService.java b/data-node/src/main/java/org/graylog/datanode/initializers/JerseyService.java index 1d226cf8f8..5de4c46c8c 100644 --- a/data-node/src/main/java/org/graylog/datanode/initializers/JerseyService.java +++ b/data-node/src/main/java/org/graylog/datanode/initializers/JerseyService.java @@ -42,6 +42,7 @@ import org.graylog.datanode.configuration.variants.KeystoreInformation; import org.graylog.datanode.configuration.variants.OpensearchSecurityConfiguration; import org.graylog.datanode.management.OpensearchConfigurationChangeEvent; import org.graylog.datanode.process.OpensearchConfiguration; +import org.graylog.datanode.rest.config.SecuredNodeAnnotationFilter; import org.graylog.security.certutil.CertConstants; import org.graylog2.bootstrap.preflight.web.BasicAuthFilter; import org.graylog2.configuration.TLSProtocolsConfiguration; @@ -231,6 +232,7 @@ public class JerseyService extends AbstractIdleService { if(isSecuredInstance) { resourceConfig.register(new BasicAuthFilter(configuration.getRootUsername(), configuration.getRootPasswordSha2(), "Datanode")); } + resourceConfig.register(new SecuredNodeAnnotationFilter(configuration.isInsecureStartup())); final HttpServer httpServer = GrizzlyHttpServerFactory.createHttpServer( listenUri, diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java index af3f362839..4e49a078c7 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java +++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java @@ -37,4 +37,8 @@ public interface OpensearchProcess extends ManagableProcess { @@ -41,18 +46,23 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro private final OpensearchProcess process; private final Provider configurationProvider; private final EventBus eventBus; + private final NodeId nodeId; @Inject public OpensearchProcessService(final DatanodeConfiguration datanodeConfiguration, final Provider configurationProvider, final EventBus eventBus, + final ClusterEventBus clusterEventBus, final CustomCAX509TrustManager trustManager, final NodeService nodeService, - final Configuration configuration) { + final Configuration configuration, + final NodeId nodeId) { this.configurationProvider = configurationProvider; this.eventBus = eventBus; + this.nodeId = nodeId; this.process = createOpensearchProcess(datanodeConfiguration, trustManager, configuration, nodeService); eventBus.register(this); + checkNotNull(clusterEventBus).registerClusterEventSubscriber(this); } private OpensearchProcess createOpensearchProcess(final DatanodeConfiguration datanodeConfiguration, final CustomCAX509TrustManager trustManager, final Configuration configuration, final NodeService nodeService) { @@ -60,16 +70,29 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro final ProcessWatchdog watchdog = new ProcessWatchdog(process, WATCHDOG_RESTART_ATTEMPTS); process.addStateMachineTracer(watchdog); process.addStateMachineTracer(new StateMachineTransitionLogger()); + process.addStateMachineTracer(new OpensearchRemovalTracer(process, configuration.getDatanodeNodeName())); return process; } @Subscribe + @SuppressWarnings("unused") public void handlePreflightConfigEvent(DataNodeProvisioningStateChangeEvent event) { switch (event.state()) { case STORED -> startWithConfig(); } } + @Subscribe + @SuppressWarnings("unused") + public void handleNodeLifecycleEvent(DataNodeLifecycleEvent event) { + if (nodeId.getNodeId().equals(event.nodeId())) { + switch (event.trigger()) { + case REMOVE -> process.onRemove(); + case RESET -> process.onReset(); + } + } + } + @Override protected void startUp() { startWithConfig(); diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchRemovalTracer.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchRemovalTracer.java new file mode 100644 index 0000000000..7aa114bbf1 --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchRemovalTracer.java @@ -0,0 +1,142 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.management; + +import com.google.common.util.concurrent.ThreadFactoryBuilder; +import org.graylog.datanode.process.ProcessEvent; +import org.graylog.datanode.process.ProcessState; +import org.graylog.datanode.process.StateMachineTracer; +import org.graylog.shaded.opensearch2.org.opensearch.OpenSearchStatusException; +import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.health.ClusterHealthRequest; +import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.health.ClusterHealthResponse; +import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.settings.ClusterGetSettingsRequest; +import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.settings.ClusterGetSettingsResponse; +import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest; +import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsResponse; +import org.graylog.shaded.opensearch2.org.opensearch.client.ClusterClient; +import org.graylog.shaded.opensearch2.org.opensearch.client.RequestOptions; +import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; +import org.graylog.shaded.opensearch2.org.opensearch.common.settings.Settings; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Optional; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * This tracer triggers functionality for node removal. + * It triggers removal on PROCESS_REMOVE and will reset the allocation exclude setting in OS on successful start of OS, + * which cannot be done after the removal due to the unavailability of the OS REST api. + */ +public class OpensearchRemovalTracer implements StateMachineTracer { + + private static final Logger LOG = LoggerFactory.getLogger(OpensearchRemovalTracer.class); + static final String CLUSTER_ROUTING_ALLOCATION_EXCLUDE_SETTING = "cluster.routing.allocation.exclude._name"; + + private final OpensearchProcess process; + private final String nodeName; + boolean allocationExcludeChecked = false; + ScheduledExecutorService executorService; + + public OpensearchRemovalTracer(OpensearchProcess process, String nodeName) { + this.process = process; + this.nodeName = nodeName; + } + + + @Override + public void trigger(ProcessEvent trigger) { + LOG.debug("Removal tracer trigger: {}", trigger); + } + + @Override + public void transition(ProcessEvent trigger, ProcessState source, ProcessState destination) { + if (destination == ProcessState.AVAILABLE && !allocationExcludeChecked) { + checkAllocationEnabledStatus(); + } else if (trigger == ProcessEvent.PROCESS_REMOVE) { + triggerRemoval(); + } + } + + private void checkAllocationEnabledStatus() { + if (process.restClient().isPresent()) { + ClusterClient clusterClient = process.restClient().get().cluster(); + try { + final ClusterGetSettingsResponse settings = + clusterClient.getSettings(new ClusterGetSettingsRequest(), RequestOptions.DEFAULT); + final String setting = settings.getSetting(CLUSTER_ROUTING_ALLOCATION_EXCLUDE_SETTING); + if (nodeName.equals(setting)) { + ClusterUpdateSettingsRequest updateSettings = new ClusterUpdateSettingsRequest(); + updateSettings.transientSettings(Settings.builder() + .putNull(CLUSTER_ROUTING_ALLOCATION_EXCLUDE_SETTING) + .build()); + clusterClient.putSettings(updateSettings, RequestOptions.DEFAULT); + } + allocationExcludeChecked = true; + } catch (IOException e) { + LOG.error("Error getting cluster settings from OpenSearch"); + } + } + } + + private void triggerRemoval() { + process.restClient().ifPresent(client -> { + final ClusterClient clusterClient = client.cluster(); + ClusterUpdateSettingsRequest settings = new ClusterUpdateSettingsRequest(); + settings.transientSettings(Settings.builder() + .put(CLUSTER_ROUTING_ALLOCATION_EXCLUDE_SETTING, nodeName) + .build()); + try { + final ClusterUpdateSettingsResponse response = + clusterClient.putSettings(settings, RequestOptions.DEFAULT); + if (response.isAcknowledged()) { + allocationExcludeChecked = false; // reset to rejoin cluster in case of failure + executorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder().setNameFormat("datanode-removal").build()); + executorService.scheduleAtFixedRate(this::checkRemovalStatus, 10, 10, TimeUnit.SECONDS); + } else { + process.onEvent(ProcessEvent.HEALTH_CHECK_FAILED); + } + } catch (IOException e) { + LOG.error("Failed to exclude node from cluster allocation", e); + process.onEvent(ProcessEvent.HEALTH_CHECK_FAILED); + } + }); + } + + void checkRemovalStatus() { + final Optional restClient = process.restClient(); + if (restClient.isPresent()) { + try { + final ClusterClient clusterClient = restClient.get().cluster(); + final ClusterHealthResponse health = clusterClient + .health(new ClusterHealthRequest(), RequestOptions.DEFAULT); + if (health.getRelocatingShards() == 0) { + process.stop(); + executorService.shutdown(); + // TODO: set the state for this datanode to removed in nodes + // and include state in OpensearchProcessImpl.writeSeedHostsList + } + } catch (IOException | OpenSearchStatusException e) { + process.onEvent(ProcessEvent.HEALTH_CHECK_FAILED); + } + } + } + +} diff --git a/data-node/src/main/java/org/graylog/datanode/periodicals/OpensearchNodeHeartbeat.java b/data-node/src/main/java/org/graylog/datanode/periodicals/OpensearchNodeHeartbeat.java index 8944c78ad0..e79b27ff60 100644 --- a/data-node/src/main/java/org/graylog/datanode/periodicals/OpensearchNodeHeartbeat.java +++ b/data-node/src/main/java/org/graylog/datanode/periodicals/OpensearchNodeHeartbeat.java @@ -47,7 +47,8 @@ public class OpensearchNodeHeartbeat extends Periodical { @Override // This method is "synchronized" because we are also calling it directly in AutomaticLeaderElectionService public synchronized void doRun() { - if (!process.isInState(ProcessState.TERMINATED) && !process.isInState(ProcessState.WAITING_FOR_CONFIGURATION)) { + if (!process.isInState(ProcessState.TERMINATED) && !process.isInState(ProcessState.WAITING_FOR_CONFIGURATION) + && !process.isInState(ProcessState.REMOVED)) { final Optional restClient = process.restClient(); if(restClient.isPresent()) { diff --git a/data-node/src/main/java/org/graylog/datanode/process/ProcessEvent.java b/data-node/src/main/java/org/graylog/datanode/process/ProcessEvent.java index 5707e6d580..0066f60dc6 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/ProcessEvent.java +++ b/data-node/src/main/java/org/graylog/datanode/process/ProcessEvent.java @@ -20,6 +20,8 @@ public enum ProcessEvent { PROCESS_STARTED, HEALTH_CHECK_OK, HEALTH_CHECK_FAILED, - PROCESS_STOPPED, // user-triggered action + PROCESS_STOPPED, + PROCESS_REMOVE, + RESET, // user-triggered action PROCESS_TERMINATED // failure from outside, not requested } diff --git a/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java b/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java index cc0c0fbf15..ffd632b537 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java +++ b/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java @@ -39,6 +39,16 @@ public enum ProcessState { * Failed to reach the opensearch REST API, but the underlying process is still alive */ FAILED, + + /** + * Removal of node from Opensearch cluster requested + */ + REMOVING, + /** + * Removal of node from Opensearch cluster completed + */ + REMOVED, + /** * The OS process is not running anymore on the underlying system, it has been terminated */ diff --git a/data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachine.java b/data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachine.java index e6c3864617..76d358f721 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachine.java +++ b/data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachine.java @@ -62,6 +62,7 @@ public class ProcessStateMachine { .permit(ProcessEvent.HEALTH_CHECK_FAILED, ProcessState.NOT_RESPONDING) .permit(ProcessEvent.PROCESS_STOPPED, ProcessState.TERMINATED) .permit(ProcessEvent.PROCESS_TERMINATED, ProcessState.TERMINATED) + .permit(ProcessEvent.PROCESS_REMOVE, ProcessState.REMOVING) .ignore(ProcessEvent.PROCESS_STARTED); // if the REST api is not responding, we'll jump to this state and count how many times the failure @@ -90,6 +91,15 @@ public class ProcessStateMachine { .ignore(ProcessEvent.PROCESS_STOPPED) .ignore(ProcessEvent.PROCESS_TERMINATED); // final state, all following terminate events are ignored + config.configure(ProcessState.REMOVING) + .ignore(ProcessEvent.HEALTH_CHECK_OK) + .permit(ProcessEvent.HEALTH_CHECK_FAILED, ProcessState.FAILED) + .permit(ProcessEvent.PROCESS_STOPPED, ProcessState.REMOVED); + + config.configure(ProcessState.REMOVED) + .permit(ProcessEvent.RESET, ProcessState.WAITING_FOR_CONFIGURATION) + .ignore(ProcessEvent.PROCESS_STOPPED); + return new StateMachine<>(ProcessState.WAITING_FOR_CONFIGURATION, config); } } diff --git a/data-node/src/main/java/org/graylog/datanode/rest/ManagementController.java b/data-node/src/main/java/org/graylog/datanode/rest/ManagementController.java new file mode 100644 index 0000000000..d5218c64ef --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/rest/ManagementController.java @@ -0,0 +1,45 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.rest; + +import org.graylog.datanode.management.OpensearchProcess; +import org.graylog.datanode.rest.config.OnlyInSecuredNode; + +import javax.inject.Inject; +import javax.ws.rs.DELETE; +import javax.ws.rs.Path; +import javax.ws.rs.Produces; +import javax.ws.rs.core.MediaType; + +@Path("/management") +@Produces(MediaType.APPLICATION_JSON) +public class ManagementController { + + private final OpensearchProcess openSearch; + + @Inject + public ManagementController(OpensearchProcess openSearch) { + this.openSearch = openSearch; + } + + @DELETE + @OnlyInSecuredNode + public void remove() { + openSearch.onRemove(); + } + +} diff --git a/data-node/src/main/java/org/graylog/datanode/rest/RestBindings.java b/data-node/src/main/java/org/graylog/datanode/rest/RestBindings.java index 095ccd26d8..fa3157a9ee 100644 --- a/data-node/src/main/java/org/graylog/datanode/rest/RestBindings.java +++ b/data-node/src/main/java/org/graylog/datanode/rest/RestBindings.java @@ -23,5 +23,6 @@ public class RestBindings extends Graylog2Module { protected void configure() { addSystemRestResource(StatusController.class); addSystemRestResource(LogsController.class); + addSystemRestResource(ManagementController.class); } } diff --git a/data-node/src/main/java/org/graylog/datanode/rest/StatusController.java b/data-node/src/main/java/org/graylog/datanode/rest/StatusController.java index 5bbfc127d9..93e5d1e456 100644 --- a/data-node/src/main/java/org/graylog/datanode/rest/StatusController.java +++ b/data-node/src/main/java/org/graylog/datanode/rest/StatusController.java @@ -48,4 +48,5 @@ public class StatusController { new StatusResponse(datanodeConfiguration.opensearchDistributionProvider().get().version(), openSearch.processInfo()) ); } + } diff --git a/data-node/src/main/java/org/graylog/datanode/rest/config/OnlyInSecuredNode.java b/data-node/src/main/java/org/graylog/datanode/rest/config/OnlyInSecuredNode.java new file mode 100644 index 0000000000..18e2939a9c --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/rest/config/OnlyInSecuredNode.java @@ -0,0 +1,27 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.rest.config; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +@Target(ElementType.METHOD) +@Retention(RetentionPolicy.RUNTIME) +public @interface OnlyInSecuredNode { +} diff --git a/data-node/src/main/java/org/graylog/datanode/rest/config/SecuredNodeAnnotationFilter.java b/data-node/src/main/java/org/graylog/datanode/rest/config/SecuredNodeAnnotationFilter.java new file mode 100644 index 0000000000..19997892b6 --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/rest/config/SecuredNodeAnnotationFilter.java @@ -0,0 +1,60 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.rest.config; + +import org.glassfish.jersey.server.ContainerRequest; +import org.glassfish.jersey.server.model.ResourceMethod; + +import javax.ws.rs.container.ContainerRequestContext; +import javax.ws.rs.container.ContainerRequestFilter; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import java.io.IOException; +import java.util.Objects; + +public class SecuredNodeAnnotationFilter implements ContainerRequestFilter { + + private final boolean isInsecureNode; + + public SecuredNodeAnnotationFilter(boolean isInsecureNode) { + this.isInsecureNode = isInsecureNode; + } + + @Override + public void filter(ContainerRequestContext requestContext) throws IOException { + if (requestContext instanceof ContainerRequest request) { + + final ResourceMethod method = request.getUriInfo().getMatchedResourceMethod(); + final OnlyInSecuredNode annotation = method.getInvocable() + .getHandlingMethod().getAnnotation(OnlyInSecuredNode.class); + + if (Objects.nonNull(annotation) && isInsecureNode) { + requestContext.abortWith(Response.status(Response.Status.UNAUTHORIZED) + .entity("This resource can only be accessed in secured data nodes.") + .type(MediaType.TEXT_PLAIN_TYPE) + .build()); + } + + } else { + requestContext.abortWith(Response.status(Response.Status.INTERNAL_SERVER_ERROR) + .entity("Server is no Jetty Server.") + .type(MediaType.TEXT_PLAIN_TYPE) + .build()); + } + + } +} diff --git a/data-node/src/test/java/org/graylog/datanode/DatanodeOpensearchWait.java b/data-node/src/test/java/org/graylog/datanode/DatanodeOpensearchWait.java deleted file mode 100644 index 8eb233f25c..0000000000 --- a/data-node/src/test/java/org/graylog/datanode/DatanodeOpensearchWait.java +++ /dev/null @@ -1,187 +0,0 @@ -/* - * Copyright (C) 2020 Graylog, Inc. - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the Server Side Public License, version 1, - * as published by MongoDB, Inc. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * Server Side Public License for more details. - * - * You should have received a copy of the Server Side Public License - * along with this program. If not, see - * . - */ -package org.graylog.datanode; - -import com.github.rholder.retry.Attempt; -import com.github.rholder.retry.RetryException; -import com.github.rholder.retry.RetryListener; -import com.github.rholder.retry.Retryer; -import com.github.rholder.retry.RetryerBuilder; -import com.github.rholder.retry.StopStrategies; -import com.github.rholder.retry.WaitStrategies; -import io.restassured.RestAssured; -import io.restassured.http.ContentType; -import io.restassured.response.ExtractableResponse; -import io.restassured.response.Response; -import io.restassured.response.ValidatableResponse; -import io.restassured.specification.RequestSpecification; -import org.apache.http.NoHttpResponseException; -import org.graylog.datanode.integration.DatanodeSecuritySetupIT; -import org.jetbrains.annotations.NotNull; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import javax.net.ssl.SSLHandshakeException; -import java.net.SocketException; -import java.security.KeyStore; -import java.util.LinkedList; -import java.util.List; -import java.util.Locale; -import java.util.Optional; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.function.Predicate; - -public class DatanodeOpensearchWait { - - private static final Logger LOG = LoggerFactory.getLogger(DatanodeOpensearchWait.class); - private static final int ATTEMPTS_COUNT = 160; - - private final int opensearchPort; - private KeyStore truststore; - private String jwtToken; - private String lastRecordedResponse; - - private DatanodeOpensearchWait(int opensearchPort) { - this.opensearchPort = opensearchPort; - } - - public static DatanodeOpensearchWait onPort(int opensearchPort) { - return new DatanodeOpensearchWait(opensearchPort); - } - - public DatanodeOpensearchWait withTruststore(KeyStore truststore) { - this.truststore = truststore; - return this; - } - - public DatanodeOpensearchWait withJwtAuth(String jwtToken) { - this.jwtToken = jwtToken; - return this; - } - - public ValidatableResponse waitForGreenStatus() throws ExecutionException, RetryException { - return waitForOpensearch( - input -> !input.extract().body().path("status").equals("green")); - } - - public ValidatableResponse waitForNodesCount(int countOfNodes) throws ExecutionException, RetryException { - return waitForOpensearch( - input -> !input.extract().body().path("discovered_cluster_manager").equals(true), - input -> !input.extract().body().path("status").equals("green"), - input -> !input.extract().body().path("number_of_nodes").equals(countOfNodes) - ); - } - - - @SafeVarargs - private ValidatableResponse waitForOpensearch(Predicate... predicates) throws ExecutionException, RetryException { - - if(LOG.isInfoEnabled()) { - LOG.info("cURL request: " + formatCurlConnectionInfo()); - } - - //noinspection UnstableApiUsage - final RetryerBuilder builder = RetryerBuilder.newBuilder() - .withWaitStrategy(WaitStrategies.fixedWait(1, TimeUnit.SECONDS)) - .withStopStrategy(StopStrategies.stopAfterAttempt(ATTEMPTS_COUNT)) - .retryIfException(input -> input instanceof NoHttpResponseException) - .retryIfException(input -> input instanceof SocketException) - .retryIfException(input -> input instanceof SSLHandshakeException) // may happen before SSL is configured properly - .retryIfResult(input -> !input.extract().contentType().startsWith("application/json")) - .withRetryListener(new RetryListener() { - @Override - public void onRetry(Attempt attempt) { - if (attempt.hasResult() && attempt.getResult() instanceof ValidatableResponse response) { - lastRecordedResponse = serializeResponse(response.extract()); - } else if (attempt.getExceptionCause() != null) { - lastRecordedResponse = attempt.getExceptionCause().getMessage(); - } - } - }); - - for (Predicate predicate : predicates) { - builder.retryIfResult(predicate::test); - } - - final Retryer retryer = builder - .build(); - try { - return retryer.call(() -> { - final RequestSpecification req = RestAssured.given() - .accept(ContentType.JSON); - - Optional.ofNullable(truststore).ifPresent(ts -> req.trustStore(truststore)); - - if (jwtToken != null) { - req.header( "Authorization", "Bearer " + jwtToken); - } - - return req.get(formatClusterHeathUrl()) - .then(); - }); - } catch (ExecutionException | RetryException e) { - if (lastRecordedResponse != null) { - LOG.warn("Last recorded opensearch response, waiting for {}: {}", formatClusterHeathUrl(), lastRecordedResponse); - } - printShardsInfo(); - throw e; - } - } - - private String formatCurlConnectionInfo() { - List builder = new LinkedList<>(); - builder.add("curl -k"); // trust self-signed certs - if(jwtToken != null) { - builder.add("-H \"Authorization: Bearer " + jwtToken + "\""); - } - builder.add(formatClusterHeathUrl()); - return String.join(" ", builder); - } - - private void printShardsInfo() { - try { - final RequestSpecification req = RestAssured.given() - .accept(ContentType.JSON); - - Optional.ofNullable(truststore).ifPresent(ts -> req.trustStore(truststore)); - - if (jwtToken != null) { - req.header( "Authorization", "Bearer " + jwtToken); - } - - final String shardsResponse = req.get("https://localhost:" + opensearchPort + "/_cat/shards?v=true&h=index,shard,prirep,state,node,unassigned.reason&s=state") - .then().extract().body().asString(); - - LOG.warn("Cluster shards status: {}", shardsResponse); - } catch (Exception e) { - LOG.warn("Failed to obtain shard info", e); - } - } - - @NotNull - private String formatClusterHeathUrl() { - return "https://localhost:" + opensearchPort + "/_cluster/health"; - } - - private String serializeResponse(ExtractableResponse extract) { - final String body = extract.body().asString(); - final String contentType = extract.contentType(); - final String headers = extract.headers().toString(); - return String.format(Locale.ROOT, "\nContent-type: %s\nheaders: %s\nbody: %s", contentType, headers, body); - } -} diff --git a/data-node/src/test/java/org/graylog/datanode/integration/DatanodeClusterIT.java b/data-node/src/test/java/org/graylog/datanode/integration/DatanodeClusterIT.java index 209b606056..f95072448b 100644 --- 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 shardNodes = osIndexClient.getShardNodes(); + Assertions.assertEquals(shardNodes.size(), 2); + + List nodes = List.of(nodeA, nodeB, nodeC); + final Optional primary = nodes.stream().filter(n -> shardNodes.get(0).equals(n.getNodeName())).findFirst(); + assertTrue(primary.isPresent()); + final Optional 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 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()); diff --git a/data-node/src/test/java/org/graylog/datanode/integration/DatanodeSecuritySetupIT.java b/data-node/src/test/java/org/graylog/datanode/integration/DatanodeSecuritySetupIT.java index 91729ea593..d1817bc0ad 100644 --- 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()); diff --git a/data-node/src/test/java/org/graylog/datanode/integration/DatanodeStartupIT.java b/data-node/src/test/java/org/graylog/datanode/integration/DatanodeStartupIT.java index 3089ad3c85..1eeccd5fad 100644 --- 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) diff --git a/data-node/src/test/java/org/graylog/datanode/management/OpensearchRemovalTracerTest.java b/data-node/src/test/java/org/graylog/datanode/management/OpensearchRemovalTracerTest.java new file mode 100644 index 0000000000..2113ac8a8a --- /dev/null +++ b/data-node/src/test/java/org/graylog/datanode/management/OpensearchRemovalTracerTest.java @@ -0,0 +1,140 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.management; + +import org.graylog.datanode.process.ProcessEvent; +import org.graylog.datanode.process.ProcessState; +import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.health.ClusterHealthResponse; +import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.settings.ClusterGetSettingsResponse; +import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest; +import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsResponse; +import org.graylog.shaded.opensearch2.org.opensearch.client.ClusterClient; +import org.graylog.shaded.opensearch2.org.opensearch.client.RequestOptions; +import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; +import org.graylog.shaded.opensearch2.org.opensearch.common.settings.Settings; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import java.io.IOException; +import java.util.Optional; +import java.util.concurrent.ScheduledExecutorService; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class OpensearchRemovalTracerTest { + + private OpensearchRemovalTracer classUnderTest; + private final String NODENAME = "datanode1"; + @Mock + private OpensearchProcess process; + @Mock + RestHighLevelClient restClient; + @Mock + ClusterClient clusterClient; + + @Before + public void setUp() { + when(process.restClient()).thenReturn(Optional.of(restClient)); + when(restClient.cluster()).thenReturn(clusterClient); + this.classUnderTest = new OpensearchRemovalTracer(process, NODENAME); + } + + @Test + public void testResetAllocation() throws IOException { + Settings settings = Settings.builder() + .put(OpensearchRemovalTracer.CLUSTER_ROUTING_ALLOCATION_EXCLUDE_SETTING, NODENAME) + .build(); + when(clusterClient.getSettings(any(), any())).thenReturn(new ClusterGetSettingsResponse(null, settings, null)); + classUnderTest.transition(ProcessEvent.HEALTH_CHECK_OK, ProcessState.STARTING, ProcessState.AVAILABLE); + + ArgumentCaptor settingsRequest = + ArgumentCaptor.forClass(ClusterUpdateSettingsRequest.class); + verify(clusterClient).putSettings(settingsRequest.capture(), eq(RequestOptions.DEFAULT)); + assertNull(settingsRequest.getValue() + .transientSettings() + .get(OpensearchRemovalTracer.CLUSTER_ROUTING_ALLOCATION_EXCLUDE_SETTING) + ); + assertTrue(classUnderTest.allocationExcludeChecked); + } + + @Test + public void testResetAllocationUnneccessary() throws IOException { + Settings settings = Settings.builder() + .put(OpensearchRemovalTracer.CLUSTER_ROUTING_ALLOCATION_EXCLUDE_SETTING, "notmynodename") + .build(); + when(clusterClient.getSettings(any(), any())).thenReturn(new ClusterGetSettingsResponse(null, settings, null)); + classUnderTest.transition(ProcessEvent.HEALTH_CHECK_OK, ProcessState.STARTING, ProcessState.AVAILABLE); + verify(clusterClient).getSettings(any(), any()); + verifyNoMoreInteractions(clusterClient); + assertTrue(classUnderTest.allocationExcludeChecked); + } + + @Test + public void testRemovalTriggered() throws IOException { + ClusterUpdateSettingsResponse response = mock(ClusterUpdateSettingsResponse.class); + when(response.isAcknowledged()).thenReturn(true); + when(clusterClient.putSettings(any(), any())).thenReturn(response); + classUnderTest.transition(ProcessEvent.PROCESS_REMOVE, ProcessState.AVAILABLE, ProcessState.REMOVING); + + ArgumentCaptor settingsRequest = + ArgumentCaptor.forClass(ClusterUpdateSettingsRequest.class); + verify(clusterClient).putSettings(settingsRequest.capture(), eq(RequestOptions.DEFAULT)); + assertEquals(NODENAME, + settingsRequest.getValue() + .transientSettings() + .get(OpensearchRemovalTracer.CLUSTER_ROUTING_ALLOCATION_EXCLUDE_SETTING)); + assertFalse(classUnderTest.allocationExcludeChecked); + } + + @Test + public void testNoShutdownWhenRelocatingShards() throws IOException { + ClusterHealthResponse health = mock(ClusterHealthResponse.class); + when(health.getRelocatingShards()).thenReturn(5); + when(clusterClient.health(any(), any())).thenReturn(health); + classUnderTest.checkRemovalStatus(); + verify(process).restClient(); + verifyNoMoreInteractions(process); + } + + @Test + public void testShutdownWhenRemovedSuccessfully() throws IOException { + ClusterHealthResponse health = mock(ClusterHealthResponse.class); + when(health.getRelocatingShards()).thenReturn(0); + when(clusterClient.health(any(), any())).thenReturn(health); + final ScheduledExecutorService executor = mock(ScheduledExecutorService.class); + classUnderTest.executorService = executor; + classUnderTest.checkRemovalStatus(); + verify(process).stop(); + verify(executor).shutdown(); + } + + +} diff --git a/data-node/src/test/java/org/graylog/datanode/process/ProcessStateMachineTest.java b/data-node/src/test/java/org/graylog/datanode/process/ProcessStateMachineTest.java index c2230e4b45..76f6f39c8c 100644 --- a/data-node/src/test/java/org/graylog/datanode/process/ProcessStateMachineTest.java +++ b/data-node/src/test/java/org/graylog/datanode/process/ProcessStateMachineTest.java @@ -99,4 +99,43 @@ class ProcessStateMachineTest { // succeeded just in time before we give up Assertions.assertEquals(ProcessState.AVAILABLE, machine.getState()); } + + @Test + void testSuccessfullRemoval() { + final StateMachine machine = ProcessStateMachine.createNew(); + Assertions.assertEquals(machine.getState(), ProcessState.WAITING_FOR_CONFIGURATION); + + machine.fire(ProcessEvent.PROCESS_STARTED); + Assertions.assertEquals(ProcessState.STARTING, machine.getState()); + + machine.fire(ProcessEvent.HEALTH_CHECK_OK); + Assertions.assertEquals(ProcessState.AVAILABLE, machine.getState()); + + machine.fire(ProcessEvent.PROCESS_REMOVE); + Assertions.assertEquals(ProcessState.REMOVING, machine.getState()); + + machine.fire(ProcessEvent.PROCESS_STOPPED); + Assertions.assertEquals(ProcessState.REMOVED, machine.getState()); + + } + + @Test + void testFailingRemoval() { + final StateMachine machine = ProcessStateMachine.createNew(); + Assertions.assertEquals(machine.getState(), ProcessState.WAITING_FOR_CONFIGURATION); + + machine.fire(ProcessEvent.PROCESS_STARTED); + Assertions.assertEquals(ProcessState.STARTING, machine.getState()); + + machine.fire(ProcessEvent.HEALTH_CHECK_OK); + Assertions.assertEquals(ProcessState.AVAILABLE, machine.getState()); + + machine.fire(ProcessEvent.PROCESS_REMOVE); + Assertions.assertEquals(ProcessState.REMOVING, machine.getState()); + + machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); + Assertions.assertEquals(ProcessState.FAILED, machine.getState()); + + } + } diff --git a/data-node/src/test/java/org/graylog/datanode/restoperations/DatanodeOpensearchWait.java b/data-node/src/test/java/org/graylog/datanode/restoperations/DatanodeOpensearchWait.java new file mode 100644 index 0000000000..90156cc501 --- /dev/null +++ b/data-node/src/test/java/org/graylog/datanode/restoperations/DatanodeOpensearchWait.java @@ -0,0 +1,38 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.restoperations; + +import com.github.rholder.retry.RetryException; +import io.restassured.response.ValidatableResponse; + +import java.util.concurrent.ExecutionException; + +public class DatanodeOpensearchWait extends WaitingRestOperation { + + public DatanodeOpensearchWait(RestOperationParameters waitingRestOperationParameters) { + super(waitingRestOperationParameters); + } + + public ValidatableResponse waitForNodesCount(int countOfNodes) throws ExecutionException, RetryException { + return waitForResponse("/_cluster/health", + input -> !input.extract().body().path("discovered_cluster_manager").equals(true), + input -> !input.extract().body().path("status").equals("green"), + input -> !input.extract().body().path("number_of_nodes").equals(countOfNodes) + ); + } + +} diff --git a/data-node/src/test/java/org/graylog/datanode/restoperations/DatanodeRestApiWait.java b/data-node/src/test/java/org/graylog/datanode/restoperations/DatanodeRestApiWait.java new file mode 100644 index 0000000000..b8b9431c34 --- /dev/null +++ b/data-node/src/test/java/org/graylog/datanode/restoperations/DatanodeRestApiWait.java @@ -0,0 +1,36 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.restoperations; + +import com.github.rholder.retry.RetryException; +import io.restassured.response.ValidatableResponse; + +import java.util.concurrent.ExecutionException; + +public class DatanodeRestApiWait extends WaitingRestOperation { + + public DatanodeRestApiWait(RestOperationParameters waitingRestOperationParameters) { + super(waitingRestOperationParameters); + } + + public ValidatableResponse waitForAvailableStatus() throws ExecutionException, RetryException { + return waitForResponse("/", + input -> !input.extract().body().path("opensearch.node.state").equals("AVAILABLE") + ); + } + +} diff --git a/data-node/src/test/java/org/graylog/datanode/restoperations/DatanodeStatusChangeOperation.java b/data-node/src/test/java/org/graylog/datanode/restoperations/DatanodeStatusChangeOperation.java new file mode 100644 index 0000000000..8f5f479e95 --- /dev/null +++ b/data-node/src/test/java/org/graylog/datanode/restoperations/DatanodeStatusChangeOperation.java @@ -0,0 +1,35 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.restoperations; + + +import javax.ws.rs.HttpMethod; + +public class DatanodeStatusChangeOperation extends RestOperation { + + public DatanodeStatusChangeOperation(RestOperationParameters waitingRestOperationParameters) { + super(waitingRestOperationParameters); + } + + public void triggerNodeRemoval() { + validatedResponse("/management", HttpMethod.DELETE, null, + "Could not trigger node removal", + r -> r.extract().statusCode() < 300 + ); + } + +} diff --git a/data-node/src/test/java/org/graylog/datanode/restoperations/OpensearchTestIndexCreation.java b/data-node/src/test/java/org/graylog/datanode/restoperations/OpensearchTestIndexCreation.java new file mode 100644 index 0000000000..718daeaeab --- /dev/null +++ b/data-node/src/test/java/org/graylog/datanode/restoperations/OpensearchTestIndexCreation.java @@ -0,0 +1,62 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.restoperations; + + +import com.github.rholder.retry.RetryException; +import io.restassured.path.json.JsonPath; +import io.restassured.response.ValidatableResponse; + +import javax.ws.rs.HttpMethod; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; + +public class OpensearchTestIndexCreation extends RestOperation { + + public static final String IT_TEST_INDEX = "it_test_index"; + + public OpensearchTestIndexCreation(RestOperationParameters waitingRestOperationParameters) { + super(waitingRestOperationParameters); + } + + public void createIndex() { + validatedResponse("/" + IT_TEST_INDEX, HttpMethod.PUT, + "{\"settings\" : { \"index.number_of_shards\": 1, \"index.number_of_replicas\": 1 }}", + "Could not create test index", + r -> r.extract().body().path("acknowledged").equals(true), + r -> r.extract().body().path("shards_acknowledged").equals(true) + ); + } + + public List getShardNodes() throws ExecutionException, RetryException { + + final ValidatableResponse response = new OpensearchTestIndexCreationWait(this.parameters).waitForIndexCreation(); + final JsonPath jsonPath = response.extract().body().jsonPath(); + + List> shards = jsonPath.getList("$"); + assertEquals(shards.size(), 2, "Expecting one primary and one replica shard."); + return shards.stream() + .peek(shard -> assertNotEquals("UNASSIGNED", shard.get("state"))) + .map(shard -> shard.get("node")) + .collect(Collectors.toList()); + } +} diff --git a/data-node/src/test/java/org/graylog/datanode/restoperations/OpensearchTestIndexCreationWait.java b/data-node/src/test/java/org/graylog/datanode/restoperations/OpensearchTestIndexCreationWait.java new file mode 100644 index 0000000000..865778fb58 --- /dev/null +++ b/data-node/src/test/java/org/graylog/datanode/restoperations/OpensearchTestIndexCreationWait.java @@ -0,0 +1,37 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.restoperations; + +import com.github.rholder.retry.RetryException; +import io.restassured.response.ValidatableResponse; + +import java.util.concurrent.ExecutionException; + +public class OpensearchTestIndexCreationWait extends WaitingRestOperation { + + public OpensearchTestIndexCreationWait(RestOperationParameters waitingRestOperationParameters) { + super(waitingRestOperationParameters); + } + + public ValidatableResponse waitForIndexCreation() throws ExecutionException, RetryException { + return waitForResponse("/_cat/shards/" + OpensearchTestIndexCreation.IT_TEST_INDEX + "?h=node,state&format=json", + input -> input.extract().body().jsonPath().getList("state").size() != 2, + input -> input.extract().body().jsonPath().getList("state").stream().anyMatch(s -> !s.equals("STARTED")) + ); + } + +} diff --git a/data-node/src/test/java/org/graylog/datanode/restoperations/RestOperation.java b/data-node/src/test/java/org/graylog/datanode/restoperations/RestOperation.java new file mode 100644 index 0000000000..33106e522f --- /dev/null +++ b/data-node/src/test/java/org/graylog/datanode/restoperations/RestOperation.java @@ -0,0 +1,98 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.restoperations; + +import io.restassured.RestAssured; +import io.restassured.http.ContentType; +import io.restassured.http.Header; +import io.restassured.response.ExtractableResponse; +import io.restassured.response.Response; +import io.restassured.response.ValidatableResponse; +import io.restassured.specification.RequestSpecification; +import org.bson.assertions.Assertions; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.ws.rs.HttpMethod; +import java.util.Arrays; +import java.util.LinkedList; +import java.util.List; +import java.util.Locale; +import java.util.Optional; +import java.util.function.Predicate; + +public abstract class RestOperation { + + static final Logger LOG = LoggerFactory.getLogger(RestOperation.class); + + final RestOperationParameters parameters; + + protected RestOperation(RestOperationParameters waitingRestOperationParameters) { + this.parameters = waitingRestOperationParameters; + } + + @SafeVarargs + final ValidatableResponse validatedResponse(String url, String method, String body, String errorMessage, Predicate... predicates) { + + if (LOG.isInfoEnabled()) { + LOG.info("cURL request: " + formatCurlConnectionInfo(parameters.port(), url)); + } + + + final RequestSpecification req = RestAssured.given().accept(ContentType.JSON); + Optional.ofNullable(parameters.truststore()).ifPresent(ts -> req.trustStore(parameters.truststore())); + parameters.addAuthorizationHeaders(req); + ValidatableResponse response; + if (method.toUpperCase(Locale.ENGLISH).equals(HttpMethod.GET)) { + response = req.get(formatUrl(parameters.port(), url)).then(); + } else if (method.toUpperCase(Locale.ENGLISH).equals(HttpMethod.PUT)) { + response = req.header(new Header("Content-Type", "application/json")).body(body).put(formatUrl(parameters.port(), url)).then(); + } else if (method.toUpperCase(Locale.ENGLISH).equals(HttpMethod.POST)) { + response = req.header(new Header("Content-Type", "application/json")).body(body).post(formatUrl(parameters.port(), url)).then(); + } else if (method.toUpperCase(Locale.ENGLISH).equals(HttpMethod.DELETE)) { + response = req.delete(formatUrl(parameters.port(), url)).then(); + } else { + throw Assertions.fail("Method not implemented"); + } + final boolean responseOk = Arrays.stream(predicates).allMatch(p -> p.test(response)); + if (!responseOk) { + LOG.error(serializeResponse(response.extract())); + Assertions.fail(errorMessage); + } + + return response; + } + + String formatCurlConnectionInfo(int port, String url) { + List builder = new LinkedList<>(); + builder.add("curl -k"); // trust self-signed certs + builder.add(parameters.formatCurlAuthentication()); + builder.add(formatUrl(port, url)); + return String.join(" ", builder); + } + + String serializeResponse(ExtractableResponse extract) { + final String body = extract.body().asString(); + final String contentType = extract.contentType(); + final String headers = extract.headers().toString(); + return String.format(Locale.ROOT, "\nContent-type: %s\nheaders: %s\nbody: %s", contentType, headers, body); + } + + String formatUrl(int port, String url) { + return "https://localhost:" + port + url; + } +} diff --git a/data-node/src/test/java/org/graylog/datanode/restoperations/RestOperationParameters.java b/data-node/src/test/java/org/graylog/datanode/restoperations/RestOperationParameters.java new file mode 100644 index 0000000000..15c7c94e3c --- /dev/null +++ b/data-node/src/test/java/org/graylog/datanode/restoperations/RestOperationParameters.java @@ -0,0 +1,87 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.restoperations; + +import com.google.auto.value.AutoValue; +import io.restassured.specification.RequestSpecification; + +import javax.annotation.Nullable; +import java.security.KeyStore; + +@AutoValue +public abstract class RestOperationParameters { + + abstract int port(); + + abstract KeyStore truststore(); + + @Nullable + abstract String username(); + + @Nullable + abstract String password(); + + @Nullable + abstract String jwtToken(); + + + private static final int DEFAULT_ATTEMPTS_COUNT = 160; + + abstract int attempts_count(); + + public void addAuthorizationHeaders(RequestSpecification req) { + if (jwtToken() != null) { + req.header("Authorization", "Bearer " + jwtToken()); + } else if (username() != null && password() != null) { + req.auth().basic(username(), password()); + } + } + + public String formatCurlAuthentication() { + if (jwtToken() != null) { + return "-H \"Authorization: Bearer " + jwtToken() + "\""; + } else if (username() != null && password() != null) { + return "-u \"" + username() + ":" + password() + "\""; + } + return ""; + } + + public static Builder builder() { + return new AutoValue_RestOperationParameters.Builder() + .attempts_count(DEFAULT_ATTEMPTS_COUNT); + } + + abstract Builder toBuilder(); + + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder port(int port); + + public abstract Builder truststore(KeyStore truststore); + + public abstract Builder username(String username); + + public abstract Builder password(String password); + + public abstract Builder jwtToken(String jwtToken); + + + public abstract Builder attempts_count(int attempts_count); + + public abstract RestOperationParameters build(); + } +} diff --git a/data-node/src/test/java/org/graylog/datanode/DatanodeRestApiWait.java b/data-node/src/test/java/org/graylog/datanode/restoperations/WaitingRestOperation.java similarity index 54% rename from data-node/src/test/java/org/graylog/datanode/DatanodeRestApiWait.java rename to data-node/src/test/java/org/graylog/datanode/restoperations/WaitingRestOperation.java index 23c63092cc..bdc62925e9 100644 --- a/data-node/src/test/java/org/graylog/datanode/DatanodeRestApiWait.java +++ b/data-node/src/test/java/org/graylog/datanode/restoperations/WaitingRestOperation.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog.datanode; +package org.graylog.datanode.restoperations; import com.github.rholder.retry.Attempt; import com.github.rholder.retry.RetryException; @@ -25,74 +25,36 @@ import com.github.rholder.retry.StopStrategies; import com.github.rholder.retry.WaitStrategies; import io.restassured.RestAssured; import io.restassured.http.ContentType; -import io.restassured.response.ExtractableResponse; -import io.restassured.response.Response; import io.restassured.response.ValidatableResponse; import io.restassured.specification.RequestSpecification; import org.apache.http.NoHttpResponseException; -import org.jetbrains.annotations.NotNull; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import javax.net.ssl.SSLHandshakeException; import java.net.SocketException; -import java.security.KeyStore; -import java.util.LinkedList; -import java.util.List; -import java.util.Locale; import java.util.Optional; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.function.Predicate; -public class DatanodeRestApiWait { +public abstract class WaitingRestOperation extends RestOperation { - private static final Logger LOG = LoggerFactory.getLogger(DatanodeRestApiWait.class); - private static final int ATTEMPTS_COUNT = 160; - - private final int datanodePort; - private KeyStore truststore; - private String restApiUsername; - private String restApiPassword; private String lastRecordedResponse; - private DatanodeRestApiWait(int opensearchPort) { - this.datanodePort = opensearchPort; + protected WaitingRestOperation(RestOperationParameters parameters) { + super(parameters); } - public static DatanodeRestApiWait onPort(int datanodePort) { - return new DatanodeRestApiWait(datanodePort); - } - - public DatanodeRestApiWait withTruststore(KeyStore truststore) { - this.truststore = truststore; - return this; - } - - public DatanodeRestApiWait withBasicAuth(String username, String password) { - this.restApiUsername = username; - this.restApiPassword = password; - return this; - } - - public ValidatableResponse waitForAvailableStatus() throws ExecutionException, RetryException { - return waitForDatanode( - input -> !input.extract().body().path("opensearch.node.state").equals("AVAILABLE") - ); - } - - @SafeVarargs - private ValidatableResponse waitForDatanode(Predicate... predicates) throws ExecutionException, RetryException { + final ValidatableResponse waitForResponse(String url, Predicate... predicates) throws ExecutionException, RetryException { - if(LOG.isInfoEnabled()) { - LOG.info("cURL request: " + formatCurlConnectionInfo()); + if (LOG.isInfoEnabled()) { + LOG.info("cURL request: " + formatCurlConnectionInfo(parameters.port(), url)); } //noinspection UnstableApiUsage final RetryerBuilder builder = RetryerBuilder.newBuilder() .withWaitStrategy(WaitStrategies.fixedWait(1, TimeUnit.SECONDS)) - .withStopStrategy(StopStrategies.stopAfterAttempt(ATTEMPTS_COUNT)) + .withStopStrategy(StopStrategies.stopAfterAttempt(parameters.attempts_count())) .retryIfException(input -> input instanceof NoHttpResponseException) .retryIfException(input -> input instanceof SocketException) .retryIfException(input -> input instanceof SSLHandshakeException) // may happen before SSL is configured properly @@ -119,42 +81,19 @@ public class DatanodeRestApiWait { final RequestSpecification req = RestAssured.given() .accept(ContentType.JSON); - Optional.ofNullable(truststore).ifPresent(ts -> req.trustStore(truststore)); + Optional.ofNullable(parameters.truststore()).ifPresent(ts -> req.trustStore(parameters.truststore())); - if (restApiUsername != null && restApiPassword != null) { - req.auth().basic(restApiUsername, restApiPassword); - } + parameters.addAuthorizationHeaders(req); - return req.get(formatDatanodeRestUrl()) + return req.get(formatUrl(parameters.port(), url)) .then(); }); } catch (Exception e) { if (lastRecordedResponse != null) { - LOG.warn("Last recorded opensearch response, waiting for {}: {}", formatDatanodeRestUrl(), lastRecordedResponse); + LOG.warn("Last recorded opensearch response, waiting for {}: {}", formatUrl(parameters.port(), url), lastRecordedResponse); } throw e; } } - private String formatCurlConnectionInfo() { - List builder = new LinkedList<>(); - builder.add("curl -k"); // trust self-signed certs - if(restApiUsername != null && restApiPassword != null) { - builder.add("-u \"" + restApiUsername + ":" + restApiPassword + "\""); - } - builder.add(formatDatanodeRestUrl()); - return String.join(" ", builder); - } - - @NotNull - private String formatDatanodeRestUrl() { - return "https://localhost:" + datanodePort; - } - - private String serializeResponse(ExtractableResponse extract) { - final String body = extract.body().asString(); - final String contentType = extract.contentType(); - final String headers = extract.headers().toString(); - return String.format(Locale.ROOT, "\nContent-type: %s\nheaders: %s\nbody: %s", contentType, headers, body); - } } diff --git a/data-node/src/test/java/org/graylog/datanode/testinfra/DatanodeContainerizedBackend.java b/data-node/src/test/java/org/graylog/datanode/testinfra/DatanodeContainerizedBackend.java index 97b64bc26c..e967953c3d 100644 --- a/data-node/src/test/java/org/graylog/datanode/testinfra/DatanodeContainerizedBackend.java +++ b/data-node/src/test/java/org/graylog/datanode/testinfra/DatanodeContainerizedBackend.java @@ -37,7 +37,7 @@ public class DatanodeContainerizedBackend { private final MongoDBTestService mongoDBTestService; private boolean shouldCloseMongodb = false; private final GenericContainer datanodeContainer; - + private String nodeName; public DatanodeContainerizedBackend() { this(new DatanodeDockerHooksAdapter()); @@ -71,6 +71,7 @@ public class DatanodeContainerizedBackend { } private GenericContainer createDatanodeContainer(String nodeName, DatanodeDockerHooks customizer) { + this.nodeName = nodeName; MavenPackager.packageJarIfNecessary(new DefaultMavenProjectDirProvider()); return new DatanodeDevContainerBuilder() @@ -125,4 +126,8 @@ public class DatanodeContainerizedBackend { public Integer getOpensearchTransportPort() { return datanodeContainer.getMappedPort(DATANODE_OPENSEARCH_TRANSPORT_PORT); } + + public String getNodeName() { + return nodeName; + } } diff --git a/graylog2-server/src/main/java/org/graylog/plugins/datanode/DataNodeModule.java b/graylog2-server/src/main/java/org/graylog/plugins/datanode/DataNodeModule.java new file mode 100644 index 0000000000..428c573bcc --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog/plugins/datanode/DataNodeModule.java @@ -0,0 +1,32 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.plugins.datanode; + +import org.graylog2.datanode.DataNodeService; +import org.graylog2.datanode.DataNodeServiceImpl; +import org.graylog2.migrations.V20231107164300_CreateDataNodeManagerRole; +import org.graylog2.plugin.PluginModule; + +public class DataNodeModule extends PluginModule { + + @Override + protected void configure() { + bind(DataNodeService.class).to(DataNodeServiceImpl.class); + addMigration(V20231107164300_CreateDataNodeManagerRole.class); + } + +} diff --git a/graylog2-server/src/main/java/org/graylog2/audit/AuditEventTypes.java b/graylog2-server/src/main/java/org/graylog2/audit/AuditEventTypes.java index ac7ef8e12a..24332c281c 100644 --- a/graylog2-server/src/main/java/org/graylog2/audit/AuditEventTypes.java +++ b/graylog2-server/src/main/java/org/graylog2/audit/AuditEventTypes.java @@ -52,6 +52,8 @@ public class AuditEventTypes implements PluginAuditEventTypes { public static final String DASHBOARD_WIDGET_POSITIONS_UPDATE = PREFIX + "dashboard_widget_positions:update"; public static final String DASHBOARD_WIDGET_UPDATE = PREFIX + "dashboard_widget:update"; public static final String DATANODE_API_REQUEST = PREFIX + "data_node:api_request"; + public static final String DATANODE_REMOVE = PREFIX + "data_node:remove"; + public static final String DATANODE_RESET = PREFIX + "data_node:reset"; public static final String ES_INDEX_CLOSE = PREFIX + "es_index:close"; public static final String ES_INDEX_CREATE = PREFIX + "es_index:create"; public static final String ES_INDEX_DELETE = PREFIX + "es_index:delete"; @@ -187,6 +189,9 @@ public class AuditEventTypes implements PluginAuditEventTypes { .add(DASHBOARD_WIDGET_POSITIONS_UPDATE) .add(DASHBOARD_WIDGET_UPDATE) .add(DATANODE_API_REQUEST) + .add(DATANODE_REMOVE) + .add(DATANODE_RESET) + .add(DATANODE_API_REQUEST) .add(ES_INDEX_CLOSE) .add(ES_INDEX_CREATE) .add(ES_INDEX_DELETE) diff --git a/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeLifecycleEvent.java b/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeLifecycleEvent.java new file mode 100644 index 0000000000..6b4e0f36db --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeLifecycleEvent.java @@ -0,0 +1,45 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog2.datanode; + +import com.fasterxml.jackson.annotation.JsonAutoDetect; +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.auto.value.AutoValue; +import org.graylog.autovalue.WithBeanGetter; + +import javax.validation.Valid; + +@AutoValue +@WithBeanGetter +@JsonAutoDetect +public abstract class DataNodeLifecycleEvent { + + @JsonProperty("node_id") + public abstract String nodeId(); + + @JsonProperty + public abstract DataNodeLifecycleTrigger trigger(); + + @JsonCreator + public static DataNodeLifecycleEvent create( + @JsonProperty("node_id") @Valid String nodeId, + @JsonProperty("trigger") @Valid DataNodeLifecycleTrigger trigger + ) { + return new AutoValue_DataNodeLifecycleEvent(nodeId, trigger); + } +} diff --git a/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeLifecycleTrigger.java b/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeLifecycleTrigger.java new file mode 100644 index 0000000000..e4fef847dc --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeLifecycleTrigger.java @@ -0,0 +1,21 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog2.datanode; + +public enum DataNodeLifecycleTrigger { + REMOVE, RESET +} diff --git a/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeService.java b/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeService.java new file mode 100644 index 0000000000..84d892df02 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeService.java @@ -0,0 +1,27 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog2.datanode; + +import com.google.inject.ImplementedBy; +import org.graylog2.cluster.NodeNotFoundException; + +@ImplementedBy(DataNodeServiceImpl.class) +public interface DataNodeService { + void removeNode(String nodeId) throws NodeNotFoundException; + + void resetNode(String nodeId) throws NodeNotFoundException; +} diff --git a/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeServiceImpl.java b/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeServiceImpl.java new file mode 100644 index 0000000000..64fac741f9 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeServiceImpl.java @@ -0,0 +1,55 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog2.datanode; + +import org.graylog2.cluster.Node; +import org.graylog2.cluster.NodeNotFoundException; +import org.graylog2.cluster.NodeService; +import org.graylog2.events.ClusterEventBus; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.inject.Inject; + +public class DataNodeServiceImpl implements DataNodeService { + + private static final Logger LOG = LoggerFactory.getLogger(DataNodeServiceImpl.class); + + private final ClusterEventBus clusterEventBus; + private final NodeService nodeService; + + @Inject + public DataNodeServiceImpl(ClusterEventBus clusterEventBus, NodeService nodeService) { + this.clusterEventBus = clusterEventBus; + this.nodeService = nodeService; + } + + @Override + public void removeNode(String nodeId) throws NodeNotFoundException { + final Node node = nodeService.byNodeId(nodeId); + DataNodeLifecycleEvent e = DataNodeLifecycleEvent.create(node.getNodeId(), DataNodeLifecycleTrigger.REMOVE); + clusterEventBus.post(e); + } + + @Override + public void resetNode(String nodeId) throws NodeNotFoundException { + final Node node = nodeService.byNodeId(nodeId); + DataNodeLifecycleEvent e = DataNodeLifecycleEvent.create(node.getNodeId(), DataNodeLifecycleTrigger.RESET); + clusterEventBus.post(e); + } + +} diff --git a/graylog2-server/src/main/java/org/graylog2/migrations/V20231107164300_CreateDataNodeManagerRole.java b/graylog2-server/src/main/java/org/graylog2/migrations/V20231107164300_CreateDataNodeManagerRole.java new file mode 100644 index 0000000000..860253f865 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/migrations/V20231107164300_CreateDataNodeManagerRole.java @@ -0,0 +1,43 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog2.migrations; + +import org.graylog2.shared.security.RestPermissions; + +import javax.inject.Inject; +import java.time.ZonedDateTime; +import java.util.Set; + +public class V20231107164300_CreateDataNodeManagerRole extends Migration { + private final MigrationHelpers helpers; + + @Inject + public V20231107164300_CreateDataNodeManagerRole(MigrationHelpers migrationHelpers) { + this.helpers = migrationHelpers; + } + + @Override + public ZonedDateTime createdAt() { + return ZonedDateTime.parse("2023-09-05T08:14:00Z"); + } + + @Override + public void upgrade() { + helpers.ensureBuiltinRole("Data Node Manager", "Grants control to manage the data node cluster (built-in)", + Set.of(RestPermissions.DATANODE_REMOVE, RestPermissions.DATANODE_RESET)); + } +} diff --git a/graylog2-server/src/main/java/org/graylog2/rest/resources/RestResourcesModule.java b/graylog2-server/src/main/java/org/graylog2/rest/resources/RestResourcesModule.java index 44b860503c..8bd4d8c2b0 100644 --- a/graylog2-server/src/main/java/org/graylog2/rest/resources/RestResourcesModule.java +++ b/graylog2-server/src/main/java/org/graylog2/rest/resources/RestResourcesModule.java @@ -34,6 +34,7 @@ import org.graylog2.rest.resources.cluster.ClusterSystemProcessingResource; import org.graylog2.rest.resources.cluster.ClusterSystemResource; import org.graylog2.rest.resources.cluster.ClusterSystemShutdownResource; import org.graylog2.rest.resources.datanodes.DataNodeApiProxyResource; +import org.graylog2.rest.resources.datanodes.DataNodeManagementResource; import org.graylog2.rest.resources.entities.preferences.EntityListPreferencesResource; import org.graylog2.rest.resources.messages.MessageResource; import org.graylog2.rest.resources.roles.RolesResource; @@ -148,6 +149,7 @@ public class RestResourcesModule extends Graylog2Module { addSystemRestResource(ContentStreamResource.class); addSystemRestResource(CertificateRenewalResource.class); addSystemRestResource(DataNodeApiProxyResource.class); + addSystemRestResource(DataNodeManagementResource.class); } private void addDebugResources() { diff --git a/graylog2-server/src/main/java/org/graylog2/rest/resources/datanodes/DataNodeApiProxyResource.java b/graylog2-server/src/main/java/org/graylog2/rest/resources/datanodes/DataNodeApiProxyResource.java index 085f018a73..2d40d5e39c 100644 --- a/graylog2-server/src/main/java/org/graylog2/rest/resources/datanodes/DataNodeApiProxyResource.java +++ b/graylog2-server/src/main/java/org/graylog2/rest/resources/datanodes/DataNodeApiProxyResource.java @@ -54,6 +54,7 @@ import static org.graylog2.audit.AuditEventTypes.DATANODE_API_REQUEST; @RequiresPermissions("*") public class DataNodeApiProxyResource extends RestResource { private static final List> allowList = List.of( + request -> request.path().startsWith("_cluster"), request -> request.path().startsWith("_cat"), request -> request.path().startsWith("_mapping") && request.method().equals("GET") ); diff --git a/graylog2-server/src/main/java/org/graylog2/rest/resources/datanodes/DataNodeManagementResource.java b/graylog2-server/src/main/java/org/graylog2/rest/resources/datanodes/DataNodeManagementResource.java new file mode 100644 index 0000000000..41b0ae0a95 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/rest/resources/datanodes/DataNodeManagementResource.java @@ -0,0 +1,81 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog2.rest.resources.datanodes; + +import io.swagger.annotations.Api; +import io.swagger.annotations.ApiOperation; +import io.swagger.annotations.ApiParam; +import org.apache.shiro.authz.annotation.RequiresAuthentication; +import org.apache.shiro.authz.annotation.RequiresPermissions; +import org.graylog2.audit.jersey.AuditEvent; +import org.graylog2.cluster.NodeNotFoundException; +import org.graylog2.datanode.DataNodeService; +import org.graylog2.shared.rest.resources.RestResource; +import org.graylog2.shared.security.RestPermissions; + +import javax.inject.Inject; +import javax.ws.rs.DELETE; +import javax.ws.rs.NotFoundException; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.ws.rs.PathParam; +import javax.ws.rs.Produces; +import javax.ws.rs.core.MediaType; + +import static org.graylog2.audit.AuditEventTypes.DATANODE_REMOVE; +import static org.graylog2.audit.AuditEventTypes.DATANODE_RESET; +import static org.graylog2.shared.rest.documentation.generator.Generator.CLOUD_VISIBLE; + +@RequiresAuthentication +@Api(value = "Datanode", description = "Data Node", tags = {CLOUD_VISIBLE}) +@Path("/datanode/") +@Produces(MediaType.APPLICATION_JSON) +public class DataNodeManagementResource extends RestResource { + + private final DataNodeService dataNodeService; + + @Inject + protected DataNodeManagementResource(DataNodeService dataNodeService) { + this.dataNodeService = dataNodeService; + } + + @DELETE + @Path("{nodeId}") + @ApiOperation("Remove node from cluster") + @AuditEvent(type = DATANODE_REMOVE) + @RequiresPermissions(RestPermissions.DATANODE_REMOVE) + public void removeNode(@ApiParam(name = "nodeId", required = true) @PathParam("nodeId") String nodeId) { + try { + dataNodeService.removeNode(nodeId); + } catch (NodeNotFoundException e) { + throw new NotFoundException("Node " + nodeId + " not found"); + } + } + + @POST + @Path("{nodeId}/reset") + @ApiOperation("Reset a removed node to rejoin the cluster") + @AuditEvent(type = DATANODE_RESET) + @RequiresPermissions(RestPermissions.DATANODE_RESET) + public void resetNode(@ApiParam(name = "nodeId", required = true) @PathParam("nodeId") String nodeId) { + try { + dataNodeService.resetNode(nodeId); + } catch (NodeNotFoundException e) { + throw new NotFoundException("Node " + nodeId + " not found"); + } + } +} diff --git a/graylog2-server/src/main/java/org/graylog2/shared/security/RestPermissions.java b/graylog2-server/src/main/java/org/graylog2/shared/security/RestPermissions.java index 99cf2c53ad..ed6afa9a49 100644 --- a/graylog2-server/src/main/java/org/graylog2/shared/security/RestPermissions.java +++ b/graylog2-server/src/main/java/org/graylog2/shared/security/RestPermissions.java @@ -162,6 +162,9 @@ public class RestPermissions implements PluginPermissions { public static final String USERS_TOKENLIST = "users:tokenlist"; public static final String USERS_TOKENREMOVE = "users:tokenremove"; + public static final String DATANODE_REMOVE = "datanode:remove"; + public static final String DATANODE_RESET = "datanode:reset"; + // This is a special permission that ONLY works with GRNs as ID/target // TODO does this belong here? public static final String ENTITY_OWN = "entity:own"; @@ -189,6 +192,8 @@ public class RestPermissions implements PluginPermissions { .add(create(DASHBOARDS_CREATE, "")) .add(create(DASHBOARDS_EDIT, "")) .add(create(DASHBOARDS_READ, "")) + .add(create(DATANODE_REMOVE, "")) + .add(create(DATANODE_RESET, "")) .add(create(DECORATORS_CREATE, "")) .add(create(DECORATORS_EDIT, "")) .add(create(DECORATORS_READ, "")) diff --git a/graylog2-server/src/test/java/org/graylog2/datanode/DataNodeServiceImplTest.java b/graylog2-server/src/test/java/org/graylog2/datanode/DataNodeServiceImplTest.java new file mode 100644 index 0000000000..f79ffc2437 --- /dev/null +++ b/graylog2-server/src/test/java/org/graylog2/datanode/DataNodeServiceImplTest.java @@ -0,0 +1,66 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog2.datanode; + +import org.graylog2.cluster.Node; +import org.graylog2.cluster.NodeNotFoundException; +import org.graylog2.cluster.NodeService; +import org.graylog2.events.ClusterEventBus; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class DataNodeServiceImplTest { + + @Mock + private ClusterEventBus clusterEventBus; + @Mock + private NodeService nodeService; + + private DataNodeServiceImpl classUnderTest; + + @Before + public void setUp() { + this.classUnderTest = new DataNodeServiceImpl(clusterEventBus, nodeService); + } + + @Test + public void removeNodePublishesClusterEvent() throws NodeNotFoundException { + Node node = mock(Node.class); + when(node.getNodeId()).thenReturn("returnedid"); // would be the same in real life + when(nodeService.byNodeId("nodeid")).thenReturn(node); + classUnderTest.removeNode("nodeid"); + verify(clusterEventBus).post(DataNodeLifecycleEvent.create(node.getNodeId(), DataNodeLifecycleTrigger.REMOVE)); + } + + @Test + public void resetNodePublishesClusterEvent() throws NodeNotFoundException { + Node node = mock(Node.class); + when(node.getNodeId()).thenReturn("returnedid"); // would be the same in real life + when(nodeService.byNodeId("nodeid")).thenReturn(node); + classUnderTest.resetNode("nodeid"); + verify(clusterEventBus).post(DataNodeLifecycleEvent.create(node.getNodeId(), DataNodeLifecycleTrigger.RESET)); + } + +} diff --git a/graylog2-server/src/test/java/org/graylog2/rest/resources/datanodes/DataNodeManagementResourceTest.java b/graylog2-server/src/test/java/org/graylog2/rest/resources/datanodes/DataNodeManagementResourceTest.java new file mode 100644 index 0000000000..37b7a1f8e3 --- /dev/null +++ b/graylog2-server/src/test/java/org/graylog2/rest/resources/datanodes/DataNodeManagementResourceTest.java @@ -0,0 +1,75 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog2.rest.resources.datanodes; + +import org.graylog2.cluster.NodeNotFoundException; +import org.graylog2.datanode.DataNodeService; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import javax.ws.rs.NotFoundException; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.verify; + +@RunWith(MockitoJUnitRunner.class) +public class DataNodeManagementResourceTest { + + public static final String NODEID = "nodeid"; + private DataNodeManagementResource classUnderTest; + + @Mock + private DataNodeService dataNodeService; + + @Before + public void setUp() { + classUnderTest = new DataNodeManagementResource(dataNodeService); + } + + @Test + public void removeUnavailableNode_throwsNotFoundException() throws NodeNotFoundException { + doThrow(NodeNotFoundException.class).when(dataNodeService).removeNode(NODEID); + Exception e = assertThrows(NotFoundException.class, () -> classUnderTest.removeNode(NODEID)); + assertEquals("Node " + NODEID + " not found", e.getMessage()); + } + + @Test + public void resetUnavailableNode_throwsNotFoundException() throws NodeNotFoundException { + doThrow(NodeNotFoundException.class).when(dataNodeService).resetNode(NODEID); + Exception e = assertThrows(NotFoundException.class, () -> classUnderTest.resetNode(NODEID)); + assertEquals("Node " + NODEID + " not found", e.getMessage()); + } + + @Test + public void verifyRemoveServiceCalled() throws NodeNotFoundException { + classUnderTest.removeNode(NODEID); + verify(dataNodeService).removeNode(NODEID); + } + + @Test + public void verifyResetServiceCalled() throws NodeNotFoundException { + classUnderTest.resetNode(NODEID); + verify(dataNodeService).resetNode(NODEID); + } + + +}