[8.x] Add LogsDB option to route on sort fields (#116687) (#119113)

* Add LogsDB option to route on sort fields (#116687)

* Add LogsDB option to route on sort fields

* fix encoding

* Update docs/changelog/116687.yaml

* tests

* tests

* tests

* fix mode

* tests

* tests

* tests

* add test

* fix test

* sync

* updates from review

* test fixes

* test fixes

* test fixes

* Move logic to SyntheticSourceIndexSettingsProvider

* fix test

* sync

* merge, no fallback

* comments

* fix test

* address comments

* address comments

* address comments

* Update x-pack/plugin/logsdb/src/main/java/org/elasticsearch/xpack/logsdb/LogsdbIndexModeSettingsProvider.java

Co-authored-by: Martijn van Groningen <[email protected]>

* [CI] Auto commit changes from spotless

* update tests

* [CI] Auto commit changes from spotless

* update tests

* fix rest compat tests

---------

Co-authored-by: Martijn van Groningen <[email protected]>
Co-authored-by: elasticsearchmachine <[email protected]>
(cherry picked from commit d80cbdd)

# Conflicts:
#	rest-api-spec/build.gradle
#	server/src/main/java/org/elasticsearch/common/TimeBasedKOrderedUUIDGenerator.java

* Update LogsIndexingIT.java

Author: kkrik-es

docs/changelog/116687.yaml

@@ -0,0 +1,5 @@
+pr: 116687
+summary: Add LogsDB option to route on sort fields
+area: Logs
+type: enhancement
+issues: []

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/logsdb/10_settings.yml

@@ -514,6 +514,48 @@ routing path not allowed in logs mode:
   - match: { error.type: "illegal_argument_exception" }
   - match: { error.reason: "[index.routing_path] requires [index.mode=time_series]" }
 
+---
+routing path allowed in logs mode with routing on sort fields:
+  - requires:
+      cluster_features: [ "routing.logsb_route_on_sort_fields" ]
+      reason: introduction of route on index sorting fields
+
+  - do:
+      indices.create:
+        index: test
+        body:
+          settings:
+            index:
+              mode: logsdb
+              number_of_replicas: 0
+              number_of_shards: 2
+              routing_path: [ host.name, agent_id ]
+              logsdb:
+                route_on_sort_fields: true
+          mappings:
+            properties:
+              "@timestamp":
+                type: date
+              host.name:
+                type: keyword
+              agent_id:
+                type: keyword
+              process_id:
+                type: integer
+              http_method:
+                type: keyword
+              message:
+                type: text
+
+  - do:
+      indices.get_settings:
+        index: test
+
+  - is_true: test
+  - match: { test.settings.index.mode: logsdb }
+  - match: { test.settings.index.logsdb.route_on_sort_fields: "true" }
+  - match: { test.settings.index.routing_path: [ host.name, agent_id ] }
+
 ---
 start time not allowed in logs mode:
   - requires:

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/tsdb/20_mapping.yml

@@ -127,7 +127,7 @@ exact match object type:
       reason: routing_path error message updated in 8.14.0
 
   - do:
-      catch: '/All fields that match routing_path must be configured with \[time_series_dimension: true\] or flattened fields with a list of dimensions in \[time_series_dimensions\] and without the \[script\] parameter. \[dim\] was \[object\]./'
+      catch: '/All fields that match routing_path must be .*flattened fields.* \[dim\] was \[object\]./'
       indices.create:
         index: tsdb_index
         body:

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/tsdb/25_id_generation.yml

@@ -427,7 +427,7 @@ delete over _bulk:
   - match: {items.0.delete.result: deleted}
   - match: {items.1.delete.result: deleted}
   - match: {items.2.delete.status: 404}
-  - match: {items.2.delete.error.reason: "invalid id [not found ++ not found] for index [id_generation_test] in time series mode"}
+  - match: {items.2.delete.error.reason: '/invalid\ id\ \[not\ found\ \+\+\ not\ found\]\ for\ index\ \[id_generation_test\]\ in\ time.series\ mode/'}
 
 ---
 routing_path matches deep object:

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/tsdb/80_index_resize.yml

@@ -95,7 +95,7 @@ split:
       reason: tsdb indexing changed in 8.2.0
 
   - do:
-      catch: /index-split is not supported because the destination index \[test\] is in time series mode/
+      catch: /index-split is not supported because the destination index \[test\] is in time.series mode/
       indices.split:
         index: test
         target: test_split

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/tsdb/90_unsupported_operations.yml

@@ -75,7 +75,7 @@ index with routing:
       reason: tsdb indexing changed in 8.2.0
 
   - do:
-      catch: /specifying routing is not supported because the destination index \[test\] is in time series mode/
+      catch: /specifying routing is not supported because the destination index \[test\] is in time.series mode/
       index:
         index:   test
         routing: foo
@@ -104,7 +104,7 @@ index with routing over _bulk:
         body:
           - '{"index": {"routing": "foo"}}'
           - '{"@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}}}}'
-  - match: {items.0.index.error.reason: "specifying routing is not supported because the destination index [test] is in time series mode"}
+  - match: {items.0.index.error.reason: '/specifying\ routing\ is\ not\ supported\ because\ the\ destination\ index\ \[test\]\ is\ in\ time.series\ mode/'}
 
 ---
 noop update:
@@ -120,7 +120,7 @@ noop update:
   - length: {hits.hits: 1}
 
   - do:
-      catch: /update is not supported because the destination index \[test\] is in time series mode/
+      catch: /update is not supported because the destination index \[test\] is in time.series mode/
       update:
         index:   test
         id:      "1"
@@ -136,7 +136,7 @@ regular update:
 
   # We fail even though the document isn't found.
   - do:
-      catch: /update is not supported because the destination index \[test\] is in time series mode/
+      catch: /update is not supported because the destination index \[test\] is in time.series mode/
       update:
         index:   test
         id:      "1"
@@ -165,7 +165,7 @@ update over _bulk:
         body:
           - '{"update": {"_id": 1}}'
           - '{"doc":{"@timestamp": "2021-04-28T18:03:24.467Z", "metricset": "pod", "k8s": {"pod": {"name": "dog", "uid":"df3145b3-0563-4d3b-a0f7-897eb2876ea9", "ip": "10.10.55.3", "network": {"tx": 1434595272, "rx": 530605511}}}}}'
-  - match: {items.0.update.error.reason: "update is not supported because the destination index [test] is in time series mode"}
+  - match: {items.0.update.error.reason: '/update\ is\ not\ supported\ because\ the\ destination\ index\ \[test\]\ is\ in\ time.series\ mode/'}
 
 ---
 search with routing:
@@ -175,7 +175,7 @@ search with routing:
 
   # We fail even though the document isn't found.
   - do:
-      catch: /searching with a specified routing is not supported because the destination index \[test\] is in time series mode/
+      catch: /searching with a specified routing is not supported because the destination index \[test\] is in time.series mode/
       search:
         index:   test
         routing: rrrr

server/src/main/java/org/elasticsearch/action/index/IndexRequest.java

@@ -51,6 +51,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
+import java.util.OptionalInt;
 import java.util.function.Supplier;
 
 import static org.elasticsearch.action.ValidateActions.addValidationError;
@@ -78,7 +79,6 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
     private static final TransportVersion PIPELINES_HAVE_RUN_FIELD_ADDED = TransportVersions.V_8_10_X;
 
     private static final Supplier<String> ID_GENERATOR = UUIDs::base64UUID;
-    private static final Supplier<String> K_SORTED_TIME_BASED_ID_GENERATOR = UUIDs::base64TimeBasedKOrderedUUID;
 
     /**
      * Max length of the source document to include into string()
@@ -705,9 +705,18 @@ public void autoGenerateId() {
     }
 
     public void autoGenerateTimeBasedId() {
+        autoGenerateTimeBasedId(OptionalInt.empty());
+    }
+
+    /**
+     *  Set the {@code #id()} to an automatically generated one, optimized for storage (compression) efficiency.
+     *  If a routing hash is passed, it is included in the generated id starting at 9 bytes before the end.
+     * @param hash optional routing hash value, used to route requests by id to the right shard.
+     */
+    public void autoGenerateTimeBasedId(OptionalInt hash) {
         assertBeforeGeneratingId();
         autoGenerateTimestamp();
-        id(K_SORTED_TIME_BASED_ID_GENERATOR.get());
+        id(UUIDs.base64TimeBasedKOrderedUUIDWithHash(hash));
     }
 
     private void autoGenerateTimestamp() {

server/src/main/java/org/elasticsearch/cluster/routing/IndexRouting.java

@@ -39,6 +39,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.OptionalInt;
 import java.util.Set;
 import java.util.function.IntConsumer;
 import java.util.function.IntSupplier;
@@ -54,6 +55,7 @@ public abstract class IndexRouting {
 
     static final NodeFeature BOOLEAN_ROUTING_PATH = new NodeFeature("routing.boolean_routing_path");
     static final NodeFeature MULTI_VALUE_ROUTING_PATH = new NodeFeature("routing.multi_value_routing_path");
+    static final NodeFeature LOGSB_ROUTE_ON_SORT_FIELDS = new NodeFeature("routing.logsb_route_on_sort_fields");
 
     /**
      * Build the routing from {@link IndexMetadata}.
@@ -164,7 +166,8 @@ private abstract static class IdAndRoutingOnly extends IndexRouting {
 
         @Override
         public void preProcess(IndexRequest indexRequest) {
-            // generate id if not already provided
+            // Generate id if not already provided.
+            // This is needed for routing, so it has to happen in pre-processing.
             final String id = indexRequest.id();
             if (id == null) {
                 if (creationVersion.onOrAfter(IndexVersions.TIME_BASED_K_ORDERED_DOC_ID_BACKPORT) && indexMode == IndexMode.LOGSDB) {
@@ -262,15 +265,20 @@ public void collectSearchShards(String routing, IntConsumer consumer) {
     public static class ExtractFromSource extends IndexRouting {
         private final Predicate<String> isRoutingPath;
         private final XContentParserConfiguration parserConfig;
+        private final IndexMode indexMode;
         private final boolean trackTimeSeriesRoutingHash;
+        private final boolean addIdWithRoutingHash;
         private int hash = Integer.MAX_VALUE;
 
         ExtractFromSource(IndexMetadata metadata) {
             super(metadata);
             if (metadata.isRoutingPartitionedIndex()) {
                 throw new IllegalArgumentException("routing_partition_size is incompatible with routing_path");
             }
-            trackTimeSeriesRoutingHash = metadata.getCreationVersion().onOrAfter(IndexVersions.TIME_SERIES_ROUTING_HASH_IN_ID);
+            indexMode = metadata.getIndexMode();
+            trackTimeSeriesRoutingHash = indexMode == IndexMode.TIME_SERIES
+                && metadata.getCreationVersion().onOrAfter(IndexVersions.TIME_SERIES_ROUTING_HASH_IN_ID);
+            addIdWithRoutingHash = indexMode == IndexMode.LOGSDB;
             List<String> routingPaths = metadata.getRoutingPaths();
             isRoutingPath = Regex.simpleMatcher(routingPaths.toArray(String[]::new));
             this.parserConfig = XContentParserConfiguration.EMPTY.withFiltering(null, Set.copyOf(routingPaths), null, true);
@@ -282,8 +290,13 @@ public boolean matchesField(String fieldName) {
 
         @Override
         public void postProcess(IndexRequest indexRequest) {
+            // Update the request with the routing hash, if needed.
+            // This needs to happen in post-processing, after the routing hash is calculated.
             if (trackTimeSeriesRoutingHash) {
                 indexRequest.routing(TimeSeriesRoutingHashFieldMapper.encode(hash));
+            } else if (addIdWithRoutingHash) {
+                assert hash != Integer.MAX_VALUE;
+                indexRequest.autoGenerateTimeBasedId(OptionalInt.of(hash));
             }
         }
 
@@ -451,12 +464,15 @@ private int idToHash(String id) {
             try {
                 idBytes = Base64.getUrlDecoder().decode(id);
             } catch (IllegalArgumentException e) {
-                throw new ResourceNotFoundException("invalid id [{}] for index [{}] in time series mode", id, indexName);
+                throw new ResourceNotFoundException("invalid id [{}] for index [{}] in " + indexMode.getName() + " mode", id, indexName);
             }
             if (idBytes.length < 4) {
-                throw new ResourceNotFoundException("invalid id [{}] for index [{}] in time series mode", id, indexName);
+                throw new ResourceNotFoundException("invalid id [{}] for index [{}] in " + indexMode.getName() + " mode", id, indexName);
             }
-            return hashToShardId(ByteUtils.readIntLE(idBytes, 0));
+            // For TSDB, the hash is stored as the id prefix.
+            // For LogsDB with routing on sort fields, the routing hash is stored in the range[id.length - 9, id.length - 5] of the id,
+            // see IndexRequest#autoGenerateTimeBasedId.
+            return hashToShardId(ByteUtils.readIntLE(idBytes, addIdWithRoutingHash ? idBytes.length - 9 : 0));
         }
 
         @Override
@@ -470,7 +486,7 @@ public void collectSearchShards(String routing, IntConsumer consumer) {
         }
 
         private String error(String operation) {
-            return operation + " is not supported because the destination index [" + indexName + "] is in time series mode";
+            return operation + " is not supported because the destination index [" + indexName + "] is in " + indexMode.getName() + " mode";
         }
     }
 

server/src/main/java/org/elasticsearch/cluster/routing/RoutingFeatures.java

@@ -20,4 +20,9 @@ public class RoutingFeatures implements FeatureSpecification {
     public Set<NodeFeature> getFeatures() {
         return Set.of(IndexRouting.BOOLEAN_ROUTING_PATH, IndexRouting.MULTI_VALUE_ROUTING_PATH);
     }
+
+    @Override
+    public Set<NodeFeature> getTestFeatures() {
+        return Set.of(IndexRouting.LOGSB_ROUTE_ON_SORT_FIELDS);
+    }
 }

server/src/main/java/org/elasticsearch/common/TimeBasedKOrderedUUIDGenerator.java

@@ -9,8 +9,11 @@
 
 package org.elasticsearch.common;
 
+import org.elasticsearch.common.util.ByteUtils;
+
 import java.nio.ByteBuffer;
 import java.util.Base64;
+import java.util.OptionalInt;
 import java.util.function.Supplier;
 
 /**
@@ -29,6 +32,7 @@
  * The result is a compact base64-encoded string, optimized for efficient compression of the _id field in an inverted index.
  */
 public class TimeBasedKOrderedUUIDGenerator extends TimeBasedUUIDGenerator {
+    static final int SIZE_IN_BYTES = 15;
 
     private static final Base64.Encoder BASE_64_NO_PADDING_URL_ENCODER = Base64.getUrlEncoder().withoutPadding();
 
@@ -42,6 +46,10 @@ public TimeBasedKOrderedUUIDGenerator(
 
     @Override
     public String getBase64UUID() {
+        return getBase64UUID(OptionalInt.empty());
+    }
+
+    public String getBase64UUID(OptionalInt hash) {
         final int sequenceId = sequenceNumber.incrementAndGet() & 0x00FF_FFFF;
 
         // Calculate timestamp to ensure ordering and avoid backward movement in case of time shifts.
@@ -53,7 +61,7 @@ public String getBase64UUID() {
             sequenceId == 0 ? (lastTimestamp, currentTimeMillis) -> Math.max(lastTimestamp, currentTimeMillis) + 1 : Math::max
         );
 
-        final byte[] uuidBytes = new byte[15];
+        final byte[] uuidBytes = new byte[SIZE_IN_BYTES + (hash.isPresent() ? 4 : 0)];
         final ByteBuffer buffer = ByteBuffer.wrap(uuidBytes);
 
         buffer.put((byte) (timestamp >>> 40)); // changes every 35 years
@@ -67,6 +75,13 @@ public String getBase64UUID() {
         assert macAddress.length == 6;
         buffer.put(macAddress, 0, macAddress.length);
 
+        // Copy the hash value if provided
+        if (hash.isPresent()) {
+            byte[] hashBytes = new byte[4];
+            ByteUtils.writeIntLE(hash.getAsInt(), hashBytes, 0);
+            buffer.put(hashBytes, 0, hashBytes.length);
+        }
+
         buffer.put((byte) (sequenceId >>> 16));
 
         // From hereinafter everything is almost like random and does not compress well

server/src/main/java/org/elasticsearch/common/UUIDs.java

@@ -11,6 +11,7 @@
 
 import org.elasticsearch.common.settings.SecureString;
 
+import java.util.OptionalInt;
 import java.util.Random;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;
@@ -23,14 +24,14 @@ public class UUIDs {
     public static final Supplier<Long> DEFAULT_TIMESTAMP_SUPPLIER = System::currentTimeMillis;
     public static final Supplier<Integer> DEFAULT_SEQUENCE_ID_SUPPLIER = sequenceNumber::incrementAndGet;
     public static final Supplier<byte[]> DEFAULT_MAC_ADDRESS_SUPPLIER = MacAddressProvider::getSecureMungedAddress;
-    private static final UUIDGenerator RANDOM_UUID_GENERATOR = new RandomBasedUUIDGenerator();
-    private static final UUIDGenerator TIME_BASED_K_ORDERED_GENERATOR = new TimeBasedKOrderedUUIDGenerator(
+    private static final RandomBasedUUIDGenerator RANDOM_UUID_GENERATOR = new RandomBasedUUIDGenerator();
+    private static final TimeBasedKOrderedUUIDGenerator TIME_BASED_K_ORDERED_GENERATOR = new TimeBasedKOrderedUUIDGenerator(
         DEFAULT_TIMESTAMP_SUPPLIER,
         DEFAULT_SEQUENCE_ID_SUPPLIER,
         DEFAULT_MAC_ADDRESS_SUPPLIER
     );
 
-    private static final UUIDGenerator TIME_UUID_GENERATOR = new TimeBasedUUIDGenerator(
+    private static final TimeBasedUUIDGenerator TIME_UUID_GENERATOR = new TimeBasedUUIDGenerator(
         DEFAULT_TIMESTAMP_SUPPLIER,
         DEFAULT_SEQUENCE_ID_SUPPLIER,
         DEFAULT_MAC_ADDRESS_SUPPLIER
@@ -51,12 +52,8 @@ public static String base64UUID() {
         return TIME_UUID_GENERATOR.getBase64UUID();
     }
 
-    public static String base64TimeBasedKOrderedUUID() {
-        return TIME_BASED_K_ORDERED_GENERATOR.getBase64UUID();
-    }
-
-    public static String base64TimeBasedUUID() {
-        return TIME_UUID_GENERATOR.getBase64UUID();
+    public static String base64TimeBasedKOrderedUUIDWithHash(OptionalInt hash) {
+        return TIME_BASED_K_ORDERED_GENERATOR.getBase64UUID(hash);
     }
 
     /**

server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -183,6 +183,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         IndexSettings.LIFECYCLE_ORIGINATION_DATE_SETTING,
         IndexSettings.LIFECYCLE_PARSE_ORIGINATION_DATE_SETTING,
         IndexSettings.TIME_SERIES_ES87TSDB_CODEC_ENABLED_SETTING,
+        IndexSettings.LOGSDB_ROUTE_ON_SORT_FIELDS,
         IndexSettings.PREFER_ILM_SETTING,
         DataStreamFailureStoreDefinition.FAILURE_STORE_DEFINITION_VERSION_SETTING,
         FieldMapper.SYNTHETIC_SOURCE_KEEP_INDEX_SETTING,