[ML] Job config document CRUD operations #32738
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -51,9 +51,13 @@ | |
| import java.io.IOException; | ||
| import java.io.InputStream; | ||
| import java.util.ArrayList; | ||
| import java.util.Collection; | ||
| import java.util.HashSet; | ||
| import java.util.Iterator; | ||
| import java.util.LinkedList; | ||
| 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; | ||
|
|
@@ -232,7 +236,8 @@ 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. | ||
| * 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. | ||
| * | ||
| * For example, given a set of names ["foo-1", "foo-2", "bar-1", bar-2"], | ||
|
|
@@ -252,34 +257,43 @@ public void onFailure(Exception e) { | |
| * 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 | ||
| */ | ||
| public void expandJobsIds(String expression, boolean allowNoJobs, ActionListener<Set<String>> listener) { | ||
|
There was a problem hiding this comment. The behaviour of Given that the option isn't documented and the new behaviour matches the description in the API spec I can live with the subtle change There was a problem hiding this comment. The intention was that In the case of The intention was that job wildcard expansion would work like index wildcard expansion, and job groups would work like aliases, so that if anybody asked how they work we'd be able to give a simple answer and people who just assumed there would be consistency between different endpoints would be correct. There was a problem hiding this comment. The original behaviour could be replicated, one option is to make a query for each token in However, maybe this can still be done with a single query. Check the results and throw if a token is not a wildcard and no job with that Id was returned or if the token is a wildcard check for matching job IDs and throw depending on the value of There was a problem hiding this comment.
I think we'd have to set up a map of token to boolean before starting, record what we saw during the paging or scroll, then throw errors at the very end. There was a problem hiding this comment. I pushed a change to replicate the old |
||
| String [] tokens = 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); | ||
|
|
||
| SearchRequest searchRequest = client.prepareSearch(AnomalyDetectorsIndex.configIndexName()) | ||
| .setIndicesOptions(IndicesOptions.lenientExpandOpen()) | ||
| .setSource(sourceBuilder).request(); | ||
|
|
||
| LinkedList<IdMatcher> requiredMatches = requiredMatches(tokens, allowNoJobs); | ||
|
|
||
| executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, searchRequest, | ||
| ActionListener.<SearchResponse>wrap( | ||
| response -> { | ||
| Set<String> jobIds = new HashSet<>(); | ||
| Set<String> groupsIds = new HashSet<>(); | ||
| SearchHit[] hits = response.getHits().getHits(); | ||
| for (SearchHit hit : hits) { | ||
| jobIds.add((String)hit.getSourceAsMap().get(Job.ID.getPreferredName())); | ||
| List<String> groups = (List<String>)hit.getSourceAsMap().get(Job.GROUPS.getPreferredName()); | ||
| if (groups != null) { | ||
| groupsIds.addAll(groups); | ||
| } | ||
| } | ||
|
|
||
| groupsIds.addAll(jobIds); | ||
| filterMatchedIds(requiredMatches, groupsIds); | ||
| if (requiredMatches.isEmpty() == false) { | ||
| // some required jobs were not found | ||
| String missing = requiredMatches.stream().map(IdMatcher::getId).collect(Collectors.joining(",")); | ||
| listener.onFailure(ExceptionsHelper.missingJobException(missing)); | ||
| return; | ||
| } | ||
|
|
||
| listener.onResponse(jobIds); | ||
| }, | ||
| listener::onFailure) | ||
|
|
@@ -298,54 +312,76 @@ public void expandJobsIds(String expression, boolean allowNoJobs, ActionListener | |
| * 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 jobs listener | ||
| */ | ||
| // NORELEASE jobs should be paged or have a mechanism to return all jobs if there are many of them | ||
| public void expandJobs(String expression, boolean allowNoJobs, ActionListener<List<Job.Builder>> listener) { | ||
| String [] tokens = tokenizeExpression(expression); | ||
| SearchSourceBuilder sourceBuilder = new SearchSourceBuilder().query(buildQuery(tokens)); | ||
| sourceBuilder.sort(Job.ID.getPreferredName()); | ||
|
|
||
| SearchRequest searchRequest = client.prepareSearch(AnomalyDetectorsIndex.configIndexName()) | ||
| .setIndicesOptions(IndicesOptions.lenientExpandOpen()) | ||
| .setSource(sourceBuilder).request(); | ||
|
|
||
| LinkedList<IdMatcher> requiredMatches = requiredMatches(tokens, allowNoJobs); | ||
|
|
||
| executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, searchRequest, | ||
| ActionListener.<SearchResponse>wrap( | ||
|
There was a problem hiding this comment. nit: is the qualification of |
||
| response -> { | ||
| List<Job.Builder> jobs = new ArrayList<>(); | ||
| Set<String> jobAndGroupIds = new HashSet<>(); | ||
|
|
||
| SearchHit[] hits = response.getHits().getHits(); | ||
| for (SearchHit hit : hits) { | ||
| try { | ||
| BytesReference source = hit.getSourceRef(); | ||
| Job.Builder job = parseJobLenientlyFromSource(source); | ||
| jobs.add(job); | ||
| jobAndGroupIds.add(job.getId()); | ||
| jobAndGroupIds.addAll(job.getGroups()); | ||
| } 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); | ||
| } | ||
| } | ||
|
|
||
| filterMatchedIds(requiredMatches, jobAndGroupIds); | ||
| if (requiredMatches.isEmpty() == false) { | ||
| // some required jobs were not found | ||
| String missing = requiredMatches.stream().map(IdMatcher::getId).collect(Collectors.joining(",")); | ||
| listener.onFailure(ExceptionsHelper.missingJobException(missing)); | ||
| return; | ||
| } | ||
|
|
||
| listener.onResponse(jobs); | ||
| }, | ||
| listener::onFailure) | ||
| , client::search); | ||
|
|
||
| } | ||
|
|
||
| private void parseJobLenientlyFromSource(BytesReference source, ActionListener<Job.Builder> jobListener) { | ||
| try (InputStream stream = source.streamInput(); | ||
| XContentParser parser = XContentFactory.xContent(XContentType.JSON) | ||
| .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, stream)) { | ||
| jobListener.onResponse(Job.LENIENT_PARSER.apply(parser, null)); | ||
| } catch (Exception e) { | ||
| jobListener.onFailure(e); | ||
| } | ||
| } | ||
|
|
||
| private Job.Builder parseJobLenientlyFromSource(BytesReference source) throws IOException { | ||
| try (InputStream stream = source.streamInput(); | ||
| XContentParser parser = XContentFactory.xContent(XContentType.JSON) | ||
| .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, stream)) { | ||
| return Job.LENIENT_PARSER.apply(parser, null); | ||
| } | ||
| } | ||
|
|
||
| private QueryBuilder buildQuery(String [] tokens) { | ||
| QueryBuilder jobQuery = new TermQueryBuilder(Job.JOB_TYPE.getPreferredName(), Job.ANOMALY_DETECTOR_JOB_TYPE); | ||
| if (isWildcardAll(tokens)) { | ||
| // match all | ||
| return jobQuery; | ||
| } | ||
|
|
||
|
|
@@ -354,7 +390,6 @@ private QueryBuilder buildQuery(String expression) { | |
| BoolQueryBuilder shouldQueries = new BoolQueryBuilder(); | ||
|
|
||
| List<String> terms = new ArrayList<>(); | ||
| for (String token : tokens) { | ||
| if (Regex.isSimpleMatchPattern(token)) { | ||
| shouldQueries.should(new WildcardQueryBuilder(Job.ID.getPreferredName(), token)); | ||
|
|
@@ -376,21 +411,131 @@ private QueryBuilder buildQuery(String expression) { | |
| return boolQueryBuilder; | ||
| } | ||
|
|
||
| /** | ||
| * Does the {@code tokens} array resolves to a wildcard all expression. | ||
| * True if {@code tokens} is empty or if it contains a single element | ||
|
There was a problem hiding this comment. If the tokens contain a full wildcard and something else then that's still a full wildcard overall. E.g. |
||
| * equal to {@link #ALL}, '*' or an empty string | ||
| * | ||
| * @param tokens Expression tokens | ||
| * @return True if tokens resolves to a wildcard all expression | ||
| */ | ||
| static boolean isWildcardAll(String [] tokens) { | ||
| if (tokens.length == 0) { | ||
| return true; | ||
| } | ||
| return tokens.length == 1 && (ALL.equals(tokens[0]) || Regex.isMatchAllPattern(tokens[0]) || tokens[0].isEmpty()); | ||
| } | ||
|
|
||
| static String [] tokenizeExpression(String expression) { | ||
| return Strings.tokenizeToStringArray(expression, ","); | ||
| } | ||
|
|
||
| /** | ||
| * Generate the list of required matches from the expressions in {@code tokens} | ||
| * | ||
| * @param tokens List of expressions that may be wildcards or full Ids | ||
| * @param allowNoJobForWildcards If true then it is not required for wildcard | ||
| * expressions to match an Id meaning they are | ||
| * not returned in the list of required matches | ||
| * @return A list of required Id matchers | ||
| */ | ||
| static LinkedList<IdMatcher> requiredMatches(String [] tokens, boolean allowNoJobForWildcards) { | ||
| LinkedList<IdMatcher> matchers = new LinkedList<>(); | ||
|
|
||
| if (isWildcardAll(tokens)) { | ||
| // if allowNoJobForWildcards == true then any number | ||
| // of jobs with any id is ok. Therefore no matches | ||
| // are required | ||
|
|
||
| if (allowNoJobForWildcards == false) { | ||
| // require something, anything to match | ||
| matchers.add(new WildcardMatcher("*")); | ||
| } | ||
| return matchers; | ||
| } | ||
|
|
||
| if (allowNoJobForWildcards) { | ||
| // matches are not required for wildcards but | ||
| // specific job Ids are | ||
| for (String token : tokens) { | ||
| if (Regex.isSimpleMatchPattern(token) == false) { | ||
| matchers.add(new EqualsIdMatcher(token)); | ||
| } | ||
| } | ||
| } else { | ||
| // Matches are required for wildcards | ||
| for (String token : tokens) { | ||
| if (Regex.isSimpleMatchPattern(token)) { | ||
| matchers.add(new WildcardMatcher(token)); | ||
| } else { | ||
| matchers.add(new EqualsIdMatcher(token)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| return matchers; | ||
| } | ||
|
|
||
| /** | ||
| * For each given {@code requiredMatchers} check there is an element | ||
| * present in {@code ids} that matches. Once a match is made the | ||
| * matcher is popped from {@code requiredMatchers}. | ||
| * | ||
| * If all matchers are satisfied the list {@code requiredMatchers} will | ||
| * be empty after the call otherwise only the unmatched remain. | ||
| * | ||
| * @param requiredMatchers This is modified by the function: all matched matchers | ||
| * are removed from the list. At the end of the call only | ||
| * the unmatched ones are in this list | ||
| * @param ids Ids required to be matched | ||
| */ | ||
| static void filterMatchedIds(LinkedList<IdMatcher> requiredMatchers, Collection<String> ids) { | ||
| for (String id: ids) { | ||
| Iterator<IdMatcher> itr = requiredMatchers.iterator(); | ||
| if (itr.hasNext() == false) { | ||
| break; | ||
| } | ||
| while (itr.hasNext()) { | ||
| if (itr.next().matches(id)) { | ||
| itr.remove(); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| abstract static class IdMatcher { | ||
| protected final String id; | ||
|
|
||
| IdMatcher(String id) { | ||
| this.id = id; | ||
| } | ||
|
|
||
| public String getId() { | ||
| return id; | ||
| } | ||
|
|
||
| public abstract boolean matches(String jobId); | ||
| } | ||
|
|
||
| static class EqualsIdMatcher extends IdMatcher { | ||
| EqualsIdMatcher(String id) { | ||
| super(id); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean matches(String id) { | ||
| return this.id.equals(id); | ||
| } | ||
| } | ||
|
|
||
| static class WildcardMatcher extends IdMatcher { | ||
| WildcardMatcher(String id) { | ||
| super(id); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean matches(String id) { | ||
| return Regex.simpleMatch(this.id, id); | ||
| } | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
since we're back at having the jobs indexed, we should double check the length limit on the ID is correct