Add option to prevent cluster-traffic from bypassing loadbalancers

.github/workflows/ccm-integration-tests.yml

@@ -129,6 +129,7 @@ jobs:
 
     env:
       CLOUDSCALE_API_TOKEN: ${{ secrets.CLOUDSCALE_API_TOKEN }}
+      HTTP_ECHO_BRANCH: ${{ vars.HTTP_ECHO_BRANCH }}
       KUBERNETES: '${{ matrix.kubernetes }}'
       SUBNET: '${{ matrix.subnet }}'
       CLUSTER_PREFIX: '${{ matrix.cluster_prefix }}'

cmd/http-echo/go.mod

@@ -0,0 +1,5 @@
+module github.com/cloudscale-ch/cloudscale-cloud-controller-manager/cmd/http-echo
+
+go 1.23.0
+
+require github.com/pires/go-proxyproto v0.7.0

cmd/http-echo/go.sum

@@ -0,0 +1,2 @@
+github.com/pires/go-proxyproto v0.7.0 h1:IukmRewDQFWC7kfnb66CSomk2q/seBuilHBYFwyq0Hs=
+github.com/pires/go-proxyproto v0.7.0/go.mod h1:Vz/1JPY/OACxWGQNIRY2BeyDmpoaWmEP40O9LbuiFR4=

cmd/http-echo/main.go

@@ -0,0 +1,73 @@
+// A http echo server to get information about connections made to it.
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"net"
+	"net/http"
+	"time"
+
+	proxyproto "github.com/pires/go-proxyproto"
+)
+
+func main() {
+	host := flag.String("host", "127.0.0.1", "Host to connect to")
+	port := flag.Int("port", 8080, "Port to connect to")
+
+	flag.Parse()
+
+	serve(*host, *port)
+}
+
+// log http requests in basic fashion
+func log(h http.Handler) http.Handler {
+	return http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			h.ServeHTTP(w, r)
+			fmt.Printf("%s %s (%s)\n", r.Method, r.RequestURI, r.RemoteAddr)
+		})
+}
+
+// serve HTTP API on the given host and port
+func serve(host string, port int) {
+	router := http.NewServeMux()
+
+	// Returns 'true' if the PROXY protocol was used for the given connection
+	router.HandleFunc("GET /proxy-protocol/used",
+		func(w http.ResponseWriter, r *http.Request) {
+			fmt.Fprintln(w, r.Context().Value("HasProxyHeader"))
+		})
+
+	addr := fmt.Sprintf("%s:%d", host, port)
+
+	server := http.Server{
+		Addr:    addr,
+		Handler: log(router),
+		ConnContext: func(ctx context.Context, c net.Conn) context.Context {
+			hasProxyHeader := false
+
+			if c, ok := c.(*proxyproto.Conn); ok {
+				hasProxyHeader = c.ProxyHeader() != nil
+			}
+
+			return context.WithValue(ctx, "HasProxyHeader", hasProxyHeader)
+		},
+	}
+
+	listener, err := net.Listen("tcp", server.Addr)
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Printf("Listening on %s\n", addr)
+
+	proxyListener := &proxyproto.Listener{
+		Listener:          listener,
+		ReadHeaderTimeout: 10 * time.Second,
+	}
+	defer proxyListener.Close()
+
+	server.Serve(proxyListener)
+}

pkg/cloudscale_ccm/loadbalancer.go

@@ -12,6 +12,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/ptr"
 )
 
 // Annotations used by the loadbalancer integration of cloudscale_ccm. Those
@@ -133,6 +134,60 @@ const (
 	// as all pools have to be recreated.
 	LoadBalancerPoolProtocol = "k8s.cloudscale.ch/loadbalancer-pool-protocol"
 
+	// LoadBalancerForceHostname forces the CCM to report a specific hostname
+	// to Kubernetes when returning the loadbalancer status, instead of
+	// reporting the IP address(es).
+	//
+	// The hostname used should point to the same IP address that would
+	// otherwise be reported. This is used as a workaround for clusters that
+	// do not support status.loadBalancer.ingress.ipMode, and use `proxy` or
+	// `proxyv2` protocol.
+	//
+	// For newer clusters, .status.loadBalancer.ingress.ipMode is automatically
+	// set to "Proxy", unless LoadBalancerIPMode is set to "VIP"
+	//
+	// For more information about this workaround see
+	// https://kubernetes.io/blog/2023/12/18/kubernetes-1-29-feature-loadbalancer-ip-mode-alpha/
+	//
+	// To illustrate, here's an example of a load balancer status shown on
+	// a Kubernetes 1.29 service with default settings:
+	//
+	//    apiVersion: v1
+	//    kind: Service
+	//    ...
+	//    status:
+	//      loadBalancer:
+	//        ingress:
+	//          - ip: 45.81.71.1
+	//          - ip: 2a06:c00::1
+	//
+	// Using the annotation causes the status to use the given value instead:
+	//
+	//    apiVersion: v1
+	//    kind: Service
+	//    metadata:
+	//      annotations:
+	//        k8s.cloudscale.ch/loadbalancer-force-hostname: example.org
+	//    status:
+	//      loadBalancer:
+	//        ingress:
+	//          - hostname: example.org
+	//
+	// If you are not using the `proxy` or `proxyv2` protocol, or if you are
+	// on Kubernetes 1.30 or newer, you probly do not need this setting.
+	//
+	// See `LoadBalancerIPMode` below.
+	LoadBalancerForceHostname = "k8s.cloudscale.ch/loadbalancer-force-hostname"
+
+	// LoadBalancerIPMode defines the IP mode reported to Kubernetes for the
+	// loadbalancers managed by this CCM. It defaults to "Proxy", where all
+	// traffic destined to the load balancer is sent through the load balancer,
+	// even if coming from inside the cluster.
+	//
+	// The older behavior, where traffic inside the cluster is directly
+	// sent to the backend service, can be activated by using "VIP" instead.
+	LoadBalancerIPMode = "k8s.cloudscale.ch/loadbalancer-ip-mode"
+
 	// LoadBalancerHealthMonitorDelayS is the delay between two successive
 	// checks, in seconds. Defaults to 2.
 	//
@@ -269,7 +324,13 @@ func (l *loadbalancer) GetLoadBalancer(
 		return nil, false, nil
 	}
 
-	return loadBalancerStatus(instance), true, nil
+	result, err := l.loadBalancerStatus(serviceInfo, instance)
+	if err != nil {
+		return nil, true, fmt.Errorf(
+			"unable to get load balancer state for %s: %w", service.Name, err)
+	}
+
+	return result, true, nil
 }
 
 // GetLoadBalancerName returns the name of the load balancer. Implementations
