Proper use of reduce transform to target go panics

[davidcpell]

I'm trying to use reduce to combine go panics into a single log line with newline delimiters. In my testing, this results in a large drop in events received by the sink:

A cleaned up version my config looks like this:

sources:
  kubernetes:
    type: "kubernetes_logs"
transforms:
  combine_go_panics:
    type: reduce
    inputs:
      - kubernetes
    starts_when:
      type: vrl
      source: |
        starts_with(string!(.message), "panic: runtime error:")
    group_by: ["file"]
    merge_strategies:
      message: concat_newline
    expire_after_ms: 1000
    max_events: 1000
  go_panic_level:
    type: remap
    inputs:
      - combine_go_panics
    source: |
      if starts_with(string!(.message), "panic: runtime error:") {
        .level = "fatal"
      }
sinks:
  http:
    type: "http"
    uri: http://vector-aggregator:${PORT_AGGREGATOR:-8000}
    encoding:
      codec: "json"
    inputs:
    - "go_panic_level"

I was under the impression starts_with would mean an aggregation begins when panic: runtime error is seen and that for any other log a reduce would not begin and it would flow through to the next component as normal. Is that accurate? Any idea on what might be causing what I'm seeing with components received dropping so low on the sink?

[pront]

I was under the impression starts_with would mean an aggregation begins when panic: runtime error is seen and that for any other log a reduce would not begin and it would flow through to the next component as normal. Is that accurate?

Hi @davidcpell, this is correct.

Any idea on what might be causing what I'm seeing with components received dropping so low on the sink?

Looking at your reduce configuration, you can potentially reduce 1000 events to 1 in 1 second intervals. So I would expect a shard decline in component_sent_events_total. Also, the stale_events_flushed_total metric might be interesting.

[pront]

I went back to do an experiment.

Input File

Starting the program...
panic: Boom

goroutine 1 [running]:
main.main()
    /path/to/file.go:7 +0x65
exit status 2
...
panic: Boom

goroutine 2 [running]:
main.main()
    /path/to/file.go:7 +0x65
exit status 2

Vector Config

data_dir: path/data/data_dir_0

sources:
  source_0:
    type: file
    include:
      - path/data/panics.txt
    read_from: "beginning"

transforms:
  t0:
    type: remap
    inputs:
      - source_0
    source: |
      .file = "a"

  combine_go_panics:
    type: reduce
    inputs:
      - t0
    starts_when:
      type: vrl
      source: |
        starts_with(string!(.message), "panic:")
    group_by: ["file"]
    merge_strategies:
      message: concat_newline
    expire_after_ms: 1000
    max_events: 1000


sinks:

  sink_0:
    type: console
    inputs:
      - combine_go_panics
    encoding:
      codec: json
      json:
        pretty: true

Vector Output

2025-03-28T19:29:43.504710Z  INFO source{component_kind="source" component_id=source_0 component_type=file}:file_server: vector::internal_events::file::source: Found new file to watch. file=/Users/pavlos.rontidis/CLionProjects/vector/pront/data/panics.txt
{
  "file": "a",
  "host": "COMP-LPF0JYPP2Q",
  "message": "Starting the program...",
  "source_type": "file",
  "timestamp": "2025-03-28T19:29:43.505538Z",
  "timestamp_end": "2025-03-28T19:29:43.505538Z"
}
{
  "file": "a",
  "host": "COMP-LPF0JYPP2Q",
  "message": "panic: Boom\n\ngoroutine 1 [running]:\nmain.main()\n    /path/to/file.go:7 +0x65\nexit status 2\n...",
  "source_type": "file",
  "timestamp": "2025-03-28T19:29:43.505689Z",
  "timestamp_end": "2025-03-28T19:29:43.505932Z"
}
2025-03-28T19:29:44.515309Z DEBUG vector::internal_events::file::source: Files checkpointed. count=1 duration_ms=7
2025-03-28T19:29:44.546020Z DEBUG source{component_kind="source" component_id=source_0 component_type=file}:file_server: file_source::file_server: event_throughput=14.000/sec bytes_throughput=196.000/sec ratios={"discovery": 0.00012023014, "other": 0.0008569367, "reading": 0.00014852721, "sending": 0.0003028562, "sleeping": 0.9985739}
{
  "file": "a",
  "host": "COMP-LPF0JYPP2Q",
  "message": "panic: Boom\n\ngoroutine 2 [running]:\nmain.main()\n    /path/to/file.go:7 +0x65\nexit status 2",
  "source_type": "file",
  "timestamp": "2025-03-28T19:29:43.505968Z",
  "timestamp_end": "2025-03-28T19:29:43.506156Z"
}

[davidcpell]

Thanks @pront.

Looking at your reduce configuration, you can potentially reduce 1000 events to 1 in 1 second intervals. So I would expect a shard decline in component_sent_events_total. Also, the stale_events_flushed_total metric might be interesting.

This is where I still have a gap in my understanding. I don't want to reduce any events aside from the panics. I used your example and added a second file with normal lines:

info.txt

log 1
log 2
log 3
log 4

When I ran Vector, I saw:

