Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[ML] Refactor common utils out of ML plugin to XPack.Core #39976

Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
/*
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
* or more contributor license agreements. Licensed under the Elastic License;
* you may not use this file except in compliance with the Elastic License.
*/
package org.elasticsearch.xpack.core.action;

import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.ActionRequestValidationException;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.xpack.core.action.util.PageParams;

import java.io.IOException;
import java.util.Objects;

public abstract class AbstractGetResourcesRequest extends ActionRequest {

private String resourceId;
private PageParams pageParams = PageParams.defaultParams();
private boolean allowNoResources = false;

public final void setResourceId(String resourceId) {
this.resourceId = resourceId;
}

public final String getResourceId() {
return resourceId;
}

public final void setPageParams(PageParams pageParams) {
this.pageParams = pageParams;
}

public final PageParams getPageParams() {
return pageParams;
}

public final void setAllowNoResources(boolean allowNoResources) {
this.allowNoResources = allowNoResources;
}

public final boolean isAllowNoResources() {
return this.allowNoResources;
}

@Override
public ActionRequestValidationException validate() {
return null;
}

@Override
public void readFrom(StreamInput in) throws IOException {
super.readFrom(in);
resourceId = in.readOptionalString();
pageParams = in.readOptionalWriteable(PageParams::new);
allowNoResources = in.readBoolean();
}

@Override
public void writeTo(StreamOutput out) throws IOException {
super.writeTo(out);
out.writeOptionalString(resourceId);
out.writeOptionalWriteable(pageParams);
out.writeBoolean(allowNoResources);
}

@Override
public int hashCode() {
return Objects.hash(resourceId, pageParams, allowNoResources);
}

@Override
public boolean equals(Object obj) {
if (obj == null) {
return false;
}
if (obj instanceof AbstractGetResourcesRequest == false) {
return false;
}
AbstractGetResourcesRequest other = (AbstractGetResourcesRequest) obj;
return Objects.equals(resourceId, other.resourceId) &&
Objects.equals(pageParams, other.pageParams) &&
allowNoResources == other.allowNoResources;
}

public abstract String getResourceIdField();
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
/*
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
* or more contributor license agreements. Licensed under the Elastic License;
* you may not use this file except in compliance with the Elastic License.
*/
package org.elasticsearch.xpack.core.action;

import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.io.stream.Writeable;
import org.elasticsearch.common.xcontent.StatusToXContentObject;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.xpack.core.action.util.QueryPage;

import java.io.IOException;
import java.util.Objects;

public abstract class AbstractGetResourcesResponse<T extends ToXContent & Writeable> extends ActionResponse
implements StatusToXContentObject {

private QueryPage<T> resources;

protected AbstractGetResourcesResponse() {}

protected AbstractGetResourcesResponse(QueryPage<T> resources) {
this.resources = Objects.requireNonNull(resources);
}

public QueryPage<T> getResources() {
return resources;
}

@Override
public void readFrom(StreamInput in) throws IOException {
super.readFrom(in);
resources = new QueryPage<>(in, getReader());
}

@Override
public void writeTo(StreamOutput out) throws IOException {
super.writeTo(out);
resources.writeTo(out);
}

@Override
public RestStatus status() {
return RestStatus.OK;
}

@Override
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
builder.startObject();
resources.doXContentBody(builder, params);
builder.endObject();
return builder;
}

@Override
public int hashCode() {
return Objects.hash(resources);
}

@Override
public boolean equals(Object obj) {
if (obj == null) {
return false;
}
if (obj instanceof AbstractGetResourcesResponse == false) {
return false;
}
AbstractGetResourcesResponse other = (AbstractGetResourcesResponse) obj;
return Objects.equals(resources, other.resources);
}

@Override
public final String toString() {
return Strings.toString(this);
}
protected abstract Reader<T> getReader();
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,193 @@
/*
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
* or more contributor license agreements. Licensed under the Elastic License;
* you may not use this file except in compliance with the Elastic License.
*/
package org.elasticsearch.xpack.core.action;

import org.elasticsearch.ResourceNotFoundException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.HandledTransportAction;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.io.stream.Writeable;
import org.elasticsearch.common.regex.Regex;
import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentParser;
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.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.transport.TransportService;
import org.elasticsearch.xpack.core.action.util.ExpandedIdsMatcher;
import org.elasticsearch.xpack.core.action.util.QueryPage;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.function.Supplier;

import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;

/**
* Abstract transport class for collecting common logic in gathering Resource objects from indices
* @param <Resource> The type of the Resource being gathered
* @param <Request> The type of the Request
* @param <Response> The type of the Response
*/
public abstract class AbstractTransportGetResourcesAction<Resource extends ToXContent & Writeable,
Request extends AbstractGetResourcesRequest, Response extends AbstractGetResourcesResponse<Resource>>
extends HandledTransportAction<Request, Response> {

private static final String ALL = "_all";

private final Client client;
private final NamedXContentRegistry xContentRegistry;

protected AbstractTransportGetResourcesAction(String actionName, TransportService transportService, ActionFilters actionFilters,
Supplier<Request> request, Client client, NamedXContentRegistry xContentRegistry) {
super(actionName, transportService, actionFilters, request);
this.client = Objects.requireNonNull(client);
this.xContentRegistry = Objects.requireNonNull(xContentRegistry);
}

protected void searchResources(AbstractGetResourcesRequest request, ActionListener<QueryPage<Resource>> listener) {
String[] tokens = Strings.tokenizeToStringArray(request.getResourceId(), ",");
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
.sort(request.getResourceIdField())
.query(buildQuery(tokens, request.getResourceIdField()));
if (request.getPageParams() != null) {
sourceBuilder.from(request.getPageParams().getFrom())
.size(request.getPageParams().getSize());
}

IndicesOptions indicesOptions = SearchRequest.DEFAULT_INDICES_OPTIONS;
SearchRequest searchRequest = new SearchRequest(getIndices())
.indicesOptions(IndicesOptions.fromOptions(true,
indicesOptions.allowNoIndices(),
indicesOptions.expandWildcardsOpen(),
indicesOptions.expandWildcardsClosed(),
indicesOptions))
.source(sourceBuilder);

executeAsyncWithOrigin(client.threadPool().getThreadContext(),
executionOrigin(),
searchRequest,
new ActionListener<SearchResponse>() {
@Override
public void onResponse(SearchResponse response) {
List<Resource> docs = new ArrayList<>();
Set<String> foundResourceIds = new HashSet<>();
for (SearchHit hit : response.getHits().getHits()) {
BytesReference docSource = hit.getSourceRef();
try (InputStream stream = docSource.streamInput();
XContentParser parser = XContentFactory.xContent(XContentType.JSON).createParser(
xContentRegistry, LoggingDeprecationHandler.INSTANCE, stream)) {
Resource resource = parse(parser);
docs.add(resource);
foundResourceIds.add(extractIdFromResource(resource));
} catch (IOException e) {
this.onFailure(e);
}
}
ExpandedIdsMatcher requiredMatches = new ExpandedIdsMatcher(tokens, request.isAllowNoResources());
requiredMatches.filterMatchedIds(foundResourceIds);
if (requiredMatches.hasUnmatchedIds()) {
listener.onFailure(notFoundException(requiredMatches.unmatchedIdsString()));
} else {
listener.onResponse(new QueryPage<>(docs, docs.size(), getResultsField()));
}
}


@Override
public void onFailure(Exception e) {
listener.onFailure(e);
}
},
client::search);
}

private QueryBuilder buildQuery(String[] tokens, String resourceIdField) {
BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();

// If the resourceId is not _all or *, we should see if it is a comma delimited string with wild-cards
// e.g. id1,id2*,id3
if (Strings.isAllOrWildcard(tokens) == false) {
BoolQueryBuilder shouldQueries = new BoolQueryBuilder();
List<String> terms = new ArrayList<>();
for (String token : tokens) {
if (Regex.isSimpleMatchPattern(token)) {
shouldQueries.should(QueryBuilders.wildcardQuery(resourceIdField, token));
} else {
terms.add(token);
}
}
if (terms.isEmpty() == false) {
shouldQueries.should(QueryBuilders.termsQuery(resourceIdField, terms));
}

if (shouldQueries.should().isEmpty() == false) {
boolQuery.filter(shouldQueries);
}
}
QueryBuilder additionalQuery = additionalQuery();
if (additionalQuery != null) {
boolQuery.filter(additionalQuery);
}
return boolQuery.hasClauses() ? boolQuery : QueryBuilders.matchAllQuery();
}

@Nullable
protected QueryBuilder additionalQuery() {
return null;
}

/**
* @return The results field parse field so that the response is properly formatted
*/
protected abstract ParseField getResultsField();

/**
* @return The indices needed to query
*/
protected abstract String[] getIndices();

/**
* @param parser Constructed XContentParser from search response hits to relay to a parser for the Resource
* @return parsed Resource typed object
*/
protected abstract Resource parse(XContentParser parser) throws IOException;
/**
* @param resourceId Resource ID or expression that was not found in the search results
* @return The exception to throw in the event that an ID or expression is not found
*/
protected abstract ResourceNotFoundException notFoundException(String resourceId);

/**
* @return The appropriate origin under which to execute the search requests
*/
protected abstract String executionOrigin();

/**
* @param resource A parsed Resource object
* @return The ID of the resource
*/
protected abstract String extractIdFromResource(Resource resource);
}
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
* or more contributor license agreements. Licensed under the Elastic License;
* you may not use this file except in compliance with the Elastic License.
*/
package org.elasticsearch.xpack.ml.job.persistence;
package org.elasticsearch.xpack.core.action.util;

import org.elasticsearch.common.Strings;
import org.elasticsearch.common.regex.Regex;
Expand Down Expand Up @@ -38,7 +38,7 @@ public final class ExpandedIdsMatcher {
* @param expression Expression containing zero or more ','s
* @return Array of tokens
*/
public static String [] tokenizeExpression(String expression) {
public static String[] tokenizeExpression(String expression) {
return Strings.tokenizeToStringArray(expression, ",");
}

Expand All @@ -50,10 +50,10 @@ public final class ExpandedIdsMatcher {
*
* @param tokens List of expressions that may be wildcards or full Ids
* @param allowNoMatchForWildcards 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
* expressions to match an Id meaning they are
* not returned in the list of required matches
*/
public ExpandedIdsMatcher(String [] tokens, boolean allowNoMatchForWildcards) {
public ExpandedIdsMatcher(String[] tokens, boolean allowNoMatchForWildcards) {
requiredMatches = new LinkedList<>();

if (Strings.isAllOrWildcard(tokens)) {
Expand Down
Loading