How to collect multiline logs in custom format

[robb83]

Question

Hi everyone,
I’m looking for suggestions on how to process the attached log file (I receive it via docker logs).

The ideal parsing rule in my view would be:
If a line starts with [Error] or [Info], then the message should continue until the next empty line.
Otherwise, each line should be a separate message.

I’ve already tried implementing a solution myself (attached), but there’s one issue:
it groups even the single-line messages together. It’s not a big problem, but I’d be curious if there’s a cleaner or more robust approach.

Is there a way to pass a tag/marker in the pipeline to indicate that the next message belongs to a multi-line block?

Thanks any suggestions!

Log source:

wall clock duration warning: isolate: f11c111a-11a1-1c11-11a1-1aaf111b11d1
early termination has been triggered: isolate: f11c111a-11a1-1c11-11a1-1aaf111b11d1
serving the request with /home/deno/functions/function-name
[Info] Listening on http://localhost:1111/

[Error] {
  endpoint: "function-name",
  ip: "127.0.0.1",
  trace: "c11f11c1df11c1111c1b1c111dc11111",
  message: {
   _errors: [],
  title: { _errors: [ "String must contain at least 5 character(s)" ] }
 }
}

wall clock duration warning: isolate: f11c111a-11a1-1c11-11a1-1aaf111b11d1
early termination has been triggered: isolate: f11c111a-11a1-1c11-11a1-1aaf111b11d1

Vector Config

sources:
  docker_functions:
    type: docker_logs
    include_containers:
      - edge-functions

transforms:
  functions_logs_remap:
    type: remap
    inputs: [docker_functions]
    source: |-
      .appname = del(.container_name)
      .project = "default"
      del(.container_created_at)
      del(.container_id)
      del(.source_type)
      del(.stream)
      del(.label)
      del(.image)
      del(.host)

  functions_logs_multiline:
    type: reduce
    inputs: [functions_logs_remap]
    group_by: [appname]
    starts_when: match(to_string(.message) ?? "", r'^(?:\s*$|\[(?:Info|Error)\])')
    merge_strategies:
      message: concat_newline
    expire_after_ms: 10000
    flush_period_ms: 1000

  functions_logs_multiline_trim:
    type: remap
    inputs: [functions_logs_multiline]
    source: |-
      .message = strip_whitespace(to_string(.message) ?? "")
      if .message == "" { abort }

Vector Logs

No response