diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java index a8bae488c0..03250e3cb4 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java +++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java @@ -26,6 +26,7 @@ import org.graylog2.cluster.nodes.DataNodeDto; import org.graylog2.cluster.nodes.NodeService; import org.graylog2.cluster.preflight.DataNodeProvisioningStateChangeEvent; import org.graylog2.datanode.DataNodeLifecycleEvent; +import org.graylog2.datanode.RemoteReindexAllowlistEvent; import org.graylog2.plugin.system.NodeId; import org.graylog2.security.CustomCAX509TrustManager; import org.slf4j.Logger; @@ -34,6 +35,8 @@ import org.slf4j.LoggerFactory; import javax.inject.Inject; import javax.inject.Provider; import javax.inject.Singleton; +import java.util.HashMap; +import java.util.Map; @Singleton public class OpensearchProcessService extends AbstractIdleService implements Provider { @@ -70,6 +73,16 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro return process; } + @Subscribe + @SuppressWarnings("unused") + public void handleRemoteReindexAllowlistEvent(RemoteReindexAllowlistEvent event) { + switch (event.action()) { + case ADD -> + startWithConfig(Map.of("reindex.remote.whitelist", event.host())); // , "action.auto_create_index", "false")); + case REMOVE -> startWithConfig(); + } + } + @Subscribe @SuppressWarnings("unused") public void handlePreflightConfigEvent(DataNodeProvisioningStateChangeEvent event) { @@ -97,19 +110,42 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro } private void startWithConfig() { - final OpensearchConfiguration config = configurationProvider.get(); + this.startWithConfig(Map.of()); + } + + private void startWithConfig(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(), + finalAdditionalConfig); + if (config.securityConfigured()) { this.process.startWithConfig(config); } else { String noConfigMessage = """ - \n - ======================================================================================================== - It seems you are starting Data node for the first time. The current configuration is not sufficient to - start the indexer process because a security configuration is missing. You have to either provide http - and transport SSL certificates or use the Graylog preflight interface to configure this Data node remotely. - ======================================================================================================== - """; + \n + ======================================================================================================== + It seems you are starting Data node for the first time. The current configuration is not sufficient to + start the indexer process because a security configuration is missing. You have to either provide http + and transport SSL certificates or use the Graylog preflight interface to configure this Data node remotely. + ======================================================================================================== + """; LOG.info(noConfigMessage); } eventBus.post(new OpensearchConfigurationChangeEvent(config)); diff --git a/full-backend-tests/src/test/java/org/graylog/datanode/RemoteReindexingMigrationIT.java b/full-backend-tests/src/test/java/org/graylog/datanode/RemoteReindexingMigrationIT.java new file mode 100644 index 0000000000..8ffce30149 --- /dev/null +++ b/full-backend-tests/src/test/java/org/graylog/datanode/RemoteReindexingMigrationIT.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; + +import com.github.rholder.retry.RetryException; +import com.github.rholder.retry.RetryerBuilder; +import com.github.rholder.retry.StopStrategies; +import com.github.rholder.retry.WaitStrategies; +import io.restassured.response.ValidatableResponse; +import org.apache.commons.lang.RandomStringUtils; +import org.assertj.core.api.Assertions; +import org.graylog.shaded.opensearch2.org.opensearch.action.index.IndexRequest; +import org.graylog.shaded.opensearch2.org.opensearch.action.index.IndexResponse; +import org.graylog.shaded.opensearch2.org.opensearch.client.indices.CreateIndexRequest; +import org.graylog.shaded.opensearch2.org.opensearch.client.indices.CreateIndexResponse; +import org.graylog.storage.opensearch2.testing.OpenSearchInstance; +import org.graylog.storage.opensearch2.testing.OpenSearchInstanceBuilder; +import org.graylog.testing.completebackend.apis.GraylogApis; +import org.graylog.testing.containermatrix.SearchServer; +import org.graylog.testing.containermatrix.annotations.ContainerMatrixTest; +import org.graylog.testing.containermatrix.annotations.ContainerMatrixTestsConfiguration; +import org.hamcrest.Matchers; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +@ContainerMatrixTestsConfiguration(searchVersions = SearchServer.DATANODE_DEV) +public class RemoteReindexingMigrationIT { + + private static final Logger LOG = LoggerFactory.getLogger(RemoteReindexingMigrationIT.class); + + private GraylogApis apis; + private OpenSearchInstance openSearchInstance; + + @BeforeEach + void setUp(GraylogApis apis) { + this.apis = apis; + openSearchInstance = OpenSearchInstanceBuilder.builder() + .network(apis.backend().network()) + .hostname("existing-opensearch-cluster") + .build(); + } + + @AfterEach + void tearDown() { + this.openSearchInstance.close(); + } + + @ContainerMatrixTest + void testRemoteReindexing() throws ExecutionException, RetryException { + + final String indexName = RandomStringUtils.randomAlphanumeric(15).toLowerCase(Locale.ROOT); + + final String messageContent = RandomStringUtils.randomAlphanumeric(20); + + final CreateIndexResponse indexCreationResponse = openSearchInstance.openSearchClient().execute((restHighLevelClient, requestOptions) -> restHighLevelClient.indices().create(new CreateIndexRequest(indexName), requestOptions)); + LOG.info("Index {} created in old Opensearch cluster", indexCreationResponse.index()); + + final IndexResponse response = openSearchInstance.openSearchClient().execute((restHighLevelClient, requestOptions) -> { + final IndexRequest req = new IndexRequest(); + req.index(indexName); + req.source(Map.of("message", messageContent)); + return restHighLevelClient.index(req, requestOptions); + }); + + + LOG.info("Document indexed: " + response); + + // flush the newly created document + openSearchInstance.client().refreshNode(); + + final String request = """ + { + "hostname": "%s", + "indices": ["%s"] + } + """.formatted(openSearchInstance.internalUri(), indexName); + + + LOG.info("Requesting remote reindex: " + request); + + final ValidatableResponse migrationResponse = apis.post("/migration/remoteReindex", request, 200); + + // one document migrated + migrationResponse.assertThat().body("results.created", Matchers.hasSize(1)); + migrationResponse.assertThat().body("results.created", Matchers.contains(1)); + + /* + + ValidatableResponse status = RetryerBuilder.newBuilder() + .withWaitStrategy(WaitStrategies.fixedWait(1, TimeUnit.SECONDS)) + .withStopStrategy(StopStrategies.stopAfterAttempt(120)) + .retryIfResult(r -> !r.extract().body().asString().equals("FINISHED")) + .withRetryListener(new RetryListener() { + @Override + public void onRetry(Attempt attempt) { + if (attempt.hasResult()) { + final String status = ((ValidatableResponse) attempt.getResult()).extract().body().asString(); + LOG.info("Current reindex status: " + status); + } + } + }).build() + .call(() -> apis.get("/migration/remoteReindex", 200)); + + Assertions.assertThat(status).isEqualTo("FINISHED"); + + */ + + final Optional> transferedMessage = RetryerBuilder.>>newBuilder() + .withWaitStrategy(WaitStrategies.fixedWait(1, TimeUnit.SECONDS)) + .withStopStrategy(StopStrategies.stopAfterAttempt(120)) + .retryIfResult(Optional::isEmpty) + .build().call(() -> apis.backend().searchServerInstance().client().findMessage(indexName, "message:" + messageContent)); + + Assertions.assertThat(transferedMessage) + .isPresent() + .hasValueSatisfying(message -> Assertions.assertThat(message.get("message")).isEqualTo(messageContent)); + } +} diff --git a/graylog-storage-elasticsearch7/src/test/java/org/graylog/storage/elasticsearch7/testing/ClientES7.java b/graylog-storage-elasticsearch7/src/test/java/org/graylog/storage/elasticsearch7/testing/ClientES7.java index 8ed13c342e..2e91776272 100644 --- a/graylog-storage-elasticsearch7/src/test/java/org/graylog/storage/elasticsearch7/testing/ClientES7.java +++ b/graylog-storage-elasticsearch7/src/test/java/org/graylog/storage/elasticsearch7/testing/ClientES7.java @@ -36,6 +36,8 @@ import org.graylog.shaded.elasticsearch7.org.elasticsearch.action.admin.indices. import org.graylog.shaded.elasticsearch7.org.elasticsearch.action.admin.indices.template.delete.DeleteIndexTemplateRequest; import org.graylog.shaded.elasticsearch7.org.elasticsearch.action.bulk.BulkRequest; import org.graylog.shaded.elasticsearch7.org.elasticsearch.action.index.IndexRequest; +import org.graylog.shaded.elasticsearch7.org.elasticsearch.action.search.SearchRequest; +import org.graylog.shaded.elasticsearch7.org.elasticsearch.action.search.SearchResponse; import org.graylog.shaded.elasticsearch7.org.elasticsearch.action.support.WriteRequest; import org.graylog.shaded.elasticsearch7.org.elasticsearch.client.Request; import org.graylog.shaded.elasticsearch7.org.elasticsearch.client.Response; @@ -56,6 +58,9 @@ import org.graylog.shaded.elasticsearch7.org.elasticsearch.cluster.metadata.Comp import org.graylog.shaded.elasticsearch7.org.elasticsearch.common.compress.CompressedXContent; import org.graylog.shaded.elasticsearch7.org.elasticsearch.common.settings.Settings; import org.graylog.shaded.elasticsearch7.org.elasticsearch.common.xcontent.XContentType; +import org.graylog.shaded.elasticsearch7.org.elasticsearch.index.query.QueryBuilders; +import org.graylog.shaded.elasticsearch7.org.elasticsearch.search.SearchHit; +import org.graylog.shaded.elasticsearch7.org.elasticsearch.search.builder.SearchSourceBuilder; import org.graylog.storage.elasticsearch7.ElasticsearchClient; import org.graylog.testing.elasticsearch.BulkIndexRequest; import org.graylog.testing.elasticsearch.Client; @@ -65,9 +70,11 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.Callable; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -87,6 +94,19 @@ public class ClientES7 implements Client { this.objectMapper = new ObjectMapperProvider().get(); } + public Optional> findMessage(String index, String query) { + return client.execute((restHighLevelClient, requestOptions) -> { + final SearchRequest searchRequest = new SearchRequest(); + searchRequest.indices(index); + final SearchSourceBuilder source = new SearchSourceBuilder(); + source.query(QueryBuilders.queryStringQuery(query)); + source.size(1); + searchRequest.source(source); + final SearchResponse response = restHighLevelClient.search(searchRequest, requestOptions); + return Arrays.stream(response.getHits().getHits()).map(SearchHit::getSourceAsMap).findFirst(); + }); + } + @Override public void createIndex(String index, int shards, int replicas) { LOG.debug("Creating index " + index); diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearch2Module.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearch2Module.java index b4163744c8..73decfbe02 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearch2Module.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearch2Module.java @@ -32,6 +32,7 @@ import org.graylog2.indexer.cluster.ClusterAdapter; import org.graylog2.indexer.cluster.NodeAdapter; import org.graylog2.indexer.counts.CountsAdapter; import org.graylog2.indexer.datanode.ProxyRequestAdapter; +import org.graylog2.indexer.datanode.RemoteReindexingMigrationAdapter; import org.graylog2.indexer.fieldtypes.IndexFieldTypePollerAdapter; import org.graylog2.indexer.fieldtypes.streamfiltered.esadapters.StreamsForFieldRetriever; import org.graylog2.indexer.indices.IndicesAdapter; @@ -77,6 +78,7 @@ public class OpenSearch2Module extends VersionAwareModule { bindForSupportedVersion(QuerySuggestionsService.class).to(QuerySuggestionsOS2.class); bindForSupportedVersion(ProxyRequestAdapter.class).to(ProxyRequestAdapterOS2.class); + bindForSupportedVersion(RemoteReindexingMigrationAdapter.class).to(RemoteReindexingMigrationAdapterOS2.class); install(new FactoryModuleBuilder().build(ScrollResultOS2.Factory.class)); 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 new file mode 100644 index 0000000000..c8fdfbde59 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RemoteReindexingMigrationAdapterOS2.java @@ -0,0 +1,201 @@ +/* + * 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.storage.opensearch2; + +import com.github.rholder.retry.RetryException; +import com.github.rholder.retry.RetryerBuilder; +import com.github.rholder.retry.StopStrategies; +import com.github.rholder.retry.WaitStrategies; +import okhttp3.Credentials; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import org.graylog.plugins.views.storage.migration.ReindexResult; +import org.graylog.plugins.views.storage.migration.RemoteReindexResult; +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; +import org.graylog.shaded.opensearch2.org.opensearch.core.xcontent.XContentBuilder; +import org.graylog.shaded.opensearch2.org.opensearch.index.reindex.ReindexRequest; +import org.graylog.shaded.opensearch2.org.opensearch.index.reindex.RemoteInfo; +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; +import org.graylog2.indexer.datanode.RemoteReindexingMigrationAdapter; +import org.graylog2.indexer.indices.Indices; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.inject.Inject; +import java.io.BufferedReader; +import java.io.IOException; +import java.io.StringReader; +import java.net.MalformedURLException; +import java.net.URI; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import static org.graylog.shaded.opensearch2.org.opensearch.index.query.QueryBuilders.matchAllQuery; + +public class RemoteReindexingMigrationAdapterOS2 implements RemoteReindexingMigrationAdapter { + private static final Logger LOG = LoggerFactory.getLogger(RemoteReindexingMigrationAdapterOS2.class); + + private static final int CONNECTION_ATTEMPTS = 40; + private static final int WAIT_BETWEEN_CONNECTION_ATTEMPTS = 3; + + private final OpenSearchClient client; + private final OkHttpClient httpClient; + private final NodeService nodeService; + private final Indices indices; + private final IndexSetRegistry indexSetRegistry; + private final ClusterEventBus eventBus; + + @Inject + public RemoteReindexingMigrationAdapterOS2(final OpenSearchClient client, + final OkHttpClient httpClient, + final NodeService nodeService, + final Indices indices, + final IndexSetRegistry indexSetRegistry, + final ClusterEventBus eventBus) { + this.client = client; + this.httpClient = httpClient; + this.nodeService = nodeService; + this.indices = indices; + this.indexSetRegistry = indexSetRegistry; + this.eventBus = eventBus; + } + + @Override + public RemoteReindexResult start(final URI uri, final String username, final String password, final List indices) { + try { + final var toReindex = isAllIndices(indices) ? getAllIndicesFrom(uri, username, password) : indices; + + final var activeNodes = getActiveNodes(); + + allowReindexingFrom(uri.getHost() + ":" + uri.getPort()); + + waitForClusterRestart(activeNodes); + + return reindex(uri, username, password, toReindex); +// removeAllowlist(uri.getHost() + ":" + uri.getPort()); + } catch (MalformedURLException e) { + throw new RuntimeException(e); + } + } + + @Override + public Status status() { + // TODO: return real migration status + return Status.STARTING; + } + + Set getActiveNodes() { + 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()); + } + + void waitForClusterRestart(final Set expectedNodes) { + // sleeping for some time to let the cluster stop so we can wait for the restart + try { + Thread.sleep(10 * 1000); + } catch (InterruptedException e) { + LOG.warn("Could not sleep..."); + } + + final var retryer = RetryerBuilder.newBuilder() + .withWaitStrategy(WaitStrategies.fixedWait(WAIT_BETWEEN_CONNECTION_ATTEMPTS, TimeUnit.SECONDS)) + .withStopStrategy(StopStrategies.stopAfterAttempt(CONNECTION_ATTEMPTS)) + .retryIfResult(response -> !response) + .retryIfException() + .build(); + + final Callable callable = () -> getActiveNodes().containsAll(expectedNodes); + + try { + var successful = retryer.call(callable); + if (!successful) { + LOG.error("Cluster failed to restart after " + CONNECTION_ATTEMPTS * WAIT_BETWEEN_CONNECTION_ATTEMPTS + " seconds."); + } + } catch (ExecutionException | RetryException e) { + LOG.error("Cluster failed to restart: " + e.getMessage(), e); + } + + } + + void removeAllowlist(final String host) { + eventBus.post(new RemoteReindexAllowlistEvent(host, RemoteReindexAllowlistEvent.ACTION.REMOVE)); + } + + void allowReindexingFrom(final String host) { + eventBus.post(new RemoteReindexAllowlistEvent(host, RemoteReindexAllowlistEvent.ACTION.ADD)); + } + + List getAllIndicesFrom(final URI uri, final String username, final String password) throws MalformedURLException { + final var host = uri.toURL().toString(); + var url = (host.endsWith("/") ? host : host + "/") + "_cat/indices?h=index"; + try (var response = httpClient.newCall(new Request.Builder().url(url).header("Authorization", Credentials.basic(username, password)).build()).execute()) { + if (response.isSuccessful() && response.body() != null) { + return new BufferedReader(new StringReader(response.body().string())).lines().toList(); + } else { + throw new RuntimeException("Could not read list of indices from " + host); + } + } catch (IOException e) { + throw new RuntimeException("Could not read list of indices from " + host + ", " + e.getMessage(), e); + } + } + + boolean isAllIndices(final List indices) { + return indices == null || indices.isEmpty() || (indices.size() == 1 && "*".equals(indices.get(0))); + } + + RemoteReindexResult reindex(final URI uri, final String username, final String password, final List indices) { + List results = new ArrayList<>(); + try (XContentBuilder builder = JsonXContent.contentBuilder().prettyPrint()) { + BytesReference query = BytesReference.bytes(matchAllQuery().toXContent(builder, ToXContent.EMPTY_PARAMS)); + + + for (var index : indices) { + if(this.indices.exists(index)) { + throw new IllegalArgumentException("Can't migrate index " + index + ", as it already exists in the target indexer"); + } + this.indices.create(index, indexSetRegistry.getForIndex(index).orElse(indexSetRegistry.getDefault())); + + final ReindexRequest reindexRequest = new ReindexRequest(); + reindexRequest + .setRemoteInfo(new RemoteInfo(uri.getScheme(), uri.getHost(), uri.getPort(), uri.getPath(), query, username, password, Map.of(), RemoteInfo.DEFAULT_SOCKET_TIMEOUT, RemoteInfo.DEFAULT_CONNECT_TIMEOUT)) + .setSourceIndices(index).setDestIndex(index); + + var reindexResult = client.execute((c, requestOptions) -> c.reindex(reindexRequest, requestOptions)); + results.add(new ReindexResult(index, reindexResult.getTook().timeUnit(), reindexResult.getCreated(), reindexResult.getBatches())); + } + } catch (IOException e) { + throw new RuntimeException(e); + } + + return new RemoteReindexResult(results); + } +} diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/ClientOS2.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/ClientOS2.java index 39c9f43a11..60654c5441 100644 --- a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/ClientOS2.java +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/ClientOS2.java @@ -35,6 +35,8 @@ import org.graylog.shaded.opensearch2.org.opensearch.action.admin.indices.settin import org.graylog.shaded.opensearch2.org.opensearch.action.admin.indices.template.delete.DeleteIndexTemplateRequest; import org.graylog.shaded.opensearch2.org.opensearch.action.bulk.BulkRequest; import org.graylog.shaded.opensearch2.org.opensearch.action.index.IndexRequest; +import org.graylog.shaded.opensearch2.org.opensearch.action.search.SearchRequest; +import org.graylog.shaded.opensearch2.org.opensearch.action.search.SearchResponse; import org.graylog.shaded.opensearch2.org.opensearch.action.support.WriteRequest; import org.graylog.shaded.opensearch2.org.opensearch.client.Request; import org.graylog.shaded.opensearch2.org.opensearch.client.Response; @@ -54,6 +56,8 @@ import org.graylog.shaded.opensearch2.org.opensearch.cluster.health.ClusterHealt import org.graylog.shaded.opensearch2.org.opensearch.cluster.metadata.ComposableIndexTemplate; import org.graylog.shaded.opensearch2.org.opensearch.common.compress.CompressedXContent; import org.graylog.shaded.opensearch2.org.opensearch.common.settings.Settings; +import org.graylog.shaded.opensearch2.org.opensearch.index.query.QueryBuilders; +import org.graylog.shaded.opensearch2.org.opensearch.search.builder.SearchSourceBuilder; import org.graylog.storage.opensearch2.OpenSearchClient; import org.graylog.testing.elasticsearch.BulkIndexRequest; import org.graylog.testing.elasticsearch.Client; @@ -63,9 +67,11 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.Callable; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -114,6 +120,19 @@ public class ClientOS2 implements Client { client.execute((c, requestOptions) -> c.indices().close(closeIndexRequest, requestOptions)); } + public Optional> findMessage(String index, String queryString) { + return client.execute((restHighLevelClient, requestOptions) -> { + final SearchRequest searchRequest = new SearchRequest(); + searchRequest.indices(index); + final SearchSourceBuilder source = new SearchSourceBuilder(); + source.query(QueryBuilders.queryStringQuery(queryString)); + source.size(1); + searchRequest.source(source); + final SearchResponse response = restHighLevelClient.search(searchRequest, requestOptions); + return Arrays.stream(response.getHits().getHits()).map(r -> r.getSourceAsMap()).findFirst(); + }); + } + @Override public boolean indicesExists(String... indices) { final GetIndexRequest getIndexRequest = new GetIndexRequest(indices); diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/ReindexResult.java b/graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/ReindexResult.java new file mode 100644 index 0000000000..beac0e953d --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/ReindexResult.java @@ -0,0 +1,23 @@ +/* + * 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.views.storage.migration; + +import java.util.concurrent.TimeUnit; + +public record ReindexResult(String index, TimeUnit took, + long created, int batches) { +} diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/RemoteReindexRequest.java b/graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/RemoteReindexRequest.java new file mode 100644 index 0000000000..66e8cf6137 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/RemoteReindexRequest.java @@ -0,0 +1,23 @@ +/* + * 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.views.storage.migration; + +import java.net.URI; +import java.util.List; + +public record RemoteReindexRequest(URI hostname, String user, String password, List indices) { +} diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/RemoteReindexResource.java b/graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/RemoteReindexResource.java new file mode 100644 index 0000000000..b9e7bdf608 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/RemoteReindexResource.java @@ -0,0 +1,69 @@ +/* + * 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.views.storage.migration; + +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.NoAuditEvent; +import org.graylog2.indexer.datanode.RemoteReindexingMigrationAdapter; +import org.graylog2.shared.security.RestPermissions; + +import javax.inject.Inject; +import javax.validation.Valid; +import javax.validation.constraints.NotNull; +import javax.ws.rs.Consumes; +import javax.ws.rs.GET; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.ws.rs.Produces; +import javax.ws.rs.core.MediaType; + +@Path("/migration") +@RequiresAuthentication +@Consumes(MediaType.APPLICATION_JSON) +@Produces(MediaType.APPLICATION_JSON) +@Api(value = "Migration", description = "Migrate data from existing cluster") +public class RemoteReindexResource { + private final RemoteReindexingMigrationAdapter migrationService; + + @Inject + public RemoteReindexResource(RemoteReindexingMigrationAdapter migrationService) { + this.migrationService = migrationService; + } + + @POST + @Path("/remoteReindex") + @NoAuditEvent("No Audit Event needed") + @RequiresPermissions(RestPermissions.DATANODE_MIGRATION) + @ApiOperation(value = "by remote reindexing", notes = "configure the host/credentials you want to use to migrate data from") + public RemoteReindexResult migrate(@ApiParam(name = "remote configuration") @NotNull @Valid RemoteReindexRequest request) { + return migrationService.start(request.hostname(), request.user(), request.password(), request.indices()); + } + + @GET + @Path("/remoteReindex") + @NoAuditEvent("No Audit Event needed") + @RequiresPermissions(RestPermissions.DATANODE_MIGRATION) + @ApiOperation(value = "status", notes = "status for a running migration") + public RemoteReindexingMigrationAdapter.Status status() { + return migrationService.status(); + } + +} diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/RemoteReindexResult.java b/graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/RemoteReindexResult.java new file mode 100644 index 0000000000..a1133ea16a --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/storage/migration/RemoteReindexResult.java @@ -0,0 +1,22 @@ +/* + * 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.views.storage.migration; + +import java.util.List; + +public record RemoteReindexResult(List results) { +} diff --git a/graylog2-server/src/main/java/org/graylog2/datanode/RemoteReindexAllowlistEvent.java b/graylog2-server/src/main/java/org/graylog2/datanode/RemoteReindexAllowlistEvent.java new file mode 100644 index 0000000000..4962dd8a20 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/datanode/RemoteReindexAllowlistEvent.java @@ -0,0 +1,23 @@ +/* + * 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 record RemoteReindexAllowlistEvent(String host, ACTION action) { + public enum ACTION { + ADD, REMOVE + } +} 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 new file mode 100644 index 0000000000..3f7aaad173 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/indexer/datanode/RemoteReindexingMigrationAdapter.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.graylog2.indexer.datanode; + +import org.graylog.plugins.views.storage.migration.RemoteReindexResult; + +import java.net.URI; +import java.util.List; + +public interface RemoteReindexingMigrationAdapter { + enum Status { + STARTING, RUNNING, ERROR, FINISHED + } + + RemoteReindexResult start(URI uri, String username, String password, List indices); + + Status status(); +} 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 59ff2cb715..3b8961bf0a 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 @@ -16,6 +16,7 @@ */ package org.graylog2.rest.resources; +import org.graylog.plugins.views.storage.migration.RemoteReindexResource; import org.graylog.plugins.views.search.engine.monitoring.data.histogram.rest.HistogramResponseWriter; import org.graylog2.Configuration; import org.graylog2.bootstrap.preflight.web.resources.CAResource; @@ -159,11 +160,12 @@ public class RestResourcesModule extends Graylog2Module { addSystemRestResource(DataNodeApiProxyResource.class); addSystemRestResource(DataNodeRestApiProxyResource.class); addSystemRestResource(DataNodeManagementResource.class); + addSystemRestResource(RemoteReindexResource.class); addSystemRestResource(CAResource.class); } private void addDebugResources() { - if(Boolean.parseBoolean(System.getenv("GRAYLOG_ENABLE_DEBUG_RESOURCES"))) { + if (Boolean.parseBoolean(System.getenv("GRAYLOG_ENABLE_DEBUG_RESOURCES"))) { // TODO: move the DebugEventsResource under this env property check as well? addSystemRestResource(DebugStreamsResource.class); } 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 46ae4508f8..2bd19fe050 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 @@ -167,6 +167,7 @@ public class RestPermissions implements PluginPermissions { public static final String DATANODE_RESET = "datanode:reset"; public static final String DATANODE_STOP = "datanode:stop"; public static final String DATANODE_START = "datanode:start"; + public static final String DATANODE_MIGRATION = "datanode:migration"; // This is a special permission that ONLY works with GRNs as ID/target // TODO does this belong here? @@ -199,6 +200,7 @@ public class RestPermissions implements PluginPermissions { .add(create(DATANODE_RESET, "")) .add(create(DATANODE_STOP, "")) .add(create(DATANODE_START, "")) + .add(create(DATANODE_MIGRATION, "")) .add(create(DECORATORS_CREATE, "")) .add(create(DECORATORS_EDIT, "")) .add(create(DECORATORS_READ, "")) diff --git a/graylog2-server/src/main/java/org/graylog2/storage/VersionAwareStorageModule.java b/graylog2-server/src/main/java/org/graylog2/storage/VersionAwareStorageModule.java index e9d21415db..e61ec41830 100644 --- a/graylog2-server/src/main/java/org/graylog2/storage/VersionAwareStorageModule.java +++ b/graylog2-server/src/main/java/org/graylog2/storage/VersionAwareStorageModule.java @@ -32,6 +32,7 @@ import org.graylog2.indexer.cluster.ClusterAdapter; import org.graylog2.indexer.cluster.NodeAdapter; import org.graylog2.indexer.counts.CountsAdapter; import org.graylog2.indexer.datanode.ProxyRequestAdapter; +import org.graylog2.indexer.datanode.RemoteReindexingMigrationAdapter; import org.graylog2.indexer.fieldtypes.IndexFieldTypePollerAdapter; import org.graylog2.indexer.fieldtypes.streamfiltered.esadapters.StreamsForFieldRetriever; import org.graylog2.indexer.indices.IndicesAdapter; @@ -50,6 +51,7 @@ import org.graylog2.storage.providers.MoreSearchAdapterProvider; import org.graylog2.storage.providers.MultiChunkResultRetrieverProvider; import org.graylog2.storage.providers.NodeAdapterProvider; import org.graylog2.storage.providers.ProxyRequestAdapterProvider; +import org.graylog2.storage.providers.RemoteReindexingMigrationAdapterProvider; import org.graylog2.storage.providers.SearchesAdapterProvider; import org.graylog2.storage.providers.StreamsForFieldRetrieverProvider; import org.graylog2.storage.providers.V20170607164210_MigrateReopenedIndicesToAliasesClusterStateAdapterProvider; @@ -82,6 +84,7 @@ public class VersionAwareStorageModule extends AbstractModule { bind(V20200730000000_AddGl2MessageIdFieldAliasForEvents.ElasticsearchAdapter.class) .toProvider(V20200730000000_AddGl2MessageIdFieldAliasForEventsElasticsearchAdapterProvider.class); bind(ProxyRequestAdapter.class).toProvider(ProxyRequestAdapterProvider.class); + bind(RemoteReindexingMigrationAdapter.class).toProvider(RemoteReindexingMigrationAdapterProvider.class); bindQueryBackend(); } diff --git a/graylog2-server/src/main/java/org/graylog2/storage/providers/RemoteReindexingMigrationAdapterProvider.java b/graylog2-server/src/main/java/org/graylog2/storage/providers/RemoteReindexingMigrationAdapterProvider.java new file mode 100644 index 0000000000..e77d28c451 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/storage/providers/RemoteReindexingMigrationAdapterProvider.java @@ -0,0 +1,33 @@ +/* + * 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.storage.providers; + +import org.graylog2.indexer.datanode.RemoteReindexingMigrationAdapter; +import org.graylog2.storage.DetectedSearchVersion; +import org.graylog2.storage.SearchVersion; +import org.graylog2.storage.VersionAwareProvider; + +import javax.inject.Inject; +import javax.inject.Provider; +import java.util.Map; + +public class RemoteReindexingMigrationAdapterProvider extends VersionAwareProvider { + @Inject + public RemoteReindexingMigrationAdapterProvider(@DetectedSearchVersion SearchVersion indexerVersion, Map> pluginBindings) { + super(indexerVersion, pluginBindings); + } +} diff --git a/graylog2-server/src/test/java/org/graylog/testing/elasticsearch/Client.java b/graylog2-server/src/test/java/org/graylog/testing/elasticsearch/Client.java index 5c95511034..1695cba600 100644 --- a/graylog2-server/src/test/java/org/graylog/testing/elasticsearch/Client.java +++ b/graylog2-server/src/test/java/org/graylog/testing/elasticsearch/Client.java @@ -19,8 +19,12 @@ package org.graylog.testing.elasticsearch; import org.graylog2.indexer.indices.Template; import java.util.Map; +import java.util.Optional; public interface Client { + + Optional> findMessage(String index, String query); + default void createIndex(String index) { createIndex(index, 1, 0); }