From e5a72e404017e72915aed69bec839b17f1b67d13 Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Mon, 29 Jun 2026 11:44:27 +0200 Subject: [PATCH 01/13] Use ephemeral ports in tests instead of pre-allocating them Remove `libbeattesting.MustAvailableTCP4Port` and the related `AvailableTCP4Port`/`AvailableTCP4Ports`/`MustAvailableTCP4Ports` helpers (and their tests). WHY: these helpers found a "free" port by binding `:0`, reading the assigned port, then closing the listener and returning the number. That leaves a time-of-check/time-of-use window: the caller binds the port for real only later, and in between the OS can hand the just-released port to another process. Running the same test many times in parallel (e.g. via `script/stresstest.sh -p 30`) makes those cross-process collisions likely, producing flaky bind failures. WHAT: convert every caller to let the OS assign an ephemeral port at bind time and read the bound port back: - otel integration tests (filebeat/metricbeat/packetbeat) set `http.port: 0` and discover the resolved monitoring port from the "Metrics endpoint listening on:" log line. - New shared helpers do the discovery: `oteltestcol.Collector.MonitoringPort` /`MonitoringPorts` (from the collector's observed logs) and `integration.BeatProc.MonitoringPort` (from the beat process log file), both backed by `libbeat/testing.ParseMonitoringPort` so the log format lives in one place. - packetbeat's test HTTP servers bind ephemeral ports and the sniffer is configured from the resolved port. - the filebeat ES-outage test and the heartbeat connection-refused tests bind `:0` in-process (releasing for the connection-refused / restart cases), removing the early pre-allocation. Verified with `script/stresstest.sh -p 30` (~5 min) on the heartbeat connection-refused tests (0 failures over ~150k/~139k runs) and by running the filebeat/metricbeat/packetbeat otel E2E tests against Elasticsearch. --- filebeat/input/net/tcp/input_test.go | 17 ++- heartbeat/monitors/active/http/http_test.go | 10 +- heartbeat/monitors/active/tcp/tcp_test.go | 12 +- libbeat/testing/available_port.go | 85 ----------- libbeat/testing/available_port_test.go | 52 ------- libbeat/testing/monitoring.go | 53 +++++++ libbeat/tests/integration/framework.go | 26 ++++ .../filebeat/tests/integration/otel_test.go | 121 ++++++++------- .../metricbeat/tests/integration/otel_test.go | 71 ++++----- x-pack/otel/oteltestcol/collector.go | 38 +++++ .../packetbeat/tests/integration/otel_test.go | 139 +++++++++--------- 11 files changed, 312 insertions(+), 312 deletions(-) delete mode 100644 libbeat/testing/available_port.go delete mode 100644 libbeat/testing/available_port_test.go create mode 100644 libbeat/testing/monitoring.go diff --git a/filebeat/input/net/tcp/input_test.go b/filebeat/input/net/tcp/input_test.go index b3bba6cc577c..e9b17148ba80 100644 --- a/filebeat/input/net/tcp/input_test.go +++ b/filebeat/input/net/tcp/input_test.go @@ -23,7 +23,6 @@ import ( "bytes" "context" "errors" - "fmt" "net" "runtime" "sync" @@ -33,7 +32,6 @@ import ( netinput "github.com/elastic/beats/v7/filebeat/input/net" "github.com/elastic/beats/v7/filebeat/input/net/nettest" v2 "github.com/elastic/beats/v7/filebeat/input/v2" - libbeattesting "github.com/elastic/beats/v7/libbeat/testing" conf "github.com/elastic/elastic-agent-libs/config" "github.com/elastic/elastic-agent-libs/logp" "github.com/elastic/elastic-agent-libs/monitoring" @@ -105,11 +103,18 @@ func TestInput(t *testing.T) { } func BenchmarkInput(b *testing.B) { - port, err := libbeattesting.AvailableTCP4Port() + // Bind an ephemeral port to discover a free address, then release it so the + // tcp input can bind it below. Benchmarks run single-process, so the brief + // window between release and re-bind does not cause cross-process port + // collisions. + l, err := net.Listen("tcp", "localhost:0") //nolint:noctx // fine for tests if err != nil { b.Fatalf("cannot find available port: %s", err) } - serverAddr := net.JoinHostPort("localhost", fmt.Sprintf("%d", port)) + serverAddr := l.Addr().String() + if err := l.Close(); err != nil { + b.Fatalf("cannot release port: %s", err) + } inp, err := configure(conf.MustNewConfigFrom(map[string]any{ "host": serverAddr, @@ -141,7 +146,7 @@ func BenchmarkInput(b *testing.B) { }() require.EventuallyWithTf(b, func(ct *assert.CollectT) { - conn, err := net.Dial("tcp", serverAddr) + conn, err := net.Dial("tcp", serverAddr) //nolint:noctx // fine for tests require.NoError(ct, err) conn.Close() }, 30*time.Second, 100*time.Millisecond, "waiting for TCP server to start") @@ -151,7 +156,7 @@ func BenchmarkInput(b *testing.B) { b.ResetTimer() b.RunParallel(func(pb *testing.PB) { for pb.Next() { - conn, err := net.Dial("tcp", serverAddr) + conn, err := net.Dial("tcp", serverAddr) //nolint:noctx // fine for tests if err != nil { b.Errorf("cannot create connection: %s", err) continue diff --git a/heartbeat/monitors/active/http/http_test.go b/heartbeat/monitors/active/http/http_test.go index e55fd2e2e98c..83e1ed12bdf3 100644 --- a/heartbeat/monitors/active/http/http_test.go +++ b/heartbeat/monitors/active/http/http_test.go @@ -57,7 +57,6 @@ import ( "github.com/elastic/beats/v7/heartbeat/scheduler/schedule" "github.com/elastic/beats/v7/libbeat/beat" "github.com/elastic/beats/v7/libbeat/common/file" - btesting "github.com/elastic/beats/v7/libbeat/testing" ) func sendSimpleTLSRequest(t *testing.T, testURL string, useUrls bool) *beat.Event { @@ -610,8 +609,15 @@ func TestHTTPSx509Auth(t *testing.T) { func TestConnRefusedJob(t *testing.T) { ip := "127.0.0.1" - port, err := btesting.AvailableTCP4Port() + // Bind an ephemeral port and release it to obtain an address where + // connections are refused (nothing is listening). Binding to :0 avoids the + // time-of-check/time-of-use race of pre-allocating a fixed port. + l, err := net.Listen("tcp", net.JoinHostPort(ip, "0")) //nolint:noctx // fine for tests require.NoError(t, err) + addr, ok := l.Addr().(*net.TCPAddr) + require.True(t, ok, "expected *net.TCPAddr from listener") + port := uint16(addr.Port) //nolint:gosec // ephemeral port fits in uint16 + require.NoError(t, l.Close()) url := fmt.Sprintf("http://%s:%d", ip, port) event := sendSimpleTLSRequest(t, url, false) diff --git a/heartbeat/monitors/active/tcp/tcp_test.go b/heartbeat/monitors/active/tcp/tcp_test.go index c5cc0dd614e6..d5c127c74b40 100644 --- a/heartbeat/monitors/active/tcp/tcp_test.go +++ b/heartbeat/monitors/active/tcp/tcp_test.go @@ -37,7 +37,6 @@ import ( "github.com/elastic/beats/v7/heartbeat/hbtest" "github.com/elastic/beats/v7/heartbeat/hbtestllext" "github.com/elastic/beats/v7/libbeat/beat" - btesting "github.com/elastic/beats/v7/libbeat/testing" ) func testTCPCheck(t *testing.T, host string, port uint16) *beat.Event { @@ -120,8 +119,15 @@ func TestUpEndpointJob(t *testing.T) { func TestConnectionRefusedEndpointJob(t *testing.T) { ip := "127.0.0.1" - port, err := btesting.AvailableTCP4Port() + // Bind an ephemeral port and release it to obtain an address where + // connections are refused (nothing is listening). Binding to :0 avoids the + // time-of-check/time-of-use race of pre-allocating a fixed port. + l, err := net.Listen("tcp", net.JoinHostPort(ip, "0")) //nolint:noctx // fine for tests require.NoError(t, err) + addr, ok := l.Addr().(*net.TCPAddr) + require.True(t, ok, "expected *net.TCPAddr from listener") + port := uint16(addr.Port) //nolint:gosec // ephemeral port fits in uint16 + require.NoError(t, l.Close()) event := testTCPCheck(t, ip, port) @@ -246,7 +252,7 @@ func TestNXDomainJob(t *testing.T) { // for the specific tests used here. func startEchoServer(t *testing.T) (host string, port uint16, ip string, close func() error, err error) { // Simple echo server - listener, err := net.Listen("tcp", "localhost:0") + listener, err := net.Listen("tcp", "localhost:0") //nolint:noctx // fine for tests if err != nil { return "", 0, "", nil, err } diff --git a/libbeat/testing/available_port.go b/libbeat/testing/available_port.go deleted file mode 100644 index e71fbdc09542..000000000000 --- a/libbeat/testing/available_port.go +++ /dev/null @@ -1,85 +0,0 @@ -// Licensed to Elasticsearch B.V. under one or more contributor -// license agreements. See the NOTICE file distributed with -// this work for additional information regarding copyright -// ownership. Elasticsearch B.V. licenses this file to you under -// the Apache License, Version 2.0 (the "License"); you may -// not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -package testing - -import ( - "fmt" - "net" - "testing" - - "github.com/stretchr/testify/require" -) - -// AvailableTCP4Port returns an unused TCP port for 127.0.0.1. -func AvailableTCP4Port() (uint16, error) { - ports, err := AvailableTCP4Ports(1) - if err != nil { - return 0, err - } - - return ports[0], nil -} - -// AvailableTCP4Ports returns n unique unused TCP ports for 127.0.0.1. -func AvailableTCP4Ports(n int) ([]uint16, error) { - if n <= 0 { - return nil, fmt.Errorf("number of ports must be positive, got %d", n) - } - - listeners := make([]*net.TCPListener, 0, n) - ports := make([]uint16, 0, n) - defer func() { - for _, l := range listeners { - _ = l.Close() - } - }() - - for i := 0; i < n; i++ { - resolved, err := net.ResolveTCPAddr("tcp4", "127.0.0.1:0") - if err != nil { - return nil, err - } - - l, err := net.ListenTCP("tcp4", resolved) - if err != nil { - return nil, err - } - listeners = append(listeners, l) - - tcpAddr, ok := l.Addr().(*net.TCPAddr) - if !ok { - return nil, fmt.Errorf("expected TCP address, got %T", l.Addr()) - } - - ports = append(ports, uint16(tcpAddr.Port)) //nolint:gosec // Safe conversion for port number - } - - return ports, nil -} - -func MustAvailableTCP4Port(t *testing.T) uint16 { - port, err := AvailableTCP4Port() - require.NoError(t, err, "failed to get available TCP4 port") - return port -} - -func MustAvailableTCP4Ports(t *testing.T, n int) []uint16 { - ports, err := AvailableTCP4Ports(n) - require.NoError(t, err, "failed to get available TCP4 ports") - return ports -} diff --git a/libbeat/testing/available_port_test.go b/libbeat/testing/available_port_test.go deleted file mode 100644 index ab639576c753..000000000000 --- a/libbeat/testing/available_port_test.go +++ /dev/null @@ -1,52 +0,0 @@ -// Licensed to Elasticsearch B.V. under one or more contributor -// license agreements. See the NOTICE file distributed with -// this work for additional information regarding copyright -// ownership. Elasticsearch B.V. licenses this file to you under -// the Apache License, Version 2.0 (the "License"); you may -// not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -package testing - -import ( - "fmt" - "net" - "testing" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -func TestAvailableTCP4Ports(t *testing.T) { - ports, err := AvailableTCP4Ports(3) - require.NoError(t, err, "failed to allocate multiple available TCP4 ports") - require.Len(t, ports, 3, "expected three allocated TCP4 ports") - - seen := map[uint16]struct{}{} - for _, port := range ports { - _, exists := seen[port] - assert.False(t, exists, "allocated duplicate TCP4 port: %d", port) - seen[port] = struct{}{} - } - - for _, port := range ports { - listener, err := net.Listen("tcp4", fmt.Sprintf("127.0.0.1:%d", port)) - require.NoError(t, err, "expected to bind allocated TCP4 port %d after helper returns", port) - require.NoError(t, listener.Close(), "failed closing listener for TCP4 port %d", port) - } -} - -func TestAvailableTCP4PortsInvalidCount(t *testing.T) { - ports, err := AvailableTCP4Ports(0) - require.Error(t, err, "expected invalid port count to fail") - assert.Nil(t, ports, "expected no ports for invalid count") -} diff --git a/libbeat/testing/monitoring.go b/libbeat/testing/monitoring.go new file mode 100644 index 000000000000..79ce3f8f568a --- /dev/null +++ b/libbeat/testing/monitoring.go @@ -0,0 +1,53 @@ +// Licensed to Elasticsearch B.V. under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Elasticsearch B.V. licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package testing + +import ( + "fmt" + "net" + "regexp" + "strconv" +) + +// MonitoringEndpointSnippet is the prefix libbeat/api logs when its HTTP +// monitoring server starts (see libbeat/api/server.go). Tests configure +// http.port: 0 and match this snippet to discover the OS-assigned port. +const MonitoringEndpointSnippet = "Metrics endpoint listening on:" + +// e.g. "Metrics endpoint listening on: 127.0.0.1:5067 (configured: localhost)". +var reMonitoringEndpoint = regexp.MustCompile(`Metrics endpoint listening on: (\S+) \(configured:`) + +// ParseMonitoringPort extracts the bound port from a MonitoringEndpointSnippet +// log line. It lets tests read the ephemeral port chosen by the OS instead of +// pre-allocating one, which avoids time-of-check/time-of-use port collisions +// when many tests run in parallel. +func ParseMonitoringPort(logLine string) (int, error) { + matches := reMonitoringEndpoint.FindStringSubmatch(logLine) + if len(matches) != 2 { + return 0, fmt.Errorf("no monitoring address found in log line: %q", logLine) + } + _, portStr, err := net.SplitHostPort(matches[1]) + if err != nil { + return 0, fmt.Errorf("could not split host:port from %q: %w", matches[1], err) + } + port, err := strconv.Atoi(portStr) + if err != nil { + return 0, fmt.Errorf("could not parse port from %q: %w", portStr, err) + } + return port, nil +} diff --git a/libbeat/tests/integration/framework.go b/libbeat/tests/integration/framework.go index 0d9d1eda8ce2..bd6c395fbb9f 100644 --- a/libbeat/tests/integration/framework.go +++ b/libbeat/tests/integration/framework.go @@ -54,6 +54,7 @@ import ( "github.com/elastic/beats/v7/dev-tools/testbin" "github.com/elastic/beats/v7/libbeat/common/proc" + libbeattesting "github.com/elastic/beats/v7/libbeat/testing" "github.com/elastic/go-elasticsearch/v8" "github.com/elastic/mock-es/pkg/api" ) @@ -647,6 +648,31 @@ func (b *BeatProc) WaitLogsContainsAnyOrder(msgs []string, timeout time.Duration ) } +// MonitoringPort waits for the Beat's HTTP monitoring server to log its +// listening address and returns the ephemeral port it bound to. +// +// Configure the Beat with `http.port: 0` so the OS assigns a free port at bind +// time. Reading the port back from the logs avoids the time-of-check/time-of-use +// race of pre-allocating a port, which causes collisions when the same test runs +// many times in parallel (e.g. via script/stresstest.sh). +func (b *BeatProc) MonitoringPort(timeout time.Duration) int { + b.t.Helper() + var port int + require.Eventuallyf(b.t, func() bool { + line := b.GetLogLine(libbeattesting.MonitoringEndpointSnippet) + if line == "" { + return false + } + p, err := libbeattesting.ParseMonitoringPort(line) + if err != nil { + return false + } + port = p + return true + }, timeout, 100*time.Millisecond, "Beat monitoring endpoint did not log its listening address") + return port +} + // WaitForLogsFromBeginning has the same behaviour as WaitForLogs, but it first // resets the log offset. func (b *BeatProc) WaitLogsContainsFromBeginning(s string, timeout time.Duration, msgAndArgs ...any) { diff --git a/x-pack/filebeat/tests/integration/otel_test.go b/x-pack/filebeat/tests/integration/otel_test.go index 504199c73f4a..d78515c519ef 100644 --- a/x-pack/filebeat/tests/integration/otel_test.go +++ b/x-pack/filebeat/tests/integration/otel_test.go @@ -36,7 +36,6 @@ import ( "github.com/gofrs/uuid/v5" "github.com/elastic/beats/v7/libbeat/features" - libbeattesting "github.com/elastic/beats/v7/libbeat/testing" "github.com/elastic/beats/v7/libbeat/tests/integration" "github.com/elastic/beats/v7/x-pack/otel/oteltest" "github.com/elastic/beats/v7/x-pack/otel/oteltestcol" @@ -56,9 +55,6 @@ func TestFilebeatOTelE2E(t *testing.T) { fbOtelIndex := "logs-integration-" + namespace fbIndex := "logs-filebeat-" + namespace - otelMonitoringPort := int(libbeattesting.MustAvailableTCP4Port(t)) - filebeatMonitoringPort := int(libbeattesting.MustAvailableTCP4Port(t)) - otelCfgFile := `receivers: filebeatreceiver: filebeat: @@ -84,7 +80,7 @@ func TestFilebeatOTelE2E(t *testing.T) { path.home: %s http.enabled: true http.host: localhost - http.port: %d + http.port: 0 management.otel.enabled: true exporters: debug: @@ -112,7 +108,11 @@ service: ` logFilePath := filepath.Join(tmpdir, "log.log") writeEventsToLogFile(t, logFilePath, numEvents) - oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, otelMonitoringPort, fbOtelIndex)) + // http.port is set to 0 so the monitoring server binds to an ephemeral + // port, avoiding the TOCTOU race of pre-allocating a port. The actual + // port is then discovered from the collector logs. + collector := oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, fbOtelIndex)) + otelMonitoringPort := collector.MonitoringPort(t) beatsCfgFile := ` filebeat.inputs: @@ -139,7 +139,7 @@ processors: - add_kubernetes_metadata: ~ http.enabled: true http.host: localhost -http.port: %d +http.port: 0 ` // start filebeat @@ -148,12 +148,16 @@ http.port: %d "filebeat", "../../filebeat.test", ) - s := fmt.Sprintf(beatsCfgFile, logFilePath, fbIndex, filebeatMonitoringPort) + s := fmt.Sprintf(beatsCfgFile, logFilePath, fbIndex) filebeat.WriteConfigFile(s) filebeat.Start() defer filebeat.Stop() + // http.port is set to 0 so filebeat's monitoring server binds to an + // ephemeral port; discover the resolved port from the logs. + filebeatMonitoringPort := filebeat.MonitoringPort(30 * time.Second) + // prepare to query ES es := integration.GetESClient(t, "http") @@ -196,6 +200,7 @@ http.port: %d assert.Equal(t, "filebeat", otelDoc.Flatten()["agent.type"], "expected agent.type field to be 'filebeat' in otel docs") assert.Equal(t, "filebeat", filebeatDoc.Flatten()["agent.type"], "expected agent.type field to be 'filebeat' in filebeat docs") assertMonitoring(t, otelMonitoringPort) + assertMonitoring(t, filebeatMonitoringPort) } func TestFilebeatOTelHTTPJSONInput(t *testing.T) { @@ -370,10 +375,9 @@ service: } type multiReceiverConfig struct { - Index int - PathHome string - InputFile string - MonitoringPort int + Index int + PathHome string + InputFile string } func renderOtelConfig(tb testing.TB, cfgTemplate string, data any) string { @@ -446,7 +450,6 @@ func TestFilebeatOTelMultipleReceiversE2E(t *testing.T) { writeEventsToLogFile(t, logFilePath, wantEvents) namespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "") - monitoringPorts := libbeattesting.MustAvailableTCP4Ports(t, 2) otelConfig := struct { Index string Receivers []multiReceiverConfig @@ -454,18 +457,18 @@ func TestFilebeatOTelMultipleReceiversE2E(t *testing.T) { Index: "logs-integration-" + namespace, Receivers: []multiReceiverConfig{ { - MonitoringPort: int(monitoringPorts[0]), - InputFile: logFilePath, - PathHome: filepath.Join(tmpdir, "r1"), + InputFile: logFilePath, + PathHome: filepath.Join(tmpdir, "r1"), }, { - MonitoringPort: int(monitoringPorts[1]), - InputFile: logFilePath, - PathHome: filepath.Join(tmpdir, "r2"), + InputFile: logFilePath, + PathHome: filepath.Join(tmpdir, "r2"), }, }, } + // http.port is 0 so each receiver's monitoring server binds an ephemeral + // port; the actual ports are discovered from the collector logs below. cfg := renderOtelConfig(t, `receivers: {{range $i, $receiver := .Receivers}} filebeatreceiver/{{$i}}: @@ -484,11 +487,9 @@ func TestFilebeatOTelMultipleReceiversE2E(t *testing.T) { - '*' queue.mem.flush.timeout: 0s path.home: {{$receiver.PathHome}} -{{if $receiver.MonitoringPort}} http.enabled: true http.host: localhost - http.port: {{$receiver.MonitoringPort}} -{{end}} + http.port: 0 {{end}} exporters: debug: @@ -524,7 +525,7 @@ service: writeEventsToLogFile(t, logFilePath, wantEvents) - oteltestcol.New(t, cfg) + collector := oteltestcol.New(t, cfg) es := integration.GetESClient(t, "http") @@ -544,8 +545,8 @@ service: assert.GreaterOrEqual(ct, otelDocs.Hits.Total.Value, wantTotalLogs, "expected at least %d events, got %d", wantTotalLogs, otelDocs.Hits.Total.Value) }, 2*time.Minute, 100*time.Millisecond, "expected at least %d events from multiple receivers", wantTotalLogs) - for _, rec := range otelConfig.Receivers { - assertMonitoring(t, rec.MonitoringPort) + for _, port := range collector.MonitoringPorts(t, len(otelConfig.Receivers)) { + assertMonitoring(t, port) } } @@ -786,21 +787,21 @@ func TestFilebeatOTelDocumentLevelRetries(t *testing.T) { index := "logs-integration-" + namespace beatsConfig := struct { - Index string - InputFile string - ESEndpoint string - MaxRetries int - MonitoringPort int - RetryOnStatus string + Index string + InputFile string + ESEndpoint string + MaxRetries int + RetryOnStatus string }{ - Index: index, - InputFile: filepath.Join(t.TempDir(), "log.log"), - ESEndpoint: server.URL, - MaxRetries: tt.maxRetries, - MonitoringPort: int(libbeattesting.MustAvailableTCP4Port(t)), - RetryOnStatus: tt.retryOnStatus, + Index: index, + InputFile: filepath.Join(t.TempDir(), "log.log"), + ESEndpoint: server.URL, + MaxRetries: tt.maxRetries, + RetryOnStatus: tt.retryOnStatus, } + // http.port is 0 so the monitoring server binds an ephemeral port; + // the actual port is discovered from the collector logs below. cfg := `receivers: filebeatreceiver: filebeat: @@ -818,7 +819,7 @@ func TestFilebeatOTelDocumentLevelRetries(t *testing.T) { setup.template.enabled: false http.enabled: true http.host: localhost - http.port: {{.MonitoringPort}} + http.port: 0 exporters: elasticsearch: auth: @@ -874,6 +875,7 @@ service: template.Must(template.New("config").Parse(cfg)).Execute(&configBuffer, beatsConfig)) collector := oteltestcol.New(t, configBuffer.String()) + monitoringPort := collector.MonitoringPort(t) writeEventsToLogFile(t, beatsConfig.InputFile, numTestEvents) // Wait for file input to be fully read @@ -926,7 +928,7 @@ service: // Confirm filebeat agreed with our accounting of ingested events require.EventuallyWithT(t, func(ct *assert.CollectT) { - address := fmt.Sprintf("http://localhost:%d", beatsConfig.MonitoringPort) + address := fmt.Sprintf("http://localhost:%d", monitoringPort) r, err := http.Get(address + "/stats") //nolint:noctx,bodyclose // fine for tests assert.NoError(ct, err) assert.Equal(ct, http.StatusOK, r.StatusCode, "incorrect status code") @@ -1260,8 +1262,9 @@ func TestNoDuplicates(t *testing.T) { namespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "") fbOtelIndex := "logs-integration-" + namespace - otelMonitoringPort := int(libbeattesting.MustAvailableTCP4Port(t)) - + // http.port is 0 so the monitoring server binds an ephemeral port. This + // test does not assert on the monitoring endpoint, so the resolved port is + // not needed. otelCfgFile := `receivers: filebeatreceiver: filebeat: @@ -1285,7 +1288,7 @@ func TestNoDuplicates(t *testing.T) { path.home: %s http.enabled: true http.host: localhost - http.port: %d + http.port: 0 management.otel.enabled: true exporters: elasticsearch/log: @@ -1348,7 +1351,7 @@ service: close(stopChan) wg.Wait() }) - collector := oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, otelMonitoringPort, fbOtelIndex)) + collector := oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, fbOtelIndex)) require.EventuallyWithT(t, func(ct *assert.CollectT) { @@ -1380,7 +1383,7 @@ service: ) // restart the collector process - collector = oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, otelMonitoringPort, fbOtelIndex)) + collector = oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, fbOtelIndex)) t.Cleanup(func() { collector.Shutdown() }) @@ -1516,7 +1519,16 @@ func TestFilebeatOTelNoEventLossDuringESOutage(t *testing.T) { namespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "") index := "logs-integration-" + namespace - serverPort := int(libbeattesting.MustAvailableTCP4Port(t)) + // Reserve an ephemeral port for the mock Elasticsearch server. The server + // is stopped and restarted on the same port to simulate an ES outage, so + // the port must be stable: bind to :0 to let the OS choose a free port, + // then release it so the mock server can claim it on demand. + portListener, err := net.Listen("tcp", "localhost:0") //nolint:noctx // it's okay for testing purposes + require.NoError(t, err) + portAddr, ok := portListener.Addr().(*net.TCPAddr) + require.True(t, ok, "expected *net.TCPAddr from listener") + serverPort := portAddr.Port + require.NoError(t, portListener.Close()) serverURL := fmt.Sprintf("http://localhost:%d", serverPort) var ingestedEvents []string @@ -1560,17 +1572,18 @@ func TestFilebeatOTelNoEventLossDuringESOutage(t *testing.T) { } beatsConfig := struct { - Index string - InputFile string - ESEndpoint string - MonitoringPort int + Index string + InputFile string + ESEndpoint string }{ - Index: index, - InputFile: logFilePath, - ESEndpoint: serverURL, - MonitoringPort: int(libbeattesting.MustAvailableTCP4Port(t)), + Index: index, + InputFile: logFilePath, + ESEndpoint: serverURL, } + // http.port is 0 so the monitoring server binds an ephemeral port. This + // test asserts on the collector's observed logs, not the monitoring + // endpoint, so the resolved port is not needed. cfg := `receivers: filebeatreceiver: filebeat: @@ -1588,7 +1601,7 @@ func TestFilebeatOTelNoEventLossDuringESOutage(t *testing.T) { setup.template.enabled: false http.enabled: true http.host: localhost - http.port: {{.MonitoringPort}} + http.port: 0 exporters: elasticsearch: auth: diff --git a/x-pack/metricbeat/tests/integration/otel_test.go b/x-pack/metricbeat/tests/integration/otel_test.go index 82d91905bd98..a0a93b1344fe 100644 --- a/x-pack/metricbeat/tests/integration/otel_test.go +++ b/x-pack/metricbeat/tests/integration/otel_test.go @@ -22,7 +22,6 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - libbeattesting "github.com/elastic/beats/v7/libbeat/testing" "github.com/elastic/beats/v7/libbeat/tests/integration" "github.com/elastic/beats/v7/x-pack/otel/oteltestcol" "github.com/elastic/elastic-agent-libs/mapstr" @@ -41,23 +40,21 @@ func TestMetricbeatOTelE2E(t *testing.T) { mbIndex := "logs-integration-mb-" + namespace mbReceiverIndex := "logs-integration-mbreceiver-" + namespace - otelMonitoringPort := int(libbeattesting.MustAvailableTCP4Port(t)) - metricbeatMonitoringPort := int(libbeattesting.MustAvailableTCP4Port(t)) - otelConfig := struct { - Index string - ESURL string - Username string - Password string - MonitoringPort int + Index string + ESURL string + Username string + Password string }{ - Index: mbReceiverIndex, - ESURL: fmt.Sprintf("%s://%s", host.Scheme, host.Host), - Username: user, - Password: password, - MonitoringPort: otelMonitoringPort, + Index: mbReceiverIndex, + ESURL: fmt.Sprintf("%s://%s", host.Scheme, host.Host), + Username: user, + Password: password, } + // http.port is 0 so the monitoring servers bind ephemeral ports, avoiding + // the TOCTOU race of pre-allocating ports. The metricbeat process port is + // discovered from its logs; the otel receiver port is not asserted here. cfg := `receivers: metricbeatreceiver: metricbeat: @@ -82,7 +79,7 @@ func TestMetricbeatOTelE2E(t *testing.T) { setup.template.enabled: false http.enabled: true http.host: localhost - http.port: {{.MonitoringPort}} + http.port: 0 management.otel.enabled: true exporters: debug: @@ -146,7 +143,7 @@ processors: - add_kubernetes_metadata: ~ http.enabled: true http.host: localhost -http.port: {{.MonitoringPort}} +http.port: 0 ` es := integration.GetESClient(t, "http") @@ -161,17 +158,15 @@ http.port: {{.MonitoringPort}} var mbConfigBuffer bytes.Buffer require.NoError(t, template.Must(template.New("config").Parse(beatsCfgFile)).Execute(&mbConfigBuffer, struct { - Index string - ESURL string - Username string - Password string - MonitoringPort int + Index string + ESURL string + Username string + Password string }{ - Index: mbIndex, - ESURL: fmt.Sprintf("%s://%s", host.Scheme, host.Host), - Username: user, - Password: password, - MonitoringPort: metricbeatMonitoringPort, + Index: mbIndex, + ESURL: fmt.Sprintf("%s://%s", host.Scheme, host.Host), + Username: user, + Password: password, })) metricbeat := integration.NewBeat(t, "metricbeat", "../../metricbeat.test") @@ -179,6 +174,9 @@ http.port: {{.MonitoringPort}} metricbeat.Start() defer metricbeat.Stop() + // http.port is 0, so discover the ephemeral monitoring port from the logs. + metricbeatMonitoringPort := metricbeat.MonitoringPort(30 * time.Second) + // Make sure find the logs var metricbeatDocs estools.Documents var otelDocs estools.Documents @@ -232,9 +230,8 @@ func TestMetricbeatOTelMultipleReceiversE2E(t *testing.T) { password, _ := host.User.Password() type receiverConfig struct { - MonitoringPort int - InputFile string - PathHome string + InputFile string + PathHome string } es := integration.GetESClient(t, "http") @@ -261,12 +258,10 @@ func TestMetricbeatOTelMultipleReceiversE2E(t *testing.T) { Password: password, Receivers: []receiverConfig{ { - MonitoringPort: int(libbeattesting.MustAvailableTCP4Port(t)), - PathHome: filepath.Join(tmpDir, "r1"), + PathHome: filepath.Join(tmpDir, "r1"), }, { - MonitoringPort: int(libbeattesting.MustAvailableTCP4Port(t)), - PathHome: filepath.Join(tmpDir, "r2"), + PathHome: filepath.Join(tmpDir, "r2"), }, }, } @@ -295,11 +290,9 @@ func TestMetricbeatOTelMultipleReceiversE2E(t *testing.T) { queue.mem.flush.timeout: 0s path.home: {{$receiver.PathHome}} management.otel.enabled: true -{{if $receiver.MonitoringPort}} http.enabled: true http.host: localhost - http.port: {{$receiver.MonitoringPort}} -{{end}} + http.port: 0 {{end}} exporters: debug: @@ -338,7 +331,7 @@ service: } }) - oteltestcol.New(t, string(configContents)) + collector := oteltestcol.New(t, string(configContents)) var r0Docs, r1Docs estools.Documents var err error @@ -371,8 +364,8 @@ service: }, 1*time.Minute, 100*time.Millisecond, "expected at least 1 log for each receiver") assertMapstrKeysEqual(t, r0Docs.Hits.Hits[0].Source, r1Docs.Hits.Hits[0].Source, nil, "expected documents keys to be equal") - for _, rec := range otelConfig.Receivers { - assertMonitoring(t, rec.MonitoringPort) + for _, port := range collector.MonitoringPorts(t, len(otelConfig.Receivers)) { + assertMonitoring(t, port) } } diff --git a/x-pack/otel/oteltestcol/collector.go b/x-pack/otel/oteltestcol/collector.go index 16f34235dead..69bb33912e21 100644 --- a/x-pack/otel/oteltestcol/collector.go +++ b/x-pack/otel/oteltestcol/collector.go @@ -12,6 +12,7 @@ import ( "testing" "time" + libbeattesting "github.com/elastic/beats/v7/libbeat/testing" "github.com/elastic/beats/v7/libbeat/version" "github.com/elastic/beats/v7/x-pack/auditbeat/abreceiver" "github.com/elastic/beats/v7/x-pack/filebeat/fbreceiver" @@ -105,6 +106,43 @@ func (c *Collector) Shutdown() { c.collector.Shutdown() } +// MonitoringPort waits for a single Beat receiver HTTP monitoring server to log +// its listening address and returns the ephemeral port it bound to. +// +// Configure the receiver with `http.host: localhost` and `http.port: 0` so the +// OS assigns a free port at bind time. Reading the port back from the logs +// avoids the time-of-check/time-of-use race of pre-allocating a port. +func (c *Collector) MonitoringPort(tb testing.TB) int { + tb.Helper() + return c.MonitoringPorts(tb, 1)[0] +} + +// MonitoringPorts waits until at least n Beat receiver HTTP monitoring servers +// have logged their listening addresses and returns the ephemeral ports they +// bound to. The ports are returned in the order they were logged; callers that +// only assert each endpoint works do not need a per-receiver mapping. +func (c *Collector) MonitoringPorts(tb testing.TB, n int) []int { + tb.Helper() + var ports []int + require.EventuallyWithT(tb, func(ct *assert.CollectT) { + seen := make(map[int]struct{}) + ports = ports[:0] + for _, entry := range c.observer.FilterMessageSnippet(libbeattesting.MonitoringEndpointSnippet).All() { + port, err := libbeattesting.ParseMonitoringPort(entry.Message) + if !assert.NoError(ct, err) { + continue + } + if _, ok := seen[port]; ok { + continue + } + seen[port] = struct{}{} + ports = append(ports, port) + } + assert.GreaterOrEqualf(ct, len(ports), n, "waiting for %d monitoring endpoints to start", n) + }, 30*time.Second, 100*time.Millisecond, "collector monitoring endpoints did not start") + return ports[:n] +} + func getComponent() (otelcol.Factories, error) { receivers, err := otelcol.MakeFactoryMap( abreceiver.NewFactory(), diff --git a/x-pack/packetbeat/tests/integration/otel_test.go b/x-pack/packetbeat/tests/integration/otel_test.go index b8ec9609620e..d300e25f7fdb 100644 --- a/x-pack/packetbeat/tests/integration/otel_test.go +++ b/x-pack/packetbeat/tests/integration/otel_test.go @@ -24,7 +24,6 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - libbeattesting "github.com/elastic/beats/v7/libbeat/testing" "github.com/elastic/beats/v7/libbeat/tests/integration" "github.com/elastic/beats/v7/x-pack/otel/oteltest" "github.com/elastic/beats/v7/x-pack/otel/oteltestcol" @@ -49,22 +48,28 @@ func skipIfNotRoot(t *testing.T) { } } -// startHTTPServer starts an HTTP server on the given port and registers -// cleanup to stop it when the test ends. -func startHTTPServer(t *testing.T, port uint16) *httptest.Server { +// startHTTPServer starts an HTTP server bound to an ephemeral port on the +// loopback interface and registers cleanup to stop it when the test ends. +// Binding to an OS-assigned port avoids the time-of-check/time-of-use race of +// pre-allocating a port. +func startHTTPServer(t *testing.T) *httptest.Server { t.Helper() - listener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", port)) //nolint:noctx // fine for tests - require.NoError(t, err) - srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) fmt.Fprintln(w, "packetbeat integration test") })) - srv.Listener = listener - srv.Start() t.Cleanup(srv.Close) return srv } +// serverPort returns the ephemeral TCP port the test HTTP server bound to. +func serverPort(t *testing.T, srv *httptest.Server) int { + t.Helper() + addr, ok := srv.Listener.Addr().(*net.TCPAddr) + require.Truef(t, ok, "expected a TCP listener address, got %T", srv.Listener.Addr()) + return addr.Port +} + // sendHTTPRequests sends numRequests GET requests to url, ignoring errors. func sendHTTPRequests(url string, numRequests int) { client := &http.Client{Timeout: 5 * time.Second} @@ -131,8 +136,8 @@ func TestPacketbeatOTelE2E(t *testing.T) { namespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "") index := "logs-integration-" + namespace - monitoringPort := int(libbeattesting.MustAvailableTCP4Port(t)) - + // http.port is 0 so the monitoring server binds an ephemeral port; the + // actual port is discovered from the collector logs below. cfg := fmt.Sprintf(`receivers: packetbeatreceiver: packetbeat: @@ -150,7 +155,7 @@ func TestPacketbeatOTelE2E(t *testing.T) { path.home: %s http.enabled: true http.host: localhost - http.port: %d + http.port: 0 management.otel.enabled: true exporters: elasticsearch/log: @@ -171,9 +176,9 @@ service: - packetbeatreceiver exporters: - elasticsearch/log -`, tmpdir, monitoringPort, index) +`, tmpdir, index) - oteltestcol.New(t, cfg) + collector := oteltestcol.New(t, cfg) es := integration.GetESClient(t, "http") @@ -188,15 +193,14 @@ service: }, 2*time.Minute, 1*time.Second, "expected packetbeat events in ES") - assertMonitoring(t, monitoringPort) + assertMonitoring(t, collector.MonitoringPort(t)) } type receiverConfig struct { - PathHome string - PcapFile string - Protocol string - Ports []int - MonitoringPort uint16 + PathHome string + PcapFile string + Protocol string + Ports []int } // TestPacketbeatOTelMultipleReceiversE2E verifies that multiple packetbeat @@ -210,24 +214,23 @@ func TestPacketbeatOTelMultipleReceiversE2E(t *testing.T) { namespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "") index := "logs-integration-" + namespace - ports := libbeattesting.MustAvailableTCP4Ports(t, 2) receivers := []receiverConfig{ { - PathHome: tmpdir + "/r0", - PcapFile: "../../../../packetbeat/tests/system/pcaps/http_x_forwarded_for.pcap", - Protocol: "http", - Ports: []int{80}, - MonitoringPort: ports[0], + PathHome: tmpdir + "/r0", + PcapFile: "../../../../packetbeat/tests/system/pcaps/http_x_forwarded_for.pcap", + Protocol: "http", + Ports: []int{80}, }, { - PathHome: tmpdir + "/r1", - PcapFile: "../../../../packetbeat/tests/system/pcaps/dns_google_com.pcap", - Protocol: "dns", - Ports: []int{53}, - MonitoringPort: ports[1], + PathHome: tmpdir + "/r1", + PcapFile: "../../../../packetbeat/tests/system/pcaps/dns_google_com.pcap", + Protocol: "dns", + Ports: []int{53}, }, } + // http.port is 0 so each receiver's monitoring server binds an ephemeral + // port; the actual ports are discovered from the collector logs below. cfg := renderOtelConfig(t, `receivers: {{range $i, $r := .Receivers}} packetbeatreceiver/{{$i}}: @@ -247,7 +250,7 @@ func TestPacketbeatOTelMultipleReceiversE2E(t *testing.T) { path.home: {{$r.PathHome}} http.enabled: true http.host: localhost - http.port: {{$r.MonitoringPort}} + http.port: 0 management.otel.enabled: true {{end}} exporters: @@ -276,7 +279,7 @@ service: "Receivers": receivers, }) - oteltestcol.New(t, cfg) + collector := oteltestcol.New(t, cfg) es := integration.GetESClient(t, "http") @@ -293,8 +296,8 @@ service: }, 2*time.Minute, 1*time.Second, "expected events from %d receivers in ES", len(receivers)) - for _, rec := range receivers { - assertMonitoring(t, int(rec.MonitoringPort)) + for _, port := range collector.MonitoringPorts(t, len(receivers)) { + assertMonitoring(t, port) } } @@ -310,8 +313,8 @@ func TestPacketbeatOTelBeatE2E(t *testing.T) { pbOtelIndex := "logs-integration-" + namespace pbIndex := "logs-packetbeat-" + namespace - otelMonitoringPort := int(libbeattesting.MustAvailableTCP4Port(t)) - + // http.port is 0 so the monitoring server binds an ephemeral port; the + // actual port is discovered from the collector logs below. otelCfg := fmt.Sprintf(`receivers: packetbeatreceiver: packetbeat: @@ -329,7 +332,7 @@ func TestPacketbeatOTelBeatE2E(t *testing.T) { path.home: %s http.enabled: true http.host: localhost - http.port: %d + http.port: 0 management.otel.enabled: true exporters: elasticsearch/log: @@ -350,9 +353,9 @@ service: - packetbeatreceiver exporters: - elasticsearch/log -`, tmpdir, otelMonitoringPort, pbOtelIndex) +`, tmpdir, pbOtelIndex) - oteltestcol.New(t, otelCfg) + collector := oteltestcol.New(t, otelCfg) standaloneCfg := fmt.Sprintf(` packetbeat.interfaces.file: ../../../../packetbeat/tests/system/pcaps/http_x_forwarded_for.pcap @@ -411,7 +414,7 @@ queue.mem.flush.timeout: 0s assert.Equal(t, "packetbeat", otelDoc.Flatten()["agent.type"], "expected agent.type to be 'packetbeat' in otel doc") assert.Equal(t, "packetbeat", pbDoc.Flatten()["agent.type"], "expected agent.type to be 'packetbeat' in standalone doc") - assertMonitoring(t, otelMonitoringPort) + assertMonitoring(t, collector.MonitoringPort(t)) } // TestPacketbeatOTelLiveInterfaceE2E verifies that a packetbeat OTel receiver @@ -425,11 +428,11 @@ func TestPacketbeatOTelLiveInterfaceE2E(t *testing.T) { namespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "") index := "logs-integration-" + namespace - ports := libbeattesting.MustAvailableTCP4Ports(t, 2) - monitoringPort := ports[0] - httpPort := ports[1] - - srv := startHTTPServer(t, httpPort) + // The HTTP server binds an ephemeral port; packetbeat is configured to + // sniff that resolved port. http.port is 0 so the monitoring server also + // binds an ephemeral port, discovered from the collector logs below. + srv := startHTTPServer(t) + httpPort := serverPort(t, srv) cfg := renderOtelConfig(t, `receivers: packetbeatreceiver: @@ -447,7 +450,7 @@ func TestPacketbeatOTelLiveInterfaceE2E(t *testing.T) { path.home: {{.PathHome}} http.enabled: true http.host: localhost - http.port: {{.MonitoringPort}} + http.port: 0 exporters: debug: use_internal_logger: false @@ -477,14 +480,13 @@ service: metrics: level: none `, map[string]any{ - "Device": loopbackDevice(), - "HTTPPort": httpPort, - "PathHome": tmpdir, - "MonitoringPort": monitoringPort, - "Index": index, + "Device": loopbackDevice(), + "HTTPPort": httpPort, + "PathHome": tmpdir, + "Index": index, }) - oteltestcol.New(t, cfg) + collector := oteltestcol.New(t, cfg) // Generate HTTP traffic continuously: the packetbeat sniffer starts // asynchronously, so a one-shot burst would race with capture startup. @@ -503,7 +505,7 @@ service: }, 2*time.Minute, 1*time.Second, "expected packetbeat events in ES") - assertMonitoring(t, int(monitoringPort)) + assertMonitoring(t, collector.MonitoringPort(t)) } // TestPacketbeatOTelLiveInterfaceMultipleReceiversE2E verifies that multiple @@ -518,22 +520,17 @@ func TestPacketbeatOTelLiveInterfaceMultipleReceiversE2E(t *testing.T) { namespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "") index := "logs-integration-" + namespace - // 4 ports: 2 monitoring + 2 HTTP servers - ports := libbeattesting.MustAvailableTCP4Ports(t, 4) - + // Each HTTP server binds an ephemeral port; packetbeat sniffs the resolved + // ports. http.port is 0 so each monitoring server binds an ephemeral port + // too, discovered from the collector logs below. type liveReceiverConfig struct { - PathHome string - HTTPPort uint16 - MonitoringPort uint16 + PathHome string + HTTPPort int } + servers := []*httptest.Server{startHTTPServer(t), startHTTPServer(t)} liveReceivers := []liveReceiverConfig{ - {PathHome: tmpdir + "/r0", MonitoringPort: ports[0], HTTPPort: ports[2]}, - {PathHome: tmpdir + "/r1", MonitoringPort: ports[1], HTTPPort: ports[3]}, - } - - servers := make([]*httptest.Server, len(liveReceivers)) - for i, rec := range liveReceivers { - servers[i] = startHTTPServer(t, rec.HTTPPort) + {PathHome: tmpdir + "/r0", HTTPPort: serverPort(t, servers[0])}, + {PathHome: tmpdir + "/r1", HTTPPort: serverPort(t, servers[1])}, } cfg := renderOtelConfig(t, `receivers: @@ -554,7 +551,7 @@ func TestPacketbeatOTelLiveInterfaceMultipleReceiversE2E(t *testing.T) { path.home: {{$r.PathHome}} http.enabled: true http.host: localhost - http.port: {{$r.MonitoringPort}} + http.port: 0 {{end}} exporters: debug: @@ -592,7 +589,7 @@ service: "Receivers": liveReceivers, }) - oteltestcol.New(t, cfg) + collector := oteltestcol.New(t, cfg) // Generate HTTP traffic continuously on each server: the packetbeat sniffer // starts asynchronously, so a one-shot burst would race with capture startup. @@ -615,7 +612,7 @@ service: }, 2*time.Minute, 1*time.Second, "expected events from %d receivers in ES", len(liveReceivers)) - for _, rec := range liveReceivers { - assertMonitoring(t, int(rec.MonitoringPort)) + for _, port := range collector.MonitoringPorts(t, len(liveReceivers)) { + assertMonitoring(t, port) } } From 7a0db431b1f1182af99c5158b966e36b167b03f9 Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Mon, 29 Jun 2026 12:44:41 +0200 Subject: [PATCH 02/13] Make test OTel collector parallel-safe The test OTel collector defaulted to a Prometheus telemetry reader on a fixed localhost:8888, so two collectors could not run at once. Stress runs (stresstest.sh -p N) of any collector-based integration test therefore collided on that port. Disable the collector's own telemetry metrics by default in oteltestcol.New (service.telemetry.metrics.level: none) unless the test set a level; the tests assert on Elasticsearch documents and the per-receiver HTTP monitoring endpoint, not the collector's own metrics, so this is safe. Collector.Shutdown now blocks until the collector has fully exited so callers can restart a collector that reuses the same path.home without racing teardown. TestNoDuplicates previously polled :8888 to detect that the previous collector had exited before restarting; the blocking Shutdown replaces that, removing the last fixed-port dependency and letting these tests run in parallel. --- .../filebeat/tests/integration/otel_test.go | 19 ++----- x-pack/otel/oteltestcol/collector.go | 54 ++++++++++++++----- 2 files changed, 46 insertions(+), 27 deletions(-) diff --git a/x-pack/filebeat/tests/integration/otel_test.go b/x-pack/filebeat/tests/integration/otel_test.go index d78515c519ef..7f07ff2b80c6 100644 --- a/x-pack/filebeat/tests/integration/otel_test.go +++ b/x-pack/filebeat/tests/integration/otel_test.go @@ -1365,23 +1365,12 @@ service: 1*time.Minute, 1*time.Second, "expected more than 0 events, got none", ) + // Shutdown blocks until the collector has fully exited, so the restart below + // cannot race the previous instance's teardown (the previous version polled + // the collector's fixed :8888 telemetry port as an exit signal, which is now + // disabled to keep collectors parallel-safe). collector.Shutdown() - // wait for 8888 port to be free (an indication that previous collector has exited) - require.Eventually(t, - func() bool { - ln, err := net.Listen("tcp", "localhost:8888") //nolint:noctx // it's okay for testing purposes - if err != nil { - return false - } - ln.Close() - return true - }, - 10*time.Second, - 100*time.Millisecond, - "port 8888 never became available", - ) - // restart the collector process collector = oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, fbOtelIndex)) t.Cleanup(func() { diff --git a/x-pack/otel/oteltestcol/collector.go b/x-pack/otel/oteltestcol/collector.go index 69bb33912e21..2d78f98f6583 100644 --- a/x-pack/otel/oteltestcol/collector.go +++ b/x-pack/otel/oteltestcol/collector.go @@ -43,11 +43,17 @@ import ( ) type Collector struct { - collector *otelcol.Collector - observer *observer.ObservedLogs + collector *otelcol.Collector + observer *observer.ObservedLogs + done chan struct{} + shutdownOnce sync.Once } // New creates and starts a new OTel collector for testing. +// +// The collector's own telemetry metrics are disabled (see metricsOffConfig) so +// multiple collectors can run in parallel, e.g. under script/stresstest.sh, +// without colliding on the fixed Prometheus port. func New(tb testing.TB, configYAML string) *Collector { tb.Helper() @@ -60,6 +66,12 @@ func New(tb testing.TB, configYAML string) *Collector { tb.Fatalf("failed to create collector: %v", err) } + // Merged after the test config to disable the collector's own telemetry + // metrics; kept in a separate file so we don't have to parse/rewrite the + // test's YAML. + metricsOffFile := filepath.Join(configDir, "metrics-off.yaml") + require.NoError(tb, os.WriteFile(metricsOffFile, []byte(metricsOffConfig), 0o644)) + var zapBuf zaptest.Buffer zapCore := zapcore.NewCore( zapcore.NewJSONEncoder(zap.NewProductionEncoderConfig()), @@ -69,23 +81,22 @@ func New(tb testing.TB, configYAML string) *Collector { observed, observer := observer.New(zapcore.DebugLevel) core := zapcore.NewTee(zapCore, observed) - settings := newCollectorSettings("file:"+configFile, core) + settings := newCollectorSettings([]string{"file:" + configFile, "file:" + metricsOffFile}, core) col, err := otelcol.NewCollector(settings) require.NoError(tb, err) - var wg sync.WaitGroup + c := &Collector{collector: col, observer: observer, done: make(chan struct{})} + tb.Cleanup(func() { - col.Shutdown() - wg.Wait() + c.Shutdown() if tb.Failed() { tb.Log("OTel Collector logs:\n" + zapBuf.String()) } }) - wg.Add(1) go func() { - defer wg.Done() + defer close(c.done) ctx, cancel := signal.NotifyContext(tb.Context(), os.Interrupt) defer cancel() assert.NoError(tb, col.Run(ctx)) @@ -95,17 +106,34 @@ func New(tb testing.TB, configYAML string) *Collector { return col.GetState() == otelcol.StateRunning }, 10*time.Second, 10*time.Millisecond, "Collector did not start in time") - return &Collector{collector: col, observer: observer} + return c } func (c *Collector) ObservedLogs() *observer.ObservedLogs { return c.observer } +// Shutdown stops the collector and blocks until it has fully exited. Blocking +// lets callers restart a collector that reuses the same path.home (or other +// resources) without racing the previous instance's teardown. It is safe to +// call multiple times. func (c *Collector) Shutdown() { - c.collector.Shutdown() + c.shutdownOnce.Do(c.collector.Shutdown) + <-c.done } +// metricsOffConfig is merged on top of the test config to turn off the +// collector's own telemetry metrics. The collector otherwise starts a +// Prometheus reader on a fixed localhost:8888, which collides when collectors +// run in parallel (e.g. under stresstest.sh). Tests assert on Elasticsearch +// documents and the per-receiver HTTP monitoring endpoint rather than the +// collector's own metrics, so disabling them is safe. +const metricsOffConfig = `service: + telemetry: + metrics: + level: none +` + // MonitoringPort waits for a single Beat receiver HTTP monitoring server to log // its listening address and returns the ephemeral port it bound to. // @@ -190,7 +218,9 @@ func getComponent() (otelcol.Factories, error) { }, nil } -func newCollectorSettings(filename string, core zapcore.Core) otelcol.CollectorSettings { +// newCollectorSettings builds the collector settings from the given config +// URIs, which are resolved and deep-merged in order (later URIs win). +func newCollectorSettings(uris []string, core zapcore.Core) otelcol.CollectorSettings { return otelcol.CollectorSettings{ BuildInfo: component.BuildInfo{ Command: "otel", @@ -205,7 +235,7 @@ func newCollectorSettings(filename string, core zapcore.Core) otelcol.CollectorS }, ConfigProviderSettings: otelcol.ConfigProviderSettings{ ResolverSettings: confmap.ResolverSettings{ - URIs: []string{filename}, + URIs: uris, ProviderFactories: []confmap.ProviderFactory{ fileprovider.NewFactory(), }, From caf0c61b0abb74c71d6f49132bc2376717b519c1 Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Mon, 29 Jun 2026 14:36:42 +0200 Subject: [PATCH 03/13] Tidy ephemeral-port test helpers Small follow-up cleanups with no behavior change: - oteltestcol.New: drop the unreachable `if err != nil` block after require.NoError and fold the WriteFile call into the assertion. - libbeat/testing.monitoring: build the endpoint regex from MonitoringEndpointSnippet via regexp.QuoteMeta so the literal lives in one place. - TestNoDuplicates: drop the redundant t.Cleanup after the collector restart; oteltestcol.New already registers shutdown via t.Cleanup. --- libbeat/testing/monitoring.go | 2 +- x-pack/filebeat/tests/integration/otel_test.go | 8 +++----- x-pack/otel/oteltestcol/collector.go | 7 +------ 3 files changed, 5 insertions(+), 12 deletions(-) diff --git a/libbeat/testing/monitoring.go b/libbeat/testing/monitoring.go index 79ce3f8f568a..57b193f70c3b 100644 --- a/libbeat/testing/monitoring.go +++ b/libbeat/testing/monitoring.go @@ -30,7 +30,7 @@ import ( const MonitoringEndpointSnippet = "Metrics endpoint listening on:" // e.g. "Metrics endpoint listening on: 127.0.0.1:5067 (configured: localhost)". -var reMonitoringEndpoint = regexp.MustCompile(`Metrics endpoint listening on: (\S+) \(configured:`) +var reMonitoringEndpoint = regexp.MustCompile(regexp.QuoteMeta(MonitoringEndpointSnippet) + ` (\S+) \(configured:`) // ParseMonitoringPort extracts the bound port from a MonitoringEndpointSnippet // log line. It lets tests read the ephemeral port chosen by the OS instead of diff --git a/x-pack/filebeat/tests/integration/otel_test.go b/x-pack/filebeat/tests/integration/otel_test.go index 7f07ff2b80c6..32d8771c7602 100644 --- a/x-pack/filebeat/tests/integration/otel_test.go +++ b/x-pack/filebeat/tests/integration/otel_test.go @@ -1371,11 +1371,9 @@ service: // disabled to keep collectors parallel-safe). collector.Shutdown() - // restart the collector process - collector = oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, fbOtelIndex)) - t.Cleanup(func() { - collector.Shutdown() - }) + // Restart the collector. New registers shutdown via t.Cleanup, so the + // returned value is not needed here. + oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, fbOtelIndex)) // wait for more docs to be published. require.EventuallyWithTf(t, diff --git a/x-pack/otel/oteltestcol/collector.go b/x-pack/otel/oteltestcol/collector.go index 2d78f98f6583..4321e22683fc 100644 --- a/x-pack/otel/oteltestcol/collector.go +++ b/x-pack/otel/oteltestcol/collector.go @@ -59,12 +59,7 @@ func New(tb testing.TB, configYAML string) *Collector { configDir := tb.TempDir() configFile := filepath.Join(configDir, "otel.yaml") - err := os.WriteFile(configFile, []byte(configYAML), 0o644) - require.NoError(tb, err) - - if err != nil { - tb.Fatalf("failed to create collector: %v", err) - } + require.NoError(tb, os.WriteFile(configFile, []byte(configYAML), 0o644)) // Merged after the test config to disable the collector's own telemetry // metrics; kept in a separate file so we don't have to parse/rewrite the From 6b02d5ec70201e32dd6b1217f89205c64cdcc6ab Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Mon, 29 Jun 2026 14:44:43 +0200 Subject: [PATCH 04/13] Drop redundant per-config telemetry metrics-off blocks oteltestcol.New now merges service.telemetry.metrics.level: none into every test collector config, so the per-config blocks that disabled the collector's own metrics (to avoid the fixed :8888 Prometheus port) are redundant. Remove them. Where a config only set the metrics level, the whole telemetry block is dropped; where it also set telemetry.logs.level, only the metrics entry is removed so the log level is preserved. --- .../tests/integration/otel_cel_test.go | 3 --- .../tests/integration/otel_kafka_test.go | 3 --- .../tests/integration/otel_lsexporter_test.go | 6 ----- .../filebeat/tests/integration/otel_test.go | 26 ------------------- .../packetbeat/tests/integration/otel_test.go | 4 --- 5 files changed, 42 deletions(-) diff --git a/x-pack/filebeat/tests/integration/otel_cel_test.go b/x-pack/filebeat/tests/integration/otel_cel_test.go index 13cd1bac4ef8..42a5659317e7 100644 --- a/x-pack/filebeat/tests/integration/otel_cel_test.go +++ b/x-pack/filebeat/tests/integration/otel_cel_test.go @@ -143,9 +143,6 @@ service: - elasticsearch receivers: - filebeatreceiver - telemetry: - metrics: - level: none ` var configBuffer bytes.Buffer diff --git a/x-pack/filebeat/tests/integration/otel_kafka_test.go b/x-pack/filebeat/tests/integration/otel_kafka_test.go index 47f44164586f..b0ce6264f701 100644 --- a/x-pack/filebeat/tests/integration/otel_kafka_test.go +++ b/x-pack/filebeat/tests/integration/otel_kafka_test.go @@ -71,9 +71,6 @@ service: - filebeatreceiver exporters: - kafka - telemetry: - metrics: - level: none `, logFilePath, tmpdir, kafkaBroker, otelTopic) oteltestcol.New(t, otelCfg) diff --git a/x-pack/filebeat/tests/integration/otel_lsexporter_test.go b/x-pack/filebeat/tests/integration/otel_lsexporter_test.go index 3a46d1122c3a..e2c9bc307823 100644 --- a/x-pack/filebeat/tests/integration/otel_lsexporter_test.go +++ b/x-pack/filebeat/tests/integration/otel_lsexporter_test.go @@ -126,9 +126,6 @@ service: - filebeatreceiver exporters: - logstash - telemetry: - metrics: - level: none `, inputFilePath, testCaseName, tmpdir) // Start OTel collector with filebeatreceiver @@ -232,9 +229,6 @@ service: - filebeatreceiver exporters: - logstash - telemetry: - metrics: - level: none `, inputFilePath, testCaseName, tmpdir) // Start OTel collector with filebeatreceiver diff --git a/x-pack/filebeat/tests/integration/otel_test.go b/x-pack/filebeat/tests/integration/otel_test.go index 32d8771c7602..a9c9a59a4a7b 100644 --- a/x-pack/filebeat/tests/integration/otel_test.go +++ b/x-pack/filebeat/tests/integration/otel_test.go @@ -519,8 +519,6 @@ service: telemetry: logs: level: DEBUG - metrics: - level: none `, otelConfig) writeEventsToLogFile(t, logFilePath, wantEvents) @@ -867,8 +865,6 @@ service: telemetry: logs: level: DEBUG - metrics: - level: none ` var configBuffer bytes.Buffer require.NoError(t, @@ -1093,9 +1089,6 @@ service: exporters: - elasticsearch/log - debug - telemetry: - metrics: - level: none # Disable collector's own metrics to prevent conflict on port 8888. We don't use those metrics anyway. receivers: filebeatreceiver: filebeat: @@ -1161,9 +1154,6 @@ service: exporters: - elasticsearch/log - debug - telemetry: - metrics: - level: none # Disable collector's own metrics to prevent conflict on port 8888. We don't use those metrics anyway. receivers: filebeatreceiver: filebeat: @@ -1633,8 +1623,6 @@ service: telemetry: logs: level: DEBUG - metrics: - level: none ` var configBuffer bytes.Buffer @@ -1769,8 +1757,6 @@ service: telemetry: logs: level: DEBUG - metrics: - level: none `, struct { Receivers []multiReceiverConfig InputFile string @@ -1903,8 +1889,6 @@ service: telemetry: logs: level: DEBUG - metrics: - level: none ` var configBuffer bytes.Buffer @@ -2014,8 +1998,6 @@ service: telemetry: logs: level: DEBUG - metrics: - level: none `, struct{ Receivers []multiReceiverConfig }{Receivers: receivers}) b.StartTimer() @@ -2068,8 +2050,6 @@ service: telemetry: logs: level: DEBUG - metrics: - level: none `, struct { PathHome string EventCount int @@ -2181,8 +2161,6 @@ service: telemetry: logs: level: warn - metrics: - level: none `, struct { EventCount int PathHome string @@ -2246,8 +2224,6 @@ func TestBeatProcessorSharedAcrossPipelines(t *testing.T) { telemetry: logs: level: debug - metrics: - level: none receivers: filebeatreceiver/1: filebeat: @@ -2314,8 +2290,6 @@ func TestBeatProcessorWhenCondition(t *testing.T) { telemetry: logs: level: debug - metrics: - level: none receivers: filebeatreceiver: filebeat: diff --git a/x-pack/packetbeat/tests/integration/otel_test.go b/x-pack/packetbeat/tests/integration/otel_test.go index d300e25f7fdb..46f5d301bf71 100644 --- a/x-pack/packetbeat/tests/integration/otel_test.go +++ b/x-pack/packetbeat/tests/integration/otel_test.go @@ -477,8 +477,6 @@ service: telemetry: logs: level: DEBUG - metrics: - level: none `, map[string]any{ "Device": loopbackDevice(), "HTTPPort": httpPort, @@ -581,8 +579,6 @@ service: telemetry: logs: level: DEBUG - metrics: - level: none `, map[string]any{ "Device": loopbackDevice(), "Index": index, From e115ab438e2779153cb52d9b7b31123dc7172c10 Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Mon, 29 Jun 2026 16:17:07 +0200 Subject: [PATCH 05/13] Cleanup BenchmarkInput --- filebeat/input/net/tcp/input_test.go | 32 +++++++++++++-------- heartbeat/monitors/active/http/http_test.go | 6 ++-- 2 files changed, 22 insertions(+), 16 deletions(-) diff --git a/filebeat/input/net/tcp/input_test.go b/filebeat/input/net/tcp/input_test.go index e9b17148ba80..1a503b4a3fee 100644 --- a/filebeat/input/net/tcp/input_test.go +++ b/filebeat/input/net/tcp/input_test.go @@ -103,18 +103,7 @@ func TestInput(t *testing.T) { } func BenchmarkInput(b *testing.B) { - // Bind an ephemeral port to discover a free address, then release it so the - // tcp input can bind it below. Benchmarks run single-process, so the brief - // window between release and re-bind does not cause cross-process port - // collisions. - l, err := net.Listen("tcp", "localhost:0") //nolint:noctx // fine for tests - if err != nil { - b.Fatalf("cannot find available port: %s", err) - } - serverAddr := l.Addr().String() - if err := l.Close(); err != nil { - b.Fatalf("cannot release port: %s", err) - } + serverAddr := ephemeralTCPAddr(b) inp, err := configure(conf.MustNewConfigFrom(map[string]any{ "host": serverAddr, @@ -182,3 +171,22 @@ func BenchmarkInput(b *testing.B) { } }) } + +// ephemeralTCPAddr binds an ephemeral localhost port, immediately releases +// it, and returns the resolved "host:port" so a caller can configure a +// server to listen on it. +// +// WARNING: racy by design. The port can become unavailable. +func ephemeralTCPAddr(tb testing.TB) string { + tb.Helper() + var lc net.ListenConfig + l, err := lc.Listen(tb.Context(), "tcp", "localhost:0") + if err != nil { + tb.Fatalf("cannot bind an ephemeral port: %s", err) + } + addr := l.Addr().String() + if err := l.Close(); err != nil { + tb.Fatalf("cannot release ephemeral port %s: %s", addr, err) + } + return addr +} diff --git a/heartbeat/monitors/active/http/http_test.go b/heartbeat/monitors/active/http/http_test.go index 83e1ed12bdf3..4a331b2da5da 100644 --- a/heartbeat/monitors/active/http/http_test.go +++ b/heartbeat/monitors/active/http/http_test.go @@ -609,10 +609,8 @@ func TestHTTPSx509Auth(t *testing.T) { func TestConnRefusedJob(t *testing.T) { ip := "127.0.0.1" - // Bind an ephemeral port and release it to obtain an address where - // connections are refused (nothing is listening). Binding to :0 avoids the - // time-of-check/time-of-use race of pre-allocating a fixed port. - l, err := net.Listen("tcp", net.JoinHostPort(ip, "0")) //nolint:noctx // fine for tests + var lc net.ListenConfig + l, err := lc.Listen(t.Context(), "tcp", net.JoinHostPort(ip, "0")) require.NoError(t, err) addr, ok := l.Addr().(*net.TCPAddr) require.True(t, ok, "expected *net.TCPAddr from listener") From 198800fe6dfdbe184721b331477ace66321534a2 Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Mon, 29 Jun 2026 16:30:20 +0200 Subject: [PATCH 06/13] comments --- x-pack/filebeat/tests/integration/otel_test.go | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/x-pack/filebeat/tests/integration/otel_test.go b/x-pack/filebeat/tests/integration/otel_test.go index a9c9a59a4a7b..216a7848c6f7 100644 --- a/x-pack/filebeat/tests/integration/otel_test.go +++ b/x-pack/filebeat/tests/integration/otel_test.go @@ -108,9 +108,6 @@ service: ` logFilePath := filepath.Join(tmpdir, "log.log") writeEventsToLogFile(t, logFilePath, numEvents) - // http.port is set to 0 so the monitoring server binds to an ephemeral - // port, avoiding the TOCTOU race of pre-allocating a port. The actual - // port is then discovered from the collector logs. collector := oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, fbOtelIndex)) otelMonitoringPort := collector.MonitoringPort(t) @@ -154,8 +151,6 @@ http.port: 0 filebeat.Start() defer filebeat.Stop() - // http.port is set to 0 so filebeat's monitoring server binds to an - // ephemeral port; discover the resolved port from the logs. filebeatMonitoringPort := filebeat.MonitoringPort(30 * time.Second) // prepare to query ES @@ -1002,10 +997,10 @@ func TestFileBeatKerberos(t *testing.T) { file_identity.native: ~ queue.mem.flush.timeout: 0s management.otel.enabled: true - path.home: {{.PathHome}} + path.home: {{.PathHome}} extensions: beatsauth: - kerberos: + kerberos: auth_type: "password" config_path: "../../../../libbeat/outputs/elasticsearch/testdata/krb5.conf" username: "beats" @@ -1022,7 +1017,7 @@ exporters: auth: authenticator: beatsauth service: - extensions: + extensions: - beatsauth pipelines: logs: From c6516cefca77e3e513408d97ddf43424307eb4fe Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Mon, 29 Jun 2026 16:35:46 +0200 Subject: [PATCH 07/13] comments --- x-pack/filebeat/tests/integration/otel_test.go | 10 ---------- x-pack/metricbeat/tests/integration/otel_test.go | 4 ---- x-pack/packetbeat/tests/integration/otel_test.go | 6 ------ 3 files changed, 20 deletions(-) diff --git a/x-pack/filebeat/tests/integration/otel_test.go b/x-pack/filebeat/tests/integration/otel_test.go index 216a7848c6f7..89accca6cb95 100644 --- a/x-pack/filebeat/tests/integration/otel_test.go +++ b/x-pack/filebeat/tests/integration/otel_test.go @@ -462,8 +462,6 @@ func TestFilebeatOTelMultipleReceiversE2E(t *testing.T) { }, } - // http.port is 0 so each receiver's monitoring server binds an ephemeral - // port; the actual ports are discovered from the collector logs below. cfg := renderOtelConfig(t, `receivers: {{range $i, $receiver := .Receivers}} filebeatreceiver/{{$i}}: @@ -793,8 +791,6 @@ func TestFilebeatOTelDocumentLevelRetries(t *testing.T) { RetryOnStatus: tt.retryOnStatus, } - // http.port is 0 so the monitoring server binds an ephemeral port; - // the actual port is discovered from the collector logs below. cfg := `receivers: filebeatreceiver: filebeat: @@ -1247,9 +1243,6 @@ func TestNoDuplicates(t *testing.T) { namespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "") fbOtelIndex := "logs-integration-" + namespace - // http.port is 0 so the monitoring server binds an ephemeral port. This - // test does not assert on the monitoring endpoint, so the resolved port is - // not needed. otelCfgFile := `receivers: filebeatreceiver: filebeat: @@ -1553,9 +1546,6 @@ func TestFilebeatOTelNoEventLossDuringESOutage(t *testing.T) { ESEndpoint: serverURL, } - // http.port is 0 so the monitoring server binds an ephemeral port. This - // test asserts on the collector's observed logs, not the monitoring - // endpoint, so the resolved port is not needed. cfg := `receivers: filebeatreceiver: filebeat: diff --git a/x-pack/metricbeat/tests/integration/otel_test.go b/x-pack/metricbeat/tests/integration/otel_test.go index a0a93b1344fe..55ed14073c0c 100644 --- a/x-pack/metricbeat/tests/integration/otel_test.go +++ b/x-pack/metricbeat/tests/integration/otel_test.go @@ -52,9 +52,6 @@ func TestMetricbeatOTelE2E(t *testing.T) { Password: password, } - // http.port is 0 so the monitoring servers bind ephemeral ports, avoiding - // the TOCTOU race of pre-allocating ports. The metricbeat process port is - // discovered from its logs; the otel receiver port is not asserted here. cfg := `receivers: metricbeatreceiver: metricbeat: @@ -174,7 +171,6 @@ http.port: 0 metricbeat.Start() defer metricbeat.Stop() - // http.port is 0, so discover the ephemeral monitoring port from the logs. metricbeatMonitoringPort := metricbeat.MonitoringPort(30 * time.Second) // Make sure find the logs diff --git a/x-pack/packetbeat/tests/integration/otel_test.go b/x-pack/packetbeat/tests/integration/otel_test.go index 46f5d301bf71..51902c71f081 100644 --- a/x-pack/packetbeat/tests/integration/otel_test.go +++ b/x-pack/packetbeat/tests/integration/otel_test.go @@ -136,8 +136,6 @@ func TestPacketbeatOTelE2E(t *testing.T) { namespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "") index := "logs-integration-" + namespace - // http.port is 0 so the monitoring server binds an ephemeral port; the - // actual port is discovered from the collector logs below. cfg := fmt.Sprintf(`receivers: packetbeatreceiver: packetbeat: @@ -229,8 +227,6 @@ func TestPacketbeatOTelMultipleReceiversE2E(t *testing.T) { }, } - // http.port is 0 so each receiver's monitoring server binds an ephemeral - // port; the actual ports are discovered from the collector logs below. cfg := renderOtelConfig(t, `receivers: {{range $i, $r := .Receivers}} packetbeatreceiver/{{$i}}: @@ -313,8 +309,6 @@ func TestPacketbeatOTelBeatE2E(t *testing.T) { pbOtelIndex := "logs-integration-" + namespace pbIndex := "logs-packetbeat-" + namespace - // http.port is 0 so the monitoring server binds an ephemeral port; the - // actual port is discovered from the collector logs below. otelCfg := fmt.Sprintf(`receivers: packetbeatreceiver: packetbeat: From 9e7a2dbf8f658ed2aef3de7b2cd018fd853318bd Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Mon, 29 Jun 2026 16:42:39 +0200 Subject: [PATCH 08/13] Move monitoring.go --- libbeat/tests/integration/framework.go | 5 ++--- libbeat/{testing => tests/integration}/monitoring.go | 2 +- x-pack/otel/oteltestcol/collector.go | 6 +++--- 3 files changed, 6 insertions(+), 7 deletions(-) rename libbeat/{testing => tests/integration}/monitoring.go (99%) diff --git a/libbeat/tests/integration/framework.go b/libbeat/tests/integration/framework.go index bd6c395fbb9f..8b692f48fcfe 100644 --- a/libbeat/tests/integration/framework.go +++ b/libbeat/tests/integration/framework.go @@ -54,7 +54,6 @@ import ( "github.com/elastic/beats/v7/dev-tools/testbin" "github.com/elastic/beats/v7/libbeat/common/proc" - libbeattesting "github.com/elastic/beats/v7/libbeat/testing" "github.com/elastic/go-elasticsearch/v8" "github.com/elastic/mock-es/pkg/api" ) @@ -659,11 +658,11 @@ func (b *BeatProc) MonitoringPort(timeout time.Duration) int { b.t.Helper() var port int require.Eventuallyf(b.t, func() bool { - line := b.GetLogLine(libbeattesting.MonitoringEndpointSnippet) + line := b.GetLogLine(MonitoringEndpointSnippet) if line == "" { return false } - p, err := libbeattesting.ParseMonitoringPort(line) + p, err := ParseMonitoringPort(line) if err != nil { return false } diff --git a/libbeat/testing/monitoring.go b/libbeat/tests/integration/monitoring.go similarity index 99% rename from libbeat/testing/monitoring.go rename to libbeat/tests/integration/monitoring.go index 57b193f70c3b..8ca9e342260a 100644 --- a/libbeat/testing/monitoring.go +++ b/libbeat/tests/integration/monitoring.go @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -package testing +package integration import ( "fmt" diff --git a/x-pack/otel/oteltestcol/collector.go b/x-pack/otel/oteltestcol/collector.go index 4321e22683fc..4b7cb7d54eda 100644 --- a/x-pack/otel/oteltestcol/collector.go +++ b/x-pack/otel/oteltestcol/collector.go @@ -12,7 +12,7 @@ import ( "testing" "time" - libbeattesting "github.com/elastic/beats/v7/libbeat/testing" + "github.com/elastic/beats/v7/libbeat/tests/integration" "github.com/elastic/beats/v7/libbeat/version" "github.com/elastic/beats/v7/x-pack/auditbeat/abreceiver" "github.com/elastic/beats/v7/x-pack/filebeat/fbreceiver" @@ -150,8 +150,8 @@ func (c *Collector) MonitoringPorts(tb testing.TB, n int) []int { require.EventuallyWithT(tb, func(ct *assert.CollectT) { seen := make(map[int]struct{}) ports = ports[:0] - for _, entry := range c.observer.FilterMessageSnippet(libbeattesting.MonitoringEndpointSnippet).All() { - port, err := libbeattesting.ParseMonitoringPort(entry.Message) + for _, entry := range c.observer.FilterMessageSnippet(integration.MonitoringEndpointSnippet).All() { + port, err := integration.ParseMonitoringPort(entry.Message) if !assert.NoError(ct, err) { continue } From b0de64831e0099dcfcba51bcc0ccc5f4849e8de7 Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Mon, 29 Jun 2026 16:51:57 +0200 Subject: [PATCH 09/13] Context --- filebeat/input/net/tcp/input_test.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/filebeat/input/net/tcp/input_test.go b/filebeat/input/net/tcp/input_test.go index 1a503b4a3fee..984efccc20a4 100644 --- a/filebeat/input/net/tcp/input_test.go +++ b/filebeat/input/net/tcp/input_test.go @@ -134,8 +134,9 @@ func BenchmarkInput(b *testing.B) { } }() + var dialer net.Dialer require.EventuallyWithTf(b, func(ct *assert.CollectT) { - conn, err := net.Dial("tcp", serverAddr) //nolint:noctx // fine for tests + conn, err := dialer.DialContext(b.Context(), "tcp", serverAddr) require.NoError(ct, err) conn.Close() }, 30*time.Second, 100*time.Millisecond, "waiting for TCP server to start") @@ -145,7 +146,7 @@ func BenchmarkInput(b *testing.B) { b.ResetTimer() b.RunParallel(func(pb *testing.PB) { for pb.Next() { - conn, err := net.Dial("tcp", serverAddr) //nolint:noctx // fine for tests + conn, err := dialer.DialContext(b.Context(), "tcp", serverAddr) if err != nil { b.Errorf("cannot create connection: %s", err) continue From 1ef00a0595eff79e36251e83f2a5cedc7afcb219 Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Mon, 29 Jun 2026 16:53:31 +0200 Subject: [PATCH 10/13] Remove cast --- heartbeat/monitors/active/http/http_test.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/heartbeat/monitors/active/http/http_test.go b/heartbeat/monitors/active/http/http_test.go index 4a331b2da5da..6fef1200f7ba 100644 --- a/heartbeat/monitors/active/http/http_test.go +++ b/heartbeat/monitors/active/http/http_test.go @@ -614,10 +614,9 @@ func TestConnRefusedJob(t *testing.T) { require.NoError(t, err) addr, ok := l.Addr().(*net.TCPAddr) require.True(t, ok, "expected *net.TCPAddr from listener") - port := uint16(addr.Port) //nolint:gosec // ephemeral port fits in uint16 require.NoError(t, l.Close()) - url := fmt.Sprintf("http://%s:%d", ip, port) + url := fmt.Sprintf("http://%s:%d", ip, addr.Port) event := sendSimpleTLSRequest(t, url, false) testslike.Test( @@ -625,7 +624,7 @@ func TestConnRefusedJob(t *testing.T) { lookslike.Strict(lookslike.Compose( hbtest.BaseChecks(ip, "down", "http"), hbtest.SummaryStateChecks(0, 1), - hbtest.ECSErrCodeChecks(ecserr.CODE_NET_COULD_NOT_CONNECT, net.JoinHostPort(ip, strconv.Itoa(int(port)))), + hbtest.ECSErrCodeChecks(ecserr.CODE_NET_COULD_NOT_CONNECT, net.JoinHostPort(ip, strconv.Itoa(addr.Port))), urlChecks(url), )), event.Fields, From b0e66823e42504b8a66789d6ca63dabc14387901 Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Mon, 29 Jun 2026 16:54:42 +0200 Subject: [PATCH 11/13] Remove cast --- heartbeat/monitors/active/http/http_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/heartbeat/monitors/active/http/http_test.go b/heartbeat/monitors/active/http/http_test.go index 6fef1200f7ba..dabddb11c29a 100644 --- a/heartbeat/monitors/active/http/http_test.go +++ b/heartbeat/monitors/active/http/http_test.go @@ -636,7 +636,7 @@ func TestUnreachableJob(t *testing.T) { // See: https://tools.ietf.org/html/rfc6890 ip := "203.0.113.1" // Port 80 is sometimes omitted in logs a non-standard one is easier to validate - port := uint16(1234) + const port = 1234 url := fmt.Sprintf("http://%s:%d", ip, port) event := sendSimpleTLSRequest(t, url, false) @@ -646,7 +646,7 @@ func TestUnreachableJob(t *testing.T) { lookslike.Strict(lookslike.Compose( hbtest.BaseChecks(ip, "down", "http"), hbtest.SummaryStateChecks(0, 1), - hbtest.ECSErrCodeChecks(ecserr.CODE_NET_COULD_NOT_CONNECT, net.JoinHostPort(ip, strconv.Itoa(int(port)))), + hbtest.ECSErrCodeChecks(ecserr.CODE_NET_COULD_NOT_CONNECT, net.JoinHostPort(ip, strconv.Itoa(port))), urlChecks(url), )), event.Fields, From 0e6f94d718312bc3c72b3f18d12550c4cecf8840 Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Mon, 29 Jun 2026 16:57:40 +0200 Subject: [PATCH 12/13] comments --- heartbeat/monitors/active/tcp/tcp_test.go | 3 --- libbeat/tests/integration/framework.go | 3 +-- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/heartbeat/monitors/active/tcp/tcp_test.go b/heartbeat/monitors/active/tcp/tcp_test.go index d5c127c74b40..2e9249522eaa 100644 --- a/heartbeat/monitors/active/tcp/tcp_test.go +++ b/heartbeat/monitors/active/tcp/tcp_test.go @@ -119,9 +119,6 @@ func TestUpEndpointJob(t *testing.T) { func TestConnectionRefusedEndpointJob(t *testing.T) { ip := "127.0.0.1" - // Bind an ephemeral port and release it to obtain an address where - // connections are refused (nothing is listening). Binding to :0 avoids the - // time-of-check/time-of-use race of pre-allocating a fixed port. l, err := net.Listen("tcp", net.JoinHostPort(ip, "0")) //nolint:noctx // fine for tests require.NoError(t, err) addr, ok := l.Addr().(*net.TCPAddr) diff --git a/libbeat/tests/integration/framework.go b/libbeat/tests/integration/framework.go index 8b692f48fcfe..123c6cecacff 100644 --- a/libbeat/tests/integration/framework.go +++ b/libbeat/tests/integration/framework.go @@ -652,8 +652,7 @@ func (b *BeatProc) WaitLogsContainsAnyOrder(msgs []string, timeout time.Duration // // Configure the Beat with `http.port: 0` so the OS assigns a free port at bind // time. Reading the port back from the logs avoids the time-of-check/time-of-use -// race of pre-allocating a port, which causes collisions when the same test runs -// many times in parallel (e.g. via script/stresstest.sh). +// race of pre-allocating a port. func (b *BeatProc) MonitoringPort(timeout time.Duration) int { b.t.Helper() var port int From 7d753257e35ddc02a570228c29adb8b63281c0bd Mon Sep 17 00:00:00 2001 From: Orestis Floros Date: Fri, 10 Jul 2026 13:48:06 +0200 Subject: [PATCH 13/13] filebeat: log tcp/udp input bound address to enable ephemeral-port tests The tcp and udp input servers now log the address they actually bound to ("Started listening for TCP/UDP connection on: ") at Info level, right after binding. This lets integration tests configure `host: :0` and read the OS-assigned port back from the logs instead of pre-allocating a fixed port, which removes the time-of-check/time-of-use race and keeps the tests parallel-safe. The bound port is also useful to operators when a config uses an ephemeral port. To consume it, add log-based port discovery mirroring the existing monitoring endpoint helpers: ParseSocketListeningPort plus BeatProc.SocketListeningPort and Collector.SocketListeningPort. TestTCPInputOTelE2E/TestUDPInputOTelE2E now bind ephemeral ports and discover them from the logs, fulfilling the TODO left by elastic/beats#51617 and dropping the hardcoded 9042/9043 ports. --- filebeat/inputsource/tcp/server.go | 6 +- filebeat/inputsource/udp/server.go | 3 + libbeat/tests/integration/framework.go | 24 +++++- libbeat/tests/integration/monitoring.go | 9 ++- libbeat/tests/integration/socketport.go | 47 ++++++++++++ libbeat/tests/integration/socketport_test.go | 73 +++++++++++++++++++ .../tests/integration/otel_tcp_udp_test.go | 34 ++++----- x-pack/otel/oteltestcol/collector.go | 23 ++++++ 8 files changed, 193 insertions(+), 26 deletions(-) create mode 100644 libbeat/tests/integration/socketport.go create mode 100644 libbeat/tests/integration/socketport_test.go diff --git a/filebeat/inputsource/tcp/server.go b/filebeat/inputsource/tcp/server.go index 85b5b4dc7475..7c0012d515ca 100644 --- a/filebeat/inputsource/tcp/server.go +++ b/filebeat/inputsource/tcp/server.go @@ -18,6 +18,7 @@ package tcp import ( + "context" "crypto/tls" "fmt" "net" @@ -79,12 +80,15 @@ func (s *Server) createServer() (net.Listener, error) { return nil, err } } else { - l, err = net.Listen(network, s.config.Host) + l, err = (&net.ListenConfig{}).Listen(context.Background(), network, s.config.Host) if err != nil { return nil, err } } + // Log the bound address so an ephemeral (host ...:0) port can be discovered. + s.logger.Infof("Started listening for TCP connection on: %s", l.Addr()) + if s.config.MaxConnections > 0 { return netutil.LimitListener(l, s.config.MaxConnections), nil } diff --git a/filebeat/inputsource/udp/server.go b/filebeat/inputsource/udp/server.go index af53d1aa5809..5bb8660ebc5a 100644 --- a/filebeat/inputsource/udp/server.go +++ b/filebeat/inputsource/udp/server.go @@ -69,6 +69,9 @@ func (u *Server) createConn() (net.PacketConn, error) { u.localaddress = listener.LocalAddr().String() + // Log the bound address so an ephemeral (host ...:0) port can be discovered. + u.logger.Infof("Started listening for UDP connection on: %s", u.localaddress) + return listener, err } diff --git a/libbeat/tests/integration/framework.go b/libbeat/tests/integration/framework.go index d99a35dc6b02..811c2560ed35 100644 --- a/libbeat/tests/integration/framework.go +++ b/libbeat/tests/integration/framework.go @@ -660,20 +660,38 @@ func (b *BeatProc) WaitLogsContainsAnyOrder(msgs []string, timeout time.Duration // time. Reading the port back from the logs avoids the time-of-check/time-of-use // race of pre-allocating a port. func (b *BeatProc) MonitoringPort(timeout time.Duration) int { + return b.portFromLog(timeout, MonitoringEndpointSnippet, ParseMonitoringPort, + "Beat monitoring endpoint did not log its listening address") +} + +// SocketListeningPort waits for a tcp or udp input to log its listening address +// and returns the ephemeral port it bound to. +// +// Configure the input with host: :0 so the OS assigns a free port at bind +// time. Reading the port back from the logs avoids the time-of-check/time-of-use +// race of pre-allocating a port. +func (b *BeatProc) SocketListeningPort(timeout time.Duration) int { + return b.portFromLog(timeout, SocketListeningSnippet, ParseSocketListeningPort, + "input did not log its listening address") +} + +// portFromLog polls the Beat's logs until a line containing snippet appears and +// parse can extract a port from it, then returns that port. +func (b *BeatProc) portFromLog(timeout time.Duration, snippet string, parse func(string) (int, error), failMsg string) int { b.t.Helper() var port int require.Eventuallyf(b.t, func() bool { - line := b.GetLogLine(MonitoringEndpointSnippet) + line := b.GetLogLine(snippet) if line == "" { return false } - p, err := ParseMonitoringPort(line) + p, err := parse(line) if err != nil { return false } port = p return true - }, timeout, 100*time.Millisecond, "Beat monitoring endpoint did not log its listening address") + }, timeout, 100*time.Millisecond, failMsg) return port } diff --git a/libbeat/tests/integration/monitoring.go b/libbeat/tests/integration/monitoring.go index 8ca9e342260a..16c55902ff2d 100644 --- a/libbeat/tests/integration/monitoring.go +++ b/libbeat/tests/integration/monitoring.go @@ -41,9 +41,14 @@ func ParseMonitoringPort(logLine string) (int, error) { if len(matches) != 2 { return 0, fmt.Errorf("no monitoring address found in log line: %q", logLine) } - _, portStr, err := net.SplitHostPort(matches[1]) + return portFromHostPort(matches[1]) +} + +// portFromHostPort parses the port out of a host:port address logged by a Beat. +func portFromHostPort(addr string) (int, error) { + _, portStr, err := net.SplitHostPort(addr) if err != nil { - return 0, fmt.Errorf("could not split host:port from %q: %w", matches[1], err) + return 0, fmt.Errorf("could not split host:port from %q: %w", addr, err) } port, err := strconv.Atoi(portStr) if err != nil { diff --git a/libbeat/tests/integration/socketport.go b/libbeat/tests/integration/socketport.go new file mode 100644 index 000000000000..960803de72ee --- /dev/null +++ b/libbeat/tests/integration/socketport.go @@ -0,0 +1,47 @@ +// Licensed to Elasticsearch B.V. under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Elasticsearch B.V. licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package integration + +import ( + "fmt" + "regexp" +) + +// SocketListeningSnippet is the fragment the tcp and udp inputs log when their +// socket server binds (see filebeat/inputsource/{tcp,udp}/server.go). Tests +// configure host: :0 and match this snippet to discover the OS-assigned +// port. +const SocketListeningSnippet = "connection on:" + +// e.g. "Started listening for TCP connection on: 127.0.0.1:54321". The address +// is the last token of the message, so the capture stops at whitespace or a +// double quote to avoid swallowing the closing quote when the message is +// embedded in a JSON log line. +var reSocketListening = regexp.MustCompile(regexp.QuoteMeta(SocketListeningSnippet) + ` ([^\s"]+)`) + +// ParseSocketListeningPort extracts the bound port from a SocketListeningSnippet +// log line. It lets tests read the ephemeral port chosen by the OS instead of +// pre-allocating one, which avoids time-of-check/time-of-use port collisions +// when many tests run in parallel. +func ParseSocketListeningPort(logLine string) (int, error) { + matches := reSocketListening.FindStringSubmatch(logLine) + if len(matches) != 2 { + return 0, fmt.Errorf("no socket listening address found in log line: %q", logLine) + } + return portFromHostPort(matches[1]) +} diff --git a/libbeat/tests/integration/socketport_test.go b/libbeat/tests/integration/socketport_test.go new file mode 100644 index 000000000000..f046bc4020ef --- /dev/null +++ b/libbeat/tests/integration/socketport_test.go @@ -0,0 +1,73 @@ +// Licensed to Elasticsearch B.V. under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Elasticsearch B.V. licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package integration + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseSocketListeningPort(t *testing.T) { + testCases := map[string]struct { + logLine string + wantPort int + wantErr bool + }{ + "tcp raw message": { + // The collector observer exposes the bare message. + logLine: "Started listening for TCP connection on: 127.0.0.1:54321", + wantPort: 54321, + }, + "udp raw message": { + logLine: "Started listening for UDP connection on: 127.0.0.1:9999", + wantPort: 9999, + }, + "embedded in a JSON log line": { + // The Beat writes structured logs; the address is followed by the + // closing quote of the message field, which must not be captured. + logLine: `{"log.level":"info","message":"Started listening for TCP connection on: 127.0.0.1:5067","service.name":"filebeat"}`, + wantPort: 5067, + }, + "ipv6 address": { + logLine: "Started listening for UDP connection on: [::1]:6000", + wantPort: 6000, + }, + "no address": { + logLine: "Started listening for UDP connection", + wantErr: true, + }, + "unrelated line": { + logLine: "some other log message", + wantErr: true, + }, + } + + for name, tc := range testCases { + t.Run(name, func(t *testing.T) { + port, err := ParseSocketListeningPort(tc.logLine) + if tc.wantErr { + assert.Error(t, err, "expected an error parsing %q", tc.logLine) + return + } + require.NoError(t, err, "unexpected error parsing %q", tc.logLine) + assert.Equal(t, tc.wantPort, port, "parsed the wrong port from %q", tc.logLine) + }) + } +} diff --git a/x-pack/filebeat/tests/integration/otel_tcp_udp_test.go b/x-pack/filebeat/tests/integration/otel_tcp_udp_test.go index 6bf6898fefde..1f97077246a7 100644 --- a/x-pack/filebeat/tests/integration/otel_tcp_udp_test.go +++ b/x-pack/filebeat/tests/integration/otel_tcp_udp_test.go @@ -32,33 +32,19 @@ const ( ) func TestTCPInputOTelE2E(t *testing.T) { - // TODO: change this to use port from log lines - // See https://github.com/elastic/beats/pull/51617 - otelServerAddr := "127.0.0.1:9042" - fbServerAddr := "127.0.0.1:9043" - runSocketInputOTelE2E( t, "tcp", tcpInputTestMsg, - otelServerAddr, - fbServerAddr, nettest.RunTCPClient, ) } func TestUDPInputOTelE2E(t *testing.T) { - // TODO: change this to use port from log lines - // See https://github.com/elastic/beats/pull/51617 - otelServerAddr := "127.0.0.1:9042" - fbServerAddr := "127.0.0.1:9043" - runSocketInputOTelE2E( t, "udp", udpInputTestMsg, - otelServerAddr, - fbServerAddr, nettest.RunUDPClient, ) } @@ -67,7 +53,7 @@ type socketClientFn func(t *testing.T, address string, data []string) func runSocketInputOTelE2E( t *testing.T, - inputType, testMessage, otelAddress, fbAddress string, + inputType, testMessage string, runClient socketClientFn, ) { t.Helper() @@ -190,16 +176,22 @@ service: PathHome: otelHome, } + // Bind to an ephemeral port (host ...:0) and read the OS-assigned port back + // from the logs. This avoids the time-of-check/time-of-use race of + // pre-allocating a fixed port, so the tests stay parallel-safe. + ephemeralHost := hostAddress(0) + var configBuffer bytes.Buffer - optionsValue.Host = otelAddress + optionsValue.Host = ephemeralHost optionsValue.Index = otelIndex require.NoError(t, template.Must(template.New("config").Parse(otelConfig)).Execute(&configBuffer, optionsValue)) - oteltestcol.New(t, configBuffer.String()) + col := oteltestcol.New(t, configBuffer.String()) + otelAddress := hostAddress(col.SocketListeningPort(t)) configBuffer.Reset() - optionsValue.Host = fbAddress + optionsValue.Host = ephemeralHost optionsValue.Index = fbIndex require.NoError(t, template.Must(template.New("config").Parse(filebeatConfig)).Execute(&configBuffer, optionsValue)) @@ -220,6 +212,8 @@ service: "filebeat did not run", ) + fbAddress := hostAddress(filebeat.SocketListeningPort(20 * time.Second)) + go runClient(t, otelAddress, data) go runClient(t, fbAddress, data) @@ -286,7 +280,7 @@ service: oteltest.AssertMapsEqual(t, filebeatDoc, otelDoc, ignoredFields, "expected documents to be equal") } -// HostAddress returns the host:port address used by net input integration tests. -func hostAddress(port uint16) string { +// hostAddress returns the host:port address used by net input integration tests. +func hostAddress(port int) string { return fmt.Sprintf("127.0.0.1:%d", port) } diff --git a/x-pack/otel/oteltestcol/collector.go b/x-pack/otel/oteltestcol/collector.go index 958e3c66c667..2b43356a10fc 100644 --- a/x-pack/otel/oteltestcol/collector.go +++ b/x-pack/otel/oteltestcol/collector.go @@ -166,6 +166,29 @@ func (c *Collector) MonitoringPorts(tb testing.TB, n int) []int { return ports[:n] } +// SocketListeningPort waits for a tcp or udp input running inside the collector +// to log its listening address and returns the ephemeral port it bound to. +// +// Configure the input with host: :0 so the OS assigns a free port at bind +// time. Reading the port back from the logs avoids the time-of-check/time-of-use +// race of pre-allocating a port. +func (c *Collector) SocketListeningPort(tb testing.TB) int { + tb.Helper() + var port int + require.EventuallyWithT(tb, func(ct *assert.CollectT) { + for _, entry := range c.observer.FilterMessageSnippet(integration.SocketListeningSnippet).All() { + p, err := integration.ParseSocketListeningPort(entry.Message) + if !assert.NoError(ct, err) { + continue + } + port = p + return + } + assert.Fail(ct, "input listening address not logged yet") + }, 30*time.Second, 100*time.Millisecond, "collector input did not start listening") + return port +} + func getComponent() (otelcol.Factories, error) { receivers, err := otelcol.MakeFactoryMap( abreceiver.NewFactory(),