libbeat/testing: use ephemeral ports to avoid TOCTOU collisions

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,7 @@ func TestInput(t *testing.T) {
 }
 
 func BenchmarkInput(b *testing.B) {
-	port, err := libbeattesting.AvailableTCP4Port()
-	if err != nil {
-		b.Fatalf("cannot find available port: %s", err)
-	}
-	serverAddr := net.JoinHostPort("localhost", fmt.Sprintf("%d", port))
+	serverAddr := ephemeralTCPAddr(b)
 
 	inp, err := configure(conf.MustNewConfigFrom(map[string]any{
 		"host":              serverAddr,
@@ -140,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)
+		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")
@@ -151,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)
+			conn, err := dialer.DialContext(b.Context(), "tcp", serverAddr)
 			if err != nil {
 				b.Errorf("cannot create connection: %s", err)
 				continue
@@ -177,3 +172,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
+}

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,18 +609,22 @@ func TestHTTPSx509Auth(t *testing.T) {
 
 func TestConnRefusedJob(t *testing.T) {
 	ip := "127.0.0.1"
-	port, err := btesting.AvailableTCP4Port()
+	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")
+	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(
 		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,
@@ -633,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)
@@ -643,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,

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,12 @@ func TestUpEndpointJob(t *testing.T) {
 
 func TestConnectionRefusedEndpointJob(t *testing.T) {
 	ip := "127.0.0.1"
-	port, err := btesting.AvailableTCP4Port()
+	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 +249,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
 	}

libbeat/tests/integration/framework.go

@@ -647,6 +647,30 @@ 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.
+func (b *BeatProc) MonitoringPort(timeout time.Duration) int {
+	b.t.Helper()
+	var port int
+	require.Eventuallyf(b.t, func() bool {
+		line := b.GetLogLine(MonitoringEndpointSnippet)
+		if line == "" {
+			return false
+		}
+		p, err := 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) {

libbeat/tests/integration/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 integration
+
+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(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
+// 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
+}

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

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)