diff --git a/filebeat/input/net/tcp/input_test.go b/filebeat/input/net/tcp/input_test.go
index c7ee4a5b63ed..e95a523647fb 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"
@@ -103,11 +101,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,
@@ -137,8 +131,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")
@@ -148,7 +143,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
@@ -174,3 +169,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/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/heartbeat/monitors/active/http/http_test.go b/heartbeat/monitors/active/http/http_test.go
index 218ac6185b5a..fd3d2a047003 100644
--- a/heartbeat/monitors/active/http/http_test.go
+++ b/heartbeat/monitors/active/http/http_test.go
@@ -58,7 +58,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 {
@@ -614,10 +613,14 @@ 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(
@@ -625,7 +628,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,
@@ -637,7 +640,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)
@@ -647,7 +650,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,
diff --git a/heartbeat/monitors/active/tcp/tcp_test.go b/heartbeat/monitors/active/tcp/tcp_test.go
index c5cc0dd614e6..2e9249522eaa 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,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
 	}
diff --git a/libbeat/testing/available_port.go b/libbeat/testing/available_port.go
deleted file mode 100644
index 93457decff22..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 range n {
-		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/tests/integration/framework.go b/libbeat/tests/integration/framework.go
index 7b8d7ad6d49d..4edaa885fdd7 100644
--- a/libbeat/tests/integration/framework.go
+++ b/libbeat/tests/integration/framework.go
@@ -651,6 +651,48 @@ 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 {
+	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: <ip>: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(snippet)
+		if line == "" {
+			return false
+		}
+		p, err := parse(line)
+		if err != nil {
+			return false
+		}
+		port = p
+		return true
+	}, timeout, 100*time.Millisecond, failMsg)
+	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/libbeat/tests/integration/monitoring.go b/libbeat/tests/integration/monitoring.go
new file mode 100644
index 000000000000..16c55902ff2d
--- /dev/null
+++ b/libbeat/tests/integration/monitoring.go
@@ -0,0 +1,58 @@
+// 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)
+	}
+	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", addr, 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/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: <ip>: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_filebeat_input_test.go b/x-pack/filebeat/tests/integration/otel_filebeat_input_test.go
index 7647bc47efc7..62d45a9333cd 100644
--- a/x-pack/filebeat/tests/integration/otel_filebeat_input_test.go
+++ b/x-pack/filebeat/tests/integration/otel_filebeat_input_test.go
@@ -481,9 +481,6 @@ service:
                 - elasticsearch
             receivers:
                 - filebeatreceiver
-    telemetry:
-        metrics:
-            level: none
 `
 
 	optionsValue := options{
diff --git a/x-pack/filebeat/tests/integration/otel_helpers_test.go b/x-pack/filebeat/tests/integration/otel_helpers_test.go
index 14f4fc093270..3f7bb7587d9d 100644
--- a/x-pack/filebeat/tests/integration/otel_helpers_test.go
+++ b/x-pack/filebeat/tests/integration/otel_helpers_test.go
@@ -68,9 +68,6 @@ const otelElasticsearchServiceYAML = `service:
                 - elasticsearch
             receivers:
                 - filebeatreceiver
-    telemetry:
-        metrics:
-            level: none
 `
 
 func otelE2ERawQueryForInputTypeAndMessage(inputType, message string) map[string]any {
diff --git a/x-pack/filebeat/tests/integration/otel_kafka_test.go b/x-pack/filebeat/tests/integration/otel_kafka_test.go
index 4ba646b38eda..641fe93ed08d 100644
--- a/x-pack/filebeat/tests/integration/otel_kafka_test.go
+++ b/x-pack/filebeat/tests/integration/otel_kafka_test.go
@@ -76,9 +76,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 1b5c4ffb33fa..52688a4ca54a 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
@@ -233,9 +230,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_tcp_udp_test.go b/x-pack/filebeat/tests/integration/otel_tcp_udp_test.go
index 8168791e1a40..5e7473a4f9a5 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()
@@ -144,16 +130,22 @@ processors:
 		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))
 
@@ -174,6 +166,8 @@ processors:
 		"filebeat did not run",
 	)
 
+	fbAddress := hostAddress(filebeat.SocketListeningPort(20 * time.Second))
+
 	go runClient(t, otelAddress, data)
 	go runClient(t, fbAddress, data)
 
@@ -216,7 +210,7 @@ processors:
 	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/filebeat/tests/integration/otel_test.go b/x-pack/filebeat/tests/integration/otel_test.go
index db0602ae690c..937e44101cc0 100644
--- a/x-pack/filebeat/tests/integration/otel_test.go
+++ b/x-pack/filebeat/tests/integration/otel_test.go
@@ -37,7 +37,6 @@ import (
 
 	"github.com/elastic/beats/v7/libbeat/features"
 	esoutput "github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
-	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"
@@ -58,9 +57,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:
@@ -86,7 +82,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:
@@ -114,7 +110,8 @@ service:
 `
 	logFilePath := filepath.Join(tmpdir, "log.log")
 	writeEventsToLogFile(t, logFilePath, numEvents)
