Skip to content

Commit

Permalink
[ML] Return missing job error when .ml-config is does not exist (#34177)
Browse files Browse the repository at this point in the history
  • Loading branch information
@davidkyle
davidkyle committed Oct 1, 2018
1 parent 2ca9d11 commit 52b891e
Show file tree
Hide file tree
Showing 5 changed files with 87 additions and 28 deletions.
10 changes: 9 additions & 1 deletion ...src/main/java/org/elasticsearch/xpack/ml/datafeed/persistence/DatafeedConfigProvider.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.engine.VersionConflictEngineException;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
Expand Down Expand Up @@ -136,6 +137,9 @@ public void putDatafeedConfig(DatafeedConfig config, Map<String, String> headers
* If the datafeed document is missing a {@code ResourceNotFoundException}
* is returned via the listener.
*
* If the .ml-config index does not exist it is treated as a missing datafeed
* error.
*
* @param datafeedId The datafeed ID
* @param datafeedConfigListener The config listener
*/
Expand All @@ -154,7 +158,11 @@ public void onResponse(GetResponse getResponse) {
}
@Override
public void onFailure(Exception e) {
datafeedConfigListener.onFailure(e);
if (e.getClass() == IndexNotFoundException.class) {
datafeedConfigListener.onFailure(ExceptionsHelper.missingDatafeedException(datafeedId));
} else {
datafeedConfigListener.onFailure(e);
}
}
});
}
Expand Down
13 changes: 11 additions & 2 deletions x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -120,7 +120,7 @@ private void setMaxModelMemoryLimit(ByteSizeValue maxModelMemoryLimit) {
}

public void jobExists(String jobId, ActionListener<Boolean> listener) {
jobConfigProvider.checkJobExists(jobId, listener);
jobConfigProvider.jobExists(jobId, true, listener);
}

