Merge pull request #532 from l1b0k/feat/eip

Support migrate eip to ack-extend-network-controller

daemon/daemon.go

@@ -12,10 +12,15 @@ import (
 	"sync"
 	"time"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/util/retry"
+
 	"github.com/AliyunContainerService/terway/deviceplugin"
 	"github.com/AliyunContainerService/terway/pkg/aliyun"
 	"github.com/AliyunContainerService/terway/pkg/aliyun/client"
 	"github.com/AliyunContainerService/terway/pkg/aliyun/credential"
+	"github.com/AliyunContainerService/terway/pkg/apis/alibabacloud.com/v1beta1"
+	"github.com/AliyunContainerService/terway/pkg/apis/crds"
 	podENITypes "github.com/AliyunContainerService/terway/pkg/apis/network.alibabacloud.com/v1beta1"
 	"github.com/AliyunContainerService/terway/pkg/backoff"
 	vswpool "github.com/AliyunContainerService/terway/pkg/controller/vswitch"
@@ -34,6 +39,7 @@ import (
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	k8sErr "k8s.io/apimachinery/pkg/api/errors"
+	k8stypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/util/flowcontrol"
 )
@@ -1099,6 +1105,108 @@ func (n *networkService) exclusiveENIFromCRD(podInfo *types.PodInfo, waitReady b
 	return netConf, nil
 }
 
+func (n *networkService) migrateEIP(ctx context.Context, objs []interface{}) error {
+	once := sync.Once{}
+
+	for _, resObj := range objs {
+		podRes, ok := resObj.(types.PodResources)
+		if !ok {
+			continue
+		}
+		if podRes.PodInfo == nil || !podRes.PodInfo.EipInfo.PodEip {
+			continue
+		}
+		for _, eipRes := range podRes.Resources {
+			if eipRes.Type != types.ResourceTypeEIP {
+				continue
+			}
+			allocType := v1beta1.IPAllocTypeAuto
+			if podRes.PodInfo.EipInfo.PodEipID != "" {
+				allocType = v1beta1.IPAllocTypeStatic
+			}
+			releaseStrategy := v1beta1.ReleaseStrategyFollow
+			releaseAfter := ""
+
+			if podRes.PodInfo.IPStickTime > 0 {
+				releaseStrategy = v1beta1.ReleaseStrategyTTL
+				releaseAfter = podRes.PodInfo.IPStickTime.String()
+			}
+
+			var err error
+			once.Do(func() {
+				err = crds.RegisterCRD([]string{crds.CRDPodEIP})
+			})
+			if err != nil {
+				return err
+			}
+
+			ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
+
+			l := serviceLog.WithField("name", fmt.Sprintf("%s/%s", podRes.PodInfo.Namespace, podRes.PodInfo.Name))
+
+			c := n.k8s.GetClient()
+			podEIP := &v1beta1.PodEIP{}
+			err = c.Get(ctx, k8stypes.NamespacedName{Namespace: podRes.PodInfo.Namespace, Name: podRes.PodInfo.Name}, podEIP)
+			if err == nil {
+				cancel()
+				l.Info("skip create podEIP, already exist")
+				continue
+			}
+			if !k8sErr.IsNotFound(err) {
+				cancel()
+				return err
+			}
+
+			err = retry.OnError(wait.Backoff{
+				Steps:    4,
+				Duration: 200 * time.Millisecond,
+				Factor:   5.0,
+				Jitter:   0.1,
+			}, func(err error) bool {
+				if k8sErr.IsTooManyRequests(err) {
+					return true
+				}
+				if k8sErr.IsInternalError(err) {
+					return true
+				}
+				return false
+			}, func() error {
+				podEIP = &v1beta1.PodEIP{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:        podRes.PodInfo.Name,
+						Namespace:   podRes.PodInfo.Namespace,
+						Annotations: map[string]string{},
+						Finalizers:  []string{"podeip-controller.alibabacloud.com/finalizer"},
+					},
+					Spec: v1beta1.PodEIPSpec{
+						AllocationID:       eipRes.ID,
+						BandwidthPackageID: podRes.PodInfo.EipInfo.PodEipBandwidthPackageID,
+						AllocationType: v1beta1.AllocationType{
+							Type:            allocType,
+							ReleaseStrategy: releaseStrategy,
+							ReleaseAfter:    releaseAfter,
+						},
+					},
+				}
+
+				l.Infof("create podEIP for %v", podRes)
+
+				err := c.Create(ctx, podEIP)
+				if k8sErr.IsAlreadyExists(err) {
+					return nil
+				}
+				return err
+			})
+			cancel()
+
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 // tracing
 func (n *networkService) Config() []tracing.MapKeyValueEntry {
 	// name, daemon_mode, configFilePath, kubeconfig, master
@@ -1344,6 +1452,10 @@ func newNetworkService(ctx context.Context, configFilePath, daemonMode string) (
 		}
 	}
 
+	if limit.TrunkPod() <= 0 {
+		config.EnableENITrunking = false
+	}
+
 	ecs := aliyun.NewAliyunImpl(aliyunClient, config.EnableENITrunking && !config.WaitTrunkENI, ipFamily, config.ENITagFilter)
 
 	netSrv.resourceDB, err = storage.NewDiskStorage(
@@ -1476,6 +1588,14 @@ func newNetworkService(ctx context.Context, configFilePath, daemonMode string) (
 		}
 	}
 
+	if config.EnableEIPMigrate {
+		err = netSrv.migrateEIP(ctx, resObjList)
+		if err != nil {
+			return nil, err
+		}
+		serviceLog.Infof("eip migrate finished")
+	}
+
 	resStr, err := json.Marshal(localResource)
 	if err != nil {
 		return nil, err
@@ -1511,25 +1631,26 @@ func newNetworkService(ctx context.Context, configFilePath, daemonMode string) (
 		if err != nil {
 			return nil, errors.Wrapf(err, "error init ENI ip resource manager")
 		}
-		if config.EnableEIPPool == conditionTrue {
-			netSrv.eipResMgr = newEipResourceManager(ecs, netSrv.k8s, config.AllowEIPRob == conditionTrue)
-		}
 		netSrv.mgrForResource = map[string]ResourceManager{
 			types.ResourceTypeENIIP: netSrv.eniIPResMgr,
-			types.ResourceTypeEIP:   netSrv.eipResMgr,
 		}
+		if config.EnableEIPPool == conditionTrue && !config.EnableEIPMigrate {
+			netSrv.eipResMgr = newEipResourceManager(ecs, netSrv.k8s, config.AllowEIPRob == conditionTrue)
+			netSrv.mgrForResource[types.ResourceTypeEIP] = netSrv.eipResMgr
+		}
+
 	case daemonModeENIOnly:
 		//init eni
 		netSrv.eniResMgr, err = newENIResourceManager(poolConfig, ecs, localResource[types.ResourceTypeENI], ipFamily, netSrv.k8s, netSrv.ipamType)
 		if err != nil {
 			return nil, errors.Wrapf(err, "error init eni resource manager")
 		}
-		if config.EnableEIPPool == conditionTrue && !config.EnableENITrunking {
-			netSrv.eipResMgr = newEipResourceManager(ecs, netSrv.k8s, config.AllowEIPRob == conditionTrue)
-		}
 		netSrv.mgrForResource = map[string]ResourceManager{
-			types.ResourceTypeENI: netSrv.eniResMgr,
-			types.ResourceTypeEIP: netSrv.eipResMgr,
+			types.ResourceTypeENIIP: netSrv.eniIPResMgr,
+		}
+		if config.EnableEIPPool == conditionTrue && !config.EnableENITrunking && !config.EnableEIPMigrate {
+			netSrv.eipResMgr = newEipResourceManager(ecs, netSrv.k8s, config.AllowEIPRob == conditionTrue)
+			netSrv.mgrForResource[types.ResourceTypeEIP] = netSrv.eipResMgr
 		}
 	default:
 		panic("unsupported daemon mode" + daemonMode)