-	oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, otelMonitoringPort, fbOtelIndex))
+	collector := oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, fbOtelIndex))
+	otelMonitoringPort := collector.MonitoringPort(t)
 
 	beatsCfgFile := `
 filebeat.inputs:
@@ -141,7 +138,7 @@ processors:
     - add_kubernetes_metadata: ~
 http.enabled: true
 http.host: localhost
-http.port: %d
+http.port: 0
 `
 
 	// start filebeat
@@ -150,12 +147,14 @@ 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()
 
+	filebeatMonitoringPort := filebeat.MonitoringPort(30 * time.Second)
+
 	// prepare to query ES
 	es := integration.GetESClient(t, "http")
 
@@ -198,13 +197,13 @@ 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)
 }
 
 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 {
@@ -277,7 +276,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
@@ -285,14 +283,12 @@ 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"),
 			},
 		},
 	}
@@ -315,11 +311,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:
@@ -349,13 +343,11 @@ service:
   telemetry:
     logs:
       level: DEBUG
-    metrics:
-      level: none
 `, otelConfig)
 
 	writeEventsToLogFile(t, logFilePath, wantEvents)
 
-	oteltestcol.New(t, cfg)
+	collector := oteltestcol.New(t, cfg)
 
 	es := integration.GetESClient(t, "http")
 
@@ -375,8 +367,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)
 	}
 }
 
@@ -611,19 +603,17 @@ 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,
 			}
 
 			cfg := `receivers:
@@ -643,7 +633,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:
@@ -691,14 +681,13 @@ service:
   telemetry:
     logs:
       level: DEBUG
-    metrics:
-      level: none
 `
 			var configBuffer bytes.Buffer
 			require.NoError(t,
 				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
@@ -745,7 +734,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")
@@ -823,10 +812,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"
@@ -843,7 +832,7 @@ exporters:
     auth:
      authenticator: beatsauth
 service:
-  extensions: 
+  extensions:
   - beatsauth
   pipelines:
     logs:
@@ -910,9 +899,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:
@@ -978,9 +964,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:
@@ -1079,8 +1062,6 @@ func TestNoDuplicates(t *testing.T) {
 	namespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "")
 	fbOtelIndex := "logs-integration-" + namespace
 
-	otelMonitoringPort := int(libbeattesting.MustAvailableTCP4Port(t))
-
 	otelCfgFile := `receivers:
   filebeatreceiver:
     filebeat:
@@ -1104,7 +1085,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:
@@ -1165,7 +1146,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) {
@@ -1179,28 +1160,15 @@ 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, otelMonitoringPort, 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,
@@ -1333,7 +1301,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
@@ -1377,15 +1354,13 @@ 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,
 	}
 
 	cfg := `receivers:
@@ -1405,7 +1380,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:
@@ -1450,8 +1425,6 @@ service:
   telemetry:
     logs:
       level: DEBUG
-    metrics:
-      level: none
 `
 
 	var configBuffer bytes.Buffer
@@ -1574,8 +1547,6 @@ service:
   telemetry:
     logs:
       level: DEBUG
-    metrics:
-      level: none
 `, struct {
 		Receivers []multiReceiverConfig
 		InputFile string
@@ -1708,8 +1679,6 @@ service:
   telemetry:
     logs:
       level: DEBUG
-    metrics:
-      level: none
 `
 
 	var configBuffer bytes.Buffer
@@ -1819,8 +1788,6 @@ service:
   telemetry:
     logs:
       level: DEBUG
