Skip to content

Commit

Permalink
add kube-ovn-controller switch for EIP and SNAT
Browse files Browse the repository at this point in the history
  • Loading branch information
@zhangzujian
zhangzujian committed Apr 19, 2022
1 parent 300a164 commit 4faa883
Show file tree
Hide file tree
Showing 3 changed files with 33 additions and 17 deletions.
3 changes: 3 additions & 0 deletions pkg/controller/config.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@ type Configuration struct {

EnableLb bool
EnableNP bool
EnableEipSnat bool
EnableExternalVpc bool
EnableEcmp bool
EnableKeepVmIP bool
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -177,6 +179,7 @@ func ParseFlags() (*Configuration, error) {
PodNicType: *argPodNicType,
EnableLb: *argEnableLb,
EnableNP: *argEnableNP,
EnableEipSnat: *argEnableEipSnat,
EnableExternalVpc: *argEnableExternalVpc,
ExternalGatewayConfigNS: *argExternalGatewayConfigNS,
ExternalGatewayNet: *argExternalGatewayNet,
Expand Down
42 changes: 26 additions & 16 deletions pkg/controller/pod.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand All @@ -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
Expand All @@ -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)

Expand Down Expand Up @@ -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) {
Expand All @@ -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{}
}
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
}
}
}
}
Expand Down
5 changes: 4 additions & 1 deletion pkg/controller/subnet.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand Down

0 comments on commit 4faa883

Please sign in to comment.