feat: support nic-hotplug to a running pod. (#2521)

pkg/controller/pod.go

@@ -324,10 +324,25 @@ func (c *Controller) enqueueUpdatePod(oldObj, newObj interface{}) {
 		}()
 		return
 	}
+	// hotplug or hotunplug nic
+	oldPodNets, err := c.getPodKubeovnNets(oldPod)
+	if err != nil {
+		klog.Errorf("failed to get oldPod nets %v", err)
+		return
+	}
 
 	podNets, err := c.getPodKubeovnNets(newPod)
 	if err != nil {
-		klog.Errorf("failed to get pod nets %v", err)
+		klog.Errorf("failed to get newPod nets %v", err)
+		return
+	}
+
+	needHotUnPlug, needHotPlug := diffKubeovnNets(oldPodNets, podNets)
+	if len(needHotUnPlug) > 0 {
+		c.updatePodQueue.Add(key)
+	}
+	if len(needHotPlug) > 0 {
+		c.addPodQueue.Add(key)
 		return
 	}
 
@@ -935,6 +950,11 @@ func (c *Controller) handleUpdatePod(key string) error {
 		return err
 	}
 
+	if err = c.syncKubeOvnNet(pod, podNets); err != nil {
+		klog.Errorf("failed to sync pod nets %v", err)
+		return err
+	}
+
 	for _, podNet := range podNets {
 		if !isOvnSubnet(podNet.Subnet) {
 			continue
@@ -1066,6 +1086,56 @@ func (c *Controller) handleUpdatePod(key string) error {
 	return nil
 }
 
+func (c *Controller) syncKubeOvnNet(pod *v1.Pod, podNets []*kubeovnNet) error {
+	podName := c.getNameByPod(pod)
+	key := fmt.Sprintf("%s/%s", pod.Namespace, podName)
+	targetPortNameList := make(map[string]struct{})
+	portsNeedToDel := []string{}
+	subnetUsedByPort := make(map[string]string)
+
+	for _, podNet := range podNets {
+		portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)
+		targetPortNameList[portName] = struct{}{}
+	}
+
+	ports, err := c.ovnClient.ListNormalLogicalSwitchPorts(true, map[string]string{"pod": key})
+	if err != nil {
+		klog.Errorf("failed to list lsps of pod '%s', %v", pod.Name, err)
+		return err
+	}
+
+	for _, port := range ports {
+		if _, ok := targetPortNameList[port.Name]; !ok {
+			portsNeedToDel = append(portsNeedToDel, port.Name)
+			subnetUsedByPort[port.Name] = port.ExternalIDs["ls"]
+		}
+	}
+
+	if len(portsNeedToDel) == 0 {
+		return nil
+	}
+
+	for _, portNeedDel := range portsNeedToDel {
+
+		if subnet, ok := c.ipam.Subnets[subnetUsedByPort[portNeedDel]]; ok {
+			subnet.ReleaseAddressWithNicName(podName, portNeedDel)
+		}
+
+		if err := c.ovnClient.DeleteLogicalSwitchPort(portNeedDel); err != nil {
+			klog.Errorf("failed to delete lsp %s, %v", portNeedDel, err)
+			return err
+		}
+		if err := c.config.KubeOvnClient.KubeovnV1().IPs().Delete(context.Background(), portNeedDel, metav1.DeleteOptions{}); err != nil {
+			if !k8serrors.IsNotFound(err) {
+				klog.Errorf("failed to delete ip %s, %v", portNeedDel, err)
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
 func isStatefulSetPod(pod *v1.Pod) (bool, string) {
 	for _, owner := range pod.OwnerReferences {
 		if owner.Kind == "StatefulSet" && strings.HasPrefix(owner.APIVersion, "apps/") {
@@ -1233,6 +1303,26 @@ type kubeovnNet struct {
 	AllowLiveMigration bool
 }
 
+func diffKubeovnNets(exist []*kubeovnNet, target []*kubeovnNet) (needHotUnPlug []*kubeovnNet, needHotPlug []*kubeovnNet) {
+	existNetMap := make(map[string]*kubeovnNet, len(exist))
+	for _, item := range exist {
+		existNetMap[item.ProviderName] = item
+	}
+
+	for _, item := range target {
+		if _, ok := existNetMap[item.ProviderName]; ok {
+			delete(existNetMap, item.ProviderName)
+		} else {
+			needHotPlug = append(needHotPlug, item)
+		}
+	}
+	for _, item := range existNetMap {
+		needHotUnPlug = append(needHotUnPlug, item)
+	}
+
+	return
+}
+
 func (c *Controller) getPodAttachmentNet(pod *v1.Pod) ([]*kubeovnNet, error) {
 	var multusNets []*multustypes.NetworkSelectionElement
 	defaultAttachNetworks := pod.Annotations[util.DefaultNetworkAnnotation]

pkg/ipam/subnet.go

@@ -570,6 +570,14 @@ func (subnet *Subnet) ReleaseAddress(podName string) {
 	}
 }
 
+func (subnet *Subnet) ReleaseAddressWithNicName(podName, nicName string) {
+	subnet.mutex.Lock()
+	defer subnet.mutex.Unlock()
+
+	subnet.releaseAddr(podName, nicName)
+	subnet.popPodNic(podName, nicName)
+}
+
 func (subnet *Subnet) ContainAddress(address IP) bool {
 	subnet.mutex.RLock()
 	defer subnet.mutex.RUnlock()