Paused forward input plugin doesn't appear to reject connections

[cjao]

Bug Report

Describe the bug

Thanks for your work on Fluent Bit. I'm evaluating Fluent Bit for an application that can potentially produce large bursts of logs that can occasionally saturate the input buffer for any feasible max buffer size. The desired response to backpressure is to simply slow down log production upstream.

I am having trouble handling backpressure correctly with the forward input plugin. It's not clear to me whether the problem originates in the server, the client, or is caused simply by misconfiguration. The client seems unaware when the forward input plugin pauses due to memory buffer exhaustion. Since message delivery does not appear to fail from the client's perspective, incoming messages can be lost.

To Reproduce

1. Start the fluent bit server with the following config under $(pwd)/etc/fluent-bit.yml

service:
  http_server: on
  http_listen: 0.0.0.0
  http_port: 2020

pipeline:
  inputs:
    - name: forward
      mem_buf_limit: 128kb
      buffer_max_size: 32kb
      buffer_chunk_size: 64kb
  outputs:
    - name: file
      match: 'fluent-client.*'
      format: plain
      path: /var/log/fluentbit-output

docker run --rm --name fluentd -p 24224:24224 -p 2020:2020\
    -v $(pwd)/etc:/etc/fluent-bit \
    -v $(pwd)/fluentbit-output:/var/log/fluentbit-output \
    docker.io/fluent/fluent-bit:latest /fluent-bit/bin/fluent-bit -c /etc/fluent-bit/fluent-bit.yml

2. Run the following fluent client program which sends 1000 16kb messages in rapid succession:

package main

import "log"
import "strings"
import "github.com/fluent/fluent-logger-golang/fluent" // used by the docker fluentd log driver

type Message struct {
	Chunk int
	Log string
}

func main() {
	c := fluent.Config{Async: false, MaxRetry: 5}
	f, err := fluent.New(c)
	if err != nil {
		log.Fatal(err.Error())
	}
	body := strings.Repeat("a", 1024*16)
	var data [1000]Message
	for i:= 0; i < 1000; i++ {
		data[i].Chunk = i
		data[i].Log = body
	}
	for i := 0; i < 1000; i++ {
		err = f.Post("fluent-client.test-128kb", data[i])
		if err != nil {
			log.Fatal(err.Error())
		}
	}
}

go run .

Observe that the server pauses and resumes the forward input plugin as the memory buffer is saturated while no errors are returned to the fluent bit client

[2024/12/17 22:05:18] [ info] [fluent bit] version=3.2.2, commit=a59c867924, pid=1
[2024/12/17 22:05:18] [ info] [storage] ver=1.5.2, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2024/12/17 22:05:18] [ info] [simd    ] disabled
[2024/12/17 22:05:18] [ info] [cmetrics] version=0.9.9
[2024/12/17 22:05:18] [ info] [ctraces ] version=0.5.7
[2024/12/17 22:05:18] [ info] [input:forward:forward.0] initializing
[2024/12/17 22:05:18] [ info] [input:forward:forward.0] storage_strategy='memory' (memory only)
[2024/12/17 22:05:18] [ info] [input:forward:forward.0] listening on 0.0.0.0:24224
[2024/12/17 22:05:18] [ info] [output:file:file.0] worker #0 started
[2024/12/17 22:05:18] [ info] [http_server] listen iface=0.0.0.0 tcp_port=2020
[2024/12/17 22:05:18] [ info] [sp] stream processor started
[2024/12/17 22:05:29] [ warn] [input] forward.0 paused (mem buf overlimit)
[2024/12/17 22:05:29] [ info] [input] pausing forward.0
[2024/12/17 22:05:30] [ info] [input] resume forward.0
[2024/12/17 22:05:30] [ info] [input] forward.0 resume (mem buf overlimit)
[2024/12/17 22:05:30] [ warn] [input] forward.0 paused (mem buf overlimit)
[2024/12/17 22:05:30] [ info] [input] pausing forward.0
[2024/12/17 22:05:31] [ info] [input] resume forward.0
[2024/12/17 22:05:31] [ info] [input] forward.0 resume (mem buf overlimit)
[2024/12/17 22:05:31] [ warn] [input] forward.0 paused (mem buf overlimit)
[2024/12/17 22:05:31] [ info] [input] pausing forward.0
[2024/12/17 22:05:32] [ info] [input] resume forward.0

3. Observe that only a fraction of the logs are actually stored:

$ ls -lh fluentbit-output/fluent-client.test-128kb 
-rw-r--r--. 1 casey casey 385K Dec 17 17:27 fluentbit-output/fluent-client.test-128kb

4. Repeat the same test with a 128mb input buffer, tagging the messages with "fluent-client.test-32mb"

service:
  http_server: on
  http_listen: 0.0.0.0
  http_port: 2020

pipeline:
  inputs:
    - name: forward
      mem_buf_limit: 128mb
      buffer_max_size: 32mb
      buffer_chunk_size: 64mb
  outputs:
    - name: file
      match: 'fluent-client.*'
      format: plain
      path: /var/log/fluentbit-output

$ l fluentbit-output/fluent-client.test-128mb
-rw-r--r--. 1 casey casey 16M Dec 17 17:05 fluentbit-output/fluent-client.test-128mb

Expected behavior
If I understand the docs on buffering and backpressure correctly, the input plugin should refuse new connections while the memory buffer is over capacity. When this happens, I would expect fluent.Post() to retry according to its retry and wait configuration, which would ultimately slow down the upstream log producer so as to avoid losing data. In actuality, the client seems unaware that the fluent-bit server is rejecting input while the forward plugin is paused.

Your Environment

• Version used: latest tag on Docker hub (v3.2.2)
• Operating System and version: Fedora 41
• Filters and plugins: Forward input plugin, file output plugin