From 6360cab08cdbb403302b55814b40fa3cf16862be Mon Sep 17 00:00:00 2001 From: David Kyle Date: Thu, 26 Apr 2018 15:19:54 +0100 Subject: [PATCH 1/5] Job manager --- .../elasticsearch/xpack/core/ml/MlTasks.java | 24 +- .../xpack/ml/MachineLearning.java | 2 +- .../action/TransportDeleteCalendarAction.java | 6 +- .../TransportDeleteCalendarEventAction.java | 6 +- .../TransportDeleteModelSnapshotAction.java | 61 +-- .../ml/action/TransportForecastJobAction.java | 96 ++-- .../ml/action/TransportGetBucketsAction.java | 47 +- .../action/TransportGetCategoriesAction.java | 15 +- .../action/TransportGetInfluencersAction.java | 28 +- .../ml/action/TransportGetJobsAction.java | 12 +- .../TransportGetModelSnapshotsAction.java | 18 +- .../TransportGetOverallBucketsAction.java | 32 +- .../ml/action/TransportGetRecordsAction.java | 29 +- .../ml/action/TransportJobTaskAction.java | 4 - .../TransportPostCalendarEventsAction.java | 6 +- .../TransportRevertModelSnapshotAction.java | 33 +- .../TransportUpdateCalendarJobAction.java | 6 +- .../action/TransportUpdateFilterAction.java | 6 +- .../xpack/ml/job/JobManager.java | 462 +++++++++--------- .../ml/job/persistence/JobConfigProvider.java | 178 ++++++- .../autodetect/AutodetectProcessManager.java | 129 ++--- .../MachineLearningLicensingTests.java | 8 + .../xpack/ml/MlSingleNodeTestCase.java | 2 +- .../elasticsearch/xpack/ml/MlTasksTests.java | 18 + .../ml/integration/JobConfigProviderIT.java | 91 ++++ .../xpack/ml/job/JobManagerTests.java | 294 ++++++++--- .../ml/job/persistence/MockClientBuilder.java | 58 +++ .../AutodetectProcessManagerTests.java | 96 ++-- 28 files changed, 1167 insertions(+), 600 deletions(-) diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/MlTasks.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/MlTasks.java index 5c17271738e32..f421ba7bf4ad8 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/MlTasks.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/MlTasks.java @@ -12,8 +12,15 @@ import org.elasticsearch.xpack.core.ml.job.config.JobState; import org.elasticsearch.xpack.core.ml.job.config.JobTaskState; +import java.util.Collection; +import java.util.Set; +import java.util.stream.Collectors; + public final class MlTasks { + public static final String JOB_TASK_PREFIX = "job-"; + public static final String DATAFEED_TASK_PREFIX = "datafeed-"; + private MlTasks() { } @@ -22,7 +29,7 @@ private MlTasks() { * A datafeed id can be used as a job id, because they are stored separately in cluster state. */ public static String jobTaskId(String jobId) { - return "job-" + jobId; + return JOB_TASK_PREFIX + jobId; } /** @@ -30,7 +37,7 @@ public static String jobTaskId(String jobId) { * A job id can be used as a datafeed id, because they are stored separately in cluster state. */ public static String datafeedTaskId(String datafeedId) { - return "datafeed-" + datafeedId; + return DATAFEED_TASK_PREFIX + datafeedId; } @Nullable @@ -67,4 +74,17 @@ public static DatafeedState getDatafeedState(String datafeedId, @Nullable Persis return DatafeedState.STOPPED; } } + + /** + * The job Ids of anomaly detector job tasks + * @param tasks Active tasks + * @return The job Ids of anomaly detector job tasks + */ + public static Set openJobIds(PersistentTasksCustomMetaData tasks) { + Collection> activeTasks = tasks.tasks(); + + return activeTasks.stream().filter(t -> t.getId().startsWith(JOB_TASK_PREFIX)) + .map(t -> t.getId().substring(JOB_TASK_PREFIX.length())) + .collect(Collectors.toSet()); + } } diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MachineLearning.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MachineLearning.java index 747440c6db4fb..fd250b63d927f 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MachineLearning.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MachineLearning.java @@ -364,7 +364,7 @@ public Collection createComponents(Client client, ClusterService cluster Auditor auditor = new Auditor(client, clusterService.nodeName()); JobResultsProvider jobResultsProvider = new JobResultsProvider(client, settings); UpdateJobProcessNotifier notifier = new UpdateJobProcessNotifier(settings, client, clusterService, threadPool); - JobManager jobManager = new JobManager(env, settings, jobResultsProvider, clusterService, auditor, client, notifier); + JobManager jobManager = new JobManager(env, settings, jobResultsProvider, clusterService, auditor, threadPool, client, notifier); JobDataCountsPersister jobDataCountsPersister = new JobDataCountsPersister(settings, client); JobResultsPersister jobResultsPersister = new JobResultsPersister(settings, client); diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteCalendarAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteCalendarAction.java index 577e981e75b69..3654edc00a8fc 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteCalendarAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteCalendarAction.java @@ -63,8 +63,10 @@ protected void doExecute(DeleteCalendarAction.Request request, ActionListener listener.onResponse(new DeleteCalendarAction.Response(true)), + listener::onFailure + )); }, listener::onFailure)); }, diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteCalendarEventAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteCalendarEventAction.java index 7d38b11f58850..4bda9cae58cb6 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteCalendarEventAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteCalendarEventAction.java @@ -104,8 +104,10 @@ public void onResponse(DeleteResponse response) { if (response.status() == RestStatus.NOT_FOUND) { listener.onFailure(new ResourceNotFoundException("No event with id [" + eventId + "]")); } else { - jobManager.updateProcessOnCalendarChanged(calendar.getJobIds()); - listener.onResponse(new DeleteCalendarEventAction.Response(true)); + jobManager.updateProcessOnCalendarChanged(calendar.getJobIds(), ActionListener.wrap( + r -> listener.onResponse(new DeleteCalendarEventAction.Response(true)), + listener::onFailure + )); } } diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteModelSnapshotAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteModelSnapshotAction.java index ae720fb40d813..2faf22b491307 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteModelSnapshotAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteModelSnapshotAction.java @@ -12,13 +12,11 @@ import org.elasticsearch.action.support.HandledTransportAction; import org.elasticsearch.client.Client; import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver; -import org.elasticsearch.cluster.service.ClusterService; import org.elasticsearch.common.inject.Inject; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.transport.TransportService; import org.elasticsearch.xpack.core.ml.action.DeleteModelSnapshotAction; -import org.elasticsearch.xpack.core.ml.job.config.Job; import org.elasticsearch.xpack.core.ml.job.messages.Messages; import org.elasticsearch.xpack.core.ml.job.persistence.JobDataDeleter; import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.ModelSnapshot; @@ -33,20 +31,20 @@ public class TransportDeleteModelSnapshotAction extends HandledTransportAction { private final Client client; + private final JobManager jobManager; private final JobResultsProvider jobResultsProvider; - private final ClusterService clusterService; private final Auditor auditor; @Inject public TransportDeleteModelSnapshotAction(Settings settings, TransportService transportService, ThreadPool threadPool, ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver, - JobResultsProvider jobResultsProvider, ClusterService clusterService, Client client, + JobManager jobManager, JobResultsProvider jobResultsProvider, Client client, Auditor auditor) { super(settings, DeleteModelSnapshotAction.NAME, threadPool, transportService, actionFilters, indexNameExpressionResolver, DeleteModelSnapshotAction.Request::new); this.client = client; + this.jobManager = jobManager; this.jobResultsProvider = jobResultsProvider; - this.clusterService = clusterService; this.auditor = auditor; } @@ -70,32 +68,39 @@ protected void doExecute(DeleteModelSnapshotAction.Request request, ActionListen ModelSnapshot deleteCandidate = deleteCandidates.get(0); // Verify the snapshot is not being used - Job job = JobManager.getJobOrThrowIfUnknown(request.getJobId(), clusterService.state()); - String currentModelInUse = job.getModelSnapshotId(); - if (currentModelInUse != null && currentModelInUse.equals(request.getSnapshotId())) { - throw new IllegalArgumentException(Messages.getMessage(Messages.REST_CANNOT_DELETE_HIGHEST_PRIORITY, - request.getSnapshotId(), request.getJobId())); - } + jobManager.getJob(request.getJobId(), ActionListener.wrap( + job -> { + String currentModelInUse = job.getModelSnapshotId(); + if (currentModelInUse != null && currentModelInUse.equals(request.getSnapshotId())) { + listener.onFailure( + new IllegalArgumentException(Messages.getMessage(Messages.REST_CANNOT_DELETE_HIGHEST_PRIORITY, + request.getSnapshotId(), request.getJobId()))); + return; + } - // Delete the snapshot and any associated state files - JobDataDeleter deleter = new JobDataDeleter(client, request.getJobId()); - deleter.deleteModelSnapshots(Collections.singletonList(deleteCandidate), new ActionListener() { - @Override - public void onResponse(BulkResponse bulkResponse) { - String msg = Messages.getMessage(Messages.JOB_AUDIT_SNAPSHOT_DELETED, deleteCandidate.getSnapshotId(), - deleteCandidate.getDescription()); - auditor.info(request.getJobId(), msg); - logger.debug("[{}] {}", request.getJobId(), msg); - // We don't care about the bulk response, just that it succeeded - listener.onResponse(new DeleteModelSnapshotAction.Response(true)); - } + // Delete the snapshot and any associated state files + JobDataDeleter deleter = new JobDataDeleter(client, request.getJobId()); + deleter.deleteModelSnapshots(Collections.singletonList(deleteCandidate), + new ActionListener() { + @Override + public void onResponse(BulkResponse bulkResponse) { + String msg = Messages.getMessage(Messages.JOB_AUDIT_SNAPSHOT_DELETED, + deleteCandidate.getSnapshotId(), deleteCandidate.getDescription()); + auditor.info(request.getJobId(), msg); + logger.debug("[{}] {}", request.getJobId(), msg); + // We don't care about the bulk response, just that it succeeded + listener.onResponse(new DeleteModelSnapshotAction.Response(true)); + } - @Override - public void onFailure(Exception e) { - listener.onFailure(e); - } - }); + @Override + public void onFailure(Exception e) { + listener.onFailure(e); + } + }); + }, + listener::onFailure + )); }, listener::onFailure); } } diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportForecastJobAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportForecastJobAction.java index bea942f8b87db..7a929c000c02f 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportForecastJobAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportForecastJobAction.java @@ -9,7 +9,6 @@ import org.elasticsearch.Version; import org.elasticsearch.action.ActionListener; import org.elasticsearch.action.support.ActionFilters; -import org.elasticsearch.cluster.ClusterState; import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver; import org.elasticsearch.cluster.service.ClusterService; import org.elasticsearch.common.inject.Inject; @@ -42,15 +41,17 @@ public class TransportForecastJobAction extends TransportJobTaskAction listener) { - ClusterState state = clusterService.state(); - Job job = JobManager.getJobOrThrowIfUnknown(task.getJobId(), state); - validate(job, request); + jobManager.getJob(task.getJobId(), ActionListener.wrap( + job -> { + validate(job, request); - ForecastParams.Builder paramsBuilder = ForecastParams.builder(); + ForecastParams.Builder paramsBuilder = ForecastParams.builder(); - if (request.getDuration() != null) { - paramsBuilder.duration(request.getDuration()); - } + if (request.getDuration() != null) { + paramsBuilder.duration(request.getDuration()); + } - if (request.getExpiresIn() != null) { - paramsBuilder.expiresIn(request.getExpiresIn()); - } + if (request.getExpiresIn() != null) { + paramsBuilder.expiresIn(request.getExpiresIn()); + } - // tmp storage might be null, we do not log here, because it might not be - // required - Path tmpStorage = processManager.tryGetTmpStorage(task, FORECAST_LOCAL_STORAGE_LIMIT); - if (tmpStorage != null) { - paramsBuilder.tmpStorage(tmpStorage.toString()); - } + // tmp storage might be null, we do not log here, because it might not be + // required + Path tmpStorage = processManager.tryGetTmpStorage(task, FORECAST_LOCAL_STORAGE_LIMIT); + if (tmpStorage != null) { + paramsBuilder.tmpStorage(tmpStorage.toString()); + } - ForecastParams params = paramsBuilder.build(); - processManager.forecastJob(task, params, e -> { - if (e == null) { - Consumer forecastRequestStatsHandler = forecastRequestStats -> { - if (forecastRequestStats == null) { - // paranoia case, it should not happen that we do not retrieve a result - listener.onFailure(new ElasticsearchException( - "Cannot run forecast: internal error, please check the logs")); - } else if (forecastRequestStats.getStatus() == ForecastRequestStats.ForecastRequestStatus.FAILED) { - List messages = forecastRequestStats.getMessages(); - if (messages.size() > 0) { - listener.onFailure(ExceptionsHelper.badRequestException("Cannot run forecast: " - + messages.get(0))); - } else { - // paranoia case, it should not be possible to have an empty message list - listener.onFailure( - new ElasticsearchException( + ForecastParams params = paramsBuilder.build(); + processManager.forecastJob(task, params, e -> { + if (e == null) { + Consumer forecastRequestStatsHandler = forecastRequestStats -> { + if (forecastRequestStats == null) { + // paranoia case, it should not happen that we do not retrieve a result + listener.onFailure(new ElasticsearchException( "Cannot run forecast: internal error, please check the logs")); + } else if (forecastRequestStats.getStatus() == ForecastRequestStats.ForecastRequestStatus.FAILED) { + List messages = forecastRequestStats.getMessages(); + if (messages.size() > 0) { + listener.onFailure(ExceptionsHelper.badRequestException("Cannot run forecast: " + + messages.get(0))); + } else { + // paranoia case, it should not be possible to have an empty message list + listener.onFailure( + new ElasticsearchException( + "Cannot run forecast: internal error, please check the logs")); + } + } else { + listener.onResponse(new ForecastJobAction.Response(true, params.getForecastId())); + } + }; + + jobResultsProvider.getForecastRequestStats(request.getJobId(), params.getForecastId(), + forecastRequestStatsHandler, listener::onFailure); + } else { + listener.onFailure(e); } - } else { - listener.onResponse(new ForecastJobAction.Response(true, params.getForecastId())); - } - }; - - jobResultsProvider.getForecastRequestStats(request.getJobId(), params.getForecastId(), - forecastRequestStatsHandler, listener::onFailure); - } else { - listener.onFailure(e); - } - }); + }); + }, + listener::onFailure + )); } static void validate(Job job, ForecastJobAction.Request request) { diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetBucketsAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetBucketsAction.java index 06fe026a2370f..fd770fa88c5cf 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetBucketsAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetBucketsAction.java @@ -38,28 +38,33 @@ public TransportGetBucketsAction(Settings settings, ThreadPool threadPool, Trans @Override protected void doExecute(GetBucketsAction.Request request, ActionListener listener) { - jobManager.getJobOrThrowIfUnknown(request.getJobId()); + jobManager.getJob(request.getJobId(), ActionListener.wrap( + job -> { + BucketsQueryBuilder query = + new BucketsQueryBuilder().expand(request.isExpand()) + .includeInterim(request.isExcludeInterim() == false) + .start(request.getStart()) + .end(request.getEnd()) + .anomalyScoreThreshold(request.getAnomalyScore()) + .sortField(request.getSort()) + .sortDescending(request.isDescending()); - BucketsQueryBuilder query = - new BucketsQueryBuilder().expand(request.isExpand()) - .includeInterim(request.isExcludeInterim() == false) - .start(request.getStart()) - .end(request.getEnd()) - .anomalyScoreThreshold(request.getAnomalyScore()) - .sortField(request.getSort()) - .sortDescending(request.isDescending()); + if (request.getPageParams() != null) { + query.from(request.getPageParams().getFrom()) + .size(request.getPageParams().getSize()); + } + if (request.getTimestamp() != null) { + query.timestamp(request.getTimestamp()); + } else { + query.start(request.getStart()); + query.end(request.getEnd()); + } + jobResultsProvider.buckets(request.getJobId(), query, q -> + listener.onResponse(new GetBucketsAction.Response(q)), listener::onFailure, client); - if (request.getPageParams() != null) { - query.from(request.getPageParams().getFrom()) - .size(request.getPageParams().getSize()); - } - if (request.getTimestamp() != null) { - query.timestamp(request.getTimestamp()); - } else { - query.start(request.getStart()); - query.end(request.getEnd()); - } - jobResultsProvider.buckets(request.getJobId(), query, q -> - listener.onResponse(new GetBucketsAction.Response(q)), listener::onFailure, client); + }, + listener::onFailure + + )); } } diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetCategoriesAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetCategoriesAction.java index 735d598dfe159..f6000e1ec6bdd 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetCategoriesAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetCategoriesAction.java @@ -37,11 +37,14 @@ public TransportGetCategoriesAction(Settings settings, ThreadPool threadPool, Tr @Override protected void doExecute(GetCategoriesAction.Request request, ActionListener listener) { - jobManager.getJobOrThrowIfUnknown(request.getJobId()); - - Integer from = request.getPageParams() != null ? request.getPageParams().getFrom() : null; - Integer size = request.getPageParams() != null ? request.getPageParams().getSize() : null; - jobResultsProvider.categoryDefinitions(request.getJobId(), request.getCategoryId(), true, from, size, - r -> listener.onResponse(new GetCategoriesAction.Response(r)), listener::onFailure, client); + jobManager.getJob(request.getJobId(), ActionListener.wrap( + job -> { + Integer from = request.getPageParams() != null ? request.getPageParams().getFrom() : null; + Integer size = request.getPageParams() != null ? request.getPageParams().getSize() : null; + jobResultsProvider.categoryDefinitions(request.getJobId(), request.getCategoryId(), true, from, size, + r -> listener.onResponse(new GetCategoriesAction.Response(r)), listener::onFailure, client); + }, + listener::onFailure + )); } } diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetInfluencersAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetInfluencersAction.java index ba370238df573..32ad802a204b4 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetInfluencersAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetInfluencersAction.java @@ -38,18 +38,22 @@ public TransportGetInfluencersAction(Settings settings, ThreadPool threadPool, T @Override protected void doExecute(GetInfluencersAction.Request request, ActionListener listener) { - jobManager.getJobOrThrowIfUnknown(request.getJobId()); - InfluencersQueryBuilder.InfluencersQuery query = new InfluencersQueryBuilder() - .includeInterim(request.isExcludeInterim() == false) - .start(request.getStart()) - .end(request.getEnd()) - .from(request.getPageParams().getFrom()) - .size(request.getPageParams().getSize()) - .influencerScoreThreshold(request.getInfluencerScore()) - .sortField(request.getSort()) - .sortDescending(request.isDescending()).build(); - jobResultsProvider.influencers(request.getJobId(), query, - page -> listener.onResponse(new GetInfluencersAction.Response(page)), listener::onFailure, client); + jobManager.getJob(request.getJobId(), ActionListener.wrap( + job -> { + InfluencersQueryBuilder.InfluencersQuery query = new InfluencersQueryBuilder() + .includeInterim(request.isExcludeInterim() == false) + .start(request.getStart()) + .end(request.getEnd()) + .from(request.getPageParams().getFrom()) + .size(request.getPageParams().getSize()) + .influencerScoreThreshold(request.getInfluencerScore()) + .sortField(request.getSort()) + .sortDescending(request.isDescending()).build(); + jobResultsProvider.influencers(request.getJobId(), query, + page -> listener.onResponse(new GetInfluencersAction.Response(page)), listener::onFailure, client); + }, + listener::onFailure) + ); } } diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetJobsAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetJobsAction.java index d0565d7cbe0ba..56b04cbb66da0 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetJobsAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetJobsAction.java @@ -18,9 +18,7 @@ import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.transport.TransportService; import org.elasticsearch.xpack.core.ml.action.GetJobsAction; -import org.elasticsearch.xpack.core.ml.action.util.QueryPage; import org.elasticsearch.xpack.ml.job.JobManager; -import org.elasticsearch.xpack.core.ml.job.config.Job; public class TransportGetJobsAction extends TransportMasterNodeReadAction { @@ -48,10 +46,14 @@ protected GetJobsAction.Response newResponse() { @Override protected void masterOperation(GetJobsAction.Request request, ClusterState state, - ActionListener listener) throws Exception { + ActionListener listener) { logger.debug("Get job '{}'", request.getJobId()); - QueryPage jobs = jobManager.expandJobs(request.getJobId(), request.allowNoJobs(), state); - listener.onResponse(new GetJobsAction.Response(jobs)); + jobManager.expandJobs(request.getJobId(), request.allowNoJobs(), ActionListener.wrap( + jobs -> { + listener.onResponse(new GetJobsAction.Response(jobs)); + }, + listener::onFailure + )); } @Override diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetModelSnapshotsAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetModelSnapshotsAction.java index 5abdb7d76a154..6a93ff79fbbde 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetModelSnapshotsAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetModelSnapshotsAction.java @@ -44,13 +44,17 @@ protected void doExecute(GetModelSnapshotsAction.Request request, ActionListener request.getJobId(), request.getSnapshotId(), request.getPageParams().getFrom(), request.getPageParams().getSize(), request.getStart(), request.getEnd(), request.getSort(), request.getDescOrder()); - jobManager.getJobOrThrowIfUnknown(request.getJobId()); - - jobResultsProvider.modelSnapshots(request.getJobId(), request.getPageParams().getFrom(), request.getPageParams().getSize(), - request.getStart(), request.getEnd(), request.getSort(), request.getDescOrder(), request.getSnapshotId(), - page -> { - listener.onResponse(new GetModelSnapshotsAction.Response(clearQuantiles(page))); - }, listener::onFailure); + jobManager.getJob(request.getJobId(), ActionListener.wrap( + job -> { + jobResultsProvider.modelSnapshots(request.getJobId(), request.getPageParams().getFrom(), + request.getPageParams().getSize(), request.getStart(), request.getEnd(), request.getSort(), + request.getDescOrder(), request.getSnapshotId(), + page -> { + listener.onResponse(new GetModelSnapshotsAction.Response(clearQuantiles(page))); + }, listener::onFailure); + }, + listener::onFailure + )); } public static QueryPage clearQuantiles(QueryPage page) { diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetOverallBucketsAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetOverallBucketsAction.java index 8d4c6dc83ce23..6bc65771df0a8 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetOverallBucketsAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetOverallBucketsAction.java @@ -73,21 +73,25 @@ public TransportGetOverallBucketsAction(Settings settings, ThreadPool threadPool @Override protected void doExecute(GetOverallBucketsAction.Request request, ActionListener listener) { - QueryPage jobsPage = jobManager.expandJobs(request.getJobId(), request.allowNoJobs(), clusterService.state()); - if (jobsPage.count() == 0) { - listener.onResponse(new GetOverallBucketsAction.Response()); - return; - } + jobManager.expandJobs(request.getJobId(), request.allowNoJobs(), ActionListener.wrap( + jobPage -> { + if (jobPage.count() == 0) { + listener.onResponse(new GetOverallBucketsAction.Response()); + return; + } - // As computing and potentially aggregating overall buckets might take a while, - // we run in a different thread to avoid blocking the network thread. - threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME).execute(() -> { - try { - getOverallBuckets(request, jobsPage.results(), listener); - } catch (Exception e) { - listener.onFailure(e); - } - }); + // As computing and potentially aggregating overall buckets might take a while, + // we run in a different thread to avoid blocking the network thread. + threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME).execute(() -> { + try { + getOverallBuckets(request, jobPage.results(), listener); + } catch (Exception e) { + listener.onFailure(e); + } + }); + }, + listener::onFailure + )); } private void getOverallBuckets(GetOverallBucketsAction.Request request, List jobs, diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetRecordsAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetRecordsAction.java index d27474bc3740d..f88011d1ea0cf 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetRecordsAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetRecordsAction.java @@ -39,18 +39,21 @@ public TransportGetRecordsAction(Settings settings, ThreadPool threadPool, Trans @Override protected void doExecute(GetRecordsAction.Request request, ActionListener listener) { - jobManager.getJobOrThrowIfUnknown(request.getJobId()); - - RecordsQueryBuilder query = new RecordsQueryBuilder() - .includeInterim(request.isExcludeInterim() == false) - .epochStart(request.getStart()) - .epochEnd(request.getEnd()) - .from(request.getPageParams().getFrom()) - .size(request.getPageParams().getSize()) - .recordScore(request.getRecordScoreFilter()) - .sortField(request.getSort()) - .sortDescending(request.isDescending()); - jobResultsProvider.records(request.getJobId(), query, page -> - listener.onResponse(new GetRecordsAction.Response(page)), listener::onFailure, client); + jobManager.getJob(request.getJobId(), ActionListener.wrap( + job -> { + RecordsQueryBuilder query = new RecordsQueryBuilder() + .includeInterim(request.isExcludeInterim() == false) + .epochStart(request.getStart()) + .epochEnd(request.getEnd()) + .from(request.getPageParams().getFrom()) + .size(request.getPageParams().getSize()) + .recordScore(request.getRecordScoreFilter()) + .sortField(request.getSort()) + .sortDescending(request.isDescending()); + jobResultsProvider.records(request.getJobId(), query, page -> + listener.onResponse(new GetRecordsAction.Response(page)), listener::onFailure, client); + }, + listener::onFailure + )); } } diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportJobTaskAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportJobTaskAction.java index d0da0907b7d5e..b09e002d5e6e8 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportJobTaskAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportJobTaskAction.java @@ -11,7 +11,6 @@ import org.elasticsearch.action.support.ActionFilters; import org.elasticsearch.action.support.tasks.BaseTasksResponse; import org.elasticsearch.action.support.tasks.TransportTasksAction; -import org.elasticsearch.cluster.ClusterState; import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver; import org.elasticsearch.cluster.service.ClusterService; import org.elasticsearch.common.io.stream.Writeable; @@ -23,7 +22,6 @@ import org.elasticsearch.xpack.core.ml.action.JobTaskRequest; import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper; import org.elasticsearch.persistent.PersistentTasksCustomMetaData; -import org.elasticsearch.xpack.ml.job.JobManager; import org.elasticsearch.xpack.ml.job.process.autodetect.AutodetectProcessManager; import java.util.List; @@ -54,8 +52,6 @@ protected void doExecute(Task task, Request request, ActionListener li String jobId = request.getJobId(); // We need to check whether there is at least an assigned task here, otherwise we cannot redirect to the // node running the job task. - ClusterState state = clusterService.state(); - JobManager.getJobOrThrowIfUnknown(jobId, state); PersistentTasksCustomMetaData tasks = clusterService.state().getMetaData().custom(PersistentTasksCustomMetaData.TYPE); PersistentTasksCustomMetaData.PersistentTask jobTask = MlTasks.getJobTask(jobId, tasks); if (jobTask == null || jobTask.isAssigned() == false) { diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportPostCalendarEventsAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportPostCalendarEventsAction.java index 0c6294e0b79b0..c2937f6fe29df 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportPostCalendarEventsAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportPostCalendarEventsAction.java @@ -84,8 +84,10 @@ protected void doExecute(PostCalendarEventsAction.Request request, new ActionListener() { @Override public void onResponse(BulkResponse response) { - jobManager.updateProcessOnCalendarChanged(calendar.getJobIds()); - listener.onResponse(new PostCalendarEventsAction.Response(events)); + jobManager.updateProcessOnCalendarChanged(calendar.getJobIds(), ActionListener.wrap( + r -> listener.onResponse(new PostCalendarEventsAction.Response(events)), + listener::onFailure + )); } @Override diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportRevertModelSnapshotAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportRevertModelSnapshotAction.java index 07b9dade4d8fa..83121043c4d07 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportRevertModelSnapshotAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportRevertModelSnapshotAction.java @@ -22,7 +22,6 @@ import org.elasticsearch.transport.TransportService; import org.elasticsearch.xpack.core.ml.MlTasks; import org.elasticsearch.xpack.core.ml.action.RevertModelSnapshotAction; -import org.elasticsearch.xpack.core.ml.job.config.Job; import org.elasticsearch.xpack.core.ml.job.config.JobState; import org.elasticsearch.xpack.core.ml.job.messages.Messages; import org.elasticsearch.xpack.core.ml.job.persistence.JobDataDeleter; @@ -72,22 +71,26 @@ protected void masterOperation(RevertModelSnapshotAction.Request request, Cluste logger.debug("Received request to revert to snapshot id '{}' for job '{}', deleting intervening results: {}", request.getSnapshotId(), request.getJobId(), request.getDeleteInterveningResults()); - Job job = JobManager.getJobOrThrowIfUnknown(request.getJobId(), state); - PersistentTasksCustomMetaData tasks = state.getMetaData().custom(PersistentTasksCustomMetaData.TYPE); - JobState jobState = MlTasks.getJobState(job.getId(), tasks); + jobManager.jobExist(request.getJobId(), ActionListener.wrap( + exists -> { + PersistentTasksCustomMetaData tasks = state.getMetaData().custom(PersistentTasksCustomMetaData.TYPE); + JobState jobState = MlTasks.getJobState(request.getJobId(), tasks); - if (jobState.equals(JobState.CLOSED) == false) { - throw ExceptionsHelper.conflictStatusException(Messages.getMessage(Messages.REST_JOB_NOT_CLOSED_REVERT)); - } + if (jobState.equals(JobState.CLOSED) == false) { + throw ExceptionsHelper.conflictStatusException(Messages.getMessage(Messages.REST_JOB_NOT_CLOSED_REVERT)); + } - getModelSnapshot(request, jobResultsProvider, modelSnapshot -> { - ActionListener wrappedListener = listener; - if (request.getDeleteInterveningResults()) { - wrappedListener = wrapDeleteOldDataListener(wrappedListener, modelSnapshot, request.getJobId()); - wrappedListener = wrapRevertDataCountsListener(wrappedListener, modelSnapshot, request.getJobId()); - } - jobManager.revertSnapshot(request, wrappedListener, modelSnapshot); - }, listener::onFailure); + getModelSnapshot(request, jobResultsProvider, modelSnapshot -> { + ActionListener wrappedListener = listener; + if (request.getDeleteInterveningResults()) { + wrappedListener = wrapDeleteOldDataListener(wrappedListener, modelSnapshot, request.getJobId()); + wrappedListener = wrapRevertDataCountsListener(wrappedListener, modelSnapshot, request.getJobId()); + } + jobManager.revertSnapshot(request, wrappedListener, modelSnapshot); + }, listener::onFailure); + }, + listener::onFailure + )); } private void getModelSnapshot(RevertModelSnapshotAction.Request request, JobResultsProvider provider, Consumer handler, diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportUpdateCalendarJobAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportUpdateCalendarJobAction.java index 6883767ce8f62..b28e24dbda85f 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportUpdateCalendarJobAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportUpdateCalendarJobAction.java @@ -45,8 +45,10 @@ protected void doExecute(UpdateCalendarJobAction.Request request, ActionListener jobResultsProvider.updateCalendar(request.getCalendarId(), jobIdsToAdd, jobIdsToRemove, c -> { - jobManager.updateProcessOnCalendarChanged(c.getJobIds()); - listener.onResponse(new PutCalendarAction.Response(c)); + jobManager.updateProcessOnCalendarChanged(c.getJobIds(), ActionListener.wrap( + r -> listener.onResponse(new PutCalendarAction.Response(c)), + listener::onFailure + )); }, listener::onFailure); } } diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportUpdateFilterAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportUpdateFilterAction.java index e58acc2dcad75..0a4ca1d680995 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportUpdateFilterAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportUpdateFilterAction.java @@ -116,8 +116,10 @@ private void indexUpdatedFilter(MlFilter filter, long version, UpdateFilterActio executeAsyncWithOrigin(client, ML_ORIGIN, IndexAction.INSTANCE, indexRequest, new ActionListener() { @Override public void onResponse(IndexResponse indexResponse) { - jobManager.notifyFilterChanged(filter, request.getAddItems(), request.getRemoveItems()); - listener.onResponse(new PutFilterAction.Response(filter)); + jobManager.notifyFilterChanged(filter, request.getAddItems(), request.getRemoveItems(), ActionListener.wrap( + response -> listener.onResponse(new PutFilterAction.Response(filter)), + listener::onFailure + )); } @Override diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java index 48bdd4682b8ca..8194e226f8f19 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java @@ -10,9 +10,7 @@ import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.support.WriteRequest; import org.elasticsearch.client.Client; -import org.elasticsearch.cluster.AckedClusterStateUpdateTask; import org.elasticsearch.cluster.ClusterState; -import org.elasticsearch.cluster.ClusterStateUpdateTask; import org.elasticsearch.cluster.metadata.MetaData; import org.elasticsearch.cluster.service.ClusterService; import org.elasticsearch.common.CheckedConsumer; @@ -28,8 +26,9 @@ import org.elasticsearch.common.xcontent.XContentFactory; import org.elasticsearch.env.Environment; import org.elasticsearch.index.analysis.AnalysisRegistry; +import org.elasticsearch.index.engine.VersionConflictEngineException; import org.elasticsearch.persistent.PersistentTasksCustomMetaData; -import org.elasticsearch.xpack.core.XPackPlugin; +import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.xpack.core.ml.MachineLearningField; import org.elasticsearch.xpack.core.ml.MlMetadata; import org.elasticsearch.xpack.core.ml.MlTasks; @@ -52,6 +51,7 @@ import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper; import org.elasticsearch.xpack.ml.MachineLearning; import org.elasticsearch.xpack.ml.job.categorization.CategorizationAnalyzer; +import org.elasticsearch.xpack.ml.job.persistence.JobConfigProvider; import org.elasticsearch.xpack.ml.job.persistence.JobResultsProvider; import org.elasticsearch.xpack.ml.job.persistence.JobResultsPersister; import org.elasticsearch.xpack.ml.job.process.autodetect.UpdateParams; @@ -61,23 +61,22 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Date; -import java.util.HashSet; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.Set; -import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import java.util.regex.Matcher; import java.util.regex.Pattern; -import java.util.stream.Collectors; /** * Allows interactions with jobs. The managed interactions include: *
    *
  • creation
    • + *
  • reading
    • *
  • deletion
    • *
  • updating
    • - *
  • starting/stopping of datafeed jobs
    • *
*/ public class JobManager extends AbstractComponent { @@ -90,7 +89,9 @@ public class JobManager extends AbstractComponent { private final ClusterService clusterService; private final Auditor auditor; private final Client client; + private final ThreadPool threadPool; private final UpdateJobProcessNotifier updateJobProcessNotifier; + private final JobConfigProvider jobConfigProvider; private volatile ByteSizeValue maxModelMemoryLimit; @@ -98,7 +99,7 @@ public class JobManager extends AbstractComponent { * Create a JobManager */ public JobManager(Environment environment, Settings settings, JobResultsProvider jobResultsProvider, - ClusterService clusterService, Auditor auditor, + ClusterService clusterService, Auditor auditor, ThreadPool threadPool, Client client, UpdateJobProcessNotifier updateJobProcessNotifier) { super(settings); this.environment = environment; @@ -106,7 +107,9 @@ public JobManager(Environment environment, Settings settings, JobResultsProvider this.clusterService = Objects.requireNonNull(clusterService); this.auditor = Objects.requireNonNull(auditor); this.client = Objects.requireNonNull(client); + this.threadPool = Objects.requireNonNull(threadPool); this.updateJobProcessNotifier = updateJobProcessNotifier; + this.jobConfigProvider = new JobConfigProvider(client, settings); maxModelMemoryLimit = MachineLearningField.MAX_MODEL_MEMORY_LIMIT.get(settings); clusterService.getClusterSettings() @@ -117,35 +120,44 @@ private void setMaxModelMemoryLimit(ByteSizeValue maxModelMemoryLimit) { this.maxModelMemoryLimit = maxModelMemoryLimit; } + public void jobExist(String jobId, ActionListener listener) { + jobConfigProvider.checkJobExists(jobId, listener); + } + /** * Gets the job that matches the given {@code jobId}. * * @param jobId the jobId - * @return The {@link Job} matching the given {code jobId} - * @throws ResourceNotFoundException if no job matches {@code jobId} + * @param jobListener the Job listener. If no job matches {@code jobId} + * a ResourceNotFoundException is returned */ - public Job getJobOrThrowIfUnknown(String jobId) { - return getJobOrThrowIfUnknown(jobId, clusterService.state()); + public void getJob(String jobId, ActionListener jobListener) { + jobConfigProvider.getJob(jobId, ActionListener.wrap( + r -> jobListener.onResponse(r.build()), // TODO JIndex we shouldn't be building the job here + e -> { + if (e instanceof ResourceNotFoundException) { + // Try to get the job from the cluster state + getJobFromClusterState(jobId, jobListener); + } else { + jobListener.onFailure(e); + } + } + )); } /** - * Gets the job that matches the given {@code jobId}. - * + * Even this this uses a listener the job is returned on the same thread * @param jobId the jobId - * @param clusterState the cluster state - * @return The {@link Job} matching the given {code jobId} - * @throws ResourceNotFoundException if no job matches {@code jobId} + * @param jobListener the Job listener. If no job matches {@code jobId} + * a ResourceNotFoundException is returned */ - public static Job getJobOrThrowIfUnknown(String jobId, ClusterState clusterState) { - Job job = MlMetadata.getMlMetadata(clusterState).getJobs().get(jobId); + private void getJobFromClusterState(String jobId, ActionListener jobListener) { + Job job = MlMetadata.getMlMetadata(clusterService.state()).getJobs().get(jobId); if (job == null) { - throw ExceptionsHelper.missingJobException(jobId); + jobListener.onFailure(ExceptionsHelper.missingJobException(jobId)); + } else { + jobListener.onResponse(job); } - return job; - } - - private Set expandJobIds(String expression, boolean allowNoJobs, ClusterState clusterState) { - return MlMetadata.getMlMetadata(clusterState).expandJobIds(expression, allowNoJobs); } /** @@ -153,19 +165,44 @@ private Set expandJobIds(String expression, boolean allowNoJobs, Cluster * Note that when the {@code jobId} is {@link MetaData#ALL} all jobs are returned. * * @param expression the jobId or an expression matching jobIds - * @param clusterState the cluster state * @param allowNoJobs if {@code false}, an error is thrown when no job matches the {@code jobId} - * @return A {@link QueryPage} containing the matching {@code Job}s + * @param jobsListener The jobs listener */ - public QueryPage expandJobs(String expression, boolean allowNoJobs, ClusterState clusterState) { - Set expandedJobIds = expandJobIds(expression, allowNoJobs, clusterState); + public void expandJobs(String expression, boolean allowNoJobs, ActionListener> jobsListener) { + Map clusterStateJobs = expandJobsFromClusterState(expression, allowNoJobs, clusterService.state()); + + jobConfigProvider.expandJobs(expression, allowNoJobs, ActionListener.wrap( + jobBuilders -> { + // Check for duplicate jobs + for (Job.Builder jb : jobBuilders) { + if (clusterStateJobs.containsKey(jb.getId())) { + jobsListener.onFailure(new IllegalStateException("Job [" + jb.getId() + "] configuration " + + "exists in both clusterstate and index")); + return; + } + } + + // Merge cluster state and index jobs + List jobs = new ArrayList<>(); + for (Job.Builder jb : jobBuilders) { + jobs.add(jb.build()); + } + + jobs.addAll(clusterStateJobs.values()); + jobsListener.onResponse(new QueryPage<>(jobs, jobs.size(), Job.RESULTS_FIELD)); + }, + jobsListener::onFailure + )); + } + + private Map expandJobsFromClusterState(String expression, boolean allowNoJobs, ClusterState clusterState) { + Set expandedJobIds = MlMetadata.getMlMetadata(clusterState).expandJobIds(expression, allowNoJobs); MlMetadata mlMetadata = MlMetadata.getMlMetadata(clusterState); - List jobs = new ArrayList<>(); + Map jobIdToJob = new HashMap<>(); for (String expandedJobId : expandedJobIds) { - jobs.add(mlMetadata.getJobs().get(expandedJobId)); + jobIdToJob.put(expandedJobId, mlMetadata.getJobs().get(expandedJobId)); } - logger.debug("Returning jobs matching [" + expression + "]"); - return new QueryPage<>(jobs, jobs.size(), Job.RESULTS_FIELD); + return jobIdToJob; } /** @@ -185,7 +222,7 @@ static void validateCategorizationAnalyzer(Job.Builder jobBuilder, AnalysisRegis } /** - * Stores a job in the cluster state + * Stores the anomaly job configuration */ public void putJob(PutJobAction.Request request, AnalysisRegistry analysisRegistry, ClusterState state, ActionListener actionListener) throws IOException { @@ -199,9 +236,7 @@ public void putJob(PutJobAction.Request request, AnalysisRegistry analysisRegist DEPRECATION_LOGGER.deprecated("Creating jobs with delimited data format is deprecated. Please use xcontent instead."); } - // pre-flight check, not necessarily required, but avoids figuring this out while on the CS update thread - XPackPlugin.checkReadyForXPackCustomMetadata(state); - + // Check for the job in the cluster state first MlMetadata currentMlMetadata = MlMetadata.getMlMetadata(state); if (currentMlMetadata.getJobs().containsKey(job.getId())) { actionListener.onFailure(ExceptionsHelper.jobAlreadyExists(job.getId())); @@ -212,19 +247,20 @@ public void putJob(PutJobAction.Request request, AnalysisRegistry analysisRegist @Override public void onResponse(Boolean indicesCreated) { - clusterService.submitStateUpdateTask("put-job-" + job.getId(), - new AckedClusterStateUpdateTask(request, actionListener) { - @Override - protected PutJobAction.Response newResponse(boolean acknowledged) { - auditor.info(job.getId(), Messages.getMessage(Messages.JOB_AUDIT_CREATED)); - return new PutJobAction.Response(job); + jobConfigProvider.putJob(job, ActionListener.wrap( + response -> { + auditor.info(job.getId(), Messages.getMessage(Messages.JOB_AUDIT_CREATED)); + actionListener.onResponse(new PutJobAction.Response(job)); + }, + e -> { + if (e instanceof VersionConflictEngineException) { + // the job already exists + actionListener.onFailure(ExceptionsHelper.jobAlreadyExists(job.getId())); + } else { + actionListener.onFailure(e); } - - @Override - public ClusterState execute(ClusterState currentState) { - return updateClusterState(job, false, currentState); - } - }); + } + )); } @Override @@ -254,13 +290,53 @@ public void onFailure(Exception e) { } public void updateJob(UpdateJobAction.Request request, ActionListener actionListener) { - Job job = getJobOrThrowIfUnknown(request.getJobId()); - validate(request.getJobUpdate(), job, ActionListener.wrap( - nullValue -> internalJobUpdate(request, actionListener), - actionListener::onFailure)); + + ActionListener postUpdateAction; + + // Autodetect must be updated if the fields that the C++ uses are changed + if (request.getJobUpdate().isAutodetectProcessUpdate()) { + postUpdateAction = ActionListener.wrap( + updatedJob -> { + JobUpdate jobUpdate = request.getJobUpdate(); + if (isJobOpen(clusterService.state(), request.getJobId())) { + updateJobProcessNotifier.submitJobUpdate(UpdateParams.fromJobUpdate(jobUpdate), ActionListener.wrap( + isUpdated -> { + if (isUpdated) { + auditJobUpdatedIfNotInternal(request); + } + }, e -> { + // No need to do anything + } + )); + } + actionListener.onResponse(new PutJobAction.Response(updatedJob)); + }, + actionListener::onFailure + ); + } else { + postUpdateAction = ActionListener.wrap(job -> { + logger.debug("[{}] No process update required for job update: {}", () -> request.getJobId(), () -> { + try { + XContentBuilder jsonBuilder = XContentFactory.jsonBuilder(); + request.getJobUpdate().toXContent(jsonBuilder, ToXContent.EMPTY_PARAMS); + return Strings.toString(jsonBuilder); + } catch (IOException e) { + return "(unprintable due to " + e.getMessage() + ")"; + } + }); + + auditJobUpdatedIfNotInternal(request); + actionListener.onResponse(new PutJobAction.Response(job)); + }, + actionListener::onFailure); + } + + + jobConfigProvider.updateJobWithValidation(request.getJobId(), request.getJobUpdate(), maxModelMemoryLimit, + this::validate, postUpdateAction); } - private void validate(JobUpdate jobUpdate, Job job, ActionListener handler) { + private void validate(Job job, JobUpdate jobUpdate, ActionListener handler) { ChainTaskExecutor chainTaskExecutor = new ChainTaskExecutor(client.threadPool().executor( MachineLearning.UTILITY_THREAD_POOL_NAME), true); validateModelSnapshotIdUpdate(job, jobUpdate.getModelSnapshotId(), chainTaskExecutor); @@ -319,86 +395,6 @@ private void validateAnalysisLimitsUpdate(Job job, AnalysisLimits newLimits, Cha }); } - private void internalJobUpdate(UpdateJobAction.Request request, ActionListener actionListener) { - if (request.isWaitForAck()) { - // Use the ack cluster state update - clusterService.submitStateUpdateTask("update-job-" + request.getJobId(), - new AckedClusterStateUpdateTask(request, actionListener) { - private AtomicReference updatedJob = new AtomicReference<>(); - - @Override - protected PutJobAction.Response newResponse(boolean acknowledged) { - return new PutJobAction.Response(updatedJob.get()); - } - - @Override - public ClusterState execute(ClusterState currentState) { - Job job = getJobOrThrowIfUnknown(request.getJobId(), currentState); - updatedJob.set(request.getJobUpdate().mergeWithJob(job, maxModelMemoryLimit)); - return updateClusterState(updatedJob.get(), true, currentState); - } - - @Override - public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) { - afterClusterStateUpdate(newState, request); - } - }); - } else { - clusterService.submitStateUpdateTask("update-job-" + request.getJobId(), new ClusterStateUpdateTask() { - private AtomicReference updatedJob = new AtomicReference<>(); - - @Override - public ClusterState execute(ClusterState currentState) throws Exception { - Job job = getJobOrThrowIfUnknown(request.getJobId(), currentState); - updatedJob.set(request.getJobUpdate().mergeWithJob(job, maxModelMemoryLimit)); - return updateClusterState(updatedJob.get(), true, currentState); - } - - @Override - public void onFailure(String source, Exception e) { - actionListener.onFailure(e); - } - - @Override - public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) { - afterClusterStateUpdate(newState, request); - actionListener.onResponse(new PutJobAction.Response(updatedJob.get())); - } - }); - } - } - - private void afterClusterStateUpdate(ClusterState newState, UpdateJobAction.Request request) { - JobUpdate jobUpdate = request.getJobUpdate(); - - // Change is required if the fields that the C++ uses are being updated - boolean processUpdateRequired = jobUpdate.isAutodetectProcessUpdate(); - - if (processUpdateRequired && isJobOpen(newState, request.getJobId())) { - updateJobProcessNotifier.submitJobUpdate(UpdateParams.fromJobUpdate(jobUpdate), ActionListener.wrap( - isUpdated -> { - if (isUpdated) { - auditJobUpdatedIfNotInternal(request); - } - }, e -> { - // No need to do anything - } - )); - } else { - logger.debug("[{}] No process update required for job update: {}", () -> request.getJobId(), () -> { - try { - XContentBuilder jsonBuilder = XContentFactory.jsonBuilder(); - jobUpdate.toXContent(jsonBuilder, ToXContent.EMPTY_PARAMS); - return Strings.toString(jsonBuilder); - } catch (IOException e) { - return "(unprintable due to " + e.getMessage() + ")"; - } - }); - - auditJobUpdatedIfNotInternal(request); - } - } - private void auditJobUpdatedIfNotInternal(UpdateJobAction.Request request) { if (request.isInternal() == false) { auditor.info(request.getJobId(), Messages.getMessage(Messages.JOB_AUDIT_UPDATED, request.getJobUpdate().getUpdateFields())); @@ -411,32 +407,48 @@ private boolean isJobOpen(ClusterState clusterState, String jobId) { return jobState == JobState.OPENED; } - private ClusterState updateClusterState(Job job, boolean overwrite, ClusterState currentState) { - MlMetadata.Builder builder = createMlMetadataBuilder(currentState); - builder.putJob(job, overwrite); - return buildNewClusterState(currentState, builder); + private Set openJobIds(ClusterState clusterState) { + PersistentTasksCustomMetaData persistentTasks = clusterState.metaData().custom(PersistentTasksCustomMetaData.TYPE); + return MlTasks.openJobIds(persistentTasks); } - public void notifyFilterChanged(MlFilter filter, Set addedItems, Set removedItems) { + public void notifyFilterChanged(MlFilter filter, Set addedItems, Set removedItems, + ActionListener updatedListener) { if (addedItems.isEmpty() && removedItems.isEmpty()) { + updatedListener.onResponse(Boolean.TRUE); return; } - ClusterState clusterState = clusterService.state(); - QueryPage jobs = expandJobs("*", true, clusterService.state()); - for (Job job : jobs.results()) { - Set jobFilters = job.getAnalysisConfig().extractReferencedFilters(); - if (jobFilters.contains(filter.getId())) { - if (isJobOpen(clusterState, job.getId())) { - updateJobProcessNotifier.submitJobUpdate(UpdateParams.filterUpdate(job.getId(), filter), - ActionListener.wrap(isUpdated -> { - auditFilterChanges(job.getId(), filter.getId(), addedItems, removedItems); - }, e -> {})); - } else { - auditFilterChanges(job.getId(), filter.getId(), addedItems, removedItems); - } - } + Set openJobIds = openJobIds(clusterService.state()); + if (openJobIds.isEmpty()) { + updatedListener.onResponse(Boolean.TRUE); + return; } + + String jobsExpression = Strings.collectionToCommaDelimitedString(openJobIds); + + // TODO JIndex probably better to query for the filter ID + jobConfigProvider.expandJobs(jobsExpression, false, ActionListener.wrap( + jobBuilders -> { + threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME).execute(() -> { + for (Job.Builder builder: jobBuilders) { + // TODO JIndex - sneaky because the job isn't being built here + Set jobFilters = builder.getAnalysisConfig().extractReferencedFilters(); + if (jobFilters.contains(filter.getId())) { + updateJobProcessNotifier.submitJobUpdate(UpdateParams.filterUpdate(builder.getId(), filter), + ActionListener.wrap(isUpdated -> { + auditFilterChanges(builder.getId(), filter.getId(), addedItems, removedItems); + }, e -> {})); + } + } + + updatedListener.onResponse(Boolean.TRUE); + }); + }, + updatedListener::onFailure + )); + + // TODO JIndex previously the filter change would also be audited for closed jobs } private void auditFilterChanges(String jobId, String filterId, Set addedItems, Set removedItems) { @@ -466,26 +478,40 @@ private static void appendCommaSeparatedSet(Set items, StringBuilder sb) sb.append("]"); } - public void updateProcessOnCalendarChanged(List calendarJobIds) { + public void updateProcessOnCalendarChanged(List calendarJobIds, ActionListener updateListener) { ClusterState clusterState = clusterService.state(); - final MlMetadata mlMetadata = MlMetadata.getMlMetadata(clusterState); - - List existingJobsOrGroups = - calendarJobIds.stream().filter(mlMetadata::isGroupOrJob).collect(Collectors.toList()); - - Set expandedJobIds = new HashSet<>(); - existingJobsOrGroups.forEach(jobId -> expandedJobIds.addAll(expandJobIds(jobId, true, clusterState))); - for (String jobId : expandedJobIds) { - if (isJobOpen(clusterState, jobId)) { - updateJobProcessNotifier.submitJobUpdate(UpdateParams.scheduledEventsUpdate(jobId), ActionListener.wrap( - isUpdated -> { - if (isUpdated) { - auditor.info(jobId, Messages.getMessage(Messages.JOB_AUDIT_CALENDARS_UPDATED_ON_PROCESS)); - } - }, e -> {} - )); - } + Set openJobIds = openJobIds(clusterState); + if (openJobIds.isEmpty()) { + updateListener.onResponse(Boolean.TRUE); + return; } + + // calendarJobIds may be a group or job + jobConfigProvider.expandGroupIds(calendarJobIds, ActionListener.wrap( + expandedIds -> { + threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME).execute(() -> { + // Merge the expended group members with the request Ids. + // Ids that aren't jobs will be filtered by by isJobOpen() + expandedIds.addAll(calendarJobIds); + + for (String jobId : expandedIds) { + if (isJobOpen(clusterState, jobId)) { + updateJobProcessNotifier.submitJobUpdate(UpdateParams.scheduledEventsUpdate(jobId), ActionListener.wrap( + isUpdated -> { + if (isUpdated) { + auditor.info(jobId, Messages.getMessage(Messages.JOB_AUDIT_CALENDARS_UPDATED_ON_PROCESS)); + } + }, + e -> logger.error("[" + jobId + "] failed submitting process update on calendar change", e) + )); + } + } + + updateListener.onResponse(Boolean.TRUE); + }); + }, + updateListener::onFailure + )); } public void deleteJob(DeleteJobAction.Request request, JobStorageDeletionTask task, @@ -494,10 +520,9 @@ public void deleteJob(DeleteJobAction.Request request, JobStorageDeletionTask ta String jobId = request.getJobId(); logger.debug("Deleting job '" + jobId + "'"); - // Step 4. When the job has been removed from the cluster state, return a response - // ------- - CheckedConsumer apiResponseHandler = jobDeleted -> { - if (jobDeleted) { + // Step 4. When the job config has been deleted, return a response + Consumer deleteConfigResponseHandler = configDeleted -> { + if (configDeleted) { logger.info("Job [" + jobId + "] deleted"); auditor.info(jobId, Messages.getMessage(Messages.JOB_AUDIT_DELETED)); actionListener.onResponse(new DeleteJobAction.Response(true)); @@ -506,32 +531,22 @@ public void deleteJob(DeleteJobAction.Request request, JobStorageDeletionTask ta } }; - // Step 3. When the physical storage has been deleted, remove from Cluster State - // ------- - CheckedConsumer deleteJobStateHandler = response -> clusterService.submitStateUpdateTask("delete-job-" + jobId, - new AckedClusterStateUpdateTask(request, ActionListener.wrap(apiResponseHandler, actionListener::onFailure)) { - - @Override - protected Boolean newResponse(boolean acknowledged) { - return acknowledged && response; - } - - @Override - public ClusterState execute(ClusterState currentState) { - MlMetadata currentMlMetadata = MlMetadata.getMlMetadata(currentState); - if (currentMlMetadata.getJobs().containsKey(jobId) == false) { + // Step 3. When the physical storage has been deleted, remove the job configuration + CheckedConsumer deleteJobStateHandler = response -> { + jobConfigProvider.deleteJob(jobId, ActionListener.wrap( + deleteResponse -> deleteConfigResponseHandler.accept(Boolean.TRUE), + e -> { + if (e instanceof ResourceNotFoundException) { // We wouldn't have got here if the job never existed so // the Job must have been deleted by another action. // Don't error in this case - return currentState; + deleteConfigResponseHandler.accept(Boolean.TRUE); + } else { + actionListener.onFailure(e); } - - MlMetadata.Builder builder = new MlMetadata.Builder(currentMlMetadata); - builder.deleteJob(jobId, currentState.getMetaData().custom(PersistentTasksCustomMetaData.TYPE)); - return buildNewClusterState(currentState, builder); } - }); - + )); + }; // Step 2. Remove the job from any calendars CheckedConsumer removeFromCalendarsHandler = response -> { @@ -539,9 +554,7 @@ public ClusterState execute(ClusterState currentState) { actionListener::onFailure )); }; - // Step 1. Delete the physical storage - // This task manages the physical deletion of the job state and results task.delete(jobId, client, clusterService.state(), removeFromCalendarsHandler, actionListener::onFailure); } @@ -575,46 +588,27 @@ public void revertSnapshot(RevertModelSnapshotAction.Request request, ActionList } }; - // Step 1. Do the cluster state update + // Step 1. update the job // ------- - Consumer clusterStateHandler = response -> clusterService.submitStateUpdateTask("revert-snapshot-" + request.getJobId(), - new AckedClusterStateUpdateTask(request, ActionListener.wrap(updateHandler, actionListener::onFailure)) { - - @Override - protected Boolean newResponse(boolean acknowledged) { - if (acknowledged) { - auditor.info(request.getJobId(), Messages.getMessage(Messages.JOB_AUDIT_REVERTED, modelSnapshot.getDescription())); - return true; - } - actionListener.onFailure(new IllegalStateException("Could not revert modelSnapshot on job [" - + request.getJobId() + "], not acknowledged by master.")); - return false; - } - - @Override - public ClusterState execute(ClusterState currentState) { - Job job = getJobOrThrowIfUnknown(request.getJobId(), currentState); - Job.Builder builder = new Job.Builder(job); - builder.setModelSnapshotId(modelSnapshot.getSnapshotId()); - builder.setEstablishedModelMemory(response); - return updateClusterState(builder.build(), true, currentState); - } - }); + Consumer updateJobHandler = response -> { + JobUpdate update = new JobUpdate.Builder(request.getJobId()) + .setModelSnapshotId(modelSnapshot.getSnapshotId()) + .setEstablishedModelMemory(response) + .build(); + + jobConfigProvider.updateJob(request.getJobId(), update, maxModelMemoryLimit, ActionListener.wrap( + job -> { + auditor.info(request.getJobId(), + Messages.getMessage(Messages.JOB_AUDIT_REVERTED, modelSnapshot.getDescription())); + updateHandler.accept(Boolean.TRUE); + }, + actionListener::onFailure + )); + }; // Step 0. Find the appropriate established model memory for the reverted job // ------- - jobResultsProvider.getEstablishedMemoryUsage(request.getJobId(), modelSizeStats.getTimestamp(), modelSizeStats, clusterStateHandler, + jobResultsProvider.getEstablishedMemoryUsage(request.getJobId(), modelSizeStats.getTimestamp(), modelSizeStats, updateJobHandler, actionListener::onFailure); } - - private static MlMetadata.Builder createMlMetadataBuilder(ClusterState currentState) { - return new MlMetadata.Builder(MlMetadata.getMlMetadata(currentState)); - } - - private static ClusterState buildNewClusterState(ClusterState currentState, MlMetadata.Builder builder) { - XPackPlugin.checkReadyForXPackCustomMetadata(currentState); - ClusterState.Builder newState = ClusterState.builder(currentState); - newState.metaData(MetaData.builder(currentState.getMetaData()).putCustom(MlMetadata.TYPE, builder.build()).build()); - return newState.build(); - } } diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java index 1b89ecb1250ce..e7cd47b4e8414 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java @@ -42,6 +42,7 @@ import org.elasticsearch.index.query.WildcardQueryBuilder; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.builder.SearchSourceBuilder; +import org.elasticsearch.search.fetch.subphase.FetchSourceContext; import org.elasticsearch.xpack.core.ml.job.config.Job; import org.elasticsearch.xpack.core.ml.job.config.JobUpdate; import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex; @@ -107,7 +108,7 @@ public void getJob(String jobId, ActionListener jobListener) { GetRequest getRequest = new GetRequest(AnomalyDetectorsIndex.configIndexName(), ElasticsearchMappings.DOC_TYPE, Job.documentId(jobId)); - executeAsyncWithOrigin(client, ML_ORIGIN, GetAction.INSTANCE, getRequest, new ActionListener() { + executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, getRequest, new ActionListener() { @Override public void onResponse(GetResponse getResponse) { if (getResponse.isExists() == false) { @@ -123,7 +124,7 @@ public void onResponse(GetResponse getResponse) { public void onFailure(Exception e) { jobListener.onFailure(e); } - }); + }, client::get); } /** @@ -156,7 +157,7 @@ public void onFailure(Exception e) { } /** - * Get the job and update it by applying {@code jobUpdater} then index the changed job + * Get the job and update it by applying {@code update} then index the changed job * setting the version in the request. Applying the update may cause a validation error * which is returned via {@code updatedJobListener} * @@ -197,26 +198,75 @@ public void onResponse(GetResponse getResponse) { return; } - try (XContentBuilder builder = XContentFactory.jsonBuilder()) { - XContentBuilder updatedSource = updatedJob.toXContent(builder, ToXContent.EMPTY_PARAMS); - IndexRequest indexRequest = client.prepareIndex(AnomalyDetectorsIndex.configIndexName(), - ElasticsearchMappings.DOC_TYPE, Job.documentId(updatedJob.getId())) - .setSource(updatedSource) - .setVersion(version) - .request(); - - executeAsyncWithOrigin(client, ML_ORIGIN, IndexAction.INSTANCE, indexRequest, ActionListener.wrap( - indexResponse -> { - assert indexResponse.getResult() == DocWriteResponse.Result.UPDATED; - updatedJobListener.onResponse(updatedJob); - }, - updatedJobListener::onFailure - )); + indexUpdatedJob(updatedJob, version, updatedJobListener); + } - } catch (IOException e) { + @Override + public void onFailure(Exception e) { + updatedJobListener.onFailure(e); + } + }); + } + + /** + * Job update validation function. + * {@code updatedListener} must be called by implementations reporting + * either an validation error or success. + */ + @FunctionalInterface + public interface UpdateValidator { + void validate(Job job, JobUpdate update, ActionListener updatedListener); + } + + /** + * Similar to {@link #updateJob(String, JobUpdate, ByteSizeValue, ActionListener)} but + * with an extra validation step which is called before the updated is applied. + * + * @param jobId The Id of the job to update + * @param update The job update + * @param maxModelMemoryLimit The maximum model memory allowed + * @param validator The job update validator + * @param updatedJobListener Updated job listener + */ + public void updateJobWithValidation(String jobId, JobUpdate update, ByteSizeValue maxModelMemoryLimit, + UpdateValidator validator, ActionListener updatedJobListener) { + GetRequest getRequest = new GetRequest(AnomalyDetectorsIndex.configIndexName(), + ElasticsearchMappings.DOC_TYPE, Job.documentId(jobId)); + + executeAsyncWithOrigin(client, ML_ORIGIN, GetAction.INSTANCE, getRequest, new ActionListener() { + @Override + public void onResponse(GetResponse getResponse) { + if (getResponse.isExists() == false) { + updatedJobListener.onFailure(ExceptionsHelper.missingJobException(jobId)); + return; + } + + long version = getResponse.getVersion(); + BytesReference source = getResponse.getSourceAsBytesRef(); + Job originalJob; + try { + originalJob = parseJobLenientlyFromSource(source).build(); + } catch (Exception e) { updatedJobListener.onFailure( - new ElasticsearchParseException("Failed to serialise job with id [" + jobId + "]", e)); + new ElasticsearchParseException("Failed to parse job configuration [" + jobId + "]", e)); + return; } + + validator.validate(originalJob, update, ActionListener.wrap( + validated -> { + Job updatedJob; + try { + // Applying the update may result in a validation error + updatedJob = update.mergeWithJob(originalJob, maxModelMemoryLimit); + } catch (Exception e) { + updatedJobListener.onFailure(e); + return; + } + + indexUpdatedJob(updatedJob, version, updatedJobListener); + }, + updatedJobListener::onFailure + )); } @Override @@ -226,6 +276,61 @@ public void onFailure(Exception e) { }); } + private void indexUpdatedJob(Job updatedJob, long version, ActionListener updatedJobListener) { + try (XContentBuilder builder = XContentFactory.jsonBuilder()) { + XContentBuilder updatedSource = updatedJob.toXContent(builder, ToXContent.EMPTY_PARAMS); + IndexRequest indexRequest = client.prepareIndex(AnomalyDetectorsIndex.configIndexName(), + ElasticsearchMappings.DOC_TYPE, Job.documentId(updatedJob.getId())) + .setSource(updatedSource) + .setVersion(version) + .request(); + + executeAsyncWithOrigin(client, ML_ORIGIN, IndexAction.INSTANCE, indexRequest, ActionListener.wrap( + indexResponse -> { + assert indexResponse.getResult() == DocWriteResponse.Result.UPDATED; + updatedJobListener.onResponse(updatedJob); + }, + updatedJobListener::onFailure + )); + + } catch (IOException e) { + updatedJobListener.onFailure( + new ElasticsearchParseException("Failed to serialise job with id [" + updatedJob.getId() + "]", e)); + } + } + + + /** + * Check a job exists. A job exists if it has a configuration document. + * + * If the job does not exist a ResourceNotFoundException is returned to the listener, + * FALSE will never be returned only TRUE or ResourceNotFoundException + * + * @param jobId The jobId to check + * @param listener Exists listener + */ + public void checkJobExists(String jobId, ActionListener listener) { + GetRequest getRequest = new GetRequest(AnomalyDetectorsIndex.configIndexName(), + ElasticsearchMappings.DOC_TYPE, Job.documentId(jobId)); + getRequest.fetchSourceContext(FetchSourceContext.DO_NOT_FETCH_SOURCE); + + executeAsyncWithOrigin(client, ML_ORIGIN, GetAction.INSTANCE, getRequest, new ActionListener() { + @Override + public void onResponse(GetResponse getResponse) { + if (getResponse.isExists() == false) { + listener.onFailure(ExceptionsHelper.missingJobException(jobId)); + } else { + listener.onResponse(Boolean.TRUE); + } + } + + @Override + public void onFailure(Exception e) { + listener.onFailure(e); + } + }); + } + /** * Expands an expression into the set of matching names. {@code expresssion} * may be a wildcard, a job group, a job ID or a list of those. @@ -351,6 +456,39 @@ public void expandJobs(String expression, boolean allowNoJobs, ActionListener
  • groupIds, ActionListener> listener) { + SearchSourceBuilder sourceBuilder = new SearchSourceBuilder() + .query(new TermsQueryBuilder(Job.GROUPS.getPreferredName(), groupIds)); + sourceBuilder.sort(Job.ID.getPreferredName()); + String [] includes = new String[] {Job.ID.getPreferredName()}; + sourceBuilder.fetchSource(includes, null); + + SearchRequest searchRequest = client.prepareSearch(AnomalyDetectorsIndex.configIndexName()) + .setIndicesOptions(IndicesOptions.lenientExpandOpen()) + .setSource(sourceBuilder).request(); + + executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, searchRequest, + ActionListener.wrap( + response -> { + Set jobIds = new HashSet<>(); + SearchHit[] hits = response.getHits().getHits(); + for (SearchHit hit : hits) { + jobIds.add((String)hit.getSourceAsMap().get(Job.ID.getPreferredName())); + } + + listener.onResponse(jobIds); + }, + listener::onFailure) + , client::search); + } + private void parseJobLenientlyFromSource(BytesReference source, ActionListener jobListener) { try (InputStream stream = source.streamInput(); XContentParser parser = XContentFactory.xContent(XContentType.JSON) diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/AutodetectProcessManager.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/AutodetectProcessManager.java index 063ab3b49d146..b0ef2dd9821fb 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/AutodetectProcessManager.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/AutodetectProcessManager.java @@ -359,10 +359,20 @@ public void writeUpdateProcessMessage(JobTask jobTask, UpdateParams updateParams updateProcessMessage.setFilter(filter); if (updateParams.isUpdateScheduledEvents()) { - Job job = jobManager.getJobOrThrowIfUnknown(jobTask.getJobId()); - DataCounts dataCounts = getStatistics(jobTask).get().v1(); - ScheduledEventsQueryBuilder query = new ScheduledEventsQueryBuilder().start(job.earliestValidTimestamp(dataCounts)); - jobResultsProvider.scheduledEventsForJob(jobTask.getJobId(), job.getGroups(), query, eventsListener); + jobManager.getJob(jobTask.getJobId(), new ActionListener() { + @Override + public void onResponse(Job job) { + DataCounts dataCounts = getStatistics(jobTask).get().v1(); + ScheduledEventsQueryBuilder query = new ScheduledEventsQueryBuilder() + .start(job.earliestValidTimestamp(dataCounts)); + jobResultsProvider.scheduledEventsForJob(jobTask.getJobId(), job.getGroups(), query, eventsListener); + } + + @Override + public void onFailure(Exception e) { + handler.accept(e); + } + }); } else { eventsListener.onResponse(null); } @@ -393,69 +403,77 @@ public void onFailure(Exception e) { public void openJob(JobTask jobTask, Consumer handler) { String jobId = jobTask.getJobId(); - Job job = jobManager.getJobOrThrowIfUnknown(jobId); - - if (job.getJobVersion() == null) { - handler.accept(ExceptionsHelper.badRequestException("Cannot open job [" + jobId - + "] because jobs created prior to version 5.5 are not supported")); - return; - } - logger.info("Opening job [{}]", jobId); - processByAllocation.putIfAbsent(jobTask.getAllocationId(), new ProcessContext(jobTask)); - jobResultsProvider.getAutodetectParams(job, params -> { - // We need to fork, otherwise we restore model state from a network thread (several GET api calls): - threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME).execute(new AbstractRunnable() { - @Override - public void onFailure(Exception e) { - handler.accept(e); - } - @Override - protected void doRun() throws Exception { - ProcessContext processContext = processByAllocation.get(jobTask.getAllocationId()); - if (processContext == null) { - logger.debug("Aborted opening job [{}] as it has been closed", jobId); - return; - } - if (processContext.getState() != ProcessContext.ProcessStateName.NOT_RUNNING) { - logger.debug("Cannot open job [{}] when its state is [{}]", jobId, processContext.getState().getClass().getName()); + jobManager.getJob(jobId, ActionListener.wrap( + // NORELEASE JIndex. Should not be doing this work on the network thread + job -> { + if (job.getJobVersion() == null) { + handler.accept(ExceptionsHelper.badRequestException("Cannot open job [" + jobId + + "] because jobs created prior to version 5.5 are not supported")); return; } - try { - createProcessAndSetRunning(processContext, params, handler); - processContext.getAutodetectCommunicator().init(params.modelSnapshot()); - setJobState(jobTask, JobState.OPENED); - } catch (Exception e1) { - // No need to log here as the persistent task framework will log it - try { - // Don't leave a partially initialised process hanging around - processContext.newKillBuilder() - .setAwaitCompletion(false) - .setFinish(false) - .kill(); - processByAllocation.remove(jobTask.getAllocationId()); - } finally { - setJobState(jobTask, JobState.FAILED, e2 -> handler.accept(e1)); - } - } - } - }); - }, e1 -> { - logger.warn("Failed to gather information required to open job [" + jobId + "]", e1); - setJobState(jobTask, JobState.FAILED, e2 -> handler.accept(e1)); - }); + + processByAllocation.putIfAbsent(jobTask.getAllocationId(), new ProcessContext(jobTask)); + jobResultsProvider.getAutodetectParams(job, params -> { + // We need to fork, otherwise we restore model state from a network thread (several GET api calls): + threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME).execute(new AbstractRunnable() { + @Override + public void onFailure(Exception e) { + handler.accept(e); + } + + @Override + protected void doRun() { + ProcessContext processContext = processByAllocation.get(jobTask.getAllocationId()); + if (processContext == null) { + logger.debug("Aborted opening job [{}] as it has been closed", jobId); + return; + } + if (processContext.getState() != ProcessContext.ProcessStateName.NOT_RUNNING) { + logger.debug("Cannot open job [{}] when its state is [{}]", + jobId, processContext.getState().getClass().getName()); + return; + } + + try { + createProcessAndSetRunning(processContext, job, params, handler); + processContext.getAutodetectCommunicator().init(params.modelSnapshot()); + setJobState(jobTask, JobState.OPENED); + } catch (Exception e1) { + // No need to log here as the persistent task framework will log it + try { + // Don't leave a partially initialised process hanging around + processContext.newKillBuilder() + .setAwaitCompletion(false) + .setFinish(false) + .kill(); + processByAllocation.remove(jobTask.getAllocationId()); + } finally { + setJobState(jobTask, JobState.FAILED, e2 -> handler.accept(e1)); + } + } + } + }); + }, e1 -> { + logger.warn("Failed to gather information required to open job [" + jobId + "]", e1); + setJobState(jobTask, JobState.FAILED, e2 -> handler.accept(e1)); + }); + }, + handler + )); + } - private void createProcessAndSetRunning(ProcessContext processContext, AutodetectParams params, Consumer handler) { + private void createProcessAndSetRunning(ProcessContext processContext, Job job, AutodetectParams params, Consumer handler) { // At this point we lock the process context until the process has been started. // The reason behind this is to ensure closing the job does not happen before // the process is started as that can result to the job getting seemingly closed // but the actual process is hanging alive. processContext.tryLock(); try { - AutodetectCommunicator communicator = create(processContext.getJobTask(), params, handler); + AutodetectCommunicator communicator = create(processContext.getJobTask(), job, params, handler); processContext.setRunning(communicator); } finally { // Now that the process is running and we have updated its state we can unlock. @@ -465,7 +483,7 @@ private void createProcessAndSetRunning(ProcessContext processContext, Autodetec } } - AutodetectCommunicator create(JobTask jobTask, AutodetectParams autodetectParams, Consumer handler) { + AutodetectCommunicator create(JobTask jobTask, Job job, AutodetectParams autodetectParams, Consumer handler) { // Closing jobs can still be using some or all threads in MachineLearning.AUTODETECT_THREAD_POOL_NAME // that an open job uses, so include them too when considering if enough threads are available. int currentRunningJobs = processByAllocation.size(); @@ -490,7 +508,6 @@ AutodetectCommunicator create(JobTask jobTask, AutodetectParams autodetectParams } } - Job job = jobManager.getJobOrThrowIfUnknown(jobId); // A TP with no queue, so that we fail immediately if there are no threads available ExecutorService autoDetectExecutorService = threadPool.executor(MachineLearning.AUTODETECT_THREAD_POOL_NAME); DataCountsReporter dataCountsReporter = new DataCountsReporter(settings, job, autodetectParams.dataCounts(), diff --git a/x-pack/plugin/ml/src/test/java/org/elasticsearch/license/MachineLearningLicensingTests.java b/x-pack/plugin/ml/src/test/java/org/elasticsearch/license/MachineLearningLicensingTests.java index 643eb8d5a5f9c..1f3f432747399 100644 --- a/x-pack/plugin/ml/src/test/java/org/elasticsearch/license/MachineLearningLicensingTests.java +++ b/x-pack/plugin/ml/src/test/java/org/elasticsearch/license/MachineLearningLicensingTests.java @@ -87,6 +87,7 @@ public void testMachineLearningPutJobActionRestricted() { } } + @AwaitsFix(bugUrl = "JIndex development") public void testMachineLearningOpenJobActionRestricted() throws Exception { String jobId = "testmachinelearningopenjobactionrestricted"; assertMLAllowed(true); @@ -138,6 +139,7 @@ public void testMachineLearningOpenJobActionRestricted() throws Exception { } } + @AwaitsFix(bugUrl = "JIndex development") public void testMachineLearningPutDatafeedActionRestricted() throws Exception { String jobId = "testmachinelearningputdatafeedactionrestricted"; String datafeedId = jobId + "-datafeed"; @@ -185,6 +187,7 @@ public void testMachineLearningPutDatafeedActionRestricted() throws Exception { } } + @AwaitsFix(bugUrl = "JIndex development") public void testAutoCloseJobWithDatafeed() throws Exception { String jobId = "testautoclosejobwithdatafeed"; String datafeedId = jobId + "-datafeed"; @@ -287,6 +290,7 @@ public void testAutoCloseJobWithDatafeed() throws Exception { }); } + @AwaitsFix(bugUrl = "JIndex development") public void testMachineLearningStartDatafeedActionRestricted() throws Exception { String jobId = "testmachinelearningstartdatafeedactionrestricted"; String datafeedId = jobId + "-datafeed"; @@ -361,6 +365,7 @@ public void testMachineLearningStartDatafeedActionRestricted() throws Exception } } + @AwaitsFix(bugUrl = "JIndex development") public void testMachineLearningStopDatafeedActionNotRestricted() throws Exception { String jobId = "testmachinelearningstopdatafeedactionnotrestricted"; String datafeedId = jobId + "-datafeed"; @@ -427,6 +432,7 @@ public void testMachineLearningStopDatafeedActionNotRestricted() throws Exceptio } } + @AwaitsFix(bugUrl = "JIndex development") public void testMachineLearningCloseJobActionNotRestricted() throws Exception { String jobId = "testmachinelearningclosejobactionnotrestricted"; assertMLAllowed(true); @@ -470,6 +476,7 @@ public void testMachineLearningCloseJobActionNotRestricted() throws Exception { } } + @AwaitsFix(bugUrl = "JIndex development") public void testMachineLearningDeleteJobActionNotRestricted() throws Exception { String jobId = "testmachinelearningclosejobactionnotrestricted"; assertMLAllowed(true); @@ -495,6 +502,7 @@ public void testMachineLearningDeleteJobActionNotRestricted() throws Exception { } } + @AwaitsFix(bugUrl = "JIndex development") public void testMachineLearningDeleteDatafeedActionNotRestricted() throws Exception { String jobId = "testmachinelearningdeletedatafeedactionnotrestricted"; String datafeedId = jobId + "-datafeed"; diff --git a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlSingleNodeTestCase.java b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlSingleNodeTestCase.java index c838b1f175a33..489e294a75789 100644 --- a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlSingleNodeTestCase.java +++ b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlSingleNodeTestCase.java @@ -78,7 +78,7 @@ protected T blockingCall(Consumer> function) throws Except AtomicReference responseHolder = new AtomicReference<>(); blockingCall(function, responseHolder, exceptionHolder); if (exceptionHolder.get() != null) { - assertNotNull(exceptionHolder.get().getMessage(), exceptionHolder.get()); + assertNull(exceptionHolder.get().getMessage(), exceptionHolder.get()); } return responseHolder.get(); } diff --git a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlTasksTests.java b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlTasksTests.java index 687292b3c85d4..53bdfbdcb3b69 100644 --- a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlTasksTests.java +++ b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlTasksTests.java @@ -15,6 +15,9 @@ import org.elasticsearch.xpack.core.ml.job.config.JobState; import org.elasticsearch.xpack.core.ml.job.config.JobTaskState; +import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.hamcrest.Matchers.empty; + public class MlTasksTests extends ESTestCase { public void testGetJobState() { PersistentTasksCustomMetaData.Builder tasksBuilder = PersistentTasksCustomMetaData.builder(); @@ -65,4 +68,19 @@ public void testGetDatafeedTask() { assertNotNull(MlTasks.getDatafeedTask("foo", tasksBuilder.build())); assertNull(MlTasks.getDatafeedTask("other", tasksBuilder.build())); } + + public void testOpenJobIds() { + PersistentTasksCustomMetaData.Builder tasksBuilder = PersistentTasksCustomMetaData.builder(); + assertThat(MlTasks.openJobIds(tasksBuilder.build()), empty()); + + tasksBuilder.addTask(MlTasks.jobTaskId("foo-1"), OpenJobAction.TASK_NAME, new OpenJobAction.JobParams("foo-1"), + new PersistentTasksCustomMetaData.Assignment("node-1", "test assignment")); + tasksBuilder.addTask(MlTasks.jobTaskId("bar"), OpenJobAction.TASK_NAME, new OpenJobAction.JobParams("bar"), + new PersistentTasksCustomMetaData.Assignment("node-1", "test assignment")); + tasksBuilder.addTask(MlTasks.datafeedTaskId("df"), StartDatafeedAction.TASK_NAME, + new StartDatafeedAction.DatafeedParams("df", 0L), + new PersistentTasksCustomMetaData.Assignment("node-1", "test assignment")); + + assertThat(MlTasks.openJobIds(tasksBuilder.build()), containsInAnyOrder("foo-1", "bar")); + } } diff --git a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java index d85d8e1d8cbcd..d478e09553c13 100644 --- a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java +++ b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java @@ -36,6 +36,7 @@ import static org.hamcrest.Matchers.containsInAnyOrder; import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.empty; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.core.IsInstanceOf.instanceOf; @@ -60,6 +61,29 @@ public void testGetMissingJob() throws InterruptedException { assertThat(exceptionHolder.get(), instanceOf(ResourceNotFoundException.class)); } + public void testCheckJobExists() throws InterruptedException { + AtomicReference jobExistsHolder = new AtomicReference<>(); + AtomicReference exceptionHolder = new AtomicReference<>(); + + blockingCall(actionListener -> jobConfigProvider.checkJobExists("missing", actionListener), jobExistsHolder, exceptionHolder); + + assertNull(jobExistsHolder.get()); + assertNotNull(exceptionHolder.get()); + assertThat(exceptionHolder.get(), instanceOf(ResourceNotFoundException.class)); + + AtomicReference indexResponseHolder = new AtomicReference<>(); + + // Create job + Job job = createJob("existing-job", null).build(new Date()); + blockingCall(actionListener -> jobConfigProvider.putJob(job, actionListener), indexResponseHolder, exceptionHolder); + + exceptionHolder.set(null); + blockingCall(actionListener -> jobConfigProvider.checkJobExists("existing-job", actionListener), jobExistsHolder, exceptionHolder); + assertNull(exceptionHolder.get()); + assertNotNull(jobExistsHolder.get()); + assertTrue(jobExistsHolder.get()); + } + public void testOverwriteNotAllowed() throws InterruptedException { final String jobId = "same-id"; @@ -163,6 +187,46 @@ public void testUpdateWithAValidationError() throws Exception { assertThat(exceptionHolder.get().getMessage(), containsString("Invalid detector rule:")); } + public void testUpdateWithValidator() throws Exception { + final String jobId = "job-update-with-validator"; + + // Create job + Job newJob = createJob(jobId, null).build(new Date()); + this.blockingCall(actionListener -> jobConfigProvider.putJob(newJob, actionListener)); + + JobUpdate jobUpdate = new JobUpdate.Builder(jobId).setDescription("This job has been updated").build(); + + JobConfigProvider.UpdateValidator validator = (job, update, listener) -> { + listener.onResponse(null); + }; + + AtomicReference exceptionHolder = new AtomicReference<>(); + AtomicReference updateJobResponseHolder = new AtomicReference<>(); + // update with the no-op validator + blockingCall(actionListener -> + jobConfigProvider.updateJobWithValidation(jobId, jobUpdate, new ByteSizeValue(32), validator, actionListener), + updateJobResponseHolder, exceptionHolder); + + assertNull(exceptionHolder.get()); + assertNotNull(updateJobResponseHolder.get()); + assertEquals("This job has been updated", updateJobResponseHolder.get().getDescription()); + + JobConfigProvider.UpdateValidator validatorWithAnError = (job, update, listener) -> { + listener.onFailure(new IllegalStateException("I don't like this update")); + }; + + updateJobResponseHolder.set(null); + // Update with a validator that errors + blockingCall(actionListener -> jobConfigProvider.updateJobWithValidation(jobId, jobUpdate, new ByteSizeValue(32), + validatorWithAnError, actionListener), + updateJobResponseHolder, exceptionHolder); + + assertNull(updateJobResponseHolder.get()); + assertNotNull(exceptionHolder.get()); + assertThat(exceptionHolder.get(), instanceOf(IllegalStateException.class)); + assertThat(exceptionHolder.get().getMessage(), containsString("I don't like this update")); + } + public void testAllowNoJobs() throws InterruptedException { AtomicReference> jobIdsHolder = new AtomicReference<>(); AtomicReference exceptionHolder = new AtomicReference<>(); @@ -297,6 +361,33 @@ public void testExpandJobs_WildCardExpansion() throws Exception { assertThat(expandedJobs, containsInAnyOrder(bar1)); } + public void testExpandGroups() throws Exception { + putJob(createJob("apples", Collections.singletonList("fruit"))); + putJob(createJob("pears", Collections.singletonList("fruit"))); + putJob(createJob("broccoli", Collections.singletonList("veg"))); + putJob(createJob("potato", Collections.singletonList("veg"))); + putJob(createJob("tomato", Arrays.asList("fruit", "veg"))); + putJob(createJob("unrelated", Collections.emptyList())); + + client().admin().indices().prepareRefresh(AnomalyDetectorsIndex.configIndexName()).get(); + + Set expandedIds = blockingCall(actionListener -> + jobConfigProvider.expandGroupIds(Collections.singletonList("fruit"), actionListener)); + assertThat(expandedIds, containsInAnyOrder("apples", "pears", "tomato")); + + expandedIds = blockingCall(actionListener -> + jobConfigProvider.expandGroupIds(Collections.singletonList("veg"), actionListener)); + assertThat(expandedIds, containsInAnyOrder("broccoli", "potato", "tomato")); + + expandedIds = blockingCall(actionListener -> + jobConfigProvider.expandGroupIds(Arrays.asList("fruit", "veg"), actionListener)); + assertThat(expandedIds, containsInAnyOrder("apples", "pears", "broccoli", "potato", "tomato")); + + expandedIds = blockingCall(actionListener -> + jobConfigProvider.expandGroupIds(Collections.singletonList("unknown-group"), actionListener)); + assertThat(expandedIds, empty()); + } + private Job.Builder createJob(String jobId, List groups) { Detector.Builder d1 = new Detector.Builder("info_content", "domain"); d1.setOverFieldName("client"); diff --git a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/JobManagerTests.java b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/JobManagerTests.java index a9162cb2ae4df..0361ba49a7ae7 100644 --- a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/JobManagerTests.java +++ b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/JobManagerTests.java @@ -8,19 +8,25 @@ import org.elasticsearch.ResourceAlreadyExistsException; import org.elasticsearch.ResourceNotFoundException; import org.elasticsearch.action.ActionListener; +import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.client.Client; import org.elasticsearch.cluster.AckedClusterStateUpdateTask; import org.elasticsearch.cluster.ClusterName; import org.elasticsearch.cluster.ClusterState; import org.elasticsearch.cluster.metadata.MetaData; import org.elasticsearch.cluster.service.ClusterService; +import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.settings.ClusterSettings; import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.xcontent.ToXContent; +import org.elasticsearch.common.xcontent.XContentBuilder; +import org.elasticsearch.common.xcontent.XContentFactory; import org.elasticsearch.env.Environment; import org.elasticsearch.env.TestEnvironment; import org.elasticsearch.index.analysis.AnalysisRegistry; import org.elasticsearch.persistent.PersistentTasksCustomMetaData; import org.elasticsearch.test.ESTestCase; +import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.xpack.core.ml.MachineLearningField; import org.elasticsearch.xpack.core.ml.MlMetadata; import org.elasticsearch.xpack.core.ml.action.PutJobAction; @@ -33,8 +39,11 @@ import org.elasticsearch.xpack.core.ml.job.config.JobState; import org.elasticsearch.xpack.core.ml.job.config.MlFilter; import org.elasticsearch.xpack.core.ml.job.config.RuleScope; +import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex; +import org.elasticsearch.xpack.ml.MachineLearning; import org.elasticsearch.xpack.ml.job.categorization.CategorizationAnalyzerTests; import org.elasticsearch.xpack.ml.job.persistence.JobResultsProvider; +import org.elasticsearch.xpack.ml.job.persistence.MockClientBuilder; import org.elasticsearch.xpack.ml.job.process.autodetect.UpdateParams; import org.elasticsearch.xpack.ml.notifications.Auditor; import org.junit.Before; @@ -43,16 +52,23 @@ import org.mockito.Mockito; import java.io.IOException; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Date; import java.util.List; import java.util.TreeSet; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; import static org.elasticsearch.xpack.core.ml.job.config.JobTests.buildJobBuilder; import static org.elasticsearch.xpack.ml.action.TransportOpenJobActionTests.addJobTask; +import static org.hamcrest.Matchers.containsInAnyOrder; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.hamcrest.Matchers.hasSize; +import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.lessThanOrEqualTo; import static org.mockito.Matchers.any; @@ -66,8 +82,8 @@ public class JobManagerTests extends ESTestCase { private Environment environment; private AnalysisRegistry analysisRegistry; - private Client client; private ClusterService clusterService; + private ThreadPool threadPool; private JobResultsProvider jobResultsProvider; private Auditor auditor; private UpdateJobProcessNotifier updateJobProcessNotifier; @@ -77,47 +93,134 @@ public void setup() throws Exception { Settings settings = Settings.builder().put(Environment.PATH_HOME_SETTING.getKey(), createTempDir()).build(); environment = TestEnvironment.newEnvironment(settings); analysisRegistry = CategorizationAnalyzerTests.buildTestAnalysisRegistry(environment); - client = mock(Client.class); clusterService = mock(ClusterService.class); + jobResultsProvider = mock(JobResultsProvider.class); auditor = mock(Auditor.class); updateJobProcessNotifier = mock(UpdateJobProcessNotifier.class); + + ExecutorService executorService = mock(ExecutorService.class); + threadPool = mock(ThreadPool.class); + org.elasticsearch.mock.orig.Mockito.doAnswer(invocation -> { + ((Runnable) invocation.getArguments()[0]).run(); + return null; + }).when(executorService).execute(any(Runnable.class)); + when(threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME)).thenReturn(executorService); } - public void testGetJobOrThrowIfUnknown_GivenUnknownJob() { - ClusterState cs = createClusterState(); - ESTestCase.expectThrows(ResourceNotFoundException.class, () -> JobManager.getJobOrThrowIfUnknown("foo", cs)); + public void testGetJobNotInIndexOrCluster() { + MlMetadata.Builder mlMetadata = new MlMetadata.Builder(); + + ClusterState clusterState = ClusterState.builder(new ClusterName("_name")) + .metaData(MetaData.builder() + .putCustom(MlMetadata.TYPE, mlMetadata.build())) + .build(); + when(clusterService.state()).thenReturn(clusterState); + + // job document does not exist + GetResponse getResponse = mock(GetResponse.class); + when(getResponse.isExists()).thenReturn(false); + MockClientBuilder mockClientBuilder = new MockClientBuilder("jm-test"); + mockClientBuilder.get(getResponse); + + JobManager jobManager = createJobManager(mockClientBuilder.build()); + + AtomicReference exceptionHolder = new AtomicReference<>(); + jobManager.getJob("non-job", ActionListener.wrap( + job -> fail("Job not expected"), + e -> exceptionHolder.set(e) + )); + + assertNotNull(exceptionHolder.get()); + assertThat(exceptionHolder.get(), instanceOf(ResourceNotFoundException.class)); } - public void testGetJobOrThrowIfUnknown_GivenKnownJob() { - Job job = buildJobBuilder("foo").build(); - MlMetadata mlMetadata = new MlMetadata.Builder().putJob(job, false).build(); - ClusterState cs = ClusterState.builder(new ClusterName("_name")) - .metaData(MetaData.builder().putCustom(MlMetadata.TYPE, mlMetadata)).build(); + public void testGetJobFromClusterWhenNotInIndex() { + String clusterJobId = "cluster-job"; + Job clusterJob = buildJobBuilder(clusterJobId).build(); + + MlMetadata.Builder mlMetadata = new MlMetadata.Builder(); + mlMetadata.putJob(clusterJob, false); + + ClusterState clusterState = ClusterState.builder(new ClusterName("_name")) + .metaData(MetaData.builder() + .putCustom(MlMetadata.TYPE, mlMetadata.build())) + .build(); + when(clusterService.state()).thenReturn(clusterState); + + // job document does not exist + GetResponse getResponse = mock(GetResponse.class); + when(getResponse.isExists()).thenReturn(false); + MockClientBuilder mockClientBuilder = new MockClientBuilder("jm-test"); + mockClientBuilder.get(getResponse); + + JobManager jobManager = createJobManager(mockClientBuilder.build()); - assertEquals(job, JobManager.getJobOrThrowIfUnknown("foo", cs)); + AtomicReference jobHolder = new AtomicReference<>(); + jobManager.getJob(clusterJobId, ActionListener.wrap( + job -> jobHolder.set(job), + e -> fail(e.getMessage()) + )); + + assertNotNull(jobHolder.get()); + assertEquals(clusterJob, jobHolder.get()); } - public void testExpandJobs_GivenAll() { + public void testExpandJobsFromClusterStateAndIndex() throws IOException { + Job csJobFoo1 = buildJobBuilder("foo-cs-1").build(); + Job csJobFoo2 = buildJobBuilder("foo-cs-2").build(); + Job csJobBar = buildJobBuilder("bar-cs").build(); + MlMetadata.Builder mlMetadata = new MlMetadata.Builder(); - for (int i = 0; i < 3; i++) { - mlMetadata.putJob(buildJobBuilder(Integer.toString(i)).build(), false); - } + mlMetadata.putJob(csJobFoo1, false); + mlMetadata.putJob(csJobFoo2, false); + mlMetadata.putJob(csJobBar, false); + ClusterState clusterState = ClusterState.builder(new ClusterName("_name")) - .metaData(MetaData.builder().putCustom(MlMetadata.TYPE, mlMetadata.build())).build(); + .metaData(MetaData.builder() + .putCustom(MlMetadata.TYPE, mlMetadata.build())) + .build(); + when(clusterService.state()).thenReturn(clusterState); + + + List docsAsBytes = new ArrayList<>(); + + Job.Builder indexJobFoo = buildJobBuilder("foo-index"); + docsAsBytes.add(toBytesReference(indexJobFoo.build())); + + MockClientBuilder mockClientBuilder = new MockClientBuilder("cluster-test"); + mockClientBuilder.prepareSearch(AnomalyDetectorsIndex.configIndexName(), docsAsBytes); + JobManager jobManager = createJobManager(mockClientBuilder.build()); + + + AtomicReference> jobsHolder = new AtomicReference<>(); + jobManager.expandJobs("_all", true, ActionListener.wrap( + jobs -> jobsHolder.set(jobs), + e -> fail(e.getMessage()) + )); + + assertNotNull(jobsHolder.get()); + assertThat(jobsHolder.get().results(), hasSize(4)); + List jobIds = jobsHolder.get().results().stream().map(Job::getId).collect(Collectors.toList()); + assertThat(jobIds, containsInAnyOrder("foo-cs-1", "foo-cs-2", "bar-cs", "foo-index")); - JobManager jobManager = createJobManager(); - QueryPage result = jobManager.expandJobs("_all", true, clusterState); + jobsHolder.set(null); + jobManager.expandJobs("foo*", true, ActionListener.wrap( + jobs -> jobsHolder.set(jobs), + e -> fail(e.getMessage()) + )); - assertThat(result.count(), equalTo(3L)); - assertThat(result.results().get(0).getId(), equalTo("0")); - assertThat(result.results().get(1).getId(), equalTo("1")); - assertThat(result.results().get(2).getId(), equalTo("2")); + assertNotNull(jobsHolder.get()); + assertThat(jobsHolder.get().results(), hasSize(3)); + jobIds = jobsHolder.get().results().stream().map(Job::getId).collect(Collectors.toList()); + assertThat(jobIds, containsInAnyOrder("foo-cs-1", "foo-cs-2", "foo-index")); } @SuppressWarnings("unchecked") public void testPutJob_AddsCreateTime() throws IOException { - JobManager jobManager = createJobManager(); + MockClientBuilder mockClientBuilder = new MockClientBuilder("cluster-test"); + JobManager jobManager = createJobManager(mockClientBuilder.build()); + PutJobAction.Request putJobRequest = new PutJobAction.Request(createJob()); doAnswer(invocation -> { @@ -153,8 +256,10 @@ public void onFailure(Exception e) { }); } - public void testPutJob_ThrowsIfJobExists() throws IOException { - JobManager jobManager = createJobManager(); + public void testPutJob_ThrowsIfJobExistsInClusterState() throws IOException { + MockClientBuilder mockClientBuilder = new MockClientBuilder("cluster-test"); + JobManager jobManager = createJobManager(mockClientBuilder.build()); + PutJobAction.Request putJobRequest = new PutJobAction.Request(createJob()); MlMetadata.Builder mlMetadata = new MlMetadata.Builder(); @@ -177,14 +282,19 @@ public void onFailure(Exception e) { public void testNotifyFilterChangedGivenNoop() { MlFilter filter = MlFilter.builder("my_filter").build(); - JobManager jobManager = createJobManager(); + MockClientBuilder mockClientBuilder = new MockClientBuilder("cluster-test"); + JobManager jobManager = createJobManager(mockClientBuilder.build()); - jobManager.notifyFilterChanged(filter, Collections.emptySet(), Collections.emptySet()); + jobManager.notifyFilterChanged(filter, Collections.emptySet(), Collections.emptySet(), ActionListener.wrap( + r -> {}, + e -> fail(e.getMessage()) + )); Mockito.verifyNoMoreInteractions(auditor, updateJobProcessNotifier); } - public void testNotifyFilterChanged() { + @AwaitsFix(bugUrl = "Closed jobs are not audited when the filter changes") + public void testNotifyFilterChanged() throws IOException { Detector.Builder detectorReferencingFilter = new Detector.Builder("count", null); detectorReferencingFilter.setByFieldName("foo"); DetectionRule filterRule = new DetectionRule.Builder(RuleScope.builder().exclude("foo", "foo_filter")).build(); @@ -192,19 +302,21 @@ public void testNotifyFilterChanged() { AnalysisConfig.Builder filterAnalysisConfig = new AnalysisConfig.Builder(Collections.singletonList( detectorReferencingFilter.build())); + List docsAsBytes = new ArrayList<>(); + Job.Builder jobReferencingFilter1 = buildJobBuilder("job-referencing-filter-1"); jobReferencingFilter1.setAnalysisConfig(filterAnalysisConfig); + docsAsBytes.add(toBytesReference(jobReferencingFilter1.build())); + Job.Builder jobReferencingFilter2 = buildJobBuilder("job-referencing-filter-2"); jobReferencingFilter2.setAnalysisConfig(filterAnalysisConfig); + docsAsBytes.add(toBytesReference(jobReferencingFilter2.build())); + Job.Builder jobReferencingFilter3 = buildJobBuilder("job-referencing-filter-3"); - jobReferencingFilter3.setAnalysisConfig(filterAnalysisConfig); - Job.Builder jobWithoutFilter = buildJobBuilder("job-without-filter"); + jobReferencingFilter2.setAnalysisConfig(filterAnalysisConfig); - MlMetadata.Builder mlMetadata = new MlMetadata.Builder(); - mlMetadata.putJob(jobReferencingFilter1.build(), false); - mlMetadata.putJob(jobReferencingFilter2.build(), false); - mlMetadata.putJob(jobReferencingFilter3.build(), false); - mlMetadata.putJob(jobWithoutFilter.build(), false); + Job.Builder jobWithoutFilter = buildJobBuilder("job-without-filter"); + docsAsBytes.add(toBytesReference(jobWithoutFilter.build())); PersistentTasksCustomMetaData.Builder tasksBuilder = PersistentTasksCustomMetaData.builder(); addJobTask(jobReferencingFilter1.getId(), "node_id", JobState.OPENED, tasksBuilder); @@ -213,8 +325,7 @@ public void testNotifyFilterChanged() { ClusterState clusterState = ClusterState.builder(new ClusterName("_name")) .metaData(MetaData.builder() - .putCustom(PersistentTasksCustomMetaData.TYPE, tasksBuilder.build()) - .putCustom(MlMetadata.TYPE, mlMetadata.build())) + .putCustom(PersistentTasksCustomMetaData.TYPE, tasksBuilder.build())) .build(); when(clusterService.state()).thenReturn(clusterState); @@ -224,12 +335,17 @@ public void testNotifyFilterChanged() { return null; }).when(updateJobProcessNotifier).submitJobUpdate(any(), any()); - JobManager jobManager = createJobManager(); + MockClientBuilder mockClientBuilder = new MockClientBuilder("cluster-test"); + mockClientBuilder.prepareSearch(AnomalyDetectorsIndex.configIndexName(), docsAsBytes); + JobManager jobManager = createJobManager(mockClientBuilder.build()); MlFilter filter = MlFilter.builder("foo_filter").setItems("a", "b").build(); jobManager.notifyFilterChanged(filter, new TreeSet<>(Arrays.asList("item 1", "item 2")), - new TreeSet<>(Collections.singletonList("item 3"))); + new TreeSet<>(Collections.singletonList("item 3")), ActionListener.wrap( + r -> {}, + e -> fail(e.getMessage()) + )); ArgumentCaptor updateParamsCaptor = ArgumentCaptor.forClass(UpdateParams.class); verify(updateJobProcessNotifier, times(2)).submitJobUpdate(updateParamsCaptor.capture(), any(ActionListener.class)); @@ -250,7 +366,8 @@ public void testNotifyFilterChanged() { Mockito.verifyNoMoreInteractions(auditor, updateJobProcessNotifier); } - public void testNotifyFilterChangedGivenOnlyAddedItems() { + @AwaitsFix(bugUrl = "Closed jobs are not audited when the filter changes") + public void testNotifyFilterChangedGivenOnlyAddedItems() throws IOException { Detector.Builder detectorReferencingFilter = new Detector.Builder("count", null); detectorReferencingFilter.setByFieldName("foo"); DetectionRule filterRule = new DetectionRule.Builder(RuleScope.builder().exclude("foo", "foo_filter")).build(); @@ -261,26 +378,33 @@ public void testNotifyFilterChangedGivenOnlyAddedItems() { Job.Builder jobReferencingFilter = buildJobBuilder("job-referencing-filter"); jobReferencingFilter.setAnalysisConfig(filterAnalysisConfig); - MlMetadata.Builder mlMetadata = new MlMetadata.Builder(); - mlMetadata.putJob(jobReferencingFilter.build(), false); + List docsAsBytes = Collections.singletonList(toBytesReference(jobReferencingFilter.build())); + PersistentTasksCustomMetaData.Builder tasksBuilder = PersistentTasksCustomMetaData.builder(); ClusterState clusterState = ClusterState.builder(new ClusterName("_name")) .metaData(MetaData.builder() - .putCustom(MlMetadata.TYPE, mlMetadata.build())) + .putCustom(PersistentTasksCustomMetaData.TYPE, tasksBuilder.build())) .build(); when(clusterService.state()).thenReturn(clusterState); - JobManager jobManager = createJobManager(); + MockClientBuilder mockClientBuilder = new MockClientBuilder("cluster-test"); + mockClientBuilder.prepareSearch(AnomalyDetectorsIndex.configIndexName(), docsAsBytes); + JobManager jobManager = createJobManager(mockClientBuilder.build()); MlFilter filter = MlFilter.builder("foo_filter").build(); - jobManager.notifyFilterChanged(filter, new TreeSet<>(Arrays.asList("a", "b")), Collections.emptySet()); + jobManager.notifyFilterChanged(filter, new TreeSet<>(Arrays.asList("a", "b")), Collections.emptySet(), + ActionListener.wrap( + r -> {}, + e -> fail(e.getMessage()) + )); verify(auditor).info(jobReferencingFilter.getId(), "Filter [foo_filter] has been modified; added items: ['a', 'b']"); Mockito.verifyNoMoreInteractions(auditor, updateJobProcessNotifier); } - public void testNotifyFilterChangedGivenOnlyRemovedItems() { + @AwaitsFix(bugUrl = "Closed jobs are not audited when the filter changes") + public void testNotifyFilterChangedGivenOnlyRemovedItems() throws IOException { Detector.Builder detectorReferencingFilter = new Detector.Builder("count", null); detectorReferencingFilter.setByFieldName("foo"); DetectionRule filterRule = new DetectionRule.Builder(RuleScope.builder().exclude("foo", "foo_filter")).build(); @@ -290,37 +414,42 @@ public void testNotifyFilterChangedGivenOnlyRemovedItems() { Job.Builder jobReferencingFilter = buildJobBuilder("job-referencing-filter"); jobReferencingFilter.setAnalysisConfig(filterAnalysisConfig); + List docsAsBytes = Collections.singletonList(toBytesReference(jobReferencingFilter.build())); - MlMetadata.Builder mlMetadata = new MlMetadata.Builder(); - mlMetadata.putJob(jobReferencingFilter.build(), false); - + PersistentTasksCustomMetaData.Builder tasksBuilder = PersistentTasksCustomMetaData.builder(); ClusterState clusterState = ClusterState.builder(new ClusterName("_name")) .metaData(MetaData.builder() - .putCustom(MlMetadata.TYPE, mlMetadata.build())) + .putCustom(PersistentTasksCustomMetaData.TYPE, tasksBuilder.build())) .build(); when(clusterService.state()).thenReturn(clusterState); + when(clusterService.state()).thenReturn(clusterState); - JobManager jobManager = createJobManager(); + MockClientBuilder mockClientBuilder = new MockClientBuilder("cluster-test"); + mockClientBuilder.prepareSearch(AnomalyDetectorsIndex.configIndexName(), docsAsBytes); + JobManager jobManager = createJobManager(mockClientBuilder.build()); MlFilter filter = MlFilter.builder("foo_filter").build(); - jobManager.notifyFilterChanged(filter, Collections.emptySet(), new TreeSet<>(Arrays.asList("a", "b"))); + jobManager.notifyFilterChanged(filter, Collections.emptySet(), new TreeSet<>(Arrays.asList("a", "b")), + ActionListener.wrap( + r -> {}, + e -> fail(e.getMessage()) + )); verify(auditor).info(jobReferencingFilter.getId(), "Filter [foo_filter] has been modified; removed items: ['a', 'b']"); Mockito.verifyNoMoreInteractions(auditor, updateJobProcessNotifier); } - public void testUpdateProcessOnCalendarChanged() { + public void testUpdateProcessOnCalendarChanged() throws IOException { + List docsAsBytes = new ArrayList<>(); Job.Builder job1 = buildJobBuilder("job-1"); + docsAsBytes.add(toBytesReference(job1.build())); Job.Builder job2 = buildJobBuilder("job-2"); +// docsAsBytes.add(toBytesReference(job2.build())); Job.Builder job3 = buildJobBuilder("job-3"); + docsAsBytes.add(toBytesReference(job3.build())); Job.Builder job4 = buildJobBuilder("job-4"); - - MlMetadata.Builder mlMetadata = new MlMetadata.Builder(); - mlMetadata.putJob(job1.build(), false); - mlMetadata.putJob(job2.build(), false); - mlMetadata.putJob(job3.build(), false); - mlMetadata.putJob(job4.build(), false); + docsAsBytes.add(toBytesReference(job4.build())); PersistentTasksCustomMetaData.Builder tasksBuilder = PersistentTasksCustomMetaData.builder(); addJobTask(job1.getId(), "node_id", JobState.OPENED, tasksBuilder); @@ -329,14 +458,19 @@ public void testUpdateProcessOnCalendarChanged() { ClusterState clusterState = ClusterState.builder(new ClusterName("_name")) .metaData(MetaData.builder() - .putCustom(PersistentTasksCustomMetaData.TYPE, tasksBuilder.build()) - .putCustom(MlMetadata.TYPE, mlMetadata.build())) + .putCustom(PersistentTasksCustomMetaData.TYPE, tasksBuilder.build())) .build(); when(clusterService.state()).thenReturn(clusterState); - JobManager jobManager = createJobManager(); + MockClientBuilder mockClientBuilder = new MockClientBuilder("cluster-test"); + mockClientBuilder.prepareSearch(AnomalyDetectorsIndex.configIndexName(), docsAsBytes); + JobManager jobManager = createJobManager(mockClientBuilder.build()); - jobManager.updateProcessOnCalendarChanged(Arrays.asList("job-1", "job-3", "job-4")); + jobManager.updateProcessOnCalendarChanged(Arrays.asList("job-1", "job-3", "job-4"), + ActionListener.wrap( + r -> {}, + e -> fail(e.getMessage()) + )); ArgumentCaptor updateParamsCaptor = ArgumentCaptor.forClass(UpdateParams.class); verify(updateJobProcessNotifier, times(2)).submitJobUpdate(updateParamsCaptor.capture(), any(ActionListener.class)); @@ -349,17 +483,17 @@ public void testUpdateProcessOnCalendarChanged() { assertThat(capturedUpdateParams.get(1).isUpdateScheduledEvents(), is(true)); } - public void testUpdateProcessOnCalendarChanged_GivenGroups() { + public void testUpdateProcessOnCalendarChanged_GivenGroups() throws IOException { Job.Builder job1 = buildJobBuilder("job-1"); job1.setGroups(Collections.singletonList("group-1")); Job.Builder job2 = buildJobBuilder("job-2"); job2.setGroups(Collections.singletonList("group-1")); Job.Builder job3 = buildJobBuilder("job-3"); - MlMetadata.Builder mlMetadata = new MlMetadata.Builder(); - mlMetadata.putJob(job1.build(), false); - mlMetadata.putJob(job2.build(), false); - mlMetadata.putJob(job3.build(), false); + List docsAsBytes = new ArrayList<>(); + docsAsBytes.add(toBytesReference(job1.build())); + docsAsBytes.add(toBytesReference(job2.build())); +// docsAsBytes.add(toBytesReference(job3.build())); PersistentTasksCustomMetaData.Builder tasksBuilder = PersistentTasksCustomMetaData.builder(); addJobTask(job1.getId(), "node_id", JobState.OPENED, tasksBuilder); @@ -368,14 +502,19 @@ public void testUpdateProcessOnCalendarChanged_GivenGroups() { ClusterState clusterState = ClusterState.builder(new ClusterName("_name")) .metaData(MetaData.builder() - .putCustom(PersistentTasksCustomMetaData.TYPE, tasksBuilder.build()) - .putCustom(MlMetadata.TYPE, mlMetadata.build())) + .putCustom(PersistentTasksCustomMetaData.TYPE, tasksBuilder.build())) .build(); when(clusterService.state()).thenReturn(clusterState); - JobManager jobManager = createJobManager(); + MockClientBuilder mockClientBuilder = new MockClientBuilder("cluster-test"); + mockClientBuilder.prepareSearch(AnomalyDetectorsIndex.configIndexName(), docsAsBytes); + JobManager jobManager = createJobManager(mockClientBuilder.build()); - jobManager.updateProcessOnCalendarChanged(Collections.singletonList("group-1")); + jobManager.updateProcessOnCalendarChanged(Collections.singletonList("group-1"), + ActionListener.wrap( + r -> {}, + e -> fail(e.getMessage()) + )); ArgumentCaptor updateParamsCaptor = ArgumentCaptor.forClass(UpdateParams.class); verify(updateJobProcessNotifier, times(2)).submitJobUpdate(updateParamsCaptor.capture(), any(ActionListener.class)); @@ -400,12 +539,12 @@ private Job.Builder createJob() { return builder; } - private JobManager createJobManager() { + private JobManager createJobManager(Client client) { ClusterSettings clusterSettings = new ClusterSettings(environment.settings(), Collections.singleton(MachineLearningField.MAX_MODEL_MEMORY_LIMIT)); when(clusterService.getClusterSettings()).thenReturn(clusterSettings); return new JobManager(environment, environment.settings(), jobResultsProvider, clusterService, - auditor, client, updateJobProcessNotifier); + auditor, threadPool, client, updateJobProcessNotifier); } private ClusterState createClusterState() { @@ -413,4 +552,11 @@ private ClusterState createClusterState() { builder.metaData(MetaData.builder()); return builder.build(); } + + private BytesReference toBytesReference(ToXContent content) throws IOException { + try (XContentBuilder xContentBuilder = XContentFactory.jsonBuilder()) { + content.toXContent(xContentBuilder, ToXContent.EMPTY_PARAMS); + return BytesReference.bytes(xContentBuilder); + } + } } diff --git a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/persistence/MockClientBuilder.java b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/persistence/MockClientBuilder.java index 61fc73d1641e1..628a46387e7e2 100644 --- a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/persistence/MockClientBuilder.java +++ b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/persistence/MockClientBuilder.java @@ -33,6 +33,7 @@ import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.index.IndexRequestBuilder; import org.elasticsearch.action.index.IndexResponse; +import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchRequestBuilder; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.search.SearchScrollRequestBuilder; @@ -42,11 +43,14 @@ import org.elasticsearch.client.Client; import org.elasticsearch.client.ClusterAdminClient; import org.elasticsearch.client.IndicesAdminClient; +import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.index.query.QueryBuilder; +import org.elasticsearch.search.SearchHit; +import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.sort.SortBuilder; import org.elasticsearch.search.sort.SortOrder; import org.elasticsearch.threadpool.ThreadPool; @@ -56,6 +60,7 @@ import org.mockito.stubbing.Answer; import java.io.IOException; +import java.util.List; import java.util.concurrent.ExecutionException; import static org.junit.Assert.assertArrayEquals; @@ -168,6 +173,19 @@ public MockClientBuilder prepareGet(String index, String type, String id, GetRes return this; } + public MockClientBuilder get(GetResponse response) { + doAnswer(new Answer() { + @Override + public Void answer(InvocationOnMock invocationOnMock) { + ActionListener listener = (ActionListener) invocationOnMock.getArguments()[1]; + listener.onResponse(response); + return null; + } + }).when(client).get(any(), any()); + + return this; + } + public MockClientBuilder prepareCreate(String index) { CreateIndexRequestBuilder createIndexRequestBuilder = mock(CreateIndexRequestBuilder.class); CreateIndexResponse response = mock(CreateIndexResponse.class); @@ -254,6 +272,46 @@ public MockClientBuilder prepareSearch(String index, String type, int from, int return this; } + /** + * Creates a {@link SearchResponse} with a {@link SearchHit} for each element of {@code docs} + * @param indexName Index being searched + * @param docs Returned in the SearchResponse + * @return + */ + @SuppressWarnings("unchecked") + public MockClientBuilder prepareSearch(String indexName, List docs) { + SearchRequestBuilder builder = mock(SearchRequestBuilder.class); + when(builder.setIndicesOptions(any())).thenReturn(builder); + when(builder.setQuery(any())).thenReturn(builder); + when(builder.setSource(any())).thenReturn(builder); + SearchRequest request = new SearchRequest(indexName); + when(builder.request()).thenReturn(request); + + when(client.prepareSearch(eq(indexName))).thenReturn(builder); + + SearchHit hits [] = new SearchHit[docs.size()]; + for (int i=0; i() { + @Override + public Void answer(InvocationOnMock invocationOnMock) { + ActionListener listener = (ActionListener) invocationOnMock.getArguments()[1]; + listener.onResponse(response); + return null; + } + }).when(client).search(eq(request), any()); + + return this; + } + public MockClientBuilder prepareSearchAnySize(String index, String type, SearchResponse response, ArgumentCaptor filter) { SearchRequestBuilder builder = mock(SearchRequestBuilder.class); when(builder.setTypes(eq(type))).thenReturn(builder); diff --git a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/process/autodetect/AutodetectProcessManagerTests.java b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/process/autodetect/AutodetectProcessManagerTests.java index 7ef329d446901..eb14d3ccacf61 100644 --- a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/process/autodetect/AutodetectProcessManagerTests.java +++ b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/process/autodetect/AutodetectProcessManagerTests.java @@ -126,7 +126,15 @@ public void setup() throws Exception { normalizerFactory = mock(NormalizerFactory.class); auditor = mock(Auditor.class); - when(jobManager.getJobOrThrowIfUnknown("foo")).thenReturn(createJobDetails("foo")); + + doAnswer(invocationOnMock -> { + @SuppressWarnings("unchecked") + ActionListener listener = (ActionListener) invocationOnMock.getArguments()[1]; + listener.onResponse(createJobDetails("foo")); + return null; + }).when(jobManager).getJob(eq("foo"), any()); + + doAnswer(invocationOnMock -> { @SuppressWarnings("unchecked") Consumer handler = (Consumer) invocationOnMock.getArguments()[1]; @@ -166,6 +174,27 @@ public void testMaxOpenJobsSetting_givenOldAndNewSettings() { + "See the breaking changes documentation for the next major version."); } + public void testOpenJob() { + Client client = mock(Client.class); + AutodetectCommunicator communicator = mock(AutodetectCommunicator.class); + doAnswer(invocationOnMock -> { + @SuppressWarnings("unchecked") + ActionListener listener = (ActionListener) invocationOnMock.getArguments()[1]; + listener.onResponse(createJobDetails("foo")); + return null; + }).when(jobManager).getJob(eq("foo"), any()); + AutodetectProcessManager manager = createManager(communicator, client); + + JobTask jobTask = mock(JobTask.class); + when(jobTask.getJobId()).thenReturn("foo"); + when(jobTask.getAllocationId()).thenReturn(1L); + manager.openJob(jobTask, e -> {}); + assertEquals(1, manager.numberOfOpenJobs()); + assertTrue(manager.jobHasActiveAutodetectProcess(jobTask)); + verify(jobTask).updatePersistentTaskState(eq(new JobTaskState(JobState.OPENED, 1L)), any()); + } + + public void testOpenJob_withoutVersion() { Client client = mock(Client.class); AutodetectCommunicator communicator = mock(AutodetectCommunicator.class); @@ -174,40 +203,32 @@ public void testOpenJob_withoutVersion() { Job job = jobBuilder.build(); assertThat(job.getJobVersion(), is(nullValue())); - when(jobManager.getJobOrThrowIfUnknown(job.getId())).thenReturn(job); - AutodetectProcessManager manager = createManager(communicator, client); + doAnswer(invocationOnMock -> { + @SuppressWarnings("unchecked") + ActionListener listener = (ActionListener) invocationOnMock.getArguments()[1]; + listener.onResponse(job); + return null; + }).when(jobManager).getJob(eq(job.getId()), any()); + AutodetectProcessManager manager = createManager(communicator, client); JobTask jobTask = mock(JobTask.class); when(jobTask.getJobId()).thenReturn(job.getId()); - AtomicReference errorHolder = new AtomicReference<>(); manager.openJob(jobTask, errorHolder::set); - Exception error = errorHolder.get(); assertThat(error, is(notNullValue())); assertThat(error.getMessage(), equalTo("Cannot open job [no_version] because jobs created prior to version 5.5 are not supported")); } - public void testOpenJob() { - Client client = mock(Client.class); - AutodetectCommunicator communicator = mock(AutodetectCommunicator.class); - when(jobManager.getJobOrThrowIfUnknown("foo")).thenReturn(createJobDetails("foo")); - AutodetectProcessManager manager = createManager(communicator, client); - - JobTask jobTask = mock(JobTask.class); - when(jobTask.getJobId()).thenReturn("foo"); - when(jobTask.getAllocationId()).thenReturn(1L); - manager.openJob(jobTask, e -> {}); - assertEquals(1, manager.numberOfOpenJobs()); - assertTrue(manager.jobHasActiveAutodetectProcess(jobTask)); - verify(jobTask).updatePersistentTaskState(eq(new JobTaskState(JobState.OPENED, 1L)), any()); - } - public void testOpenJob_exceedMaxNumJobs() { - when(jobManager.getJobOrThrowIfUnknown("foo")).thenReturn(createJobDetails("foo")); - when(jobManager.getJobOrThrowIfUnknown("bar")).thenReturn(createJobDetails("bar")); - when(jobManager.getJobOrThrowIfUnknown("baz")).thenReturn(createJobDetails("baz")); - when(jobManager.getJobOrThrowIfUnknown("foobar")).thenReturn(createJobDetails("foobar")); + for (String jobId : new String [] {"foo", "bar", "baz", "foobar"}) { + doAnswer(invocationOnMock -> { + @SuppressWarnings("unchecked") + ActionListener listener = (ActionListener) invocationOnMock.getArguments()[1]; + listener.onResponse(createJobDetails(jobId)); + return null; + }).when(jobManager).getJob(eq(jobId), any()); + } Client client = mock(Client.class); ThreadPool threadPool = mock(ThreadPool.class); @@ -577,7 +598,14 @@ public void testCreate_notEnoughThreads() throws IOException { doThrow(new EsRejectedExecutionException("")).when(executorService).submit(any(Runnable.class)); when(threadPool.executor(anyString())).thenReturn(executorService); when(threadPool.scheduleWithFixedDelay(any(), any(), any())).thenReturn(mock(ThreadPool.Cancellable.class)); - when(jobManager.getJobOrThrowIfUnknown("my_id")).thenReturn(createJobDetails("my_id")); + Job job = createJobDetails("my_id"); + doAnswer(invocationOnMock -> { + @SuppressWarnings("unchecked") + ActionListener listener = (ActionListener) invocationOnMock.getArguments()[1]; + listener.onResponse(job); + return null; + }).when(jobManager).getJob(eq("my_id"), any()); + AutodetectProcess autodetectProcess = mock(AutodetectProcess.class); AutodetectProcessFactory autodetectProcessFactory = (j, autodetectParams, e, onProcessCrash) -> autodetectProcess; @@ -588,7 +616,7 @@ public void testCreate_notEnoughThreads() throws IOException { JobTask jobTask = mock(JobTask.class); when(jobTask.getJobId()).thenReturn("my_id"); expectThrows(EsRejectedExecutionException.class, - () -> manager.create(jobTask, buildAutodetectParams(), e -> {})); + () -> manager.create(jobTask, job, buildAutodetectParams(), e -> {})); verify(autodetectProcess, times(1)).close(); } @@ -598,7 +626,7 @@ public void testCreate_givenFirstTime() { JobTask jobTask = mock(JobTask.class); when(jobTask.getJobId()).thenReturn("foo"); - manager.create(jobTask, buildAutodetectParams(), e -> {}); + manager.create(jobTask, createJobDetails("foo"), buildAutodetectParams(), e -> {}); String expectedNotification = "Loading model snapshot [N/A], job latest_record_timestamp [N/A]"; verify(auditor).info("foo", expectedNotification); @@ -614,7 +642,7 @@ public void testCreate_givenExistingModelSnapshot() { JobTask jobTask = mock(JobTask.class); when(jobTask.getJobId()).thenReturn("foo"); - manager.create(jobTask, buildAutodetectParams(), e -> {}); + manager.create(jobTask, createJobDetails("foo"), buildAutodetectParams(), e -> {}); String expectedNotification = "Loading model snapshot [snapshot-1] with " + "latest_record_timestamp [1970-01-01T00:00:00.000Z], " + @@ -633,7 +661,7 @@ public void testCreate_givenNonZeroCountsAndNoModelSnapshotNorQuantiles() { JobTask jobTask = mock(JobTask.class); when(jobTask.getJobId()).thenReturn("foo"); - manager.create(jobTask, buildAutodetectParams(), e -> {}); + manager.create(jobTask, createJobDetails("foo"), buildAutodetectParams(), e -> {}); String expectedNotification = "Loading model snapshot [N/A], " + "job latest_record_timestamp [1970-01-01T00:00:00.000Z]"; @@ -650,7 +678,13 @@ private AutodetectProcessManager createNonSpyManager(String jobId) { ExecutorService executorService = mock(ExecutorService.class); when(threadPool.executor(anyString())).thenReturn(executorService); when(threadPool.scheduleWithFixedDelay(any(), any(), any())).thenReturn(mock(ThreadPool.Cancellable.class)); - when(jobManager.getJobOrThrowIfUnknown(jobId)).thenReturn(createJobDetails(jobId)); + doAnswer(invocationOnMock -> { + @SuppressWarnings("unchecked") + ActionListener listener = (ActionListener) invocationOnMock.getArguments()[1]; + listener.onResponse(createJobDetails(jobId)); + return null; + }).when(jobManager).getJob(eq(jobId), any()); + AutodetectProcess autodetectProcess = mock(AutodetectProcess.class); AutodetectProcessFactory autodetectProcessFactory = (j, autodetectParams, e, onProcessCrash) -> autodetectProcess; @@ -684,7 +718,7 @@ private AutodetectProcessManager createManager(AutodetectCommunicator communicat autodetectProcessFactory, normalizerFactory, new NamedXContentRegistry(Collections.emptyList()), auditor); manager = spy(manager); - doReturn(communicator).when(manager).create(any(), eq(buildAutodetectParams()), any()); + doReturn(communicator).when(manager).create(any(), any(), eq(buildAutodetectParams()), any()); return manager; } From fdf1405f6b507688138249b862bd786576cbef83 Mon Sep 17 00:00:00 2001 From: Dimitris Athanasiou Date: Fri, 24 Aug 2018 17:38:09 +0100 Subject: [PATCH 2/5] Addresses the filter update issue by finding all jobs with custom rules --- .../xpack/ml/job/JobManager.java | 30 +++++++------- .../ml/job/persistence/JobConfigProvider.java | 39 +++++++++++++++++++ 2 files changed, 52 insertions(+), 17 deletions(-) diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java index 8194e226f8f19..a6f2db0e8eb2e 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java @@ -52,8 +52,8 @@ import org.elasticsearch.xpack.ml.MachineLearning; import org.elasticsearch.xpack.ml.job.categorization.CategorizationAnalyzer; import org.elasticsearch.xpack.ml.job.persistence.JobConfigProvider; -import org.elasticsearch.xpack.ml.job.persistence.JobResultsProvider; import org.elasticsearch.xpack.ml.job.persistence.JobResultsPersister; +import org.elasticsearch.xpack.ml.job.persistence.JobResultsProvider; import org.elasticsearch.xpack.ml.job.process.autodetect.UpdateParams; import org.elasticsearch.xpack.ml.notifications.Auditor; import org.elasticsearch.xpack.ml.utils.ChainTaskExecutor; @@ -419,26 +419,22 @@ public void notifyFilterChanged(MlFilter filter, Set addedItems, Set openJobIds = openJobIds(clusterService.state()); - if (openJobIds.isEmpty()) { - updatedListener.onResponse(Boolean.TRUE); - return; - } - - String jobsExpression = Strings.collectionToCommaDelimitedString(openJobIds); - - // TODO JIndex probably better to query for the filter ID - jobConfigProvider.expandJobs(jobsExpression, false, ActionListener.wrap( + jobConfigProvider.findJobsWithCustomRules(ActionListener.wrap( jobBuilders -> { threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME).execute(() -> { - for (Job.Builder builder: jobBuilders) { - // TODO JIndex - sneaky because the job isn't being built here - Set jobFilters = builder.getAnalysisConfig().extractReferencedFilters(); + for (Job job: jobBuilders) { + Set jobFilters = job.getAnalysisConfig().extractReferencedFilters(); + ClusterState clusterState = clusterService.state(); if (jobFilters.contains(filter.getId())) { - updateJobProcessNotifier.submitJobUpdate(UpdateParams.filterUpdate(builder.getId(), filter), + if (isJobOpen(clusterState, job.getId())) { + updateJobProcessNotifier.submitJobUpdate(UpdateParams.filterUpdate(job.getId(), filter), ActionListener.wrap(isUpdated -> { - auditFilterChanges(builder.getId(), filter.getId(), addedItems, removedItems); - }, e -> {})); + auditFilterChanges(job.getId(), filter.getId(), addedItems, removedItems); + }, e -> { + })); + } else { + auditFilterChanges(job.getId(), filter.getId(), addedItems, removedItems); + } } } diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java index e7cd47b4e8414..a9d12968b8655 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java @@ -5,6 +5,7 @@ */ package org.elasticsearch.xpack.ml.job.persistence; +import org.apache.lucene.search.join.ScoreMode; import org.elasticsearch.ElasticsearchParseException; import org.elasticsearch.action.ActionListener; import org.elasticsearch.action.DocWriteRequest; @@ -37,12 +38,15 @@ import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.query.QueryBuilder; +import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.index.query.TermQueryBuilder; import org.elasticsearch.index.query.TermsQueryBuilder; import org.elasticsearch.index.query.WildcardQueryBuilder; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.elasticsearch.search.fetch.subphase.FetchSourceContext; +import org.elasticsearch.xpack.core.ml.job.config.AnalysisConfig; +import org.elasticsearch.xpack.core.ml.job.config.Detector; import org.elasticsearch.xpack.core.ml.job.config.Job; import org.elasticsearch.xpack.core.ml.job.config.JobUpdate; import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex; @@ -52,6 +56,7 @@ import java.io.IOException; import java.io.InputStream; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashSet; import java.util.List; import java.util.Set; @@ -489,6 +494,40 @@ public void expandGroupIds(List groupIds, ActionListener> l , client::search); } + public void findJobsWithCustomRules(ActionListener> listener) { + String customRulesPath = Strings.collectionToDelimitedString(Arrays.asList(Job.ANALYSIS_CONFIG.getPreferredName(), + AnalysisConfig.DETECTORS.getPreferredName(), Detector.CUSTOM_RULES_FIELD.getPreferredName()), "."); + SearchSourceBuilder sourceBuilder = new SearchSourceBuilder() + .query(QueryBuilders.nestedQuery(customRulesPath, QueryBuilders.existsQuery(customRulesPath), ScoreMode.None)) + .size(10000); + + SearchRequest searchRequest = client.prepareSearch(AnomalyDetectorsIndex.configIndexName()) + .setIndicesOptions(IndicesOptions.lenientExpandOpen()) + .setSource(sourceBuilder).request(); + + executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, searchRequest, + ActionListener.wrap( + response -> { + List jobs = new ArrayList<>(); + + SearchHit[] hits = response.getHits().getHits(); + for (SearchHit hit : hits) { + try { + BytesReference source = hit.getSourceRef(); + Job job = parseJobLenientlyFromSource(source).build(); + jobs.add(job); + } catch (IOException e) { + // TODO A better way to handle this rather than just ignoring the error? + logger.error("Error parsing anomaly detector job configuration [" + hit.getId() + "]", e); + } + } + + listener.onResponse(jobs); + }, + listener::onFailure) + , client::search); + } + private void parseJobLenientlyFromSource(BytesReference source, ActionListener jobListener) { try (InputStream stream = source.streamInput(); XContentParser parser = XContentFactory.xContent(XContentType.JSON) From 291829a7592722c6e4b1ebd9a95277f277ede3dd Mon Sep 17 00:00:00 2001 From: David Kyle Date: Tue, 28 Aug 2018 12:53:35 +0100 Subject: [PATCH 3/5] Address review comments. Rename jobExist(s) function, reformat code, use doc values in job Id search and sort jobs by Id. --- .../ml/action/TransportForecastJobAction.java | 49 ++++++++++--------- .../TransportRevertModelSnapshotAction.java | 2 +- .../xpack/ml/job/JobManager.java | 20 ++++---- .../ml/job/persistence/JobConfigProvider.java | 43 +++++++++++----- .../ml/integration/JobConfigProviderIT.java | 5 +- .../xpack/ml/job/JobManagerTests.java | 5 +- 6 files changed, 72 insertions(+), 52 deletions(-) diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportForecastJobAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportForecastJobAction.java index 7a929c000c02f..c927346da9d74 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportForecastJobAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportForecastJobAction.java @@ -89,29 +89,7 @@ protected void taskOperation(ForecastJobAction.Request request, TransportOpenJob ForecastParams params = paramsBuilder.build(); processManager.forecastJob(task, params, e -> { if (e == null) { - Consumer forecastRequestStatsHandler = forecastRequestStats -> { - if (forecastRequestStats == null) { - // paranoia case, it should not happen that we do not retrieve a result - listener.onFailure(new ElasticsearchException( - "Cannot run forecast: internal error, please check the logs")); - } else if (forecastRequestStats.getStatus() == ForecastRequestStats.ForecastRequestStatus.FAILED) { - List messages = forecastRequestStats.getMessages(); - if (messages.size() > 0) { - listener.onFailure(ExceptionsHelper.badRequestException("Cannot run forecast: " - + messages.get(0))); - } else { - // paranoia case, it should not be possible to have an empty message list - listener.onFailure( - new ElasticsearchException( - "Cannot run forecast: internal error, please check the logs")); - } - } else { - listener.onResponse(new ForecastJobAction.Response(true, params.getForecastId())); - } - }; - - jobResultsProvider.getForecastRequestStats(request.getJobId(), params.getForecastId(), - forecastRequestStatsHandler, listener::onFailure); +; getForecastRequestStats(request.getJobId(), params.getForecastId(), listener); } else { listener.onFailure(e); } @@ -121,6 +99,31 @@ protected void taskOperation(ForecastJobAction.Request request, TransportOpenJob )); } + private void getForecastRequestStats(String jobId, String forecastId, ActionListener listener) { + Consumer forecastRequestStatsHandler = forecastRequestStats -> { + if (forecastRequestStats == null) { + // paranoia case, it should not happen that we do not retrieve a result + listener.onFailure(new ElasticsearchException( + "Cannot run forecast: internal error, please check the logs")); + } else if (forecastRequestStats.getStatus() == ForecastRequestStats.ForecastRequestStatus.FAILED) { + List messages = forecastRequestStats.getMessages(); + if (messages.size() > 0) { + listener.onFailure(ExceptionsHelper.badRequestException("Cannot run forecast: " + + messages.get(0))); + } else { + // paranoia case, it should not be possible to have an empty message list + listener.onFailure( + new ElasticsearchException( + "Cannot run forecast: internal error, please check the logs")); + } + } else { + listener.onResponse(new ForecastJobAction.Response(true, forecastId)); + } + }; + + jobResultsProvider.getForecastRequestStats(jobId, forecastId, forecastRequestStatsHandler, listener::onFailure); + } + static void validate(Job job, ForecastJobAction.Request request) { if (job.getJobVersion() == null || job.getJobVersion().before(Version.V_6_1_0)) { throw ExceptionsHelper.badRequestException( diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportRevertModelSnapshotAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportRevertModelSnapshotAction.java index 83121043c4d07..5c84e39a6dd29 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportRevertModelSnapshotAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportRevertModelSnapshotAction.java @@ -71,7 +71,7 @@ protected void masterOperation(RevertModelSnapshotAction.Request request, Cluste logger.debug("Received request to revert to snapshot id '{}' for job '{}', deleting intervening results: {}", request.getSnapshotId(), request.getJobId(), request.getDeleteInterveningResults()); - jobManager.jobExist(request.getJobId(), ActionListener.wrap( + jobManager.jobExists(request.getJobId(), ActionListener.wrap( exists -> { PersistentTasksCustomMetaData tasks = state.getMetaData().custom(PersistentTasksCustomMetaData.TYPE); JobState jobState = MlTasks.getJobState(request.getJobId(), tasks); diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java index a6f2db0e8eb2e..970c38e62ab0e 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java @@ -60,6 +60,8 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; import java.util.Date; import java.util.HashMap; import java.util.List; @@ -120,7 +122,7 @@ private void setMaxModelMemoryLimit(ByteSizeValue maxModelMemoryLimit) { this.maxModelMemoryLimit = maxModelMemoryLimit; } - public void jobExist(String jobId, ActionListener listener) { + public void jobExists(String jobId, ActionListener listener) { jobConfigProvider.checkJobExists(jobId, listener); } @@ -146,7 +148,9 @@ public void getJob(String jobId, ActionListener jobListener) { } /** - * Even this this uses a listener the job is returned on the same thread + * Read a job from the cluster state. + * The job is returned on the same thread even though a listener is used. + * * @param jobId the jobId * @param jobListener the Job listener. If no job matches {@code jobId} * a ResourceNotFoundException is returned @@ -189,6 +193,7 @@ public void expandJobs(String expression, boolean allowNoJobs, ActionListener(jobs, jobs.size(), Job.RESULTS_FIELD)); }, jobsListener::onFailure @@ -252,14 +257,7 @@ public void onResponse(Boolean indicesCreated) { auditor.info(job.getId(), Messages.getMessage(Messages.JOB_AUDIT_CREATED)); actionListener.onResponse(new PutJobAction.Response(job)); }, - e -> { - if (e instanceof VersionConflictEngineException) { - // the job already exists - actionListener.onFailure(ExceptionsHelper.jobAlreadyExists(job.getId())); - } else { - actionListener.onFailure(e); - } - } + actionListener::onFailure )); } @@ -487,7 +485,7 @@ public void updateProcessOnCalendarChanged(List calendarJobIds, ActionLi expandedIds -> { threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME).execute(() -> { // Merge the expended group members with the request Ids. - // Ids that aren't jobs will be filtered by by isJobOpen() + // Ids that aren't jobs will be filtered by isJobOpen() expandedIds.addAll(calendarJobIds); for (String jobId : expandedIds) { diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java index a9d12968b8655..ae13d0371a3a5 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java @@ -36,6 +36,7 @@ import org.elasticsearch.common.xcontent.XContentFactory; import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentType; +import org.elasticsearch.index.engine.VersionConflictEngineException; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.query.QueryBuilder; import org.elasticsearch.index.query.QueryBuilders; @@ -60,6 +61,7 @@ import java.util.HashSet; import java.util.List; import java.util.Set; +import java.util.stream.Collectors; import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN; import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin; @@ -94,7 +96,16 @@ public void putJob(Job job, ActionListener listener) { .setOpType(DocWriteRequest.OpType.CREATE) .request(); - executeAsyncWithOrigin(client, ML_ORIGIN, IndexAction.INSTANCE, indexRequest, listener); + executeAsyncWithOrigin(client, ML_ORIGIN, IndexAction.INSTANCE, indexRequest, ActionListener.wrap( + listener::onResponse, + e -> { + if (e instanceof VersionConflictEngineException) { + // the job already exists + listener.onFailure(ExceptionsHelper.jobAlreadyExists(job.getId())); + } else { + listener.onFailure(e); + } + })); } catch (IOException e) { listener.onFailure(new ElasticsearchParseException("Failed to serialise job with id [" + job.getId() + "]", e)); @@ -338,10 +349,13 @@ public void onFailure(Exception e) { /** * Expands an expression into the set of matching names. {@code expresssion} - * may be a wildcard, a job group, a job ID or a list of those. + * may be a wildcard, a job group, a job Id or a list of those. * If {@code expression} == 'ALL', '*' or the empty string then all - * job IDs are returned. - * Job groups are expanded to all the jobs IDs in that group. + * job Ids are returned. + * Job groups are expanded to all the jobs Ids in that group. + * + * If {@code expression} contains a job Id or a Group name then it + * is an error if the job or group do not exist. * * For example, given a set of names ["foo-1", "foo-2", "bar-1", bar-2"], * expressions resolve follows: @@ -359,14 +373,15 @@ public void onFailure(Exception e) { * @param allowNoJobs if {@code false}, an error is thrown when no name matches the {@code expression}. * This only applies to wild card expressions, if {@code expression} is not a * wildcard then setting this true will not suppress the exception - * @param listener The expanded job IDs listener + * @param listener The expanded job Ids listener */ public void expandJobsIds(String expression, boolean allowNoJobs, ActionListener> listener) { String [] tokens = ExpandedIdsMatcher.tokenizeExpression(expression); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder().query(buildQuery(tokens)); sourceBuilder.sort(Job.ID.getPreferredName()); - String [] includes = new String[] {Job.ID.getPreferredName(), Job.GROUPS.getPreferredName()}; - sourceBuilder.fetchSource(includes, null); + sourceBuilder.fetchSource(false); + sourceBuilder.docValueField(Job.ID.getPreferredName()); + sourceBuilder.docValueField(Job.GROUPS.getPreferredName()); SearchRequest searchRequest = client.prepareSearch(AnomalyDetectorsIndex.configIndexName()) .setIndicesOptions(IndicesOptions.lenientExpandOpen()) @@ -381,10 +396,10 @@ public void expandJobsIds(String expression, boolean allowNoJobs, ActionListener Set groupsIds = new HashSet<>(); SearchHit[] hits = response.getHits().getHits(); for (SearchHit hit : hits) { - jobIds.add((String)hit.getSourceAsMap().get(Job.ID.getPreferredName())); - List groups = (List)hit.getSourceAsMap().get(Job.GROUPS.getPreferredName()); + jobIds.add(hit.field(Job.ID.getPreferredName()).getValue()); + List groups = hit.field(Job.GROUPS.getPreferredName()).getValues(); if (groups != null) { - groupsIds.addAll(groups); + groupsIds.addAll(groups.stream().map(Object::toString).collect(Collectors.toList())); } } @@ -463,6 +478,8 @@ public void expandJobs(String expression, boolean allowNoJobs, ActionListener
  • groupIds, ActionListener> l SearchSourceBuilder sourceBuilder = new SearchSourceBuilder() .query(new TermsQueryBuilder(Job.GROUPS.getPreferredName(), groupIds)); sourceBuilder.sort(Job.ID.getPreferredName()); - String [] includes = new String[] {Job.ID.getPreferredName()}; - sourceBuilder.fetchSource(includes, null); + sourceBuilder.fetchSource(false); + sourceBuilder.docValueField(Job.ID.getPreferredName()); SearchRequest searchRequest = client.prepareSearch(AnomalyDetectorsIndex.configIndexName()) .setIndicesOptions(IndicesOptions.lenientExpandOpen()) @@ -485,7 +502,7 @@ public void expandGroupIds(List groupIds, ActionListener> l Set jobIds = new HashSet<>(); SearchHit[] hits = response.getHits().getHits(); for (SearchHit hit : hits) { - jobIds.add((String)hit.getSourceAsMap().get(Job.ID.getPreferredName())); + jobIds.add(hit.field(Job.ID.getPreferredName()).getValue()); } listener.onResponse(jobIds); diff --git a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java index d478e09553c13..8dde7974deb5f 100644 --- a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java +++ b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java @@ -6,13 +6,13 @@ package org.elasticsearch.xpack.ml.integration; import org.elasticsearch.ElasticsearchStatusException; +import org.elasticsearch.ResourceAlreadyExistsException; import org.elasticsearch.ResourceNotFoundException; import org.elasticsearch.action.DocWriteResponse; import org.elasticsearch.action.delete.DeleteResponse; import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.unit.ByteSizeValue; -import org.elasticsearch.index.engine.VersionConflictEngineException; import org.elasticsearch.xpack.core.ml.job.config.AnalysisConfig; import org.elasticsearch.xpack.core.ml.job.config.DataDescription; import org.elasticsearch.xpack.core.ml.job.config.DetectionRule; @@ -101,7 +101,8 @@ public void testOverwriteNotAllowed() throws InterruptedException { blockingCall(actionListener -> jobConfigProvider.putJob(jobWithSameId, actionListener), indexResponseHolder, exceptionHolder); assertNull(indexResponseHolder.get()); assertNotNull(exceptionHolder.get()); - assertThat(exceptionHolder.get(), instanceOf(VersionConflictEngineException.class)); + assertThat(exceptionHolder.get(), instanceOf(ResourceAlreadyExistsException.class)); + assertEquals("The job cannot be created with the Id 'same-id'. The Id is already used.", exceptionHolder.get().getMessage()); } public void testCrud() throws InterruptedException { diff --git a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/JobManagerTests.java b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/JobManagerTests.java index 0361ba49a7ae7..98c406694ec5c 100644 --- a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/JobManagerTests.java +++ b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/JobManagerTests.java @@ -64,6 +64,7 @@ import static org.elasticsearch.xpack.core.ml.job.config.JobTests.buildJobBuilder; import static org.elasticsearch.xpack.ml.action.TransportOpenJobActionTests.addJobTask; +import static org.hamcrest.Matchers.contains; import static org.hamcrest.Matchers.containsInAnyOrder; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.greaterThanOrEqualTo; @@ -202,7 +203,7 @@ public void testExpandJobsFromClusterStateAndIndex() throws IOException { assertNotNull(jobsHolder.get()); assertThat(jobsHolder.get().results(), hasSize(4)); List jobIds = jobsHolder.get().results().stream().map(Job::getId).collect(Collectors.toList()); - assertThat(jobIds, containsInAnyOrder("foo-cs-1", "foo-cs-2", "bar-cs", "foo-index")); + assertThat(jobIds, contains("bar-cs", "foo-cs-1", "foo-cs-2", "foo-index")); jobsHolder.set(null); jobManager.expandJobs("foo*", true, ActionListener.wrap( @@ -213,7 +214,7 @@ public void testExpandJobsFromClusterStateAndIndex() throws IOException { assertNotNull(jobsHolder.get()); assertThat(jobsHolder.get().results(), hasSize(3)); jobIds = jobsHolder.get().results().stream().map(Job::getId).collect(Collectors.toList()); - assertThat(jobIds, containsInAnyOrder("foo-cs-1", "foo-cs-2", "foo-index")); + assertThat(jobIds, contains("foo-cs-1", "foo-cs-2", "foo-index")); } @SuppressWarnings("unchecked") From ebe9a072ada7ebde19f745d915c0202378ac731e Mon Sep 17 00:00:00 2001 From: Dimitris Athanasiou Date: Tue, 28 Aug 2018 15:27:07 +0100 Subject: [PATCH 4/5] Add integration test for findJobsWithCustomRules() --- .../ml/integration/JobConfigProviderIT.java | 39 ++++++++++++++++++- 1 file changed, 38 insertions(+), 1 deletion(-) diff --git a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java index 8dde7974deb5f..712206e37556a 100644 --- a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java +++ b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java @@ -19,6 +19,8 @@ import org.elasticsearch.xpack.core.ml.job.config.Detector; import org.elasticsearch.xpack.core.ml.job.config.Job; import org.elasticsearch.xpack.core.ml.job.config.JobUpdate; +import org.elasticsearch.xpack.core.ml.job.config.Operator; +import org.elasticsearch.xpack.core.ml.job.config.RuleCondition; import org.elasticsearch.xpack.core.ml.job.config.RuleScope; import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex; import org.elasticsearch.xpack.ml.MlSingleNodeTestCase; @@ -38,6 +40,7 @@ import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.empty; import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.is; import static org.hamcrest.core.IsInstanceOf.instanceOf; public class JobConfigProviderIT extends MlSingleNodeTestCase { @@ -389,7 +392,34 @@ public void testExpandGroups() throws Exception { assertThat(expandedIds, empty()); } - private Job.Builder createJob(String jobId, List groups) { + public void testFindJobsWithCustomRules_GivenNoJobs() throws Exception { + List foundJobs = blockingCall(listener -> jobConfigProvider.findJobsWithCustomRules(listener)); + assertThat(foundJobs.isEmpty(), is(true)); + } + + public void testFindJobsWithCustomRules() throws Exception { + putJob(createJob("job-without-rules", Collections.emptyList())); + + DetectionRule rule = new DetectionRule.Builder(Collections.singletonList( + new RuleCondition(RuleCondition.AppliesTo.ACTUAL, Operator.GT, 0.0))).build(); + + Job.Builder jobWithRules1 = createJob("job-with-rules-1", Collections.emptyList()); + jobWithRules1 = addCustomRule(jobWithRules1, rule); + putJob(jobWithRules1); + Job.Builder jobWithRules2 = createJob("job-with-rules-2", Collections.emptyList()); + jobWithRules2 = addCustomRule(jobWithRules2, rule); + putJob(jobWithRules2); + + client().admin().indices().prepareRefresh(AnomalyDetectorsIndex.configIndexName()).get(); + + List foundJobs = blockingCall(listener -> jobConfigProvider.findJobsWithCustomRules(listener)); + + Set foundJobIds = foundJobs.stream().map(Job::getId).collect(Collectors.toSet()); + assertThat(foundJobIds.size(), equalTo(2)); + assertThat(foundJobIds, containsInAnyOrder(jobWithRules1.getId(), jobWithRules2.getId())); + } + + private static Job.Builder createJob(String jobId, List groups) { Detector.Builder d1 = new Detector.Builder("info_content", "domain"); d1.setOverFieldName("client"); AnalysisConfig.Builder ac = new AnalysisConfig.Builder(Collections.singletonList(d1.build())); @@ -404,6 +434,13 @@ private Job.Builder createJob(String jobId, List groups) { return builder; } + private static Job.Builder addCustomRule(Job.Builder job, DetectionRule rule) { + JobUpdate.Builder update1 = new JobUpdate.Builder(job.getId()); + update1.setDetectorUpdates(Collections.singletonList(new JobUpdate.DetectorUpdate(0, null, Collections.singletonList(rule)))); + Job updatedJob = update1.build().mergeWithJob(job.build(new Date()), null); + return new Job.Builder(updatedJob); + } + private Job putJob(Job.Builder job) throws Exception { Job builtJob = job.build(new Date()); this.blockingCall(actionListener -> jobConfigProvider.putJob(builtJob, actionListener)); From 5d0f0f6596e589d7d1f01faa1c10aaa11942c37f Mon Sep 17 00:00:00 2001 From: Dimitris Athanasiou Date: Tue, 28 Aug 2018 17:12:42 +0100 Subject: [PATCH 5/5] Remove todo and unused import --- .../main/java/org/elasticsearch/xpack/ml/job/JobManager.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java index 970c38e62ab0e..fd0dc33138341 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java @@ -26,7 +26,6 @@ import org.elasticsearch.common.xcontent.XContentFactory; import org.elasticsearch.env.Environment; import org.elasticsearch.index.analysis.AnalysisRegistry; -import org.elasticsearch.index.engine.VersionConflictEngineException; import org.elasticsearch.persistent.PersistentTasksCustomMetaData; import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.xpack.core.ml.MachineLearningField; @@ -441,8 +440,6 @@ public void notifyFilterChanged(MlFilter filter, Set addedItems, Set addedItems, Set removedItems) {