Add a firewall rule to open the LB health check and service port(s) when using service.beta.kubernetes.io/do-loadbalancer-type=REGIONAL_NETWORK (#748)

bbassingthwaite · web-flow · commit 59b7b066309a · 2024-07-26T22:21:46.000+02:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,5 +1,9 @@
 ## unreleased
 
+* When using the LoadBalancer `service.beta.kubernetes.io/do-loadbalancer-type=REGIONAL_NETWORK` (under closed beta), firewall rules
+are added to open up the underlying health check port and all the defined (port, protocols) defined on the service. This is to
+permit traffic to arrive directly on the underlying worker nodes.
+
 ## v0.1.54 (beta) - June 12, 2024
 
 * Fixes an issue with load balancer health checks when the LB is using PROXY protocol. The new health check 
diff --git a/cloud-controller-manager/do/firewall_controller.go b/cloud-controller-manager/do/firewall_controller.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"sort"
 	"strconv"
 	"time"
 
@@ -270,20 +271,25 @@ func (fm *firewallManager) updateFirewall(ctx context.Context, fwID string, fr *
 	return fm.executeInstrumentedFirewallOperationUpdate(ctx, fwID, fr)
 }
 
+type portProtocol struct {
+	port     int
+	protocol string
+}
+
 // createReconciledFirewallRequest creates a firewall request that has the correct rules, name and tag
-func (fm *firewallManager) createReconciledFirewallRequest(serviceList []*v1.Service) *godo.FirewallRequest {
+func (fm *firewallManager) createReconciledFirewallRequest(serviceList []*v1.Service) (*godo.FirewallRequest, error) {
 	var nodePortInboundRules []godo.InboundRule
+	loadBalancerPorts := make(map[portProtocol]struct{})
 	for _, svc := range serviceList {
-		managed, err := isManaged(svc)
-		if err != nil {
-			klog.Warningf("managing service %s/%s for which no correct management flag setting could be detected: %s", svc.Namespace, svc.Name, err)
-			managed = true
-		}
-		if !managed {
-			continue
-		}
-
 		if svc.Spec.Type == v1.ServiceTypeNodePort {
+			managed, err := isManaged(svc)
+			if err != nil {
+				klog.Warningf("managing service %s/%s for which no correct management flag setting could be detected: %s", svc.Namespace, svc.Name, err)
+				managed = true
+			}
+			if !managed {
+				continue
+			}
 			// this is a nodeport service so we should check for existing inbound rules on all ports.
 			for _, servicePort := range svc.Spec.Ports {
 				// In the odd case that a failure is asynchronous causing the NodePort to be set to zero.
@@ -312,14 +318,62 @@ func (fm *firewallManager) createReconciledFirewallRequest(serviceList []*v1.Ser
 					},
 				)
 			}
+		} else if svc.Spec.Type == v1.ServiceTypeLoadBalancer {
+			lbType, err := getType(svc)
+			if err != nil {
+				return nil, fmt.Errorf("failed to get load balancer type for service %s/%s: %v", svc.Namespace, svc.Name, err)
+			}
+			lbNetwork, err := getNetwork(svc)
+			if err != nil {
+				return nil, fmt.Errorf("failed to get load balancer network for service %s/%s: %v", svc.Namespace, svc.Name, err)
+			}
+			if lbType == godo.LoadBalancerTypeRegionalNetwork && lbNetwork == godo.LoadBalancerNetworkTypeExternal {
+				// Add the health check port
+				_, healthCheckPort := healthCheckPathAndPort(svc)
+				if healthCheckPort == 0 {
+					continue
+				}
+				loadBalancerPorts[portProtocol{protocol: "tcp", port: healthCheckPort}] = struct{}{}
+
+				// Add the services (port, protocol)
+				var protocol string
+				for _, servicePort := range svc.Spec.Ports {
+					switch servicePort.Protocol {
+					case v1.ProtocolTCP:
+						protocol = "tcp"
+					case v1.ProtocolUDP:
+						protocol = "udp"
+					default:
+						klog.Warningf("unsupported service protocol %v, skipping service port %v", servicePort.Protocol, servicePort.Name)
+						continue
+					}
+					loadBalancerPorts[portProtocol{protocol: protocol, port: int(servicePort.Port)}] = struct{}{}
+				}
+			}
 		}
 	}
+	for p := range loadBalancerPorts {
+		nodePortInboundRules = append(nodePortInboundRules, godo.InboundRule{
+			Protocol:  p.protocol,
+			PortRange: strconv.Itoa(p.port),
+			Sources: &godo.Sources{
+				Addresses: []string{"0.0.0.0/0", "::/0"},
+			},
+		})
+	}
+	// Sort for deterministic output
+	sort.SliceStable(nodePortInboundRules, func(i, j int) bool {
+		if nodePortInboundRules[i].Protocol == nodePortInboundRules[j].Protocol {
+			return nodePortInboundRules[i].PortRange < nodePortInboundRules[j].PortRange
+		}
+		return nodePortInboundRules[i].Protocol < nodePortInboundRules[j].Protocol
+	})
 	return &godo.FirewallRequest{
 		Name:          fm.workerFirewallName,
 		InboundRules:  nodePortInboundRules,
 		OutboundRules: allowAllOutboundRules,
 		Tags:          fm.workerFirewallTags,
-	}
+	}, nil
 }
 
 // isManaged returns if the given Service should be firewall-managed based on the
