[ML] Create the ML annotations index (#36731)
The ML UI now provides the ability for users to annotate time periods with arbitrary text to add insight to what happened. This change makes the backend create the index for these annotations, together with read and write aliases to make future upgrades possible without adding complexity to the UI. It also adds read and write permission to the index for all ML users (not just admins). The spec for the index is in https://github.com/elastic/kibana/pull/26034/files#diff-c5c6ac3dbb0e7c91b6d127aa06121b2cR7 Relates #33376 Relates elastic/kibana#26034
This commit is contained in:
David Roberts
GitHub
parent
4bc9bffb4d
commit
624307410e
|
|
@ -0,0 +1,244 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.core.ml.annotations;
|
||||
|
||||
import org.elasticsearch.common.ParseField;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.io.stream.Writeable;
|
||||
import org.elasticsearch.common.xcontent.ObjectParser;
|
||||
import org.elasticsearch.common.xcontent.ToXContentObject;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.xpack.core.ml.job.config.Job;
|
||||
import org.elasticsearch.xpack.core.ml.utils.time.TimeUtils;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Date;
|
||||
import java.util.Objects;
|
||||
|
||||
public class Annotation implements ToXContentObject, Writeable {
|
||||
|
||||
public static final ParseField ANNOTATION = new ParseField("annotation");
|
||||
public static final ParseField CREATE_TIME = new ParseField("create_time");
|
||||
public static final ParseField CREATE_USERNAME = new ParseField("create_username");
|
||||
public static final ParseField TIMESTAMP = new ParseField("timestamp");
|
||||
public static final ParseField END_TIMESTAMP = new ParseField("end_timestamp");
|
||||
public static final ParseField MODIFIED_TIME = new ParseField("modified_time");
|
||||
public static final ParseField MODIFIED_USERNAME = new ParseField("modified_username");
|
||||
public static final ParseField TYPE = new ParseField("type");
|
||||
|
||||
public static final ObjectParser<Annotation, Void> PARSER = new ObjectParser<>(TYPE.getPreferredName(), true, Annotation::new);
|
||||
|
||||
static {
|
||||
PARSER.declareString(Annotation::setAnnotation, ANNOTATION);
|
||||
PARSER.declareField(Annotation::setCreateTime,
|
||||
p -> TimeUtils.parseTimeField(p, CREATE_TIME.getPreferredName()), CREATE_TIME, ObjectParser.ValueType.VALUE);
|
||||
PARSER.declareString(Annotation::setCreateUsername, CREATE_USERNAME);
|
||||
PARSER.declareField(Annotation::setTimestamp,
|
||||
p -> TimeUtils.parseTimeField(p, TIMESTAMP.getPreferredName()), TIMESTAMP, ObjectParser.ValueType.VALUE);
|
||||
PARSER.declareField(Annotation::setEndTimestamp,
|
||||
p -> TimeUtils.parseTimeField(p, END_TIMESTAMP.getPreferredName()), END_TIMESTAMP, ObjectParser.ValueType.VALUE);
|
||||
PARSER.declareString(Annotation::setJobId, Job.ID);
|
||||
PARSER.declareField(Annotation::setModifiedTime,
|
||||
p -> TimeUtils.parseTimeField(p, MODIFIED_TIME.getPreferredName()), MODIFIED_TIME, ObjectParser.ValueType.VALUE);
|
||||
PARSER.declareString(Annotation::setModifiedUsername, MODIFIED_USERNAME);
|
||||
PARSER.declareString(Annotation::setType, TYPE);
|
||||
}
|
||||
|
||||
private String annotation;
|
||||
private Date createTime;
|
||||
private String createUsername;
|
||||
private Date timestamp;
|
||||
private Date endTimestamp;
|
||||
/**
|
||||
* Unlike most ML classes, this may be <code>null</code> or wildcarded
|
||||
*/
|
||||
private String jobId;
|
||||
private Date modifiedTime;
|
||||
private String modifiedUsername;
|
||||
private String type;
|
||||
|
||||
private Annotation() {
|
||||
}
|
||||
|
||||
public Annotation(String annotation, Date createTime, String createUsername, Date timestamp, Date endTimestamp, String jobId,
|
||||
Date modifiedTime, String modifiedUsername, String type) {
|
||||
this.annotation = Objects.requireNonNull(annotation);
|
||||
this.createTime = Objects.requireNonNull(createTime);
|
||||
this.createUsername = Objects.requireNonNull(createUsername);
|
||||
this.timestamp = Objects.requireNonNull(timestamp);
|
||||
this.endTimestamp = endTimestamp;
|
||||
this.jobId = jobId;
|
||||
this.modifiedTime = modifiedTime;
|
||||
this.modifiedUsername = modifiedUsername;
|
||||
this.type = Objects.requireNonNull(type);
|
||||
}
|
||||
|
||||
public Annotation(StreamInput in) throws IOException {
|
||||
annotation = in.readString();
|
||||
createTime = new Date(in.readLong());
|
||||
createUsername = in.readString();
|
||||
timestamp = new Date(in.readLong());
|
||||
if (in.readBoolean()) {
|
||||
endTimestamp = new Date(in.readLong());
|
||||
} else {
|
||||
endTimestamp = null;
|
||||
}
|
||||
jobId = in.readOptionalString();
|
||||
if (in.readBoolean()) {
|
||||
modifiedTime = new Date(in.readLong());
|
||||
} else {
|
||||
modifiedTime = null;
|
||||
}
|
||||
modifiedUsername = in.readOptionalString();
|
||||
type = in.readString();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(StreamOutput out) throws IOException {
|
||||
out.writeString(annotation);
|
||||
out.writeLong(createTime.getTime());
|
||||
out.writeString(createUsername);
|
||||
out.writeLong(timestamp.getTime());
|
||||
if (endTimestamp != null) {
|
||||
out.writeBoolean(true);
|
||||
out.writeLong(endTimestamp.getTime());
|
||||
} else {
|
||||
out.writeBoolean(false);
|
||||
}
|
||||
out.writeOptionalString(jobId);
|
||||
if (modifiedTime != null) {
|
||||
out.writeBoolean(true);
|
||||
out.writeLong(modifiedTime.getTime());
|
||||
} else {
|
||||
out.writeBoolean(false);
|
||||
}
|
||||
out.writeOptionalString(modifiedUsername);
|
||||
out.writeString(type);
|
||||
|
||||
}
|
||||
|
||||
public String getAnnotation() {
|
||||
return annotation;
|
||||
}
|
||||
|
||||
public void setAnnotation(String annotation) {
|
||||
this.annotation = Objects.requireNonNull(annotation);
|
||||
}
|
||||
|
||||
public Date getCreateTime() {
|
||||
return createTime;
|
||||
}
|
||||
|
||||
public void setCreateTime(Date createTime) {
|
||||
this.createTime = Objects.requireNonNull(createTime);
|
||||
}
|
||||
|
||||
public String getCreateUsername() {
|
||||
return createUsername;
|
||||
}
|
||||
|
||||
public void setCreateUsername(String createUsername) {
|
||||
this.createUsername = Objects.requireNonNull(createUsername);
|
||||
}
|
||||
|
||||
public Date getTimestamp() {
|
||||
return timestamp;
|
||||
}
|
||||
|
||||
public void setTimestamp(Date timestamp) {
|
||||
this.timestamp = Objects.requireNonNull(timestamp);
|
||||
}
|
||||
|
||||
public Date getEndTimestamp() {
|
||||
return endTimestamp;
|
||||
}
|
||||
|
||||
public void setEndTimestamp(Date endTimestamp) {
|
||||
this.endTimestamp = endTimestamp;
|
||||
}
|
||||
|
||||
public String getJobId() {
|
||||
return jobId;
|
||||
}
|
||||
|
||||
public void setJobId(String jobId) {
|
||||
this.jobId = jobId;
|
||||
}
|
||||
|
||||
public Date getModifiedTime() {
|
||||
return modifiedTime;
|
||||
}
|
||||
|
||||
public void setModifiedTime(Date modifiedTime) {
|
||||
this.modifiedTime = modifiedTime;
|
||||
}
|
||||
|
||||
public String getModifiedUsername() {
|
||||
return modifiedUsername;
|
||||
}
|
||||
|
||||
public void setModifiedUsername(String modifiedUsername) {
|
||||
this.modifiedUsername = modifiedUsername;
|
||||
}
|
||||
|
||||
public String getType() {
|
||||
return type;
|
||||
}
|
||||
|
||||
public void setType(String type) {
|
||||
this.type = Objects.requireNonNull(type);
|
||||
}
|
||||
|
||||
@Override
|
||||
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
|
||||
builder.startObject();
|
||||
builder.field(ANNOTATION.getPreferredName(), annotation);
|
||||
builder.timeField(CREATE_TIME.getPreferredName(), CREATE_TIME.getPreferredName() + "_string", createTime.getTime());
|
||||
builder.field(CREATE_USERNAME.getPreferredName(), createUsername);
|
||||
builder.timeField(TIMESTAMP.getPreferredName(), TIMESTAMP.getPreferredName() + "_string", timestamp.getTime());
|
||||
if (endTimestamp != null) {
|
||||
builder.timeField(END_TIMESTAMP.getPreferredName(), END_TIMESTAMP.getPreferredName() + "_string", endTimestamp.getTime());
|
||||
}
|
||||
if (jobId != null) {
|
||||
builder.field(Job.ID.getPreferredName(), jobId);
|
||||
}
|
||||
if (modifiedTime != null) {
|
||||
builder.timeField(MODIFIED_TIME.getPreferredName(), MODIFIED_TIME.getPreferredName() + "_string", modifiedTime.getTime());
|
||||
}
|
||||
if (modifiedUsername != null) {
|
||||
builder.field(MODIFIED_USERNAME.getPreferredName(), modifiedUsername);
|
||||
}
|
||||
builder.field(TYPE.getPreferredName(), type);
|
||||
builder.endObject();
|
||||
return builder;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(annotation, createTime, createUsername, timestamp, endTimestamp, jobId, modifiedTime, modifiedUsername, type);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if (obj == this) {
|
||||
return true;
|
||||
}
|
||||
if (obj == null || getClass() != obj.getClass()) {
|
||||
return false;
|
||||
}
|
||||
Annotation other = (Annotation) obj;
|
||||
return Objects.equals(annotation, other.annotation) &&
|
||||
Objects.equals(createTime, other.createTime) &&
|
||||
Objects.equals(createUsername, other.createUsername) &&
|
||||
Objects.equals(timestamp, other.timestamp) &&
|
||||
Objects.equals(endTimestamp, other.endTimestamp) &&
|
||||
Objects.equals(jobId, other.jobId) &&
|
||||
Objects.equals(modifiedTime, other.modifiedTime) &&
|
||||
Objects.equals(modifiedUsername, other.modifiedUsername) &&
|
||||
Objects.equals(type, other.type);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,147 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.core.ml.annotations;
|
||||
|
||||
import org.elasticsearch.ResourceAlreadyExistsException;
|
||||
import org.elasticsearch.action.ActionListener;
|
||||
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
|
||||
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
|
||||
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
|
||||
import org.elasticsearch.action.support.master.AcknowledgedResponse;
|
||||
import org.elasticsearch.client.Client;
|
||||
import org.elasticsearch.cluster.ClusterState;
|
||||
import org.elasticsearch.cluster.metadata.AliasOrIndex;
|
||||
import org.elasticsearch.cluster.metadata.IndexMetaData;
|
||||
import org.elasticsearch.cluster.routing.UnassignedInfo;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.xpack.core.ml.MachineLearningField;
|
||||
import org.elasticsearch.xpack.core.ml.job.config.Job;
|
||||
import org.elasticsearch.xpack.core.ml.job.persistence.ElasticsearchMappings;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.SortedMap;
|
||||
|
||||
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
|
||||
import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
|
||||
import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;
|
||||
|
||||
public class AnnotationIndex {
|
||||
|
||||
public static final String READ_ALIAS_NAME = ".ml-annotations-read";
|
||||
public static final String WRITE_ALIAS_NAME = ".ml-annotations-write";
|
||||
// Exposed for testing, but always use the aliases in non-test code
|
||||
public static final String INDEX_NAME = ".ml-annotations-6";
|
||||
|
||||
/**
|
||||
* Create the .ml-annotations index with correct mappings.
|
||||
* This index is read and written by the UI results views,
|
||||
* so needs to exist when there might be ML results to view.
|
||||
*/
|
||||
public static void createAnnotationsIndex(Settings settings, Client client, ClusterState state,
|
||||
final ActionListener<Boolean> finalListener) {
|
||||
|
||||
final ActionListener<Boolean> createAliasListener = ActionListener.wrap(success -> {
|
||||
final IndicesAliasesRequest request = client.admin().indices().prepareAliases()
|
||||
.addAlias(INDEX_NAME, READ_ALIAS_NAME)
|
||||
.addAlias(INDEX_NAME, WRITE_ALIAS_NAME).request();
|
||||
executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, request,
|
||||
ActionListener.<AcknowledgedResponse>wrap(r -> finalListener.onResponse(r.isAcknowledged()), finalListener::onFailure),
|
||||
client.admin().indices()::aliases);
|
||||
}, finalListener::onFailure);
|
||||
|
||||
// Only create the index or aliases if some other ML index exists - saves clutter if ML is never used.
|
||||
SortedMap<String, AliasOrIndex> mlLookup = state.getMetaData().getAliasAndIndexLookup().tailMap(".ml");
|
||||
if (mlLookup.isEmpty() == false && mlLookup.firstKey().startsWith(".ml")) {
|
||||
|
||||
// Create the annotations index if it doesn't exist already.
|
||||
if (mlLookup.containsKey(INDEX_NAME) == false) {
|
||||
|
||||
final TimeValue delayedNodeTimeOutSetting;
|
||||
// Whether we are using native process is a good way to detect whether we are in dev / test mode:
|
||||
if (MachineLearningField.AUTODETECT_PROCESS.get(settings)) {
|
||||
delayedNodeTimeOutSetting = UnassignedInfo.INDEX_DELAYED_NODE_LEFT_TIMEOUT_SETTING.get(settings);
|
||||
} else {
|
||||
delayedNodeTimeOutSetting = TimeValue.ZERO;
|
||||
}
|
||||
|
||||
CreateIndexRequest createIndexRequest = new CreateIndexRequest(INDEX_NAME);
|
||||
try (XContentBuilder annotationsMapping = AnnotationIndex.annotationsMapping()) {
|
||||
createIndexRequest.mapping(ElasticsearchMappings.DOC_TYPE, annotationsMapping);
|
||||
createIndexRequest.settings(Settings.builder()
|
||||
.put(IndexMetaData.SETTING_AUTO_EXPAND_REPLICAS, "0-1")
|
||||
.put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, "1")
|
||||
.put(UnassignedInfo.INDEX_DELAYED_NODE_LEFT_TIMEOUT_SETTING.getKey(), delayedNodeTimeOutSetting));
|
||||
|
||||
executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, createIndexRequest,
|
||||
ActionListener.<CreateIndexResponse>wrap(
|
||||
r -> createAliasListener.onResponse(r.isAcknowledged()),
|
||||
e -> {
|
||||
// Possible that the index was created while the request was executing,
|
||||
// so we need to handle that possibility
|
||||
if (e instanceof ResourceAlreadyExistsException) {
|
||||
// Create the alias
|
||||
createAliasListener.onResponse(true);
|
||||
} else {
|
||||
finalListener.onFailure(e);
|
||||
}
|
||||
}
|
||||
), client.admin().indices()::create);
|
||||
} catch (IOException e) {
|
||||
finalListener.onFailure(e);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// Recreate the aliases if they've gone even though the index still exists.
|
||||
if (mlLookup.containsKey(READ_ALIAS_NAME) == false || mlLookup.containsKey(WRITE_ALIAS_NAME) == false) {
|
||||
createAliasListener.onResponse(true);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Nothing to do, but respond to the listener
|
||||
finalListener.onResponse(false);
|
||||
}
|
||||
|
||||
public static XContentBuilder annotationsMapping() throws IOException {
|
||||
return jsonBuilder()
|
||||
.startObject()
|
||||
.startObject(ElasticsearchMappings.DOC_TYPE)
|
||||
.startObject(ElasticsearchMappings.PROPERTIES)
|
||||
.startObject(Annotation.ANNOTATION.getPreferredName())
|
||||
.field(ElasticsearchMappings.TYPE, ElasticsearchMappings.TEXT)
|
||||
.endObject()
|
||||
.startObject(Annotation.CREATE_TIME.getPreferredName())
|
||||
.field(ElasticsearchMappings.TYPE, ElasticsearchMappings.DATE)
|
||||
.endObject()
|
||||
.startObject(Annotation.CREATE_USERNAME.getPreferredName())
|
||||
.field(ElasticsearchMappings.TYPE, ElasticsearchMappings.KEYWORD)
|
||||
.endObject()
|
||||
.startObject(Annotation.TIMESTAMP.getPreferredName())
|
||||
.field(ElasticsearchMappings.TYPE, ElasticsearchMappings.DATE)
|
||||
.endObject()
|
||||
.startObject(Annotation.END_TIMESTAMP.getPreferredName())
|
||||
.field(ElasticsearchMappings.TYPE, ElasticsearchMappings.DATE)
|
||||
.endObject()
|
||||
.startObject(Job.ID.getPreferredName())
|
||||
.field(ElasticsearchMappings.TYPE, ElasticsearchMappings.KEYWORD)
|
||||
.endObject()
|
||||
.startObject(Annotation.MODIFIED_TIME.getPreferredName())
|
||||
.field(ElasticsearchMappings.TYPE, ElasticsearchMappings.DATE)
|
||||
.endObject()
|
||||
.startObject(Annotation.MODIFIED_USERNAME.getPreferredName())
|
||||
.field(ElasticsearchMappings.TYPE, ElasticsearchMappings.KEYWORD)
|
||||
.endObject()
|
||||
.startObject(Annotation.TYPE.getPreferredName())
|
||||
.field(ElasticsearchMappings.TYPE, ElasticsearchMappings.KEYWORD)
|
||||
.endObject()
|
||||
.endObject()
|
||||
.endObject()
|
||||
.endObject();
|
||||
}
|
||||
}
|
||||
|
|
@ -136,13 +136,22 @@ public class ReservedRolesStore implements BiConsumer<Set<String>, ActionListene
|
|||
.put(UsernamesField.APM_ROLE, new RoleDescriptor(UsernamesField.APM_ROLE,
|
||||
new String[] { "monitor", MonitoringBulkAction.NAME}, null, null, MetadataUtils.DEFAULT_RESERVED_METADATA))
|
||||
.put("machine_learning_user", new RoleDescriptor("machine_learning_user", new String[] { "monitor_ml" },
|
||||
new RoleDescriptor.IndicesPrivileges[] { RoleDescriptor.IndicesPrivileges.builder().indices(".ml-anomalies*",
|
||||
".ml-notifications").privileges("view_index_metadata", "read").build() },
|
||||
new RoleDescriptor.IndicesPrivileges[] {
|
||||
RoleDescriptor.IndicesPrivileges.builder().indices(".ml-anomalies*", ".ml-notifications*")
|
||||
.privileges("view_index_metadata", "read").build(),
|
||||
RoleDescriptor.IndicesPrivileges.builder().indices(".ml-annotations*")
|
||||
.privileges("view_index_metadata", "read", "write").build()
|
||||
},
|
||||
null, MetadataUtils.DEFAULT_RESERVED_METADATA))
|
||||
.put("machine_learning_admin", new RoleDescriptor("machine_learning_admin", new String[] { "manage_ml" },
|
||||
new RoleDescriptor.IndicesPrivileges[] {
|
||||
RoleDescriptor.IndicesPrivileges.builder().indices(".ml-*").privileges("view_index_metadata", "read")
|
||||
.build() }, null, MetadataUtils.DEFAULT_RESERVED_METADATA))
|
||||
RoleDescriptor.IndicesPrivileges.builder()
|
||||
.indices(".ml-anomalies*", ".ml-notifications*", ".ml-state*", ".ml-meta*")
|
||||
.privileges("view_index_metadata", "read").build(),
|
||||
RoleDescriptor.IndicesPrivileges.builder().indices(".ml-annotations*")
|
||||
.privileges("view_index_metadata", "read", "write").build()
|
||||
},
|
||||
null, MetadataUtils.DEFAULT_RESERVED_METADATA))
|
||||
.put("watcher_admin", new RoleDescriptor("watcher_admin", new String[] { "manage_watcher" },
|
||||
new RoleDescriptor.IndicesPrivileges[] {
|
||||
RoleDescriptor.IndicesPrivileges.builder().indices(Watch.INDEX, TriggeredWatchStoreField.INDEX_NAME,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,38 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.core.ml.annotations;
|
||||
|
||||
import org.elasticsearch.common.io.stream.Writeable;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
import org.elasticsearch.test.AbstractSerializingTestCase;
|
||||
|
||||
import java.util.Date;
|
||||
|
||||
public class AnnotationTests extends AbstractSerializingTestCase<Annotation> {
|
||||
|
||||
@Override
|
||||
protected Annotation doParseInstance(XContentParser parser) {
|
||||
return Annotation.PARSER.apply(parser, null);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Annotation createTestInstance() {
|
||||
return new Annotation(randomAlphaOfLengthBetween(100, 1000),
|
||||
new Date(randomNonNegativeLong()),
|
||||
randomAlphaOfLengthBetween(5, 20),
|
||||
new Date(randomNonNegativeLong()),
|
||||
randomBoolean() ? new Date(randomNonNegativeLong()) : null,
|
||||
randomBoolean() ? randomAlphaOfLengthBetween(10, 30) : null,
|
||||
randomBoolean() ? new Date(randomNonNegativeLong()) : null,
|
||||
randomBoolean() ? randomAlphaOfLengthBetween(5, 20) : null,
|
||||
randomAlphaOfLengthBetween(10, 15));
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Writeable.Reader<Annotation> instanceReader() {
|
||||
return Annotation::new;
|
||||
}
|
||||
}
|
||||
|
|
@ -424,7 +424,7 @@ public class MachineLearning extends Plugin implements ActionPlugin, AnalysisPlu
|
|||
jobResultsProvider,
|
||||
jobManager,
|
||||
autodetectProcessManager,
|
||||
new MlInitializationService(threadPool, clusterService, client),
|
||||
new MlInitializationService(settings, threadPool, clusterService, client),
|
||||
jobDataCountsPersister,
|
||||
datafeedManager,
|
||||
auditor,
|
||||
|
|
|
|||
|
|
@ -5,23 +5,32 @@
|
|||
*/
|
||||
package org.elasticsearch.xpack.ml;
|
||||
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
import org.elasticsearch.action.ActionListener;
|
||||
import org.elasticsearch.client.Client;
|
||||
import org.elasticsearch.cluster.ClusterChangedEvent;
|
||||
import org.elasticsearch.cluster.ClusterStateListener;
|
||||
import org.elasticsearch.cluster.service.ClusterService;
|
||||
import org.elasticsearch.common.component.LifecycleListener;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.gateway.GatewayService;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.xpack.core.ml.annotations.AnnotationIndex;
|
||||
|
||||
class MlInitializationService implements ClusterStateListener {
|
||||
|
||||
private static final Logger logger = LogManager.getLogger(MlInitializationService.class);
|
||||
|
||||
private final Settings settings;
|
||||
private final ThreadPool threadPool;
|
||||
private final ClusterService clusterService;
|
||||
private final Client client;
|
||||
|
||||
private volatile MlDailyMaintenanceService mlDailyMaintenanceService;
|
||||
|
||||
MlInitializationService(ThreadPool threadPool, ClusterService clusterService, Client client) {
|
||||
MlInitializationService(Settings settings, ThreadPool threadPool, ClusterService clusterService, Client client) {
|
||||
this.settings = settings;
|
||||
this.threadPool = threadPool;
|
||||
this.clusterService = clusterService;
|
||||
this.client = client;
|
||||
|
|
@ -37,6 +46,13 @@ class MlInitializationService implements ClusterStateListener {
|
|||
|
||||
if (event.localNodeMaster()) {
|
||||
installDailyMaintenanceService();
|
||||
AnnotationIndex.createAnnotationsIndex(settings, client, event.state(), ActionListener.wrap(
|
||||
r -> {
|
||||
if (r) {
|
||||
logger.info("Created ML annotations index and aliases");
|
||||
}
|
||||
},
|
||||
e -> logger.error("Error creating ML annotations index or aliases", e)));
|
||||
} else {
|
||||
uninstallDailyMaintenanceService();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ import org.elasticsearch.cluster.metadata.MetaData;
|
|||
import org.elasticsearch.cluster.node.DiscoveryNode;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNodes;
|
||||
import org.elasticsearch.cluster.service.ClusterService;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.transport.TransportAddress;
|
||||
import org.elasticsearch.test.ESTestCase;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
|
|
@ -61,7 +62,7 @@ public class MlInitializationServiceTests extends ESTestCase {
|
|||
}
|
||||
|
||||
public void testInitialize() {
|
||||
MlInitializationService initializationService = new MlInitializationService(threadPool, clusterService, client);
|
||||
MlInitializationService initializationService = new MlInitializationService(Settings.EMPTY, threadPool, clusterService, client);
|
||||
|
||||
ClusterState cs = ClusterState.builder(new ClusterName("_name"))
|
||||
.nodes(DiscoveryNodes.builder()
|
||||
|
|
@ -76,7 +77,7 @@ public class MlInitializationServiceTests extends ESTestCase {
|
|||
}
|
||||
|
||||
public void testInitialize_noMasterNode() {
|
||||
MlInitializationService initializationService = new MlInitializationService(threadPool, clusterService, client);
|
||||
MlInitializationService initializationService = new MlInitializationService(Settings.EMPTY, threadPool, clusterService, client);
|
||||
|
||||
ClusterState cs = ClusterState.builder(new ClusterName("_name"))
|
||||
.nodes(DiscoveryNodes.builder()
|
||||
|
|
@ -89,7 +90,7 @@ public class MlInitializationServiceTests extends ESTestCase {
|
|||
}
|
||||
|
||||
public void testInitialize_alreadyInitialized() {
|
||||
MlInitializationService initializationService = new MlInitializationService(threadPool, clusterService, client);
|
||||
MlInitializationService initializationService = new MlInitializationService(Settings.EMPTY, threadPool, clusterService, client);
|
||||
|
||||
ClusterState cs = ClusterState.builder(new ClusterName("_name"))
|
||||
.nodes(DiscoveryNodes.builder()
|
||||
|
|
@ -107,7 +108,7 @@ public class MlInitializationServiceTests extends ESTestCase {
|
|||
}
|
||||
|
||||
public void testNodeGoesFromMasterToNonMasterAndBack() {
|
||||
MlInitializationService initializationService = new MlInitializationService(threadPool, clusterService, client);
|
||||
MlInitializationService initializationService = new MlInitializationService(Settings.EMPTY, threadPool, clusterService, client);
|
||||
MlDailyMaintenanceService initialDailyMaintenanceService = mock(MlDailyMaintenanceService.class);
|
||||
initializationService.setDailyMaintenanceService(initialDailyMaintenanceService);
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,91 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.ml.integration;
|
||||
|
||||
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
|
||||
import org.elasticsearch.cluster.ClusterState;
|
||||
import org.elasticsearch.cluster.metadata.AliasMetaData;
|
||||
import org.elasticsearch.common.collect.ImmutableOpenMap;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.plugins.Plugin;
|
||||
import org.elasticsearch.xpack.core.XPackSettings;
|
||||
import org.elasticsearch.xpack.core.ml.annotations.AnnotationIndex;
|
||||
import org.elasticsearch.xpack.ml.LocalStateMachineLearning;
|
||||
import org.elasticsearch.xpack.ml.MachineLearning;
|
||||
import org.elasticsearch.xpack.ml.MlSingleNodeTestCase;
|
||||
import org.elasticsearch.xpack.ml.notifications.Auditor;
|
||||
import org.junit.Before;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
|
||||
public class AnnotationIndexIT extends MlSingleNodeTestCase {
|
||||
|
||||
@Override
|
||||
protected Settings nodeSettings() {
|
||||
Settings.Builder newSettings = Settings.builder();
|
||||
newSettings.put(super.nodeSettings());
|
||||
newSettings.put(XPackSettings.MONITORING_ENABLED.getKey(), false);
|
||||
newSettings.put(XPackSettings.SECURITY_ENABLED.getKey(), false);
|
||||
newSettings.put(XPackSettings.WATCHER_ENABLED.getKey(), false);
|
||||
return newSettings.build();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Collection<Class<? extends Plugin>> getPlugins() {
|
||||
return pluginList(LocalStateMachineLearning.class);
|
||||
}
|
||||
|
||||
// TODO remove this when the jindex feature branches are merged, as this is in the base class then
|
||||
@Before
|
||||
public void waitForMlTemplates() throws Exception {
|
||||
// Block until the templates are installed
|
||||
assertBusy(() -> {
|
||||
ClusterState state = client().admin().cluster().prepareState().get().getState();
|
||||
assertTrue("Timed out waiting for the ML templates to be installed",
|
||||
MachineLearning.allTemplatesInstalled(state));
|
||||
});
|
||||
}
|
||||
|
||||
public void testNotCreatedWhenNoOtherMlIndices() {
|
||||
|
||||
// Ask a few times to increase the chance of failure if the .ml-annotations index is created when no other ML index exists
|
||||
for (int i = 0; i < 10; ++i) {
|
||||
assertFalse(annotationsIndexExists());
|
||||
assertEquals(0, numberOfAnnotationsAliases());
|
||||
}
|
||||
}
|
||||
|
||||
public void testCreatedWhenAfterOtherMlIndex() throws Exception {
|
||||
|
||||
Auditor auditor = new Auditor(client(), "node_1");
|
||||
auditor.info("whatever", "blah");
|
||||
|
||||
// Creating a document in the .ml-notifications index should cause .ml-annotations
|
||||
// to be created, as it should get created as soon as any other ML index exists
|
||||
|
||||
assertBusy(() -> {
|
||||
assertTrue(annotationsIndexExists());
|
||||
assertEquals(2, numberOfAnnotationsAliases());
|
||||
});
|
||||
}
|
||||
|
||||
private boolean annotationsIndexExists() {
|
||||
return client().admin().indices().prepareExists(AnnotationIndex.INDEX_NAME).get().isExists();
|
||||
}
|
||||
|
||||
private int numberOfAnnotationsAliases() {
|
||||
int count = 0;
|
||||
ImmutableOpenMap<String, List<AliasMetaData>> aliases = client().admin().indices()
|
||||
.prepareGetAliases(AnnotationIndex.READ_ALIAS_NAME, AnnotationIndex.WRITE_ALIAS_NAME).get().getAliases();
|
||||
if (aliases != null) {
|
||||
for (ObjectObjectCursor<String, List<AliasMetaData>> entry : aliases) {
|
||||
count += entry.value.size();
|
||||
}
|
||||
}
|
||||
return count;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue