Merge remote-tracking branch 'upstream/main' into component-templates-tracking

* upstream/main:
  Fix MergeWithLowDiskSpaceIT testRelocationWhileForceMerging (elastic#131806)
  [ML] Prevent the trained model deployment memory estimation from double-counting allocations. (elastic#131990)
  ES|QL Assert current thread during query planning and execution (elastic#131807)
  Add ElasticsearchIndexDeletionPolicy and EngineConfig policy wrapper (elastic#130442)
  [TEST] Adds tests for ESTestCase randomSubset methods (elastic#131745)
  Simplify esql session (elastic#131925)
  Simplify EsqlExecution info serialization (elastic#131823)
  Add utility to check for project global block (elastic#131927)
  [DOCS] Update ES|QL applies to's (elastic#131805)
  Handle structured log messages (elastic#131027)
  Mute org.elasticsearch.test.rest.yaml.RcsCcsCommonYamlTestSuiteIT test {p0=search/600_flattened_ignore_above/flattened ignore_above multi-value field} elastic#131967
  Mute org.elasticsearch.xpack.remotecluster.CrossClusterEsqlRCS2EnrichUnavailableRemotesIT testEsqlEnrichWithSkipUnavailable elastic#131965
  Mute org.elasticsearch.xpack.restart.FullClusterRestartIT testWatcherWithApiKey {cluster=UPGRADED} elastic#131964
  [ES|QL] Fix aggregate_metric_double sorting and mv_expand issues (elastic#131658)
  Reduce logging levels for meter usage tests (elastic#131935)

Author: szybia

docs/changelog/131027.yaml

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

docs/changelog/131658.yaml

@@ -0,0 +1,5 @@
+pr: 131658
+summary: Fix `aggregate_metric_double` sorting and `mv_expand` issues
+area: ES|QL
+type: bug
+issues: []

docs/changelog/131990.yaml

@@ -0,0 +1,6 @@
+pr: 131990
+summary: Prevent the trained model deployment memory estimation from double-counting
+  allocations
+area: Machine Learning
+type: bug
+issues: []

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"
+        }
+      }
+    }
+  }
+}
+```

docs/reference/query-languages/esql/_snippets/functions/layout/categorize.md

@@ -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() {}
 

docs/reference/query-languages/esql/_snippets/functions/layout/sample.md

@@ -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 {