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 9ea01ba7f4..2c7601f5b0 100644 --- a/data-node/src/main/java/org/graylog/datanode/Configuration.java +++ b/data-node/src/main/java/org/graylog/datanode/Configuration.java @@ -215,9 +215,7 @@ public class Configuration extends BaseConfiguration { @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"; @@ -386,10 +384,6 @@ public class Configuration extends BaseConfiguration { return metricsTimestamp; } - public String getMetricsTemplate() { - return metricsTemplate; - } - public String getMetricsStream() { return metricsStream; } 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 83d02af957..33502b6182 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 @@ -32,6 +32,7 @@ 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.indexer.fieldtypes.IndexFieldTypesService; import org.graylog2.plugin.system.NodeId; import org.graylog2.security.CustomCAX509TrustManager; import org.slf4j.Logger; @@ -50,6 +51,7 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro private final Provider configurationProvider; private final EventBus eventBus; private final NodeId nodeId; + private final IndexFieldTypesService indexFieldTypesService; private final ObjectMapper objectMapper; @Inject @@ -59,11 +61,14 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro final CustomCAX509TrustManager trustManager, final NodeService nodeService, final Configuration configuration, - final NodeId nodeId, ObjectMapper objectMapper) { + final NodeId nodeId, + final IndexFieldTypesService indexFieldTypesService, + final ObjectMapper objectMapper) { this.configurationProvider = configurationProvider; this.eventBus = eventBus; this.nodeId = nodeId; this.objectMapper = objectMapper; + this.indexFieldTypesService = indexFieldTypesService; this.process = createOpensearchProcess(datanodeConfiguration, trustManager, configuration, nodeService, objectMapper); eventBus.register(this); } @@ -75,7 +80,7 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro process.addStateMachineTracer(watchdog); process.addStateMachineTracer(new StateMachineTransitionLogger()); process.addStateMachineTracer(new OpensearchRemovalTracer(process, configuration.getDatanodeNodeName())); - process.addStateMachineTracer(new ConfigureMetricsIndexSettings(process, configuration, objectMapper)); + process.addStateMachineTracer(new ConfigureMetricsIndexSettings(process, configuration, indexFieldTypesService, 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 index 543e5d7e3a..eeac146191 100644 --- a/data-node/src/main/java/org/graylog/datanode/metrics/ConfigureMetricsIndexSettings.java +++ b/data-node/src/main/java/org/graylog/datanode/metrics/ConfigureMetricsIndexSettings.java @@ -33,8 +33,9 @@ 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.graylog2.indexer.datastream.DataStreamService; +import org.graylog2.indexer.datastream.DataStreamServiceImpl; +import org.graylog2.indexer.fieldtypes.IndexFieldTypesService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -48,13 +49,15 @@ public class ConfigureMetricsIndexSettings implements StateMachineTracer { private final OpensearchProcess process; private final Configuration configuration; + private final IndexFieldTypesService indexFieldTypesService; private final ObjectMapper objectMapper; - private DataStreamAdapter dataStreamAdapter; + private DataStreamService dataStreamService; - public ConfigureMetricsIndexSettings(OpensearchProcess process, Configuration configuration, ObjectMapper objectMapper) { + public ConfigureMetricsIndexSettings(OpensearchProcess process, Configuration configuration, IndexFieldTypesService indexFieldTypesService, ObjectMapper objectMapper) { this.process = process; this.configuration = configuration; this.objectMapper = objectMapper; + this.indexFieldTypesService = indexFieldTypesService; } @Override @@ -65,34 +68,25 @@ public class ConfigureMetricsIndexSettings implements StateMachineTracer { public void transition(ProcessEvent trigger, ProcessState source, ProcessState destination) { if (destination == ProcessState.AVAILABLE && source == ProcessState.STARTING) { process.openSearchClient().ifPresent(client -> { - if (dataStreamAdapter == null) { + if (dataStreamService == null) { final IsmApi ismApi = new IsmApi(client, objectMapper); - dataStreamAdapter = new DataStreamAdapterOS2(client, objectMapper, ismApi); + dataStreamService = new DataStreamServiceImpl( + new DataStreamAdapterOS2(client, objectMapper, ismApi), + indexFieldTypesService + ); } - updateDataStreamTemplate(); - dataStreamAdapter.createDataStream(configuration.getMetricsStream()); - configureMetricsIsm(configuration); + dataStreamService.createDataStream(configuration.getMetricsStream(), + configuration.getMetricsTimestamp(), + createMappings(), + createPolicy(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") - ); + private Map> createMappings() { + Map> mappings = new HashMap<>(); 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)); + return mappings; } private IsmPolicy createPolicy(Configuration configuration) { @@ -142,7 +136,7 @@ public class ConfigureMetricsIndexSettings implements StateMachineTracer { ); 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"))); + final List transitions = ImmutableList.of(new Policy.Transition(nextState, new Policy.Condition(configuration.getMetricsRetention()))); return new Policy.State("rollup", actions, transitions); } 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 6767fc0909..550bd2f6e5 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 @@ -18,6 +18,9 @@ package org.graylog.storage.opensearch2.views; import com.google.common.collect.Maps; import io.opentelemetry.instrumentation.annotations.WithSpan; +import jakarta.inject.Inject; +import jakarta.inject.Named; +import jakarta.inject.Provider; import org.graylog.plugins.views.search.Filter; import org.graylog.plugins.views.search.GlobalOverride; import org.graylog.plugins.views.search.Query; @@ -53,14 +56,11 @@ import org.graylog2.indexer.ElasticsearchException; import org.graylog2.indexer.FieldTypeException; import org.graylog2.plugin.Message; import org.graylog2.plugin.Tools; +import org.graylog2.plugin.streams.Stream; import org.joda.time.DateTime; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import jakarta.inject.Inject; -import jakarta.inject.Named; -import jakarta.inject.Provider; - import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -167,7 +167,7 @@ public class OpenSearchBackend implements QueryBackend ) ); - if (effectiveStreamIds.stream().noneMatch(s -> s.startsWith(IndexLookup.DATASTREAM_PREFIX))) { + if (effectiveStreamIds.stream().noneMatch(s -> s.startsWith(Stream.DATASTREAM_PREFIX))) { searchTypeOverrides .must(QueryBuilders.termsQuery(Message.FIELD_STREAMS, effectiveStreamIds)); } diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/search/elasticsearch/FieldTypesLookup.java b/graylog2-server/src/main/java/org/graylog/plugins/views/search/elasticsearch/FieldTypesLookup.java index fa45c2a725..6978e11b44 100644 --- a/graylog2-server/src/main/java/org/graylog/plugins/views/search/elasticsearch/FieldTypesLookup.java +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/search/elasticsearch/FieldTypesLookup.java @@ -18,27 +18,34 @@ package org.graylog.plugins.views.search.elasticsearch; import com.google.common.collect.Sets; import org.graylog2.indexer.fieldtypes.FieldTypeDTO; +import org.graylog2.indexer.fieldtypes.IndexFieldTypesDTO; import org.graylog2.indexer.fieldtypes.IndexFieldTypesService; import jakarta.inject.Inject; +import org.graylog2.streams.StreamService; +import org.jetbrains.annotations.NotNull; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; public class FieldTypesLookup { private final IndexFieldTypesService indexFieldTypesService; + private final StreamService streamService; @Inject - public FieldTypesLookup(IndexFieldTypesService indexFieldTypesService) { + public FieldTypesLookup(IndexFieldTypesService indexFieldTypesService, StreamService streamService) { this.indexFieldTypesService = indexFieldTypesService; + this.streamService = streamService; } private Map> get(Set streamIds) { - return this.indexFieldTypesService.findForStreamIds(streamIds) + return findForStreamIds(streamIds) .stream() .flatMap(indexFieldTypes -> indexFieldTypes.fields().stream()) .collect(Collectors.toMap( @@ -48,6 +55,20 @@ public class FieldTypesLookup { )); } + private Collection findForStreamIds(Collection streamIds) { + final Set indexSetIds = getIndexSetIds(streamIds); + return indexFieldTypesService.findForIndexSets(indexSetIds); + } + + @NotNull + private Set getIndexSetIds(Collection streamIds) { + return streamService.loadByIds(streamIds) + .stream() + .filter(Objects::nonNull) + .map(stream -> stream.getIndexSet().getConfig().id()) + .collect(Collectors.toSet()); + } + public Optional getType(Set streamIds, String field) { final Map> allFieldTypes = this.get(streamIds); final Set fieldTypes = allFieldTypes.get(field); 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 fae209cf46..548d2ece6e 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 @@ -32,7 +32,6 @@ 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; @@ -60,10 +59,10 @@ public class IndexLookup { } Set result = streamIds.stream() - .filter(s -> s.startsWith(DATASTREAM_PREFIX)) - .map(s -> s.substring(DATASTREAM_PREFIX.length())) + .filter(s -> s.startsWith(Stream.DATASTREAM_PREFIX)) + .map(s -> s.substring(Stream.DATASTREAM_PREFIX.length())) .collect(Collectors.toSet()); - streamIds = streamIds.stream().filter(s -> !s.startsWith(DATASTREAM_PREFIX)).collect(Collectors.toSet()); + streamIds = streamIds.stream().filter(s -> !s.startsWith(Stream.DATASTREAM_PREFIX)).collect(Collectors.toSet()); final Set usedStreams = streamService.loadByIds(streamIds); final SortedSet candidateIndices = indexRangeService.find(timeRange.getFrom(), timeRange.getTo()); diff --git a/graylog2-server/src/main/java/org/graylog2/bindings/PersistenceServicesBindings.java b/graylog2-server/src/main/java/org/graylog2/bindings/PersistenceServicesBindings.java index 6d65f6d41b..666a605fbc 100644 --- a/graylog2-server/src/main/java/org/graylog2/bindings/PersistenceServicesBindings.java +++ b/graylog2-server/src/main/java/org/graylog2/bindings/PersistenceServicesBindings.java @@ -30,6 +30,8 @@ import org.graylog2.database.suggestions.EntitySuggestionService; import org.graylog2.database.suggestions.MongoEntitySuggestionService; import org.graylog2.indexer.IndexFailureService; import org.graylog2.indexer.IndexFailureServiceImpl; +import org.graylog2.indexer.datastream.DataStreamService; +import org.graylog2.indexer.datastream.DataStreamServiceImpl; import org.graylog2.indexer.ranges.IndexRangeService; import org.graylog2.indexer.ranges.MongoIndexRangeService; import org.graylog2.inputs.InputService; @@ -76,5 +78,6 @@ public class PersistenceServicesBindings extends AbstractModule { bind(EntityListPreferencesService.class).to(EntityListPreferencesServiceImpl.class); bind(EntitySuggestionService.class).to(MongoEntitySuggestionService.class); bind(EntityTitleService.class).to(EntityTitleServiceImpl.class); + bind(DataStreamService.class).to(DataStreamServiceImpl.class); } } diff --git a/graylog2-server/src/main/java/org/graylog2/indexer/datastream/DataStreamService.java b/graylog2-server/src/main/java/org/graylog2/indexer/datastream/DataStreamService.java new file mode 100644 index 0000000000..ef9462c923 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/indexer/datastream/DataStreamService.java @@ -0,0 +1,39 @@ +/* + * 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 java.util.Map; + +public interface DataStreamService { + + /** + * Create a data stream. + * This will create a data stream template with the given mappings, adding a timestamp mapping if none + * is specified for the timestampField in the mappings. Mappings will be made available to the FieldTypeMappingsService. + * If no data stream with the given name exists, the data stream will be created. + * A policy for handling lifecycle management of the data stream and its backing indices can be specified. + * Existing templates, mappings and policies will be updated. + * + * @param dataStreamName name of the data stream + * @param timestampField timestamp field + * @param mappings field type mappings in the data stream + * @param ismPolicy ism policy for backing index lifecycle management + */ + void createDataStream(String dataStreamName, String timestampField, Map> mappings, + Policy ismPolicy); + +} diff --git a/graylog2-server/src/main/java/org/graylog2/indexer/datastream/DataStreamServiceImpl.java b/graylog2-server/src/main/java/org/graylog2/indexer/datastream/DataStreamServiceImpl.java new file mode 100644 index 0000000000..c5bfc9aae8 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/indexer/datastream/DataStreamServiceImpl.java @@ -0,0 +1,74 @@ +/* + * 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 com.google.common.collect.ImmutableMap; +import jakarta.inject.Inject; +import org.graylog2.indexer.fieldtypes.FieldTypeDTO; +import org.graylog2.indexer.fieldtypes.IndexFieldTypesDTO; +import org.graylog2.indexer.fieldtypes.IndexFieldTypesService; +import org.graylog2.indexer.indices.Template; +import org.graylog2.plugin.streams.Stream; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +public class DataStreamServiceImpl implements DataStreamService { + + private final DataStreamAdapter dataStreamAdapter; + private final IndexFieldTypesService indexFieldTypesService; + + @Inject + public DataStreamServiceImpl(DataStreamAdapter dataStreamAdapter, IndexFieldTypesService indexFieldTypesService) { + this.dataStreamAdapter = dataStreamAdapter; + this.indexFieldTypesService = indexFieldTypesService; + } + + @Override + public void createDataStream(String dataStreamName, String timestampField, Map> mappings, + Policy ismPolicy) { + updateDataStreamTemplate(dataStreamName, timestampField, mappings); + dataStreamAdapter.createDataStream(dataStreamName); + dataStreamAdapter.applyIsmPolicy(dataStreamName, ismPolicy); + } + + private void updateDataStreamTemplate(String dataStreamName, String timestampField, Map> mappings) { + if (!mappings.containsKey(timestampField)) { + mappings.put(timestampField, ImmutableMap.of( + "type", "date", + "format", "yyyy-MM-dd HH:mm:ss.SSS||strict_date_optional_time||epoch_millis") + ); + } + Template template = new Template(List.of(dataStreamName + "*"), + new Template.Mappings(ImmutableMap.of("properties", mappings)), 99999L, new Template.Settings(Map.of())); + dataStreamAdapter.ensureDataStreamTemplate(dataStreamName + "-template", template, timestampField); + createFieldTypes(dataStreamName, mappings); + } + + private void createFieldTypes(String metricsStream, Map> mappings) { + final Set fields = mappings.entrySet().stream() + .map(mapping -> FieldTypeDTO.builder() + .fieldName(mapping.getKey()) + .physicalType(mapping.getValue().get("type")) + .build()) + .collect(Collectors.toSet()); + IndexFieldTypesDTO dto = IndexFieldTypesDTO.create(Stream.DATASTREAM_PREFIX + metricsStream, metricsStream, fields); + indexFieldTypesService.upsert(dto); + } +} diff --git a/graylog2-server/src/main/java/org/graylog2/indexer/fieldtypes/IndexFieldTypesService.java b/graylog2-server/src/main/java/org/graylog2/indexer/fieldtypes/IndexFieldTypesService.java index 4ffe2919bf..0884d908ac 100644 --- a/graylog2-server/src/main/java/org/graylog2/indexer/fieldtypes/IndexFieldTypesService.java +++ b/graylog2-server/src/main/java/org/graylog2/indexer/fieldtypes/IndexFieldTypesService.java @@ -23,27 +23,22 @@ import com.mongodb.client.AggregateIterable; import com.mongodb.client.MongoCollection; import com.mongodb.client.model.Aggregates; import com.mongodb.client.model.Projections; +import jakarta.inject.Inject; import org.bson.Document; import org.bson.types.ObjectId; import org.graylog2.bindings.providers.MongoJackObjectMapperProvider; import org.graylog2.database.MongoConnection; import org.graylog2.database.MongoDBUpsertRetryer; -import org.graylog2.streams.StreamService; import org.mongojack.DBQuery; import org.mongojack.JacksonDBCollection; import org.mongojack.WriteResult; -import jakarta.inject.Inject; - import java.util.ArrayList; import java.util.Collection; import java.util.LinkedList; import java.util.List; import java.util.Locale; -import java.util.Objects; import java.util.Optional; -import java.util.Set; -import java.util.stream.Collectors; import static com.mongodb.client.model.Filters.and; import static com.mongodb.client.model.Filters.eq; @@ -64,15 +59,12 @@ public class IndexFieldTypesService { private static final String FIELDS_FIELD_NAMES = String.format(Locale.US, "%s.%s", FIELD_FIELDS, FIELD_NAME); private final JacksonDBCollection db; - private final StreamService streamService; private final MongoCollection mongoCollection; @Inject public IndexFieldTypesService(final MongoConnection mongoConnection, - final StreamService streamService, final MongoJackObjectMapperProvider objectMapperProvider) { - this.streamService = streamService; this.mongoCollection = mongoConnection.getMongoDatabase().getCollection("index_field_types"); this.db = JacksonDBCollection.wrap(mongoConnection.getDatabase().getCollection("index_field_types"), IndexFieldTypesDTO.class, @@ -191,16 +183,6 @@ public class IndexFieldTypesService { return findByQuery(query); } - public Collection findForStreamIds(Collection streamIds) { - final Set indexSetIds = streamService.loadByIds(streamIds) - .stream() - .filter(Objects::nonNull) - .map(stream -> stream.getIndexSet().getConfig().id()) - .collect(Collectors.toSet()); - - return findForIndexSets(indexSetIds); - } - public Collection findAll() { return findByQuery(DBQuery.empty()); } diff --git a/graylog2-server/src/main/java/org/graylog2/plugin/streams/Stream.java b/graylog2-server/src/main/java/org/graylog2/plugin/streams/Stream.java index 761f1481ad..39b52c3289 100644 --- a/graylog2-server/src/main/java/org/graylog2/plugin/streams/Stream.java +++ b/graylog2-server/src/main/java/org/graylog2/plugin/streams/Stream.java @@ -44,6 +44,10 @@ public interface Stream extends Persisted { * The ID of the stream for message failures. */ String FAILURES_STREAM_ID = "000000000000000000000004"; + /** + * The prefix of all streams managed by OpenSearch as data streams + */ + String DATASTREAM_PREFIX = "datastream:"; /** * Contains all default event streams. (e.g. events and system events) */ diff --git a/graylog2-server/src/main/java/org/graylog2/streams/StreamServiceImpl.java b/graylog2-server/src/main/java/org/graylog2/streams/StreamServiceImpl.java index f90a7defdb..eba5741f8c 100644 --- a/graylog2-server/src/main/java/org/graylog2/streams/StreamServiceImpl.java +++ b/graylog2-server/src/main/java/org/graylog2/streams/StreamServiceImpl.java @@ -26,6 +26,7 @@ import com.mongodb.DBCursor; import com.mongodb.DBObject; import com.mongodb.QueryBuilder; import com.mongodb.WriteResult; +import jakarta.inject.Inject; import org.bson.types.ObjectId; import org.graylog.security.entities.EntityOwnershipService; import org.graylog2.database.MongoConnection; @@ -50,9 +51,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.annotation.Nullable; - -import jakarta.inject.Inject; - import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -266,14 +264,21 @@ public class StreamServiceImpl extends PersistedServiceImpl implements StreamSer @Override public Set indexSetIdsByIds(Collection streamIds) { + Set dataStreamIds = streamIds.stream() + .filter(s -> s.startsWith(Stream.DATASTREAM_PREFIX)) + .collect(Collectors.toSet()); + final Set objectIds = streamIds.stream() + .filter(s -> !s.startsWith(Stream.DATASTREAM_PREFIX)) .map(ObjectId::new) .collect(Collectors.toSet()); final DBObject query = QueryBuilder.start(StreamImpl.FIELD_ID).in(objectIds).get(); final DBObject onlyIndexSetIdField = DBProjection.include(FIELD_INDEX_SET_ID); - return StreamSupport.stream(collection(StreamImpl.class).find(query, onlyIndexSetIdField).spliterator(), false) + Set indexSets = StreamSupport.stream(collection(StreamImpl.class).find(query, onlyIndexSetIdField).spliterator(), false) .map(s -> s.get(FIELD_INDEX_SET_ID).toString()) .collect(Collectors.toSet()); + indexSets.addAll(dataStreamIds); + return indexSets; } protected Set loadOutputsForRawStream(DBObject stream) { diff --git a/graylog2-server/src/test/java/org/graylog/plugins/views/search/elasticsearch/FieldTypesLookupTest.java b/graylog2-server/src/test/java/org/graylog/plugins/views/search/elasticsearch/FieldTypesLookupTest.java index 4d83d2e65d..0e06db22ed 100644 --- a/graylog2-server/src/test/java/org/graylog/plugins/views/search/elasticsearch/FieldTypesLookupTest.java +++ b/graylog2-server/src/test/java/org/graylog/plugins/views/search/elasticsearch/FieldTypesLookupTest.java @@ -17,137 +17,151 @@ package org.graylog.plugins.views.search.elasticsearch; import com.google.common.collect.ImmutableSet; +import org.apache.commons.lang3.tuple.Pair; import org.graylog2.indexer.fieldtypes.FieldTypeDTO; import org.graylog2.indexer.fieldtypes.IndexFieldTypesDTO; import org.graylog2.indexer.fieldtypes.IndexFieldTypesService; -import org.junit.jupiter.api.BeforeEach; +import org.graylog2.plugin.streams.Stream; +import org.graylog2.streams.StreamService; import org.junit.jupiter.api.Test; +import org.mockito.Mockito; +import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.Optional; import java.util.Set; +import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; public class FieldTypesLookupTest { - private IndexFieldTypesService indexFieldTypesService; - private FieldTypesLookup fieldTypesLookup; - @BeforeEach - void setUp() { - this.indexFieldTypesService = mock(IndexFieldTypesService.class); - this.fieldTypesLookup = new FieldTypesLookup(indexFieldTypesService); - } @Test void returnsEmptyOptionalIfFieldTypesAreEmpty() { - final Optional result = this.fieldTypesLookup.getType(Collections.singleton("SomeStream"), "somefield"); + final Pair services = mockServices(); + final FieldTypesLookup lookup = new FieldTypesLookup(services.getLeft(), services.getRight()); + final Optional result = lookup.getType(Collections.singleton("SomeStream"), "somefield"); assertThat(result).isEmpty(); } @Test void returnsEmptyOptionalIfStreamsAreEmpty() { - final Optional result = this.fieldTypesLookup.getType(Collections.emptySet(), "somefield"); + final Pair services = mockServices(); + final FieldTypesLookup lookup = new FieldTypesLookup(services.getLeft(), services.getRight()); + final Optional result = lookup.getType(Collections.emptySet(), "somefield"); assertThat(result).isEmpty(); } @Test void returnsEmptyOptionalIfMultipleTypesExistForField() { - when(this.indexFieldTypesService.findForStreamIds(Collections.singleton("stream1"))).thenReturn(ImmutableSet.of( + final Pair services = mockServices( IndexFieldTypesDTO.create("indexSet1", "stream1", ImmutableSet.of( FieldTypeDTO.create("somefield", "long") )), IndexFieldTypesDTO.create("indexSet2", "stream1", ImmutableSet.of( FieldTypeDTO.create("somefield", "float") )) - )); + ); - final Optional result = this.fieldTypesLookup.getType(Collections.singleton("stream1"), "somefield"); + final FieldTypesLookup lookup = new FieldTypesLookup(services.getLeft(), services.getRight()); + final Optional result = lookup.getType(Collections.singleton("stream1"), "somefield"); assertThat(result).isEmpty(); } @Test void returnsEmptyOptionalIfNoTypesExistForStream() { - when(this.indexFieldTypesService.findForStreamIds(Collections.singleton("stream1"))).thenReturn(ImmutableSet.of( + final Pair services = mockServices( IndexFieldTypesDTO.create("indexSet1", "stream1", ImmutableSet.of( FieldTypeDTO.create("somefield", "long") )) - )); + ); - final Optional result = this.fieldTypesLookup.getType(Collections.singleton("stream2"), "somefield"); + final FieldTypesLookup lookup = new FieldTypesLookup(services.getLeft(), services.getRight()); + final Optional result = lookup.getType(Collections.singleton("stream2"), "somefield"); assertThat(result).isEmpty(); } @Test void returnsFieldTypeIfSingleTypeExistsForFieldInStream() { - when(this.indexFieldTypesService.findForStreamIds(Collections.singleton("stream1"))).thenReturn(ImmutableSet.of( + final Pair services = mockServices( IndexFieldTypesDTO.create("indexSet1", "stream1", ImmutableSet.of( FieldTypeDTO.create("somefield", "long") )) - )); + ); - final Optional result = this.fieldTypesLookup.getType(Collections.singleton("stream1"), "somefield"); + final FieldTypesLookup lookup = new FieldTypesLookup(services.getLeft(), services.getRight()); + final Optional result = lookup.getType(Collections.singleton("stream1"), "somefield"); assertThat(result).contains("long"); } @Test void returnsFieldTypeIfSingleTypeExistsForFieldInAllStreams() { - when(this.indexFieldTypesService.findForStreamIds(ImmutableSet.of("stream1", "stream2"))).thenReturn(ImmutableSet.of( + final Pair services = mockServices( IndexFieldTypesDTO.create("indexSet1", "stream1", ImmutableSet.of( FieldTypeDTO.create("somefield", "long") )), IndexFieldTypesDTO.create("indexSet2", "stream2", ImmutableSet.of( FieldTypeDTO.create("somefield", "long") )) - )); + ); - final Optional result = this.fieldTypesLookup.getType(ImmutableSet.of("stream1", "stream2"), "somefield"); + final FieldTypesLookup lookup = new FieldTypesLookup(services.getLeft(), services.getRight()); + final Optional result = lookup.getType(ImmutableSet.of("stream1", "stream2"), "somefield"); assertThat(result).contains("long"); } @Test void getTypesReturnsEmptyMapIfFieldTypesAreEmpty() { - assertThat(fieldTypesLookup.getTypes(Set.of("SomeStream"), Set.of("somefield"))).isEmpty(); + final Pair services = mockServices(); + final FieldTypesLookup lookup = new FieldTypesLookup(services.getLeft(), services.getRight()); + assertThat(lookup.getTypes(Set.of("SomeStream"), Set.of("somefield"))).isEmpty(); } @Test void getTypesReturnsEmptyMapIfStreamsAreEmpty() { - assertThat(fieldTypesLookup.getTypes(Set.of(), Set.of("somefield"))).isEmpty(); + final Pair services = mockServices(); + final FieldTypesLookup lookup = new FieldTypesLookup(services.getLeft(), services.getRight()); + assertThat(lookup.getTypes(Set.of(), Set.of("somefield"))).isEmpty(); } @Test void getTypesReturnsEmptyMapIfMultipleTypesExistForField() { - when(this.indexFieldTypesService.findForStreamIds(Collections.singleton("stream1"))).thenReturn(ImmutableSet.of( + final Pair services = mockServices( IndexFieldTypesDTO.create("indexSet1", "stream1", ImmutableSet.of( FieldTypeDTO.create("somefield", "long") )), IndexFieldTypesDTO.create("indexSet2", "stream1", ImmutableSet.of( FieldTypeDTO.create("somefield", "float") )) - )); + ); - assertThat(fieldTypesLookup.getTypes(Set.of("stream1"), Set.of("somefield"))).isEmpty(); + final FieldTypesLookup lookup = new FieldTypesLookup(services.getLeft(), services.getRight()); + assertThat(lookup.getTypes(Set.of("stream1"), Set.of("somefield"))).isEmpty(); } @Test void getTypesReturnsEmptyMapForNonExistingFields() { - when(this.indexFieldTypesService.findForStreamIds(ImmutableSet.of("stream1", "stream2"))).thenReturn(ImmutableSet.of( + final Pair services = mockServices( IndexFieldTypesDTO.create("indexSet1", "stream1", ImmutableSet.of( FieldTypeDTO.create("existing_field", "long") )), IndexFieldTypesDTO.create("indexSet2", "stream2", ImmutableSet.of( FieldTypeDTO.create("existing_field", "long") )) - )); + ); - assertThat(fieldTypesLookup.getTypes(Set.of("stream1", "stream2"), Set.of("somefield1", "somefield2"))).isEmpty(); + final FieldTypesLookup lookup = new FieldTypesLookup(services.getLeft(), services.getRight()); + assertThat(lookup.getTypes(Set.of("stream1", "stream2"), Set.of("somefield1", "somefield2"))).isEmpty(); } @Test void getTypesBehavesCorrectlyIfMultipleScenariosAreMixed() { - when(this.indexFieldTypesService.findForStreamIds(Set.of("stream1", "stream2"))).thenReturn(Set.of( + final Pair services = mockServices( IndexFieldTypesDTO.create("indexSet1", "stream1", Set.of( FieldTypeDTO.create("good_field", "long"), FieldTypeDTO.create("bad_field", "ip") @@ -156,12 +170,43 @@ public class FieldTypesLookupTest { FieldTypeDTO.create("good_field", "long"), FieldTypeDTO.create("bad_field", "text") )) - )); + ); - assertThat(fieldTypesLookup.getTypes(Set.of("stream1", "stream2"), Set.of("good_field", "bad_field", "unknown_field"))) + final FieldTypesLookup lookup = new FieldTypesLookup(services.getLeft(), services.getRight()); + + assertThat(lookup.getTypes(Set.of("stream1", "stream2"), Set.of("good_field", "bad_field", "unknown_field"))) .satisfies(result -> assertThat(result).doesNotContainKey("bad_field")) //different type in different streams .satisfies(result -> assertThat(result).doesNotContainKey("unknown_field")) //no type info .satisfies(result -> assertThat(result).containsEntry("good_field", "long")); //proper type info } + + /** + * @param dtos what we expect as a result of {@link IndexFieldTypesService#findForIndexSets(Collection)} + * @return pair of services that can be used as dependencies for the {@link FieldTypesLookup]. + */ + Pair mockServices(IndexFieldTypesDTO... dtos) { + + // caution, we interchange stream and index names in this test freely + final Set expectedStreams = Arrays.stream(dtos).map(IndexFieldTypesDTO::indexName).collect(Collectors.toSet()); + + final Set expectedIndexSets = Arrays.stream(dtos).map(IndexFieldTypesDTO::indexSetId).collect(Collectors.toSet()); + + final StreamService streamService = mock(StreamService.class); + final IndexFieldTypesService indexFieldTypeService = mock(IndexFieldTypesService.class); + + Mockito.doAnswer(invocation -> Arrays.stream(dtos) + .map(dto -> { + final Stream mock = mock(Stream.class, RETURNS_DEEP_STUBS); + when(mock.getIndexSet().getConfig().id()).thenReturn(dto.indexSetId()); + return mock; + }) + .collect(Collectors.toSet()) + ).when(streamService).loadByIds(expectedStreams); + + Mockito.when(indexFieldTypeService.findForIndexSets(expectedIndexSets)) + .thenReturn(Set.of(dtos)); + + return Pair.of(indexFieldTypeService, streamService); + } } diff --git a/graylog2-server/src/test/java/org/graylog2/indexer/datastream/DataStreamServiceImplTest.java b/graylog2-server/src/test/java/org/graylog2/indexer/datastream/DataStreamServiceImplTest.java new file mode 100644 index 0000000000..053691f0e3 --- /dev/null +++ b/graylog2-server/src/test/java/org/graylog2/indexer/datastream/DataStreamServiceImplTest.java @@ -0,0 +1,122 @@ +/* + * 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.fieldtypes.IndexFieldTypesDTO; +import org.graylog2.indexer.fieldtypes.IndexFieldTypesService; +import org.graylog2.indexer.indices.Template; +import org.graylog2.plugin.streams.Stream; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; + +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +public class DataStreamServiceImplTest { + @Rule + public final MockitoRule mockitoRule = MockitoJUnit.rule(); + @Mock + private DataStreamAdapter dataStreamAdapter; + @Mock + private IndexFieldTypesService indexFieldTypesService; + + private DataStreamService dataStreamService; + + @Before + public void setUp() { + dataStreamService = new DataStreamServiceImpl(dataStreamAdapter, indexFieldTypesService); + } + + @Test + public void createDataStreamPerformsFunctions() { + final String name = "teststream"; + final String ts = "ts"; + final Map> mappings = new HashMap<>(); + final Policy policy = mock(Policy.class); + dataStreamService.createDataStream(name, ts, mappings, policy); + verify(dataStreamAdapter).ensureDataStreamTemplate(eq(name + "-template"), any(), eq(ts)); + verify(indexFieldTypesService).upsert(any()); + verify(dataStreamAdapter).createDataStream(name); + verify(dataStreamAdapter).applyIsmPolicy(name, policy); + } + + @SuppressWarnings("unchecked") + @Test + public void templateCreatesTimestampMapping() { + final Map> mappings = new HashMap<>(); + String ts = "ts"; + dataStreamService.createDataStream("teststream", ts, mappings, mock(Policy.class)); + ArgumentCaptor