Skip to content

[ML] Don't treat stale FAILED jobs as OPENING in job allocation #31800

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 2 commits into from
Jul 5, 2018
Merged

[ML] Don't treat stale FAILED jobs as OPENING in job allocation #31800

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ccef6e7
[ML] Don't treat stale FAILED jobs as OPENING in job allocation
Jul 4, 2018
1617e84
Use isAnyOf()
Jul 5, 2018
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
21 changes: 16 additions & 5 deletions x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportOpenJobAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -210,16 +210,27 @@ static PersistentTasksCustomMetaData.Assignment selectLeastLoadedMlNode(String j
for (PersistentTasksCustomMetaData.PersistentTask<?> assignedTask : assignedTasks) {
JobTaskState jobTaskState = (JobTaskState) assignedTask.getState();
JobState jobState;
if (jobTaskState == null || // executor node didn't have the chance to set job status to OPENING
// previous executor node failed and current executor node didn't have the chance to set job status to OPENING
jobTaskState.isStatusStale(assignedTask)) {
if (jobTaskState == null) {
// executor node didn't have the chance to set job status to OPENING
++numberOfAllocatingJobs;
jobState = JobState.OPENING;
} else {
jobState = jobTaskState.getState();
if (jobTaskState.isStatusStale(assignedTask)) {
if (jobState == JobState.CLOSING) {
// previous executor node failed while the job was closing - it won't
// be reopened, so consider it CLOSED for resource usage purposes
jobState = JobState.CLOSED;
} else if (jobState != JobState.FAILED) {
// previous executor node failed and current executor node didn't
// have the chance to set job status to OPENING
++numberOfAllocatingJobs;
jobState = JobState.OPENING;
}
}
}
// Don't count FAILED jobs, as they don't consume native memory
if (jobState != JobState.FAILED) {
// Don't count CLOSED or FAILED jobs, as they don't consume native memory
if (jobState.isAnyOf(JobState.CLOSED, JobState.FAILED) == false) {
++numberOfAssignedJobs;
String assignedJobId = ((OpenJobAction.JobParams) assignedTask.getParams()).getJobId();
Job assignedJob = mlMetadata.getJobs().get(assignedJobId);
Expand Down
58 changes: 53 additions & 5 deletions ...lugin/ml/src/test/java/org/elasticsearch/xpack/ml/action/TransportOpenJobActionTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,6 @@
import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
Expand Down Expand Up @@ -285,7 +284,7 @@ public void testSelectLeastLoadedMlNode_maxConcurrentOpeningJobs() {
nodeAttr, Collections.emptySet(), Version.CURRENT))
.build();

PersistentTasksCustomMetaData.Builder tasksBuilder = PersistentTasksCustomMetaData.builder();
PersistentTasksCustomMetaData.Builder tasksBuilder = PersistentTasksCustomMetaData.builder();
addJobTask("job_id1", "_node_id1", null, tasksBuilder);
addJobTask("job_id2", "_node_id1", null, tasksBuilder);
addJobTask("job_id3", "_node_id2", null, tasksBuilder);
Expand Down Expand Up @@ -340,6 +339,55 @@ public void testSelectLeastLoadedMlNode_maxConcurrentOpeningJobs() {
assertTrue(result.getExplanation().contains("because node exceeds [2] the maximum number of jobs [2] in opening state"));
}

public void testSelectLeastLoadedMlNode_concurrentOpeningJobsAndStaleFailedJob() {
Map<String, String> nodeAttr = new HashMap<>();
nodeAttr.put(MachineLearning.ML_ENABLED_NODE_ATTR, "true");
DiscoveryNodes nodes = DiscoveryNodes.builder()
.add(new DiscoveryNode("_node_name1", "_node_id1", new TransportAddress(InetAddress.getLoopbackAddress(), 9300),
nodeAttr, Collections.emptySet(), Version.CURRENT))
.add(new DiscoveryNode("_node_name2", "_node_id2", new TransportAddress(InetAddress.getLoopbackAddress(), 9301),
nodeAttr, Collections.emptySet(), Version.CURRENT))
.add(new DiscoveryNode("_node_name3", "_node_id3", new TransportAddress(InetAddress.getLoopbackAddress(), 9302),
nodeAttr, Collections.emptySet(), Version.CURRENT))
.build();

PersistentTasksCustomMetaData.Builder tasksBuilder = PersistentTasksCustomMetaData.builder();
addJobTask("job_id1", "_node_id1", JobState.fromString("failed"), tasksBuilder);
// This will make the allocation stale for job_id1
tasksBuilder.reassignTask(MlMetadata.jobTaskId("job_id1"), new Assignment("_node_id1", "test assignment"));
addJobTask("job_id2", "_node_id1", null, tasksBuilder);
addJobTask("job_id3", "_node_id2", null, tasksBuilder);
addJobTask("job_id4", "_node_id2", null, tasksBuilder);
addJobTask("job_id5", "_node_id3", null, tasksBuilder);
addJobTask("job_id6", "_node_id3", null, tasksBuilder);
PersistentTasksCustomMetaData tasks = tasksBuilder.build();

ClusterState.Builder csBuilder = ClusterState.builder(new ClusterName("_name"));
csBuilder.nodes(nodes);
MetaData.Builder metaData = MetaData.builder();
RoutingTable.Builder routingTable = RoutingTable.builder();
addJobAndIndices(metaData, routingTable, "job_id1", "job_id2", "job_id3", "job_id4", "job_id5", "job_id6", "job_id7", "job_id8");
csBuilder.routingTable(routingTable.build());
metaData.putCustom(PersistentTasksCustomMetaData.TYPE, tasks);
csBuilder.metaData(metaData);

ClusterState cs = csBuilder.build();
// Allocation won't be possible if the stale failed job is treated as opening
Assignment result = TransportOpenJobAction.selectLeastLoadedMlNode("job_id7", cs, 2, 10, 30, logger);
assertEquals("_node_id1", result.getExecutorNode());

tasksBuilder = PersistentTasksCustomMetaData.builder(tasks);
addJobTask("job_id7", "_node_id1", null, tasksBuilder);
tasks = tasksBuilder.build();

csBuilder = ClusterState.builder(cs);
csBuilder.metaData(MetaData.builder(cs.metaData()).putCustom(PersistentTasksCustomMetaData.TYPE, tasks));
cs = csBuilder.build();
result = TransportOpenJobAction.selectLeastLoadedMlNode("job_id8", cs, 2, 10, 30, logger);
assertNull("no node selected, because OPENING state", result.getExecutorNode());
assertTrue(result.getExplanation().contains("because node exceeds [2] the maximum number of jobs [2] in opening state"));
}

public void testSelectLeastLoadedMlNode_noCompatibleJobTypeNodes() {
Map<String, String> nodeAttr = new HashMap<>();
nodeAttr.put(MachineLearning.ML_ENABLED_NODE_ATTR, "true");
Expand Down Expand Up @@ -710,13 +758,13 @@ private ClusterState getClusterStateWithMappingsWithMetaData(Map<String, Object>

private static Function<String, Job> jobWithRulesCreator() {
return jobId -> {
DetectionRule rule = new DetectionRule.Builder(Arrays.asList(
DetectionRule rule = new DetectionRule.Builder(Collections.singletonList(
new RuleCondition(RuleCondition.AppliesTo.TYPICAL, Operator.LT, 100.0)
)).build();

Detector.Builder detector = new Detector.Builder("count", null);
detector.setRules(Arrays.asList(rule));
AnalysisConfig.Builder analysisConfig = new AnalysisConfig.Builder(Arrays.asList(detector.build()));
detector.setRules(Collections.singletonList(rule));
AnalysisConfig.Builder analysisConfig = new AnalysisConfig.Builder(Collections.singletonList(detector.build()));
DataDescription.Builder dataDescription = new DataDescription.Builder();
Job.Builder job = new Job.Builder(jobId);
job.setAnalysisConfig(analysisConfig);
Expand Down