Skip to content

Commit

Permalink
REST high-level client: add simulate pipeline API (#31158)
Browse files Browse the repository at this point in the history 
relates to #27205
  • Loading branch information
@sohaibiftikhar @javanna
sohaibiftikhar authored and javanna committed Jun 26, 2018
1 parent 626ff21 commit adbfd1d
Show file tree
Hide file tree
Showing 21 changed files with 1,186 additions and 129 deletions.
35 changes: 35 additions & 0 deletions client/rest-high-level/src/main/java/org/elasticsearch/client/IngestClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@
import org.elasticsearch.action.ingest.GetPipelineRequest;
import org.elasticsearch.action.ingest.GetPipelineResponse;
import org.elasticsearch.action.ingest.PutPipelineRequest;
import org.elasticsearch.action.ingest.SimulatePipelineRequest;
import org.elasticsearch.action.ingest.SimulatePipelineResponse;
import org.elasticsearch.action.ingest.WritePipelineResponse;

import java.io.IOException;
Expand Down Expand Up @@ -125,4 +127,37 @@ public void deletePipelineAsync(DeletePipelineRequest request, RequestOptions op
restHighLevelClient.performRequestAsyncAndParseEntity( request, RequestConverters::deletePipeline, options,
WritePipelineResponse::fromXContent, listener, emptySet());
}

/**
* Simulate a pipeline on a set of documents provided in the request
* <p>
* See
* <a href="https://www.elastic.co/guide/en/elasticsearch/reference/master/simulate-pipeline-api.html">
* Simulate Pipeline API on elastic.co</a>
* @param request the request
* @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
* @return the response
* @throws IOException in case there is a problem sending the request or parsing back the response
*/
public SimulatePipelineResponse simulatePipeline(SimulatePipelineRequest request, RequestOptions options) throws IOException {
return restHighLevelClient.performRequestAndParseEntity( request, RequestConverters::simulatePipeline, options,
SimulatePipelineResponse::fromXContent, emptySet());
}

/**
* Asynchronously simulate a pipeline on a set of documents provided in the request
* <p>
* See
* <a href="https://www.elastic.co/guide/en/elasticsearch/reference/master/simulate-pipeline-api.html">
* Simulate Pipeline API on elastic.co</a>
* @param request the request
* @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
* @param listener the listener to be notified upon request completion
*/
public void simulatePipelineAsync(SimulatePipelineRequest request,
RequestOptions options,
ActionListener<SimulatePipelineResponse> listener) {
restHighLevelClient.performRequestAsyncAndParseEntity( request, RequestConverters::simulatePipeline, options,
SimulatePipelineResponse::fromXContent, listener, emptySet());
}
}
15 changes: 15 additions & 0 deletions client/rest-high-level/src/main/java/org/elasticsearch/client/RequestConverters.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,7 @@
import org.elasticsearch.action.ingest.DeletePipelineRequest;
import org.elasticsearch.action.ingest.PutPipelineRequest;
import org.elasticsearch.action.ingest.GetPipelineRequest;
import org.elasticsearch.action.ingest.SimulatePipelineRequest;
import org.elasticsearch.action.search.ClearScrollRequest;
import org.elasticsearch.action.search.MultiSearchRequest;
import org.elasticsearch.action.search.SearchRequest;
Expand Down Expand Up @@ -927,6 +928,20 @@ static Request validateQuery(ValidateQueryRequest validateQueryRequest) throws I
return request;
}

static Request simulatePipeline(SimulatePipelineRequest simulatePipelineRequest) throws IOException {
EndpointBuilder builder = new EndpointBuilder().addPathPartAsIs("_ingest/pipeline");
if (simulatePipelineRequest.getId() != null && !simulatePipelineRequest.getId().isEmpty()) {
builder.addPathPart(simulatePipelineRequest.getId());
}
builder.addPathPartAsIs("_simulate");
String endpoint = builder.build();
Request request = new Request(HttpPost.METHOD_NAME, endpoint);
Params params = new Params(request);
params.putParam("verbose", Boolean.toString(simulatePipelineRequest.isVerbose()));
request.setEntity(createEntity(simulatePipelineRequest, REQUEST_BODY_CONTENT_TYPE));
return request;
}

