Writing raw mapping_error events from the Logstash DLQ to Elasticsearch

[Yor_On]

Hi,

I'm trying to find a way to get the raw data of messages that fail on mapping errors from my Logstash's Dead Letter Queues to Elasticsearch so I can look into the reasons why the messages ended up in the DLQ in the first place. (Logstash and ES versions are both 7.7.0)

I found this question asked a while ago: Logstash Dead Letter Queue, send raw message to ElasticSearch, which describes exactly what I want to do, but it looks like original solution the OP asked for was never provided.

I don't care about "fixing" the DLQ'd events or them not reaching their original "destination", I'm only interested in bringing the raw messages as strings to Elasticsearch, preferably with the reason for the mapping error from the DLQ metadata, so I can have a record of all those same unindexed messages and the reasons why they happened.

At the moment, with a straightforward pipeline using the dead_letter_queue input plugin and the elasticsearch output plugin, the events from the DLQ get written "as is" to the ES index (sometimes causing even more mapping errors)

I'm struggling to find how I can "encapsulate" the raw JSON message and add the DLQ metadata through the filter section with the mutate plugin, because I can't find a way to reference the entirety of the JSON message that I'm filtering over from the dead_letter_queue input plugin, or the DLQ metadata....

For example, I need to go from:

{
    "object_with_mapping_error" :
    {
        "int_field": "oops",
        "string_field": 123
    }
}

to:

{
    "message" : "{\"object_with_mapping_error\":{\"int_field\": \"oops\", \"string_field\": 123}}",
    "error":
    {
        "type" : "mapper_parsing_exception",
        "reason": <corresponding value from DLQ metadata>,
        "caused_by" : {
            "type" : <corresponding value from DLQ metadata>,
            "reason": <corresponding value from DLQ metadata>
        }
    }
    
}

Any and all help/pointers in figuring out the filter section needed to get this done would be greatly appreciated!