Protobuf default recursion depth limit of 32 - fails disk buffering, codec, and Vector source/sink

[sbalmos]

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

Vector's protobuf decoder has an implied by-default recursion depth limit of 32. This will cause events that are nested deeper than this limit to fail delivery at any point in a pipeline that uses protobuf encoding - whether protobuf/native codec, using the Vector native source/sink, or using a sink with disk buffering enabled. Depending on the configured retry behavior, an event of this type may cause infinite retries, effectively stopping the pipeline as a poison event.

This is directly related to the default recursion limit of 100-deep in the Prost protobuf crate, and is resolved by enabling the no-recursion-limit prost crate feature.

The error message differs between the usage, whether from a vector source/sink, or from disk buffering.

Vector sink:
2023-12-05T19:13:00.920742Z WARN sink{component_kind="sink" component_id=vector_out component_type=vector}:request{request_id=2}: vector::sinks::util::retries: Retrying after error. error=Request failed: status: Internal, message: "failed to decode Protobuf message: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: ValueMap.fields: Value.kind: Log.fields: EventWrapper.event: PushEventsRequest.events: recursion limit reached", details: [], metadata: MetadataMap { headers: {"content-type": "application/grpc", "date": "Tue, 05 Dec 2023 19:13:00 GMT", "content-length": "0"} } internal_log_rate_limit=true

(note the smoking gun error message of "recursion limit reached")

Any disk-buffered sink:
2023-12-05T19:17:32.250361Z ERROR sink{component_kind="sink" component_id=sender component_type=http}: vector_buffers::internal_events: Error encountered during buffer read. error=failed to decoded record: InvalidProtobufPayload error_code="decode_failed" error_type="reader_failed" stage="processing" internal_log_rate_limit=true

Configuration

Vector source/sink example:
--
test_recursion_sender.yaml:
data_dir: /tmp
api:
  enabled: true
  address: 0.0.0.0:8686
  playground: false
sources:
  stdin:
    type: stdin
transforms:
  parse_json:
    type: remap
    inputs:
      - stdin
    source: |-
      .message = parse_json!(string!(.message), 128)
sinks:
  sender:
    type: vector
    inputs:
      - parse_json
    address: 127.0.0.1:6000

test_recursion_receiver.yaml:
data_dir: /tmp
api:
  enabled: true
  address: 0.0.0.0:8687
  playground: false
sources:
  receiver:
    type: vector
    address: 0.0.0.0:6000
sinks:
  console_out:
    type: console
    target: stdout
    inputs:
      - receiver
    encoding:
      codec: json

Disk buffering example:
--
test_recursion_sender.yaml:
data_dir: /tmp
api:
  enabled: true
  address: 0.0.0.0:8686
  playground: false
sources:
  stdin:
    type: stdin
transforms:
  parse_json:
    type: remap
    inputs:
      - stdin
    source: |-
      .message = parse_json!(string!(.message), 128)
sinks:
  sender:
    type: http
    inputs:
      - parse_json
    uri: http://127.0.0.1:80
    encoding:
      codec: json
    buffer:
      type: disk
      max_size: 268435488

test_recursion_receiver.yaml:
data_dir: /tmp
api:
  enabled: true
  address: 0.0.0.0:8687
  playground: false
sources:
  receiver:
    type: http_server
    address: 0.0.0.0:80
    decoding:
      codec: json
sinks:
  console_out:
    type: console
    target: stdout
    inputs:
      - receiver
    encoding:
      codec: json

Version

vector 0.35.0-custom-ca1b6d14a (x86_64-unknown-linux-gnu debug=full)

Debug Output

No response

Example Data

Any trivial 32+-level-deep JSON structure is usable, such as below.

