Best way to drop an event

[Bouska]

Hi,

I've create a pipeline that takes various logs from different embedded devices. The logs are either in kernel format or process format. Anything that are not those formats are to be dropped. The way I've done it is to have one transform per type of logs; I raise an error with parse_regex and I drop the event on error. So all the events goes in parallel in both transforms, the event will either match one of the two regex and continue or be dropped.

transforms:
  parse_kernel_log:
    type: "remap"
    inputs: ["*_extract_log"]
    drop_on_error: true
    source: |
      . |= parse_regex!(.message, r'^\[(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\] \[(?P<severity>\w+)\] \[\s*(?P<uptime>[0-9]+[.][0-9]{6})\] (?P<message>.+)$')

      # Coerce parsed fields
      .timestamp = parse_timestamp!(.timestamp, "%Y-%m-%d %H:%M:%S")
      .pid = 0
      .process = "kernel"
      .message = strip_whitespace(.message)
      .uptime = to_float!(.uptime)

  parse_process_log:
    type: "remap"
    inputs: ["*_extract_log"]
    drop_on_error: true
    source: |
      . |= parse_regex!(.message, r'^\[(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\] \[(?P<severity>\w+)\] (?P<process>\w+)\[(?P<pid>\d+)\]: (?P<message>.+)$')

      # Coerce parsed fields
      .timestamp = parse_timestamp!(.timestamp, "%Y-%m-%d %H:%M:%S")
      .pid = to_int!(.pid)
      .message = strip_whitespace(.message)

This works as intended but there is two drawbacks:

• the Vector logs are flooded with the errors from parse_regex
• all the events are checked twice; ideally the checks could be short-cut if one has succeed

2025-05-14T09:40:58.986449Z ERROR transform{component_kind="transform" component_id=parse_kernel_log component_type=remap}: vector_common::internal_event::component_events_dropped: Internal log [Events dropped] has been suppressed 6171 times.
2025-05-14T09:40:58.986486Z ERROR transform{component_kind="transform" component_id=parse_kernel_log component_type=remap}: vector_common::internal_event::component_events_dropped: Events dropped intentional=false count=1 reason="Mapping failed with event." internal_log_rate_limit=true
2025-05-14T09:40:58.986493Z ERROR transform{component_kind="transform" component_id=parse_process_log component_type=remap}: vector_common::internal_event::component_events_dropped: Internal log [Events dropped] has been suppressed 2445 times.
2025-05-14T09:40:58.986513Z ERROR transform{component_kind="transform" component_id=parse_process_log component_type=remap}: vector_common::internal_event::component_events_dropped: Events dropped intentional=false count=1 reason="Mapping failed with event." internal_log_rate_limit=true
2025-05-14T09:40:58.986523Z ERROR transform{component_kind="transform" component_id=parse_kernel_log component_type=remap}: vector::internal_events::remap: Internal log [Mapping failed with event.] is being suppressed to avoid flooding.
2025-05-14T09:40:58.986553Z ERROR transform{component_kind="transform" component_id=parse_kernel_log component_type=remap}: vector_common::internal_event::component_events_dropped: Internal log [Events dropped] is being suppressed to avoid flooding.
2025-05-14T09:40:58.988682Z ERROR transform{component_kind="transform" component_id=parse_process_log component_type=remap}: vector::internal_events::remap: Internal log [Mapping failed with event.] is being suppressed to avoid flooding.
2025-05-14T09:40:58.988713Z ERROR transform{component_kind="transform" component_id=parse_process_log component_type=remap}: vector_common::internal_event::component_events_dropped: Internal log [Events dropped] is being suppressed to avoid flooding.

If we had custom function in VRL, I could do something like that parse_process_log(.message) ?? parse_kernel_log(.message) ?? {} and then filter out the empty events.

Is there a better way to do what I'm currently doing?

[satellite-no]

You have a couple of options here, before you transform the data I would use an exclusive route to split the streams, then you have one stream far process and another far kernel and you can process the data as expected.

example

type: exclusive_route
inputs:
  - '*'
routes:
  - name: "kernel"
    condition: match(to_string!(.message), r'parse_kernel_log')
  - name: "process"
     condition: match(to_string!(.message), r'parse_process_log')

Alternatively, you could convert to one remap and try to process the data on error.

example

transforms:
  parse_kernel_log:
    type: "remap"
    inputs: ["*_extract_log"]
    drop_on_error: true
    source: |
      ., err |= parse_regex(.message, r'^\[(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\] \[(?P<severity>\w+)\] \[\s*(?P<uptime>[0-9]+[.][0-9]{6})\] (?P<message>.+)$')

      if err != null {
        . |= parse_regex!(.message, r'^\[(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\] \[(?P<severity>\w+)\] (?P<process>\w+)\[(?P<pid>\d+)\]: (?P<message>.+)$')
       }

      # Coerce parsed fields
      .timestamp = parse_timestamp!(.timestamp, "%Y-%m-%d %H:%M:%S")
      .message = strip_whitespace(.message)
      .uptime = to_float!(.uptime)

[Bouska]

The transform exclusive_route is definitely what I needed. I've added a blackhole sink for the _unmatched to properly "drop" the unwanted events, and no more ERROR nor WARN in my logs 🥳

Thanks!