Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Create data stream aliases from template #73867

Merged
merged 27 commits into from
Jul 22, 2021
Merged
Show file tree
Hide file tree
Changes from 26 commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
52c9402
wip
danhermann Jun 7, 2021
ebdbce5
fix test compile error
danhermann Jun 7, 2021
cc8c7c4
add missing parser
danhermann Jun 7, 2021
5d21f2f
bwc
danhermann Jun 7, 2021
738b9e6
fix compilation
danhermann Jun 7, 2021
b0848d6
unused import
danhermann Jun 7, 2021
36847c8
fix test
danhermann Jun 7, 2021
0aebdb4
Merge branch 'master' into 66163_create_alias_from_template
elasticmachine Jun 16, 2021
49968e4
Merge branch 'master' into 66163_create_alias_from_template
elasticmachine Jul 13, 2021
001b0d5
fix merge error
danhermann Jul 14, 2021
39973f6
auto-create alias when data stream is created
danhermann Jul 16, 2021
991a36b
Merge branch 'master' into 66163_create_alias_from_template
elasticmachine Jul 16, 2021
e12bd60
remove AliasMetadata::isDataStream
danhermann Jul 19, 2021
1399903
reuse existing alias definitions
danhermann Jul 19, 2021
59e57af
Merge branch 'master' into 66163_create_alias_from_template
danhermann Jul 19, 2021
0f5c6a9
fix NPE
danhermann Jul 19, 2021
43e1d5f
add test
danhermann Jul 19, 2021
45cc1f5
remove a couple more unnecessary flags
danhermann Jul 19, 2021
86dc1f3
another test + minor reformatting
danhermann Jul 20, 2021
f75425b
checkstyle
danhermann Jul 20, 2021
4edd676
fix test
danhermann Jul 20, 2021
3b70768
review comments
danhermann Jul 21, 2021
f3a97c8
missed one spot
danhermann Jul 21, 2021
77e2a32
Merge branch 'master' into 66163_create_alias_from_template
elasticmachine Jul 21, 2021
8d1e1e7
fix merge error
danhermann Jul 21, 2021
33f2359
fix test
danhermann Jul 21, 2021
833677a
review comments
danhermann Jul 22, 2021
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -157,12 +157,11 @@ static ClusterState createDataStream(MetadataCreateIndexService metadataCreateIn
}