{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":false}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}

Additional Context

No response

References

#18130

[smo921]

We have been able to reproduce this locally with a trivial configuration on v0.35.1

sources/files.yaml:

type: file
include:
  - /log_files/*
read_from: beginning

transforms/debug.yaml:

type: remap
inputs:
  - files
source: |
  log(., level: "debug", rate_limit_secs: 500)
  
  m, e = parse_json(.message)
  if e == null {
    .message = m
  }
  log(., level: "debug", rate_limit_secs: 500)

sinks/blackhole.yaml

cat vector/sinks/blackhole.yaml 
type: blackhole
inputs:
  - debug
print_interval_secs: 10
buffer:
  max_size: 268435488
  type: disk
  when_full: block

Our test data is the same JSON object as above. The relevant output is captured below.

vector_playground-1  | 2024-04-17T18:39:29.454494Z DEBUG transform{component_kind="transform" component_id=debug component_type=remap}: vrl::stdlib::log: {"file":"/log_files/bad_file.json","host":"60efe62b5dca","message":"{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":{\"hi\":false}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}","source_type":"file","timestamp":"2024-04-17T18:39:29.454399Z"} internal_log_rate_secs=500 vrl_position=0

vector_playground-1  | 2024-04-17T18:39:29.454534Z DEBUG transform{component_kind="transform" component_id=debug component_type=remap}: vrl::stdlib::log: {"file":"/log_files/bad_file.json","host":"60efe62b5dca","message":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":{"hi":false}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}},"source_type":"file","timestamp":"2024-04-17T18:39:29.454399Z"} internal_log_rate_secs=500 vrl_position=106

vector_playground-1  | 2024-04-17T18:39:29.454955Z ERROR sink{component_kind="sink" component_id=blackhole component_type=blackhole}: vector_buffers::internal_events: Error encountered during buffer read. error=failed to decoded record: InvalidProtobufPayload error_code="decode_failed" error_type="reader_failed" stage="processing" internal_log_rate_limit=true

As a work around we have set max_depth: 30 on the parse_json() call. This is similar to the issue noted in #20212

[Keniamin]

Hi! I want to argue "type: enhancement" tag here. From my point of view this is "type: bug", and very dangerous one that must be fixed ASAP. Why? Because it leads to unexpected and unpredictable losing of events:

1. If buffer is not full at the moment when "broken" (too deep) event is written, it stores the event. Then, at the moment of reading, an error occurs here: Error encountered during buffer read. error=failed to decoded record: InvalidProtobufPayload error_code="decode_failed" error_type="reader_failed". This is not good, but acceptable, because there is a loop that goes further and the only lost event is the "broken" one.
2. But if the buffer is full, Vector for some reason immediately checks if event is decodable. For "broken" events this leads to panic with message writer entered inconsistent state.
3. And now - combo! If two such events go in a row and a buffer is nearly full, we get: first event goes into the buffer, second one overflows the buffer and leads to Vector restart, and from this moment Vector is dead because after restart it gets error occurred when building buffer: failed to build individual stage 0: failed to seek to position where writer left off: failed to validate the last written record: failed to decoded record, and fails.
4. The only way to get Vector out of this infinite restart loop is to remove buffer file manually, which leads to loose of all events in it (and remember: buffer is guaranteed to be nearly full at this moment).

This is not just theoretical race condition, but real sad practice. We've put into our Vector a stream from k8s audit log (as in this issue) and Vector started to stuck with broken buffer every 8-10 hours.

Can we discuss, please, how complex would be a full solution of the problem? Is there a possibility that fix can be done and included in the next Vector release?

P.S. The main problem for me is that I collect by Vector many log streams (100+ at the moment) that I do not control. So event with problematic depth may come from any of them at every moment. I'm really not enjoying sitting on a powder keg :) And the only workaround looks like add a limit for parse_json to every of 100+ parser, which is obviously a bit uncomfortable and, you know, just looks weird...

[pront]

Can we discuss, please, how complex would be a full solution of the problem?

Without diving into the source code, I wonder if we can expose the recursion limit as a configuration option for the protobuf codecs. Then we can keep the current default limit and users can overwrite it. Hence why this was marked an enhancement. The event loss sounds like a buffers bug.

Is there a possibility that fix can be done and included in the next Vector release?

That depends how fast we can get a PR tested and merged.

[Keniamin]

First of all, thank you for your attention and fast responce!

Without diving into the source code, I wonder if we can expose the recursion limit as a configuration option for the protobuf codecs.

Yes, I understand this. I'm not really familiar with Rust and its crates, but I see this idea in the issue description:

This is directly related to the default recursion limit of 100-deep in the Prost protobuf crate, and is resolved by enabling the no-recursion-limit prost crate feature.

This sounds like turning on a compilation/building option, and a bit tune on the usage of crate to provide custom limit. Must be quite simple to implement? But I may be wrong about it, of course.

Then we can keep the current default limit and users can overwrite it.

Do you thinks it is neccessary? This limitation sounds like a bit artificial by nature, I don't think someone can really relies on it in any manner. Either user has no problem like this (then raising of the limit makes no difference for him), or he has and would be glad to have no limit, no dropped events and no data loss in buffers :)

The event loss sounds like a buffers bug.

Yes, and it was discussed already, for example, here (the last paragraf and citations below). Of course it would be great to fix this issue and make buffer to behave after restart like in normal: skip only one broken event and go further with others. But:

1. It sounds much more complex. I guess this logic about "failed to seek to position where writer left off" was added not arbitrarily, but because there is a real conceptual problem: if some record is broken how can we determine its size to skip? This requires to introduce something like "magic sync word" into a buffer after every successful recording, which majory changes the buffer structure and leads to complicated code refactoring and migration.

2. After all, this would prevent fatal state and buffer loss after restart, but would not prevent loss of "too nested" events themselves. They still would be skipped either in normal workflow, or after restart.

So I think fixing the root cause of the problem (InvalidProtobufPayload exception) looks like both more simple and more strict solution.

[pront]

This sounds like turning on a compilation/building option, and a bit tune on the usage of crate to provide custom limit. Must be quite simple to implement? But I may be wrong about it, of course.

That is my read on this issue as well. It doesn't seem hard to implement but as always, issues might arise as someone dives into it.

Do you thinks it is necessary?

Yes, we want to enforce bounds everywhere. Those can be very generous but unbounded values are issues waiting to happen. If this is about increasing the limit, I can see how that can improve the situation but it's not a real solution. We need to ensure that if that limit is hit an error is thrown early without affecting anything else in the system (e.g. buffers).

[Keniamin]

Well, while I'm agree in general, from my opinion in this particular case unbounded value is not "an issue waiting to happen", vice versa bounded value is – because I certanly expect deserialize(serialize(data)) to be strictly equal to data, regardless of what exactly stored in data :)

More other, for me as a user of the system these protobufs are a deep implementation detail, I can't affect them or interact with them somehow. I just put events on input (eg. Kafka topic) and wait for them on the other side of pipeline (eg. my custom HTTP endpoint). It's quite strange and unexpected, that I send event with nesting level 15 and it works, nesting level 25 works, and level 35 suddenly lost (even if it would not break Vector instance).

But ok, I understood your position too. Thanks for explanation.

[pront]

I certanly expect deserialize(serialize(data)) to be strictly equal to data, regardless of what exactly stored in data :)

That's the ideal. But even the library we use imposes a max limit. If they remove that, we can also allow users to opt-in for unlimited recursion depths. But to manage expectations, there always will be limits and not just on this layer. Anyway practically, I am optimistic that a configurable recursion limit will solve your issues.

[dobesv]

I think I'm running into this issue now with mongodb log output. A vector agent has stopped sending logs and it's logging that message.

I restarted the vector agent and it seems happy again now, but I guess it will get stuck again if a slow query log comes through with too much nesting in the JSON.

I didn't seem to have any disk buffering in this case, so a restart was enough. For users who have disk buffers this suggested startup script over here might help:

#18130 (comment)