diff --git a/data-node/pom.xml b/data-node/pom.xml index 06497c24a5..ef9b39ee37 100644 --- a/data-node/pom.xml +++ b/data-node/pom.xml @@ -63,6 +63,11 @@ graylog2-server ${project.version} + + org.graylog + graylog-storage-opensearch2 + ${project.version} + com.github.rvesse @@ -448,9 +453,18 @@ datanode + + src/main/resources + + **/*.properties + + src/main/resources true + + **/*.properties + @@ -590,7 +604,6 @@ - @@ -602,7 +615,6 @@ - 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 5f5be034c4..9ea01ba7f4 100644 --- a/data-node/src/main/java/org/graylog/datanode/Configuration.java +++ b/data-node/src/main/java/org/graylog/datanode/Configuration.java @@ -213,6 +213,26 @@ public class Configuration extends BaseConfiguration { @Parameter(value = "http_allow_embedding") private boolean httpAllowEmbedding = false; + @Parameter(value = "metrics_timestamp") + private String metricsTimestamp = "timestamp"; + + @Parameter(value = "metrics_template") + private String metricsTemplate = "gl-datanode-metrics-template"; + @Parameter(value = "metrics_stream") + private String metricsStream = "gl-datanode-metrics"; + + @Parameter(value = "metrics_retention") + private String metricsRetention = "14d"; + + @Parameter(value = "metrics_daily_retention") + private String metricsDailyRetention = "365d"; + + @Parameter(value = "metrics_daily_index") + private String metricsDailyIndex = "gl-datanode-metrics-daily"; + + @Parameter(value = "metrics_policy") + private String metricsPolicy = "gl-datanode-metrics-ism"; + public boolean isInsecureStartup() { return insecureStartup; } @@ -361,6 +381,35 @@ public class Configuration extends BaseConfiguration { return clustername; } + + public String getMetricsTimestamp() { + return metricsTimestamp; + } + + public String getMetricsTemplate() { + return metricsTemplate; + } + + public String getMetricsStream() { + return metricsStream; + } + + public String getMetricsRetention() { + return metricsRetention; + } + + public String getMetricsDailyRetention() { + return metricsDailyRetention; + } + + public String getMetricsDailyIndex() { + return metricsDailyIndex; + } + + public String getMetricsPolicy() { + return metricsPolicy; + } + public static class NodeIdFileValidator implements Validator { @Override public void validate(String name, String path) throws ValidationException { @@ -582,4 +631,5 @@ public class Configuration extends BaseConfiguration { public String getRootPasswordSha2() { return rootPasswordSha2; } + } diff --git a/data-node/src/main/java/org/graylog/datanode/bindings/PeriodicalBindings.java b/data-node/src/main/java/org/graylog/datanode/bindings/PeriodicalBindings.java index 3767a00feb..92cde4c866 100644 --- a/data-node/src/main/java/org/graylog/datanode/bindings/PeriodicalBindings.java +++ b/data-node/src/main/java/org/graylog/datanode/bindings/PeriodicalBindings.java @@ -20,6 +20,7 @@ import com.google.inject.AbstractModule; import com.google.inject.multibindings.Multibinder; import org.graylog.datanode.bootstrap.preflight.DataNodeConfigurationPeriodical; import org.graylog.datanode.periodicals.ClusterManagerDiscovery; +import org.graylog.datanode.periodicals.MetricsCollector; import org.graylog.datanode.periodicals.NodePingPeriodical; import org.graylog.datanode.periodicals.OpensearchNodeHeartbeat; import org.graylog2.events.ClusterEventCleanupPeriodical; @@ -37,5 +38,6 @@ public class PeriodicalBindings extends AbstractModule { // periodicalBinder.addBinding().to(UserSessionTerminationPeriodical.class); periodicalBinder.addBinding().to(NodePingPeriodical.class); periodicalBinder.addBinding().to(DataNodeConfigurationPeriodical.class); + periodicalBinder.addBinding().to(MetricsCollector.class); } } 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 5c4d5ee3be..691f552676 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 @@ -19,6 +19,7 @@ package org.graylog.datanode.management; import org.graylog.datanode.process.OpensearchConfiguration; import org.graylog.datanode.process.OpensearchInfo; import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; +import org.graylog.storage.opensearch2.OpenSearchClient; import java.net.URI; import java.util.List; @@ -30,6 +31,8 @@ public interface OpensearchProcess extends ManagableProcess restClient(); + Optional openSearchClient(); + boolean isLeaderNode(); void setLeaderNode(boolean isManagerNode); List stdOutLogs(); @@ -42,4 +45,5 @@ public interface OpensearchProcess extends ManagableProcess opensearchConfiguration = Optional.empty(); @SuppressWarnings("OptionalUsedAsFieldOrParameterType") private Optional restClient = Optional.empty(); + private Optional openSearchClient = Optional.empty(); private final StateMachine processState; @@ -73,9 +76,11 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { private final CustomCAX509TrustManager trustManager; private final NodeService nodeService; private final Configuration configuration; + private final ObjectMapper objectMapper; - OpensearchProcessImpl(DatanodeConfiguration datanodeConfiguration, int logsCacheSize, final CustomCAX509TrustManager trustManager, final Configuration configuration, final NodeService nodeService) { + OpensearchProcessImpl(DatanodeConfiguration datanodeConfiguration, int logsCacheSize, final CustomCAX509TrustManager trustManager, + final Configuration configuration, final NodeService nodeService, ObjectMapper objectMapper) { this.datanodeConfiguration = datanodeConfiguration; this.processState = ProcessStateMachine.createNew(); tracerAggregator = new StateMachineTracerAggregator(); @@ -85,6 +90,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { this.trustManager = trustManager; this.nodeService = nodeService; this.configuration = configuration; + this.objectMapper = objectMapper; } private RestHighLevelClient createRestClient(OpensearchConfiguration configuration) { @@ -105,6 +111,10 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { return restClient; } + public Optional openSearchClient() { + return openSearchClient; + } + public OpensearchInfo processInfo() { return new OpensearchInfo(configuration.getDatanodeNodeName(), processState.getState(), isLeaderNode, getOpensearchBaseUrl().toString(), commandLineProcess != null ? commandLineProcess.processInfo() : ProcessInformation.empty()); } @@ -185,6 +195,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { commandLineProcess = new OpensearchCommandLineProcess(config, this); commandLineProcess.start(); restClient = Optional.of(createRestClient(config)); + openSearchClient = restClient.map(c -> new OpenSearchClient(c, objectMapper)); }), () -> {throw new IllegalArgumentException("Opensearch configuration required but not supplied!");} ); 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 d25182ece6..83d02af957 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 @@ -16,6 +16,7 @@ */ package org.graylog.datanode.management; +import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.eventbus.EventBus; import com.google.common.eventbus.Subscribe; import com.google.common.util.concurrent.AbstractIdleService; @@ -24,6 +25,7 @@ import jakarta.inject.Provider; import jakarta.inject.Singleton; import org.graylog.datanode.Configuration; import org.graylog.datanode.configuration.DatanodeConfiguration; +import org.graylog.datanode.metrics.ConfigureMetricsIndexSettings; import org.graylog.datanode.process.OpensearchConfiguration; import org.graylog2.cluster.nodes.DataNodeDto; import org.graylog2.cluster.nodes.NodeService; @@ -48,6 +50,7 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro private final Provider configurationProvider; private final EventBus eventBus; private final NodeId nodeId; + private final ObjectMapper objectMapper; @Inject public OpensearchProcessService(final DatanodeConfiguration datanodeConfiguration, @@ -56,20 +59,23 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro final CustomCAX509TrustManager trustManager, final NodeService nodeService, final Configuration configuration, - final NodeId nodeId) { + final NodeId nodeId, ObjectMapper objectMapper) { this.configurationProvider = configurationProvider; this.eventBus = eventBus; this.nodeId = nodeId; - this.process = createOpensearchProcess(datanodeConfiguration, trustManager, configuration, nodeService); + this.objectMapper = objectMapper; + this.process = createOpensearchProcess(datanodeConfiguration, trustManager, configuration, nodeService, objectMapper); eventBus.register(this); } - private OpensearchProcess createOpensearchProcess(final DatanodeConfiguration datanodeConfiguration, final CustomCAX509TrustManager trustManager, final Configuration configuration, final NodeService nodeService) { - final OpensearchProcessImpl process = new OpensearchProcessImpl(datanodeConfiguration, datanodeConfiguration.processLogsBufferSize(), trustManager, configuration, nodeService); + private OpensearchProcess createOpensearchProcess(final DatanodeConfiguration datanodeConfiguration, final CustomCAX509TrustManager trustManager, final Configuration configuration, + final NodeService nodeService, final ObjectMapper objectMapper) { + final OpensearchProcessImpl process = new OpensearchProcessImpl(datanodeConfiguration, datanodeConfiguration.processLogsBufferSize(), trustManager, configuration, nodeService, objectMapper); final ProcessWatchdog watchdog = new ProcessWatchdog(process, WATCHDOG_RESTART_ATTEMPTS); process.addStateMachineTracer(watchdog); process.addStateMachineTracer(new StateMachineTransitionLogger()); process.addStateMachineTracer(new OpensearchRemovalTracer(process, configuration.getDatanodeNodeName())); + process.addStateMachineTracer(new ConfigureMetricsIndexSettings(process, configuration, objectMapper)); return process; } diff --git a/data-node/src/main/java/org/graylog/datanode/metrics/ConfigureMetricsIndexSettings.java b/data-node/src/main/java/org/graylog/datanode/metrics/ConfigureMetricsIndexSettings.java new file mode 100644 index 0000000000..543e5d7e3a --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/metrics/ConfigureMetricsIndexSettings.java @@ -0,0 +1,155 @@ +/* + * 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.metrics; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.graylog.datanode.Configuration; +import org.graylog.datanode.management.OpensearchProcess; +import org.graylog.datanode.process.ProcessEvent; +import org.graylog.datanode.process.ProcessState; +import org.graylog.datanode.process.StateMachineTracer; +import org.graylog.storage.opensearch2.DataStreamAdapterOS2; +import org.graylog.storage.opensearch2.ism.IsmApi; +import org.graylog.storage.opensearch2.ism.policy.IsmPolicy; +import org.graylog.storage.opensearch2.ism.policy.Policy; +import org.graylog.storage.opensearch2.ism.policy.actions.Action; +import org.graylog.storage.opensearch2.ism.policy.actions.DeleteAction; +import org.graylog.storage.opensearch2.ism.policy.actions.RolloverAction; +import org.graylog.storage.opensearch2.ism.policy.actions.RollupAction; +import org.graylog2.indexer.datastream.DataStreamAdapter; +import org.graylog2.indexer.indices.Template; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class ConfigureMetricsIndexSettings implements StateMachineTracer { + + private final Logger log = LoggerFactory.getLogger(ConfigureMetricsIndexSettings.class); + + private final OpensearchProcess process; + private final Configuration configuration; + private final ObjectMapper objectMapper; + private DataStreamAdapter dataStreamAdapter; + + public ConfigureMetricsIndexSettings(OpensearchProcess process, Configuration configuration, ObjectMapper objectMapper) { + this.process = process; + this.configuration = configuration; + this.objectMapper = objectMapper; + } + + @Override + public void trigger(ProcessEvent trigger) { + } + + @Override + public void transition(ProcessEvent trigger, ProcessState source, ProcessState destination) { + if (destination == ProcessState.AVAILABLE && source == ProcessState.STARTING) { + process.openSearchClient().ifPresent(client -> { + if (dataStreamAdapter == null) { + final IsmApi ismApi = new IsmApi(client, objectMapper); + dataStreamAdapter = new DataStreamAdapterOS2(client, objectMapper, ismApi); + } + updateDataStreamTemplate(); + dataStreamAdapter.createDataStream(configuration.getMetricsStream()); + configureMetricsIsm(configuration); + }); + + + } + } + + private void updateDataStreamTemplate() { + Map mappings = new HashMap<>(); + mappings.put(configuration.getMetricsTimestamp(), ImmutableMap.of( + "type", "date", + "format", "yyyy-MM-dd HH:mm:ss.SSS||strict_date_optional_time||epoch_millis") + ); + mappings.put("node", ImmutableMap.of("type", "keyword")); + Template template = new Template(List.of(configuration.getMetricsStream() + "*"), + new Template.Mappings(ImmutableMap.of("properties", mappings)), 99999L, new Template.Settings(Map.of())); + dataStreamAdapter.ensureDataStreamTemplate(configuration.getMetricsTemplate(), template, configuration.getMetricsTimestamp()); + } + + private void configureMetricsIsm(Configuration configuration) { + dataStreamAdapter.applyIsmPolicy(configuration.getMetricsStream(), + createPolicy(configuration)); + } + + private IsmPolicy createPolicy(Configuration configuration) { + // states defined from last to first to use name in previous step + Policy.State stateDelete = ismDeleteState(); + Policy.State stateRollup = ismRollupState(stateDelete.name(), configuration); + Policy.State stateOpen = ismOpenState(stateRollup.name()); + + Policy policy = new Policy(null, + "Manages rollover, rollup and deletion of data note metrics indices", + null, + stateOpen.name(), ImmutableList.of(stateOpen, stateRollup, stateDelete)); + + try { + log.debug("Creating ISM configuration for metrics data stream {}", + objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(policy)); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + + return new IsmPolicy(configuration.getMetricsPolicy(), policy); + } + + private Policy.State ismDeleteState() { + final List actions = ImmutableList.of(new Action(new DeleteAction())); + final List transitions = ImmutableList.of(); + return new Policy.State("delete", actions, transitions); + } + + private Policy.State ismRollupState(String nextState, Configuration configuration) { + + final RollupAction.IsmRollup ismRollup = new RollupAction.IsmRollup( + configuration.getMetricsDailyIndex(), + "Rollup Data Stream Index", + 1000, + ImmutableList.of( + new RollupAction.IsmRollup.DateHistogram( + configuration.getMetricsTimestamp(), + "60m", "UTC") + ), + ImmutableList.of( //TODO dynamically rollup all configured metrics + new RollupAction.IsmRollup.Metric( + "jvm_heap", + ImmutableList.of(new RollupAction.IsmRollup.AvgMetric()) + ) + ) + ); + RollupAction rollupAction = new RollupAction(ismRollup); + final List actions = ImmutableList.of(new Action(rollupAction)); + final List transitions = ImmutableList.of(new Policy.Transition(nextState, new Policy.Condition("13d"))); + return new Policy.State("rollup", actions, transitions); + } + + private Policy.State ismOpenState(String nextState) { + final List actions = ImmutableList.of(new Action(new RolloverAction("1d"))); + final List transitions = ImmutableList.of(new Policy.Transition(nextState, null)); + return new Policy.State("open", actions, transitions); + } + +} diff --git a/data-node/src/main/java/org/graylog/datanode/periodicals/MetricsCollector.java b/data-node/src/main/java/org/graylog/datanode/periodicals/MetricsCollector.java new file mode 100644 index 0000000000..dd273e86f8 --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/periodicals/MetricsCollector.java @@ -0,0 +1,111 @@ +/* + * 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.periodicals; + +import jakarta.inject.Inject; +import org.graylog.datanode.Configuration; +import org.graylog.datanode.management.OpensearchProcess; +import org.graylog.datanode.process.ProcessState; +import org.graylog.shaded.opensearch2.org.joda.time.DateTime; +import org.graylog.shaded.opensearch2.org.joda.time.DateTimeZone; +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.RequestOptions; +import org.graylog.shaded.opensearch2.org.opensearch.core.action.ActionListener; +import org.graylog2.plugin.periodical.Periodical; +import org.graylog2.plugin.system.NodeId; +import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.Map; + +public class MetricsCollector extends Periodical { + + private static final Logger LOG = LoggerFactory.getLogger(MetricsCollector.class); + private final OpensearchProcess process; + private final Configuration configuration; + private final NodeId nodeId; + + @Inject + public MetricsCollector(OpensearchProcess process, Configuration configuration, NodeId nodeId) { + this.process = process; + this.configuration = configuration; + this.nodeId = nodeId; + } + + @Override + public boolean runsForever() { + return false; + } + + @Override + public boolean stopOnGracefulShutdown() { + return false; + } + + @Override + public boolean startOnThisNode() { + return true; + } + + @Override + public boolean isDaemon() { + return true; + } + + @Override + public int getInitialDelaySeconds() { + return 0; + } + + @Override + public int getPeriodSeconds() { + return 60; + } + + @NotNull + @Override + protected Logger getLogger() { + return LOG; + } + + @Override + public void doRun() { + if (process.isInState(ProcessState.AVAILABLE)) { + process.restClient().ifPresent(client -> { + final IndexRequest indexRequest = new IndexRequest(configuration.getMetricsStream()); + Map metrics = new HashMap(); + metrics.put(configuration.getMetricsTimestamp(), new DateTime(DateTimeZone.UTC)); + metrics.put("node", configuration.getHostname()); + metrics.put("jvm_heap", Runtime.getRuntime().totalMemory()); + indexRequest.source(metrics); + client.indexAsync(indexRequest, RequestOptions.DEFAULT, new ActionListener() { + @Override + public void onResponse(IndexResponse indexResponse) { + } + + @Override + public void onFailure(Exception e) { + LOG.error("Error indexing metrics", e); + } + }); + }); + } + } +} diff --git a/data-node/src/main/resources/org/graylog/datanode/metrics/metrics-ism.json b/data-node/src/main/resources/org/graylog/datanode/metrics/metrics-ism.json new file mode 100644 index 0000000000..b0ee80afbd --- /dev/null +++ b/data-node/src/main/resources/org/graylog/datanode/metrics/metrics-ism.json @@ -0,0 +1,79 @@ +{ + "policy": { + "description": "Rollover and rollup index after one day, delete after 14 days", + "default_state": "open", + "states": [ + { + "name": "open", + "actions": [ + { + "rollover": { + "min_index_age": "1d" + } + } + ], + "transitions": [ + { + "state_name": "rollup" + } + ] + }, + { + "name": "rollup", + "actions": [ + { + "rollup": { + "ism_rollup": { + "target_index": "gl-datanode-metrics-daily", + "description": "Rollup index", + "page_size": 200, + "dimensions": [ + { + "date_histogram": { + "source_field": "timestamp", + "fixed_interval": "60m", + "timezone": "America/Los_Angeles" + } + } + ], + "metrics": [ + { + "source_field": "jvm_heap", + "metrics": [ + { + "avg": {} + } + ] + } + ] + } + } + } + ], + "transitions": [ + { + "state_name": "delete", + "conditions": { + "min_index_age": "13d" + } + } + ] + }, + { + "name": "delete", + "actions": [ + { + "delete": {} + } + ], + "transitions": [] + } + ], + "ism_template": { + "index_patterns": [ + "gl-datanode-metrics" + ], + "priority": 1 + } + } +} diff --git a/data-node/src/test/java/org/graylog/datanode/integration/DataNodePluginsIT.java b/data-node/src/test/java/org/graylog/datanode/integration/DataNodePluginsIT.java index 1eed74a16e..3f3a45ee21 100644 --- a/data-node/src/test/java/org/graylog/datanode/integration/DataNodePluginsIT.java +++ b/data-node/src/test/java/org/graylog/datanode/integration/DataNodePluginsIT.java @@ -53,7 +53,6 @@ public class DataNodePluginsIT { Matchers.not(Matchers.containsString("opensearch-alerting")), Matchers.not(Matchers.containsString("opensearch-custom-codecs")), Matchers.not(Matchers.containsString("opensearch-geospatial")), - Matchers.not(Matchers.containsString("opensearch-index-management")), Matchers.not(Matchers.containsString("opensearch-knn")), Matchers.not(Matchers.containsString("opensearch-neural-search")), Matchers.not(Matchers.containsString("opensearch-notifications")), diff --git a/graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/DataStreamAdapterES7.java b/graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/DataStreamAdapterES7.java new file mode 100644 index 0000000000..08975b6dbe --- /dev/null +++ b/graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/DataStreamAdapterES7.java @@ -0,0 +1,41 @@ +/* + * 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.elasticsearch7; + +import org.graylog2.indexer.datastream.DataStreamAdapter; +import org.graylog2.indexer.datastream.Policy; +import org.graylog2.indexer.indices.Template; +import org.jetbrains.annotations.NotNull; + +public class DataStreamAdapterES7 implements DataStreamAdapter { + private static final String ERROR_MESSAGE = "Data Streams not supported in Elastic Search"; + + @Override + public boolean ensureDataStreamTemplate(@NotNull String templateName, @NotNull Template template, @NotNull String timestampField) { + throw new UnsupportedOperationException(ERROR_MESSAGE); + } + + @Override + public void createDataStream(String dataStreamName) { + throw new UnsupportedOperationException(ERROR_MESSAGE); + } + + @Override + public void applyIsmPolicy(@NotNull String dataStreamName, @NotNull Policy policy) { + throw new UnsupportedOperationException(ERROR_MESSAGE); + } +} diff --git a/graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/Elasticsearch7Module.java b/graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/Elasticsearch7Module.java index 13b2517c73..d3087e09e2 100644 --- a/graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/Elasticsearch7Module.java +++ b/graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/Elasticsearch7Module.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.datastream.DataStreamAdapter; import org.graylog2.indexer.fieldtypes.IndexFieldTypePollerAdapter; import org.graylog2.indexer.fieldtypes.streamfiltered.esadapters.StreamsForFieldRetriever; import org.graylog2.indexer.indices.IndicesAdapter; @@ -57,6 +58,7 @@ public class Elasticsearch7Module extends VersionAwareModule { bindForSupportedVersion(CountsAdapter.class).to(CountsAdapterES7.class); bindForSupportedVersion(ClusterAdapter.class).to(ClusterAdapterES7.class); bindForSupportedVersion(IndicesAdapter.class).to(IndicesAdapterES7.class); + bindForSupportedVersion(DataStreamAdapter.class).to(DataStreamAdapterES7.class); if (useComposableIndexTemplates) { bind(IndexTemplateAdapter.class).to(ComposableIndexTemplateAdapter.class); } else { @@ -77,7 +79,7 @@ public class Elasticsearch7Module extends VersionAwareModule { bindForSupportedVersion(QuerySuggestionsService.class).to(QuerySuggestionsES7.class); bindForSupportedVersion(ProxyRequestAdapter.class).to(ProxyRequestAdapterES7.class); - + install(new FactoryModuleBuilder().build(ScrollResultES7.Factory.class)); bind(RestHighLevelClient.class).toProvider(RestHighLevelClientProvider.class); diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/DataStreamAdapterOS2.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/DataStreamAdapterOS2.java new file mode 100644 index 0000000000..1c46f8e2ed --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/DataStreamAdapterOS2.java @@ -0,0 +1,137 @@ +/* + * 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.fasterxml.jackson.databind.ObjectMapper; +import org.graylog.shaded.opensearch2.org.opensearch.action.support.master.AcknowledgedResponse; +import org.graylog.shaded.opensearch2.org.opensearch.client.indices.CreateDataStreamRequest; +import org.graylog.shaded.opensearch2.org.opensearch.client.indices.DeleteComposableIndexTemplateRequest; +import org.graylog.shaded.opensearch2.org.opensearch.client.indices.DeleteDataStreamRequest; +import org.graylog.shaded.opensearch2.org.opensearch.client.indices.GetDataStreamRequest; +import org.graylog.shaded.opensearch2.org.opensearch.client.indices.GetDataStreamResponse; +import org.graylog.shaded.opensearch2.org.opensearch.client.indices.PutComposableIndexTemplateRequest; +import org.graylog.shaded.opensearch2.org.opensearch.cluster.metadata.ComposableIndexTemplate; +import org.graylog.shaded.opensearch2.org.opensearch.cluster.metadata.DataStream; +import org.graylog.shaded.opensearch2.org.opensearch.common.compress.CompressedXContent; +import org.graylog.storage.opensearch2.ism.IsmApi; +import org.graylog.storage.opensearch2.ism.policy.IsmPolicy; +import org.graylog2.indexer.IndexNotFoundException; +import org.graylog2.indexer.datastream.DataStreamAdapter; +import org.graylog2.indexer.datastream.Policy; +import org.graylog2.indexer.indices.Template; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; +import jakarta.inject.Inject; +import java.io.IOException; +import java.util.List; +import java.util.Objects; +import java.util.Optional; + +public class DataStreamAdapterOS2 implements DataStreamAdapter { + + private final Logger log = LoggerFactory.getLogger(DataStreamAdapterOS2.class); + private final OpenSearchClient client; + private final ObjectMapper objectMapper; + private final IsmApi ismApi; + + @Inject + public DataStreamAdapterOS2(OpenSearchClient client, ObjectMapper objectMapper, IsmApi ismApi) { + this.client = client; + this.objectMapper = objectMapper; + this.ismApi = ismApi; + } + + @Override + public boolean ensureDataStreamTemplate(@Nonnull String templateName, @Nonnull Template template, @Nonnull String timestampField) { + final ComposableIndexTemplate.DataStreamTemplate datastreamTemplate = + new ComposableIndexTemplate.DataStreamTemplate(new DataStream.TimestampField(timestampField)); + + CompressedXContent serializedMapping = null; + try { + serializedMapping = new CompressedXContent(objectMapper.writeValueAsString(template.mappings())); + } catch (IOException e) { + throw new RuntimeException("Could not serialize mappings for data stream", e); + } + var settings = org.graylog.shaded.opensearch2.org.opensearch.common.settings.Settings.builder().loadFromMap(template.settings()).build(); + var osTemplate = new org.graylog.shaded.opensearch2.org.opensearch.cluster.metadata.Template(settings, serializedMapping, null); + var indexTemplate = new ComposableIndexTemplate(template.indexPatterns(), osTemplate, null, template.order(), null, null, datastreamTemplate); + var request = new PutComposableIndexTemplateRequest() + .name(templateName) + .indexTemplate(indexTemplate); + + final AcknowledgedResponse result = client.execute((c, requestOptions) -> c.indices().putIndexTemplate(request, requestOptions), + "Unable to create data stream template " + templateName); + + return result.isAcknowledged(); + } + + protected boolean deleteDataStreamTemplate(@Nonnull String templateName) { + final DeleteComposableIndexTemplateRequest request = new DeleteComposableIndexTemplateRequest(templateName); + final AcknowledgedResponse result = client.execute((c, requestOptions) -> c.indices().deleteIndexTemplate(request, requestOptions), + "Unable to delete data stream template " + templateName); + return result.isAcknowledged(); + } + + @Override + public void createDataStream(@Nonnull String dataStreamName) { + CreateDataStreamRequest createDataStreamRequest = new CreateDataStreamRequest(dataStreamName); + final GetDataStreamResponse dataStream; + try { + client.execute((c, requestOptions) -> + c.indices().getDataStream(new GetDataStreamRequest(dataStreamName), requestOptions)); + } catch (IndexNotFoundException e) { + client.execute((c, requestOptions) -> c.indices().createDataStream(createDataStreamRequest, requestOptions)); + } + } + + public boolean deleteDataStream(@Nonnull String dataStreamName) { + return client.execute((c, requestOptions) -> + c.indices().deleteDataStream(new DeleteDataStreamRequest(dataStreamName), requestOptions)).isAcknowledged(); + } + + protected List getDataStream(@Nonnull String dataStreamName) { + final GetDataStreamResponse dataStream = client.execute((c, requestOptions) -> + c.indices().getDataStream(new GetDataStreamRequest(dataStreamName), requestOptions)); + return dataStream.getDataStreams(); + } + + @Override + public void applyIsmPolicy(@Nonnull String dataStreamName, @Nonnull Policy policy) { + // this might need to be adjusted in the future to using versioning for the ism policy. + // for the time being, we will just remove and reapply the policy to the data stream. + IsmPolicy ismPolicy = (IsmPolicy) policy; + final String id = ismPolicy.id(); + if (Objects.isNull(id)) { + throw new IllegalArgumentException("Policy Id may not be null"); + } + final Optional osPolicy = ismApi.getPolicy(id); + if (osPolicy.isPresent()) { + ismApi.removePolicyFromIndex(dataStreamName); + ismApi.deletePolicy(id); + } + ismApi.createPolicy(ismPolicy.id(), new IsmPolicy(ismPolicy.policy())); + ismApi.addPolicyToIndex(id, dataStreamName); + } + + public void deleteIsmPolicy(@Nonnull String policyId) { + ismApi.deletePolicy(policyId); + } + + +} 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 73decfbe02..b92543ac83 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.datastream.DataStreamAdapter; import org.graylog2.indexer.datanode.RemoteReindexingMigrationAdapter; import org.graylog2.indexer.fieldtypes.IndexFieldTypePollerAdapter; import org.graylog2.indexer.fieldtypes.streamfiltered.esadapters.StreamsForFieldRetriever; @@ -58,6 +59,7 @@ public class OpenSearch2Module extends VersionAwareModule { bindForSupportedVersion(CountsAdapter.class).to(CountsAdapterOS2.class); bindForSupportedVersion(ClusterAdapter.class).to(ClusterAdapterOS2.class); bindForSupportedVersion(IndicesAdapter.class).to(IndicesAdapterOS2.class); + bindForSupportedVersion(DataStreamAdapter.class).to(DataStreamAdapterOS2.class); if (useComposableIndexTemplates) { bind(IndexTemplateAdapter.class).to(ComposableIndexTemplateAdapter.class); } else { diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/IsmApi.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/IsmApi.java new file mode 100644 index 0000000000..fd4ef2c897 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/IsmApi.java @@ -0,0 +1,119 @@ +/* + * 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.ism; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.inject.Inject; +import org.graylog.shaded.opensearch2.org.opensearch.client.Request; +import org.graylog.shaded.opensearch2.org.opensearch.client.Response; +import org.graylog.shaded.opensearch2.org.opensearch.client.ResponseException; +import org.graylog.storage.opensearch2.OpenSearchClient; +import org.graylog.storage.opensearch2.ism.policy.IsmPolicy; + +import java.io.IOException; +import java.util.Objects; +import java.util.Optional; + +import static org.graylog2.shared.utilities.StringUtils.f; + +public class IsmApi { + + private final ObjectMapper objectMapper; + private final OpenSearchClient client; + + @Inject + public IsmApi(OpenSearchClient client, ObjectMapper objectMapper) { + this.objectMapper = objectMapper; + this.client = client; + } + + public Optional getPolicy(String policyId) { + final Request request = request("GET", "policies/" + policyId, null); + return perform(request, new TypeReference<>() {}, "Could not get ism policy"); + } + + public void createPolicy(String policyId, IsmPolicy policy) { + // remove id from policy since it is not allowed on creation + if (!Objects.isNull(policy.id())) { + if (!policyId.equals(policy.id())) { + throw new IllegalArgumentException("Policy id present in policy does not match provided id."); + } + policy = new IsmPolicy(policy.policy()); + } + final Request request = request("PUT", "policies/" + policyId, policy); + perform(request, + new TypeReference() {}, + "Unable to create ism policy"); + } + + public void addPolicyToIndex(String policyId, String index) { + final Request request = request("POST", "add/" + index, null); + request.setJsonEntity(f("{\"policy_id\":\"%s\"}", policyId)); + perform(request, + new TypeReference() {}, + "Unable to add policy to index"); + } + + public void removePolicyFromIndex(String index) { + final Request request = request("POST", "remove/" + index, null); + perform(request, + new TypeReference() {}, + "Unable to remove policy from index"); + } + + public void deletePolicy(String policyId) { + final Request request = request("DELETE", "policies/" + policyId, null); + perform(request, + new TypeReference() {}, + "Unable to delete policy"); + } + + private Optional perform(Request request, TypeReference responseClass, String errorMessage) { + return client.execute((c, requestOptions) -> { + request.setOptions(requestOptions); + + Response response; + try { + response = c.getLowLevelClient().performRequest(request); + } catch (ResponseException e) { + if (e.getResponse().getStatusLine().getStatusCode() == 404) + return Optional.empty(); + throw e; + } + return Optional.of(returnType(response, responseClass)); + }, errorMessage); + } + + private R returnType(Response response, TypeReference responseClass) throws IOException { + return objectMapper.readValue(response.getEntity().getContent(), responseClass); + } + + private Request request(@SuppressWarnings("SameParameterValue") String method, String endpoint, Object body) { + final Request request = new Request(method, "/_plugins/_ism/" + endpoint); + request.addParameter("format", "json"); + try { + request.setJsonEntity(objectMapper.writeValueAsString(body)); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + return request; + } + +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/IsmPolicy.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/IsmPolicy.java new file mode 100644 index 0000000000..7025cbd49d --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/IsmPolicy.java @@ -0,0 +1,41 @@ +/* + * 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.ism.policy; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; + +import javax.annotation.Nullable; + +/** + * Model for a ISM Policy as defined in .... + * This model is not yet complete, but can be adjusted as further actions or parameters become relevant. + */ +@JsonInclude(JsonInclude.Include.NON_NULL) +public record IsmPolicy(@JsonProperty("_id") @Nullable String id, + @JsonProperty("_version") @Nullable String version, + @JsonProperty("_primary_term") @Nullable String primaryTerm, + @JsonProperty("_seq_no") @Nullable String seqNo, + Policy policy) implements org.graylog2.indexer.datastream.Policy { + public IsmPolicy(Policy policy) { + this(null, null, null, null, policy); + } + + public IsmPolicy(String id, Policy policy) { + this(id, null, null, null, policy); + } +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/Policy.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/Policy.java new file mode 100644 index 0000000000..0ebc2af4af --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/Policy.java @@ -0,0 +1,48 @@ +/* + * 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.ism.policy; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonInclude; +import org.graylog.storage.opensearch2.ism.policy.actions.Action; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import java.util.List; + +@JsonInclude(JsonInclude.Include.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public record Policy(@Nullable String policyId, + @Nonnull String description, + @Nullable String lastUpdatedTime, + @Nonnull String defaultState, + @Nonnull List states) { + + public Policy(@Nonnull String description, @Nonnull String defaultState, @Nonnull List states) { + this(null, description, null, defaultState, states); + } + + + public record State(@Nonnull String name, @Nonnull List actions, + @Nonnull List transitions) {} + + @JsonInclude(JsonInclude.Include.NON_NULL) + public record Transition(@Nonnull String stateName, @Nullable Condition conditions) {} + + public record Condition(String minIndexAge) {} + +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/Action.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/Action.java new file mode 100644 index 0000000000..2b5478aee8 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/Action.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.storage.opensearch2.ism.policy.actions; + +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +@JsonSerialize(using = ActionSerializer.class) +@JsonDeserialize(using = ActionDeserializer.class) +public record Action(@Nullable Retry retry, @Nonnull WrappedAction action) { + public Action(@Nonnull WrappedAction action) { + this(null, action); + } + +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/ActionDeserializer.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/ActionDeserializer.java new file mode 100644 index 0000000000..578ee70141 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/ActionDeserializer.java @@ -0,0 +1,59 @@ +/* + * 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.ism.policy.actions; + +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.databind.DeserializationContext; +import com.fasterxml.jackson.databind.JsonDeserializer; +import com.fasterxml.jackson.databind.JsonNode; + +import java.io.IOException; +import java.util.Iterator; +import java.util.Locale; + +public class ActionDeserializer extends JsonDeserializer { + @Override + public Action deserialize(JsonParser p, DeserializationContext ctxt) throws IOException { + JsonNode node = p.getCodec().readTree(p); + + // parse retry node + Retry retry = node.has("retry") ? ctxt.readTreeAsValue(node.get("retry"), Retry.class) : null; + + // resolve action type and deserialize based on name of second node + String actionPropertyName = findActionPropertyName(node); + final WrappedAction.Type type = WrappedAction.Type.valueOf(actionPropertyName.toUpperCase(Locale.ROOT)); + WrappedAction action = ctxt.readTreeAsValue(node.get(actionPropertyName), type.implementingClass); + + return new Action(retry, action); + } + + /** + * The node other than `retry` must be the action name + * + * @param node action node + * @return action name + */ + private String findActionPropertyName(JsonNode node) { + for (Iterator it = node.fieldNames(); it.hasNext(); ) { + final String fieldName = it.next(); + if (!"retry".equals(fieldName)) { + return fieldName; + } + } + throw new RuntimeException("Unable to determine the correct ism action"); + } +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/ActionSerializer.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/ActionSerializer.java new file mode 100644 index 0000000000..dc38330648 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/ActionSerializer.java @@ -0,0 +1,53 @@ +/* + * 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.ism.policy.actions; + +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.databind.JsonSerializer; +import com.fasterxml.jackson.databind.SerializerProvider; +import com.fasterxml.jackson.databind.jsontype.TypeSerializer; + +import java.io.IOException; +import java.util.Locale; + +public class ActionSerializer extends JsonSerializer { + @Override + public void serialize(Action value, JsonGenerator gen, SerializerProvider serializers) throws IOException { + gen.writeStartObject(); + if (value.retry() != null) { + gen.writeObjectField("retry", value.retry()); + } + + // Get the actual subtype of WrappedAction + String actionName = value.action().getType().name().toLowerCase(Locale.ROOT); + + // Set the property name based on the subtype + gen.writeObjectField(actionName, value.action()); + + gen.writeEndObject(); + } + + @Override + public void serializeWithType(Action value, JsonGenerator gen, SerializerProvider serializers, TypeSerializer typeSer) throws IOException { + // Implement if needed + typeSer.writeTypePrefixForObject(value, gen); + serialize(value, gen, serializers); + typeSer.writeTypeSuffixForObject(value, gen); + } + + +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/DeleteAction.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/DeleteAction.java new file mode 100644 index 0000000000..ef82e5f3db --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/DeleteAction.java @@ -0,0 +1,25 @@ +/* + * 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.ism.policy.actions; + +public record DeleteAction() implements WrappedAction { + + @Override + public Type getType() { + return Type.DELETE; + } +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/Retry.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/Retry.java new file mode 100644 index 0000000000..ad0c0e4750 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/Retry.java @@ -0,0 +1,20 @@ +/* + * 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.ism.policy.actions; + +public record Retry(int count, String backoff, String delay) { +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/RolloverAction.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/RolloverAction.java new file mode 100644 index 0000000000..426c150033 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/RolloverAction.java @@ -0,0 +1,31 @@ +/* + * 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.ism.policy.actions; + +import javax.annotation.Nonnull; + +public record RolloverAction(@Nonnull String minIndexAge, boolean copyAlias) implements WrappedAction { + + public RolloverAction(@Nonnull String minIndexAge) { + this(minIndexAge, false); + } + + @Override + public Type getType() { + return Type.ROLLOVER; + } +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/RollupAction.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/RollupAction.java new file mode 100644 index 0000000000..1338c73b78 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/RollupAction.java @@ -0,0 +1,57 @@ +/* + * 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.ism.policy.actions; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; + +import java.util.List; + +public record RollupAction(IsmRollup ismRollup) implements WrappedAction { + + @Override + public Type getType() { + return Type.ROLLUP; + } + + @JsonInclude(JsonInclude.Include.NON_NULL) + public record IsmRollup(String targetIndex, String description, int pageSize, + List dimensions, List metrics) { + + @JsonTypeInfo(include = JsonTypeInfo.As.WRAPPER_OBJECT, use = JsonTypeInfo.Id.NAME) + @JsonSubTypes({ + @JsonSubTypes.Type(value = DateHistogram.class, name = "date_histogram"), + }) + public interface Dimension {} + + @JsonIgnoreProperties(ignoreUnknown = true) + public record DateHistogram(String sourceField, String fixedInterval, String timezone) implements Dimension {} + + public record Metric(String sourceField, List metrics) {} + + @JsonTypeInfo(include = JsonTypeInfo.As.WRAPPER_OBJECT, use = JsonTypeInfo.Id.NAME) + @JsonSubTypes({ + @JsonSubTypes.Type(value = AvgMetric.class, name = "avg"), + }) + public interface AggregationMetric {} + + public record AvgMetric() implements AggregationMetric {} + + } +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/WrappedAction.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/WrappedAction.java new file mode 100644 index 0000000000..02e9352510 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ism/policy/actions/WrappedAction.java @@ -0,0 +1,41 @@ +/* + * 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.ism.policy.actions; + +import com.fasterxml.jackson.annotation.JsonIgnore; + +public interface WrappedAction { + + // actions need to be registered here as @JsonTypeInfo cannot be used due to the sibling "retry" node. + // enum name must match action name as defined in + // https://opensearch.org/docs/latest/im-plugin/ism/policies/#ism-supported-operations + enum Type { + DELETE(DeleteAction.class), + ROLLOVER(RolloverAction.class), + ROLLUP(RollupAction.class); + + public final Class implementingClass; + + Type(Class implementingClass) { + this.implementingClass = implementingClass; + } + } + + @JsonIgnore + Type getType(); + +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/views/OpenSearchBackend.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/views/OpenSearchBackend.java index 71d60ae846..6767fc0909 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/views/OpenSearchBackend.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/views/OpenSearchBackend.java @@ -165,8 +165,12 @@ public class OpenSearchBackend implements QueryBackend ), "Timerange for search type " + searchType.id() + " cannot be found in query or search type." ) - ) + ); + + if (effectiveStreamIds.stream().noneMatch(s -> s.startsWith(IndexLookup.DATASTREAM_PREFIX))) { + searchTypeOverrides .must(QueryBuilders.termsQuery(Message.FIELD_STREAMS, effectiveStreamIds)); + } searchType.query().ifPresent(searchTypeQuery -> { final QueryBuilder normalizedSearchTypeQuery = translateQueryString(searchTypeQuery.queryString()); diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/DataStreamAdapterOS2IT.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/DataStreamAdapterOS2IT.java new file mode 100644 index 0000000000..868125a3ac --- /dev/null +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/DataStreamAdapterOS2IT.java @@ -0,0 +1,85 @@ +/* + * 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.fasterxml.jackson.databind.ObjectMapper; +import org.graylog.shaded.opensearch2.org.opensearch.client.indices.DataStream; +import org.graylog.storage.opensearch2.ism.IsmApi; +import org.graylog.storage.opensearch2.ism.policy.IsmPolicy; +import org.graylog.storage.opensearch2.ism.policy.IsmPolicyTest; +import org.graylog.storage.opensearch2.testing.OpenSearchInstance; +import org.graylog2.indexer.indices.Template; +import org.graylog2.shared.bindings.providers.ObjectMapperProvider; +import org.junit.Rule; +import org.junit.Test; + +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +public class DataStreamAdapterOS2IT { + + @Rule + public final OpenSearchInstance openSearchInstance = OpenSearchInstance.create(); + + ObjectMapper objectMapper = new ObjectMapperProvider().get(); + + DataStreamAdapterOS2 dataStreamAdapter = new DataStreamAdapterOS2(openSearchInstance.openSearchClient(), + objectMapper, new IsmApi(openSearchInstance.openSearchClient(), objectMapper)); + + + @Test + public void testCreateDataStreamAndApplyPolicy() { + String stream = "testdatastream"; + + // create template for data stream + Template template = new Template(List.of(stream + "*"), new Template.Mappings(Map.of()), 1L, + new Template.Settings(Map.of("number_of_replicas", 0))); + String templateName = "datastream-test-template"; + boolean ack = dataStreamAdapter.ensureDataStreamTemplate(templateName, template, "timestamp"); + assertThat(ack).isTrue(); + + // create data stream and backing index + dataStreamAdapter.createDataStream(stream); + List dataStreams = dataStreamAdapter.getDataStream(stream); + dataStreamAdapter.createDataStream(stream); + assertThat(dataStreams).hasSize(1); + + // assert that existing data stream will not be recreated + List dataStreams2 = dataStreamAdapter.getDataStream(stream); + assertThat(dataStreams2).hasSize(1); + assertThat(dataStreams.get(0)).isEqualTo(dataStreams2.get(0)); + + // apply ism policy + IsmPolicy simpleTestPolicy = IsmPolicyTest.createSimpleTestPolicy(); + dataStreamAdapter.applyIsmPolicy(stream, simpleTestPolicy); + + // clean up to avoid exception when deleting indices + ack = dataStreamAdapter.deleteDataStream(stream); + assertThat(ack).isTrue(); + + assert simpleTestPolicy.id() != null; + dataStreamAdapter.deleteIsmPolicy(simpleTestPolicy.id()); + + ack = dataStreamAdapter.deleteDataStreamTemplate(templateName); + assertThat(ack).isTrue(); + + } + + +} diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ism/IsmApiIT.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ism/IsmApiIT.java new file mode 100644 index 0000000000..b06ca8d74e --- /dev/null +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ism/IsmApiIT.java @@ -0,0 +1,54 @@ +/* + * 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.ism; + +import org.graylog.storage.opensearch2.ism.policy.IsmPolicy; +import org.graylog.storage.opensearch2.ism.policy.IsmPolicyTest; +import org.graylog.storage.opensearch2.testing.OpenSearchInstance; +import org.graylog2.shared.bindings.providers.ObjectMapperProvider; +import org.junit.Rule; +import org.junit.Test; + +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; + +public class IsmApiIT { + + @Rule + public final OpenSearchInstance openSearchInstance = OpenSearchInstance.create(); + + IsmApi ismApi = new IsmApi(openSearchInstance.openSearchClient(), new ObjectMapperProvider().get()); + + @Test + public void testGetUnknownPolicyReturnsEmpty() { + final Optional policy = ismApi.getPolicy("unknown"); + assertThat(policy).isEmpty(); + } + + @Test + public void testCreatePolicyAndGet() { + IsmPolicy policy = IsmPolicyTest.createSimpleTestPolicy(); + final String testPolicy = policy.id(); + ismApi.createPolicy(testPolicy, policy); + Optional storedPolicy = ismApi.getPolicy(testPolicy); + assertThat(storedPolicy).isPresent(); + assertThat(storedPolicy.get().id()).isEqualTo(testPolicy); + ismApi.deletePolicy(testPolicy); + } + +} diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ism/policy/IsmPolicyTest.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ism/policy/IsmPolicyTest.java new file mode 100644 index 0000000000..96228e46c3 --- /dev/null +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ism/policy/IsmPolicyTest.java @@ -0,0 +1,101 @@ +/* + * 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.ism.policy; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.collect.ImmutableList; +import org.graylog.storage.opensearch2.ism.policy.actions.Action; +import org.graylog.storage.opensearch2.ism.policy.actions.DeleteAction; +import org.graylog.storage.opensearch2.ism.policy.actions.Retry; +import org.graylog2.shared.bindings.providers.ObjectMapperProvider; +import org.junit.Test; + +import java.io.IOException; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +public class IsmPolicyTest { + + public static IsmPolicy createSimpleTestPolicy() { + Policy.State deleteState = deleteState(); + Policy.State initialState = transitionState(deleteState.name()); + Policy policy = new Policy("Test Policy", initialState.name(), ImmutableList.of(initialState, deleteState)); + return new IsmPolicy("graylog-ism-test-policy", policy); + } + + private static Policy.State transitionState(String nextState) { + final List actions = ImmutableList.of(); + final List transitions = ImmutableList.of( + new Policy.Transition(nextState, new Policy.Condition("1d")) + ); + return new Policy.State("transition", actions, transitions); + } + + private static Policy.State deleteState() { + final List actions = ImmutableList.of(new Action(new Retry(3, "exponential", "1s"), new DeleteAction())); + final List transitions = ImmutableList.of(); + return new Policy.State("delete", actions, transitions); + } + + String simpleTestPolicyJson = """ + { + "_id" : "graylog-ism-test-policy", + "policy" : { + "description" : "Test Policy", + "default_state" : "transition", + "states" : [ { + "name" : "transition", + "actions" : [ ], + "transitions" : [ { + "state_name" : "delete", + "conditions" : { + "min_index_age" : "1d" + } + } ] + }, { + "name" : "delete", + "actions" : [ { + "retry" : { + "count" : 3, + "backoff" : "exponential", + "delay" : "1s" + }, + "delete" : { } + } ], + "transitions" : [ ] + } ] + } + }"""; + + + @Test + public void testPolicySerializationWorks() throws IOException { + ObjectMapper objectMapper = new ObjectMapperProvider().get(); + final String s = objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(createSimpleTestPolicy()); + assertThat(s).isEqualTo(simpleTestPolicyJson); + } + + @Test + public void testPolicyDeserializationWorks() throws JsonProcessingException { + ObjectMapper objectMapper = new ObjectMapperProvider().get(); + final IsmPolicy policy = objectMapper.readValue(simpleTestPolicyJson, IsmPolicy.class); + assertThat(policy.policy().states()).hasSize(2); + } + +} 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 60654c5441..7cb393bfee 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 @@ -43,7 +43,9 @@ import org.graylog.shaded.opensearch2.org.opensearch.client.Response; import org.graylog.shaded.opensearch2.org.opensearch.client.indices.CloseIndexRequest; import org.graylog.shaded.opensearch2.org.opensearch.client.indices.CreateIndexRequest; import org.graylog.shaded.opensearch2.org.opensearch.client.indices.DeleteComposableIndexTemplateRequest; +import org.graylog.shaded.opensearch2.org.opensearch.client.indices.DeleteDataStreamRequest; import org.graylog.shaded.opensearch2.org.opensearch.client.indices.GetComposableIndexTemplateRequest; +import org.graylog.shaded.opensearch2.org.opensearch.client.indices.GetDataStreamRequest; import org.graylog.shaded.opensearch2.org.opensearch.client.indices.GetIndexRequest; import org.graylog.shaded.opensearch2.org.opensearch.client.indices.GetIndexTemplatesRequest; import org.graylog.shaded.opensearch2.org.opensearch.client.indices.GetMappingsRequest; @@ -114,6 +116,14 @@ public class ClientOS2 implements Client { } } + public void deleteDataStreams() { + client.execute((c, requestOptions) -> c.indices().getDataStream(new GetDataStreamRequest("*"), requestOptions)) + .getDataStreams() + .forEach(ds -> client.execute((c, requestOptions) -> + c.indices().deleteDataStream(new DeleteDataStreamRequest(ds.getName()), requestOptions)) + ); + } + @Override public void closeIndex(String index) { final CloseIndexRequest closeIndexRequest = new CloseIndexRequest(index); @@ -312,6 +322,7 @@ public class ClientOS2 implements Client { @Override public void cleanUp() { LOG.debug("Removing indices: " + String.join(",", existingIndices())); + deleteDataStreams(); deleteIndices(existingIndices()); deleteTemplates(existingTemplates()); refreshNode(); diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/OpenSearchInstance.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/OpenSearchInstance.java index 201258dfe2..82d5059adf 100644 --- a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/OpenSearchInstance.java +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/OpenSearchInstance.java @@ -29,6 +29,10 @@ 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.client.indices.GetIndexRequest; import org.graylog.shaded.opensearch2.org.opensearch.client.indices.GetIndexResponse; +import org.graylog.shaded.opensearch2.org.opensearch.client.indices.PutComposableIndexTemplateRequest; +import org.graylog.shaded.opensearch2.org.opensearch.cluster.metadata.ComposableIndexTemplate; +import org.graylog.shaded.opensearch2.org.opensearch.cluster.metadata.Template; +import org.graylog.shaded.opensearch2.org.opensearch.common.settings.Settings; import org.graylog.storage.opensearch2.OpenSearchClient; import org.graylog.storage.opensearch2.RestHighLevelClientProvider; import org.graylog.testing.containermatrix.SearchServer; @@ -96,6 +100,18 @@ public class OpenSearchInstance extends TestableSearchServerInstance { if (version().satisfies(SearchVersion.Distribution.OPENSEARCH, "2.9.0")) { fixNumberOfReplicaForMlPlugin(); } + if (version().isOpenSearch()) { + fixDefaultNumberOfReplicasForIsmConfigs(); + } + } + + private void fixDefaultNumberOfReplicasForIsmConfigs() { + PutComposableIndexTemplateRequest request = new PutComposableIndexTemplateRequest(); + request.name("ism-zero-replica-template"); + request.indexTemplate(new ComposableIndexTemplate(List.of(".opendistro-ism-config"), + new Template(Settings.builder().put("number_of_replicas", 0).build(), null, null), + null, Long.MAX_VALUE, null, null)); + openSearchClient().execute((client, requestOptions) -> client.indices().putIndexTemplate(request, requestOptions)); } /** diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/search/elasticsearch/IndexLookup.java b/graylog2-server/src/main/java/org/graylog/plugins/views/search/elasticsearch/IndexLookup.java index f9e48a83ef..fae209cf46 100644 --- a/graylog2-server/src/main/java/org/graylog/plugins/views/search/elasticsearch/IndexLookup.java +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/search/elasticsearch/IndexLookup.java @@ -31,6 +31,8 @@ import java.util.SortedSet; import java.util.stream.Collectors; public class IndexLookup { + + public final static String DATASTREAM_PREFIX = "datastream:"; private final IndexRangeService indexRangeService; private final StreamService streamService; private final IndexRangeContainsOneOfStreams indexRangeContainsOneOfStreams; @@ -51,19 +53,27 @@ public class IndexLookup { this.indexRangeContainsOneOfStreams = indexRangeContainsOneOfStreams; } - public Set indexNamesForStreamsInTimeRange(final Set streamIds, + public Set indexNamesForStreamsInTimeRange(Set streamIds, final TimeRange timeRange) { if (streamIds.isEmpty()) { return Collections.emptySet(); } + Set result = streamIds.stream() + .filter(s -> s.startsWith(DATASTREAM_PREFIX)) + .map(s -> s.substring(DATASTREAM_PREFIX.length())) + .collect(Collectors.toSet()); + streamIds = streamIds.stream().filter(s -> !s.startsWith(DATASTREAM_PREFIX)).collect(Collectors.toSet()); + final Set usedStreams = streamService.loadByIds(streamIds); final SortedSet candidateIndices = indexRangeService.find(timeRange.getFrom(), timeRange.getTo()); - return candidateIndices.stream() + result.addAll(candidateIndices.stream() .filter(i -> indexRangeContainsOneOfStreams.test(i, usedStreams)) .map(IndexRange::indexName) - .collect(Collectors.toSet()); + .collect(Collectors.toSet()) + ); + return result; } } diff --git a/graylog2-server/src/main/java/org/graylog2/indexer/datastream/DataStreamAdapter.java b/graylog2-server/src/main/java/org/graylog2/indexer/datastream/DataStreamAdapter.java new file mode 100644 index 0000000000..bf99929492 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/indexer/datastream/DataStreamAdapter.java @@ -0,0 +1,31 @@ +/* + * 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.datastream; + +import org.graylog2.indexer.indices.Template; + +import javax.annotation.Nonnull; + +public interface DataStreamAdapter { + + boolean ensureDataStreamTemplate(@Nonnull String templateName, @Nonnull Template template, @Nonnull String timestampField); + + void createDataStream(String dataStreamName); + + void applyIsmPolicy(@Nonnull String dataStreamName, @Nonnull Policy policy); + +} diff --git a/graylog2-server/src/main/java/org/graylog2/indexer/datastream/Policy.java b/graylog2-server/src/main/java/org/graylog2/indexer/datastream/Policy.java new file mode 100644 index 0000000000..c2596c3a9e --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/indexer/datastream/Policy.java @@ -0,0 +1,20 @@ +/* + * 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.datastream; + +public interface Policy { +} 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 e61ec41830..c7907a9ac1 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.datastream.DataStreamAdapter; import org.graylog2.indexer.datanode.RemoteReindexingMigrationAdapter; import org.graylog2.indexer.fieldtypes.IndexFieldTypePollerAdapter; import org.graylog2.indexer.fieldtypes.streamfiltered.esadapters.StreamsForFieldRetriever; @@ -42,6 +43,7 @@ import org.graylog2.indexer.searches.SearchesAdapter; import org.graylog2.migrations.V20170607164210_MigrateReopenedIndicesToAliases; import org.graylog2.storage.providers.ClusterAdapterProvider; import org.graylog2.storage.providers.CountsAdapterProvider; +import org.graylog2.storage.providers.DataStreamAdapterProvider; import org.graylog2.storage.providers.ElasticsearchBackendProvider; import org.graylog2.storage.providers.IndexFieldTypePollerAdapterProvider; import org.graylog2.storage.providers.IndexToolsAdapterProvider; @@ -71,6 +73,7 @@ public class VersionAwareStorageModule extends AbstractModule { bind(StreamsForFieldRetriever.class).toProvider(StreamsForFieldRetrieverProvider.class); bind(CountsAdapter.class).toProvider(CountsAdapterProvider.class); bind(IndicesAdapter.class).toProvider(IndicesAdapterProvider.class); + bind(DataStreamAdapter.class).toProvider(DataStreamAdapterProvider.class); bind(SearchesAdapter.class).toProvider(SearchesAdapterProvider.class); bind(MultiChunkResultRetriever.class).toProvider(MultiChunkResultRetrieverProvider.class); bind(MoreSearchAdapter.class).toProvider(MoreSearchAdapterProvider.class); diff --git a/graylog2-server/src/main/java/org/graylog2/storage/providers/DataStreamAdapterProvider.java b/graylog2-server/src/main/java/org/graylog2/storage/providers/DataStreamAdapterProvider.java new file mode 100644 index 0000000000..6c679cd785 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/storage/providers/DataStreamAdapterProvider.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.datastream.DataStreamAdapter; +import org.graylog2.storage.DetectedSearchVersion; +import org.graylog2.storage.SearchVersion; +import org.graylog2.storage.VersionAwareProvider; + +import jakarta.inject.Inject; +import jakarta.inject.Provider; +import java.util.Map; + +public class DataStreamAdapterProvider extends VersionAwareProvider { + @Inject + public DataStreamAdapterProvider(@DetectedSearchVersion SearchVersion version, Map> pluginBindings) { + super(version, pluginBindings); + } +}