parse podNetworking to podNetworks

Author: l1b0k

cmd/terway-controlplane/terway-controlplane.go

@@ -21,10 +21,11 @@ import (
 	"math/rand"
 	"time"
 
-	"github.com/AliyunContainerService/terway/pkg/aliyun/client"
+	aliyun "github.com/AliyunContainerService/terway/pkg/aliyun/client"
 	"github.com/AliyunContainerService/terway/pkg/aliyun/credential"
 	"github.com/AliyunContainerService/terway/pkg/apis/crds"
 	networkv1beta1 "github.com/AliyunContainerService/terway/pkg/apis/network.alibabacloud.com/v1beta1"
+	"github.com/AliyunContainerService/terway/pkg/backoff"
 	"github.com/AliyunContainerService/terway/pkg/cert"
 	register "github.com/AliyunContainerService/terway/pkg/controller"
 	_ "github.com/AliyunContainerService/terway/pkg/controller/all"
@@ -35,7 +36,6 @@ import (
 	"github.com/AliyunContainerService/terway/pkg/utils"
 	"github.com/AliyunContainerService/terway/pkg/version"
 	"github.com/AliyunContainerService/terway/types/controlplane"
-
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
@@ -60,14 +60,25 @@ func init() {
 
 func main() {
 	rand.Seed(time.Now().UnixNano())
+	var (
+		configFilePath     string
+		credentialFilePath string
+	)
+	flag.StringVar(&configFilePath, "config", "/etc/config/ctrl-config.yaml", "config file for controlplane")
+	flag.StringVar(&credentialFilePath, "credential", "/etc/credential/ctrl-secret.yaml", "secret file for controlplane")
+
 	flag.Parse()
 
 	ctrl.SetLogger(klogr.New())
 	log.Info(version.Version)
 
-	cfg := controlplane.GetConfig()
-	log.Info("using config", "config", cfg)
+	ctx := ctrl.SetupSignalHandler()
 
+	cfg, err := controlplane.ParseAndValidate(configFilePath, credentialFilePath)
+	if err != nil {
+		panic(err)
+	}
+	backoff.OverrideBackoff(cfg.BackoffOverride)
 	utils.SetStsKinds(cfg.CustomStatefulWorkloadKinds)
 
 	restConfig := ctrl.GetConfigOrDie()
@@ -76,7 +87,9 @@ func main() {
 	restConfig.UserAgent = version.UA
 	utils.RegisterClients(restConfig)
 
-	err := crds.RegisterCRDs()
+	log.Info("using config", "config", cfg)
+
+	err = crds.RegisterCRDs()
 	if err != nil {
 		panic(err)
 	}
@@ -91,13 +104,11 @@ func main() {
 		panic(err)
 	}
 
-	aliyunClient, err := client.New(clientSet, flowcontrol.NewTokenBucketRateLimiter(cfg.ReadOnlyQPS, cfg.ReadOnlyBurst), flowcontrol.NewTokenBucketRateLimiter(cfg.MutatingQPS, cfg.MutatingBurst))
+	aliyunClient, err := aliyun.New(clientSet, flowcontrol.NewTokenBucketRateLimiter(cfg.ReadOnlyQPS, cfg.ReadOnlyBurst), flowcontrol.NewTokenBucketRateLimiter(cfg.MutatingQPS, cfg.MutatingBurst))
 	if err != nil {
 		panic(err)
 	}
 
-	ctx := ctrl.SetupSignalHandler()
-
 	mgr, err := ctrl.NewManager(restConfig, ctrl.Options{
 		Scheme:                     scheme,
 		HealthProbeBindAddress:     cfg.HealthzBindAddress,

daemon/daemon.go

@@ -29,6 +29,7 @@ import (
 	"github.com/AliyunContainerService/terway/pkg/utils"
 	"github.com/AliyunContainerService/terway/rpc"
 	"github.com/AliyunContainerService/terway/types"
+	"github.com/AliyunContainerService/terway/types/daemon"
 
 	"github.com/containernetworking/cni/libcni"
 	containertypes "github.com/containernetworking/cni/pkg/types"
@@ -1347,7 +1348,7 @@ func newNetworkService(configFilePath, kubeconfig, master, daemonMode string) (r
 
 	var err error
 
-	globalConfig, err := types.GetConfigFromFileWithMerge(configFilePath, nil)
+	globalConfig, err := daemon.GetConfigFromFileWithMerge(configFilePath, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -1364,7 +1365,7 @@ func newNetworkService(configFilePath, kubeconfig, master, daemonMode string) (r
 		dynamicCfg = ""
 	}
 
-	config, err := types.GetConfigFromFileWithMerge(configFilePath, []byte(dynamicCfg))
+	config, err := daemon.GetConfigFromFileWithMerge(configFilePath, []byte(dynamicCfg))
 	if err != nil {
 		return nil, fmt.Errorf("failed parse config: %v", err)
 	}
@@ -1594,7 +1595,7 @@ func restoreLocalENIRes(ecs ipam.API, k8s Kubernetes, resourceDB storage.Storage
 }
 
 // setup default value
-func setDefault(cfg *types.Configure) error {
+func setDefault(cfg *daemon.Config) error {
 	if cfg.EniCapRatio == 0 {
 		cfg.EniCapRatio = 1
 	}
@@ -1611,7 +1612,7 @@ func setDefault(cfg *types.Configure) error {
 	return nil
 }
 
-func validateConfig(cfg *types.Configure) error {
+func validateConfig(cfg *daemon.Config) error {
 	switch cfg.IPStack {
 	case "", string(types.IPStackIPv4), string(types.IPStackDual):
 	default:
@@ -1621,7 +1622,7 @@ func validateConfig(cfg *types.Configure) error {
 	return nil
 }
 
-func getPoolConfig(cfg *types.Configure, ipamType types.IPAMType) (*types.PoolConfig, error) {
+func getPoolConfig(cfg *daemon.Config, ipamType types.IPAMType) (*types.PoolConfig, error) {
 	poolConfig := &types.PoolConfig{
 		MaxPoolSize:               cfg.MaxPoolSize,
 		MinPoolSize:               cfg.MinPoolSize,

daemon/k8s.go

@@ -21,6 +21,7 @@ import (
 	"github.com/AliyunContainerService/terway/pkg/utils"
 	"github.com/AliyunContainerService/terway/pkg/version"
 	"github.com/AliyunContainerService/terway/types"
+	"github.com/AliyunContainerService/terway/types/daemon"
 
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
@@ -269,7 +270,7 @@ func (k *k8s) WaitTrunkReady() (string, error) {
 }
 
 // newK8S return Kubernetes service by pod spec and daemon mode
-func newK8S(master, kubeconfig string, daemonMode string, globalConfig *types.Configure) (Kubernetes, error) {
+func newK8S(master, kubeconfig string, daemonMode string, globalConfig *daemon.Config) (Kubernetes, error) {
 
 	k8sRestConfig, err := clientcmd.BuildConfigFromFlags(master, kubeconfig)
 	if err != nil {

pkg/controller/common/match.go

@@ -18,70 +18,11 @@ package common
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/AliyunContainerService/terway/pkg/apis/network.alibabacloud.com/v1beta1"
-	"github.com/AliyunContainerService/terway/pkg/utils"
-
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-// MatchOnePodNetworking will range all podNetworking and try to found a matched podNetworking for this pod
-// for stateless pod Fixed ip config is never matched
-func MatchOnePodNetworking(pod *corev1.Pod, ns *corev1.Namespace, networkings []v1beta1.PodNetworking) (*v1beta1.PodNetworking, error) {
-	podLabels := labels.Set(pod.Labels)
-	nsLabels := labels.Set(ns.Labels)
-	for _, podNetworking := range networkings {
-		if podNetworking.Status.Status != v1beta1.NetworkingStatusReady {
-			continue
-		}
-		if !utils.IsStsPod(pod) {
-			// for fixed ip , only match sts pod
-			if podNetworking.Spec.AllocationType.Type == v1beta1.IPAllocTypeFixed {
-				continue
-			}
-		}
-
-		matchOne := false
-		if podNetworking.Spec.Selector.PodSelector != nil {
-			ok, err := PodMatchSelector(podNetworking.Spec.Selector.PodSelector, podLabels)
-			if err != nil {
-				return nil, fmt.Errorf("error match pod selector, %w", err)
-			}
-			if !ok {
-				continue
-			}
-			matchOne = true
-		}
-		if podNetworking.Spec.Selector.NamespaceSelector != nil {
-			ok, err := PodMatchSelector(podNetworking.Spec.Selector.NamespaceSelector, nsLabels)
-			if err != nil {
-				return nil, fmt.Errorf("error match namespace selector, %w", err)
-			}
-			if !ok {
-				continue
-			}
-			matchOne = true
-		}
-		if matchOne {
-			return &podNetworking, nil
-		}
-	}
-	return nil, nil
-}
-
-// PodMatchSelector pod is selected by selector
-func PodMatchSelector(labelSelector *metav1.LabelSelector, l labels.Set) (bool, error) {
-	selector, err := metav1.LabelSelectorAsSelector(labelSelector)
-	if err != nil {
-		return false, err
-	}
-	return selector.Matches(l), nil
-}
-
 // UpdatePodENI update cr
 func UpdatePodENI(ctx context.Context, c client.Client, update *v1beta1.PodENI) (*v1beta1.PodENI, error) {
 	var err error
@@ -107,16 +48,3 @@ func UpdatePodENIStatus(ctx context.Context, c client.Client, update *v1beta1.Po
 	}
 	return nil, err
 }
-
-// PatchPodENIStatus set cr status
-func PatchPodENIStatus(ctx context.Context, c client.Client, update, old *v1beta1.PodENI) (*v1beta1.PodENI, error) {
-	var err error
-	for i := 0; i < 2; i++ {
-		err = c.Status().Patch(ctx, update, client.MergeFrom(old))
-		if err != nil {
-			continue
-		}
-		return update, nil
-	}
-	return nil, err
-}

pkg/controller/node/node_controller.go

@@ -15,6 +15,7 @@ import (
 	"github.com/AliyunContainerService/terway/pkg/controller/vswitch"
 	"github.com/AliyunContainerService/terway/types"
 	"github.com/AliyunContainerService/terway/types/controlplane"
+	"github.com/AliyunContainerService/terway/types/daemon"
 	"k8s.io/client-go/util/retry"
 
 	"github.com/aliyun/alibaba-cloud-sdk-go/services/ecs"
@@ -209,7 +210,7 @@ func (m *ReconcileNode) ensureTrunkENI(ctx context.Context, node *corev1.Node) (
 		l.Info("no eni found, will create trunk eni")
 
 		// create and attach eni
-		eniConfig, err := common.ConfigFromConfigMConfigFromConfigMap(ctx, m.client, node.Name)
+		eniConfig, err := daemon.ConfigFromConfigMap(ctx, m.client, node.Name)
 		if err != nil {
 			return reconcile.Result{}, err
 		}
@@ -284,7 +285,7 @@ func (m *ReconcileNode) initENIManagerForNode(ctx context.Context, node *corev1.
 		return err
 	}
 
-	eniConfig, err := common.ConfigFromConfigMConfigFromConfigMap(ctx, m.client, node.Name)
+	eniConfig, err := daemon.ConfigFromConfigMap(ctx, m.client, node.Name)
 	if err != nil {
 		return err
 	}

pkg/controller/pod-eni/eni_controller.go

@@ -595,7 +595,8 @@ func (m *ReconcilePodENI) gcCRPodENIs(ctx context.Context) {
 				ll.V(5).Info("update pod lastSeen to now")
 				update := podENI.DeepCopy()
 				update.Status.PodLastSeen = metav1.Now()
-				_, err = common.PatchPodENIStatus(ctx, m.client, update, &podENI)
+
+				err = m.client.Status().Patch(ctx, update, client.MergeFrom(&podENI))
 				if err != nil {
 					ll.Error(err, "error update timestamp")
 				}

pkg/controller/pod-networking/networking_test.go

@@ -1,5 +1,4 @@
 //go:build test_env
-// +build test_env
 
 package podnetworking
 

pkg/controller/pod-networking/suite_test.go

@@ -1,5 +1,4 @@
 //go:build test_env
-// +build test_env
 
 package podnetworking
 
@@ -12,6 +11,7 @@ import (
 	networkv1beta1 "github.com/AliyunContainerService/terway/pkg/apis/network.alibabacloud.com/v1beta1"
 	register "github.com/AliyunContainerService/terway/pkg/controller"
 	"github.com/AliyunContainerService/terway/pkg/controller/vswitch"
+	"github.com/AliyunContainerService/terway/types/controlplane"
 
 	"github.com/aliyun/alibaba-cloud-sdk-go/services/vpc"
 	. "github.com/onsi/ginkgo"
@@ -74,7 +74,12 @@ var _ = BeforeSuite(func() {
 			},
 		},
 	}
-	err = register.Controllers[controllerName](k8sManager, fakeClient, vsw)
+	err = register.Controllers[controllerName].Creator(k8sManager, &register.ControllerCtx{
+		Config:         &controlplane.Config{},
+		VSwitchPool:    vsw,
+		AliyunClient:   fakeClient,
+		DelegateClient: fakeClient,
+	})
 	Expect(err).ToNot(HaveOccurred())
 
 	go func() {

pkg/controller/pod/pod_controller.go

@@ -49,6 +49,7 @@ import (
 )
 
 const controllerName = "pod"
+const defaultInterface = "eth0"
 
 func init() {
 	register.Add(controllerName, func(mgr manager.Manager, ctrlCtx *register.ControllerCtx) error {
@@ -471,7 +472,7 @@ func (m *ReconcilePod) reConfig(ctx context.Context, pod *corev1.Pod, prePodENI
 	for i, n := range anno.PodNetworks {
 		name := n.Interface
 		if name == "" {
-			name = "eth0"
+			name = defaultInterface
 		}
 		targets[name] = i
 	}
@@ -481,7 +482,7 @@ func (m *ReconcilePod) reConfig(ctx context.Context, pod *corev1.Pod, prePodENI
 		alloc := update.Spec.Allocations[i]
 		name := alloc.Interface
 		if name == "" {
-			name = "eth0"
+			name = defaultInterface
 		}
 
 		if _, ok := targets[name]; ok {

pkg/controller/pod/pod_controller_default.go

@@ -4,14 +4,66 @@ package pod
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/AliyunContainerService/terway/pkg/apis/network.alibabacloud.com/v1beta1"
+	"github.com/AliyunContainerService/terway/pkg/controller/common"
+	"github.com/AliyunContainerService/terway/pkg/controller/vswitch"
 	"github.com/AliyunContainerService/terway/types/controlplane"
 )
 
 // ParsePodNetworksFromAnnotation parse alloc
 func (m *ReconcilePod) ParsePodNetworksFromAnnotation(ctx context.Context, zoneID string, anno *controlplane.PodNetworksAnnotation) ([]*v1beta1.Allocation, error) {
-	return nil, nil
+	if zoneID == "" {
+		return nil, fmt.Errorf("zoneID is empty")
+	}
+	var allocs []*v1beta1.Allocation
+
+	for _, c := range anno.PodNetworks {
+		ifName := c.Interface
+		if ifName == "" {
+			ifName = defaultInterface
+		}
+
+		if len(c.VSwitchOptions) == 0 || len(c.SecurityGroupIDs) == 0 {
+			return nil, fmt.Errorf("vSwitchOptions or securityGroupIDs is missing")
+		}
+
+		alloc := &v1beta1.Allocation{
+			ENI: v1beta1.ENI{
+				SecurityGroupIDs: c.SecurityGroupIDs,
+			},
+			Interface:   ifName,
+			ExtraConfig: map[string]string{},
+		}
+
+		ctx := common.WithCtx(ctx, alloc)
+
+		// allow config route without
+		realClient, _, err := common.Became(ctx, m.aliyun)
+		if err != nil {
+			return nil, err
+		}
+
+		sw, err := m.swPool.GetOne(ctx, realClient, zoneID, c.VSwitchOptions, &vswitch.SelectOptions{
+			IgnoreZone: false,
+		})
+		if err != nil {
+			return nil, err
+		}
+		alloc.ENI.VSwitchID = sw.ID
+		alloc.IPv4CIDR = sw.IPv4CIDR
+		alloc.IPv6CIDR = sw.IPv6CIDR
+		var routes []v1beta1.Route
+		for _, r := range c.ExtraRoutes {
+			routes = append(routes, v1beta1.Route{Dst: r.Dst})
+		}
+		alloc.ExtraRoutes = routes
+
+		allocs = append(allocs, alloc)
+	}
+
+	return allocs, nil
 }
 
 func (m *ReconcilePod) PostENICreate(ctx context.Context, alloc *v1beta1.Allocation) error {