{
  "file": "panics.txt",
  "host": "H577TVK23Y",
  "message": "Starting the program...",
  "source_type": "file",
  "timestamp": "2025-04-02T15:47:18.796726Z",
  "timestamp_end": "2025-04-02T15:47:18.796726Z"
}
{
  "file": "panics.txt",
  "host": "H577TVK23Y",
  "message": "panic: Boom\n\ngoroutine 1 [running]:\nmain.main()\n    /path/to/file.go:7 +0x65\nexit status 2\n...",
  "source_type": "file",
  "timestamp": "2025-04-02T15:47:18.796737Z",
  "timestamp_end": "2025-04-02T15:47:18.796745Z"
}
{
  "file": "info.txt",
  "host": "H577TVK23Y",
  "message": "log 1\nlog 2\nlog 3\nlog 4",
  "source_type": "file",
  "timestamp": "2025-04-02T15:47:18.796755Z",
  "timestamp_end": "2025-04-02T15:47:18.796759Z"
}
{
  "file": "panics.txt",
  "host": "H577TVK23Y",
  "message": "panic: Boom\n\ngoroutine 2 [running]:\nmain.main()\n    /path/to/file.go:7 +0x65\nexit status 2\n",
  "source_type": "file",
  "timestamp": "2025-04-02T15:47:18.796746Z",
  "timestamp_end": "2025-04-02T15:47:18.796754Z"
}

Because of starts_with targeting the first lines of the panics, why are the logs from info.txt also being reduced?

EDIT

I went back and re-read the reduce doc, and it seems like I just misread what starts_with actually does:

A condition used to distinguish the first event of a transaction. If this condition resolves to true for an event, the previous transaction is flushed (without this event) and a new transaction is started.

So for log files that never have a panic, literally every log will be reduced until the timer runs out.

Does what I'm trying to do make sense to you, though? Is there some better way? With reduce it seems like I'd need an ends_with that applies to everything but a panic log, which seems hard to get right.

[pront]

I went back and re-read the reduce doc, and it seems like I just misread what starts_with actually does:

A condition used to distinguish the first event of a transaction. If this condition resolves to true for an event, the previous transaction is flushed (without this event) and a new transaction is started.

So for log files that never have a panic, literally every log will be reduced until the timer runs out.

Correct.

Does what I'm trying to do make sense to you, though? Is there some better way? With reduce it seems like I'd need an ends_with that applies to everything but a panic log, which seems hard to get right.

Right, this is a problem when it comes to aggregating multi-line panic logs. If I owned the files, I would preprocess them (merge into one line or add markers). Otherwise, we might be able to come up with an enhancement for the reduce processor.

[davidcpell]

Thanks @pront. A teammate and I are working on this and he got a lua transform working, but I'm concerned about introducing lua bc we don't use it anywhere else and concerned about the performance warnings and maintaining lua longer-term.

No idea what an enhancement might look like or how hard it would be to add, but an intuitive (to me at least) way to use this would be starts_with and ends_with to indicate when the aggregation should begin and end, with everything else simply passing through the component unchanged. For a panic, the end would be EOF or just a timeout, since after a panic we wouldn't expect more logs to be written to the file.

[pront]

with everything else simply passing through the component unchanged.

I see, basically you want to:

1. tell reduce to start reducing
2. tell reduce to stop reducing
3. if not currently reducing, then just forward events to the next component (I think this is what is missing now)

[davidcpell]

Yes, having thought about this more over the last couple days, I think it's the "Transaction" model used by the transform that confused me. When I just read through the config, I imagined using starts_when: match(.message, r'^panic') and ends_when(.message, r'^exit status') to indicate a transaction begins on the panic lines and ends on the last line of the panic.

But when the documentation starts_when states, "A condition used to distinguish the first event of a transaction", I now understand that the 'first' transaction begins as soon as the first line is processed:

Starting the program...
normal log 1
normal log 2
panic: Boom

goroutine 1 [running]:
main.main()
    /path/to/file.go:7 +0x65
exit status 2

So lines 1-3 are being treated as a transaction already, and then my starts_with targeting line 4 begins a new transaction, which flushes the first transaction which has reduced the first 3 lines into one. I can see uses of that, like collapsing all the logs related to a single request_id or what have you into a single event.

Thinking of the transform as always reducing helps me understand why I'm seeing the results I'm getting. This works fine with the Ruby exception example at the bottom of the doc bc Ruby exceptions always nested every line of the exception using whitespace and the word "from" for lines 2-N, but Go panics don't have a pattern like that, and in my environment things are complicated by the fact that my vector agents are reading files from applications using different languages or, taking Go example, some having adopted our standard internal Go logging package (where we can differentiate panics from non-panics by the non-panics being in JSON) and others not. So what might work to target only go panics in one file will cause problems for other files. Then we get into possibly using a router and it's just a rabbit hole where we would need to do something like add annotations to kubernetes pods giving us the info we'd need to route on, transmitted by the metadata added by the kubernetes_logs source.

[davidcpell]

This seems like it would solve our problem fwiw, since we are using kubernetes_logs: #22689