From 1572342d600d1ad596fb95ae00624a845e649f3b Mon Sep 17 00:00:00 2001 From: Tomas Dvorak Date: Wed, 17 Apr 2024 14:39:39 +0200 Subject: [PATCH] Async cluster restart during remote reindex migration (#19080) * Async cluster restart during remote reindex migration, configuration cleanup * Safer and faster way to restart and wait for datanode cluster during remote reindex --------- Co-authored-by: Matthias Oesterheld <33032967+moesterheld@users.noreply.github.com> --- .../org/graylog/datanode/Configuration.java | 5 + .../OpensearchConfigurationProvider.java | 26 ++-- .../management/OpensearchProcessService.java | 45 ++----- .../RemoteReindexingMigrationAdapterOS2.java | 118 +++++++++--------- .../RemoteReindexingMigrationAdapter.java | 2 +- .../indexer/migration/RemoteReindexIndex.java | 6 +- .../migration/RemoteReindexMigration.java | 18 +-- .../migration/RemoteReindexMigrationTest.java | 41 +++++- 8 files changed, 142 insertions(+), 119 deletions(-) diff --git a/data-node/src/main/java/org/graylog/datanode/Configuration.java b/data-node/src/main/java/org/graylog/datanode/Configuration.java index d5c8f43280..1dd0f98fa8 100644 --- a/data-node/src/main/java/org/graylog/datanode/Configuration.java +++ b/data-node/src/main/java/org/graylog/datanode/Configuration.java @@ -280,6 +280,11 @@ public class Configuration extends BaseConfiguration { return opensearchAuditLog; } + /** + * The insecure flag causes problems on many places. We should replace it with autosecurity option, that would + * configure all the CA and certs automatically. + */ + @Deprecated public boolean isInsecureStartup() { return insecureStartup; } diff --git a/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java b/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java index 0aa83925af..611967cb72 100644 --- a/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java +++ b/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java @@ -17,6 +17,10 @@ package org.graylog.datanode.configuration; import com.google.common.collect.ImmutableMap; +import jakarta.inject.Inject; +import jakarta.inject.Named; +import jakarta.inject.Provider; +import jakarta.inject.Singleton; import org.graylog.datanode.Configuration; import org.graylog.datanode.configuration.variants.InSecureConfiguration; import org.graylog.datanode.configuration.variants.MongoCertSecureConfiguration; @@ -25,22 +29,17 @@ import org.graylog.datanode.configuration.variants.SecurityConfigurationVariant; import org.graylog.datanode.configuration.variants.UploadedCertFilesSecureConfiguration; import org.graylog.datanode.process.OpensearchConfiguration; import org.graylog.security.certutil.ca.exceptions.KeyStoreStorageException; -import org.graylog2.bootstrap.preflight.PreflightConfigResult; -import org.graylog2.bootstrap.preflight.PreflightConfigService; import org.graylog2.cluster.Node; import org.graylog2.cluster.nodes.DataNodeDto; import org.graylog2.cluster.nodes.NodeService; -import jakarta.inject.Inject; -import jakarta.inject.Named; -import jakarta.inject.Provider; -import jakarta.inject.Singleton; - import java.io.IOException; import java.nio.charset.StandardCharsets; import java.security.GeneralSecurityException; import java.util.List; +import java.util.Map; import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; @Singleton @@ -52,9 +51,13 @@ public class OpensearchConfigurationProvider implements Provider nodeService; - private final PreflightConfigService preflightConfigService; private final S3RepositoryConfiguration s3RepositoryConfiguration; + /** + * This configuration won't survive datanode restart. But it can be repeatedly provided to the managed opensearch + */ + private final Map transientConfiguration = new ConcurrentHashMap<>(); + @Inject public OpensearchConfigurationProvider(final Configuration localConfiguration, final DatanodeConfiguration datanodeConfiguration, @@ -62,7 +65,6 @@ public class OpensearchConfigurationProvider implements Provider nodeService, - final PreflightConfigService preflightConfigService, final @Named("password_secret") String passwordSecret, final S3RepositoryConfiguration s3RepositoryConfiguration) { this.localConfiguration = localConfiguration; @@ -72,7 +74,6 @@ public class OpensearchConfigurationProvider implements Provider { @@ -57,21 +53,20 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro private static final int WATCHDOG_RESTART_ATTEMPTS = 3; private final OpensearchProcess process; - private final Provider configurationProvider; + private final OpensearchConfigurationProvider configurationProvider; private final EventBus eventBus; private final NodeId nodeId; private final DataNodeProvisioningService dataNodeProvisioningService; private final IndexFieldTypesService indexFieldTypesService; private final ClusterEventBus clusterEventBus; private final DatanodeDirectoriesLockfileCheck lockfileCheck; - private final ClusterConfigService clusterConfigService; private final PreflightConfigService preflightConfigService; private final Configuration configuration; @Inject public OpensearchProcessService(final DatanodeConfiguration datanodeConfiguration, - final Provider configurationProvider, + final OpensearchConfigurationProvider configurationProvider, final EventBus eventBus, final CustomCAX509TrustManager trustManager, final NodeService nodeService, @@ -83,7 +78,6 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro final ProcessStateMachine processStateMachine, final ClusterEventBus clusterEventBus, final DatanodeDirectoriesLockfileCheck lockfileCheck, - final ClusterConfigService clusterConfigService, final PreflightConfigService preflightConfigService) { this.configurationProvider = configurationProvider; this.configuration = configuration; @@ -93,7 +87,6 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro this.indexFieldTypesService = indexFieldTypesService; this.clusterEventBus = clusterEventBus; this.lockfileCheck = lockfileCheck; - this.clusterConfigService = clusterConfigService; this.preflightConfigService = preflightConfigService; this.process = createOpensearchProcess(datanodeConfiguration, trustManager, configuration, nodeService, objectMapper, processStateMachine); eventBus.register(this); @@ -117,7 +110,8 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro switch (event.action()) { case ADD -> { this.process.stop(); - configure(Map.of("reindex.remote.allowlist", event.allowlist())); // , "action.auto_create_index", "false")); + this.configurationProvider.setTransientConfiguration("reindex.remote.allowlist", event.allowlist()); + configure(); // , "action.auto_create_index", "false")); this.process.start(); } case REMOVE -> { @@ -178,33 +172,8 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro } } - protected void configure() { - this.configure(Map.of()); - } - - private void configure(Map additionalConfig) { - final OpensearchConfiguration original = configurationProvider.get(); - - final var finalAdditionalConfig = new HashMap(); - finalAdditionalConfig.putAll(original.additionalConfiguration()); - finalAdditionalConfig.putAll(additionalConfig); - - final var config = new OpensearchConfiguration( - original.opensearchDistribution(), - original.datanodeDirectories(), - original.bindAddress(), - original.hostname(), - original.httpPort(), - original.transportPort(), - original.clusterName(), - original.nodeName(), - original.nodeRoles(), - original.discoverySeedHosts(), - original.opensearchSecurityConfiguration(), - original.s3RepositoryConfiguration(), - original.nodeSearchCacheSize(), - finalAdditionalConfig); - + private void configure() { + final OpensearchConfiguration config = configurationProvider.get(); if (config.securityConfigured()) { this.process.configure(config); } else { diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RemoteReindexingMigrationAdapterOS2.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RemoteReindexingMigrationAdapterOS2.java index 9646bc2dc3..9124d6984c 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RemoteReindexingMigrationAdapterOS2.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RemoteReindexingMigrationAdapterOS2.java @@ -17,7 +17,9 @@ package org.graylog.storage.opensearch2; import com.fasterxml.jackson.databind.ObjectMapper; +import com.github.rholder.retry.Attempt; import com.github.rholder.retry.RetryException; +import com.github.rholder.retry.RetryListener; import com.github.rholder.retry.RetryerBuilder; import com.github.rholder.retry.StopStrategies; import com.github.rholder.retry.WaitStrategies; @@ -30,11 +32,14 @@ import okhttp3.Credentials; import okhttp3.OkHttpClient; import okhttp3.Request; import org.apache.commons.lang.time.DurationFormatUtils; +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.client.RequestOptions; import org.graylog.shaded.opensearch2.org.opensearch.client.Response; import org.graylog.shaded.opensearch2.org.opensearch.client.tasks.TaskSubmissionResponse; +import org.graylog.shaded.opensearch2.org.opensearch.cluster.health.ClusterHealthStatus; import org.graylog.shaded.opensearch2.org.opensearch.common.xcontent.json.JsonXContent; import org.graylog.shaded.opensearch2.org.opensearch.core.common.bytes.BytesReference; import org.graylog.shaded.opensearch2.org.opensearch.core.xcontent.ToXContent; @@ -42,10 +47,6 @@ import org.graylog.shaded.opensearch2.org.opensearch.core.xcontent.XContentBuild import org.graylog.shaded.opensearch2.org.opensearch.index.reindex.ReindexRequest; import org.graylog.shaded.opensearch2.org.opensearch.index.reindex.RemoteInfo; import org.graylog.shaded.opensearch2.org.opensearch.tasks.Task; -import org.graylog2.cluster.nodes.DataNodeDto; -import org.graylog2.cluster.nodes.DataNodeStatus; -import org.graylog2.cluster.nodes.NodeDto; -import org.graylog2.cluster.nodes.NodeService; import org.graylog2.datanode.RemoteReindexAllowlistEvent; import org.graylog2.events.ClusterEventBus; import org.graylog2.indexer.IndexSetRegistry; @@ -69,6 +70,8 @@ import org.joda.time.Duration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import jakarta.annotation.Nonnull; + import javax.annotation.Nullable; import java.io.BufferedReader; import java.io.IOException; @@ -82,8 +85,6 @@ import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Optional; -import java.util.Set; -import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -102,7 +103,6 @@ public class RemoteReindexingMigrationAdapterOS2 implements RemoteReindexingMigr private final OpenSearchClient client; private final OkHttpClient httpClient; - private final NodeService nodeService; private final Indices indices; private final IndexSetRegistry indexSetRegistry; private final ClusterEventBus eventBus; @@ -113,14 +113,12 @@ public class RemoteReindexingMigrationAdapterOS2 implements RemoteReindexingMigr @Inject public RemoteReindexingMigrationAdapterOS2(final OpenSearchClient client, final OkHttpClient httpClient, - final NodeService nodeService, final Indices indices, final IndexSetRegistry indexSetRegistry, final ClusterEventBus eventBus, final ObjectMapper objectMapper, RemoteReindexMigrationService reindexMigrationService) { this.client = client; this.httpClient = httpClient; - this.nodeService = nodeService; this.indices = indices; this.indexSetRegistry = indexSetRegistry; this.eventBus = eventBus; @@ -145,9 +143,11 @@ public class RemoteReindexingMigrationAdapterOS2 implements RemoteReindexingMigr private void doStartMigration(MigrationConfiguration migration, RemoteReindexRequest request) { try { - prepareCluster(request.allowlist()); - createIndicesInNewCluster(migration); - startAsyncTasks(migration, request); + new Thread(() -> { + prepareCluster(request.allowlist(), migration); + createIndicesInNewCluster(migration); + startAsyncTasks(migration, request); + }).start(); } catch (Exception e) { LOG.error("Failed to start remote reindex migration", e); throw new RuntimeException(e); @@ -164,21 +164,18 @@ public class RemoteReindexingMigrationAdapterOS2 implements RemoteReindexingMigr }); } - private void prepareCluster(String allowlistAsString) { + private void prepareCluster(String allowlistAsString, MigrationConfiguration migration) { Preconditions.checkArgument(allowlistAsString != null, "Allowlist has to be provided in the request."); - final var activeNodes = getAllActiveNodeIDs(); List allowlist = Arrays.stream(allowlistAsString.split(",")).map(String::trim).toList(); - try { - verifyRemoteReindexAllowlistSetting(allowlist); - } catch (RemoteReindexNotAllowedException e) { + if (!isRemoteReindexAllowed(allowlist)) { // this is expected state for fresh datanode cluster - there is no value configured in the reindex.remote.allowlist // we have to add it to the configuration and wait till the whole cluster restarts + logInfo(migration, "Preparing cluster for remote reindexing, setting allowlist to: " + allowlistAsString); allowReindexingFrom(allowlist); - waitForClusterRestart(activeNodes); + waitForClusterRestart(allowlist, migration); + } else { + logInfo(migration, "Remote reindex allowlist already configured, skipping cluster configuration and restart."); } - - // verify again, just to be sure that all the configuration is in place and vali - verifyRemoteReindexAllowlistSetting(allowlist); } private ReindexRequest createReindexRequest(final String index, final BytesReference query, URI uri, String username, String password, MigrationConfiguration migration) { @@ -196,7 +193,7 @@ public class RemoteReindexingMigrationAdapterOS2 implements RemoteReindexingMigr final List indices = migrationConfiguration.indices() .parallelStream() .map(indexConfig -> indexConfig.taskId().flatMap(this::getTask).map(task -> taskToIndex(indexConfig.indexName(), task)) - .orElse(RemoteReindexIndex.notStartedYet(indexConfig.indexName()))) + .orElse(RemoteReindexIndex.noBackgroundTaskYet(indexConfig.indexName()))) .sorted(Comparator.comparing(RemoteReindexIndex::name)) .collect(Collectors.toList()); return new RemoteReindexMigration(migrationID, indices, migrationConfiguration.logs()); @@ -217,7 +214,7 @@ public class RemoteReindexingMigrationAdapterOS2 implements RemoteReindexingMigr return new RemoteReindexIndex(indexName, Status.FINISHED, created, duration, progress, null); } } else { - return new RemoteReindexIndex(indexName, Status.RUNNING, created, duration,progress, null); + return new RemoteReindexIndex(indexName, Status.RUNNING, created, duration, progress, null); } } @@ -245,63 +242,64 @@ public class RemoteReindexingMigrationAdapterOS2 implements RemoteReindexingMigr } } - public void verifyRemoteReindexAllowlistSetting(List allowlistEntries) throws RemoteReindexNotAllowedException { + private boolean isRemoteReindexAllowed(List allowlistEntries) { final String allowlistSettingValue = client.execute((restHighLevelClient, requestOptions) -> { final ClusterGetSettingsRequest request = new ClusterGetSettingsRequest(); request.includeDefaults(true); final ClusterGetSettingsResponse settings = restHighLevelClient.cluster().getSettings(request, requestOptions); return settings.getSetting("reindex.remote.allowlist"); }); - // the value is not proper json, just something like [localhost:9201]. It should be safe to simply use String.contains, // but there is maybe a chance for mismatches and then we'd have to parse the value - final boolean isRemoteReindexAllowed = !allowlistEntries.stream().allMatch(entry -> allowlistSettingValue.contains(entry)); - if (isRemoteReindexAllowed) { - final String message = "Failed to configure reindex.remote.allowlist setting in the datanode cluster. Current setting value: " + allowlistSettingValue; - LOG.error(message); - throw new RemoteReindexNotAllowedException(message); - } + return allowlistEntries.stream().allMatch(allowlistSettingValue::contains); } - private Set getAllActiveNodeIDs() { - return nodeService.allActive().values().stream() - .filter(dn -> dn.getDataNodeStatus() == DataNodeStatus.AVAILABLE) // we have to wait till the datanode is not just alive but all started properly and the indexer accepts connections - .map(NodeDto::getNodeId).collect(Collectors.toSet()); - } - - private void waitForClusterRestart(final Set expectedNodes) { - // We are currently unable to detect that datanodes stopped and are starting again. We just hope that - // these 10 seconds give them enough time and they will restart the opensearch process in the background - // after 10s we'll wait till all the previously known nodes are up and healthy again. - try { - Thread.sleep(10 * 1000); - } catch (InterruptedException e) { - LOG.warn("Could not sleep..."); - } - - final var retryer = RetryerBuilder.newBuilder() + private void waitForClusterRestart(List allowlist, MigrationConfiguration migration) { + final var retryer = RetryerBuilder.newBuilder() .withWaitStrategy(WaitStrategies.fixedWait(WAIT_BETWEEN_CONNECTION_ATTEMPTS, TimeUnit.SECONDS)) .withStopStrategy(StopStrategies.stopAfterAttempt(CONNECTION_ATTEMPTS)) - .retryIfResult(response -> !response) + .withRetryListener(createClusterRestartWaitListener(migration)) .retryIfException() + .retryIfResult(status -> !status.isClusterReady()) .build(); - - final Callable callable = () -> getAllActiveNodeIDs().containsAll(expectedNodes); - try { - var successful = retryer.call(callable); - if (!successful) { - final String message = "Cluster failed to restart after " + CONNECTION_ATTEMPTS * WAIT_BETWEEN_CONNECTION_ATTEMPTS + " seconds."; - LOG.error(message); - throw new IllegalStateException(message); - } + retryer.call(() -> remoteReindexClusterState(allowlist)); + logInfo(migration, "Datanode cluster successfully reconfigured and restarted."); } catch (ExecutionException | RetryException e) { - final String message = "Cluster failed to restart: " + e.getMessage(); - LOG.error(message, e); + final String message = "Cluster failed to restart after " + CONNECTION_ATTEMPTS * WAIT_BETWEEN_CONNECTION_ATTEMPTS + " seconds."; + logError(migration, message, e); throw new RuntimeException(message); } } + @Nonnull + private RetryListener createClusterRestartWaitListener(MigrationConfiguration migration) { + return new RetryListener() { + @Override + public void onRetry(Attempt attempt) { + if (attempt.hasResult()) { + final RemoteReindexConfigurationStatus status = (RemoteReindexConfigurationStatus) attempt.getResult(); + final String message = String.format(Locale.ROOT, "Waiting for datanode cluster to reconfigure and restart, attempt %d. Cluster health: %s, allowlist configured: %b.", attempt.getAttemptNumber(), status.status(), status.allowlistConfigured()); + logInfo(migration, message); + } else { + logInfo(migration, "Waiting for datanode cluster to reconfigure and restart, attempt #" + attempt.getAttemptNumber()); + } + } + }; + } + + private RemoteReindexConfigurationStatus remoteReindexClusterState(List allowlist) { + final ClusterHealthResponse clusterHealth = client.execute((restHighLevelClient, requestOptions) -> restHighLevelClient.cluster().health(new ClusterHealthRequest(), RequestOptions.DEFAULT)); + final boolean remoteReindexAllowed = isRemoteReindexAllowed(allowlist); + return new RemoteReindexConfigurationStatus(remoteReindexAllowed, clusterHealth.getStatus()); + } + + private record RemoteReindexConfigurationStatus(boolean allowlistConfigured, ClusterHealthStatus status) { + private boolean isClusterReady() { + return allowlistConfigured && status.equals(ClusterHealthStatus.GREEN); + } + } + void allowReindexingFrom(final List allowlist) { eventBus.post(new RemoteReindexAllowlistEvent(allowlist, RemoteReindexAllowlistEvent.ACTION.ADD)); } diff --git a/graylog2-server/src/main/java/org/graylog2/indexer/datanode/RemoteReindexingMigrationAdapter.java b/graylog2-server/src/main/java/org/graylog2/indexer/datanode/RemoteReindexingMigrationAdapter.java index 197380e2e5..4c748756e7 100644 --- a/graylog2-server/src/main/java/org/graylog2/indexer/datanode/RemoteReindexingMigrationAdapter.java +++ b/graylog2-server/src/main/java/org/graylog2/indexer/datanode/RemoteReindexingMigrationAdapter.java @@ -24,7 +24,7 @@ import java.net.URI; public interface RemoteReindexingMigrationAdapter { enum Status { - NOT_STARTED, STARTING, RUNNING, ERROR, FINISHED + NOT_STARTED, RUNNING, ERROR, FINISHED } /** diff --git a/graylog2-server/src/main/java/org/graylog2/indexer/migration/RemoteReindexIndex.java b/graylog2-server/src/main/java/org/graylog2/indexer/migration/RemoteReindexIndex.java index 12744e6c77..a2b0423575 100644 --- a/graylog2-server/src/main/java/org/graylog2/indexer/migration/RemoteReindexIndex.java +++ b/graylog2-server/src/main/java/org/graylog2/indexer/migration/RemoteReindexIndex.java @@ -26,8 +26,12 @@ public record RemoteReindexIndex(String name, Status status, DateTime created, D IndexMigrationProgress progress, String errorMsg) { - public static RemoteReindexIndex notStartedYet(String indexName) { + public static RemoteReindexIndex noBackgroundTaskYet(String indexName) { return new RemoteReindexIndex(indexName, Status.NOT_STARTED, null, null, null, null); } + public boolean isCompleted() { + return status() == Status.FINISHED || status == Status.ERROR; + } + } diff --git a/graylog2-server/src/main/java/org/graylog2/indexer/migration/RemoteReindexMigration.java b/graylog2-server/src/main/java/org/graylog2/indexer/migration/RemoteReindexMigration.java index c8cb42fa2f..23c72913ae 100644 --- a/graylog2-server/src/main/java/org/graylog2/indexer/migration/RemoteReindexMigration.java +++ b/graylog2-server/src/main/java/org/graylog2/indexer/migration/RemoteReindexMigration.java @@ -65,16 +65,18 @@ public class RemoteReindexMigration { @JsonProperty("status") public Status status() { - if (indices.isEmpty()) { + if (indices.isEmpty() || indices.stream().allMatch(i -> i.status() == Status.NOT_STARTED)) { return Status.NOT_STARTED; - } - if (indices().stream().map(RemoteReindexIndex::status).anyMatch(i -> i == Status.RUNNING)) { + } else if (indices.stream().allMatch(RemoteReindexIndex::isCompleted)) { + // all are now completed, either finished or errored + if (indices.stream().anyMatch(i -> i.status() == Status.ERROR)) { + return Status.ERROR; + } else { + return Status.FINISHED; + } + } else { return Status.RUNNING; } - if (indices().stream().map(RemoteReindexIndex::status).anyMatch(i -> i == Status.ERROR)) { - return Status.ERROR; - } - return Status.FINISHED; } /** @@ -94,7 +96,7 @@ public class RemoteReindexMigration { .map(relativeProgress -> relativeProgress * indexPortion) .sum(); - return Math.min((int)Math.ceil(overallProgress), 100); + return Math.min((int) Math.ceil(overallProgress), 100); } public List getLogs() { diff --git a/graylog2-server/src/test/java/org/graylog2/indexer/migration/RemoteReindexMigrationTest.java b/graylog2-server/src/test/java/org/graylog2/indexer/migration/RemoteReindexMigrationTest.java index 02ee0eba64..c2e04f2f47 100644 --- a/graylog2-server/src/test/java/org/graylog2/indexer/migration/RemoteReindexMigrationTest.java +++ b/graylog2-server/src/test/java/org/graylog2/indexer/migration/RemoteReindexMigrationTest.java @@ -19,6 +19,7 @@ package org.graylog2.indexer.migration; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.jayway.jsonpath.JsonPath; +import jakarta.annotation.Nonnull; import org.assertj.core.api.Assertions; import org.graylog2.indexer.datanode.RemoteReindexingMigrationAdapter; import org.graylog2.shared.bindings.providers.ObjectMapperProvider; @@ -88,8 +89,46 @@ class RemoteReindexMigrationTest { Assertions.assertThat(progress).isEqualTo(50); } - @NotNull + @Nonnull private static RemoteReindexMigration withIndices(RemoteReindexIndex... indices) { return new RemoteReindexMigration(UUID.randomUUID().toString(), Arrays.asList(indices), Collections.emptyList()); } + + @Test + void testStatusNotStarted() { + final RemoteReindexMigration migration = withIndices( + index("one", RemoteReindexingMigrationAdapter.Status.NOT_STARTED), + index("two", RemoteReindexingMigrationAdapter.Status.NOT_STARTED) + ); + Assertions.assertThat(migration.status()).isEqualTo(RemoteReindexingMigrationAdapter.Status.NOT_STARTED); + } + + @Test + void testStatusCompletedWithError() { + final RemoteReindexMigration migration = withIndices( + index("one", RemoteReindexingMigrationAdapter.Status.FINISHED), + index("two", RemoteReindexingMigrationAdapter.Status.ERROR) + ); + Assertions.assertThat(migration.status()).isEqualTo(RemoteReindexingMigrationAdapter.Status.ERROR); + } + + + @Test + void testStatusRunningWithError() { + final RemoteReindexMigration migration = withIndices( + index("one", RemoteReindexingMigrationAdapter.Status.FINISHED), + index("two", RemoteReindexingMigrationAdapter.Status.ERROR), + index("three", RemoteReindexingMigrationAdapter.Status.RUNNING) + ); + Assertions.assertThat(migration.status()).isEqualTo(RemoteReindexingMigrationAdapter.Status.RUNNING); + } + + @Test + void testStatusFinished() { + final RemoteReindexMigration migration = withIndices( + index("one", RemoteReindexingMigrationAdapter.Status.FINISHED), + index("two", RemoteReindexingMigrationAdapter.Status.FINISHED) + ); + Assertions.assertThat(migration.status()).isEqualTo(RemoteReindexingMigrationAdapter.Status.FINISHED); + } }