@@ -361,7 +422,13 @@ func (l *loadbalancer) EnsureLoadBalancer(
 			"unable to annotate service %s: %w", service.Name, err)
 	}
 
-	return loadBalancerStatus(actual.lb), nil
+	result, err := l.loadBalancerStatus(serviceInfo, actual.lb)
+	if err != nil {
+		return nil, fmt.Errorf(
+			"unable to get load balancer state for %s: %w", service.Name, err)
+	}
+
+	return result, nil
 }
 
 // UpdateLoadBalancer updates hosts under the specified load balancer.
@@ -432,6 +499,53 @@ func (l *loadbalancer) EnsureLoadBalancerDeleted(
 	})
 }
 
+// loadBalancerStatus generates the v1.LoadBalancerStatus for the given
+// loadbalancer, as required by Kubernetes.
+func (l *loadbalancer) loadBalancerStatus(
+	serviceInfo *serviceInfo,
+	lb *cloudscale.LoadBalancer,
+) (*v1.LoadBalancerStatus, error) {
+
+	status := v1.LoadBalancerStatus{}
+
+	// When forcing the use of a hostname, there's exactly one ingress item
+	hostname := serviceInfo.annotation(LoadBalancerForceHostname)
+	if len(hostname) > 0 {
+		status.Ingress = []v1.LoadBalancerIngress{{Hostname: hostname}}
+		return &status, nil
+	}
+
+	// Otherwise there as many items as there are addresses
+	status.Ingress = make([]v1.LoadBalancerIngress, len(lb.VIPAddresses))
+
+	var ipmode *v1.LoadBalancerIPMode
+	switch serviceInfo.annotation(LoadBalancerIPMode) {
+	case "Proxy":
+		ipmode = ptr.To(v1.LoadBalancerIPModeProxy)
+	case "VIP":
+		ipmode = ptr.To(v1.LoadBalancerIPModeVIP)
+	default:
+		return nil, fmt.Errorf(
+			"unsupported IP mode: '%s', must be 'Proxy' or 'VIP'", *ipmode)
+	}
+
+	// On newer releases, we explicitly configure the IP mode
+	supportsIPMode, err := kubeutil.IsKubernetesReleaseOrNewer(l.k8s, 1, 30)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get load balancer status: %w", err)
+	}
+
+	for i, address := range lb.VIPAddresses {
+		status.Ingress[i].IP = address.Address
+
+		if supportsIPMode {
+			status.Ingress[i].IPMode = ipmode
+		}
+	}
+
+	return &status, nil
+}
+
 // ensureValidConfig ensures that the configuration can be applied at all,
 // acting as a gate that ensures certain invariants before any changes are
 // made.
@@ -545,17 +659,3 @@ func (l *loadbalancer) findIPsAssignedElsewhere(
 
 	return conflicts, nil
 }
-
-// loadBalancerStatus generates the v1.LoadBalancerStatus for the given
-// loadbalancer, as required by Kubernetes.
-func loadBalancerStatus(lb *cloudscale.LoadBalancer) *v1.LoadBalancerStatus {
-
-	status := v1.LoadBalancerStatus{}
-	status.Ingress = make([]v1.LoadBalancerIngress, len(lb.VIPAddresses))
-
-	for i, address := range lb.VIPAddresses {
-		status.Ingress[i].IP = address.Address
-	}
-
-	return &status
-}

pkg/cloudscale_ccm/service_info.go

@@ -82,6 +82,10 @@ func (s serviceInfo) annotation(key string) string {
 		return s.annotationOrDefault(key, "")
 	case LoadBalancerPoolProtocol:
 		return s.annotationOrDefault(key, "tcp")
+	case LoadBalancerForceHostname:
+		return s.annotationOrDefault(key, "")
+	case LoadBalancerIPMode:
+		return s.annotationOrDefault(key, "Proxy")
 	case LoadBalancerFlavor:
 		return s.annotationOrDefault(key, "lb-standard")
 	case LoadBalancerVIPAddresses: