[apm-data] Set event.dataset if empty for logs (#129074) (#129537)

For APM logs, set event.dataset to data_stream.dataset if event.dataset is empty, to satisfy Anomaly Detection's requirement to have event.dataset in every logs-* data stream.

(cherry picked from commit 466afba)

# Conflicts:
#	test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java

Author: carsonip

docs/changelog/129074.yaml

@@ -0,0 +1,5 @@
+pr: 129074
+summary: "[apm-data] Set `event.dataset` if empty for logs"
+area: Data streams
+type: bug
+issues: []

x-pack/plugin/apm-data/src/main/resources/index-templates/[email protected]

@@ -24,4 +24,4 @@ template:
   settings:
     index:
       default_pipeline: logs-apm.app@default-pipeline
-      final_pipeline: apm@pipeline
+      final_pipeline: logs-apm@pipeline

x-pack/plugin/apm-data/src/main/resources/index-templates/[email protected]

@@ -31,4 +31,4 @@ template:
   settings:
     index:
       default_pipeline: logs-apm.error@default-pipeline
-      final_pipeline: apm@pipeline
+      final_pipeline: logs-apm@pipeline

x-pack/plugin/apm-data/src/main/resources/ingest-pipelines/[email protected]

@@ -0,0 +1,14 @@
+---
+version: ${xpack.apmdata.template.version}
+_meta:
+  managed: true
+description: Built-in ingest pipeline for logs-apm.*-* data streams
+processors:
+# Set event.dataset if unset to meet Anomaly Detection requirements
+- set:
+    field: event.dataset
+    copy_from: "data_stream.dataset"
+    ignore_empty_value: true
+    override: false
+- pipeline:
+    name: apm@pipeline

x-pack/plugin/apm-data/src/main/resources/resources.yaml

@@ -1,7 +1,7 @@
 # "version" holds the version of the templates and ingest pipelines installed
 # by xpack-plugin apm-data. This must be increased whenever an existing template or
 # pipeline is changed, in order for it to be updated on Elasticsearch upgrade.
-version: 15
+version: 16
 
 component-templates:
   # Data lifecycle.
@@ -97,6 +97,9 @@ ingest-pipelines:
   - metrics-apm@pipeline:
       dependencies:
         - apm@pipeline
+  - logs-apm@pipeline:
+      dependencies:
+        - apm@pipeline
 
 lifecycle-policies:
   - logs-apm.app_logs-default_policy

x-pack/plugin/apm-data/src/yamlRestTest/resources/rest-api-spec/test/20_logs_pipeline.yml

@@ -0,0 +1,70 @@
+---
+setup:
+  - do:
+      cluster.health:
+        wait_for_events: languid
+---
+"Test logs-apm.error-* event.dataset field":
+  - do:
+      bulk:
+        index: logs-apm.error-eventdataset
+        refresh: true
+        body:
+          # data_stream.dataset present, event.dataset not present
+          - create: {}
+          - '{"@timestamp": "2017-06-22", "data_stream": {"type": "logs", "dataset": "apm.error", "namespace": "eventdataset"}, "log": {"level": "error"}, "error": {"log": {"message": "loglevel"}, "exception": [{"message": "exception_used"}]}}'
+          # data_stream.dataset present, event.dataset present
+          - create: {}
+          - '{"@timestamp": "2017-06-22", "data_stream": {"type": "logs", "dataset": "apm.error", "namespace": "eventdataset"}, "event": {"dataset": "foo"}, "log": {"level": "error"}, "error": {"log": {"message": "loglevel"}, "exception": [{"message": "exception_used"}]}}'
+          # unlikely: data_stream.dataset not present, event.dataset not present
+          - create: {}
+          - '{"@timestamp": "2017-06-22", "log": {"level": "error"}, "error": {"log": {"message": "loglevel"}, "exception": [{"message": "exception_used"}]}}'
+          # unlikely: data_stream.dataset not present, event.dataset present
+          - create: {}
+          - '{"@timestamp": "2017-06-22", "event": {"dataset": "foo"}, "log": {"level": "error"}, "error": {"log": {"message": "loglevel"}, "exception": [{"message": "exception_used"}]}}'
+
+  - is_false: errors
+
+  - do:
+      search:
+        index: logs-apm.error-eventdataset
+        body:
+          fields: ["event.dataset"]
+  - length: { hits.hits: 4 }
+  - match: { hits.hits.0.fields: { "event.dataset": ["apm.error"] } }
+  - match: { hits.hits.1.fields: { "event.dataset": ["foo"] } }
+  - match: { hits.hits.2.fields: null }
+  - match: { hits.hits.3.fields: { "event.dataset": ["foo"] } }
+---
+"Test logs-apm.app.*-* event.dataset field":
+  - do:
+      bulk:
+        index: logs-apm.app.foo-eventdataset
+        refresh: true
+        body:
+          # data_stream.dataset present, event.dataset not present
+          - create: {}
+          - '{"@timestamp": "2017-06-22", "data_stream": {"type": "logs", "dataset": "apm.app.foo", "namespace": "eventdataset"}, "message": "foo"}'
+          # data_stream.dataset present, event.dataset present
+          - create: {}
+          - '{"@timestamp": "2017-06-22", "data_stream": {"type": "logs", "dataset": "apm.app.foo", "namespace": "eventdataset"}, "event": {"dataset": "foo"}, "message": "foo"}'
+          # unlikely: data_stream.dataset not present, event.dataset not present
+          - create: {}
+          - '{"@timestamp": "2017-06-22", "message": "foo"}'
+          # unlikely: data_stream.dataset not present, event.dataset present
+          - create: {}
+          - '{"@timestamp": "2017-06-22", "event": {"dataset": "foo"}, "message": "foo"}'
+
+  - is_false: errors
+
+  - do:
+      search:
+        index: logs-apm.app.foo-eventdataset
+        body:
+          fields: ["event.dataset"]
+  - length: { hits.hits: 4 }
+  - match: { hits.hits.0.fields: { "event.dataset": ["apm.app.foo"] } }
+  - match: { hits.hits.1.fields: { "event.dataset": ["foo"] } }
+  - match: { hits.hits.2.fields: null }
+  - match: { hits.hits.3.fields: { "event.dataset": ["foo"] } }
+