Skip to content

Commit

Permalink
[ML] Job In Index: Migrate config from the clusterstate (#35834)
Browse files Browse the repository at this point in the history 
Migrate ML configuration from clusterstate to index for closed jobs
only once all nodes are v6.6.0 or higher
  • Loading branch information
@davidkyle
davidkyle committed Dec 5, 2018
1 parent 29fc10d commit 81549e6
Show file tree
Hide file tree
Showing 16 changed files with 980 additions and 105 deletions.
4 changes: 2 additions & 2 deletions x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/MlMetadata.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -405,14 +405,14 @@ private void checkDatafeedIsStopped(Supplier<String> msg, String datafeedId, Per
}
}

private Builder putJobs(Collection<Job> jobs) {
public Builder putJobs(Collection<Job> jobs) {
for (Job job : jobs) {
putJob(job, true);
}
return this;
}

private Builder putDatafeeds(Collection<DatafeedConfig> datafeeds) {
public Builder putDatafeeds(Collection<DatafeedConfig> datafeeds) {
for (DatafeedConfig datafeed : datafeeds) {
this.datafeeds.put(datafeed.getId(), datafeed);
}
Expand Down
17 changes: 17 additions & 0 deletions x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/MlTasks.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,23 @@ public static Set<String> openJobIds(@Nullable PersistentTasksCustomMetaData tas
.collect(Collectors.toSet());
}

/**
* The datafeed Ids of started datafeed tasks
*
* @param tasks Persistent tasks. If null an empty set is returned.
* @return The Ids of running datafeed tasks
*/
public static Set<String> startedDatafeedIds(@Nullable PersistentTasksCustomMetaData tasks) {
if (tasks == null) {
return Collections.emptySet();
}

return tasks.findTasks(DATAFEED_TASK_NAME, task -> true)
.stream()
.map(t -> t.getId().substring(DATAFEED_TASK_ID_PREFIX.length()))
.collect(Collectors.toSet());
}

/**
* Is there an ml anomaly detector job task for the job {@code jobId}?
* @param jobId The job id
Expand Down
20 changes: 20 additions & 0 deletions x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/MlTasksTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,26 @@ public void testOpenJobIds_GivenNull() {
assertThat(MlTasks.openJobIds(null), empty());
}

public void testStartedDatafeedIds() {
PersistentTasksCustomMetaData.Builder tasksBuilder = PersistentTasksCustomMetaData.builder();
assertThat(MlTasks.openJobIds(tasksBuilder.build()), empty());

tasksBuilder.addTask(MlTasks.jobTaskId("job-1"), MlTasks.JOB_TASK_NAME, new OpenJobAction.JobParams("foo-1"),
new PersistentTasksCustomMetaData.Assignment("node-1", "test assignment"));
tasksBuilder.addTask(MlTasks.datafeedTaskId("df1"), MlTasks.DATAFEED_TASK_NAME,
new StartDatafeedAction.DatafeedParams("df1", 0L),
new PersistentTasksCustomMetaData.Assignment("node-1", "test assignment"));
tasksBuilder.addTask(MlTasks.datafeedTaskId("df2"), MlTasks.DATAFEED_TASK_NAME,
new StartDatafeedAction.DatafeedParams("df2", 0L),
new PersistentTasksCustomMetaData.Assignment("node-2", "test assignment"));

assertThat(MlTasks.startedDatafeedIds(tasksBuilder.build()), containsInAnyOrder("df1", "df2"));
}

public void testStartedDatafeedIds_GivenNull() {
assertThat(MlTasks.startedDatafeedIds(null), empty());
}

public void testTaskExistsForJob() {
PersistentTasksCustomMetaData.Builder tasksBuilder = PersistentTasksCustomMetaData.builder();
assertFalse(MlTasks.taskExistsForJob("job-1", tasksBuilder.build()));
Expand Down
4 changes: 2 additions & 2 deletions x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MachineLearning.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -438,7 +438,7 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
jobDataCountsPersister,
datafeedManager,
auditor,
new MlAssignmentNotifier(auditor, clusterService),
new MlAssignmentNotifier(auditor, threadPool, client, clusterService),
memoryTracker
);
}
Expand All @@ -453,7 +453,7 @@ public List<PersistentTasksExecutor<?>> getPersistentTasksExecutor(ClusterServic

return Arrays.asList(
new TransportOpenJobAction.OpenJobPersistentTasksExecutor(settings, clusterService, autodetectProcessManager.get(),
memoryTracker.get()),
memoryTracker.get(), client),
new TransportStartDatafeedAction.StartDatafeedPersistentTasksExecutor( datafeedManager.get())
);
}
Expand Down
42 changes: 38 additions & 4 deletions x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MlAssignmentNotifier.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,11 @@

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.Version;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterChangedEvent;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.ClusterStateListener;
import org.elasticsearch.cluster.LocalNodeMasterListener;
import org.elasticsearch.cluster.node.DiscoveryNode;
Expand All @@ -30,12 +34,23 @@ public class MlAssignmentNotifier implements ClusterStateListener, LocalNodeMast

private final Auditor auditor;
private final ClusterService clusterService;

private final MlConfigMigrator mlConfigMigrator;
private final ThreadPool threadPool;
private final AtomicBoolean enabled = new AtomicBoolean(false);

MlAssignmentNotifier(Auditor auditor, ClusterService clusterService) {
MlAssignmentNotifier(Auditor auditor, ThreadPool threadPool, Client client, ClusterService clusterService) {
this.auditor = auditor;
this.clusterService = clusterService;
this.mlConfigMigrator = new MlConfigMigrator(client, clusterService);
this.threadPool = threadPool;
clusterService.addLocalNodeMasterListener(this);
}

MlAssignmentNotifier(Auditor auditor, ThreadPool threadPool, MlConfigMigrator mlConfigMigrator, ClusterService clusterService) {
this.auditor = auditor;
this.clusterService = clusterService;
this.mlConfigMigrator = mlConfigMigrator;
this.threadPool = threadPool;
clusterService.addLocalNodeMasterListener(this);
}

Expand Down Expand Up @@ -72,6 +87,25 @@ public void clusterChanged(ClusterChangedEvent event) {
return;
}

Version minNodeVersion = event.state().nodes().getMinNodeVersion();
if (minNodeVersion.onOrAfter(Version.V_6_6_0)) {
// ok to migrate
mlConfigMigrator.migrateConfigsWithoutTasks(event.state(), ActionListener.wrap(
response -> threadPool.executor(executorName()).execute(() -> auditChangesToMlTasks(current, previous, event.state())),
e -> {
logger.error("error migrating ml configurations", e);
threadPool.executor(executorName()).execute(() -> auditChangesToMlTasks(current, previous, event.state()));
}
));
} else {
threadPool.executor(executorName()).execute(() -> auditChangesToMlTasks(current, previous, event.state()));
}

}

private void auditChangesToMlTasks(PersistentTasksCustomMetaData current, PersistentTasksCustomMetaData previous,
ClusterState state) {

for (PersistentTask<?> currentTask : current.tasks()) {
Assignment currentAssignment = currentTask.getAssignment();
PersistentTask<?> previousTask = previous != null ? previous.getTask(currentTask.getId()) : null;
Expand All @@ -84,7 +118,7 @@ public void clusterChanged(ClusterChangedEvent event) {
if (currentAssignment.getExecutorNode() == null) {
auditor.warning(jobId, "No node found to open job. Reasons [" + currentAssignment.getExplanation() + "]");
} else {
DiscoveryNode node = event.state().nodes().get(currentAssignment.getExecutorNode());
DiscoveryNode node = state.nodes().get(currentAssignment.getExecutorNode());
auditor.info(jobId, "Opening job on node [" + node.toString() + "]");
}
} else if (MlTasks.DATAFEED_TASK_NAME.equals(currentTask.getTaskName())) {
Expand All @@ -98,7 +132,7 @@ public void clusterChanged(ClusterChangedEvent event) {
auditor.warning(jobId, msg);
}
} else {
DiscoveryNode node = event.state().nodes().get(currentAssignment.getExecutorNode());
DiscoveryNode node = state.nodes().get(currentAssignment.getExecutorNode());
if (jobId != null) {
auditor.info(jobId, "Starting datafeed [" + datafeedParams.getDatafeedId() + "] on node [" + node + "]");
}
Expand Down
Loading

0 comments on commit 81549e6

Please sign in to comment.