Feat/ipam2

cmd/terway-controlplane/terway-controlplane.go

@@ -34,6 +34,7 @@
 	oteltrace "go.opentelemetry.io/otel/trace"
 	corev1 "k8s.io/api/core/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	k8sErr "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -56,13 +57,19 @@
 	"github.com/AliyunContainerService/terway/pkg/cert"
 	register "github.com/AliyunContainerService/terway/pkg/controller"
 	_ "github.com/AliyunContainerService/terway/pkg/controller/all"
+	multiipnode "github.com/AliyunContainerService/terway/pkg/controller/multi-ip/node"
+	multiippod "github.com/AliyunContainerService/terway/pkg/controller/multi-ip/pod"
+	"github.com/AliyunContainerService/terway/pkg/controller/node"
+	"github.com/AliyunContainerService/terway/pkg/controller/preheating"
 	"github.com/AliyunContainerService/terway/pkg/controller/webhook"
 	"github.com/AliyunContainerService/terway/pkg/metric"
 	"github.com/AliyunContainerService/terway/pkg/utils"
 	"github.com/AliyunContainerService/terway/pkg/utils/k8sclient"
 	"github.com/AliyunContainerService/terway/pkg/version"
 	"github.com/AliyunContainerService/terway/pkg/vswitch"
+	"github.com/AliyunContainerService/terway/types"
 	"github.com/AliyunContainerService/terway/types/controlplane"
+	"github.com/AliyunContainerService/terway/types/daemon"
 )
 
 var (
@@ -125,6 +132,12 @@
 		}
 	})
 
+	err = detectMultiIP(ctx, directClient, cfg)
+	if err != nil {
+		log.Error(err, "unable to detect multi IP")
+		os.Exit(1)
+	}
+
 	ws := wh.NewServer(wh.Options{
 		Port:    cfg.WebhookPort,
 		CertDir: cfg.CertDir,
@@ -255,6 +268,13 @@
 		}
 	}
 
+	if err = (&preheating.DummyReconcile{
+		RegisterResource: ctrlCtx.RegisterResource,
+	}).SetupWithManager(mgr); err != nil {
+		log.Error(err, "unable to create controller", "controller", "preheating")
+		os.Exit(1)
+	}
+
 	log.Info("controller started")
 	err = mgr.Start(ctx)
 	if err != nil {
@@ -300,3 +320,40 @@
 
 	return traceProvider, nil
 }
+
+func detectMultiIP(ctx context.Context, directClient client.Client, cfg *controlplane.Config) error {
+	if !lo.Contains(cfg.Controllers, multiipnode.ControllerName) {
+		return nil
+	}
+
+	var daemonConfig *daemon.Config
+	var innerErr error
+	err := wait.PollUntilContextTimeout(ctx, 1*time.Second, 10*time.Second, true, func(ctx context.Context) (done bool, err error) {
+		daemonConfig, innerErr = daemon.ConfigFromConfigMap(ctx, directClient, "")
+		if innerErr != nil {
+			if k8sErr.IsNotFound(innerErr) {
+				return false, nil
+			}
+			log.Error(innerErr, "failed to get ConfigMap eni-config")
+			return false, nil
+		}
+		return true, nil
+	})
+	if err != nil {
+		return fmt.Errorf("error waiting for daemon to be configured: %w, innerErr %s", err, innerErr)
+	}
+	switch daemonConfig.IPAMType {
+	case types.IPAMTypeCRD:
+		return nil
+	}
+
+	cfg.Controllers = lo.Reject(cfg.Controllers, func(item string, index int) bool {
+		switch item {
+		case node.ControllerName, multiipnode.ControllerName, multiippod.ControllerName:
+			return true
+		}
+		return false
+	})
+	log.Info("daemon is not at crd mode, disable v2 ipam")
+	return nil
+}

cmd/terway-controlplane/terway-controlplane_test.go

