feat: support running emulator on the same host with Omni

hack/compose/docker-compose.yml

@@ -5,6 +5,7 @@ services:
       - state:/_out/state
     container_name: talemu
     restart: on-failure
+    network_mode: host
     cap_add:
       - NET_ADMIN
     build:

internal/pkg/kubefactory/etcd.go

@@ -48,6 +48,17 @@ func NewEmbeddedEtcd(ctx context.Context, path string, logger *zap.Logger) (*Etc
 	cfg.ExperimentalCompactHashCheckEnabled = true
 	cfg.ExperimentalInitialCorruptCheck = true
 
+	// run the emulator etcd on a different port to avoid clashing with Omni etcd
+	caddr, _ := url.Parse("http://localhost:2400") //nolint:errcheck
+	paddr, _ := url.Parse("http://localhost:2401") //nolint:errcheck
+
+	cfg.InitialCluster = fmt.Sprintf("default=%s", paddr.String())
+	cfg.ListenClientUrls = []url.URL{*caddr}
+	cfg.AdvertiseClientUrls = []url.URL{*caddr}
+
+	cfg.ListenPeerUrls = []url.URL{*paddr}
+	cfg.AdvertisePeerUrls = []url.URL{*paddr}
+
 	peerURL, err := url.Parse("http://localhost:0")
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse URL: %w", err)

internal/pkg/machine/controllers/grpc_tls.go

@@ -9,6 +9,7 @@ import (
 	stdlibx509 "crypto/x509"
 	"fmt"
 	"net"
+	"net/netip"
 	"strings"
 	"time"
 
@@ -335,6 +336,7 @@ func (ctrl *GRPCTLSController) generateWorker(ctx context.Context, r controller.
 	remoteGen, err := gen.NewRemoteGenerator(func(ctx context.Context, addr string) (net.Conn, error) {
 		var dialer net.Dialer
 
+		dialer.LocalAddr = net.TCPAddrFromAddrPort(netip.AddrPortFrom(address.TypedSpec().Address.Addr(), 0))
 		dialer.Control = emunet.BindToInterface(address.TypedSpec().LinkName)
 
 		return dialer.DialContext(ctx, "tcp", addr)

internal/pkg/machine/controllers/log_sink.go

@@ -59,7 +59,7 @@ func (ctrl *LogSinkController) Run(ctx context.Context, r controller.Runtime, _
 
 		if err = addresses.ForEachErr(func(r *network.AddressStatus) error {
 			if strings.HasPrefix(r.TypedSpec().LinkName, constants.SideroLinkName) {
-				return ctrl.LogSink.ConfigureInterface(ctx, r.TypedSpec().LinkName)
+				return ctrl.LogSink.ConfigureInterface(ctx, r)
 			}
 
 			return nil

internal/pkg/machine/logging/logging.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/siderolabs/go-circular"
+	"github.com/siderolabs/talos/pkg/machinery/resources/network"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 )
@@ -56,8 +57,8 @@ func NewZapCore(endpoint string) (*ZapCore, error) {
 }
 
 // ConfigureInterface inits the sender.
-func (core *ZapCore) ConfigureInterface(ctx context.Context, iface string) error {
-	core.sender.configure(iface)
+func (core *ZapCore) ConfigureInterface(ctx context.Context, res *network.AddressStatus) error {
+	core.sender.configure(res.TypedSpec().LinkName, res.TypedSpec().Address)
 
 	ctx, cancel := context.WithTimeout(ctx, time.Second)
 	defer cancel()

internal/pkg/machine/logging/sender.go

@@ -9,6 +9,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net"
+	"net/netip"
 	"net/url"
 	"sync"
 	"time"
@@ -18,11 +19,12 @@ import (
 
 // LogSender writes zap logs to the remote destination.
 type LogSender struct {
-	endpoint *url.URL
-	conn     net.Conn
-	sema     chan struct{}
-	iface    string
-	mu       sync.Mutex
+	endpoint  *url.URL
+	conn      net.Conn
+	sema      chan struct{}
+	iface     string
+	localAddr netip.Prefix
+	mu        sync.Mutex
 }
 
 // NewLogSender returns log sender that sends logs in JSON over TCP (newline-delimited)
@@ -45,11 +47,12 @@ func (j *LogSender) ready() bool {
 	return j.iface != ""
 }
 
-func (j *LogSender) configure(iface string) {
+func (j *LogSender) configure(iface string, localAddr netip.Prefix) {
 	j.mu.Lock()
 	defer j.mu.Unlock()
 
 	j.iface = iface
+	j.localAddr = localAddr
 }
 
 func (j *LogSender) tryLock(ctx context.Context) (unlock func()) {
@@ -80,6 +83,7 @@ func (j *LogSender) marshalJSON(e *LogEvent) ([]byte, error) {
 func (j *LogSender) Send(ctx context.Context, e *LogEvent) error {
 	j.mu.Lock()
 	iface := j.iface
+	localAddr := j.localAddr
 	j.mu.Unlock()
 
 	if iface == "" {
@@ -104,6 +108,7 @@ func (j *LogSender) Send(ctx context.Context, e *LogEvent) error {
 
 	var dialer net.Dialer
 
+	dialer.LocalAddr = net.TCPAddrFromAddrPort(netip.AddrPortFrom(localAddr.Addr(), 0))
 	dialer.Control = network.BindToInterface(iface)
 
 	// Connect (or "connect" for UDP) if no connection is established already.