/**
Expand Down Expand Up @@ -281,7 +281,16 @@ public void onFailure(Exception e) {
actionListener::onFailure
);

jobResultsProvider.checkForLeftOverDocuments(job, checkForLeftOverDocs);
jobConfigProvider.jobExists(job.getId(), false, ActionListener.wrap(
jobExists -> {
if (jobExists) {
actionListener.onFailure(ExceptionsHelper.jobAlreadyExists(job.getId()));
} else {
jobResultsProvider.checkForLeftOverDocuments(job, checkForLeftOverDocs);
}
},
actionListener::onFailure
));
}

public void updateJob(UpdateJobAction.Request request, ActionListener<PutJobAction.Response> actionListener) {
Expand Down
41 changes: 33 additions & 8 deletions ...plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.engine.VersionConflictEngineException;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
Expand Down Expand Up @@ -122,6 +123,9 @@ public void putJob(Job job, ActionListener<IndexResponse> listener) {
* If the job is missing a {@code ResourceNotFoundException} is returned
* via the listener.
*
* If the .ml-config index does not exist it is treated as a missing job
* error.
*
* @param jobId The job ID
* @param jobListener Job listener
*/
Expand All @@ -143,7 +147,11 @@ public void onResponse(GetResponse getResponse) {

@Override
public void onFailure(Exception e) {
jobListener.onFailure(e);
if (e.getClass() == IndexNotFoundException.class) {
jobListener.onFailure(ExceptionsHelper.missingJobException(jobId));
} else {
jobListener.onFailure(e);
}
}
}, client::get);
}
Expand Down Expand Up @@ -368,14 +376,19 @@ private void indexUpdatedJob(Job updatedJob, long version, ActionListener<Job> u

/**
* Check a job exists. A job exists if it has a configuration document.
* If the .ml-config index does not exist it is treated as a missing job
* error.
*
* If the job does not exist a ResourceNotFoundException is returned to the listener,
* FALSE will never be returned only TRUE or ResourceNotFoundException
* Depending on the value of {@code errorIfMissing} if the job does not
* exist a ResourceNotFoundException is returned to the listener,
* otherwise false is returned in the response.
*
* @param jobId The jobId to check
* @param listener Exists listener
* @param jobId The jobId to check
* @param errorIfMissing If true and the job is missing the listener fails with
* a ResourceNotFoundException else false is returned.
* @param listener Exists listener
*/
public void checkJobExists(String jobId, ActionListener<Boolean> listener) {
public void jobExists(String jobId, boolean errorIfMissing, ActionListener<Boolean> listener) {
GetRequest getRequest = new GetRequest(AnomalyDetectorsIndex.configIndexName(),
ElasticsearchMappings.DOC_TYPE, Job.documentId(jobId));
getRequest.fetchSourceContext(FetchSourceContext.DO_NOT_FETCH_SOURCE);
Expand All @@ -384,15 +397,27 @@ public void checkJobExists(String jobId, ActionListener<Boolean> listener) {
@Override
public void onResponse(GetResponse getResponse) {
if (getResponse.isExists() == false) {
listener.onFailure(ExceptionsHelper.missingJobException(jobId));
if (errorIfMissing) {
listener.onFailure(ExceptionsHelper.missingJobException(jobId));
} else {
listener.onResponse(Boolean.FALSE);
}
} else {
listener.onResponse(Boolean.TRUE);
}
}

@Override
public void onFailure(Exception e) {
listener.onFailure(e);
if (e.getClass() == IndexNotFoundException.class) {
if (errorIfMissing) {
listener.onFailure(ExceptionsHelper.missingJobException(jobId));
} else {
listener.onResponse(Boolean.FALSE);
}
} else {
listener.onFailure(e);
}
}
});
}
Expand Down
17 changes: 13 additions & 4 deletions ...gin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/DatafeedConfigProviderIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,15 @@ public void testCrud() throws InterruptedException {
assertEquals(DocWriteResponse.Result.DELETED, deleteResponseHolder.get().getResult());
}

public void testGetDatafeedConfig_missing() throws InterruptedException {
AtomicReference<Exception> exceptionHolder = new AtomicReference<>();
AtomicReference<DatafeedConfig.Builder> configBuilderHolder = new AtomicReference<>();
blockingCall(actionListener -> datafeedConfigProvider.getDatafeedConfig("missing", actionListener),
configBuilderHolder, exceptionHolder);
assertNull(configBuilderHolder.get());
assertEquals(ResourceNotFoundException.class, exceptionHolder.get().getClass());
}

public void testMultipleCreateAndDeletes() throws InterruptedException {
String datafeedId = "df2";

Expand All @@ -127,7 +136,7 @@ public void testMultipleCreateAndDeletes() throws InterruptedException {
indexResponseHolder, exceptionHolder);
assertNull(indexResponseHolder.get());
assertThat(exceptionHolder.get(), instanceOf(ResourceAlreadyExistsException.class));
assertEquals("A datafeed with Id [df2] already exists", exceptionHolder.get().getMessage());
assertEquals("A datafeed with id [df2] already exists", exceptionHolder.get().getMessage());

// delete
exceptionHolder.set(null);
Expand All @@ -142,7 +151,7 @@ public void testMultipleCreateAndDeletes() throws InterruptedException {
blockingCall(actionListener -> datafeedConfigProvider.deleteDatafeedConfig(datafeedId, actionListener),
deleteResponseHolder, exceptionHolder);
assertNull(deleteResponseHolder.get());
assertThat(exceptionHolder.get(), instanceOf(ResourceNotFoundException.class));
assertEquals(ResourceNotFoundException.class, exceptionHolder.get().getClass());
}

public void testUpdateWhenApplyingTheUpdateThrows() throws Exception {
Expand Down Expand Up @@ -202,7 +211,7 @@ public void testAllowNoDatafeeds() throws InterruptedException {

assertNull(datafeedIdsHolder.get());
assertNotNull(exceptionHolder.get());
assertThat(exceptionHolder.get(), IsInstanceOf.instanceOf(ResourceNotFoundException.class));
assertEquals(ResourceNotFoundException.class, exceptionHolder.get().getClass());
assertThat(exceptionHolder.get().getMessage(), containsString("No datafeed with id [*] exists"));

exceptionHolder.set(null);
Expand All @@ -217,7 +226,7 @@ public void testAllowNoDatafeeds() throws InterruptedException {

assertNull(datafeedsHolder.get());
assertNotNull(exceptionHolder.get());
assertThat(exceptionHolder.get(), IsInstanceOf.instanceOf(ResourceNotFoundException.class));
assertEquals(ResourceNotFoundException.class, exceptionHolder.get().getClass());
assertThat(exceptionHolder.get().getMessage(), containsString("No datafeed with id [*] exists"));

exceptionHolder.set(null);
Expand Down
34 changes: 21 additions & 13 deletions ...k/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/JobConfigProviderIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -68,18 +68,25 @@ public void testGetMissingJob() throws InterruptedException {

assertNull(jobHolder.get());
assertNotNull(exceptionHolder.get());
assertThat(exceptionHolder.get(), instanceOf(ResourceNotFoundException.class));
assertEquals(ResourceNotFoundException.class, exceptionHolder.get().getClass());
}

public void testCheckJobExists() throws InterruptedException {
AtomicReference<Boolean> jobExistsHolder = new AtomicReference<>();
AtomicReference<Exception> exceptionHolder = new AtomicReference<>();

blockingCall(actionListener -> jobConfigProvider.checkJobExists("missing", actionListener), jobExistsHolder, exceptionHolder);

assertNull(jobExistsHolder.get());
assertNotNull(exceptionHolder.get());
assertThat(exceptionHolder.get(), instanceOf(ResourceNotFoundException.class));
boolean throwIfMissing = randomBoolean();
blockingCall(actionListener ->
jobConfigProvider.jobExists("missing", throwIfMissing, actionListener), jobExistsHolder, exceptionHolder);

if (throwIfMissing) {
assertNull(jobExistsHolder.get());
assertNotNull(exceptionHolder.get());
assertEquals(ResourceNotFoundException.class, exceptionHolder.get().getClass());
} else {
assertFalse(jobExistsHolder.get());
assertNull(exceptionHolder.get());
}

AtomicReference<IndexResponse> indexResponseHolder = new AtomicReference<>();

Expand All @@ -88,7 +95,8 @@ public void testCheckJobExists() throws InterruptedException {
blockingCall(actionListener -> jobConfigProvider.putJob(job, actionListener), indexResponseHolder, exceptionHolder);

exceptionHolder.set(null);
blockingCall(actionListener -> jobConfigProvider.checkJobExists("existing-job", actionListener), jobExistsHolder, exceptionHolder);
blockingCall(actionListener ->
jobConfigProvider.jobExists("existing-job", throwIfMissing, actionListener), jobExistsHolder, exceptionHolder);
assertNull(exceptionHolder.get());
assertNotNull(jobExistsHolder.get());
assertTrue(jobExistsHolder.get());
Expand Down Expand Up @@ -159,15 +167,15 @@ public void testCrud() throws InterruptedException {
getJobResponseHolder.set(null);
blockingCall(actionListener -> jobConfigProvider.getJob(jobId, actionListener), getJobResponseHolder, exceptionHolder);
assertNull(getJobResponseHolder.get());
assertThat(exceptionHolder.get(), instanceOf(ResourceNotFoundException.class));
assertEquals(ResourceNotFoundException.class, exceptionHolder.get().getClass());

// Delete deleted job
deleteJobResponseHolder.set(null);
exceptionHolder.set(null);
blockingCall(actionListener -> jobConfigProvider.deleteJob(jobId, actionListener),
deleteJobResponseHolder, exceptionHolder);
assertNull(deleteJobResponseHolder.get());
assertThat(exceptionHolder.get(), instanceOf(ResourceNotFoundException.class));
assertEquals(ResourceNotFoundException.class, exceptionHolder.get().getClass());
}

public void testGetJobs() throws Exception {
Expand Down Expand Up @@ -263,7 +271,7 @@ public void testAllowNoJobs() throws InterruptedException {

assertNull(jobIdsHolder.get());
assertNotNull(exceptionHolder.get());
assertThat(exceptionHolder.get(), instanceOf(ResourceNotFoundException.class));
assertEquals(ResourceNotFoundException.class, exceptionHolder.get().getClass());
assertThat(exceptionHolder.get().getMessage(), containsString("No known job with id"));

exceptionHolder.set(null);
Expand All @@ -278,7 +286,7 @@ public void testAllowNoJobs() throws InterruptedException {

assertNull(jobsHolder.get());
assertNotNull(exceptionHolder.get());
assertThat(exceptionHolder.get(), instanceOf(ResourceNotFoundException.class));
assertEquals(ResourceNotFoundException.class, exceptionHolder.get().getClass());
assertThat(exceptionHolder.get().getMessage(), containsString("No known job with id"));

exceptionHolder.set(null);
Expand Down Expand Up @@ -315,7 +323,7 @@ public void testExpandJobs_GroupsAndJobIds() throws Exception {
jobIdsHolder, exceptionHolder);
assertNull(jobIdsHolder.get());
assertNotNull(exceptionHolder.get());
assertThat(exceptionHolder.get(), instanceOf(ResourceNotFoundException.class));
assertEquals(ResourceNotFoundException.class, exceptionHolder.get().getClass());
assertThat(exceptionHolder.get().getMessage(), equalTo("No known job with id 'missing1,missing2'"));

// Job builders
Expand Down Expand Up @@ -344,7 +352,7 @@ public void testExpandJobs_GroupsAndJobIds() throws Exception {
jobsHolder, exceptionHolder);
assertNull(jobsHolder.get());
assertNotNull(exceptionHolder.get());
assertThat(exceptionHolder.get(), instanceOf(ResourceNotFoundException.class));
assertEquals(ResourceNotFoundException.class, exceptionHolder.get().getClass());
assertThat(exceptionHolder.get().getMessage(), equalTo("No known job with id 'missing1,missing2'"));
}

Expand Down

0 comments on commit 52b891e

Please sign in to comment.