Add create rollup job api to high level rest client

client/rest-high-level/src/main/java/org/elasticsearch/client/RequestConverters.java

@@ -112,6 +112,7 @@
 import org.elasticsearch.protocol.xpack.license.PutLicenseRequest;
 import org.elasticsearch.protocol.xpack.migration.IndexUpgradeInfoRequest;
 import org.elasticsearch.protocol.xpack.ml.PutJobRequest;
+import org.elasticsearch.protocol.xpack.rollup.PutRollupJobRequest;
 import org.elasticsearch.protocol.xpack.watcher.DeleteWatchRequest;
 import org.elasticsearch.protocol.xpack.watcher.PutWatchRequest;
 import org.elasticsearch.rest.action.search.RestSearchAction;
@@ -1213,6 +1214,18 @@ static Request getMigrationAssistance(IndexUpgradeInfoRequest indexUpgradeInfoRe
         return request;
     }
 
+    static Request putRollupJob(PutRollupJobRequest putRollupJobRequest) throws IOException {
+        String endpoint = new EndpointBuilder()
+            .addPathPartAsIs("_xpack")
+            .addPathPartAsIs("rollup")
+            .addPathPartAsIs("job")
+            .addPathPart(putRollupJobRequest.getConfig().getId())
+            .build();
+        Request request = new Request(HttpPut.METHOD_NAME, endpoint);
+        request.setEntity(createEntity(putRollupJobRequest, REQUEST_BODY_CONTENT_TYPE));
+        return request;
+    }
+
     private static HttpEntity createEntity(ToXContent toXContent, XContentType xContentType) throws IOException {
         BytesRef source = XContentHelper.toXContent(toXContent, xContentType, false).toBytesRef();
         return new ByteArrayEntity(source.bytes, source.offset, source.length, createContentType(xContentType));

client/rest-high-level/src/main/java/org/elasticsearch/client/RestHighLevelClient.java

@@ -212,6 +212,7 @@ public class RestHighLevelClient implements Closeable {
     private final LicenseClient licenseClient = new LicenseClient(this);
     private final MigrationClient migrationClient = new MigrationClient(this);
     private final MachineLearningClient machineLearningClient = new MachineLearningClient(this);
+    private final RollupClient rollupClient = new RollupClient(this);
 
     /**
      * Creates a {@link RestHighLevelClient} given the low level {@link RestClientBuilder} that allows to build the
@@ -361,6 +362,16 @@ public MachineLearningClient machineLearning() {
         return machineLearningClient;
     }
 
+    /**
+     * Provides methods for accessing the Elastic Licensed Rollup APIs that
+     * are shipped with the default distribution of Elasticsearch. All of
+     * these APIs will 404 if run against the OSS distribution of Elasticsearch.
+     * <p>
+     * See the <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/rollup-apis.html">
+     * Watcher APIs on elastic.co</a> for more information.
+     */
+    public RollupClient rollup() { return rollupClient; }
+
     /**
      * Executes a bulk request using the Bulk API.
      * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html">Bulk API on elastic.co</a>

client/rest-high-level/src/main/java/org/elasticsearch/client/RollupClient.java

@@ -0,0 +1,71 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.client;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.protocol.xpack.rollup.PutRollupJobRequest;
+import org.elasticsearch.protocol.xpack.rollup.PutRollupJobResponse;
+
+import java.io.IOException;
+
+import static java.util.Collections.emptySet;
+
+/**
+ * A wrapper for the {@link RestHighLevelClient} that provides methods for
+ * accessing the Elastic Rollup-related methods
+ * <p>
+ * See the <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/rollup-apis.html">
+ * X-Pack Rollup APIs on elastic.co</a> for more information.
+ */
+public class RollupClient {
+
+    private final RestHighLevelClient restHighLevelClient;
+
+    RollupClient(RestHighLevelClient restHighLevelClient) {
+        this.restHighLevelClient = restHighLevelClient;
+    }
+
+    /**
+     * Put a rollup job into the cluster
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/rollup-put-job.html">
+     * the docs</a> for more.
+     * @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 PutRollupJobResponse putRollupJob(PutRollupJobRequest request, RequestOptions options) throws IOException {
+        return restHighLevelClient.performRequestAndParseEntity(request, RequestConverters::putRollupJob, options,
+            PutRollupJobResponse::fromXContent, emptySet());
+    }
+
+    /**
+     * Asynchronously put a rollup job into the cluster
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/rollup-put-job.html">
+     * the docs</a> for more.
+     * @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 putRollupJobAsync(PutRollupJobRequest request, RequestOptions options, ActionListener<PutRollupJobResponse> listener) {
+        restHighLevelClient.performRequestAsyncAndParseEntity(request, RequestConverters::putRollupJob, options,
+            PutRollupJobResponse::fromXContent, listener, emptySet());
+    }
+}

client/rest-high-level/src/test/java/org/elasticsearch/client/RestHighLevelClientTests.java

@@ -757,6 +757,7 @@ public void testApiNamingConventions() throws Exception {
                         if (apiName.startsWith("xpack.") == false &&
                             apiName.startsWith("license.") == false &&
                             apiName.startsWith("machine_learning.") == false &&
+                            apiName.startsWith("rollup.") == false &&
                             apiName.startsWith("watcher.") == false &&
                             apiName.startsWith("migration.") == false) {
                             apiNotFound.add(apiName);

client/rest-high-level/src/test/java/org/elasticsearch/client/RollupIT.java

@@ -0,0 +1,158 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client;
+
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.protocol.xpack.rollup.PutRollupJobRequest;
+import org.elasticsearch.protocol.xpack.rollup.PutRollupJobResponse;
+import org.elasticsearch.protocol.xpack.rollup.job.DateHistogramGroupConfig;
+import org.elasticsearch.protocol.xpack.rollup.job.GroupConfig;
+import org.elasticsearch.protocol.xpack.rollup.job.MetricConfig;
+import org.elasticsearch.protocol.xpack.rollup.job.RollupJobConfig;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.elasticsearch.protocol.xpack.rollup.RollupField.SUPPORTED_METRICS;
+
+public class RollupIT extends ESRestHighLevelClientTestCase {
+
+    @SuppressWarnings("unchecked")
+    public void testPutRollupJob() throws Exception {
+        final Set<Integer> values = new HashSet<>();
+        double sum = 0.0d;
+        int max = Integer.MIN_VALUE;
+        int min = Integer.MAX_VALUE;
+
+        final BulkRequest bulkRequest = new BulkRequest();
+        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+        for (int minute = 0; minute < 60; minute++) {
+            for (int second = 0; second < 60; second = second + 10) {
+                final int value = randomIntBetween(0, 100);
+
+                final IndexRequest indexRequest = new IndexRequest("docs", "doc");
+                indexRequest.source(jsonBuilder()
+                    .startObject()
+                    .field("value", value)
+                    .field("date", String.format(Locale.ROOT, "2018-01-01T00:%02d:%02dZ", minute, second))
+                    .endObject());
+                bulkRequest.add(indexRequest);
+
+                values.add(value);
+                sum += value;
+                if (value > max) {
+                    max = value;
+                }
+                if (value < min) {
+                    min = value;
+                }
+            }
+        }
+
+        final int numDocs = bulkRequest.numberOfActions();
+
+        BulkResponse bulkResponse = highLevelClient().bulk(bulkRequest, RequestOptions.DEFAULT);
+        assertEquals(RestStatus.OK, bulkResponse.status());
+        if (bulkResponse.hasFailures())     {
+            for (BulkItemResponse itemResponse : bulkResponse.getItems()) {
+                if (itemResponse.isFailed()) {
+                    logger.fatal(itemResponse.getFailureMessage());
+                }
+            }
+        }
+        assertFalse(bulkResponse.hasFailures());
+
+        RefreshResponse refreshResponse = highLevelClient().indices().refresh(new RefreshRequest("docs"), RequestOptions.DEFAULT);
+        assertEquals(0, refreshResponse.getFailedShards());
+
+        final String id = randomAlphaOfLength(10);
+        final String indexPattern = randomFrom("docs", "d*", "doc*");
+        final String rollupIndex = randomFrom("rollup", "test");
+        final String cron = "*/1 * * * * ?";
+        final int pageSize = randomIntBetween(numDocs, numDocs * 10);
+        // TODO expand this to also test with histogram and terms?
+        final GroupConfig groups = new GroupConfig(new DateHistogramGroupConfig("date", DateHistogramInterval.DAY));
+        final List<MetricConfig> metrics = Collections.singletonList(new MetricConfig("value", SUPPORTED_METRICS));
+        final TimeValue timeout = TimeValue.timeValueSeconds(randomIntBetween(30, 600));
+
+        PutRollupJobRequest putRollupJobRequest =
+            new PutRollupJobRequest(new RollupJobConfig(id, indexPattern, rollupIndex, cron, pageSize, groups, metrics, timeout));
+
+        final RollupClient rollupClient = highLevelClient().rollup();
+        PutRollupJobResponse response = execute(putRollupJobRequest, rollupClient::putRollupJob, rollupClient::putRollupJobAsync);
+        assertTrue(response.isAcknowledged());
+
+        // TODO Replace this with the Rollup Start Job API
+        Response startResponse = client().performRequest(new Request("POST", "/_xpack/rollup/job/" + id + "/_start"));
+        assertEquals(RestStatus.OK.getStatus(), startResponse.getHttpResponse().getStatusLine().getStatusCode());
+
+        int finalMin = min;
+        int finalMax = max;
+        double finalSum = sum;
+        assertBusy(() -> {
+            SearchResponse searchResponse = highLevelClient().search(new SearchRequest(rollupIndex), RequestOptions.DEFAULT);
+            assertEquals(0, searchResponse.getFailedShards());
+            assertEquals(1L, searchResponse.getHits().getTotalHits());
+
+            SearchHit searchHit = searchResponse.getHits().getAt(0);
+            Map<String, Object> source = searchHit.getSourceAsMap();
+            assertNotNull(source);
+
+            assertEquals(numDocs, source.get("date.date_histogram._count"));
+            assertEquals(groups.getDateHistogram().getInterval().toString(), source.get("date.date_histogram.interval"));
+            assertEquals(groups.getDateHistogram().getTimeZone(), source.get("date.date_histogram.time_zone"));
+
+            for (MetricConfig metric : metrics) {
+                for (String name : metric.getMetrics()) {
+                    Number value = (Number) source.get(metric.getField() + "." + name + ".value");
+                    if ("min".equals(name)) {
+                        assertEquals(finalMin, value.intValue());
+                    } else if ("max".equals(name)) {
+                        assertEquals(finalMax, value.intValue());
+                    } else if ("sum".equals(name)) {
+                        assertEquals(finalSum, value.doubleValue(), 0.0d);
+                    } else if ("avg".equals(name)) {
+                        assertEquals(finalSum, value.doubleValue(), 0.0d);
+                        Number avgCount = (Number) source.get(metric.getField() + "." + name + "._count");
+                        assertEquals(numDocs, avgCount.intValue());
+                    } else if ("value_count".equals(name)) {
+                        assertEquals(numDocs, value.intValue());
+                    }
+                }
+            }
+        });
+    }
+}