static Request getAlias(GetAliasesRequest getAliasesRequest) {
String[] indices = getAliasesRequest.indices() == null ? Strings.EMPTY_ARRAY : getAliasesRequest.indices();
String[] aliases = getAliasesRequest.aliases() == null ? Strings.EMPTY_ARRAY : getAliasesRequest.aliases();
Expand Down
10 changes: 7 additions & 3 deletions ...rest-high-level/src/test/java/org/elasticsearch/client/ESRestHighLevelClientTestCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -123,9 +123,7 @@ private HighLevelClient(RestClient restClient) {
}
}

protected static XContentBuilder buildRandomXContentPipeline() throws IOException {
XContentType xContentType = randomFrom(XContentType.values());
XContentBuilder pipelineBuilder = XContentBuilder.builder(xContentType.xContent());
protected static XContentBuilder buildRandomXContentPipeline(XContentBuilder pipelineBuilder) throws IOException {
pipelineBuilder.startObject();
{
pipelineBuilder.field(Pipeline.DESCRIPTION_KEY, "some random set of processors");
Expand All @@ -152,6 +150,12 @@ protected static XContentBuilder buildRandomXContentPipeline() throws IOExceptio
return pipelineBuilder;
}

protected static XContentBuilder buildRandomXContentPipeline() throws IOException {
XContentType xContentType = randomFrom(XContentType.values());
XContentBuilder pipelineBuilder = XContentBuilder.builder(xContentType.xContent());
return buildRandomXContentPipeline(pipelineBuilder);
}

protected static void createPipeline(String pipelineId) throws IOException {
XContentBuilder builder = buildRandomXContentPipeline();
createPipeline(new PutPipelineRequest(pipelineId, BytesReference.bytes(builder), builder.contentType()));
Expand Down
99 changes: 99 additions & 0 deletions client/rest-high-level/src/test/java/org/elasticsearch/client/IngestClientIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,12 +23,22 @@
import org.elasticsearch.action.ingest.GetPipelineRequest;
import org.elasticsearch.action.ingest.GetPipelineResponse;
import org.elasticsearch.action.ingest.PutPipelineRequest;
import org.elasticsearch.action.ingest.SimulateDocumentBaseResult;
import org.elasticsearch.action.ingest.SimulateDocumentResult;
import org.elasticsearch.action.ingest.SimulateDocumentVerboseResult;
import org.elasticsearch.action.ingest.SimulatePipelineRequest;
import org.elasticsearch.action.ingest.SimulatePipelineResponse;
import org.elasticsearch.action.ingest.WritePipelineResponse;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.ingest.PipelineConfiguration;

import java.io.IOException;
import java.util.List;

import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.core.IsInstanceOf.instanceOf;

public class IngestClientIT extends ESRestHighLevelClientTestCase {

Expand Down Expand Up @@ -80,4 +90,93 @@ public void testDeletePipeline() throws IOException {
execute(request, highLevelClient().ingest()::deletePipeline, highLevelClient().ingest()::deletePipelineAsync);
assertTrue(response.isAcknowledged());
}

public void testSimulatePipeline() throws IOException {
testSimulatePipeline(false, false);
}

public void testSimulatePipelineWithFailure() throws IOException {
testSimulatePipeline(false, true);
}

public void testSimulatePipelineVerbose() throws IOException {
testSimulatePipeline(true, false);
}

public void testSimulatePipelineVerboseWithFailure() throws IOException {
testSimulatePipeline(true, true);
}

private void testSimulatePipeline(boolean isVerbose,
boolean isFailure) throws IOException {
XContentType xContentType = randomFrom(XContentType.values());
XContentBuilder builder = XContentBuilder.builder(xContentType.xContent());
String rankValue = isFailure ? "non-int" : Integer.toString(1234);
builder.startObject();
{
builder.field("pipeline");
buildRandomXContentPipeline(builder);
builder.startArray("docs");
{
builder.startObject()
.field("_index", "index")
.field("_type", "doc")
.field("_id", "doc_" + 1)
.startObject("_source").field("foo", "rab_" + 1).field("rank", rankValue).endObject()
.endObject();
}
builder.endArray();
}
builder.endObject();

SimulatePipelineRequest request = new SimulatePipelineRequest(
BytesReference.bytes(builder),
builder.contentType()
);
request.setVerbose(isVerbose);
SimulatePipelineResponse response =
execute(request, highLevelClient().ingest()::simulatePipeline, highLevelClient().ingest()::simulatePipelineAsync);
List<SimulateDocumentResult> results = response.getResults();
assertEquals(1, results.size());
if (isVerbose) {
assertThat(results.get(0), instanceOf(SimulateDocumentVerboseResult.class));
SimulateDocumentVerboseResult verboseResult = (SimulateDocumentVerboseResult) results.get(0);
assertEquals(2, verboseResult.getProcessorResults().size());
if (isFailure) {
assertNotNull(verboseResult.getProcessorResults().get(1).getFailure());
assertThat(verboseResult.getProcessorResults().get(1).getFailure().getMessage(),
containsString("unable to convert [non-int] to integer"));
} else {
assertEquals(
verboseResult.getProcessorResults().get(0).getIngestDocument()
.getFieldValue("foo", String.class),
"bar"
);
assertEquals(
Integer.valueOf(1234),
verboseResult.getProcessorResults().get(1).getIngestDocument()
.getFieldValue("rank", Integer.class)
);
}
} else {
assertThat(results.get(0), instanceOf(SimulateDocumentBaseResult.class));
SimulateDocumentBaseResult baseResult = (SimulateDocumentBaseResult)results.get(0);
if (isFailure) {
assertNotNull(baseResult.getFailure());
assertThat(baseResult.getFailure().getMessage(),
containsString("unable to convert [non-int] to integer"));
} else {
assertNotNull(baseResult.getIngestDocument());
assertEquals(
baseResult.getIngestDocument().getFieldValue("foo", String.class),
"bar"
);
assertEquals(
Integer.valueOf(1234),
baseResult.getIngestDocument()
.getFieldValue("rank", Integer.class)
);
}
}
}
}
29 changes: 29 additions & 0 deletions client/rest-high-level/src/test/java/org/elasticsearch/client/RequestConvertersTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,7 @@
import org.elasticsearch.action.ingest.DeletePipelineRequest;
import org.elasticsearch.action.ingest.GetPipelineRequest;
import org.elasticsearch.action.ingest.PutPipelineRequest;
import org.elasticsearch.action.ingest.SimulatePipelineRequest;
import org.elasticsearch.action.search.ClearScrollRequest;
import org.elasticsearch.action.search.MultiSearchRequest;
import org.elasticsearch.action.search.SearchRequest;
Expand Down Expand Up @@ -1622,6 +1623,34 @@ public void testDeletePipeline() {
assertEquals(expectedParams, expectedRequest.getParameters());
}

public void testSimulatePipeline() throws IOException {
String pipelineId = randomBoolean() ? "some_pipeline_id" : null;
boolean verbose = randomBoolean();
String json = "{\"pipeline\":{" +
"\"description\":\"_description\"," +
"\"processors\":[{\"set\":{\"field\":\"field2\",\"value\":\"_value\"}}]}," +
"\"docs\":[{\"_index\":\"index\",\"_type\":\"_doc\",\"_id\":\"id\",\"_source\":{\"foo\":\"rab\"}}]}";
SimulatePipelineRequest request = new SimulatePipelineRequest(
new BytesArray(json.getBytes(StandardCharsets.UTF_8)),
XContentType.JSON
);
request.setId(pipelineId);
request.setVerbose(verbose);
Map<String, String> expectedParams = new HashMap<>();
expectedParams.put("verbose", Boolean.toString(verbose));

Request expectedRequest = RequestConverters.simulatePipeline(request);
StringJoiner endpoint = new StringJoiner("/", "/", "");
endpoint.add("_ingest/pipeline");
if (pipelineId != null && !pipelineId.isEmpty())
endpoint.add(pipelineId);
endpoint.add("_simulate");
assertEquals(endpoint.toString(), expectedRequest.getEndpoint());
assertEquals(HttpPost.METHOD_NAME, expectedRequest.getMethod());
assertEquals(expectedParams, expectedRequest.getParameters());
assertToXContentBody(request, expectedRequest.getEntity());
}

public void testClusterHealth() {
ClusterHealthRequest healthRequest = new ClusterHealthRequest();
Map<String, String> expectedParams = new HashMap<>();
Expand Down
111 changes: 111 additions & 0 deletions ...vel/src/test/java/org/elasticsearch/client/documentation/IngestClientDocumentationIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,12 @@
import org.elasticsearch.action.ingest.GetPipelineRequest;
import org.elasticsearch.action.ingest.GetPipelineResponse;
import org.elasticsearch.action.ingest.PutPipelineRequest;
import org.elasticsearch.action.ingest.SimulateDocumentBaseResult;
import org.elasticsearch.action.ingest.SimulateDocumentResult;
import org.elasticsearch.action.ingest.SimulateDocumentVerboseResult;
import org.elasticsearch.action.ingest.SimulatePipelineRequest;
import org.elasticsearch.action.ingest.SimulatePipelineResponse;
import org.elasticsearch.action.ingest.SimulateProcessorResult;
import org.elasticsearch.action.ingest.WritePipelineResponse;
import org.elasticsearch.client.ESRestHighLevelClientTestCase;
import org.elasticsearch.client.RequestOptions;
Expand Down Expand Up @@ -277,4 +283,109 @@ public void onFailure(Exception e) {
}
}

public void testSimulatePipeline() throws IOException {
RestHighLevelClient client = highLevelClient();

{
// tag::simulate-pipeline-request
String source =
"{\"" +
"pipeline\":{" +
"\"description\":\"_description\"," +
"\"processors\":[{\"set\":{\"field\":\"field2\",\"value\":\"_value\"}}]" +
"}," +
"\"docs\":[" +
"{\"_index\":\"index\",\"_type\":\"_doc\",\"_id\":\"id\",\"_source\":{\"foo\":\"bar\"}}," +
"{\"_index\":\"index\",\"_type\":\"_doc\",\"_id\":\"id\",\"_source\":{\"foo\":\"rab\"}}" +
"]" +
"}";
SimulatePipelineRequest request = new SimulatePipelineRequest(
new BytesArray(source.getBytes(StandardCharsets.UTF_8)), // <1>
XContentType.JSON // <2>
);
// end::simulate-pipeline-request

// tag::simulate-pipeline-request-pipeline-id
request.setId("my-pipeline-id"); // <1>
// end::simulate-pipeline-request-pipeline-id

// For testing we set this back to null
request.setId(null);

// tag::simulate-pipeline-request-verbose
request.setVerbose(true); // <1>
// end::simulate-pipeline-request-verbose

// tag::simulate-pipeline-execute
SimulatePipelineResponse response = client.ingest().simulatePipeline(request, RequestOptions.DEFAULT); // <1>
// end::simulate-pipeline-execute

// tag::simulate-pipeline-response
for (SimulateDocumentResult result: response.getResults()) { // <1>
if (request.isVerbose()) {
assert result instanceof SimulateDocumentVerboseResult;
SimulateDocumentVerboseResult verboseResult = (SimulateDocumentVerboseResult)result; // <2>
for (SimulateProcessorResult processorResult: verboseResult.getProcessorResults()) { // <3>
processorResult.getIngestDocument(); // <4>
processorResult.getFailure(); // <5>
}
} else {
assert result instanceof SimulateDocumentBaseResult;
SimulateDocumentBaseResult baseResult = (SimulateDocumentBaseResult)result; // <6>
baseResult.getIngestDocument(); // <7>
baseResult.getFailure(); // <8>
}
}
// end::simulate-pipeline-response
assert(response.getResults().size() > 0);
}
}

public void testSimulatePipelineAsync() throws Exception {
RestHighLevelClient client = highLevelClient();

{
String source =
"{\"" +
"pipeline\":{" +
"\"description\":\"_description\"," +
"\"processors\":[{\"set\":{\"field\":\"field2\",\"value\":\"_value\"}}]" +
"}," +
"\"docs\":[" +
"{\"_index\":\"index\",\"_type\":\"_doc\",\"_id\":\"id\",\"_source\":{\"foo\":\"bar\"}}," +
"{\"_index\":\"index\",\"_type\":\"_doc\",\"_id\":\"id\",\"_source\":{\"foo\":\"rab\"}}" +
"]" +
"}";
SimulatePipelineRequest request = new SimulatePipelineRequest(
new BytesArray(source.getBytes(StandardCharsets.UTF_8)),
XContentType.JSON
);

// tag::simulate-pipeline-execute-listener
ActionListener<SimulatePipelineResponse> listener =
new ActionListener<SimulatePipelineResponse>() {
@Override
public void onResponse(SimulatePipelineResponse response) {
// <1>
}

@Override
public void onFailure(Exception e) {
// <2>
}
};
// end::simulate-pipeline-execute-listener

// Replace the empty listener by a blocking listener in test
final CountDownLatch latch = new CountDownLatch(1);
listener = new LatchedActionListener<>(listener, latch);

// tag::simulate-pipeline-execute-async
client.ingest().simulatePipelineAsync(request, RequestOptions.DEFAULT, listener); // <1>
// end::simulate-pipeline-execute-async

assertTrue(latch.await(30L, TimeUnit.SECONDS));
}
}

}
Loading

0 comments on commit adbfd1d

Please sign in to comment.