[feat] : add support for configuring nodebalancers with VPC (#337)

* intial commit for nb-vpc support

* fix loadbalancer update and add documentation

* add unittests for changes

* add e2e test

Author: rahulait

Makefile

@@ -162,6 +162,7 @@ create-capl-cluster:
 .PHONY: patch-linode-ccm
 patch-linode-ccm:
 	KUBECONFIG=$(KUBECONFIG_PATH) kubectl patch -n kube-system daemonset ccm-linode --type='json' -p="[{'op': 'replace', 'path': '/spec/template/spec/containers/0/image', 'value': '${IMG}'}]"
+	KUBECONFIG=$(KUBECONFIG_PATH) kubectl patch -n kube-system daemonset ccm-linode --type='json' -p='[{"op": "add", "path": "/spec/template/spec/containers/0/env/-", "value": {"name": "LINODE_API_VERSION", "value": "v4beta"}}]'
 	KUBECONFIG=$(KUBECONFIG_PATH) kubectl rollout status -n kube-system daemonset/ccm-linode --timeout=600s
 	KUBECONFIG=$(KUBECONFIG_PATH) kubectl -n kube-system get daemonset/ccm-linode -o yaml
 

cloud/annotations/annotations.go

@@ -34,4 +34,9 @@ const (
 	AnnLinodeHostUUID      = "node.k8s.linode.com/host-uuid"
 
 	AnnLinodeNodeIPSharingUpdated = "node.k8s.linode.com/ip-sharing-updated"
+
+	NodeBalancerBackendIPv4Range = "service.beta.kubernetes.io/linode-loadbalancer-backend-ipv4-range"
+
+	NodeBalancerBackendVPCName    = "service.beta.kubernetes.io/linode-loadbalancer-backend-vpc-name"
+	NodeBalancerBackendSubnetName = "service.beta.kubernetes.io/linode-loadbalancer-backend-subnet-name"
 )

cloud/linode/fake_linode_test.go

@@ -19,12 +19,15 @@ import (
 const apiVersion = "v4"
 
 type fakeAPI struct {
-	t   *testing.T
-	nb  map[string]*linodego.NodeBalancer
-	nbc map[string]*linodego.NodeBalancerConfig
-	nbn map[string]*linodego.NodeBalancerNode
-	fw  map[int]*linodego.Firewall               // map of firewallID -> firewall
-	fwd map[int]map[int]*linodego.FirewallDevice // map of firewallID -> firewallDeviceID:FirewallDevice
+	t      *testing.T
+	nb     map[string]*linodego.NodeBalancer
+	nbc    map[string]*linodego.NodeBalancerConfig
+	nbn    map[string]*linodego.NodeBalancerNode
+	fw     map[int]*linodego.Firewall               // map of firewallID -> firewall
+	fwd    map[int]map[int]*linodego.FirewallDevice // map of firewallID -> firewallDeviceID:FirewallDevice
+	nbvpcc map[string]*linodego.NodeBalancerVPCConfig
+	vpc    map[int]*linodego.VPC
+	subnet map[int]*linodego.VPCSubnet
 
 	requests map[fakeRequest]struct{}
 	mux      *http.ServeMux
@@ -44,6 +47,9 @@ func newFake(t *testing.T) *fakeAPI {
 		nbn:      make(map[string]*linodego.NodeBalancerNode),
 		fw:       make(map[int]*linodego.Firewall),
 		fwd:      make(map[int]map[int]*linodego.FirewallDevice),
+		nbvpcc:   make(map[string]*linodego.NodeBalancerVPCConfig),
+		vpc:      make(map[int]*linodego.VPC),
+		subnet:   make(map[int]*linodego.VPCSubnet),
 		requests: make(map[fakeRequest]struct{}),
 		mux:      http.NewServeMux(),
 	}
@@ -117,6 +123,54 @@ func (f *fakeAPI) setupRoutes() {
 		_, _ = w.Write(rr)
 	})
 
+	f.mux.HandleFunc("GET /v4/vpcs", func(w http.ResponseWriter, r *http.Request) {
+		res := 0
+		data := []linodego.VPC{}
+		filter := r.Header.Get("X-Filter")
+		if filter == "" {
+			for _, v := range f.vpc {
+				data = append(data, *v)
+			}
+		} else {
+			var fs map[string]string
+			err := json.Unmarshal([]byte(filter), &fs)
+			if err != nil {
+				f.t.Fatal(err)
+			}
+			for _, v := range f.vpc {
+				if v.Label != "" && fs["label"] != "" && v.Label == fs["label"] {
+					data = append(data, *v)
+				}
+			}
+		}
+
+		resp := paginatedResponse[linodego.VPC]{
+			Page:    1,
+			Pages:   1,
+			Results: res,
+			Data:    data,
+		}
+		rr, _ := json.Marshal(resp)
+		_, _ = w.Write(rr)
+	})
+
+	f.mux.HandleFunc("GET /v4/vpcs/{vpcId}/subnets", func(w http.ResponseWriter, r *http.Request) {
+		res := 0
+		vpcID, err := strconv.Atoi(r.PathValue("vpcId"))
+		if err != nil {
+			f.t.Fatal(err)
+		}
+
+		resp := paginatedResponse[linodego.VPCSubnet]{
+			Page:    1,
+			Pages:   1,
+			Results: res,
+			Data:    f.vpc[vpcID].Subnets,
+		}
+		rr, _ := json.Marshal(resp)
+		_, _ = w.Write(rr)
+	})
+
 	f.mux.HandleFunc("GET /v4/nodebalancers/{nodeBalancerId}", func(w http.ResponseWriter, r *http.Request) {
 		nb, found := f.nb[r.PathValue("nodeBalancerId")]
 		if !found {
@@ -462,6 +516,54 @@ func (f *fakeAPI) setupRoutes() {
 		_, _ = w.Write(resp)
 	})
 
+	f.mux.HandleFunc("POST /v4/vpcs", func(w http.ResponseWriter, r *http.Request) {
+		vco := linodego.VPCCreateOptions{}
+		if err := json.NewDecoder(r.Body).Decode(&vco); err != nil {
+			f.t.Fatal(err)
+		}
+
+		subnets := []linodego.VPCSubnet{}
+		for _, s := range vco.Subnets {
+			subnet := linodego.VPCSubnet{
+				ID:    rand.Intn(9999),
+				IPv4:  s.IPv4,
+				Label: s.Label,
+			}
+			subnets = append(subnets, subnet)
+			f.subnet[subnet.ID] = &subnet
+		}
+		vpc := linodego.VPC{
+			ID:          rand.Intn(9999),
+			Label:       vco.Label,
+			Description: vco.Description,
+			Region:      vco.Region,
+			Subnets:     subnets,
+		}
+
+		f.vpc[vpc.ID] = &vpc
+		resp, err := json.Marshal(vpc)
+		if err != nil {
+			f.t.Fatal(err)
+		}
+		_, _ = w.Write(resp)
+	})
+
+	f.mux.HandleFunc("DELETE /v4/vpcs/{vpcId}", func(w http.ResponseWriter, r *http.Request) {
+		vpcid, err := strconv.Atoi(r.PathValue("vpcId"))
+		if err != nil {
+			f.t.Fatal(err)
+		}
+
+		for k, v := range f.vpc {
+			if v.ID == vpcid {
+				for _, s := range v.Subnets {
+					delete(f.subnet, s.ID)
+				}
+				delete(f.vpc, k)
+			}
+		}
+	})
+
 	f.mux.HandleFunc("POST /v4/networking/firewalls/{firewallId}/devices", func(w http.ResponseWriter, r *http.Request) {
 		fdco := linodego.FirewallDeviceCreateOptions{}
 		if err := json.NewDecoder(r.Body).Decode(&fdco); err != nil {

cloud/linode/loadbalancers.go

@@ -383,8 +383,18 @@ func (l *loadbalancers) updateNodeBalancer(
 		}
 		// Add all of the Nodes to the config
 		newNBNodes := make([]linodego.NodeBalancerConfigRebuildNodeOptions, 0, len(nodes))
+		subnetID := 0
+		_, ok := service.GetAnnotations()[annotations.NodeBalancerBackendIPv4Range]
+		if ok {
+			id, err := l.getSubnetIDForSVC(ctx, service)
+			if err != nil {
+				sentry.CaptureError(ctx, err)
+				return fmt.Errorf("Error getting subnet ID for service %s: %v", service.Name, err)
+			}
+			subnetID = id
+		}
 		for _, node := range nodes {
-			newNodeOpts := l.buildNodeBalancerNodeConfigRebuildOptions(node, port.NodePort)
+			newNodeOpts := l.buildNodeBalancerNodeConfigRebuildOptions(node, port.NodePort, subnetID)
 			oldNodeID, ok := oldNBNodeIDs[newNodeOpts.Address]
 			if ok {
 				newNodeOpts.ID = oldNodeID
@@ -652,6 +662,20 @@ func (l *loadbalancers) createNodeBalancer(ctx context.Context, clusterName stri
 		Type:               nbType,
 	}
 
+	backendIPv4Range, ok := service.GetAnnotations()[annotations.NodeBalancerBackendIPv4Range]
+	if ok {
+		subnetID, err := l.getSubnetIDForSVC(ctx, service)
+		if err != nil {
+			return nil, err
+		}
+		createOpts.VPCs = []linodego.NodeBalancerVPCOptions{
+			{
+				SubnetID:  subnetID,
+				IPv4Range: backendIPv4Range,
+			},
+		}
+	}
+
 	fwid, ok := service.GetAnnotations()[annotations.AnnLinodeCloudFirewallID]
 	if ok {
 		firewallID, err := strconv.Atoi(fwid)
@@ -768,6 +792,28 @@ func (l *loadbalancers) addTLSCert(ctx context.Context, service *v1.Service, nbC
 	return nil
 }
 
+// getSubnetIDForSVC returns the subnet ID for the service's VPC and subnet.
+// By default, first VPCName and SubnetName are used to calculate subnet id for the service.
+// If the service has annotations specifying VPCName and SubnetName, they are used instead.
+func (l *loadbalancers) getSubnetIDForSVC(ctx context.Context, service *v1.Service) (int, error) {
+	if Options.VPCNames == "" {
+		return 0, fmt.Errorf("CCM not configured with VPC, cannot create NodeBalancer with specified annotation")
+	}
+	vpcName := strings.Split(Options.VPCNames, ",")[0]
+	if specifiedVPCName, ok := service.GetAnnotations()[annotations.NodeBalancerBackendVPCName]; ok {
+		vpcName = specifiedVPCName
+	}
+	vpcID, err := GetVPCID(ctx, l.client, vpcName)
+	if err != nil {
+		return 0, err
+	}
+	subnetName := strings.Split(Options.SubnetNames, ",")[0]
+	if specifiedSubnetName, ok := service.GetAnnotations()[annotations.NodeBalancerBackendSubnetName]; ok {
+		subnetName = specifiedSubnetName
+	}
+	return GetSubnetID(ctx, l.client, vpcID, subnetName)
+}
+
 // buildLoadBalancerRequest returns a linodego.NodeBalancer
 // requests for service across nodes.
 func (l *loadbalancers) buildLoadBalancerRequest(ctx context.Context, clusterName string, service *v1.Service, nodes []*v1.Node) (*linodego.NodeBalancer, error) {
@@ -777,6 +823,16 @@ func (l *loadbalancers) buildLoadBalancerRequest(ctx context.Context, clusterNam
 	ports := service.Spec.Ports
 	configs := make([]*linodego.NodeBalancerConfigCreateOptions, 0, len(ports))
 
+	subnetID := 0
+	_, ok := service.GetAnnotations()[annotations.NodeBalancerBackendIPv4Range]
+	if ok {
+		id, err := l.getSubnetIDForSVC(ctx, service)
+		if err != nil {
+			return nil, err
+		}
+		subnetID = id
+	}
+
 	for _, port := range ports {
 		if port.Protocol == v1.ProtocolUDP {
 			return nil, fmt.Errorf("error creating NodeBalancer Config: ports with the UDP protocol are not supported")
@@ -789,7 +845,7 @@ func (l *loadbalancers) buildLoadBalancerRequest(ctx context.Context, clusterNam
 		createOpt := config.GetCreateOptions()
 
 		for _, n := range nodes {
-			createOpt.Nodes = append(createOpt.Nodes, l.buildNodeBalancerNodeConfigRebuildOptions(n, port.NodePort).NodeBalancerNodeCreateOptions)
+			createOpt.Nodes = append(createOpt.Nodes, l.buildNodeBalancerNodeConfigRebuildOptions(n, port.NodePort, subnetID).NodeBalancerNodeCreateOptions)
 		}
 
 		configs = append(configs, &createOpt)
@@ -809,17 +865,21 @@ func coerceString(s string, minLen, maxLen int, padding string) string {
 	return s
 }
 
-func (l *loadbalancers) buildNodeBalancerNodeConfigRebuildOptions(node *v1.Node, nodePort int32) linodego.NodeBalancerConfigRebuildNodeOptions {
-	return linodego.NodeBalancerConfigRebuildNodeOptions{
+func (l *loadbalancers) buildNodeBalancerNodeConfigRebuildOptions(node *v1.Node, nodePort int32, subnetID int) linodego.NodeBalancerConfigRebuildNodeOptions {
+	nodeOptions := linodego.NodeBalancerConfigRebuildNodeOptions{
 		NodeBalancerNodeCreateOptions: linodego.NodeBalancerNodeCreateOptions{
-			Address: fmt.Sprintf("%v:%v", getNodePrivateIP(node), nodePort),
+			Address: fmt.Sprintf("%v:%v", getNodePrivateIP(node, subnetID), nodePort),
 			// NodeBalancer backends must be 3-32 chars in length
 			// If < 3 chars, pad node name with "node-" prefix
 			Label:  coerceString(node.Name, 3, 32, "node-"),
 			Mode:   "accept",
 			Weight: 100,
 		},
 	}
+	if subnetID != 0 {
+		nodeOptions.NodeBalancerNodeCreateOptions.SubnetID = subnetID
+	}
+	return nodeOptions
 }
 
 func (l *loadbalancers) retrieveKubeClient() error {
@@ -926,13 +986,17 @@ func getPortConfigAnnotation(service *v1.Service, port int) (portConfigAnnotatio
 	return annotation, nil
 }
 
-// getNodePrivateIP should provide the Linode Private IP the NodeBalance
-// will communicate with. When using a VLAN or VPC for the Kubernetes cluster
-// network, this will not be the NodeInternalIP, so this prefers an annotation
-// cluster operators may specify in such a situation.
-func getNodePrivateIP(node *v1.Node) string {
-	if address, exists := node.Annotations[annotations.AnnLinodeNodePrivateIP]; exists {
-		return address
+// getNodePrivateIP provides the Linode Backend IP the NodeBalancer will communicate with.
+// If a service specifies NodeBalancerBackendIPv4Range annotation, it will
+// use NodeInternalIP of node.
+// For services which don't have NodeBalancerBackendIPv4Range annotation,
+// Backend IP can be overwritten to the one specified using AnnLinodeNodePrivateIP
+// annotation over the NodeInternalIP.
+func getNodePrivateIP(node *v1.Node, subnetID int) string {
+	if subnetID == 0 {
+		if address, exists := node.Annotations[annotations.AnnLinodeNodePrivateIP]; exists {
+			return address
+		}
 	}
 
 	klog.Infof("Node %s, assigned IP addresses: %v", node.Name, node.Status.Addresses)