-    metrics:
-      level: none
 `, struct{ Receivers []multiReceiverConfig }{Receivers: receivers})
 
 		b.StartTimer()
@@ -1873,8 +1840,6 @@ service:
   telemetry:
     logs:
       level: DEBUG
-    metrics:
-      level: none
 `, struct {
 					PathHome        string
 					EventCount      int
@@ -2111,8 +2076,6 @@ service:
   telemetry:
     logs:
       level: warn
-    metrics:
-      level: none
 `, struct {
 			EventCountPerReceiver int
 			PathHome              string
@@ -2194,8 +2157,6 @@ func TestBeatProcessorSharedAcrossPipelines(t *testing.T) {
   telemetry:
     logs:
       level: debug
-    metrics:
-      level: none
 receivers:
   filebeatreceiver/1:
     filebeat:
@@ -2262,8 +2223,6 @@ func TestBeatProcessorWhenCondition(t *testing.T) {
   telemetry:
     logs:
       level: debug
-    metrics:
-      level: none
 receivers:
   filebeatreceiver:
     filebeat:
diff --git a/x-pack/metricbeat/tests/integration/otel_test.go b/x-pack/metricbeat/tests/integration/otel_test.go
index 82d91905bd98..55ed14073c0c 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,21 +40,16 @@ 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,
 	}
 
 	cfg := `receivers:
@@ -82,7 +76,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 +140,7 @@ processors:
     - add_kubernetes_metadata: ~
 http.enabled: true
 http.host: localhost
-http.port: {{.MonitoringPort}}
+http.port: 0
 `
 
 	es := integration.GetESClient(t, "http")
@@ -161,17 +155,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 +171,8 @@ http.port: {{.MonitoringPort}}
 	metricbeat.Start()
 	defer metricbeat.Stop()
 
+	metricbeatMonitoringPort := metricbeat.MonitoringPort(30 * time.Second)
+
 	// Make sure find the logs
 	var metricbeatDocs estools.Documents
 	var otelDocs estools.Documents
@@ -232,9 +226,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 +254,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 +286,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 +327,7 @@ service:
 		}
 	})
 
-	oteltestcol.New(t, string(configContents))
+	collector := oteltestcol.New(t, string(configContents))
 
 	var r0Docs, r1Docs estools.Documents
 	var err error
@@ -371,8 +360,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 21f72bf81922..2b43356a10fc 100644
--- a/x-pack/otel/oteltestcol/collector.go
+++ b/x-pack/otel/oteltestcol/collector.go
@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"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"
@@ -42,22 +43,29 @@ 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()
 
 	configDir := tb.TempDir()
 	configFile := filepath.Join(configDir, "otel.yaml")
-	err := os.WriteFile(configFile, []byte(configYAML), 0o644)
-	require.NoError(tb, err)
+	require.NoError(tb, os.WriteFile(configFile, []byte(configYAML), 0o644))
 
-	if err != nil {
-		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(
@@ -68,39 +76,117 @@ 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.Go(func() {
+	go func() {
+		defer close(c.done)
 		ctx, cancel := signal.NotifyContext(tb.Context(), os.Interrupt)
 		defer cancel()
 		assert.NoError(tb, col.Run(ctx))
-	})
+	}()
 
 	require.Eventually(tb, func() bool {
 		return col.GetState() == otelcol.StateRunning
 	}, 15*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.
+//
+// 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(integration.MonitoringEndpointSnippet).All() {
+			port, err := integration.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]
+}
+
+// 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: <ip>: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) {
@@ -150,7 +236,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",
@@ -165,7 +253,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(),
 				},
diff --git a/x-pack/packetbeat/tests/integration/otel_test.go b/x-pack/packetbeat/tests/integration/otel_test.go
index b8ec9609620e..51902c71f081 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,6 @@ func TestPacketbeatOTelE2E(t *testing.T) {
 	namespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "")
 	index := "logs-integration-" + namespace
 
-	monitoringPort := int(libbeattesting.MustAvailableTCP4Port(t))
-
 	cfg := fmt.Sprintf(`receivers:
   packetbeatreceiver:
     packetbeat:
@@ -150,7 +153,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 +174,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 +191,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,21 +212,18 @@ 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},
 		},
 	}
 
@@ -247,7 +246,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 +275,7 @@ service:
 		"Receivers": receivers,
 	})
 
-	oteltestcol.New(t, cfg)
+	collector := oteltestcol.New(t, cfg)
 
 	es := integration.GetESClient(t, "http")
 
@@ -293,8 +292,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 +309,6 @@ func TestPacketbeatOTelBeatE2E(t *testing.T) {
 	pbOtelIndex := "logs-integration-" + namespace
 	pbIndex := "logs-packetbeat-" + namespace
 
-	otelMonitoringPort := int(libbeattesting.MustAvailableTCP4Port(t))
-
 	otelCfg := fmt.Sprintf(`receivers:
   packetbeatreceiver:
     packetbeat:
@@ -329,7 +326,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 +347,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 +408,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 +422,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 +444,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
@@ -474,17 +471,14 @@ service:
   telemetry:
     logs:
       level: DEBUG
-    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 +497,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 +512,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 +543,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:
@@ -584,15 +573,13 @@ service:
   telemetry:
     logs:
       level: DEBUG
-    metrics:
-      level: none
 `, map[string]any{
 		"Device":    loopbackDevice(),
 		"Index":     index,
 		"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 +602,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)
 	}
 }