Skip to content

[ML] Switch ML internal index templates to composable templates #73232

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.

Sign up for GitHub

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

Merged
merged 5 commits into from
May 24, 2021
Merged

[ML] Switch ML internal index templates to composable templates #73232

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
7bcebd1
[ML] Switch ML internal index templates to composable templates
May 19, 2021
200c007
Remove unused import
May 19, 2021
8aeac2b
Fix mappings vs template test
May 19, 2021
0e953fa
Fix another place where we wait for templates
May 19, 2021
4d384a6
Merge branch 'master' into composable_templates_for_ml
May 19, 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
43 changes: 34 additions & 9 deletions ...core/src/main/java/org/elasticsearch/xpack/core/common/notifications/AbstractAuditor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,17 +8,24 @@

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchParseException;
import org.elasticsearch.Version;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.template.put.PutComposableIndexTemplateAction;
import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.OriginSettingClient;
import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.DeprecationHandler;
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.xpack.core.ml.utils.MlIndexAndAlias;
import org.elasticsearch.xpack.core.template.IndexTemplateConfig;

Expand All @@ -39,43 +46,61 @@ public abstract class AbstractAuditor<T extends AbstractAuditMessage> {

private static final Logger logger = LogManager.getLogger(AbstractAuditor.class);
static final int MAX_BUFFER_SIZE = 1000;
static final TimeValue MASTER_TIMEOUT = TimeValue.timeValueMinutes(1);

private final OriginSettingClient client;
private final String nodeName;
private final String auditIndex;
private final String templateName;
private final Supplier<PutIndexTemplateRequest> templateSupplier;
private final Version versionComposableTemplateExpected;
private final Supplier<PutIndexTemplateRequest> legacyTemplateSupplier;
private final Supplier<PutComposableIndexTemplateAction.Request> templateSupplier;
private final AbstractAuditMessageFactory<T> messageFactory;
private final AtomicBoolean hasLatestTemplate;

private Queue<ToXContent> backlog;
private final ClusterService clusterService;
private final AtomicBoolean putTemplateInProgress;


protected AbstractAuditor(OriginSettingClient client,
String auditIndex,
Version versionComposableTemplateExpected,
IndexTemplateConfig legacyTemplateConfig,
IndexTemplateConfig templateConfig,
String nodeName,
AbstractAuditMessageFactory<T> messageFactory,
ClusterService clusterService) {

this(client, auditIndex, templateConfig.getTemplateName(),
() -> new PutIndexTemplateRequest(templateConfig.getTemplateName()).source(templateConfig.loadBytes(), XContentType.JSON),
this(client, auditIndex, templateConfig.getTemplateName(), versionComposableTemplateExpected,
() -> new PutIndexTemplateRequest(legacyTemplateConfig.getTemplateName())
.source(legacyTemplateConfig.loadBytes(), XContentType.JSON).masterNodeTimeout(MASTER_TIMEOUT),
() -> {
try {
return new PutComposableIndexTemplateAction.Request(templateConfig.getTemplateName())
.indexTemplate(ComposableIndexTemplate.parse(JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY,
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, templateConfig.loadBytes())))
.masterNodeTimeout(MASTER_TIMEOUT);
} catch (IOException e) {
throw new ElasticsearchParseException("unable to parse composable template " + templateConfig.getTemplateName(), e);
}
},
nodeName, messageFactory, clusterService);
}


protected AbstractAuditor(OriginSettingClient client,
String auditIndex,
String templateName,
Supplier<PutIndexTemplateRequest> templateSupplier,
Version versionComposableTemplateExpected,
Supplier<PutIndexTemplateRequest> legacyTemplateSupplier,
Supplier<PutComposableIndexTemplateAction.Request> templateSupplier,
String nodeName,
AbstractAuditMessageFactory<T> messageFactory,
ClusterService clusterService) {
this.client = Objects.requireNonNull(client);
this.auditIndex = Objects.requireNonNull(auditIndex);
this.templateName = Objects.requireNonNull(templateName);
this.versionComposableTemplateExpected = versionComposableTemplateExpected;
this.legacyTemplateSupplier = Objects.requireNonNull(legacyTemplateSupplier);
this.templateSupplier = Objects.requireNonNull(templateSupplier);
this.messageFactory = Objects.requireNonNull(messageFactory);
this.clusterService = Objects.requireNonNull(clusterService);
Expand Down Expand Up @@ -111,7 +136,7 @@ protected void indexDoc(ToXContent toXContent) {
return;
}

if (MlIndexAndAlias.hasIndexTemplate(clusterService.state(), templateName)) {
if (MlIndexAndAlias.hasIndexTemplate(clusterService.state(), templateName, templateName, versionComposableTemplateExpected)) {
synchronized (this) {
// synchronized so nothing can be added to backlog while this value changes
hasLatestTemplate.set(true);
Expand Down Expand Up @@ -152,8 +177,8 @@ protected void indexDoc(ToXContent toXContent) {

// stop multiple invocations
if (putTemplateInProgress.compareAndSet(false, true)) {
MlIndexAndAlias.installIndexTemplateIfRequired(clusterService.state(), client, templateSupplier.get(),
putTemplateListener);
MlIndexAndAlias.installIndexTemplateIfRequired(clusterService.state(), client, versionComposableTemplateExpected,
legacyTemplateSupplier.get(), templateSupplier.get(), putTemplateListener);
}
return;
}
Expand Down
4 changes: 4 additions & 0 deletions x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/MlStatsIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,10 @@ public class MlStatsIndex {

private MlStatsIndex() {}

public static String wrappedMapping() {
return "{\n\"_doc\" : " + mapping() + "\n}";
}

public static String mapping() {
return TemplateUtils.loadTemplate("/org/elasticsearch/xpack/core/ml/stats_index_mappings.json",
Version.CURRENT.toString(), MAPPINGS_VERSION_VARIABLE);
Expand Down
4 changes: 4 additions & 0 deletions .../src/main/java/org/elasticsearch/xpack/core/ml/job/persistence/AnomalyDetectorsIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,10 @@ public static void createStateIndexAndAliasIfNecessary(Client client, ClusterSta
finalListener);
}

public static String wrappedResultsMapping() {
return "{\n\"_doc\" : " + resultsMapping() + "\n}";
}

public static String resultsMapping() {
return TemplateUtils.loadTemplate(RESOURCE_PATH + "results_index_mappings.json",
Version.CURRENT.toString(), RESULTS_MAPPINGS_VERSION_VARIABLE);
Expand Down
11 changes: 11 additions & 0 deletions .../core/src/main/java/org/elasticsearch/xpack/core/ml/notifications/NotificationsIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,20 @@
*/
package org.elasticsearch.xpack.core.ml.notifications;

import org.elasticsearch.Version;
import org.elasticsearch.xpack.core.template.TemplateUtils;

public final class NotificationsIndex {

public static final String NOTIFICATIONS_INDEX = ".ml-notifications-000001";

private static final String RESOURCE_PATH = "/org/elasticsearch/xpack/core/ml/";
private static final String MAPPINGS_VERSION_VARIABLE = "xpack.ml.version";

private NotificationsIndex() {}

public static String mapping() {
return TemplateUtils.loadTemplate(RESOURCE_PATH + "notifications_index_mappings.json",
Version.CURRENT.toString(), MAPPINGS_VERSION_VARIABLE);
}
}
84 changes: 62 additions & 22 deletions x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/utils/MlIndexAndAlias.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,9 @@
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.message.ParameterizedMessage;
import org.elasticsearch.ElasticsearchParseException;
import org.elasticsearch.ResourceAlreadyExistsException;
import org.elasticsearch.Version;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
Expand All @@ -19,20 +21,26 @@
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.template.put.PutComposableIndexTemplateAction;
import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateRequest;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
import org.elasticsearch.cluster.metadata.IndexMetadata;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.DeprecationHandler;
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.indices.SystemIndexDescriptor;
import org.elasticsearch.xpack.core.template.IndexTemplateConfig;

import java.io.IOException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Optional;
Expand Down Expand Up @@ -303,25 +311,40 @@ private static void updateWriteAlias(Client client,
public static void installIndexTemplateIfRequired(
ClusterState clusterState,
Client client,
Version versionComposableTemplateExpected,
IndexTemplateConfig legacyTemplateConfig,
IndexTemplateConfig templateConfig,
TimeValue masterTimeout,
ActionListener<Boolean> listener
) {
String legacyTemplateName = legacyTemplateConfig.getTemplateName();
String templateName = templateConfig.getTemplateName();

// The check for existence of the template is against the cluster state, so very cheap
if (hasIndexTemplate(clusterState, templateName)) {
if (hasIndexTemplate(clusterState, legacyTemplateName, templateName, versionComposableTemplateExpected)) {
listener.onResponse(true);
return;
}

PutIndexTemplateRequest request = new PutIndexTemplateRequest(templateName)
.source(templateConfig.loadBytes(), XContentType.JSON);
PutIndexTemplateRequest legacyRequest = new PutIndexTemplateRequest(legacyTemplateName)
.source(legacyTemplateConfig.loadBytes(), XContentType.JSON).masterNodeTimeout(masterTimeout);

PutComposableIndexTemplateAction.Request request;
try {
request = new PutComposableIndexTemplateAction.Request(templateConfig.getTemplateName())
.indexTemplate(ComposableIndexTemplate.parse(JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY,
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, templateConfig.loadBytes())))
.masterNodeTimeout(masterTimeout);
} catch (IOException e) {
throw new ElasticsearchParseException("unable to parse composable template " + templateConfig.getTemplateName(), e);
}

installIndexTemplateIfRequired(clusterState, client, request, listener);
installIndexTemplateIfRequired(clusterState, client, versionComposableTemplateExpected, legacyRequest, request, listener);
}

/**
* See {@link #installIndexTemplateIfRequired(ClusterState, Client, IndexTemplateConfig, ActionListener)}.
* See {@link #installIndexTemplateIfRequired(ClusterState, Client, Version, IndexTemplateConfig, IndexTemplateConfig, TimeValue,
* ActionListener)}.
*
* Overload takes a {@code PutIndexTemplateRequest} instead of {@code IndexTemplateConfig}
*
Expand All @@ -333,34 +356,51 @@ public static void installIndexTemplateIfRequired(
public static void installIndexTemplateIfRequired(
ClusterState clusterState,
Client client,
PutIndexTemplateRequest templateRequest,
Version versionComposableTemplateExpected,
PutIndexTemplateRequest legacyTemplateRequest,
PutComposableIndexTemplateAction.Request templateRequest,
ActionListener<Boolean> listener
) {
String templateName = templateRequest.name();

// The check for existence of the template is against the cluster state, so very cheap
if (hasIndexTemplate(clusterState, templateRequest.name())) {
if (hasIndexTemplate(clusterState, legacyTemplateRequest.name(), templateRequest.name(), versionComposableTemplateExpected)) {
listener.onResponse(true);
return;
}

templateRequest.masterNodeTimeout(TimeValue.timeValueMinutes(1));
if (versionComposableTemplateExpected != null &&
clusterState.nodes().getMinNodeVersion().onOrAfter(versionComposableTemplateExpected)) {
ActionListener<AcknowledgedResponse> innerListener = ActionListener.wrap(
response -> {
if (response.isAcknowledged() == false) {
logger.warn("error adding template [{}], request was not acknowledged", templateRequest.name());
}
listener.onResponse(response.isAcknowledged());
},
listener::onFailure);

ActionListener<AcknowledgedResponse> innerListener = ActionListener.wrap(
response -> {
if (response.isAcknowledged() == false) {
logger.warn("error adding legacy template [{}], request was not acknowledged", templateName);
}
listener.onResponse(response.isAcknowledged());
},
listener::onFailure);
executeAsyncWithOrigin(client, ML_ORIGIN, PutComposableIndexTemplateAction.INSTANCE, templateRequest, innerListener);
} else {
ActionListener<AcknowledgedResponse> innerListener = ActionListener.wrap(
response -> {
if (response.isAcknowledged() == false) {
logger.warn("error adding legacy template [{}], request was not acknowledged", legacyTemplateRequest.name());
}
listener.onResponse(response.isAcknowledged());
},
listener::onFailure);

executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, templateRequest, innerListener,
client.admin().indices()::putTemplate);
executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, legacyTemplateRequest, innerListener,
client.admin().indices()::putTemplate);
}
}

public static boolean hasIndexTemplate(ClusterState state, String templateName) {
return state.getMetadata().getTemplates().containsKey(templateName);
public static boolean hasIndexTemplate(ClusterState state, String legacyTemplateName,
String templateName, Version versionComposableTemplateExpected) {
if (versionComposableTemplateExpected != null && state.nodes().getMinNodeVersion().onOrAfter(versionComposableTemplateExpected)) {
return state.getMetadata().templatesV2().containsKey(templateName);
} else {
return state.getMetadata().getTemplates().containsKey(legacyTemplateName);
}
}

public static boolean hasIndex(ClusterState state, String index) {
Expand Down
Loading