Feature Implement Unix Domain Socket based connectivity for Envoy->Gateway XDS Server.

Implement Unix Domain Socket based connectivity for Envoy to connect to XDS server for gateway.

Author: ashishkpathak-connect

pkg/gateway/controller.go

@@ -20,7 +20,7 @@ import (
 	"context"
 	"fmt"
 	"net"
-	"sort"
+	"os"
 	"sync/atomic"
 	"time"
 
@@ -67,6 +67,7 @@ import (
 const (
 	controllerName = "kind.sigs.k8s.io/gateway-controller"
 	GWClassName    = "cloud-provider-kind"
+	GWSocketName   = "/var/run/cloudproviderkind.sock"
 	maxRetries     = 5
 	workers        = 5
 )
@@ -120,11 +121,9 @@ type Controller struct {
 	referenceGrantLister       gatewaylistersv1beta1.ReferenceGrantLister
 	referenceGrantListerSynced cache.InformerSynced
 
-	xdscache        cachev3.SnapshotCache
-	xdsserver       serverv3.Server
-	xdsLocalAddress string
-	xdsLocalPort    int
-	xdsVersion      atomic.Uint64
+	xdscache   cachev3.SnapshotCache
+	xdsserver  serverv3.Server
+	xdsVersion atomic.Uint64
 
 	tunnelManager *tunnels.TunnelManager
 }
@@ -428,29 +427,35 @@ func (c *Controller) Run(ctx context.Context) error {
 	secretv3.RegisterSecretDiscoveryServiceServer(grpcServer, c.xdsserver)
 	runtimev3.RegisterRuntimeDiscoveryServiceServer(grpcServer, c.xdsserver)
 
-	address, err := GetControlPlaneAddress()
-	if err != nil {
-		return err
+	// Cleanup any existing socket file remaining incase from previous crash.
+	if err := os.RemoveAll(GWSocketName); err != nil {
+		return fmt.Errorf("failed to remove old socket: %s", err.Error())
 	}
-	listener, err := net.Listen("tcp", fmt.Sprintf("%s:0", address))
+
+	// Start a Unix Domain Socket
+	listener, err := net.Listen("unix", GWSocketName)
 	if err != nil {
 		return err
 	}
-	defer listener.Close()
 
-	addr := listener.Addr()
-	tcpAddr, ok := addr.(*net.TCPAddr)
-	if !ok {
-		return fmt.Errorf("could not assert listener address to TCPAddr: %s", addr.String())
+	defer func() {
+		listener.Close()
+		logger.Info("Cleaning up socket")
+		if err := os.RemoveAll(GWSocketName); err != nil {
+			logger.Error(err, "failed to cleanup socket")
+		}
+	}()
+
+	if err := os.Chmod(GWSocketName, 0666); err != nil {
+		listener.Close()
+		return err
 	}
 
-	c.xdsLocalAddress = address
-	c.xdsLocalPort = tcpAddr.Port
 	go func() {
 		logger.Info(
 			"XDS management server listening",
-			"address", c.xdsLocalAddress,
-			"port", c.xdsLocalPort)
+			"Unix Domain Socket", GWSocketName,
+		)
 		if err = grpcServer.Serve(listener); err != nil {
 			logger.Error(err, "gRPC server error:")
 		}
@@ -650,7 +655,6 @@ func (c *Controller) processNextGatewayItem(ctx context.Context) bool {
 		return false
 	}
 	defer c.gatewayqueue.Done(key)
-
 	err := c.syncGateway(ctx, key)
 	c.handleGatewayErr(err, key)
 	return true
@@ -673,54 +677,6 @@ func (c *Controller) handleGatewayErr(err error, key string) {
 	klog.Infof("Dropping Gateway %q out of the queue: %v", key, err)
 }
 
-func GetControlPlaneAddress() (string, error) {
-	interfaces, err := net.Interfaces()
-	if err != nil {
-		return "", err
-	}
-
-	sort.Slice(interfaces, func(i, j int) bool {
-		nameI := interfaces[i].Name
-		nameJ := interfaces[j].Name
-
-		if nameI == "docker0" {
-			return true
-		}
-		if nameJ == "docker0" {
-			return false
-		}
-
-		if nameI == "eth0" {
-			return nameJ != "docker0"
-		}
-		if nameJ == "eth0" {
-			return false
-		}
-
-		return nameI < nameJ
-	})
-
-	for _, iface := range interfaces {
-		if iface.Flags&net.FlagUp == 0 || iface.Flags&net.FlagLoopback != 0 {
-			continue
-		}
-
-		addrs, err := iface.Addrs()
-		if err != nil {
-			continue
-		}
-
-		for _, addr := range addrs {
-			ipNet, ok := addr.(*net.IPNet)
-			if ok && ipNet.IP.To4() != nil && !ipNet.IP.IsLinkLocalUnicast() && !ipNet.IP.IsLoopback() {
-				return ipNet.IP.String(), nil
-			}
-		}
-	}
-
-	return "", fmt.Errorf("no suitable global unicast IPv4 address found on any active non-loopback interface")
-}
-
 func (c *Controller) UpdateXDSServer(ctx context.Context, nodeid string, resources map[resourcev3.Type][]envoyproxytypes.Resource) error {
 	c.xdsVersion.Add(1)
 

pkg/gateway/envoy.go

@@ -21,6 +21,7 @@ import (
 const (
 	proxyConfigPath = "/home/envoy/envoy.yaml"
 	envoyAdminPort  = 10000
+	envoySocketName = "/var/run/envoy.sock"
 
 	// well known dns to reach host from containers
 	// https://github.com/containerd/nerdctl/issues/747
@@ -46,7 +47,7 @@ dynamic_resources:
 
 static_resources:
   clusters:
-  - type: STRICT_DNS
+  - type: STATIC
     typed_extension_protocol_options:
       envoy.extensions.upstreams.http.v3.HttpProtocolOptions:
         "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
@@ -59,19 +60,8 @@ static_resources:
       - lb_endpoints:
         - endpoint:
             address:
-              socket_address:
-                address: {{ .ControlPlaneAddress }}
-                port_value: {{ .ControlPlanePort }}
-        - endpoint:
-            address:
-              socket_address:
-                address: host.docker.internal
-                port_value: {{ .ControlPlanePort }}
-        - endpoint:
-            address:
-              socket_address:
-                address: host.lima.internal
-                port_value: {{ .ControlPlanePort }}
+              pipe:
+                path: {{ .EnvoySocket }}
 
 admin:
   access_log_path: /dev/stdout
@@ -82,20 +72,17 @@ admin:
 `
 
 type configData struct {
-	Cluster             string
-	ID                  string
-	AdminPort           int
-	ControlPlaneAddress string
-	ControlPlanePort    int
+	Cluster     string
+	ID          string
+	AdminPort   int
+	EnvoySocket string
 }
 
 // generateEnvoyConfig returns an envoy config generated from config data
 func generateEnvoyConfig(data *configData) (string, error) {
 	if data.Cluster == "" ||
 		data.ID == "" ||
-		data.AdminPort == 0 ||
-		data.ControlPlaneAddress == "" ||
-		data.ControlPlanePort == 0 {
+		data.AdminPort == 0 {
 		return "", fmt.Errorf("missing parameters")
 	}
 
@@ -128,15 +115,14 @@ func gatewaySimpleName(clusterName, namespace, name string) string {
 }
 
 // createGateway create a docker container with a gateway
-func createGateway(clusterName string, nameserver string, localAddress string, localPort int, gateway *gatewayv1.Gateway, enableTunnel bool) error {
+func createGateway(clusterName string, nameserver string, gateway *gatewayv1.Gateway, enableTunnel bool) error {
 	name := gatewayName(clusterName, gateway.Namespace, gateway.Name)
 	simpleName := gatewaySimpleName(clusterName, gateway.Namespace, gateway.Name)
 	envoyConfigData := &configData{
-		ID:                  name,
-		Cluster:             simpleName,
-		AdminPort:           envoyAdminPort,
-		ControlPlaneAddress: localAddress,
-		ControlPlanePort:    localPort,
+		ID:          name,
+		Cluster:     simpleName,
+		AdminPort:   envoyAdminPort,
+		EnvoySocket: envoySocketName,
 	}
 	dynamicFilesystemConfig, err := generateEnvoyConfig(envoyConfigData)
 	if err != nil {
@@ -153,6 +139,8 @@ func createGateway(clusterName string, nameserver string, localAddress string, l
 		"--user=0",
 		"--label", fmt.Sprintf("%s=%s", constants.NodeCCMLabelKey, clusterName),
 		"--label", fmt.Sprintf("%s=%s", constants.GatewayNameLabelKey, simpleName),
+		"--env", "ENVOY_UID=0",
+		"--volume", fmt.Sprintf("%s:%s", GWSocketName, envoySocketName),
 		"--net", networkName,
 		"--dns", nameserver,
 		"--init=false",

pkg/gateway/kindcluster.go

@@ -196,7 +196,7 @@ func (c *Controller) ensureGatewayContainer(ctx context.Context, gw *gatewayv1.G
 	if !container.Exist(containerName) {
 		klog.V(2).Infof("creating container %s for gateway  %s/%s on cluster %s", containerName, namespace, name, c.clusterName)
 		enableTunnels := c.tunnelManager != nil || config.DefaultConfig.LoadBalancerConnectivity == config.Portmap
-		err := createGateway(c.clusterName, c.clusterNameserver, c.xdsLocalAddress, c.xdsLocalPort, gw, enableTunnels)
+		err := createGateway(c.clusterName, c.clusterNameserver, gw, enableTunnels)
 		if err != nil {
 			return err
 		}