[oocm] support for externalTrafficPolicy: Local (kubernetes#1720)

* support for externalTrafficPolicy local

with externalTrafficPolicy local we create an http healthmonitor that checks the dedicated healthCheckNodePort.

Signed-off-by: Fabian Ruff <fabian.ruff@sap.com>

* remove limitation from docs

Signed-off-by: Fabian Ruff <fabian.ruff@sap.com>

* address feedback

* Add some note that health monitors are mandatory for externalTrafficPolicy: Local

Author: databus23

docs/openstack-cloud-controller-manager/expose-applications-using-loadbalancer-type-service.md

@@ -173,7 +173,7 @@ Request Body:
 
 - `loadbalancer.openstack.org/enable-health-monitor`
 
-  Defines whether to create health monitor for the load balancer pool, if not specified, use `create-monitor` config. The health monitor can be created or deleted dynamically.
+  Defines whether to create health monitor for the load balancer pool, if not specified, use `create-monitor` config. The health monitor can be created or deleted dynamically. A health monitor is required for services with `externalTrafficPolicy: Local`.
 
   Not supported when `lb-provider=ovn` is configured in openstack-cloud-controller-manager.
 

docs/openstack-cloud-controller-manager/using-openstack-cloud-controller-manager.md

@@ -16,7 +16,6 @@
   - [Metrics](#metrics)
   - [Limitation](#limitation)
     - [OpenStack availability zone must not contain blank](#openstack-availability-zone-must-not-contain-blank)
-    - [externalTrafficPolicy support](#externaltrafficpolicy-support)
 
 <!-- END doctoc generated TOC please keep comment here to allow auto update -->
 
@@ -209,7 +208,7 @@ Although the openstack-cloud-controller-manager was initially implemented with N
   This option is not supported for Octavia. The worker nodes and the Octavia amphorae are usually in the same subnet, so it's sufficient to config the port security group rules manually for worker nodes, to allow the traffic coming from the the subnet IP range to the node port range(i.e. 30000-32767).
 
 * `create-monitor`
-  Indicates whether or not to create a health monitor for the service load balancer. Default: false
+  Indicates whether or not to create a health monitor for the service load balancer. A health monitor required for services that declare `externalTrafficPolicy: Local`. Default: false
 
 * `monitor-delay`
   The time, in seconds, between sending probes to members of the load balancer. Default: 5
@@ -289,9 +288,3 @@ Refer to [Metrics for openstack-cloud-controller-manager](../metrics.md)
 ### OpenStack availability zone must not contain blank
 
 `topology.kubernetes.io/zone` is used to label node and its value comes from availability zone of the node, according to [label spec](https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#syntax-and-character-set) it does not support blank (' ') but OpenStack availability zone supports blank. So your OpenStack availability zone must not contain blank otherwise it will lead to node that belongs to this availability zone register failure, see [#1379](https://github.com/kubernetes/cloud-provider-openstack/issues/1379) for further information.
-
-### externalTrafficPolicy support
-
-`externalTrafficPolicy` denotes if this Service desires to route external traffic to node-local or cluster-wide endpoints. "Local" preserves the client source IP and avoids a second hop for LoadBalancer and Nodeport type services, but risks potentially imbalanced traffic spreading. "Cluster" obscures the client source IP and may cause a second hop to another node, but should have good overall load-spreading.
-
-openstack-cloud-controller-manager only supports `externalTrafficPolicy: Cluster` for now.

pkg/openstack/loadbalancer.go

@@ -337,6 +337,7 @@ type serviceConfig struct {
 	lbID                 string
 	lbName               string
 	supportLBTags        bool
+	healthCheckNodePort  int
 }
 
 type listenerKey struct {
@@ -623,7 +624,7 @@ func (lbaas *LbaasV2) createFullyPopulatedOctaviaLoadBalancer(name, clusterName
 		poolCreateOpt.Name = fmt.Sprintf("%s_%d_pool", listenerCreateOpt.Protocol, int(port.Port))
 		var withHealthMonitor string
 		if svcConf.enableMonitor {
-			opts := lbaas.buildMonitorCreateOpts(port)
+			opts := lbaas.buildMonitorCreateOpts(svcConf, port)
 			poolCreateOpt.Monitor = &opts
 			withHealthMonitor = " with healthmonitor"
 		}
@@ -1132,10 +1133,25 @@ func (lbaas *LbaasV2) getServiceAddress(clusterName string, service *corev1.Serv
 func (lbaas *LbaasV2) ensureOctaviaHealthMonitor(lbID string, name string, pool *v2pools.Pool, port corev1.ServicePort, svcConf *serviceConfig) error {
 	monitorID := pool.MonitorID
 
+	if monitorID != "" {
+		monitor, err := openstackutil.GetHealthMonitor(lbaas.lb, monitorID)
+		if err != nil {
+			return err
+		}
+		//Recreate health monitor with correct protocol if externalTrafficPolicy was changed
+		if (svcConf.healthCheckNodePort > 0 && monitor.Type != "HTTP") ||
+			(svcConf.healthCheckNodePort == 0 && monitor.Type == "HTTP") {
+			klog.InfoS("Recreating health monitor for the pool", "pool", pool.ID, "oldMonitor", monitorID)
+			if err := openstackutil.DeleteHealthMonitor(lbaas.lb, monitorID, lbID); err != nil {
+				return err
+			}
+			monitorID = ""
+		}
+	}
 	if monitorID == "" && svcConf.enableMonitor {
 		klog.V(2).Infof("Creating monitor for pool %s", pool.ID)
 
-		createOpts := lbaas.buildMonitorCreateOpts(port)
+		createOpts := lbaas.buildMonitorCreateOpts(svcConf, port)
 		// Populate PoolID, attribute is omitted for consumption of the createOpts for fully populated Loadbalancer
 		createOpts.PoolID = pool.ID
 		createOpts.Name = name
@@ -1144,6 +1160,7 @@ func (lbaas *LbaasV2) ensureOctaviaHealthMonitor(lbID string, name string, pool
 			return err
 		}
 		monitorID = monitor.ID
+		klog.Infof("Health monitor %s for pool %s created.", monitorID, pool.ID)
 	} else if monitorID != "" && !svcConf.enableMonitor {
 		klog.Infof("Deleting health monitor %s for pool %s", monitorID, pool.ID)
 
@@ -1152,19 +1169,18 @@ func (lbaas *LbaasV2) ensureOctaviaHealthMonitor(lbID string, name string, pool
 		}
 	}
 
-	if monitorID != "" {
-		klog.Infof("Health monitor %s for pool %s created.", monitorID, pool.ID)
-	}
-
 	return nil
 }
 
 //buildMonitorCreateOpts returns a v2monitors.CreateOpts without PoolID for consumption of both, fully popuplated Loadbalancers and Monitors.
-func (lbaas *LbaasV2) buildMonitorCreateOpts(port corev1.ServicePort) v2monitors.CreateOpts {
+func (lbaas *LbaasV2) buildMonitorCreateOpts(svcConf *serviceConfig, port corev1.ServicePort) v2monitors.CreateOpts {
 	monitorProtocol := string(port.Protocol)
 	if port.Protocol == corev1.ProtocolUDP {
 		monitorProtocol = "UDP-CONNECT"
 	}
+	if svcConf.healthCheckNodePort > 0 {
+		monitorProtocol = "HTTP"
+	}
 	return v2monitors.CreateOpts{
 		Type:       monitorProtocol,
 		Delay:      int(lbaas.opts.MonitorDelay.Duration.Seconds()),
@@ -1209,17 +1225,16 @@ func (lbaas *LbaasV2) ensureOctaviaPool(lbID string, name string, listener *list
 		if err != nil {
 			return nil, err
 		}
+		klog.V(2).Infof("Pool %s created for listener %s", pool.ID, listener.ID)
 	}
 
-	klog.V(2).Infof("Pool %s created for listener %s", pool.ID, listener.ID)
-
 	curMembers := sets.NewString()
 	poolMembers, err := openstackutil.GetMembersbyPool(lbaas.lb, pool.ID)
 	if err != nil {
 		klog.Errorf("failed to get members in the pool %s: %v", pool.ID, err)
 	}
 	for _, m := range poolMembers {
-		curMembers.Insert(fmt.Sprintf("%s-%d", m.Address, m.ProtocolPort))
+		curMembers.Insert(fmt.Sprintf("%s-%d-%d", m.Address, m.ProtocolPort, m.MonitorPort))
 	}
 
 	members, newMembers, err := lbaas.buildBatchUpdateMemberOpts(port, nodes, svcConf)
@@ -1294,8 +1309,11 @@ func (lbaas *LbaasV2) buildBatchUpdateMemberOpts(port corev1.ServicePort, nodes
 			Name:         &node.Name,
 			SubnetID:     &svcConf.lbMemberSubnetID,
 		}
+		if svcConf.healthCheckNodePort > 0 {
+			member.MonitorPort = &svcConf.healthCheckNodePort
+		}
 		members = append(members, member)
-		newMembers.Insert(fmt.Sprintf("%s-%d", addr, member.ProtocolPort))
+		newMembers.Insert(fmt.Sprintf("%s-%d-%d", addr, member.ProtocolPort, svcConf.healthCheckNodePort))
 	}
 	return members, newMembers, nil
 }
@@ -1679,6 +1697,9 @@ func (lbaas *LbaasV2) checkService(service *corev1.Service, nodes []*corev1.Node
 	}
 
 	svcConf.enableMonitor = getBoolFromServiceAnnotation(service, ServiceAnnotationLoadBalancerEnableHealthMonitor, lbaas.opts.CreateMonitor)
+	if svcConf.enableMonitor && lbaas.opts.UseOctavia && service.Spec.ExternalTrafficPolicy == corev1.ServiceExternalTrafficPolicyTypeLocal && service.Spec.HealthCheckNodePort > 0 {
+		svcConf.healthCheckNodePort = int(service.Spec.HealthCheckNodePort)
+	}
 
 	return nil
 }

pkg/util/openstack/loadbalancer.go

@@ -668,3 +668,14 @@ func CreateHealthMonitor(client *gophercloud.ServiceClient, opts monitors.Create
 
 	return monitor, nil
 }
+
+// GetHealthMonitor gets details about loadbalancer health monitor.
+func GetHealthMonitor(client *gophercloud.ServiceClient, monitorID string) (*monitors.Monitor, error) {
+	mc := metrics.NewMetricContext("loadbalancer_healthmonitor", "get")
+	monitor, err := monitors.Get(client, monitorID).Extract()
+	if mc.ObserveRequest(err) != nil {
+		return nil, fmt.Errorf("failed to get healthmonitor: %v", err)
+	}
+
+	return monitor, nil
+}