@@ -0,0 +1,124 @@
+package main
+
+import (
+	"context"
+	"testing"
+
+	"github.com/samber/lo"
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	"github.com/AliyunContainerService/terway/types/controlplane"
+)
+
+func Test_detectMultiIP(t *testing.T) {
+
+	type args struct {
+		ctx          context.Context
+		directClient client.Client
+		cfg          *controlplane.Config
+	}
+	tests := []struct {
+		name      string
+		args      args
+		wantErr   bool
+		checkFunc func(t *testing.T, cfg *controlplane.Config)
+	}{
+		{
+			name: "test empty config",
+			args: args{
+				ctx: context.Background(),
+				directClient: func() client.Client {
+					nodeReader := fake.NewClientBuilder()
+					return nodeReader.Build()
+				}(),
+				cfg: &controlplane.Config{},
+			},
+			wantErr:   false,
+			checkFunc: func(t *testing.T, cfg *controlplane.Config) {},
+		},
+		{
+			name: "test config not found",
+			args: args{
+				ctx: context.Background(),
+				directClient: func() client.Client {
+					nodeReader := fake.NewClientBuilder()
+					return nodeReader.Build()
+				}(),
+				cfg: &controlplane.Config{
+					Controllers: []string{"multi-ip-node", "node", "multi-ip-pod"},
+				},
+			},
+			wantErr:   true,
+			checkFunc: func(t *testing.T, cfg *controlplane.Config) {},
+		},
+		{
+			name: "test config ipamv1",
+			args: args{
+				ctx: context.Background(),
+				directClient: func() client.Client {
+					nodeReader := fake.NewClientBuilder()
+					nodeReader.WithObjects(
+						&corev1.ConfigMap{
+							ObjectMeta: metav1.ObjectMeta{Name: "eni-config", Namespace: "kube-system"},
+							Data: map[string]string{
+								"eni_conf": "{\"ipam_type\":\"\"}",
+							},
+						},
+					)
+					return nodeReader.Build()
+				}(),
+				cfg: &controlplane.Config{
+					Controllers: []string{"multi-ip-node", "node", "multi-ip-pod", "foo"},
+				},
+			},
+			wantErr: false,
+			checkFunc: func(t *testing.T, cfg *controlplane.Config) {
+				assert.False(t, lo.Contains(cfg.Controllers, "multi-ip-node"))
+				assert.False(t, lo.Contains(cfg.Controllers, "node"))
+				assert.False(t, lo.Contains(cfg.Controllers, "multi-ip-pod"))
+				assert.True(t, lo.Contains(cfg.Controllers, "foo"))
+			},
+		},
+		{
+			name: "test config ipamv2",
+			args: args{
+				ctx: context.Background(),
+				directClient: func() client.Client {
+					nodeReader := fake.NewClientBuilder()
+					nodeReader.WithObjects(
+						&corev1.ConfigMap{
+							ObjectMeta: metav1.ObjectMeta{Name: "eni-config", Namespace: "kube-system"},
+							Data: map[string]string{
+								"eni_conf": "{\"ipam_type\":\"crd\"}",
+							},
+						},
+					)
+					return nodeReader.Build()
+				}(),
+				cfg: &controlplane.Config{
+					Controllers: []string{"multi-ip-node", "node", "multi-ip-pod", "foo"},
+				},
+			},
+			wantErr: false,
+			checkFunc: func(t *testing.T, cfg *controlplane.Config) {
+				assert.True(t, lo.Contains(cfg.Controllers, "multi-ip-node"))
+				assert.True(t, lo.Contains(cfg.Controllers, "node"))
+				assert.True(t, lo.Contains(cfg.Controllers, "multi-ip-pod"))
+				assert.True(t, lo.Contains(cfg.Controllers, "foo"))
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := detectMultiIP(tt.args.ctx, tt.args.directClient, tt.args.cfg); (err != nil) != tt.wantErr {
+				t.Errorf("detectMultiIP() error = %v, wantErr %v", err, tt.wantErr)
+			} else {
+				tt.checkFunc(t, tt.args.cfg)
+			}
+		})
+	}
+}

pkg/aliyun/credential/aliyun_client_mgr.go

