Vector memory explodes with recursion

[MadsRC]

I have a Vector source of type Splunk HEC and the workload that ships data to it sends it to the /services/collector/raw endpoint. The data shipped is newline delimited JSON (fi t helps, it is logs from Cloudflare, using their LogPush to Splunk HEC).

When piping that directly to a sink, Vectors memory usage idles at around 30-40MB. For that I'm using this config:

[sources.http]
type = "splunk_hec"
address = "0.0.0.0:8080"
valid_tokens = [ "${HEC_TOKEN}" ]

[sinks.queue]
type = "kafka"
inputs = [ "filter" ]
bootstrap_servers = "${KAFKA_BROKERS}"
topic = "${KAFKA_TOPIC}"

  [sinks.queue.encoding]
  codec = "json"
  [sinks.queue.tls]
  enabled = true

However, since one big blog of newline delimited JSON is of limited use, I want to explode the ndjson to individual events. When doing that with this config, Vectors memory usage explodes to 5GB+

[sources.http]
type = "splunk_hec"
address = "0.0.0.0:8080"
valid_tokens = [ "${HEC_TOKEN}" ]

[transforms.filter]
type = "remap"
inputs = [ "http" ]
source = '''
  events = []
  buf = split!(.message, "\n")
  map_values(buf) -> |value| {
    if value != "" {
        temp = {}
        temp.event = parse_json!(value)
        events = push(events, temp)
    }
  }
  . = events
'''

[sinks.queue]
type = "kafka"
inputs = [ "filter" ]
bootstrap_servers = "${KAFKA_BROKERS}"
topic = "${KAFKA_TOPIC}"

  [sinks.queue.encoding]
  codec = "json"
  [sinks.queue.tls]
  enabled = true

I've tried replicating it on a local instance of Vector, but I can't seem to replicate it there. The environment where it happens is a Kubernetes cluster running the 0.27.1-debian image (which I also tried using locally to replicate the issue).

Does anyone have any idea why the above config would cause memory to explode like that?

[MadsRC]

Weird thing is that if I replace the single transform with these 3 transforms, memory stays at 30-40MB

    [transforms.split]
    type = "remap"
    inputs = [ "http" ]
    source = '''
      . = split!(.message, "\n")
    '''

    [transforms.filter]
    type = "filter"
    inputs = [ "split" ]
    condition = '.message != ""'

    [transforms.remap]
    type = "remap"
    inputs = [ "filter" ]
    source = '''
      .event = parse_json!(.message)
      del(.message)
    '''

[fuchsnj]

@MadsRC This may be related to https://github.com/vectordotdev/vector/issues/14336 (modifying a value while it's being iterated)

What I think is happening, since you are using map_values to iterate, .message is modified by the return value of the closure in map_values. The return value of

if value != "" {
    temp = {}
    temp.event = parse_json!(value)
    events = push(events, temp)
}

is the entire events array. So I believe your .message is becoming really large. Try returning the original value in the closure to prevent map_values from modifying .message.

Suggestion:

events = []
  buf = split!(.message, "\n")
  map_values(buf) -> |value| {
    if value != "" {
        temp = {}
        temp.event = parse_json!(value)
        events = push(events, temp)
    }
   value
  }
  . = events

There is work to add a recursive version of for_each which would be a better to use, but until then, please try to suggestion above and see if that helps.