honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[ML] Only check licensing in the transport action of the open job and start datafeed apis. 

Original commit: elastic/x-pack-elasticsearch@a59ef8884c
This commit is contained in:
Martijn van Groningen 2017-04-14 08:57:44 +02:00
parent 10390a5e50
commit e93b447b9c
3 changed files with 24 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
plugin/src/main/java/org/elasticsearch/xpack/ml/MachineLearning.java
View File

@ -307,8 +307,8 @@ public class MachineLearning implements ActionPlugin {
new InvalidLicenseEnforcer(settings, licenseState, threadPool, datafeedManager, autodetectProcessManager); new InvalidLicenseEnforcer(settings, licenseState, threadPool, datafeedManager, autodetectProcessManager);
PersistentTasksExecutorRegistry persistentTasksExecutorRegistry = new PersistentTasksExecutorRegistry(Settings.EMPTY, Arrays.asList( PersistentTasksExecutorRegistry persistentTasksExecutorRegistry = new PersistentTasksExecutorRegistry(Settings.EMPTY, Arrays.asList(
new OpenJobAction.OpenJobPersistentTasksExecutor(settings, licenseState, clusterService, autodetectProcessManager), new OpenJobAction.OpenJobPersistentTasksExecutor(settings, clusterService, autodetectProcessManager),
new StartDatafeedAction.StartDatafeedPersistentTasksExecutor(settings, licenseState, datafeedManager) new StartDatafeedAction.StartDatafeedPersistentTasksExecutor(settings, datafeedManager)
)); ));
return Arrays.asList( return Arrays.asList(

32
plugin/src/main/java/org/elasticsearch/xpack/ml/action/OpenJobAction.java
View File

@ -447,15 +447,13 @@ public class OpenJobAction extends Action<OpenJobAction.Request, OpenJobAction.R
public static class OpenJobPersistentTasksExecutor extends PersistentTasksExecutor<JobParams> { public static class OpenJobPersistentTasksExecutor extends PersistentTasksExecutor<JobParams> {
private final AutodetectProcessManager autodetectProcessManager; private final AutodetectProcessManager autodetectProcessManager;
private final XPackLicenseState licenseState;
private final int maxNumberOfOpenJobs; private final int maxNumberOfOpenJobs;
private volatile int maxConcurrentJobAllocations; private volatile int maxConcurrentJobAllocations;
public OpenJobPersistentTasksExecutor(Settings settings, XPackLicenseState licenseState, public OpenJobPersistentTasksExecutor(Settings settings, ClusterService clusterService,
ClusterService clusterService, AutodetectProcessManager autodetectProcessManager) { AutodetectProcessManager autodetectProcessManager) {
super(settings, TASK_NAME, ThreadPool.Names.MANAGEMENT); super(settings, TASK_NAME, ThreadPool.Names.MANAGEMENT);
this.licenseState = licenseState;
this.autodetectProcessManager = autodetectProcessManager; this.autodetectProcessManager = autodetectProcessManager;
this.maxNumberOfOpenJobs = AutodetectProcessManager.MAX_RUNNING_JOBS_PER_NODE.get(settings); this.maxNumberOfOpenJobs = AutodetectProcessManager.MAX_RUNNING_JOBS_PER_NODE.get(settings);
this.maxConcurrentJobAllocations = MachineLearning.CONCURRENT_JOB_ALLOCATIONS.get(settings); this.maxConcurrentJobAllocations = MachineLearning.CONCURRENT_JOB_ALLOCATIONS.get(settings);
@ -470,21 +468,17 @@ public class OpenJobAction extends Action<OpenJobAction.Request, OpenJobAction.R
@Override @Override
public void validate(JobParams params, ClusterState clusterState) { public void validate(JobParams params, ClusterState clusterState) {
if (licenseState.isMachineLearningAllowed()) { // If we already know that we can't find an ml node because all ml nodes are running at capacity or
// If we already know that we can't find an ml node because all ml nodes are running at capacity or // simply because there are no ml nodes in the cluster then we fail quickly here:
// simply because there are no ml nodes in the cluster then we fail quickly here: MlMetadata mlMetadata = clusterState.metaData().custom(MlMetadata.TYPE);
MlMetadata mlMetadata = clusterState.metaData().custom(MlMetadata.TYPE); OpenJobAction.validate(params.getJobId(), mlMetadata);
OpenJobAction.validate(params.getJobId(), mlMetadata); Assignment assignment = selectLeastLoadedMlNode(params.getJobId(), clusterState, maxConcurrentJobAllocations,
Assignment assignment = selectLeastLoadedMlNode(params.getJobId(), clusterState, maxConcurrentJobAllocations, maxNumberOfOpenJobs, logger);
maxNumberOfOpenJobs, logger); if (assignment.getExecutorNode() == null) {
if (assignment.getExecutorNode() == null) { String msg = "Could not open job because no suitable nodes were found, allocation explanation ["
String msg = "Could not open job because no suitable nodes were found, allocation explanation [" + assignment.getExplanation() + "]";
+ assignment.getExplanation() + "]"; logger.warn("[{}] {}", params.getJobId(), msg);
logger.warn("[{}] {}", params.getJobId(), msg); throw new ElasticsearchStatusException(msg, RestStatus.TOO_MANY_REQUESTS);
throw new ElasticsearchStatusException(msg, RestStatus.TOO_MANY_REQUESTS);
}
} else {
throw LicenseUtils.newComplianceException(XPackPlugin.MACHINE_LEARNING);
} }
} }

24
plugin/src/main/java/org/elasticsearch/xpack/ml/action/StartDatafeedAction.java
View File

@ -473,12 +473,10 @@ public class StartDatafeedAction
public static class StartDatafeedPersistentTasksExecutor extends PersistentTasksExecutor<DatafeedParams> { public static class StartDatafeedPersistentTasksExecutor extends PersistentTasksExecutor<DatafeedParams> {
private final DatafeedManager datafeedManager; private final DatafeedManager datafeedManager;
private final XPackLicenseState licenseState;
private final IndexNameExpressionResolver resolver; private final IndexNameExpressionResolver resolver;
public StartDatafeedPersistentTasksExecutor(Settings settings, XPackLicenseState licenseState, DatafeedManager datafeedManager) { public StartDatafeedPersistentTasksExecutor(Settings settings, DatafeedManager datafeedManager) {
super(settings, TASK_NAME, ThreadPool.Names.MANAGEMENT); super(settings, TASK_NAME, ThreadPool.Names.MANAGEMENT);
this.licenseState = licenseState;
this.datafeedManager = datafeedManager; this.datafeedManager = datafeedManager;
this.resolver = new IndexNameExpressionResolver(settings); this.resolver = new IndexNameExpressionResolver(settings);
} }
@ -490,18 +488,14 @@ public class StartDatafeedAction
@Override @Override
public void validate(DatafeedParams params, ClusterState clusterState) { public void validate(DatafeedParams params, ClusterState clusterState) {
if (licenseState.isMachineLearningAllowed()) { MlMetadata mlMetadata = clusterState.metaData().custom(MlMetadata.TYPE);
MlMetadata mlMetadata = clusterState.metaData().custom(MlMetadata.TYPE); PersistentTasksCustomMetaData tasks = clusterState.getMetaData().custom(PersistentTasksCustomMetaData.TYPE);
PersistentTasksCustomMetaData tasks = clusterState.getMetaData().custom(PersistentTasksCustomMetaData.TYPE); StartDatafeedAction.validate(params.getDatafeedId(), mlMetadata, tasks);
StartDatafeedAction.validate(params.getDatafeedId(), mlMetadata, tasks); Assignment assignment = selectNode(logger, params.getDatafeedId(), clusterState, resolver);
Assignment assignment = selectNode(logger, params.getDatafeedId(), clusterState, resolver); if (assignment.getExecutorNode() == null) {
if (assignment.getExecutorNode() == null) { String msg = "No node found to start datafeed [" + params.getDatafeedId()
String msg = "No node found to start datafeed [" + params.getDatafeedId() + "], allocation explanation [" + assignment.getExplanation() + "]";
+ "], allocation explanation [" + assignment.getExplanation() + "]"; throw new ElasticsearchException(msg);
throw new ElasticsearchException(msg);
}
} else {
throw LicenseUtils.newComplianceException(XPackPlugin.MACHINE_LEARNING);
} }
} }