Merge pull request #133407 from richabanker/mvp-egress-dialer

Add egress dialer to proxyClientConfig for Mixed Version Proxy

pkg/controlplane/apiserver/config.go

@@ -330,7 +330,9 @@ func CreateConfig(
 				opts.PeerAdvertiseAddress,
 				genericConfig.APIServerID,
 				config.Extra.PeerEndpointLeaseReconciler,
-				config.Generic.Serializer)
+				config.Generic.Serializer,
+				config.Generic.EgressSelector,
+			)
 			if err != nil {
 				return nil, nil, err
 			}

pkg/controlplane/apiserver/peer.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/client-go/transport"
 
 	genericapiserver "k8s.io/apiserver/pkg/server"
+	"k8s.io/apiserver/pkg/server/egressselector"
 	serverstorage "k8s.io/apiserver/pkg/server/storage"
 	utilpeerproxy "k8s.io/apiserver/pkg/util/peerproxy"
 	coordinationv1informers "k8s.io/client-go/informers/coordination/v1"
@@ -52,7 +53,9 @@ func BuildPeerProxy(
 	peerAdvertiseAddress reconcilers.PeerAdvertiseAddress,
 	apiServerID string,
 	reconciler reconcilers.PeerEndpointLeaseReconciler,
-	serializer runtime.NegotiatedSerializer) (utilpeerproxy.Interface, error) {
+	serializer runtime.NegotiatedSerializer,
+	es *egressselector.EgressSelector,
+) (utilpeerproxy.Interface, error) {
 	if proxyClientCertFile == "" {
 		return nil, fmt.Errorf("error building peer proxy handler, proxy-cert-file not specified")
 	}
@@ -69,6 +72,19 @@ func BuildPeerProxy(
 			ServerName: "kubernetes.default.svc",
 		}}
 
