Google SecOps Parser Extension - How to reference extracted.fields["_raw.log"] from base parser?

Any help is highly appreciate!

​@cmorris or someone from goggle rep Help us!

The base parser successfully populates:

extracted.fields["_raw.component"]

extracted.fields["_raw.log"]

but my extension cannot access:

[_raw][component]

[_raw][log]

Is there a supported way to reference values already stored in extracted.fields from a Parser Extension?
 

Hi ​@spanuganti  - The extracted fields are populated via the auto extraction feature (https://docs.cloud.google.com/chronicle/docs/event-processing/auto-extraction) instead of via the parser. As a result, the parser extension will need to focus on the log rather than the extracted fields.

Thanks for the clarification. That helps explain why extracted.fields["_raw.component"] and extracted.fields["_raw.log"] are not available in the Parser Extension.

My remaining question is about the original raw event.

• The incoming event looks like:
  {

"_raw": {

"component": "nginx",

"log": "127.0.0.6 - - [07/Aug/2026:18:12:19 +0000] \"GET /api/v2/organizations/uwm/workspaces HTTP/1.1\" 304 0 ..."

},

"cribl_group": "AzureEastUS2-Upper"

}

• In the Parser Extension, I attempted to parse the event using:

  json {

  source => "message"

  }

• and then reference:
   

  [_raw][component]

  [_raw][log]

• • However, validation fails with errors such as:
    "_raw.log" not found in state data

• Is message the correct source field for accessing the original raw payload in a Parser Extension?

  If not, what is the recommended way to access nested fields such as:
   

  _raw.component

  _raw.log
   

  from the original event so that I can parse nginx/atlas/task-worker logs and map them into UDM fields?

  Thanks!

   

Please see the below that will extract the 127.0.0.6 IP from your log to principal.ip (as an example of what the extension would look like):

filter {
  # Initialize variables and the UDM event object to prevent 'empty field' errors
  mutate {
    replace => {
      "principal_ip" => ""
      "_json_parse_error" => "false"
      "_grok_error" => "false"
    }
  }

  # Parse the outer JSON structure
  json {
    source => "message"
    on_error => "_json_parse_error"
  }

  if ![_json_parse_error] {
    # Check if the nested log field is populated before attempting extraction
    if [_raw][log] != "" {
      grok {
        match => {
          "_raw.log" => "^(?P<principal_ip>%{IPORHOST})"
        }
        overwrite => ["principal_ip"]
        on_error => "_grok_error"
      }
    }

    # principal.ip is a repeated field, so we use merge
    if [principal_ip] != "" {
      mutate {
        merge => {
          "event.idm.read_only_udm.principal.ip" => "principal_ip"
        }
      }
    }
  }

  # Output the final UDM event
  mutate {
    merge => {
      "@output" => "event"
    }
  }
}

Thanks for the previous guidance. We successfully validated separate parser logic for:

• nginx logs
• atlas logs
• plain Terraform agent logs

Component logs arrive as:

{

"_raw": {

"component": "atlas",

"log": "..."

}

}

`

• while agent logs arrive as:

{

"_raw": "2026-08-12T20:24:14.390Z [INFO] agent: Starting..."

}

• A combined parser extension fails with:
  "_raw.component" not found in state data

• when evaluating:
  if [_raw][component] == "nginx"
  
  What is the recommended way to safely support both _raw as an object and _raw as a string within the same parser extension?
  
   

  • "_raw": {

  "component": "atlas",

  "log": "..."

  }
  and
  "_raw": "2026-08-12T20:24:14.390Z [INFO] ..."

Can someone from Google team help us ​@cmorris ​@matthewnichols 

Try this:

filter {
  mutate {
    replace => {
      "principal_ip" => ""
    }
  }

  json {
    source => "message"
    on_error => "_json_parse_error"
  }

  if ![_json_parse_error] {

    grok {
      match => {
        "_raw" => "^(?P<is_string>.*)"
      }
      on_error => "_raw_is_object"
    }

    if [_raw_is_object] {

      # --- NGINX LOGIC ---
      if [_raw][component] == "nginx" {

        grok {
          match => {
            "message" => "log[^a-zA-Z0-9]+(?P<principal_ip>[^ ]+)"
          }
          overwrite => ["principal_ip"]
          on_error => "_grok_error"
        }

        if ![_grok_error] {
          mutate {
            merge => {
              "event.idm.read_only_udm.principal.ip" => "principal_ip"
            }
          }
        }

      # --- ATLAS LOGIC ---
      } else if [_raw][component] == "atlas" {
        # add your logic
      }

    # raw log w/o component
    } else {

      grok {
        match => {
          "_raw" => "agent: (?P<status>.*)"
        }
        on_error => "_grok_status_failed"
      }

      if ![_grok_status_failed] {

        mutate {
          replace => {
            "status_field.key" => "status"
            "status_field.value.string_value" => "%{status}"
          }
        }

        mutate {
          merge => {
            "event.idm.read_only_udm.additional.fields" => "status_field"
          }
        }
      }
    }
  }

  mutate {
    merge => {
      "@output" => "event"
    }
  }
}

For testing an example, for your raw string log, this extension maps “Starting...” to additional.fields[“status”] and continues to map the IP for the nginx component log.

Thanks, the _raw_is_object gate works and validates correctly.
atlas :
_raw_is_object = true

component_found = atlas

- However, if we add:
if [_raw][component] == "atlas"

-  the Agent sample still fails with:
"_raw.component" not found in state data
 

It appears Chronicle evaluates [_raw][component] even when [_raw_is_object] == false.

Is there a supported way to safely access _raw.component only when _raw is an object?

Do you have a sample sanitized atlas log? The nginx log works appropriately, I would expect the atlas to work similarly

{

"_raw": {

"component": "atlas",

"log": "2026-08-07 16:17:11 [INFO] [REQUEST-ID] [dd.service=atlas dd.trace_id=TRACE-ID dd.span_id=SPAN-ID ddsource=ruby] {\"method\":\"GET\",\"path\":\"/api/v2/runs/run-EXAMPLE/run-events\",\"format\":\"jsonapi\",\"status\":200,\"allocations\":12345,\"duration\":63.95,\"view\":26.67,\"db\":21.97,\"dd\":{\"trace_id\":\"TRACE-ID\",\"span_id\":\"SPAN-ID\",\"env\":\"\",\"service\":\"atlas\",\"version\":\"\"},\"ddsource\":[\"ruby\"],\"uuid\":\"UUID-EXAMPLE\",\"remote_ip\":\" [removed by moderator] \",\"request_id\":\"REQUEST-ID\",\"user_agent\":\"Mozilla/5.0\",\"user\":\"testuser\",\"auth_source\":\"ui\",\"auth_error_code\":null,\"pid\":1234}"

}

}

the object detection test suceeds:
_raw_is_object = true

component_found = atlas

However, when testing the Agent format where _raw is a string, references to [_raw][component] still produce:
"_raw.component" not found in state data

even when wrapped inside the _raw_is_object conditional.

Added some logic to the atlas section. The parser now works with nginx, atlas, and the raw string. For nginx, IP is mapped to principal.ip; for atlas, remote_ip is mapped to tager_ip; and for the string log, Starting… is mapped to an additional field with key ‘status’. You will need to expand out each section with the mappings you need, but the overall structure should work.

filter {
  mutate {
    replace => {
      "principal_ip" => ""
      "target_ip" => ""
    }
  }

  json {
    source => "message"
    on_error => "_json_parse_error"
  }

  if ![_json_parse_error] {

    # Check string or object
    grok {
      match => {
        "_raw" => "^(?P<is_string>.*)"
      }
      on_error => "_raw_is_object"
    }

    # if object
    if [_raw_is_object] {

      # nginx logic
      if [_raw][component] == "nginx" {

        grok {
          match => {
            "message" => "log[^a-zA-Z0-9]+(?P<principal_ip>[^ ]+)"
          }
          overwrite => ["principal_ip"]
          on_error => "_grok_error"
        }

        if ![_grok_error] {
          mutate {
            merge => {
              "event.idm.read_only_udm.principal.ip" => "principal_ip"
            }
          }
        }

      # atlas logic
      } else if [_raw][component] == "atlas" {        
        grok {
          match => {
            "message" => "remote_ip[^0-9]+(?P<target_ip>%{IP})"
          }
          overwrite => ["target_ip"]
          on_error => "_grok_atlas_error"
        }

        if ![_grok_atlas_error] {
          mutate {
            merge => {
              "event.idm.read_only_udm.target.ip" => "target_ip"
            }
          }
        }
      }

    # else not object - ex. string
    } else {

      grok {
        match => {
          "_raw" => "agent: (?P<status>.*)"
        }
        on_error => "_grok_status_failed"
      }

      if ![_grok_status_failed] {

        mutate {
          replace => {
            "status_field.key" => "status"
            "status_field.value.string_value" => "%{status}"
          }
        }

        mutate {
          merge => {
            "event.idm.read_only_udm.additional.fields" => "status_field"
          }
        }
      }
    }
  }

  mutate {
    merge => {
      "@output" => "event"
    }
  }
}

Thanks, I tested the parser and it works successfully.

Atlas logs are entering the atlas branch and remote_ip is being mapped successfully to target.ip.

I'd now like to extend the Atlas section to map:

• method
• path
• status
• user
• user_agent
• request_id
• auth_source

Since those fields are embedded inside   _raw.log, I'm going to test extracting the JSON payload from _raw.log. I'll share my results if I run into issues.

Hi,

I'm seeing two different event formats from the same source.

Format 1 (works with my current parser):

• {

  "_raw": {

  "component": "task-worker",

  "log": "{\"@message\":\"request complete\"}"

  }

  }
   

• Format 2 (validation fails):

  {

  "_raw": "2026-08-24T17:19:32.308Z [INFO] agent: Core plugin is shutting down",

  "cribl_group": "AzureEastUS2-Upper"

  }

• My parser contains component-specific conditions such as:
   

  if [_raw][component] == "task-worker" {

  ...

  }

• The parser validates successfully for structured component logs, but when it encounters the plaintext format, validation fails with:
  
  "_raw.component" not found in state data
  
   

  because _raw appears to be a string rather than an object.

  Questions:

• What is the recommended approach for handling mixed Terraform Enterprise log formats where _raw can be either:

  • an object containing component and log
  • or a plain string log message
• • Is there a supported way to safely determine whether _raw is an object before referencing
     - [_raw][component]
     
  • Is there a parser pattern or example that handles both formats in a single parser extension?

  • Any guidance would be appreciated.

    Thanks!

When a raw log hits a parser, the log payload and its contents are always treated as a string. It only becomes an object once you start extracting the data from the raw log into (nested) state variables, e.g. by using the json{...} filter.

For your Format 1, using a json filter will give you string values in _raw.component and _raw.log. For your Format 2, using a json filter will give you only a string value in _raw. This is why you get a “does not exist” error since there is no log or component sub-field in this case. As such, you’ll need to handle the two cases separately based on conditions.

 Looking at the parser code ​@cmorris shared in his last reply, this is handled via:
- lines 16-22 determining if nested or not
- subsequent if conditions conditioning on the outcome of this check

This is far from the only way to build the condition you are trying to build, but it is a clever solution and in reading through it, looks like it should gracefully handle the two log examples you shared.

Are you having issues with the parser code above or with a custom derivative you are trying to build? If the latter, sharing your code would be helpful for us to help pinpoint and reproduce the issue you are seeing.

 

​@matt-amastra 

 

Thanks, this is helpful.

I am using a modified version of the parser previously shared in this thread.

The parser successfully handles the structured Terraform Enterprise logs such as:

{

"_raw": {

"component": "task-worker",

"log": "..."

}

}

and I currently have Atlas, Archivist, Task Worker, and Metrics components parsing successfully.

The issue occurs when validation encounters plaintext events such as:

{

"_raw": "2026-08-24T17:19:32.308Z [INFO] agent: Core plugin is shutting down"

}

At that point, references such as:

if [_raw][component] == "task-worker"

cause validation to fail because _raw is not nested.

Could you point me to the specific logic from the earlier example that determines whether the event is nested versus plaintext before evaluating component-specific conditions?

If it helps, I can also share my current parser version.

​@spanuganti, the reason this happens is due to the capture type, as ​@matt-amastra  and ​@cmorris are pointing to in a few ways. 

The `json{...}` filter works on your JSON log example, as it’s simply splitting on key/value pairs, one of which is a `component` field to populate that state variable to key off. 

The syslog string in the other log has nothing for the JSON filter to key off, so it goes to the `else` case on ​@cmorris’s line 66 in the example. This section is only representative (more work to parse syslog than JSON extraction :) ), but has no capture clause setting `component` akin to how you may be expecting since it’s “just there” on the JSON path. 

If still unclear, feel free to share the snippet of your parser code and we can refer to direct lines in your example to explain further. 

Thanks, that makes sense.

My current parser is a derivative of the parser shared earlier in this thread.

Atlas, Archivist, Task Worker, and Metrics component logs are all parsing successfully because the JSON filter populates _raw.component and _raw.log.

The remaining issue is validation of plaintext events such as:

{

"_raw": "2026-08-24T17:19:32.308Z [INFO] agent: Core plugin is shutting down"

}

I now understand that these events never populate _raw.component because there is no JSON structure for the parser to extract.

Could you share or point me to the specific nested vs non-nested condition from Chris's earlier example (lines 16-22 referenced previously) so I can implement the same handling pattern in my parser derivative?

If useful, I can also post my current parser code.

Yes, please post your current parser code ​@spanuganti

filter {

 mutate {

 replace => {

 "principal_ip" => ""

 "target_ip" => ""

 "component" => ""

 }

 }

#   mutate {

#     gsub => ["message","\\r\\n\\t\\t",""]

#   }

#   mutate {

#     gsub => ["message","\\r\\n\\t",""]

#   }

#   mutate {

#     gsub => ["message","\\r\\n",""]

#   }

#   mutate {

#     gsub => ["message","message","msg"]

#   }

json {

        source => "message"

        array_function => "split_columns"

        on_error => "not_a_valid_json"

    }

# PLAINTEXT TERRAFORM LOGS

grok {

 match => {

 "_raw" => "^(?P<log_timestamp>[^ ]+) \\[(?P<log_level>[A-Z]+)\\] (?P<log_source>[^:]+): (?P<log_message>.*)$"

 }

 on_error => "_terraform_plaintext_error"

}

#  if ![_json_parse_error] {

#  grok {

#  match => {

#  "_raw" => "^(?P<is_string>.*)"

#  }

#  on_error => "_raw_is_object"

#  }

#  if [_raw_is_object] {

#  if [_raw][component] == "nginx" {

#  grok {

#  match => {

#  "message" => "log[^a-zA-Z0-9]+(?P<principal_ip>[^ ]+)"

#  }

#  overwrite => ["principal_ip"]

#  on_error => "_grok_error"

#  }

#  if ![_grok_error] {

#  mutate {

#  merge => {

#  "event.idm.read_only_udm.principal.ip" => "principal_ip"

#  }

#  }

#  }

#  } else

# //////////////////////// Atlas Block is ready

  if [_raw][component] == "atlas" {

    mutate {

  replace => {

    "atlas_log" => "%{_raw.log}"

  }

    }

grok {

  match => {

    "atlas_log" => ".*\\{\"method\":\"(?P<atlas_method>[^\"]+)\",\"path\":\"(?P<atlas_path>[^\"]+)\".*\"status\":(?P<atlas_status>[0-9]+).*\"user_agent\":\"(?P<atlas_user_agent>[^\"]+)\".*\"user\":\"(?P<atlas_user>[^\"]+)\".*"

  }

  on_error => "_atlas_field_extract_error"

}

}

 #  //////////////Done Atlas

# ////Archivist Block

if [_raw][component] == "archivist" {

  mutate {

    replace => {

      "archivist_log" => "%{_raw.log}"

    }

  }

  grok {

    match => {

      "archivist_log" => ".*\\\"@message\\\":\\\"(?P<archivist_message>[^\\\"]+)\\\".*"

    }

    on_error => "_archivist_message_error"

  }

}

grok {

  match => {

    "archivist_log" => ".*\\\"@module\\\":\\\"(?P<archivist_module>[^\\\"]+)\\\".*"

  }

  on_error => "_archivist_module_error"

}

grok {

  match => {

    "archivist_log" => ".*\\\"req.filename\\\":\\\"(?P<archivist_filename>[^\\\"]+)\\\".*"

  }

  on_error => "_archivist_filename_error"

}

grok {

  match => {

    "archivist_log" => ".*\\\"req.mode\\\":\\\"(?P<archivist_mode>[^\\\"]+)\\\".*"

  }

  on_error => "_archivist_mode_error"

}

grok {

  match => {

    "archivist_log" => ".*\\\"obj.key\\\":\\\"(?P<archivist_obj_key>[^\\\"]+)\\\".*"

  }

  on_error => "_archivist_obj_key_error"

}

grok {

  match => {

    "archivist_log" => ".*\\\"req.ttl\\\":\\\"(?P<archivist_ttl>[^\\\"]+)\\\".*"

  }

  on_error => "_archivist_ttl_error"

}

grok {

  match => {

    "archivist_log" => ".*\\\"req.key\\\":\\\"(?P<archivist_req_key>[^\\\"]+)\\\".*"

  }

  on_error => "_archivist_req_key_error"

}

# ////////////////

    mutate{

        replace => {

            "event.idm.read_only_udm.metadata.description" => "%{_raw.log}"

        }

        on_error => "no_component"

    }

#  if [_raw][component]!= ""{

#     mutate{

#         replace => {

#             "event.idm.read_only_udm.principal.hostname" => "%{_raw.component}"

#         }

#         on_error => "no_component"

#     }

#  }

# Taskworker ///////////

if [_raw][component] == "task-worker" {

  mutate {

    replace => {

      "taskworker_log" => "%{_raw.log}"

    }

  }

}

 grok {

  match => {

    "taskworker_log" => ".*\\\"@message\\\":\\\"(?P<taskworker_message>[^\\\"]+)\\\".*"

  }

  on_error => "_taskworker_message_error"

}

grok {

  match => {

    "taskworker_log" => ".*\\\"message\\\":\\\"(?P<taskworker_error_message>[^\\\"]+)\\\".*"

  }

  on_error => "_taskworker_error_message_error"

}

grok {

  match => {

    "taskworker_log" => ".*\\\"@module\\\":\\\"(?P<taskworker_module>[^\\\"]+)\\\".*"

  }

  on_error => "_taskworker_module_error"

}

grok {

  match => {

    "taskworker_log" => ".*\\\"method\\\":\\\"(?P<taskworker_method>[^\\\"]+)\\\".*"

  }

  on_error => "_taskworker_method_error"

}

grok {

  match => {

    "taskworker_log" => ".*\\\"path\\\":\\\"(?P<taskworker_path>[^\\\"]+)\\\".*"

  }

  on_error => "_taskworker_path_error"

}

#   if [cribl_group]!= ""{

#     mutate{

#         replace => {

#             "event.idm.read_only_udm.target.hostname" => "%{cribl_group}"

#         }

#     }

#  }

# if [cribl_group] {

#     mutate {

#         replace => {

#             "event.idm.read_only_udm.target.hostname" => "%{cribl_group}"

#         }

#     }

# }

# Metrics ///////////

if [_raw][component] == "metrics" {

  mutate {

    replace => {

      "metrics_log" => "%{_raw.log}"

    }

  }

  grok {

    match => {

      "metrics_log" => ".*\\\"@message\\\":\\\"(?P<metrics_message>[^\\\"]+)\\\".*"

    }

    on_error => "_metrics_message_error"

  }

  grok {

    match => {

      "metrics_log" => ".*\\\"@module\\\":\\\"(?P<metrics_module>[^\\\"]+)\\\".*"

    }

    on_error => "_metrics_module_error"

  }

  grok {

    match => {

      "metrics_log" => ".*\\\"Name\\\":\\\"(?P<metrics_name>[^\\\"]+)\\\".*"

    }

    on_error => "_metrics_name_error"

  }

  grok {

    match => {

      "metrics_log" => ".*\\\"Type\\\":\\\"(?P<metrics_type>[^\\\"]+)\\\".*"

    }

    on_error => "_metrics_type_error"

  }

  grok {

    match => {

      "metrics_log" => ".*\\\"Value\\\":(?P<metrics_value>[0-9]+).*"

    }

    on_error => "_metrics_value_error"

  }

}

# ////////////

 mutate {

 merge => {

 "@output" => "event"

 }

 }

    # statedump{}

}

raw log:
{ "_raw": "2026-08-25T18:40:02.278Z [INFO] terraform: Generating and uploading plan JSON", "cribl_group": "AzureEastUS2-Upper" }

Error:
generic::unknown: pipeline.ParseLogEntry failed: LOG_PARSING_CBN_ERROR: "generic::invalid_argument: pipeline failed: filter conditional (3) failed: failed to evaluate expression: generic::invalid_argument: \"_raw.component\" not found in state data"

​@spanuganti I updated your parser and tested against every log sample you have sent in this thread and all validated properly. Each code path now contains a representative example of how to handle each log format + in-place `TODO` comments for what you should handle next. Use the UDM Field List to determine relevant destination fields for each source field and the Parser Syntax Reference for documentation on available functions/helpers and how to use them. Grok pattern reference is also helpful building grok match expressions.

filter {
  # INIT
  mutate {
    replace => {
      "principal_ip" => ""
      "target_ip" => ""
      "component" => ""
      # Required Fields
      "event.idm.read_only_udm.metadata.event_type" => "GENERIC_EVENT"
    }
  }

  json {
    source => "message"
    array_function => "split_columns"
    on_error => "_not_a_valid_json"
  }

  if ![_not_a_valid_json] {
    # Determine if _raw is string or object
    grok {
      match => {
        "_raw" => "^(?P<is_string>.*)"
      }
      on_error => "_raw_is_object"
    }
    # If object
    if [_raw_is_object] {
      # Store raw log in description
      mutate{
        replace => {
          "event.idm.read_only_udm.metadata.description" => "%{_raw.log}"
        }
        on_error => "_no_raw_log"
      }
      #
      # TODO: Map _raw.component to a UDM field
      #
      json {
        source => "_raw.log"
        array_function => "split_columns"
        on_error => "_log_not_json"
      }
      # Handle logs based on component type
      # ATLAS
      if [_raw][component] == "atlas" {
        grok {
          match => {
            "_raw.log" => ["%{TIMESTAMP_ISO8601:log_timestamp}%{SPACE}\\[%{WORD:log_level}\\]%{SPACE}\\[%{DATA:field0}\\]%{SPACE}\\[%{DATA:kv}\\]%{SPACE}%{GREEDYDATA:json}",
            "%{TIMESTAMP_ISO8601:log_timestamp}%{SPACE}\\[%{WORD:log_level}\\]%{SPACE}%{GREEDYDATA:other_fields}"]
          }
          on_error => "_atlas_raw_log_grok_mismatch"
        }
        if ![_atlas_raw_log_grok_mismatch] {
          kv {
            source => "kv"
            field_split => " "
            value_split => "="
            on_error => "_not_a_valid_inner_kv"
          }
          json {
            source => "json"
            array_function => "split_columns"
            on_error => "_not_a_valid_inner_json"
          }
        }
        #
        # TODO: Map all extracted fields to UDM fields
        #
      }
      # ARCHIVIST
      if [_raw][component] == "archivist" {
        #
        # TODO: Extract data from all archivist formats and map to UDM
        #
      }
      # TASKWORKER
      if [_raw][component] == "task-worker" {
        #
        # TODO: Extract data from all task-worker formats and map to UDM
        #
      }
      # METRICS
      if [_raw][component] == "metrics" {
        #
        # TODO: Extract data from all metrics formats and map to UDM
        #
      }
      #
      # TODO: Handle other component types
      #
    }
    # If string
    else {
      # PLAINTEXT TERRAFORM LOGS
      grok {
        match => {
          "_raw" => "^(?P<log_timestamp>[^ ]+) \\[(?P<log_level>[A-Z]+)\\] (?P<log_source>[^:]+): (?P<log_message>.*)$"
        }
        on_error => "_terraform_plaintext_error"
      }
      date {
        match => ["log_timestamp", "ISO8601"]
        on_error => "_no_date_match"
      }
      mutate{
        replace => {
          "event.idm.read_only_udm.metadata.description" => "%{log_message}"
        }
        on_error => "_no_log_message"
      }
      #
      # TODO: Map log_source to a UDM field
      #
      mutate {
        replace => {
          "security_result.severity_details" => "%{log_level}"
        }
        on_error => "_no_log_level"
      }
      if [log_level] == "INFO" {
        mutate {
          replace => {
            "security_result.severity" => "INFORMATIONAL"
          }
        }
      }
      #
      # TODO: Map other log_level values to UDM severity ENUM
      #
      mutate {
        merge => {
          "event.idm.read_only_udm.security_result" => "security_result"
        }
        on_error => "-_no_security_result"
      }
    }
  }
  # Not JSON
  else {
    # Dump whole log contents in description
    mutate{
      replace => {
        "event.idm.read_only_udm.metadata.description" => "%{message}"
      }
      on_error => "no_raw_log"
    }
  }
  statedump{}
  # OUTPUT
  mutate {
    merge => {
      "@output" => "event"
    }
  }
}