From 4faa88311d5988af2604456654a20585d9a9a0ae Mon Sep 17 00:00:00 2001 From: zhangzujian Date: Tue, 19 Apr 2022 10:46:38 +0800 Subject: [PATCH] add kube-ovn-controller switch for EIP and SNAT --- pkg/controller/config.go | 3 +++ pkg/controller/pod.go | 42 +++++++++++++++++++++++++--------------- pkg/controller/subnet.go | 5 ++++- 3 files changed, 33 insertions(+), 17 deletions(-) diff --git a/pkg/controller/config.go b/pkg/controller/config.go index d8fa8e0dc42..968aefd8f9d 100644 --- a/pkg/controller/config.go +++ b/pkg/controller/config.go @@ -70,6 +70,7 @@ type Configuration struct { EnableLb bool EnableNP bool + EnableEipSnat bool EnableExternalVpc bool EnableEcmp bool EnableKeepVmIP bool @@ -118,6 +119,7 @@ func ParseFlags() (*Configuration, error) { argPodNicType = pflag.String("pod-nic-type", "veth-pair", "The default pod network nic implementation type") argEnableLb = pflag.Bool("enable-lb", true, "Enable load balancer") argEnableNP = pflag.Bool("enable-np", true, "Enable network policy support") + argEnableEipSnat = pflag.Bool("enable-eip-snat", true, "Enable EIP and SNAT") argEnableExternalVpc = pflag.Bool("enable-external-vpc", true, "Enable external vpc support") argEnableEcmp = pflag.Bool("enable-ecmp", false, "Enable ecmp route for centralized subnet") argKeepVmIP = pflag.Bool("keep-vm-ip", false, "Whether to keep ip for kubevirt pod when pod is rebuild") @@ -177,6 +179,7 @@ func ParseFlags() (*Configuration, error) { PodNicType: *argPodNicType, EnableLb: *argEnableLb, EnableNP: *argEnableNP, + EnableEipSnat: *argEnableEipSnat, EnableExternalVpc: *argEnableExternalVpc, ExternalGatewayConfigNS: *argExternalGatewayConfigNS, ExternalGatewayNet: *argExternalGatewayNet, diff --git a/pkg/controller/pod.go b/pkg/controller/pod.go index d21528e0217..e0f291b4fb1 100644 --- a/pkg/controller/pod.go +++ b/pkg/controller/pod.go @@ -324,14 +324,14 @@ func (c *Controller) processNextDeletePodWorkItem() bool { utilruntime.HandleError(fmt.Errorf("expected pod in workqueue but got %#v", obj)) return nil } - klog.Infof("handle delete pod %s", pod.Name) + klog.Infof("handle delete pod %s/%s", pod.Namespace, pod.Name) if err := c.handleDeletePod(pod); err != nil { c.deletePodQueue.AddRateLimited(obj) return fmt.Errorf("error syncing '%s': %s, requeuing", pod.Name, err.Error()) } c.deletePodQueue.Forget(obj) last := time.Since(now) - klog.Infof("take %d ms to handle delete pod %s", last.Milliseconds(), pod.Name) + klog.Infof("take %d ms to handle delete pod %s/%s", last.Milliseconds(), pod.Namespace, pod.Name) return nil }(obj) @@ -349,6 +349,7 @@ func (c *Controller) processNextUpdatePodWorkItem() bool { return false } + now := time.Now() err := func(obj interface{}) error { defer c.updatePodQueue.Done(obj) var key string @@ -358,11 +359,14 @@ func (c *Controller) processNextUpdatePodWorkItem() bool { utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj)) return nil } + klog.Infof("handle update pod %s", key) if err := c.handleUpdatePod(key); err != nil { c.updatePodQueue.AddRateLimited(key) return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error()) } c.updatePodQueue.Forget(obj) + last := time.Since(now) + klog.Infof("take %d ms to handle update pod %s", last.Milliseconds(), key) return nil }(obj) @@ -430,13 +434,15 @@ func (c *Controller) getPodKubeovnNets(pod *v1.Pod) ([]*kubeovnNet, error) { } func (c *Controller) handleAddPod(key string) error { - c.podKeyMutex.Lock(key) - defer c.podKeyMutex.Unlock(key) namespace, name, err := cache.SplitMetaNamespaceKey(key) if err != nil { utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key)) return nil } + + c.podKeyMutex.Lock(key) + defer c.podKeyMutex.Unlock(key) + oripod, err := c.podsLister.Pods(namespace).Get(name) if err != nil { if k8serrors.IsNotFound(err) { @@ -458,7 +464,7 @@ func (c *Controller) handleAddPod(key string) error { } op := "replace" - if pod.Annotations == nil || len(pod.Annotations) == 0 { + if len(pod.Annotations) == 0 { op = "add" pod.Annotations = map[string]string{} } @@ -736,13 +742,15 @@ func (c *Controller) handleUpdatePodSecurity(key string) error { } func (c *Controller) handleUpdatePod(key string) error { - c.podKeyMutex.Lock(key) - defer c.podKeyMutex.Unlock(key) namespace, name, err := cache.SplitMetaNamespaceKey(key) if err != nil { utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key)) return nil } + + c.podKeyMutex.Lock(key) + defer c.podKeyMutex.Unlock(key) + oripod, err := c.podsLister.Pods(namespace).Get(name) if err != nil { if k8serrors.IsNotFound(err) { @@ -776,7 +784,7 @@ func (c *Controller) handleUpdatePod(key string) error { subnet = podNet.Subnet if podIP != "" && subnet.Spec.Vlan == "" && subnet.Spec.Vpc == util.DefaultVpc { - if pod.Annotations[util.EipAnnotation] != "" || pod.Annotations[util.SnatAnnotation] != "" { + if c.config.EnableEipSnat && (pod.Annotations[util.EipAnnotation] != "" || pod.Annotations[util.SnatAnnotation] != "") { cm, err := c.configMapsLister.ConfigMaps(c.config.ExternalGatewayConfigNS).Get(util.ExternalGatewayConfig) if err != nil { klog.Errorf("failed to get ex-gateway config, %v", err) @@ -833,15 +841,17 @@ func (c *Controller) handleUpdatePod(key string) error { } } - for _, ipStr := range strings.Split(podIP, ",") { - if err := c.ovnClient.UpdateNatRule("dnat_and_snat", ipStr, pod.Annotations[util.EipAnnotation], c.config.ClusterRouter, pod.Annotations[util.MacAddressAnnotation], fmt.Sprintf("%s.%s", podName, pod.Namespace)); err != nil { - klog.Errorf("failed to add nat rules, %v", err) - return err - } + if c.config.EnableEipSnat { + for _, ipStr := range strings.Split(podIP, ",") { + if err := c.ovnClient.UpdateNatRule("dnat_and_snat", ipStr, pod.Annotations[util.EipAnnotation], c.config.ClusterRouter, pod.Annotations[util.MacAddressAnnotation], fmt.Sprintf("%s.%s", podName, pod.Namespace)); err != nil { + klog.Errorf("failed to add nat rules, %v", err) + return err + } - if err := c.ovnClient.UpdateNatRule("snat", ipStr, pod.Annotations[util.SnatAnnotation], c.config.ClusterRouter, "", ""); err != nil { - klog.Errorf("failed to add nat rules, %v", err) - return err + if err := c.ovnClient.UpdateNatRule("snat", ipStr, pod.Annotations[util.SnatAnnotation], c.config.ClusterRouter, "", ""); err != nil { + klog.Errorf("failed to add nat rules, %v", err) + return err + } } } } diff --git a/pkg/controller/subnet.go b/pkg/controller/subnet.go index 8169be73182..34e37468e34 100644 --- a/pkg/controller/subnet.go +++ b/pkg/controller/subnet.go @@ -897,7 +897,10 @@ func (c *Controller) reconcileGateway(subnet *kubeovnv1.Subnet) error { } for _, pod := range pods { - if !isPodAlive(pod) || pod.Annotations[util.IpAddressAnnotation] == "" || pod.Annotations[util.LogicalSwitchAnnotation] != subnet.Name { + if !isPodAlive(pod) { + continue + } + if c.config.EnableEipSnat && (pod.Annotations[util.EipAnnotation] != "" || pod.Annotations[util.SnatAnnotation] != "") { continue }