static ClusterState createDataStream(MetadataCreateIndexService metadataCreateIndexService,
ClusterState currentState,
String dataStreamName,
List<IndexMetadata> backingIndices,
IndexMetadata writeIndex,
SystemDataStreamDescriptor systemDataStreamDescriptor) throws Exception
{
ClusterState currentState,
String dataStreamName,
List<IndexMetadata> backingIndices,
IndexMetadata writeIndex,
SystemDataStreamDescriptor systemDataStreamDescriptor) throws Exception {
Objects.requireNonNull(metadataCreateIndexService);
Objects.requireNonNull(currentState);
Objects.requireNonNull(backingIndices);
Expand All @@ -177,8 +176,8 @@ static ClusterState createDataStream(MetadataCreateIndexService metadataCreateIn
throw new IllegalArgumentException("data_stream [" + dataStreamName + "] must be lowercase");
}
if (dataStreamName.startsWith(DataStream.BACKING_INDEX_PREFIX)) {
throw new IllegalArgumentException("data_stream [" + dataStreamName + "] must not start with '"
+ DataStream.BACKING_INDEX_PREFIX + "'");
throw new IllegalArgumentException(
"data_stream [" + dataStreamName + "] must not start with '" + DataStream.BACKING_INDEX_PREFIX + "'");
}

final boolean isSystem = systemDataStreamDescriptor != null;
Expand Down Expand Up @@ -222,6 +221,13 @@ static ClusterState createDataStream(MetadataCreateIndexService metadataCreateIn
logger.info("adding data stream [{}] with write index [{}] and backing indices [{}]", dataStreamName,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

maybe enhance this log line with the fact that aliases were created as well (if template has aliases)?

writeIndex.getIndex().getName(),
Strings.arrayToCommaDelimitedString(backingIndices.stream().map(i -> i.getIndex().getName()).toArray()));

if (template.template() != null && template.template().aliases() != null) {
for (var alias : template.template().aliases().values()) {
builder.put(alias.getAlias(), dataStreamName, alias.writeIndex(), alias.filter() == null ? null : alias.filter().string());
}
}

return ClusterState.builder(currentState).metadata(builder).build();
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -504,7 +504,8 @@ private ClusterState applyCreateIndexRequestWithV2Template(final ClusterState cu
logger.debug("applying create index request using composable template [{}]", templateName);

ComposableIndexTemplate template = currentState.getMetadata().templatesV2().get(templateName);
if (request.dataStreamName() == null && template.getDataStreamTemplate() != null) {
final boolean isDataStream = template.getDataStreamTemplate() != null;
if (isDataStream && request.dataStreamName() == null) {
throw new IllegalArgumentException("cannot create index with name [" + request.index() +
"], because it matches with template [" + templateName + "] that creates data streams only, " +
"use create data stream api instead");
Expand All @@ -519,14 +520,28 @@ private ClusterState applyCreateIndexRequestWithV2Template(final ClusterState cu
int routingNumShards = getIndexNumberOfRoutingShards(aggregatedIndexSettings, null);
IndexMetadata tmpImd = buildAndValidateTemporaryIndexMetadata(aggregatedIndexSettings, request, routingNumShards);

return applyCreateIndexWithTemporaryService(currentState, request, silent, null, tmpImd, mappings,
indexService -> resolveAndValidateAliases(request.index(), request.aliases(),
MetadataIndexTemplateService.resolveAliases(currentState.metadata(), templateName), currentState.metadata(),
// the context is only used for validation so it's fine to pass fake values for the
// shard id and the current timestamp
aliasValidator, xContentRegistry, indexService.newSearchExecutionContext(0, 0, null, () -> 0L, null, emptyMap()),
indexService.dateMathExpressionResolverAt(request.getNameResolvedAt())),
Collections.singletonList(templateName), metadataTransformer);
return applyCreateIndexWithTemporaryService(
currentState,
request,
silent,
null,
tmpImd,
mappings,
indexService -> resolveAndValidateAliases(
request.index(),
// data stream aliases are created separately in MetadataCreateDataStreamService::createDataStream
isDataStream ? Set.of() : request.aliases(),
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

maybe add a comment here that data stream aliases are created elsewhere?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, @martijnvg. I've made those changes.

isDataStream ? List.of() : MetadataIndexTemplateService.resolveAliases(currentState.metadata(), templateName),
currentState.metadata(),
aliasValidator,
xContentRegistry,
// the context is used ony for validation so it's fine to pass fake values for the shard id and the current timestamp
indexService.newSearchExecutionContext(0, 0, null, () -> 0L, null, emptyMap()),
indexService.dateMathExpressionResolverAt(request.getNameResolvedAt())
),
Collections.singletonList(templateName),
metadataTransformer
);
}

private ClusterState applyCreateIndexRequestForSystemDataStream(final ClusterState currentState,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1146,12 +1146,6 @@ static List<Map<String, AliasMetadata>> resolveAliases(final ComposableIndexTemp
.map(Template::aliases)
.ifPresent(aliases::add);

// A template that creates data streams can't also create aliases.
// (otherwise we end up with aliases pointing to backing indices of data streams)
if (aliases.size() > 0 && template.getDataStreamTemplate() != null) {
throw new IllegalArgumentException("template [" + templateName + "] has alias and data stream definitions");
}

// Aliases are applied in order, but subsequent alias configuration from the same name is
// ignored, so in order for the order to be correct, alias configuration should be in order
// of precedence (with the index template first)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -635,82 +635,6 @@ protected String contentType() {
}
}

public void testRolloverDataStreamWorksWithTemplateThatAlsoCreatesAliases() throws Exception {
final DataStream dataStream = DataStreamTestHelper.randomInstance()
// ensure no replicate data stream
.promoteDataStream();
ComposableIndexTemplate template = new ComposableIndexTemplate.Builder().indexPatterns(List.of(dataStream.getName() + "*"))
.template(new Template(null, null, Map.of("my-alias", AliasMetadata.newAliasMetadataBuilder("my-alias").build())))
.dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate()).build();
Metadata.Builder builder = Metadata.builder();
builder.put("template", template);
for (Index index : dataStream.getIndices()) {
builder.put(DataStreamTestHelper.getIndexMetadataBuilderForIndex(index));
}
builder.put(dataStream);
final ClusterState clusterState = ClusterState.builder(new ClusterName("test")).metadata(builder).build();

ThreadPool testThreadPool = new TestThreadPool(getTestName());
try {
DateFieldMapper dateFieldMapper = new DateFieldMapper.Builder(
"@timestamp",
DateFieldMapper.Resolution.MILLISECONDS,
null,
ScriptCompiler.NONE,
false,
Version.CURRENT).build(new ContentPath());
MappedFieldType mockedTimestampFieldType = mock(MappedFieldType.class);
when(mockedTimestampFieldType.name()).thenReturn("_data_stream_timestamp");
MetadataFieldMapper mockedTimestampField = new MetadataFieldMapper(mockedTimestampFieldType) {
@Override
protected String contentType() {
return null;
}
};
MetadataFieldMapper[] metadataFieldMappers = {new MetadataIndexTemplateServiceTests.MetadataTimestampFieldMapper(true)};
RootObjectMapper.Builder root = new RootObjectMapper.Builder("_doc");
root.add(new DateFieldMapper.Builder(dataStream.getTimeStampField().getName(), DateFieldMapper.Resolution.MILLISECONDS,
DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER, ScriptCompiler.NONE, true, Version.CURRENT));
Mapping mapping = new Mapping(root.build(new ContentPath("")), metadataFieldMappers, Collections.emptyMap());
MappingLookup mappingLookup = MappingLookup.fromMappers(
mapping,
List.of(mockedTimestampField, dateFieldMapper),
List.of(),
List.of());
ClusterService clusterService = ClusterServiceUtils.createClusterService(testThreadPool);
Environment env = mock(Environment.class);
when(env.sharedDataFile()).thenReturn(null);
AllocationService allocationService = mock(AllocationService.class);
when(allocationService.reroute(any(ClusterState.class), any(String.class))).then(i -> i.getArguments()[0]);
IndicesService indicesService = mockIndicesServices(mappingLookup);
IndexNameExpressionResolver mockIndexNameExpressionResolver = mock(IndexNameExpressionResolver.class);
when(mockIndexNameExpressionResolver.resolveDateMathExpression(any())).then(returnsFirstArg());

ShardLimitValidator shardLimitValidator = new ShardLimitValidator(Settings.EMPTY, clusterService);
MetadataCreateIndexService createIndexService = new MetadataCreateIndexService(Settings.EMPTY,
clusterService, indicesService, allocationService, new AliasValidator(), shardLimitValidator, env,
IndexScopedSettings.DEFAULT_SCOPED_SETTINGS, testThreadPool, null, EmptySystemIndices.INSTANCE, false);
MetadataIndexAliasesService indexAliasesService = new MetadataIndexAliasesService(clusterService, indicesService,
new AliasValidator(), null, xContentRegistry());
MetadataRolloverService rolloverService = new MetadataRolloverService(testThreadPool, createIndexService, indexAliasesService,
mockIndexNameExpressionResolver, EmptySystemIndices.INSTANCE);

MaxDocsCondition condition = new MaxDocsCondition(randomNonNegativeLong());
List<Condition<?>> metConditions = Collections.singletonList(condition);
CreateIndexRequest createIndexRequest = new CreateIndexRequest("_na_");

// Ensure that a warning header is emitted
Exception e = expectThrows(
IllegalArgumentException.class,
() -> rolloverService.rolloverClusterState(clusterState, dataStream.getName(), null, createIndexRequest, metConditions,
randomBoolean(), false)
);
assertThat(e.getMessage(), equalTo("template [template] has alias and data stream definitions"));
} finally {
testThreadPool.shutdown();
}
}

public void testValidation() throws Exception {
final String rolloverTarget;
final String sourceIndexName;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,12 +23,15 @@
import org.elasticsearch.indices.SystemIndices.Feature;
import org.elasticsearch.test.ESTestCase;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.createFirstBackingIndex;
import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.createTimestampField;
import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.generateMapping;
import static org.hamcrest.Matchers.arrayContainingInAnyOrder;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
Expand Down Expand Up @@ -66,6 +69,41 @@ public void testCreateDataStream() throws Exception {
assertThat(newState.metadata().index(DataStream.getDefaultBackingIndexName(dataStreamName, 1)).isSystem(), is(false));
}

public void testCreateDataStreamWithAliasFromTemplate() throws Exception {
final MetadataCreateIndexService metadataCreateIndexService = getMetadataCreateIndexService();
final String dataStreamName = "my-data-stream";
final int aliasCount = randomIntBetween(0, 3);
Map<String, AliasMetadata> aliases = new HashMap<>(aliasCount);
for (int k = 0; k < aliasCount; k++) {
final String aliasName = randomAlphaOfLength(6);
aliases.put(aliasName, AliasMetadata.newAliasMetadataBuilder(aliasName).build());
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

maybe also test the case here with aliases with write data streams and filter? (Either by randomly adding and asserting it or a new test)

}
ComposableIndexTemplate template = new ComposableIndexTemplate.Builder()
.indexPatterns(List.of(dataStreamName + "*"))
.dataStreamTemplate(new DataStreamTemplate())
.template(new Template(null, null, aliases))
.build();
ClusterState cs = ClusterState.builder(new ClusterName("_name"))
.metadata(Metadata.builder().put("template", template).build())
.build();
CreateDataStreamClusterStateUpdateRequest req =
new CreateDataStreamClusterStateUpdateRequest(dataStreamName, TimeValue.ZERO, TimeValue.ZERO);
ClusterState newState = MetadataCreateDataStreamService.createDataStream(metadataCreateIndexService, cs, req);
assertThat(newState.metadata().dataStreams().size(), equalTo(1));
assertThat(newState.metadata().dataStreams().get(dataStreamName).getName(), equalTo(dataStreamName));
assertThat(newState.metadata().dataStreams().get(dataStreamName).isSystem(), is(false));
assertThat(newState.metadata().dataStreams().get(dataStreamName).isHidden(), is(false));
assertThat(newState.metadata().dataStreams().get(dataStreamName).isReplicated(), is(false));
assertThat(newState.metadata().dataStreamAliases().size(), is(aliasCount));
assertThat(newState.metadata().dataStreamAliases().values().stream().map(DataStreamAlias::getName).toArray(),
arrayContainingInAnyOrder (new ArrayList<>(aliases.keySet()).toArray()));
assertThat(newState.metadata().index(DataStream.getDefaultBackingIndexName(dataStreamName, 1)), notNullValue());
assertThat(newState.metadata().index(DataStream.getDefaultBackingIndexName(dataStreamName, 1)).getAliases().size(), is(0));
assertThat(newState.metadata().index(DataStream.getDefaultBackingIndexName(dataStreamName, 1)).getSettings().get("index.hidden"),
equalTo("true"));
assertThat(newState.metadata().index(DataStream.getDefaultBackingIndexName(dataStreamName, 1)).isSystem(), is(false));
}

public void testCreateSystemDataStream() throws Exception {
final MetadataCreateIndexService metadataCreateIndexService = getMetadataCreateIndexService();
final String dataStreamName = ".system-data-stream";
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1207,31 +1207,6 @@ public void testResolveAliases() throws Exception {
assertThat(resolvedAliases, equalTo(List.of(a3, a1, a2)));
}

public void testResolveAliasesDataStreams() throws Exception {
Map<String, AliasMetadata> a1 = new HashMap<>();
a1.put("logs", AliasMetadata.newAliasMetadataBuilder("logs").build());

// index template can't have data streams and aliases
ComposableIndexTemplate it = new ComposableIndexTemplate(List.of("logs-*"),
new Template(null, null, a1), null, 0L, 1L, null, new ComposableIndexTemplate.DataStreamTemplate(), null);
ClusterState state1 = ClusterState.builder(ClusterState.EMPTY_STATE)
.metadata(Metadata.builder().put("1", it).build())
.build();
Exception e =
expectThrows(IllegalArgumentException.class, () -> MetadataIndexTemplateService.resolveAliases(state1.metadata(), "1"));
assertThat(e.getMessage(), equalTo("template [1] has alias and data stream definitions"));

// index template can't have data streams and a component template with an aliases
ComponentTemplate componentTemplate = new ComponentTemplate(new Template(null, null, a1), null, null);
it = new ComposableIndexTemplate(List.of("logs-*"), null, List.of("c1"), 0L, 1L, null,
new ComposableIndexTemplate.DataStreamTemplate(), null);
ClusterState state2 = ClusterState.builder(ClusterState.EMPTY_STATE)
.metadata(Metadata.builder().put("1", it).put("c1", componentTemplate).build())
.build();
e = expectThrows(IllegalArgumentException.class, () -> MetadataIndexTemplateService.resolveAliases(state2.metadata(), "1"));
assertThat(e.getMessage(), equalTo("template [1] has alias and data stream definitions"));
}

public void testAddInvalidTemplate() throws Exception {
ComposableIndexTemplate template = new ComposableIndexTemplate(Collections.singletonList("a"), null,
Arrays.asList("good", "bad"), null, null, null);
Expand Down
Loading