daemon/logger/fluentd/fluentd_test.go - third_party/github.com/moby/moby - Git at Google

 package fluentd

 import (
 	"bufio"
 	"context"
 	"net"
 	"path/filepath"
 	"runtime"
 	"sync"
 	"testing"
 	"time"

 	"github.com/google/go-cmp/cmp"
 	"github.com/moby/moby/v2/daemon/logger"
 	"gotest.tools/v3/assert"
 )

 func TestValidateLogOptReconnectInterval(t *testing.T) {
 	invalidIntervals := []string{"-1", "1", "-1s", "99ms", "11s"}
 	for _, v := range invalidIntervals {
 		t.Run("invalid "+v, func(t *testing.T) {
 			err := ValidateLogOpt(map[string]string{asyncReconnectIntervalKey: v})
 			assert.ErrorContains(t, err, "invalid value for fluentd-async-reconnect-interval:")
 		})
 	}

 	validIntervals := []string{"100ms", "10s"}
 	for _, v := range validIntervals {
 		t.Run("valid "+v, func(t *testing.T) {
 			err := ValidateLogOpt(map[string]string{asyncReconnectIntervalKey: v})
 			assert.NilError(t, err)
 		})
 	}
 }

 func TestValidateLogOptAddress(t *testing.T) {
 	// ports to try, and their results
 	validPorts := map[string]int{
 		"":       defaultPort,
 		":":      defaultPort,
 		":123":   123,
 		":65535": 65535,
 	}
 	// paths to try, which should result in an error
 	paths := []string{"/", "/some-path"}

 	tests := []struct {
 		addr        string
 		ports       map[string]int // combinations of addr + port -> expected port
 		paths       []string       // paths to append to addr, should be an error for tcp/udp
 		expected    location
 		expectedErr string
 	}{
 		{
 			addr: "",
 			expected: location{
 				protocol: defaultProtocol,
 				host:     defaultHost,
 				port:     defaultPort,
 			},
 		},
 		{
 			addr:  "192.168.1.1",
 			ports: validPorts,
 			paths: paths,
 			expected: location{
 				protocol: defaultProtocol,
 				host:     "192.168.1.1",
 			},
 		},
 		{
 			addr:  "[::1]",
 			ports: validPorts,
 			paths: paths,
 			expected: location{
 				protocol: defaultProtocol,
 				host:     "::1",
 			},
 		},
 		{
 			addr:  "example.com",
 			ports: validPorts,
 			paths: paths,
 			expected: location{
 				protocol: defaultProtocol,
 				host:     "example.com",
 			},
 		},
 		{
 			addr:  "tcp://",
 			paths: paths,
 			expected: location{
 				protocol: "tcp",
 				host:     defaultHost,
 				port:     defaultPort,
 			},
 		},
 		{
 			addr:  "tcp://example.com",
 			ports: validPorts,
 			paths: paths,
 			expected: location{
 				protocol: "tcp",
 				host:     "example.com",
 			},
 		},
 		{
 			addr:  "tls://",
 			paths: paths,
 			expected: location{
 				protocol: "tls",
 				host:     defaultHost,
 				port:     defaultPort,
 			},
 		},
 		{
 			addr:  "tls://example.com",
 			ports: validPorts,
 			paths: paths,
 			expected: location{
 				protocol: "tls",
 				host:     "example.com",
 			},
 		},
 		{
 			addr:        "://",
 			expectedErr: "missing protocol scheme",
 		},
 		{
 			addr:        "something://",
 			expectedErr: "unsupported scheme: 'something'",
 		},
 		{
 			addr:        "udp://",
 			expectedErr: "unsupported scheme: 'udp'",
 		},
 		{
 			addr:        "unixgram://",
 			expectedErr: "unsupported scheme: 'unixgram'",
 		},
 		{
 			addr:        "tcp+tls://",
 			expectedErr: "unsupported scheme: 'tcp+tls'",
 		},
 		{
 			addr:        "corrupted:c",
 			expectedErr: "invalid port",
 		},
 		{
 			addr:        "tcp://example.com:port",
 			expectedErr: "invalid port",
 		},
 		{
 			addr:        "tcp://example.com:-1",
 			expectedErr: "invalid port",
 		},
 		{
 			addr:        "tcp://example.com:65536",
 			expectedErr: "invalid port",
 		},
 		{
 			addr:        "unix://",
 			expectedErr: "path is empty",
 		},
 		{
 			addr: "unix:///some/socket.sock",
 			expected: location{
 				protocol: "unix",
 				path:     "/some/socket.sock",
 			},
 		},
 		{
 			addr: "unix:///some/socket.sock:80", // unusual, but technically valid
 			expected: location{
 				protocol: "unix",
 				path:     "/some/socket.sock:80",
 			},
 		},
 	}
 	for _, tc := range tests {
 		if len(tc.ports) == 0 {
 			tc.ports = map[string]int{"": tc.expected.port}
 		}

 		// always try empty paths; add paths to try if the test specifies it
 		tc.paths = append([]string{""}, tc.paths...)
 		for port, expectedPort := range tc.ports {
 			for _, path := range tc.paths {
 				address := tc.addr + port + path
 				expected := tc.expected
 				expected.port = expectedPort
 				t.Run(address, func(t *testing.T) {
 					err := ValidateLogOpt(map[string]string{addressKey: address})
 					if path != "" {
 						assert.ErrorContains(t, err, "should not contain a path element")
 						return
 					}
 					if tc.expectedErr != "" {
 						assert.ErrorContains(t, err, tc.expectedErr)
 						return
 					}

 					assert.NilError(t, err)
 					addr, _ := parseAddress(address)
 					assert.DeepEqual(t, expected, *addr, cmp.AllowUnexported(location{}))
 				})
 			}
 		}
 	}
 }

 func TestValidateWriteTimeoutDuration(t *testing.T) {
 	invalidDurations := []string{"-1", "1", "-1s"}
 	for _, d := range invalidDurations {
 		t.Run("invalid "+d, func(t *testing.T) {
 			err := ValidateLogOpt(map[string]string{writeTimeoutKey: d})
 			assert.ErrorContains(t, err, "invalid value for fluentd-write-timeout:")
 		})
 	}

 	validDurations := map[string]time.Duration{
 		"100ms": 100 * time.Millisecond,
 		"10s":   10 * time.Second,
 		"":      0,
 	}
 	for k, v := range validDurations {
 		t.Run("valid "+k, func(t *testing.T) {
 			err := ValidateLogOpt(map[string]string{writeTimeoutKey: k})
 			assert.NilError(t, err)
 			cfg, err := parseConfig(map[string]string{writeTimeoutKey: k})
 			// This check is mostly redundant since it's checked in ValidateLogOpt as well.
 			// This is here to guard against potential regressions in the future.
 			assert.NilError(t, err)
 			assert.Equal(t, cfg.WriteTimeout, v)
 		})
 	}
 }

 // TestReadWriteTimeoutsAreEffective tests that read and write timeout values are effective
 // for fluentd.
 func TestReadWriteTimeoutsAreEffective(t *testing.T) {
 	if runtime.GOOS == "windows" {
 		t.Skip("windows not supported")
 	}

 	for _, tc := range []struct {
 		name              string
 		cfg               map[string]string
 		connectionHandler connHandler
 		rwValidator       func(ctx context.Context, f logger.Logger)
 	}{
 		{
 			// This test case tests that writes timeout when the server is unresponsive.
 			// The test ensures that instead of hanging forever, the fluentd write operation
 			// returns an error when writes cannot be completed within the specified duration.
 			name: "write timeout",
 			cfg: map[string]string{
 				"fluentd-write-timeout": "1ms",
 			},
 			connectionHandler: blackholeConnectionHandler,
 			rwValidator: func(ctx context.Context, f logger.Logger) {
 				// Attempt writing 1MiB worth of log data (all 0's) repeatedly. We should see a failure
 				// after the 1st or the 2nd attempt depending on when the connection's write buffer gets
 				// filled up.
 				// If we don't set a write timeout on the connection, this will hang forever. But, because
 				// we have a write timeout, we expect the `Log` method to return an error.
 				data := make([]byte, 1024*1024)
 				var err error
 				for range 10 {
 					err = f.Log(&logger.Message{
 						Line:      data,
 						Timestamp: time.Now(),
 					})
 					if err != nil {
 						break
 					}
 				}

 				// Checks if the error contains the expected message. The full message is of the format:
 				// "fluent#write: failed to write after %d attempts".
 				assert.ErrorContains(t, err, "fluent#write: failed to write after")
 			},
 		},
 		{
 			// This test case tests that reads timeout when the server is unresponsive and unable to
 			// send acks back.
 			name: "read timeout",
 			cfg: map[string]string{
 				"fluentd-read-timeout": "1ms",
 				"fluentd-request-ack":  "true",
 			},
 			connectionHandler: noAckConnectionHandler,
 			rwValidator: func(ctx context.Context, f logger.Logger) {
 				data := make([]byte, 1024*1024)
 				done := make(chan error, 1)
 				go func() {
 					// Log will hang forever if the read timeout is not set. Hence, we invoke that
 					// asynchronously so that we can timeout the test if something goes wrong.
 					done <- f.Log(&logger.Message{
 						Line:      data,
 						Timestamp: time.Now(),
 					})
 				}()

 				select {
 				case err := <-done:
 					// In an ideal world, we would expect the following error to be returned by fluentd:
 					// "fluent#write: error reading message response ack"
 					// (Ref: https://github.com/fluent/fluent-logger-golang/blob/6b31033c91e794274fd6b77c692412ae945d7d67/fluent/fluent.go#L686)
 					// However, the write method returns a generic message back to the caller and consumes
 					// the original error message without returning it to the caller.
 					// (Ref: https://github.com/fluent/fluent-logger-golang/blob/6b31033c91e794274fd6b77c692412ae945d7d67/fluent/fluent.go#L597)
 					assert.ErrorContains(t, err, "fluent#write: failed to write after 1 attempts")
 				case <-ctx.Done():
 					t.Log("Test timed out, which is unexpected")
 					t.Fail()
 				}
 			},
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			// Create a temporary directory for the socket file.
 			tmpDir := t.TempDir()
 			socketFile := filepath.Join(tmpDir, "fluent-logger-golang.sock")
 			l, err := net.Listen("unix", socketFile)
 			assert.NilError(t, err, "unable to create listener for socket %s", socketFile)
 			defer l.Close()

 			// This is to guard against potential run-away test scenario so that a future change
 			// doesn't cause the tests suite to timeout.
 			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 			defer cancel()

 			var connectedWG sync.WaitGroup
 			connectedWG.Add(1)

 			// Start accepting connections.
 			go acceptConnection(t, ctx, l, tc.connectionHandler, &connectedWG)

 			// Create a base configuration for fluentd logger, agnostic of the test.
 			cfg := map[string]string{
 				"fluentd-address": "unix://" + socketFile,
 				"tag":             "{{.Name}}/{{.FullID}}",
 				// Disabling async behavior with limited retries and buffer size lets us test this in a more
 				// preditable manner for failures. The "fluentd-read-timeout" flag should be equally effective
 				// regardless of async being enabled/disabled.
 				"fluentd-async":        "false",
 				"fluentd-max-retries":  "1",
 				"fluentd-retry-wait":   "10ms",
 				"fluentd-buffer-limit": "1",
 			}
 			// Update the config with test specific configs.
 			for k, v := range tc.cfg {
 				cfg[k] = v
 			}

 			f, err := New(logger.Info{
 				ContainerName: "/test-container",
 				ContainerID:   "container-abcdefghijklmnopqrstuvwxyz01234567890",
 				Config:        cfg,
 			})
 			assert.NilError(t, err)
 			defer closeLoggerWithContext(ctx, f)

 			// Ensure that the server is ready to accept connections since we have disabled async mode
 			// in fluentd options.
 			connectedWG.Wait()

 			tc.rwValidator(ctx, f)
 		})
 	}
 }

 // closeLoggerWithContext enables the caller to return early if logger's Close() method is stuck.
 // Fluentd's Close() will be stuck as long as there's a pending write in progress since the mutex
 // associated with the connection will be locked.
 // Ref: https://github.com/fluent/fluent-logger-golang/blob/6b31033c91e794274fd6b77c692412ae945d7d67/fluent/fluent.go#L600
 func closeLoggerWithContext(ctx context.Context, f logger.Logger) {
 	doneClose := make(chan error, 1)
 	go func() {
 		doneClose <- f.Close()
 	}()
 	// Return if either Close() returns or if the context is done.
 	select {
 	case <-doneClose:
 	case <-ctx.Done():
 	}
 }

 func acceptConnection(
 	t *testing.T,
 	ctx context.Context,
 	l net.Listener,
 	handleConnection connHandler,
 	wg *sync.WaitGroup,
 ) {
 	wg.Done()

 	for {
 		conn, err := l.Accept()
 		if err != nil {
 			// Unable to process the connection. This can happen if the connection is closed.
 			select {
 			case <-ctx.Done():
 				// If the context is canceled, there's nothing for us to do here.
 				return
 			default:
 				t.Logf("Unable to accept connection: %v", err)
 				continue
 			}
 		}

 		// Handle an incoming connection.
 		go handleConnection(ctx, conn)
 	}
 }

 type connHandler func(ctx context.Context, conn net.Conn)

 func blackholeConnectionHandler(ctx context.Context, conn net.Conn) {
 	// Simulate unresponsive server: do nothing with the connection. We're essentially blackholing this
 	// by not reading from or writing to the connection.
 	<-ctx.Done()
 	_ = conn.Close()
 }

 func noAckConnectionHandler(ctx context.Context, conn net.Conn) {
 	// Create a buffered reader to read from the connection.
 	reader := bufio.NewReader(conn)
 	// Read data from the connection.
 	_, err := reader.ReadString('\n')
 	if err != nil {
 		// If there's an error reading from the connection, it means the connection is closed.
 		select {
 		case <-ctx.Done():
 			// If the context is canceled, there's nothing for us to do here.
 			return
 		default:
 			return
 		}
 	}
 	// Don't write an ack back. The fluent configuration is set to expect an ack from the server.
 	<-ctx.Done()
 	_ = conn.Close()
 }
	package fluentd

	import (
	"bufio"
	"context"
	"net"
	"path/filepath"
	"runtime"
	"sync"
	"testing"
	"time"

	"github.com/google/go-cmp/cmp"
	"github.com/moby/moby/v2/daemon/logger"
	"gotest.tools/v3/assert"
	)

	func TestValidateLogOptReconnectInterval(t *testing.T) {
	invalidIntervals := []string{"-1", "1", "-1s", "99ms", "11s"}
	for _, v := range invalidIntervals {
	t.Run("invalid "+v, func(t *testing.T) {
	err := ValidateLogOpt(map[string]string{asyncReconnectIntervalKey: v})
	assert.ErrorContains(t, err, "invalid value for fluentd-async-reconnect-interval:")
	})
	}

	validIntervals := []string{"100ms", "10s"}
	for _, v := range validIntervals {
	t.Run("valid "+v, func(t *testing.T) {
	err := ValidateLogOpt(map[string]string{asyncReconnectIntervalKey: v})
	assert.NilError(t, err)
	})
	}
	}

	func TestValidateLogOptAddress(t *testing.T) {
	// ports to try, and their results
	validPorts := map[string]int{
	"": defaultPort,
	":": defaultPort,
	":123": 123,
	":65535": 65535,
	}
	// paths to try, which should result in an error
	paths := []string{"/", "/some-path"}

	tests := []struct {
	addr string
	ports map[string]int // combinations of addr + port -> expected port
	paths []string // paths to append to addr, should be an error for tcp/udp
	expected location
	expectedErr string
	}{
	{
	addr: "",
	expected: location{
	protocol: defaultProtocol,
	host: defaultHost,
	port: defaultPort,
	},
	},
	{
	addr: "192.168.1.1",
	ports: validPorts,
	paths: paths,
	expected: location{
	protocol: defaultProtocol,
	host: "192.168.1.1",
	},
	},
	{
	addr: "[::1]",
	ports: validPorts,
	paths: paths,
	expected: location{
	protocol: defaultProtocol,
	host: "::1",
	},
	},
	{
	addr: "example.com",
	ports: validPorts,
	paths: paths,
	expected: location{
	protocol: defaultProtocol,
	host: "example.com",
	},
	},
	{
	addr: "tcp://",
	paths: paths,
	expected: location{
	protocol: "tcp",
	host: defaultHost,
	port: defaultPort,
	},
	},
	{
	addr: "tcp://example.com",
	ports: validPorts,
	paths: paths,
	expected: location{
	protocol: "tcp",
	host: "example.com",
	},
	},
	{
	addr: "tls://",
	paths: paths,
	expected: location{
	protocol: "tls",
	host: defaultHost,
	port: defaultPort,
	},
	},
	{
	addr: "tls://example.com",
	ports: validPorts,
	paths: paths,
	expected: location{
	protocol: "tls",
	host: "example.com",
	},
	},
	{
	addr: "://",
	expectedErr: "missing protocol scheme",
	},
	{
	addr: "something://",
	expectedErr: "unsupported scheme: 'something'",
	},
	{
	addr: "udp://",
	expectedErr: "unsupported scheme: 'udp'",
	},
	{
	addr: "unixgram://",
	expectedErr: "unsupported scheme: 'unixgram'",
	},
	{
	addr: "tcp+tls://",
	expectedErr: "unsupported scheme: 'tcp+tls'",
	},
	{
	addr: "corrupted:c",
	expectedErr: "invalid port",
	},
	{
	addr: "tcp://example.com:port",
	expectedErr: "invalid port",
	},
	{
	addr: "tcp://example.com:-1",
	expectedErr: "invalid port",
	},
	{
	addr: "tcp://example.com:65536",
	expectedErr: "invalid port",
	},
	{
	addr: "unix://",
	expectedErr: "path is empty",
	},
	{
	addr: "unix:///some/socket.sock",
	expected: location{
	protocol: "unix",
	path: "/some/socket.sock",
	},
	},
	{
	addr: "unix:///some/socket.sock:80", // unusual, but technically valid
	expected: location{
	protocol: "unix",
	path: "/some/socket.sock:80",
	},
	},
	}
	for _, tc := range tests {
	if len(tc.ports) == 0 {
	tc.ports = map[string]int{"": tc.expected.port}
	}

	// always try empty paths; add paths to try if the test specifies it
	tc.paths = append([]string{""}, tc.paths...)
	for port, expectedPort := range tc.ports {
	for _, path := range tc.paths {
	address := tc.addr + port + path
	expected := tc.expected
	expected.port = expectedPort
	t.Run(address, func(t *testing.T) {
	err := ValidateLogOpt(map[string]string{addressKey: address})
	if path != "" {
	assert.ErrorContains(t, err, "should not contain a path element")
	return
	}
	if tc.expectedErr != "" {
	assert.ErrorContains(t, err, tc.expectedErr)
	return
	}

	assert.NilError(t, err)
	addr, _ := parseAddress(address)
	assert.DeepEqual(t, expected, *addr, cmp.AllowUnexported(location{}))
	})
	}
	}
	}
	}

	func TestValidateWriteTimeoutDuration(t *testing.T) {
	invalidDurations := []string{"-1", "1", "-1s"}
	for _, d := range invalidDurations {
	t.Run("invalid "+d, func(t *testing.T) {
	err := ValidateLogOpt(map[string]string{writeTimeoutKey: d})
	assert.ErrorContains(t, err, "invalid value for fluentd-write-timeout:")
	})
	}

	validDurations := map[string]time.Duration{
	"100ms": 100 * time.Millisecond,
	"10s": 10 * time.Second,
	"": 0,
	}
	for k, v := range validDurations {
	t.Run("valid "+k, func(t *testing.T) {
	err := ValidateLogOpt(map[string]string{writeTimeoutKey: k})
	assert.NilError(t, err)
	cfg, err := parseConfig(map[string]string{writeTimeoutKey: k})
	// This check is mostly redundant since it's checked in ValidateLogOpt as well.
	// This is here to guard against potential regressions in the future.
	assert.NilError(t, err)
	assert.Equal(t, cfg.WriteTimeout, v)
	})
	}
	}

	// TestReadWriteTimeoutsAreEffective tests that read and write timeout values are effective
	// for fluentd.
	func TestReadWriteTimeoutsAreEffective(t *testing.T) {
	if runtime.GOOS == "windows" {
	t.Skip("windows not supported")
	}

	for _, tc := range []struct {
	name string
	cfg map[string]string
	connectionHandler connHandler
	rwValidator func(ctx context.Context, f logger.Logger)
	}{
	{
	// This test case tests that writes timeout when the server is unresponsive.
	// The test ensures that instead of hanging forever, the fluentd write operation
	// returns an error when writes cannot be completed within the specified duration.
	name: "write timeout",
	cfg: map[string]string{
	"fluentd-write-timeout": "1ms",
	},
	connectionHandler: blackholeConnectionHandler,
	rwValidator: func(ctx context.Context, f logger.Logger) {
	// Attempt writing 1MiB worth of log data (all 0's) repeatedly. We should see a failure
	// after the 1st or the 2nd attempt depending on when the connection's write buffer gets
	// filled up.
	// If we don't set a write timeout on the connection, this will hang forever. But, because
	// we have a write timeout, we expect the `Log` method to return an error.
	data := make([]byte, 1024*1024)
	var err error
	for range 10 {
	err = f.Log(&logger.Message{
	Line: data,
	Timestamp: time.Now(),
	})
	if err != nil {
	break
	}
	}

	// Checks if the error contains the expected message. The full message is of the format:
	// "fluent#write: failed to write after %d attempts".
	assert.ErrorContains(t, err, "fluent#write: failed to write after")
	},
	},
	{
	// This test case tests that reads timeout when the server is unresponsive and unable to
	// send acks back.
	name: "read timeout",
	cfg: map[string]string{
	"fluentd-read-timeout": "1ms",
	"fluentd-request-ack": "true",
	},
	connectionHandler: noAckConnectionHandler,
	rwValidator: func(ctx context.Context, f logger.Logger) {
	data := make([]byte, 1024*1024)
	done := make(chan error, 1)
	go func() {
	// Log will hang forever if the read timeout is not set. Hence, we invoke that
	// asynchronously so that we can timeout the test if something goes wrong.
	done <- f.Log(&logger.Message{
	Line: data,
	Timestamp: time.Now(),
	})
	}()

	select {
	case err := <-done:
	// In an ideal world, we would expect the following error to be returned by fluentd:
	// "fluent#write: error reading message response ack"
	// (Ref: https://github.com/fluent/fluent-logger-golang/blob/6b31033c91e794274fd6b77c692412ae945d7d67/fluent/fluent.go#L686)
	// However, the write method returns a generic message back to the caller and consumes
	// the original error message without returning it to the caller.
	// (Ref: https://github.com/fluent/fluent-logger-golang/blob/6b31033c91e794274fd6b77c692412ae945d7d67/fluent/fluent.go#L597)
	assert.ErrorContains(t, err, "fluent#write: failed to write after 1 attempts")
	case <-ctx.Done():
	t.Log("Test timed out, which is unexpected")
	t.Fail()
	}
	},
	},
	} {
	t.Run(tc.name, func(t *testing.T) {
	// Create a temporary directory for the socket file.
	tmpDir := t.TempDir()
	socketFile := filepath.Join(tmpDir, "fluent-logger-golang.sock")
	l, err := net.Listen("unix", socketFile)
	assert.NilError(t, err, "unable to create listener for socket %s", socketFile)
	defer l.Close()

	// This is to guard against potential run-away test scenario so that a future change
	// doesn't cause the tests suite to timeout.
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	var connectedWG sync.WaitGroup
	connectedWG.Add(1)

	// Start accepting connections.
	go acceptConnection(t, ctx, l, tc.connectionHandler, &connectedWG)

	// Create a base configuration for fluentd logger, agnostic of the test.
	cfg := map[string]string{
	"fluentd-address": "unix://" + socketFile,
	"tag": "{{.Name}}/{{.FullID}}",
	// Disabling async behavior with limited retries and buffer size lets us test this in a more
	// preditable manner for failures. The "fluentd-read-timeout" flag should be equally effective
	// regardless of async being enabled/disabled.
	"fluentd-async": "false",
	"fluentd-max-retries": "1",
	"fluentd-retry-wait": "10ms",
	"fluentd-buffer-limit": "1",
	}
	// Update the config with test specific configs.
	for k, v := range tc.cfg {
	cfg[k] = v
	}

	f, err := New(logger.Info{
	ContainerName: "/test-container",
	ContainerID: "container-abcdefghijklmnopqrstuvwxyz01234567890",
	Config: cfg,
	})
	assert.NilError(t, err)
	defer closeLoggerWithContext(ctx, f)

	// Ensure that the server is ready to accept connections since we have disabled async mode
	// in fluentd options.
	connectedWG.Wait()

	tc.rwValidator(ctx, f)
	})
	}
	}

	// closeLoggerWithContext enables the caller to return early if logger's Close() method is stuck.
	// Fluentd's Close() will be stuck as long as there's a pending write in progress since the mutex
	// associated with the connection will be locked.
	// Ref: https://github.com/fluent/fluent-logger-golang/blob/6b31033c91e794274fd6b77c692412ae945d7d67/fluent/fluent.go#L600
	func closeLoggerWithContext(ctx context.Context, f logger.Logger) {
	doneClose := make(chan error, 1)
	go func() {
	doneClose <- f.Close()
	}()
	// Return if either Close() returns or if the context is done.
	select {
	case <-doneClose:
	case <-ctx.Done():
	}
	}

	func acceptConnection(
	t *testing.T,
	ctx context.Context,
	l net.Listener,
	handleConnection connHandler,
	wg *sync.WaitGroup,
	) {
	wg.Done()

	for {
	conn, err := l.Accept()
	if err != nil {
	// Unable to process the connection. This can happen if the connection is closed.
	select {
	case <-ctx.Done():
	// If the context is canceled, there's nothing for us to do here.
	return
	default:
	t.Logf("Unable to accept connection: %v", err)
	continue
	}
	}

	// Handle an incoming connection.
	go handleConnection(ctx, conn)
	}
	}

	type connHandler func(ctx context.Context, conn net.Conn)

	func blackholeConnectionHandler(ctx context.Context, conn net.Conn) {
	// Simulate unresponsive server: do nothing with the connection. We're essentially blackholing this
	// by not reading from or writing to the connection.
	<-ctx.Done()
	_ = conn.Close()
	}

	func noAckConnectionHandler(ctx context.Context, conn net.Conn) {
	// Create a buffered reader to read from the connection.
	reader := bufio.NewReader(conn)
	// Read data from the connection.
	_, err := reader.ReadString('\n')
	if err != nil {
	// If there's an error reading from the connection, it means the connection is closed.
	select {
	case <-ctx.Done():
	// If the context is canceled, there's nothing for us to do here.
	return
	default:
	return
	}
	}
	// Don't write an ack back. The fluent configuration is set to expect an ack from the server.
	<-ctx.Done()
	_ = conn.Close()
	}