Handle structured log messages (#131027)

Author: eyalkoren

docs/changelog/131027.yaml

@@ -0,0 +1,6 @@
+pr: 131027
+summary: Handle structured log messages
+area: Ingest Node
+type: feature
+issues:
+ - 130333

docs/reference/enrich-processor/normalize-for-stream.md

@@ -153,3 +153,87 @@ will be normalized into the following form:
  "trace_id": "abcdef1234567890abcdef1234567890"
 }
 ```
+## Structured `message` field
+
+If the `message` field in the ingested document is structured as a JSON, the
+processor will determine whether it is in ECS format or not, based on the
+existence or absence of the `@timestamp` field. If the `@timestamp` field is
+present, the `message` field will be considered to be in ECS format, and its
+contents will be merged into the root of the document and then normalized as
+described above. The `@timestamp` from the `message` field will override the
+root `@timestamp` field in the resulting document.
+If the `@timestamp` field is absent, the `message` field will be moved to
+the `body.structured` field as is, without any further normalization.
+
+For example, if the `message` field is an ECS-JSON, as follows:
+
+```json
+{
+ "@timestamp": "2023-10-01T12:00:00Z",
+ "message": "{\"@timestamp\":\"2023-10-01T12:01:00Z\",\"log.level\":\"INFO\",\"service.name\":\"my-service\",\"message\":\"The actual log message\",\"http\":{\"method\":\"GET\",\"url\":{\"path\":\"/api/v1/resource\"}}}"
+
+}
+```
+it will be normalized into the following form:
+
+```json
+{
+ "@timestamp": "2023-10-01T12:01:00Z",
+ "severity_text": "INFO",
+ "body": {
+ "text": "The actual log message"
+ },
+ "resource": {
+ "attributes": {
+ "service.name": "my-service"
+ }
+ },
+ "attributes": {
+ "http.method": "GET",
+ "http.url.path": "/api/v1/resource"
+ }
+}
+```
+
+However, if the `message` field is not recognized as ECS format, as follows:
+
+```json
+{
+ "@timestamp": "2023-10-01T12:00:00Z",
+ "log": {
+ "level": "INFO"
+ },
+ "service": {
+ "name": "my-service"
+ },
+ "tags": ["user-action", "api-call"],
+ "message": "{\"root_cause\":\"Network error\",\"http\":{\"method\":\"GET\",\"url\":{\"path\":\"/api/v1/resource\"}}}"
+}
+```
+it will be normalized into the following form:
+
+```json
+{
+ "@timestamp": "2023-10-01T12:00:00Z",
+ "severity_text": "INFO",
+ "resource": {
+ "attributes": {
+ "service.name": "my-service"
+ }
+ },
+ "attributes": {
+ "tags": ["user-action", "api-call"]
+ },
+ "body": {
+ "structured": {
+ "root_cause": "Network error",
+ "http": {
+ "method": "GET",
+ "url": {
+ "path": "/api/v1/resource"
+ }
+ }
+ }
+ }
+}
+```

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/IngestCommonPlugin.java

@@ -21,6 +21,7 @@
 import org.elasticsearch.ingest.PipelineProcessor;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.plugins.ActionPlugin;
+import org.elasticsearch.plugins.ExtensiblePlugin;
 import org.elasticsearch.plugins.IngestPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.rest.RestController;
@@ -33,7 +34,7 @@
 
 import static java.util.Map.entry;
 
-public class IngestCommonPlugin extends Plugin implements ActionPlugin, IngestPlugin {
+public class IngestCommonPlugin extends Plugin implements ActionPlugin, IngestPlugin, ExtensiblePlugin {
 
  public IngestCommonPlugin() {}
 

modules/ingest-otel/build.gradle

@@ -12,6 +12,14 @@ apply plugin: 'elasticsearch.internal-yaml-rest-test'
 esplugin {
  description = 'Ingest processor that normalizes ECS documents to OpenTelemetry-compatible namespaces'
  classname ='org.elasticsearch.ingest.otel.NormalizeForStreamPlugin'
+ extendedPlugins = ['ingest-common']
+}
+
+dependencies {
+ compileOnly(project(':modules:ingest-common'))
+ compileOnly project(':modules:lang-painless:spi')
+ clusterModules project(':modules:ingest-common')
+ clusterModules project(':modules:lang-painless')
 }
 
 restResources {

modules/ingest-otel/src/main/java/module-info.java

@@ -10,4 +10,6 @@
 module org.elasticsearch.ingest.otel {
  requires org.elasticsearch.base;
  requires org.elasticsearch.server;
+ requires org.apache.logging.log4j;
+ requires org.elasticsearch.ingest.common;
 }

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

@@ -9,11 +9,14 @@
 
 package org.elasticsearch.ingest.otel;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
+import org.elasticsearch.ingest.common.JsonProcessor;
 
 import java.util.HashMap;
 import java.util.HashSet;
@@ -60,6 +63,8 @@ public class NormalizeForStreamProcessor extends AbstractProcessor {
  * OpenTelemetry-compatible fields that are renamed by the processor.
  */
  private static final Set<String> KEEP_KEYS;
+ private static final Logger log = LogManager.getLogger(NormalizeForStreamProcessor.class);
+
  static {
  Set<String> keepKeys = new HashSet<>(Set.of("@timestamp", "attributes", "resource"));
  Set<String> renamedTopLevelFields = new HashSet<>();
@@ -103,6 +108,41 @@ public IngestDocument execute(IngestDocument document) {
 
  // non-OTel document
 
+ // handling structured messages
+ Map<String, Object> body = null;
+ try {
+ String message = document.getFieldValue("message", String.class, true);
+ if (message != null) {
+ message = message.trim();
+ if (message.startsWith("{") && message.endsWith("}")) {
+ // if the message is a JSON object, we assume it is a structured log
+ Object parsedMessage = JsonProcessor.apply(message, true, true);
+ if (parsedMessage instanceof Map) {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> messageMap = (Map<String, Object>) parsedMessage;
+ if (messageMap.containsKey("@timestamp")) {
+ log.debug(
+ "Handling structured message with @timestamp field, assuming ECS-JSON format, merging into root document"
+ );
+ source.remove("message");
+ JsonProcessor.recursiveMerge(source, messageMap);
+ } else {
+ log.debug(
+ "Handling structured message without @timestamp field, assuming non-ECS format, moving to 'body.structured'"
+ );
+ body = new HashMap<>();
+ body.put(STRUCTURED_KEY, messageMap);
+ source.remove("message");
+ }
+ } else {
+ log.debug("Structured message is not a JSON object, keeping it as a string in 'body.text' field: {}", message);
+ }
+ }
+ }
+ } catch (Exception e) {
+ log.warn("Failed to parse structured message, keeping it as a string in 'body.text' field: {}", e.getMessage());
+ }
+
  Map<String, Object> newAttributes = new HashMap<>();
  // The keep keys indicate the fields that should be kept at the top level later on when applying the namespacing.
  // However, at this point we need to move their original values (if they exist) to the one of the new attributes namespaces, except
@@ -117,6 +157,11 @@ public IngestDocument execute(IngestDocument document) {
  }
  }
 
+ // if the body is not null, it means we have a structured log that we need to move to the body.structured field.
+ if (body != null) {
+ source.put(BODY_KEY, body);
+ }
+
  source.put(ATTRIBUTES_KEY, newAttributes);
 
  renameSpecialKeys(document);