@@ -181,7 +181,7 @@ func (c *ClientMgr) refreshToken() (bool, error) {
 		if err != nil {
 			return false, err
 		}
-		c.ecs.SetEndpointRules(c.ecs.EndpointMap, "regional", "public")
+		c.ecs.SetEndpointRules(c.ecs.EndpointMap, "regional", "vpc")
 
 		if c.ecsDomainOverride != "" {
 			c.ecs.Domain = c.ecsDomainOverride
@@ -191,7 +191,7 @@ func (c *ClientMgr) refreshToken() (bool, error) {
 		if err != nil {
 			return false, err
 		}
-		c.vpc.SetEndpointRules(c.vpc.EndpointMap, "regional", "public")
+		c.vpc.SetEndpointRules(c.vpc.EndpointMap, "regional", "vpc")
 
 		if c.vpcDomainOverride != "" {
 			c.vpc.Domain = c.vpcDomainOverride
@@ -201,7 +201,7 @@ func (c *ClientMgr) refreshToken() (bool, error) {
 		if err != nil {
 			return false, err
 		}
-		c.eflo.SetEndpointRules(c.eflo.EndpointMap, "regional", "public")
+		c.eflo.SetEndpointRules(c.eflo.EndpointMap, "regional", "vpc")
 
 		if c.efloDomainOverride != "" {
 			c.eflo.Domain = c.efloDomainOverride

pkg/aliyun/credential/aliyun_client_mgr_test.go

@@ -0,0 +1,20 @@
+package credential
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestClientMgr_refreshToken(t *testing.T) {
+	mgr, err := NewClientMgr("foo", &AKPairProvider{
+		accessKeyID:     "foo",
+		accessKeySecret: "bar",
+	})
+	assert.NoError(t, err)
+	ok, err := mgr.refreshToken()
+	assert.True(t, ok)
+	assert.NoError(t, err)
+
+	assert.Equalf(t, "vpc", mgr.ecs.Network, "default endpoint should be vpc")
+}

pkg/controller/endpoint/endpoint.go

@@ -20,10 +20,10 @@ import (
 
 var log = ctrl.Log.WithName("endpoint")
 
-const controllerName = "endpoint"
+const ControllerName = "endpoint"
 
 func init() {
-	register.Add(controllerName, func(mgr manager.Manager, ctrlCtx *register.ControllerCtx) error {
+	register.Add(ControllerName, func(mgr manager.Manager, ctrlCtx *register.ControllerCtx) error {
 		ipStr := os.Getenv("MY_POD_IP")
 		if ipStr == "" {
 			return fmt.Errorf("podIP is not found")

pkg/controller/multi-ip/node/pool.go

@@ -47,6 +47,8 @@
 )
 
 const (
+	ControllerName = "multi-ip-node"
+
 	finalizer = "network.alibabacloud.com/node-controller"
 
 	batchSize = 10
@@ -55,8 +57,7 @@
 var EventCh = make(chan event.GenericEvent, 1)
 
 func init() {
-	controllerName := "multi-ip-node"
-	register.Add(controllerName, func(mgr manager.Manager, ctrlCtx *register.ControllerCtx) error {
+	register.Add(ControllerName, func(mgr manager.Manager, ctrlCtx *register.ControllerCtx) error {
 		fullSyncPeriod, err := time.ParseDuration(ctrlCtx.Config.MultiIPNodeSyncPeriod)
 		if err != nil {
 			return err
@@ -77,17 +78,17 @@
 			return err
 		}
 
-		// metirc and tracer
+		// metric and tracer
 
 		metrics.Registry.MustRegister(ResourcePoolTotal)
-		tracer := ctrlCtx.TracerProvider.Tracer(controllerName)
+		tracer := ctrlCtx.TracerProvider.Tracer(ControllerName)
 
-		ctrl, err := controller.New(controllerName, mgr, controller.Options{
-			MaxConcurrentReconciles: ctrlCtx.Config.MultiIPMaxConcurrent,
+		ctrl, err := controller.New(ControllerName, mgr, controller.Options{
+			MaxConcurrentReconciles: ctrlCtx.Config.MultiIPNodeMaxConcurrent,
 			Reconciler: &ReconcileNode{
 				client:             mgr.GetClient(),
 				scheme:             mgr.GetScheme(),
-				record:             mgr.GetEventRecorderFor(controllerName),
+				record:             mgr.GetEventRecorderFor(ControllerName),
 				aliyun:             ctrlCtx.AliyunClient,
 				vswpool:            ctrlCtx.VSwitchPool,
 				fullSyncNodePeriod: fullSyncPeriod,
@@ -1199,26 +1200,3 @@
 		in[ip.IP] = ip
 	}
 }
-
-//
-//func (n *ReconcileNode) loadToMetadata(ctx context.Context, name string) {
-//	prev, ok := n.cache.Load(name)
-//	if !ok {
-//		return
-//	}
-//	status := prev.(*NodeStatus)
-//
-//	ctxMeta := MetaCtx(ctx)
-//	ctxMeta.needSyncOpenAPI.Store(status.NeedSyncOpenAPI)
-//	ctxMeta.lastGCTime = status.LastGCTime
-//	ctxMeta.lastReconcileTime = status.LastReconcileTime
-//}
-//
-//func (n *ReconcileNode) fromMetadata(ctx context.Context, name string) {
-//	ctxMeta := MetaCtx(ctx)
-//
-//	n.cache.Store(name, &NodeStatus{
-//		NeedSyncOpenAPI: ctxMeta.needSyncOpenAPI.Load(),
-//		LastGCTime:      ctxMeta.lastGCTime,
-//	})
-//}

pkg/controller/multi-ip/pod/pod.go

@@ -21,19 +21,20 @@ import (
 	"github.com/AliyunContainerService/terway/pkg/controller/multi-ip/node"
 )
 
-const controllerName = "multi-ip-pod"
+const ControllerName = "multi-ip-pod"
 
 func init() {
-	register.Add(controllerName, func(mgr manager.Manager, ctrlCtx *register.ControllerCtx) error {
+	register.Add(ControllerName, func(mgr manager.Manager, ctrlCtx *register.ControllerCtx) error {
+		ctrlCtx.RegisterResource = append(ctrlCtx.RegisterResource, &corev1.Pod{})
 		return ctrl.NewControllerManagedBy(mgr).
 			WithOptions(controller.Options{
-				MaxConcurrentReconciles: 10,
+				MaxConcurrentReconciles: ctrlCtx.Config.MultiIPPodMaxConcurrent,
 			}).
 			For(&corev1.Pod{}, builder.WithPredicates(&predicateForPodEvent{})).
 			Complete(&ReconcilePod{
 				client: mgr.GetClient(),
 				scheme: mgr.GetScheme(),
-				record: mgr.GetEventRecorderFor(controllerName),
+				record: mgr.GetEventRecorderFor(ControllerName),
 			})
 	}, false)
 }