+	if es != nil {
+		egressDialer, err := es.Lookup(egressselector.ControlPlane.AsNetworkContext())
+		if err != nil {
+			return nil, err
+		}
+
+		// We only set the DialHolder if an egress dialer is explicitly configured for the
+		// ControlPlane network context, otherwise the transport will fall back to the default direct dialer.
+		if egressDialer != nil {
+			proxyClientConfig.DialHolder = &transport.DialHolder{Dial: egressDialer}
+		}
+	}
+
 	return utilpeerproxy.NewPeerProxyHandler(
 		apiServerID,
 		IdentityLeaseComponentLabelKey+"="+KubeAPIServer,

test/integration/apiserver/peerproxy/peer_proxy_test.go

@@ -18,6 +18,13 @@ package peerproxy
 
 import (
 	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -201,6 +208,77 @@ func TestPeerProxiedRequestToThirdServerAfterFirstDies(t *testing.T) {
 	assert.Equal(t, job.Name, jobsB.Items[0].Name)
 }
 
+func TestPeerProxy_EgressDialerIsUsed(t *testing.T) {
+	ktesting.SetDefaultVerbosity(1)
+
+	// Start a HTTP CONNECT proxy to act as a fake egress dialer over UDS
+	dialerHit, udsName := runMockEgressProxy(t)
+
+	// Write a temporary egress selector config file pointing to our proxy
+	egressSelectorConfigFile, err := os.CreateTemp("", "egress_selector_configuration.yaml")
+	require.NoError(t, err)
+	defer func() { _ = os.Remove(egressSelectorConfigFile.Name()) }()
+
+	configYAML := fmt.Sprintf(`
+apiVersion: apiserver.k8s.io/v1beta1
+kind: EgressSelectorConfiguration
+egressSelections:
+- name: controlplane
+  connection:
+      proxyProtocol: HTTPConnect
+      transport:
+        uds:
+          udsName: "%s"
+`, udsName)
+	err = os.WriteFile(egressSelectorConfigFile.Name(), []byte(configYAML), 0644)
+	require.NoError(t, err)
+
+	featuregatetesting.SetFeatureGatesDuringTest(t, utilfeature.DefaultFeatureGate, featuregatetesting.FeatureOverrides{
+		features.APIServerIdentity:                   true,
+		features.UnknownVersionInteroperabilityProxy: true,
+	})
+
+	etcd := framework.SharedEtcd()
+	proxyCA, err := createProxyCertContent()
+	require.NoError(t, err)
+
+	// Start two apiservers, one with batch/v1 disabled
+	server.SetHostnameFuncForTests("test-server-egress-a")
+	serverA := kastesting.StartTestServerOrDie(t, &kastesting.TestServerInstanceOptions{
+		EnableCertAuth: true,
+		ProxyCA:        &proxyCA},
+		[]string{"--runtime-config=api/all=true", "--egress-selector-config-file=" + egressSelectorConfigFile.Name()}, etcd)
+	t.Cleanup(serverA.TearDownFn)
+
+	server.SetHostnameFuncForTests("test-server-egress-b")
+	serverB := kastesting.StartTestServerOrDie(t, &kastesting.TestServerInstanceOptions{
+		EnableCertAuth: true,
+		ProxyCA:        &proxyCA},
+		[]string{"--runtime-config=api/all=true,batch/v1=false", "--egress-selector-config-file=" + egressSelectorConfigFile.Name()}, etcd)
+	t.Cleanup(serverB.TearDownFn)
+
+	kubeClientSetA, err := kubernetes.NewForConfig(serverA.ClientConfig)
+	require.NoError(t, err)
+	kubeClientSetB, err := kubernetes.NewForConfig(serverB.ClientConfig)
+	require.NoError(t, err)
+
+	// create jobs resource using serverA
+	job := createJobResource()
+	_, err = kubeClientSetA.BatchV1().Jobs("default").Create(context.Background(), job, metav1.CreateOptions{})
+	require.NoError(t, err)
+
+	// List jobs using ServerB (should be proxied to ServerA, hitting our dialer)
+	jobsB, err := kubeClientSetB.BatchV1().Jobs("default").List(context.Background(), metav1.ListOptions{})
+	require.NoError(t, err)
+	assert.NotEmpty(t, jobsB)
+	assert.Equal(t, job.Name, jobsB.Items[0].Name)
+
+	// Assert that our dialer was hit at least once
+	if atomic.LoadInt32(dialerHit) == 0 {
+		t.Errorf("expected egress dialer to be used, but it was not hit")
+	}
+}
+
 func createProxyCertContent() (kastesting.ProxyCA, error) {
 	result := kastesting.ProxyCA{}
 	proxySigningKey, err := testutil.NewPrivateKey()
@@ -240,3 +318,70 @@ func createJobResource() *v1.Job {
 		},
 	}
 }
+
+func runMockEgressProxy(t *testing.T) (*int32, string) {
+	t.Helper()
+	udsFile, err := os.CreateTemp("/tmp", "egress-*.sock")
+	require.NoError(t, err)
+	udsName := udsFile.Name()
+	_ = udsFile.Close()
+	_ = os.Remove(udsName) // Must remove so we can bind
+
+	listener, err := net.Listen("unix", udsName)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_ = listener.Close()
+		_ = os.Remove(udsName)
+	})
+
+	dialerHit := new(int32)
+	proxy := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		atomic.AddInt32(dialerHit, 1)
+
+		// A standard HTTP CONNECT proxy proxies opaque TCP streams.
+		if req.Method == http.MethodConnect {
+			// Actively establish a TCP connection to the destination API server
+			destConn, err := net.DialTimeout("tcp", req.Host, 5*time.Second)
+			if err != nil {
+				http.Error(w, err.Error(), http.StatusServiceUnavailable)
+				return
+			}
+
+			// Acknowledge the connection stream has opened successfully
+			w.WriteHeader(http.StatusOK)
+
+			// 'Hijack' the underlying network connection away from Go's
+			// generic HTTP Server so we can use it as a raw TCP proxy socket.
+			hijacker, ok := w.(http.Hijacker)
+			if !ok {
+				http.Error(w, "Hijacking not supported", http.StatusInternalServerError)
+				return
+			}
+			clientConn, _, err := hijacker.Hijack()
+			if err != nil {
+				http.Error(w, err.Error(), http.StatusServiceUnavailable)
+				return
+			}
+
+			// Spin up two concurrent forwarding pipes to pass traffic between
+			// the client and the destination over our UDS socket.
+			go func() {
+				_, _ = io.Copy(destConn, clientConn)
+				_ = destConn.Close()
+				_ = clientConn.Close()
+			}()
+			go func() {
+				_, _ = io.Copy(clientConn, destConn)
+				_ = destConn.Close()
+				_ = clientConn.Close()
+			}()
+		} else {
+			http.Error(w, "this is a proxy", http.StatusBadRequest)
+		}
+	}))
+	proxy.Listener = listener
+	proxy.Start()
+	t.Cleanup(proxy.Close)
+
+	return dialerHit, udsName
+}