[apm-data] Set event.dataset if empty for logs

[carsonip]

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.

Fixes elastic/apm-server#16397

[elasticsearchmachine]

Pinging @elastic/es-data-management (Team:Data Management)

[elasticsearchmachine]

Hi @carsonip, I've created a changelog YAML for you.

[carsonip]

Testing notes

Tested with adding the processor to default pipeline.

DELETE _data_stream/logs-apm.app.foo-bar

POST /logs-apm.app.foo-bar/_doc
{
  "@timestamp": 1,
  "data_stream": {
    "type": "logs",
    "dataset": "apm.app.foo",
    "namespace": "bar"
  }
}

POST /logs-apm.app.foo-bar/_doc
{
  "@timestamp": 1,
  "data_stream": {
    "type": "logs",
    "dataset": "apm.app.foo",
    "namespace": "bar"
  },
  "event": {
    "dataset": "ds"
  }
}

GET /logs-apm.app.foo-bar/_search

Output:

{
  "took": 1,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 2,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": ".ds-logs-apm.app.foo-bar-2025.06.09-000001",
        "_id": "_WCQVZcBbmWL1cSKlTSi",
        "_score": 1,
        "_source": {
          "@timestamp": 1,
          "data_stream": {
            "namespace": "bar",
            "type": "logs",
            "dataset": "apm.app.foo"
          },
          "event": {
            "ingested": "2025-06-09T16:40:28Z",
            "dataset": "apm.app.foo"
          }
        }
      },
      {
        "_index": ".ds-logs-apm.app.foo-bar-2025.06.09-000001",
        "_id": "_mCQVZcBbmWL1cSKoTRm",
        "_score": 1,
        "_source": {
          "@timestamp": 1,
          "data_stream": {
            "namespace": "bar",
            "type": "logs",
            "dataset": "apm.app.foo"
          },
          "event": {
            "ingested": "2025-06-09T16:40:31Z",
            "dataset": "ds"
          }
        }
      }
    ]
  }
}

[axw]

Looks good, just a question about sending docs to logs-apm.* without explicitly setting data_stream.dataset. Not sure if that's a thing we need to cater for.

[axw]

Nit: should we use copy_from and ignore_empty_value, and remove the if and value?
(See https://www.elastic.co/docs/reference/enrich-processor/set-processor)

[axw]

Will data_stream.dataset be automatically set for the logs-apm.* data stream documents, if they're not specified in _source? If that's the case, I'm not sure they will be in ctx.*, and we may need to have a fallback.

Can you add test cases for that?

[carsonip]

Will data_stream.dataset be automatically set for the logs-apm.* data stream documents, if they're not specified in _source?

No, at least not visible to the log ingest pipeline.

If that's the case, I'm not sure they will be in ctx.*

Correct, they will not be in ctx.

we may need to have a fallback.

Do you mean a fallback to a non-null event.dataset in the case where both data_stream.dataset and event.dataset is missing? I'm inclined to not perform any special case handling to make things up because it sounds more like a misconfiguration issue. Unless the user explicitly remove the DS fields in an intermediate ingest pipeline, DS fields should always be present in docs from apm-server (even if reroute processor is used).

In any case, I've added test in 957d9ef to confirm this behavior.

[axw]

Do you mean a fallback to a non-null event.dataset in the case where both data_stream.dataset and event.dataset is missing? I'm inclined to not perform any special case handling to make things up because it sounds more like a misconfiguration issue.

I'm OK with this. I don't think it's a real thing we need to handle, even if it's technically possible.

[dakrone]

Is there a reason why you switched away from apm@pipeline for the final pipeline? When I look at this it appears that these logs will no longer be doing the processing in that file, instead only doing the event.dataset configuration from the new final pipeline.

Can this be considered a breaking change in behavior?

[carsonip]

Is there a reason why you switched away from apm@pipeline for the final pipeline? When I look at this it appears that these logs will no longer be doing the processing in that file, instead only doing the event.dataset configuration from the new final pipeline.

The new logs-apm@pipeline in turn calls apm@pipeline, just like the other final pipelines e.g. metrics-apm@pipeline and traces-apm@pipeline. It is needed just because we needed log-specific processor in the final pipeline and we cannot just add it to apm@pipeline directly.

[dakrone]

Ahh okay, that makes sense. I missed the call out to the original apm@pipeline (I am not used to reading pipelines in yaml…). Thanks for the explanation.

[axw]

LGTM, thanks, and sorry for the delay

[elasticsearchmachine]

💔 Backport failed

Status	Branch	Result
❌	8.19	Commit could not be cherrypicked due to conflicts

You can use sqren/backport to manually backport by running backport --upstream elastic/elasticsearch --pr 129074

[carsonip]

💚 All backports created successfully

Status	Branch	Result
✅	8.19

Questions ?

Please refer to the Backport tool documentation

[rubvs]

Tested manually with steps outlined in #129074 (comment)

With version 9.0.3

    "hits": [
      {
        "_index": ".ds-logs-apm.app.foo-bar-2025.06.23-000001",
        "_id": "JNXnnZcBQ3xVa0c-8bvn",
        "_score": 1,
        "_source": {
          "@timestamp": 1,
          "data_stream": {
            "namespace": "bar",
            "type": "logs",
            "dataset": "apm.app.foo"
          },
          "tags": [
            "_geoip_database_unavailable_GeoLite2-City.mmdb"
          ]
        }
      },
      {
        "_index": ".ds-logs-apm.app.foo-bar-2025.06.23-000001",
        "_id": "xU_nnZcBCJmh9scm-w5_",
        "_score": 1,
        "_source": {
          "@timestamp": 1,
          "data_stream": {
            "namespace": "bar",
            "type": "logs",
            "dataset": "apm.app.foo"
          },
          "event": {
            "dataset": "ds"
          },
          "tags": [
            "_geoip_database_unavailable_GeoLite2-City.mmdb"
          ]
        }
      }
    ]

With version 9.1.0

    "hits": [
      {
        "_index": ".ds-logs-apm.app.foo-bar-2025.06.23-000001",
        "_id": "hFznnZcBHYim8ksPeDdl",
        "_score": 1,
        "_source": {
          "@timestamp": 1,
          "data_stream": {
            "namespace": "bar",
            "type": "logs",
            "dataset": "apm.app.foo"
          },
          "event": {
            "dataset": "apm.app.foo"
          },
          "tags": [
            "_geoip_database_unavailable_GeoLite2-City.mmdb"
          ]
        }
      },
      {
        "_index": ".ds-logs-apm.app.foo-bar-2025.06.23-000001",
        "_id": "hVznnZcBHYim8ksPjTfk",
        "_score": 1,
        "_source": {
          "@timestamp": 1,
          "data_stream": {
            "namespace": "bar",
            "type": "logs",
            "dataset": "apm.app.foo"
          },
          "event": {
            "dataset": "ds"
          },
          "tags": [
            "_geoip_database_unavailable_GeoLite2-City.mmdb"
          ]
        }
      }
    ]

By inspecting the outputs above, version 9.0.3 does not include the event.dataset field, while 9.1.0 does include it. Concluding that the data_stream.dataset field is copied properly.

"event": {
    "dataset": "apm.app.foo"
},