@@ -399,7 +453,10 @@ func (fc *FirewallController) ensureReconciledFirewall(ctx context.Context) (ski
 	if err != nil {
 		return false, fmt.Errorf("failed to list services: %v", err)
 	}
-	fr := fc.fwManager.createReconciledFirewallRequest(serviceList)
+	fr, err := fc.fwManager.createReconciledFirewallRequest(serviceList)
+	if err != nil {
+		return false, fmt.Errorf("failed to create reconciled firewall request: %v", err)
+	}
 
 	fw, err := fc.fwManager.GetPreferFromCache(ctx)
 	if err != nil {
diff --git a/cloud-controller-manager/do/firewall_controller_test.go b/cloud-controller-manager/do/firewall_controller_test.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strconv"
 	"testing"
 	"time"
 
@@ -296,6 +297,7 @@ func TestFirewallController_createReconciledFirewallRequest(t *testing.T) {
 		firewallRequest    *godo.FirewallRequest
 		firewallController FirewallController
 		serviceList        []*v1.Service
+		expectedError      error
 	}{
 		{
 			name: "nothing to reconcile when there are no changes",
@@ -425,21 +427,21 @@ func TestFirewallController_createReconciledFirewallRequest(t *testing.T) {
 				InboundRules: []godo.InboundRule{
 					{
 						Protocol:  "tcp",
-						PortRange: "31000",
+						PortRange: "30000",
 						Sources: &godo.Sources{
 							Addresses: []string{"0.0.0.0/0", "::/0"},
 						},
 					},
 					{
-						Protocol:  "udp",
+						Protocol:  "tcp",
 						PortRange: "31000",
 						Sources: &godo.Sources{
 							Addresses: []string{"0.0.0.0/0", "::/0"},
 						},
 					},
 					{
 						Protocol:  "tcp",
-						PortRange: "30000",
+						PortRange: "32727",
 						Sources: &godo.Sources{
 							Addresses: []string{"0.0.0.0/0", "::/0"},
 						},
@@ -452,8 +454,8 @@ func TestFirewallController_createReconciledFirewallRequest(t *testing.T) {
 						},
 					},
 					{
-						Protocol:  "tcp",
-						PortRange: "32727",
+						Protocol:  "udp",
+						PortRange: "31000",
 						Sources: &godo.Sources{
 							Addresses: []string{"0.0.0.0/0", "::/0"},
 						},
@@ -524,6 +526,119 @@ func TestFirewallController_createReconciledFirewallRequest(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "reconcile firewall with REGIONAL_NETWORK LB w/ externalTrafficPolicy=Cluster",
+			firewallRequest: &godo.FirewallRequest{
+				Name: testWorkerFWName,
+				InboundRules: []godo.InboundRule{
+					{
+						Protocol:  "tcp",
+						PortRange: strconv.Itoa(kubeProxyHealthPort),
+						Sources: &godo.Sources{
+							Addresses: []string{"0.0.0.0/0", "::/0"},
+						},
+					},
+					{
+						Protocol:  "tcp",
+						PortRange: strconv.Itoa(443),
+						Sources: &godo.Sources{
+							Addresses: []string{"0.0.0.0/0", "::/0"},
+						},
+					},
+					{
+						Protocol:  "tcp",
+						PortRange: strconv.Itoa(80),
+						Sources: &godo.Sources{
+							Addresses: []string{"0.0.0.0/0", "::/0"},
+						},
+					},
+				},
+				OutboundRules: testOutboundRules,
+				Tags:          testWorkerFWTags,
+			},
+			serviceList: []*v1.Service{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "regional_network",
+						UID:  "abc123",
+						Annotations: map[string]string{
+							annDOType: godo.LoadBalancerTypeRegionalNetwork,
+						},
+					},
+					Spec: v1.ServiceSpec{
+						Type:                  v1.ServiceTypeLoadBalancer,
+						ExternalTrafficPolicy: v1.ServiceExternalTrafficPolicyCluster,
+						Ports: []v1.ServicePort{
+							{
+								Protocol: v1.ProtocolTCP,
+								Port:     80,
+							},
+							{
+								Protocol: v1.ProtocolTCP,
+								Port:     443,
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "reconcile firewall with REGIONAL_NETWORK LB w/ externalTrafficPolicy=Local",
+			firewallRequest: &godo.FirewallRequest{
+				Name: testWorkerFWName,
+				InboundRules: []godo.InboundRule{
+					{
+						Protocol:  "tcp",
+						PortRange: strconv.Itoa(15000),
+						Sources: &godo.Sources{
+							Addresses: []string{"0.0.0.0/0", "::/0"},
+						},
+					},
+					{
+						Protocol:  "tcp",
+						PortRange: strconv.Itoa(443),
+						Sources: &godo.Sources{
+							Addresses: []string{"0.0.0.0/0", "::/0"},
+						},
+					},
+					{
+						Protocol:  "tcp",
+						PortRange: strconv.Itoa(80),
+						Sources: &godo.Sources{
+							Addresses: []string{"0.0.0.0/0", "::/0"},
+						},
+					},
+				},
+				OutboundRules: testOutboundRules,
+				Tags:          testWorkerFWTags,
+			},
+			serviceList: []*v1.Service{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "regional_network",
+						UID:  "abc123",
+						Annotations: map[string]string{
+							annDOType: godo.LoadBalancerTypeRegionalNetwork,
+						},
+					},
+					Spec: v1.ServiceSpec{
+						Type:                  v1.ServiceTypeLoadBalancer,
+						ExternalTrafficPolicy: v1.ServiceExternalTrafficPolicyLocal,
+						HealthCheckNodePort:   15000,
+						Ports: []v1.ServicePort{
+							{
+								Protocol: v1.ProtocolTCP,
+								Port:     80,
+							},
+							{
+								Protocol: v1.ProtocolTCP,
+								Port:     443,
+							},
+						},
+					},
+				},
+			},
+		},
 		{
 			name: "reconcile firewall with management flag",
 			firewallRequest: &godo.FirewallRequest{
@@ -638,7 +753,11 @@ func TestFirewallController_createReconciledFirewallRequest(t *testing.T) {
 				workerFirewallTags: testWorkerFWTags,
 				workerFirewallName: testWorkerFWName,
 			}
-			fwReq := fm.createReconciledFirewallRequest(test.serviceList)
+			fwReq, err := fm.createReconciledFirewallRequest(test.serviceList)
+
+			if (err != nil && test.expectedError == nil) || (err == nil && test.expectedError != nil) {
+				t.Fatalf("expected error %q, got %q", test.expectedError, err)
+			}
 			if diff := cmp.Diff(test.firewallRequest, fwReq); diff != "" {
 				t.Errorf("createReconciledFirewallRequest() mismatch (-want +got):\n%s", diff)
 			}
@@ -675,15 +794,18 @@ func TestFirewallController_ensureReconciledFirewall(t *testing.T) {
 		}
 	}
 
-	serviceToFirewall := func(fm *firewallManager, svc *v1.Service) *godo.Firewall {
-		fr := fm.createReconciledFirewallRequest([]*v1.Service{svc})
+	serviceToFirewall := func(fm *firewallManager, svc *v1.Service) (*godo.Firewall, error) {
+		fr, err := fm.createReconciledFirewallRequest([]*v1.Service{svc})
+		if err != nil {
+			return nil, err
+		}
 		return &godo.Firewall{
 			ID:            "id",
 			Name:          fr.Name,
 			InboundRules:  fr.InboundRules,
 			OutboundRules: fr.OutboundRules,
 			Tags:          fr.Tags,
-		}
+		}, nil
 	}
 
 	tests := []struct {
@@ -757,7 +879,10 @@ func TestFirewallController_ensureReconciledFirewall(t *testing.T) {
 
 			// Populate the firewall cache.
 			if test.nodePortForCachedFirewall != nil {
-				fw := serviceToFirewall(fwManager, nodePortToService(*test.nodePortForCachedFirewall))
+				fw, err := serviceToFirewall(fwManager, nodePortToService(*test.nodePortForCachedFirewall))
+				if err != nil {
+					t.Fatal(err)
+				}
 				fwManager.fwCache.updateCache(fw)
 			}
 
