OTLP: store units in mappings (#134709)

This is enabled by adding a new `dynamic_template_params` option to the bulk request header. It can be used in combination with the `dynamic_templates` option to provide parameters that can be used within a dynamic template.

The dynamic template can control which parameters it accepts and can provide a default value in case clients don't provide the parameters in the bulk request. If no default value is provided, and the bulk request also doesn't specify the parameter, the value is kept as it is defined in the template (including braces).

For example, the dynamic template can define a `unit` placeholder with an empty string as the default value.
```
{
  "dynamic_templates": [
    {
      "gauge_double": {
        "mapping": {
          "type": "double",
          "time_series_metric": "gauge",
          "meta": {
            "unit": "{unit:}"
          }
        }
      }
    }
  ]
}
```

Clients can then set provide a value for the unit in the bulk request header like so:
```
{"create": {"dynamic_templates":{"metrics.foo.bar":"gauge_double"},"dynamic_template_params":{"metrics.foo.bar":{"unit": "By"}}}}
```

Author: felixbarny

docs/changelog/134709.yaml

@@ -0,0 +1,5 @@
+pr: 134709
+summary: "OTLP: store units in mappings"
+area: Mapping
+type: enhancement
+issues: []

server/src/internalClusterTest/java/org/elasticsearch/action/bulk/IncrementalBulkIT.java

@@ -203,7 +203,7 @@ public void testIncrementalBulkHighWatermarkBackOff() throws Exception {
         AtomicBoolean nextPage = new AtomicBoolean(false);
 
         ArrayList<IncrementalBulkService.Handler> handlers = new ArrayList<>();
-        for (int i = 0; i < 4; ++i) {
+        for (int i = 0; i < 5; ++i) {
             ArrayList<DocWriteRequest<?>> requests = new ArrayList<>();
             add512BRequests(requests, index);
             IncrementalBulkService.Handler handler = incrementalBulkService.newBulkRequest();
@@ -230,6 +230,8 @@ public void testIncrementalBulkHighWatermarkBackOff() throws Exception {
         // Test that a request larger than SPLIT_BULK_HIGH_WATERMARK_SIZE (1KB) is throttled
         add512BRequests(requestsThrottle, index);
         add512BRequests(requestsThrottle, index);
+        // Ensure we'll be above SPLIT_BULK_HIGH_WATERMARK
+        assertThat(indexingPressure.stats().getCurrentCombinedCoordinatingAndPrimaryBytes() + 1024, greaterThan(4096L));
 
         CountDownLatch finishLatch = new CountDownLatch(1);
         blockWriteCoordinationPool(threadPool, finishLatch);

server/src/internalClusterTest/java/org/elasticsearch/monitor/metrics/NodeIndexingMetricsIT.java

@@ -796,7 +796,7 @@ public void testIncrementalBulkHighWatermarkSplitMetrics() throws Exception {
         AtomicBoolean nextPage = new AtomicBoolean(false);
 
         ArrayList<IncrementalBulkService.Handler> handlers = new ArrayList<>();
-        for (int i = 0; i < 4; ++i) {
+        for (int i = 0; i < 5; ++i) {
             ArrayList<DocWriteRequest<?>> requests = new ArrayList<>();
             add512BRequests(requests, index);
             IncrementalBulkService.Handler handler = incrementalBulkService.newBulkRequest();
@@ -838,6 +838,8 @@ public void testIncrementalBulkHighWatermarkSplitMetrics() throws Exception {
         // Test that a request larger than SPLIT_BULK_HIGH_WATERMARK_SIZE (1KB) is throttled
         add512BRequests(requestsThrottle, index);
         add512BRequests(requestsThrottle, index);
+        // Ensure we'll be above SPLIT_BULK_HIGH_WATERMARK
+        assertThat(indexingPressure.stats().getCurrentCombinedCoordinatingAndPrimaryBytes() + 1024, greaterThan(4096L));
 
         CountDownLatch finishLatch = new CountDownLatch(1);
         blockWriteCoordinationPool(threadPool, finishLatch);

server/src/main/java/org/elasticsearch/action/bulk/BulkRequestParser.java

@@ -68,6 +68,7 @@ public final class BulkRequestParser {
     private static final ParseField REQUIRE_DATA_STREAM = new ParseField(DocWriteRequest.REQUIRE_DATA_STREAM);
     private static final ParseField LIST_EXECUTED_PIPELINES = new ParseField(DocWriteRequest.LIST_EXECUTED_PIPELINES);
     private static final ParseField DYNAMIC_TEMPLATES = new ParseField("dynamic_templates");
+    private static final ParseField DYNAMIC_TEMPLATE_PARAMS = new ParseField("dynamic_template_params");
 
     // TODO: Remove this parameter once the BulkMonitoring endpoint has been removed
     // for CompatibleApi V7 this means to deprecate on type, for V8+ it means to throw an error
@@ -359,6 +360,7 @@ private boolean parseActionLine(BytesReference data, int from, int to) throws IO
                 boolean requireAlias = defaultRequireAlias != null && defaultRequireAlias;
                 boolean requireDataStream = defaultRequireDataStream != null && defaultRequireDataStream;
                 Map<String, String> dynamicTemplates = Map.of();
+                Map<String, Map<String, String>> dynamicTemplatesParms = Map.of();
 
                 // at this stage, next token can either be END_OBJECT (and use default index and type, with auto generated id)
                 // or START_OBJECT which will have another set of parameters
@@ -427,19 +429,22 @@ private boolean parseActionLine(BytesReference data, int from, int to) throws IO
                             && DYNAMIC_TEMPLATES.match(currentFieldName, parser.getDeprecationHandler())) {
                                 dynamicTemplates = parser.mapStrings();
                             } else if (token == XContentParser.Token.START_OBJECT
-                                && SOURCE.match(currentFieldName, parser.getDeprecationHandler())) {
-                                    currentFetchSourceContext = FetchSourceContext.fromXContent(parser);
-                                } else if (token != XContentParser.Token.VALUE_NULL) {
-                                    throw new IllegalArgumentException(
-                                        "Malformed action/metadata line ["
-                                            + line
-                                            + "], expected a simple value for field ["
-                                            + currentFieldName
-                                            + "] but found ["
-                                            + token
-                                            + "]"
-                                    );
-                                }
+                                && DYNAMIC_TEMPLATE_PARAMS.match(currentFieldName, parser.getDeprecationHandler())) {
+                                    dynamicTemplatesParms = parser.map(HashMap::new, XContentParser::mapStrings);
+                                } else if (token == XContentParser.Token.START_OBJECT
+                                    && SOURCE.match(currentFieldName, parser.getDeprecationHandler())) {
+                                        currentFetchSourceContext = FetchSourceContext.fromXContent(parser);
+                                    } else if (token != XContentParser.Token.VALUE_NULL) {
+                                        throw new IllegalArgumentException(
+                                            "Malformed action/metadata line ["
+                                                + line
+                                                + "], expected a simple value for field ["
+                                                + currentFieldName
+                                                + "] but found ["
+                                                + token
+                                                + "]"
+                                        );
+                                    }
                     }
                 } else if (token != XContentParser.Token.END_OBJECT) {
                     throw new IllegalArgumentException(
@@ -462,6 +467,11 @@ private boolean parseActionLine(BytesReference data, int from, int to) throws IO
                             "Delete request in line [" + line + "] does not accept " + DYNAMIC_TEMPLATES.getPreferredName()
                         );
                     }
+                    if (dynamicTemplatesParms.isEmpty() == false) {
+                        throw new IllegalArgumentException(
+                            "Update request in line [" + line + "] does not accept " + DYNAMIC_TEMPLATE_PARAMS.getPreferredName()
+                        );
+                    }
                     currentRequest = new DeleteRequest(index).id(id)
                         .routing(routing)
                         .version(version)
@@ -480,6 +490,7 @@ private boolean parseActionLine(BytesReference data, int from, int to) throws IO
                             .setIfSeqNo(ifSeqNo)
                             .setIfPrimaryTerm(ifPrimaryTerm)
                             .setDynamicTemplates(dynamicTemplates)
+                            .setDynamicTemplateParams(dynamicTemplatesParms)
                             .setRequireAlias(requireAlias)
                             .setRequireDataStream(requireDataStream)
                             .setListExecutedPipelines(currentListExecutedPipelines)
@@ -508,6 +519,11 @@ private boolean parseActionLine(BytesReference data, int from, int to) throws IO
                                 "Update request in line [" + line + "] does not accept " + DYNAMIC_TEMPLATES.getPreferredName()
                             );
                         }
+                        if (dynamicTemplatesParms.isEmpty() == false) {
+                            throw new IllegalArgumentException(
+                                "Update request in line [" + line + "] does not accept " + DYNAMIC_TEMPLATE_PARAMS.getPreferredName()
+                            );
+                        }
                         UpdateRequest updateRequest = new UpdateRequest().index(index)
                             .id(id)
                             .routing(routing)

server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -433,6 +433,7 @@ static boolean executeBulkItemRequest(
                 request.getContentType(),
                 request.routing(),
                 request.getDynamicTemplates(),
+                request.getDynamicTemplateParams(),
                 request.getIncludeSourceOnError(),
                 meteringParserDecorator,
                 request.tsid()
@@ -755,6 +756,7 @@ private static Engine.Result performOpOnReplica(
                     indexRequest.getContentType(),
                     indexRequest.routing(),
                     Map.of(),
+                    Map.of(),
                     true,
                     XContentMeteringParserDecorator.NOOP,
                     indexRequest.tsid()

server/src/main/java/org/elasticsearch/action/bulk/TransportSimulateBulkAction.java

@@ -206,6 +206,7 @@ private ValidationResult validateMappings(
             request.getContentType(),
             request.routing(),
             request.getDynamicTemplates(),
+            request.getDynamicTemplateParams(),
             request.getIncludeSourceOnError(),
             XContentMeteringParserDecorator.NOOP,
             request.tsid()

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

@@ -79,6 +79,9 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
     private static final TransportVersion PIPELINES_HAVE_RUN_FIELD_ADDED = TransportVersions.V_8_10_X;
     private static final TransportVersion INDEX_REQUEST_INCLUDE_TSID = TransportVersion.fromName("index_request_include_tsid");
     private static final TransportVersion INDEX_SOURCE = TransportVersion.fromName("index_source");
+    static final TransportVersion INGEST_REQUEST_DYNAMIC_TEMPLATE_PARAMS = TransportVersion.fromName(
+        "ingest_request_dynamic_template_params"
+    );
 
     private static final Supplier<String> ID_GENERATOR = UUIDs::base64UUID;
 
@@ -147,6 +150,7 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
     private long ifPrimaryTerm = UNASSIGNED_PRIMARY_TERM;
 
     private Map<String, String> dynamicTemplates = Map.of();
+    private Map<String, Map<String, String>> dynamicTemplateParams = Map.of();
 
     /**
      * rawTimestamp field is used on the coordinate node, it doesn't need to be serialised.
@@ -234,6 +238,10 @@ public IndexRequest(@Nullable ShardId shardId, StreamInput in) throws IOExceptio
         if (in.getTransportVersion().supports(INDEX_REQUEST_INCLUDE_TSID)) {
             tsid = in.readBytesRefOrNullIfEmpty();
         }
+
+        if (in.getTransportVersion().supports(INGEST_REQUEST_DYNAMIC_TEMPLATE_PARAMS)) {
+            dynamicTemplateParams = in.readMap(StreamInput::readString, i -> i.readMap(StreamInput::readString));
+        }
     }
 
     public IndexRequest() {
@@ -823,6 +831,9 @@ private void writeBody(StreamOutput out) throws IOException {
         if (out.getTransportVersion().supports(INDEX_REQUEST_INCLUDE_TSID)) {
             out.writeBytesRef(tsid);
         }
+        if (out.getTransportVersion().supports(INGEST_REQUEST_DYNAMIC_TEMPLATE_PARAMS)) {
+            out.writeMap(dynamicTemplateParams, StreamOutput::writeString, (o, v) -> o.writeMap(v, StreamOutput::writeString));
+        }
     }
 
     @Override
@@ -982,6 +993,15 @@ public Map<String, String> getDynamicTemplates() {
         return dynamicTemplates;
     }
 
+    public IndexRequest setDynamicTemplateParams(Map<String, Map<String, String>> dynamicTemplateParams) {
+        this.dynamicTemplateParams = dynamicTemplateParams;
+        return this;
+    }
+
+    public Map<String, Map<String, String>> getDynamicTemplateParams() {
+        return dynamicTemplateParams;
+    }
+
     public Object getRawTimestamp() {
         return rawTimestamp;
     }

server/src/main/java/org/elasticsearch/index/mapper/DocumentParserContext.java

@@ -922,6 +922,28 @@ public final DynamicTemplate findDynamicTemplate(String fieldName, DynamicTempla
         return null;
     }
 
+    /**
+     * Provides parameters for the {@link DynamicTemplate} returned by
+     * {@link #findDynamicTemplate(String, DynamicTemplate.XContentFieldType)}.
+     * The {@link DynamicTemplate} can use placeholders in its definition that
+     * will be replaced by the values returned by this method.
+     * For example, a dynamic template may contain a snippet like this:
+     * <pre>
+     * "meta": {
+     *   "unit": "{{unit}}"
+     * }
+     * </pre>
+     * When applying the dynamic template to a field, the placeholder <code>{{unit}}</code>
+     * will be replaced by the value returned by this method for the key <code>unit</code>.
+     *
+     * @param fieldName the name of the field
+     * @return a map of parameter names to values; empty map if no parameters exist
+     */
+    public final Map<String, String> getDynamicTemplateParams(String fieldName) {
+        final String pathAsString = path().pathAsText(fieldName);
+        return sourceToParse.dynamicTemplateParams().getOrDefault(pathAsString, Map.of());
+    }
+
     // XContentParser that wraps an existing parser positioned on a value,
     // and a field name, and returns a stream that looks like { 'field' : 'value' }
     private static class CopyToParser extends FilterXContentParserWrapper {

server/src/main/java/org/elasticsearch/index/mapper/DynamicFieldsBuilder.java

@@ -237,7 +237,7 @@ private static boolean applyMatchingTemplate(
         String dynamicType = dynamicTemplate.isRuntimeMapping() ? matchType.defaultRuntimeMappingType() : matchType.defaultMappingType();
 
         String mappingType = dynamicTemplate.mappingType(dynamicType);
-        Map<String, Object> mapping = dynamicTemplate.mappingForName(name, dynamicType);
+        Map<String, Object> mapping = dynamicTemplate.mappingForName(name, dynamicType, context.getDynamicTemplateParams(name));
         if (dynamicTemplate.isRuntimeMapping()) {
             MappingParserContext parserContext = context.dynamicTemplateParserContext(dateFormatter);
             RuntimeField.Parser parser = parserContext.runtimeFieldParser(mappingType);
@@ -262,7 +262,7 @@ private static Mapper.Builder findTemplateBuilderForObject(DocumentParserContext
         }
         String dynamicType = matchType.defaultMappingType();
         String mappingType = dynamicTemplate.mappingType(dynamicType);
-        Map<String, Object> mapping = dynamicTemplate.mappingForName(name, dynamicType);
+        Map<String, Object> mapping = dynamicTemplate.mappingForName(name, dynamicType, context.getDynamicTemplateParams(name));
         return parseDynamicTemplateMapping(name, mappingType, mapping, null, context);
     }