Add support for flexible access pattern to NormalizeForStreamProcessor

docs/changelog/134524.yaml

@@ -0,0 +1,5 @@
+pr: 134524
+summary: Add support for flexible access pattern to `NormalizeForStreamProcessor`
+area: Ingest Node
+type: bug
+issues: []

modules/ingest-otel/build.gradle

@@ -24,6 +24,6 @@ dependencies {
 
 restResources {
  restApi {
- include '_common', 'indices', 'index', 'cluster', 'nodes', 'get', 'ingest'
+ include '_common', 'indices', 'index', 'cluster', 'nodes', 'get', 'ingest', 'capabilities'
  }
 }

modules/ingest-otel/src/main/java/org/elasticsearch/ingest/otel/NormalizeForStreamProcessor.java

@@ -49,7 +49,7 @@ public class NormalizeForStreamProcessor extends AbstractProcessor {
  /**
  * Mapping of ECS field names to their corresponding OpenTelemetry-compatible counterparts.
  */
- private static final Map<String, String> RENAME_KEYS = Map.ofEntries(
+ static final Map<String, String> RENAME_KEYS = Map.ofEntries(
  entry("span.id", "span_id"),
  entry("message", "body.text"),
  entry("log.level", "severity_text"),
@@ -69,12 +69,12 @@ public class NormalizeForStreamProcessor extends AbstractProcessor {
  Set<String> keepKeys = new HashSet<>(Set.of("@timestamp", "attributes", "resource"));
  Set<String> renamedTopLevelFields = new HashSet<>();
  for (String value : RENAME_KEYS.values()) {
- // if the renamed field is nested, we only need to know the top level field
- int dotIndex = value.indexOf('.');
- if (dotIndex != -1) {
-  renamedTopLevelFields.add(value.substring(0, dotIndex));
- } else {
- renamedTopLevelFields.add(value);
+ // if the renamed field is nested, generate the full list of paths that it could be rooted under
+ String workingKey = null;
+ String[] values = value.split("\\.");
+ for (String part : values) {
+  workingKey = workingKey == null ? part : workingKey + "." + part;
+ renamedTopLevelFields.add(workingKey);
  }
  }
  keepKeys.addAll(renamedTopLevelFields);
@@ -244,7 +244,29 @@ static boolean isOTelDocument(Map<String, Object> source) {
  }
 
  /**
- * Renames specific ECS keys in the given document to their OpenTelemetry-compatible counterparts, based on the {@code RENAME_KEYS} map.
+ * Renames specific ECS keys in the given document to their OpenTelemetry-compatible counterparts in a way that is compatible with the
+ * current access pattern on the IngestDocument.
+ *
+ * <p>This method performs the following operations:
+ * <ul>
+ * <li>For each key in the {@code RENAME_KEYS} map, it checks if a corresponding field exists in the document.</li>
+ * <li>If the field exists, it removes it from the document and adds a new field with the corresponding name from the
+ * {@code RENAME_KEYS} map and the same value.</li>
+ * <li>If the key is nested (contains dots), it recursively removes empty parent fields after renaming.</li>
+ * </ul>
+ *
+ * @param document the document to process
+ */
+ static void renameSpecialKeys(IngestDocument document) {
+ switch (document.getCurrentAccessPatternSafe()) {
+ case CLASSIC -> renameSpecialKeysClassic(document);
+ case FLEXIBLE -> renameSpecialKeysFlexible(document);
+ }
+ }
+
+ /**
+ * Renames specific ECS keys in the given document to their OpenTelemetry-compatible counterparts using logic compatible with the
+ * {@link org.elasticsearch.ingest.IngestPipelineFieldAccessPattern#CLASSIC} access pattern and based on the {@code RENAME_KEYS} map.
  *
  * <p>This method performs the following operations:
  * <ul>
@@ -257,7 +279,7 @@ static boolean isOTelDocument(Map<String, Object> source) {
  *
  * @param document the document to process
  */
- static void renameSpecialKeys(IngestDocument document) {
+ static void renameSpecialKeysClassic(IngestDocument document) {
  RENAME_KEYS.forEach((nonOtelName, otelName) -> {
  boolean fieldExists = false;
  Object value = null;
@@ -272,7 +294,7 @@ static void renameSpecialKeys(IngestDocument document) {
  String parentName = nonOtelName.substring(0, lastDot);
  // parent should never be null and must be a map if we are here
  @SuppressWarnings("unchecked")
- Map<String, Object> parent = (Map<String, Object>) document.getFieldValue(parentName, Map.class);
+ Map<String, Object> parent = document.getFieldValue(parentName, Map.class);
  if (parent.isEmpty()) {
  document.removeField(parentName);
  } else {
@@ -294,6 +316,76 @@ static void renameSpecialKeys(IngestDocument document) {
  });
  }
 
+ /**
+ * Renames specific ECS keys in the given document to their OpenTelemetry-compatible counterparts using logic compatible with the
+ * {@link org.elasticsearch.ingest.IngestPipelineFieldAccessPattern#FLEXIBLE} access pattern and based on the {@code RENAME_KEYS} map.
+ *
+ * <p>This method performs the following operations:
+ * <ul>
+ * <li>For each key in the {@code RENAME_KEYS} map, it checks if a corresponding field exists in the document.</li>
+ * <li>If the field exists, it removes it from the document and adds a new field with the corresponding name from the
+ * {@code RENAME_KEYS} map and the same value. If a field's parent objects do not exist, it will progressively build
+ * each parent object instead of concatenating the field names together.</li>
+ * <li>If the key is nested (contains dots), it recursively removes empty parent fields after renaming.</li>
+ * </ul>
+ *
+ * @param document the document to process
+ */
+ static void renameSpecialKeysFlexible(IngestDocument document) {
+ RENAME_KEYS.forEach((nonOtelName, otelName) -> {
+ boolean fieldExists = false;
+ Object value = null;
+ if (document.hasField(nonOtelName)) {
+ // Dotted fields are treated the same as normalized fields in flexible mode
+ fieldExists = true;
+ value = document.getFieldValue(nonOtelName, Object.class, true);
+ document.removeField(nonOtelName);
+ // recursively remove empty parent fields
+ int lastDot = nonOtelName.lastIndexOf('.');
+ while (lastDot > 0) {
+ String parentName = nonOtelName.substring(0, lastDot);
+ // In flexible mode, dotted field names can be removed. Parent paths may not exist since they might be included
+ // by the dotted field removal (e.g. For the doc {a:{b.c:1}}, removing a.b.c will not leave an a.b field because
+ // there is no a.b field to start with.
+ @SuppressWarnings("unchecked")
+ Map<String, Object> parent = document.getFieldValue(parentName, Map.class, true);
+ if (parent != null) {
+ if (parent.isEmpty()) {
+ document.removeField(parentName);
+ } else {
+ break;
+ }
+ }
+ lastDot = parentName.lastIndexOf('.');
+ }
+ }
+ if (fieldExists) {
+ // Flexible mode creates dotted field names when parent fields are not present. We expect the rename keys to be
+ // normalized after processing, so we progressively build each field's parents if it's a dotted field.
+ Map<String, Object> source = document.getSource();
+ String remainingPath = otelName;
+ int dot = remainingPath.indexOf('.');
+ while (dot > 0) {
+ // Dotted field, emulate classic mode by building out each parent object
+ String fieldName = remainingPath.substring(0, dot);
+ remainingPath = remainingPath.substring(dot + 1);
+ Object existingParent = source.get(fieldName);
+ if (existingParent instanceof Map) {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> castAssignment = (Map<String, Object>) existingParent;
+ source = castAssignment;
+ } else {
+ Map<String, Object> map = new HashMap<>();
+ source.put(fieldName, map);
+ source = map;
+ }
+ dot = remainingPath.indexOf('.');
+ }
+ source.put(remainingPath, value);
+ }
+ });
+ }
+
  private static void moveResourceAttributes(Map<String, Object> attributes, Map<String, Object> resourceAttributes) {
  Set<String> ecsResourceFields = EcsOTelResourceAttributes.LATEST;
  Iterator<Map.Entry<String, Object>> attributeIterator = attributes.entrySet().iterator();