Create a mode: time_series option for indices

libs/x-content/src/main/java/org/elasticsearch/common/xcontent/support/MapXContentParser.java

@@ -163,7 +163,7 @@ public Object objectBytes() throws IOException {
 
     @Override
     public boolean hasTextCharacters() {
-        throw new UnsupportedOperationException("use text() instead");
+        return false;
     }
 
     @Override

libs/x-content/src/test/java/org/elasticsearch/common/xcontent/MapXContentParserTests.java

@@ -10,6 +10,7 @@
 
 import org.elasticsearch.core.CheckedConsumer;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.common.xcontent.support.MapXContentParser;
 import org.elasticsearch.test.ESTestCase;
 
@@ -57,19 +58,51 @@ public void testSimpleMap() throws IOException {
         });
     }
 
-
     public void testRandomObject() throws IOException {
         compareTokens(builder -> generateRandomObject(builder, randomIntBetween(0, 10)));
     }
 
-    public void compareTokens(CheckedConsumer<XContentBuilder, IOException> consumer) throws IOException {
+    /**
+     * Assert that {@link XContentParser#hasTextCharacters()} returns false because
+     * we don't support {@link XContentParser#textCharacters()}.
+     */
+    public void testHasTextCharacters() throws IOException {
+        assertFalse(
+            new MapXContentParser(
+                xContentRegistry(),
+                LoggingDeprecationHandler.INSTANCE,
+                Map.of("a", "b"),
+                randomFrom(XContentType.values())
+            ).hasTextCharacters()
+        );
+    }
+
+    public void testCopyCurrentStructure() throws IOException {
+        try (
+            XContentParser parser = new MapXContentParser(
+                xContentRegistry(),
+                LoggingDeprecationHandler.INSTANCE,
+                Map.of("a", "b"),
+                randomFrom(XContentType.values())
+            )
+        ) {
+            try (
+                XContentBuilder builder = JsonXContent.contentBuilder().copyCurrentStructure(parser);
+                XContentParser copied = createParser(builder)
+            ) {
+                assertEquals(copied.map(), Map.of("a", "b"));
+            }
+        }
+    }
+
+    private void compareTokens(CheckedConsumer<XContentBuilder, IOException> consumer) throws IOException {
         for (XContentType xContentType : EnumSet.allOf(XContentType.class)) {
             logger.info("--> testing with xcontent type: {}", xContentType);
             compareTokens(consumer, xContentType);
         }
     }
 
-    public void compareTokens(CheckedConsumer<XContentBuilder, IOException> consumer, XContentType xContentType) throws IOException {
+    private void compareTokens(CheckedConsumer<XContentBuilder, IOException> consumer, XContentType xContentType) throws IOException {
         try (XContentBuilder builder = XContentBuilder.builder(xContentType.xContent())) {
             consumer.accept(builder);
             final Map<String, Object> map;

plugins/repository-hdfs/build.gradle

@@ -76,7 +76,7 @@ dependencies {
 
 restResources {
   restApi {
-    include '_common', 'cluster', 'nodes', 'indices', 'index', 'snapshot'
+    include '_common', 'bulk', 'cluster', 'index', 'indices', 'nodes', 'search', 'snapshot'
   }
 }
 

plugins/repository-hdfs/src/test/resources/rest-api-spec/test/hdfs_repository/50_tsdb.yml

@@ -0,0 +1,148 @@
+---
+"Create a snapshot and then restore it":
+  - skip:
+      version: " - 7.99.99"
+      reason: introduced in 8.0.0 to be backported to 7.15.0
+      features: ["allowed_warnings"]
+
+  # Create repository
+  - do:
+      snapshot.create_repository:
+        repository: test_restore_repository
+        body:
+          type: hdfs
+          settings:
+            uri: "hdfs://localhost:9999"
+            path: "test/restore"
+
+  # Create index
+  - do:
+      indices.create:
+          index: test_index
+          body:
+            settings:
+              index:
+                mode: time_series
+                number_of_replicas: 0
+                number_of_shards: 2
+            mappings:
+              properties:
+                "@timestamp":
+                  type: date
+                metricset:
+                  type: keyword
+                  dimension: true
+                k8s:
+                  properties:
+                    pod:
+                      properties:
+                        uid:
+                          type: keyword
+                          dimension: true
+                        name:
+                          type: keyword
+                        ip:
+                          type: ip
+                        network:
+                          properties:
+                            tx:
+                              type: long
+                            rx:
+                              type: long
+  - do:
+      bulk:
+        refresh: true
+        index: test_index
+        body:
+          - '{"index": {}}'
+          - '{"@timestamp": "2021-04-28T18:50:04.467Z", "metricset": "pod", "k8s": {"pod": {"name": "cat", "uid":"947e4ced-1786-4e53-9e0c-5c447e959507", "ip": "10.10.55.1", "network": {"tx": 2001818691, "rx": 802133794}}}}'
+          - '{"index": {}}'
+          - '{"@timestamp": "2021-04-28T18:50:24.467Z", "metricset": "pod", "k8s": {"pod": {"name": "cat", "uid":"947e4ced-1786-4e53-9e0c-5c447e959507", "ip": "10.10.55.1", "network": {"tx": 2005177954, "rx": 801479970}}}}'
+          - '{"index": {}}'
+          - '{"@timestamp": "2021-04-28T18:50:44.467Z", "metricset": "pod", "k8s": {"pod": {"name": "cat", "uid":"947e4ced-1786-4e53-9e0c-5c447e959507", "ip": "10.10.55.1", "network": {"tx": 2006223737, "rx": 802337279}}}}'
+          - '{"index": {}}'
+          - '{"@timestamp": "2021-04-28T18:51:04.467Z", "metricset": "pod", "k8s": {"pod": {"name": "cat", "uid":"947e4ced-1786-4e53-9e0c-5c447e959507", "ip": "10.10.55.2", "network": {"tx": 2012916202, "rx": 803685721}}}}'
+          - '{"index": {}}'
+          - '{"@timestamp": "2021-04-28T18:50:03.142Z", "metricset": "pod", "k8s": {"pod": {"name": "dog", "uid":"df3145b3-0563-4d3b-a0f7-897eb2876ea9", "ip": "10.10.55.3", "network": {"tx": 1434521831, "rx": 530575198}}}}'
+          - '{"index": {}}'
+          - '{"@timestamp": "2021-04-28T18:50:23.142Z", "metricset": "pod", "k8s": {"pod": {"name": "dog", "uid":"df3145b3-0563-4d3b-a0f7-897eb2876ea9", "ip": "10.10.55.3", "network": {"tx": 1434577921, "rx": 530600088}}}}'
+          - '{"index": {}}'
+          - '{"@timestamp": "2021-04-28T18:50:53.142Z", "metricset": "pod", "k8s": {"pod": {"name": "dog", "uid":"df3145b3-0563-4d3b-a0f7-897eb2876ea9", "ip": "10.10.55.3", "network": {"tx": 1434587694, "rx": 530604797}}}}'
+          - '{"index": {}}'
+          - '{"@timestamp": "2021-04-28T18:51:03.142Z", "metricset": "pod", "k8s": {"pod": {"name": "dog", "uid":"df3145b3-0563-4d3b-a0f7-897eb2876ea9", "ip": "10.10.55.3", "network": {"tx": 1434595272, "rx": 530605511}}}}'
+          - '{"index": {}}'
+          - '{"@timestamp": "2021-04-28T18:50:03.142Z", "metricset": "pod", "k8s": {"pod": {"name": "cow", "uid":"1c4fc7b8-93b7-4ba8-b609-2a48af2f8e39", "ip": "10.10.55.4", "network": {"tx": 1434521831, "rx": 530575198}}}}'
+          - '{"index": {}}'
+          - '{"@timestamp": "2021-04-28T18:50:23.142Z", "metricset": "pod", "k8s": {"pod": {"name": "cow", "uid":"1c4fc7b8-93b7-4ba8-b609-2a48af2f8e39", "ip": "10.10.55.4", "network": {"tx": 1434577921, "rx": 530600088}}}}'
+          - '{"index": {}}'
+          - '{"@timestamp": "2021-04-28T18:50:53.142Z", "metricset": "pod", "k8s": {"pod": {"name": "cow", "uid":"1c4fc7b8-93b7-4ba8-b609-2a48af2f8e39", "ip": "10.10.55.4", "network": {"tx": 1434587694, "rx": 530604797}}}}'
+          - '{"index": {}}'
+          - '{"@timestamp": "2021-04-28T18:51:03.142Z", "metricset": "pod", "k8s": {"pod": {"name": "cow", "uid":"1c4fc7b8-93b7-4ba8-b609-2a48af2f8e39", "ip": "10.10.55.4", "network": {"tx": 1434595272, "rx": 530605511}}}}'
+
+
+  # Wait for green
+  - do:
+      cluster.health:
+        wait_for_status: green
+
+  # Take snapshot
+  - do:
+      snapshot.create:
+        repository: test_restore_repository
+        snapshot: test_restore_tsdb
+        wait_for_completion: true
+
+  - match: { snapshot.snapshot: test_restore_tsdb }
+  - match: { snapshot.state : SUCCESS }
+  - match: { snapshot.shards.successful: 2 }
+  - match: { snapshot.shards.failed : 0 }
+  - is_true: snapshot.version
+  - gt: { snapshot.version_id: 0}
+
+  # Close index
+  - do:
+      indices.close:
+        index : test_index
+      allowed_warnings:
+        - "the default value for the ?wait_for_active_shards parameter will change from '0' to 'index-setting' in version 8; specify '?wait_for_active_shards=index-setting' to adopt the future default behaviour, or '?wait_for_active_shards=0' to preserve today's behaviour"
+
+  # Restore index
+  - do:
+      snapshot.restore:
+        repository: test_restore_repository
+        snapshot: test_restore_tsdb
+        wait_for_completion: true
+
+  # Check recovery stats
+  - do:
+      indices.recovery:
+        index: test_index
+
+  - match: { test_index.shards.0.type: SNAPSHOT }
+  - match: { test_index.shards.0.stage: DONE }
+  - match: { test_index.shards.0.index.files.recovered: 1}
+  - gt:    { test_index.shards.0.index.size.recovered_in_bytes: 0}
+
+  # Remove our snapshot
+  - do:
+      snapshot.delete:
+        repository: test_restore_repository
+        snapshot: test_restore_tsdb
+
+  # Remove our repository
+  - do:
+     snapshot.delete_repository:
+       repository: test_restore_repository
+
+  - do:
+      search:
+        index: test_index
+        body:
+          fields:
+            - field: _tsid
+          query:
+            query_string:
+              query: '+@timestamp:"2021-04-28T18:51:04.467Z" +k8s.pod.name:cat'
+
+  - match: {hits.total.value: 1}
+  - match: {hits.hits.0.fields._tsid: [{k8s.pod.uid: 947e4ced-1786-4e53-9e0c-5c447e959507, metricset: pod}]}

qa/full-cluster-restart/src/test/java/org/elasticsearch/upgrades/FullClusterRestartIT.java

@@ -26,6 +26,7 @@
 import org.elasticsearch.core.Booleans;
 import org.elasticsearch.core.CheckedFunction;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.rest.action.admin.indices.RestPutIndexTemplateAction;
 import org.elasticsearch.test.NotEqualMessageBuilder;
 import org.elasticsearch.test.XContentTestUtils;
@@ -45,6 +46,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -151,7 +153,7 @@ public void testSearch() throws Exception {
         assertStoredBinaryFields(count);
     }
 
-    public void testNewReplicasWork() throws Exception {
+    public void testNewReplicas() throws Exception {
         if (isRunningAgainstOldCluster()) {
             XContentBuilder mappingsAndSettings = jsonBuilder();
             mappingsAndSettings.startObject();
@@ -189,7 +191,7 @@ public void testNewReplicasWork() throws Exception {
             logger.debug("--> creating [{}] replicas for index [{}]", numReplicas, index);
             Request setNumberOfReplicas = new Request("PUT", "/" + index + "/_settings");
             setNumberOfReplicas.setJsonEntity("{ \"index\": { \"number_of_replicas\" : " + numReplicas + " }}");
-            Response response = client().performRequest(setNumberOfReplicas);
+            client().performRequest(setNumberOfReplicas);
 
             ensureGreenLongWait(index);
 
@@ -210,6 +212,91 @@ public void testNewReplicasWork() throws Exception {
         }
     }
 
+    public void testSearchTimeSeriesMode() throws Exception {
+        assumeTrue("time series mode introduced in 8.0.0 to be backported to 7.15.0", getOldClusterVersion().onOrAfter(Version.V_8_0_0));
+        int numDocs;
+        if (isRunningAgainstOldCluster()) {
+            numDocs = createTimeSeriesModeIndex(1);
+        } else {
+            numDocs = countOfIndexedRandomDocuments();
+        }
+        assertCountAll(numDocs);
+    }
+
+    public void testNewReplicasTimeSeriesMode() throws Exception {
+        assumeTrue("time series mode introduced in 8.0.0 to be backported to 7.15.0", getOldClusterVersion().onOrAfter(Version.V_8_0_0));
+        if (isRunningAgainstOldCluster()) {
+            createTimeSeriesModeIndex(0);
+        } else {
+            final int numReplicas = 1;
+            final long startTime = System.currentTimeMillis();
+            logger.debug("--> creating [{}] replicas for index [{}]", numReplicas, index);
+            Request setNumberOfReplicas = new Request("PUT", "/" + index + "/_settings");
+            setNumberOfReplicas.setJsonEntity("{ \"index\": { \"number_of_replicas\" : " + numReplicas + " }}");
+            client().performRequest(setNumberOfReplicas);
+
+            ensureGreenLongWait(index);
+
+            logger.debug("--> index [{}] is green, took [{}] ms", index, (System.currentTimeMillis() - startTime));
+            Map<String, Object> recoverRsp = entityAsMap(client().performRequest(new Request("GET", "/" + index + "/_recovery")));
+            logger.debug("--> recovery status:\n{}", recoverRsp);
+
+            Set<Integer> counts = new HashSet<>();
+            for (String node : dataNodes(index, client())) {
+                Request search = new Request("GET", "/" + index + "/_search");
+                search.addParameter("preference", "_only_nodes:" + node);
+                Map<String, Object> responseBody = entityAsMap(client().performRequest(search));
+                assertNoFailures(responseBody);
+                int hits = extractTotalHits(responseBody);
+                counts.add(hits);
+            }
+            assertEquals("All nodes should have a consistent number of documents", 1, counts.size());
+        }
+    }
+
+    private int createTimeSeriesModeIndex(int replicas) throws IOException {
+        XContentBuilder mappingsAndSettings = jsonBuilder();
+        mappingsAndSettings.startObject();
+        {
+            mappingsAndSettings.startObject("settings");
+            mappingsAndSettings.field("number_of_shards", 1);
+            mappingsAndSettings.field("number_of_replicas", replicas);
+            mappingsAndSettings.field("mode", "time_series");
+            mappingsAndSettings.endObject();
+        }
+        {
+            mappingsAndSettings.startObject("mappings");
+            mappingsAndSettings.startObject("properties");
+            {
+                mappingsAndSettings.startObject("@timestamp").field("type", "date").endObject();
+                mappingsAndSettings.startObject("dim").field("type", "keyword").field("dimension", true).endObject();
+            }
+            mappingsAndSettings.endObject();
+            mappingsAndSettings.endObject();
+        }
+        mappingsAndSettings.endObject();
+
+        Request createIndex = new Request("PUT", "/" + index);
+        createIndex.setJsonEntity(Strings.toString(mappingsAndSettings));
+        client().performRequest(createIndex);
+
+        int numDocs = randomIntBetween(2000, 3000);
+        long basetime = DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER.parseMillis("2021-01-01T00:00:00Z");
+        indexRandomDocuments(
+            numDocs,
+            true,
+            true,
+            i -> JsonXContent.contentBuilder()
+                .startObject()
+                .field("@timestamp", basetime + TimeUnit.MINUTES.toMillis(i))
+                .field("dim", "value")
+                .endObject()
+        );
+        logger.info("Refreshing [{}]", index);
+        client().performRequest(new Request("POST", "/" + index + "/_refresh"));
+        return numDocs;
+    }
+
     public void testClusterState() throws Exception {
         if (isRunningAgainstOldCluster()) {
             XContentBuilder mappingsAndSettings = jsonBuilder();
@@ -459,14 +546,18 @@ public void testRollover() throws IOException {
         assertEquals(expectedCount, extractTotalHits(count));
     }
 
+    void assertCountAll(int count) throws IOException {
+        Map<String, Object> response = entityAsMap(client().performRequest(new Request("GET", "/" + index + "/_search")));
+        assertNoFailures(response);
+        int numDocs = extractTotalHits(response);
+        logger.info("Found {} in old index", numDocs);
+        assertEquals(count, numDocs);
+    }
+
     void assertBasicSearchWorks(int count) throws IOException {
         logger.info("--> testing basic search");
         {
-            Map<String, Object> response = entityAsMap(client().performRequest(new Request("GET", "/" + index + "/_search")));
-            assertNoFailures(response);
-            int numDocs = extractTotalHits(response);
-            logger.info("Found {} in old index", numDocs);
-            assertEquals(count, numDocs);
+            assertCountAll(count);
         }
 
         logger.info("--> testing basic search with sort");
@@ -1136,8 +1227,6 @@ private void checkSnapshot(final String snapshotName, final int count, final Ver
         }
     }
 
-    // TODO tests for upgrades after shrink. We've had trouble with shrink in the past.
-
     private void indexRandomDocuments(
             final int count,
             final boolean flushAllowed,
@@ -1629,5 +1718,4 @@ public static void assertNumHits(String index, int numHits, int totalShards) thr
         assertThat(XContentMapValues.extractValue("_shards.successful", resp), equalTo(totalShards));
         assertThat(extractTotalHits(resp), equalTo(numHits));
     }
-
 }