Merge pull request #457 from jwsui/shared-vpc

Support shared vpc

pkg/controllers/namespace/namespace_controller.go

@@ -11,7 +11,9 @@ import (
 	"strings"
 
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	apimachineryruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/util/retry"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -82,6 +84,54 @@ func (r *NamespaceReconciler) createVPCCR(ctx *context.Context, obj client.Objec
 	return vpcCR, nil
 }
 
+func (r *NamespaceReconciler) createDefaultSubnetSet(ns string) error {
+	defaultSubnetSets := map[string]string{
+		types.DefaultVMSubnetSet:  types.LabelDefaultVMSubnetSet,
+		types.DefaultPodSubnetSet: types.LabelDefaultPodSubnetSet,
+	}
+	for name, subnetSetType := range defaultSubnetSets {
+		if err := retry.OnError(retry.DefaultRetry, func(err error) bool {
+			return err != nil
+		}, func() error {
+			list := &v1alpha1.SubnetSetList{}
+			label := client.MatchingLabels{
+				types.LabelDefaultSubnetSet: subnetSetType,
+			}
+			if err := r.Client.List(context.Background(), list, label, client.InNamespace(ns)); err != nil {
+				return err
+			}
+			if len(list.Items) > 0 {
+				log.Info("default SubnetSet already exists", types.LabelDefaultSubnetSet, subnetSetType)
+				return nil
+			}
+			obj := &v1alpha1.SubnetSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: ns,
+					Name:      name,
+					Labels: map[string]string{
+						types.LabelDefaultSubnetSet: subnetSetType,
+					},
+				},
+				Spec: v1alpha1.SubnetSetSpec{
+					AdvancedConfig: v1alpha1.AdvancedConfig{
+						StaticIPAllocation: v1alpha1.StaticIPAllocation{
+							Enable: true,
+						},
+					},
+				},
+			}
+			if err := r.Client.Create(context.Background(), obj); err != nil {
+				return err
+			}
+			return nil
+		}); err != nil {
+			log.Error(err, "failed to create SubnetSet", "Namespace", ns, "Name", name)
+			return err
+		}
+	}
+	return nil
+}
+
 func (r *NamespaceReconciler) namespaceError(ctx *context.Context, k8sObj client.Object, msg string, err error) {
 	logErr := util.If(err == nil, errors.New(msg), err).(error)
 	log.Error(logErr, msg)
@@ -137,6 +187,7 @@ func (r *NamespaceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 	if obj.ObjectMeta.DeletionTimestamp.IsZero() {
 		metrics.CounterInc(r.NSXConfig, metrics.ControllerUpdateTotal, common.MetricResTypeNamespace)
 		log.Info("start processing namespace create/update event", "namespace", ns)
+
 		ctx := context.Background()
 		annotations := obj.GetAnnotations()
 		r.insertNamespaceNetworkconfigBinding(ns, annotations)
@@ -145,7 +196,7 @@ func (r *NamespaceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		// skip the event.
 		ncName, ncExist := annotations[types.AnnotationVPCNetworkConfig]
 		vpcName, nameExist := annotations[types.AnnotationVPCName]
-		var create_vpc_name *string
+		var createVpcName *string
 		if nameExist {
 			log.Info("read ns annotation vpcName", "VPCNAME", vpcName)
 			res := strings.Split(vpcName, "/")
@@ -162,7 +213,7 @@ func (r *NamespaceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 				log.Info("name space is using shared vpc, with vpc name anno", "VPCNAME", vpcName, "Namespace", ns)
 				return common.ResultNormal, nil
 			}
-			create_vpc_name = &res[1]
+			createVpcName = &res[1]
 			log.Info("creating vpc using customer defined vpc name", "VPCName", res[1])
 		}
 
@@ -172,7 +223,10 @@ func (r *NamespaceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 			ncName = types.DefaultNetworkConfigName
 		}
 
-		if _, err := r.createVPCCR(&ctx, obj, ns, ncName, create_vpc_name); err != nil {
+		if _, err := r.createVPCCR(&ctx, obj, ns, ncName, createVpcName); err != nil {
+			return common.ResultRequeueAfter10sec, nil
+		}
+		if err := r.createDefaultSubnetSet(ns); err != nil {
 			return common.ResultRequeueAfter10sec, nil
 		}
 		return common.ResultNormal, nil

pkg/controllers/subnet/subnet_controller.go

@@ -100,8 +100,11 @@ func (r *SubnetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 		for k, v := range nsObj.Labels {
 			tags = append(tags, model.Tag{Scope: servicecommon.String(k), Tag: servicecommon.String(v)})
 		}
-
-		if _, err := r.Service.CreateOrUpdateSubnet(obj, tags); err != nil {
+		vpcInfo, err := common.ServiceMediator.GetNamespaceVPCInfo(req.Namespace)
+		if err != nil {
+			return commonctl.ResultRequeueAfter10sec, nil
+		}
+		if _, err := r.Service.CreateOrUpdateSubnet(obj, *vpcInfo, tags); err != nil {
 			log.Error(err, "operate failed, would retry exponentially", "subnet", req.NamespacedName)
 			updateFail(r, &ctx, obj)
 			return ResultRequeue, err

pkg/controllers/subnetset/subnetset_controller.go

@@ -32,8 +32,6 @@ var (
 	ResultRequeue           = common.ResultRequeue
 	ResultRequeueAfter5mins = common.ResultRequeueAfter5mins
 	MetricResTypeSubnetSet  = common.MetricResTypeSubnetSet
-	//TODO rename this
-	defaultSubnet = "default-subnet"
 )
 
 // SubnetSetReconciler reconciles a SubnetSet object
@@ -211,9 +209,6 @@ func (r *SubnetSetReconciler) setupWithManager(mgr ctrl.Manager) error {
 		WithOptions(controller.Options{
 			MaxConcurrentReconciles: runtime.NumCPU(),
 		}).
-		Watches(&v1alpha1.VPC{},
-			&VPCHandler{Client: mgr.GetClient()},
-			builder.WithPredicates(VPCPredicate)).
 		Watches(
 			&v1.Namespace{},
 			&EnqueueRequestForNamespace{Client: mgr.GetClient()},

pkg/nsx/services/mediator/mediator.go

@@ -82,7 +82,24 @@ func (serviceMediator *ServiceMediator) GetAvailableSubnet(subnetSet *v1alpha1.S
 		tags = append(tags, model.Tag{Scope: common.String(k), Tag: common.String(v)})
 	}
 	log.Info("the existing subnets are not available, creating new subnet", "subnetList", subnetList, "subnetSet.Name", subnetSet.Name, "subnetSet.Namespace", subnetSet.Namespace)
-	return serviceMediator.CreateOrUpdateSubnet(subnetSet, tags)
+	vpcInfo, err := serviceMediator.GetNamespaceVPCInfo(subnetSet.Namespace)
+	if err != nil {
+		return "", err
+	}
+	return serviceMediator.CreateOrUpdateSubnet(subnetSet, *vpcInfo, tags)
+}
+
+func (serviceMediator *ServiceMediator) GetNamespaceVPCInfo(ns string) (*common.VPCResourceInfo, error) {
+	vpcList := serviceMediator.GetVPCsByNamespace(ns)
+	if len(vpcList) == 0 {
+		return nil, fmt.Errorf("no vpc found for ns %s", ns)
+	}
+	vpcInfo, err := common.ParseVPCResourcePath(*vpcList[0].Path)
+	if err != nil {
+		err := fmt.Errorf("failed to parse NSX VPC path for VPC %s: %s", *vpcList[0].Id, err)
+		return nil, err
+	}
+	return &vpcInfo, nil
 }
 
 func (serviceMediator *ServiceMediator) GetPortsOfSubnet(nsxSubnetID string) (ports []model.VpcSubnetPort) {

pkg/nsx/services/subnet/subnet.go

@@ -97,23 +97,7 @@ func InitializeSubnetService(service common.Service) (*SubnetService, error) {
 	return subnetService, nil
 }
 
-func (service *SubnetService) CreateOrUpdateSubnet(obj client.Object, tags []model.Tag) (string, error) {
-	vpcList := &v1alpha1.VPCList{}
-	if err := service.Client.List(context.Background(), vpcList, client.InNamespace(obj.GetNamespace())); err != nil {
-		log.Error(err, "fail to list VPC", "ns", obj.GetNamespace())
-		return "", err
-	}
-	if len(vpcList.Items) == 0 {
-		err := errors.New("no VPC found")
-		log.Error(err, "", "ns", obj.GetNamespace())
-		return "", err
-	}
-	vpc := vpcList.Items[0]
-	vpcInfo, err := common.ParseVPCResourcePath(vpc.Status.NSXResourcePath)
-	if err != nil {
-		err := fmt.Errorf("failed to parse NSX VPC path for VPC %s: %s", vpc.UID, err)
-		return "", err
-	}
+func (service *SubnetService) CreateOrUpdateSubnet(obj client.Object, vpcInfo common.VPCResourceInfo, tags []model.Tag) (string, error) {
 	uid := string(obj.GetUID())
 	nsxSubnet, err := service.buildSubnet(obj, tags)
 	if err != nil {

pkg/nsx/services/vpc/vpc.go

@@ -620,7 +620,7 @@ func (service *VPCService) CreateOrUpdateAVIRule(vpc *model.Vpc, namespace strin
 
 	if allowrule != nil {
 		if !service.needUpdateRule(allowrule, externalCIDRs) {
-			log.Info("avi rule is not changed, skip updating avi rulee")
+			log.Info("avi rule is not changed, skip updating avi rule")
 			return nil
 		} else {
 			log.Info("avi rule changed", "previous", allowrule.DestinationGroups, "current", externalCIDRs)

test/e2e/manifest/testSubnet/shared_ns.yaml

@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Namespace
+metadata:
+  annotations:
+    nsx.vmware.com/vpc_name: target-ns/target-ns-vpc
+  name: target-ns
+
+---
+
+apiVersion: v1
+kind: Namespace
+metadata:
+  annotations:
+    nsx.vmware.com/vpc_name: target-ns/target-ns-vpc
+  name: subnet-e2e-shared

test/e2e/manifest/testSubnet/subnetport_3.yaml

@@ -0,0 +1,6 @@
+apiVersion: nsx.vmware.com/v1alpha1
+kind: SubnetPort
+metadata:
+  name: port-3
+  namespace: subnet-e2e-shared
+spec:

test/e2e/nsx_subnet_test.go

@@ -3,6 +3,7 @@ package e2e
 import (
 	"context"
 	"log"
+	"net"
 	"path/filepath"
 	"testing"
 	"time"
@@ -16,6 +17,8 @@ import (
 const (
 	SubnetSetCRType        = "subnetsets"
 	E2ENamespace           = "subnet-e2e"
+	E2ENamespaceShared     = "subnet-e2e-shared"
+	E2ENamespaceTarget     = "target-ns"
 	VPCNetworkConfigCRName = "default"
 	UserSubnetSet          = "user-pod-subnetset"
 	// SubnetDeletionTimeout requires a bigger value than defaultTimeout, it's because that it takes some time for NSX to
@@ -45,11 +48,23 @@ func verifySubnetSetCR(subnetSet string) bool {
 	return true
 }
 
-func TestDefaultSubnetSet(t *testing.T) {
+func TestSubnetSet(t *testing.T) {
 	setupTest(t, E2ENamespace)
+	nsPath, _ := filepath.Abs("./manifest/testSubnet/shared_ns.yaml")
+	err := applyYAML(nsPath, "")
+	assert_nil(t, err)
+
 	defer teardownTest(t, E2ENamespace, SubnetDeletionTimeout)
+	defer teardownTest(t, E2ENamespaceShared, SubnetDeletionTimeout)
+	defer teardownTest(t, E2ENamespaceTarget, SubnetDeletionTimeout)
+
+	t.Run("case=DefaultSubnetSet", defaultSubnetSet)
+	t.Run("case=UserSubnetSet", userSubnetSet)
+	t.Run("case=SharedSubnetSet", sharedSubnetSet)
+}
 
-	// 1. Check whether vm-default, pod-default are created.
+func defaultSubnetSet(t *testing.T) {
+	// 1. Check whether default-vm-subnetset and default-pod-subnetset are created.
 	err := testData.waitForCRReadyOrDeleted(defaultTimeout, SubnetSetCRType, E2ENamespace, common.DefaultVMSubnetSet, Ready)
 	assert_nil(t, err)
 	err = testData.waitForCRReadyOrDeleted(defaultTimeout, SubnetSetCRType, E2ENamespace, common.DefaultPodSubnetSet, Ready)
@@ -128,10 +143,7 @@ func TestDefaultSubnetSet(t *testing.T) {
 	assert_false(t, found, "Failed to delete tags for NSX subnet %s", vpcInfo.ID)
 }
 
-func TestUserSubnetSet(t *testing.T) {
-	setupTest(t, E2ENamespace)
-	defer teardownTest(t, E2ENamespace, SubnetDeletionTimeout)
-
+func userSubnetSet(t *testing.T) {
 	// 1. Check SubnetSet created by user.
 	subnetSetPath, _ := filepath.Abs("./manifest/testSubnet/subnetset.yaml")
 	err := applyYAML(subnetSetPath, E2ENamespace)
@@ -157,3 +169,37 @@ func TestUserSubnetSet(t *testing.T) {
 	_, err = testData.nsxClient.SubnetsClient.Get(vpcInfo.OrgID, vpcInfo.ProjectID, vpcInfo.VPCID, vpcInfo.ID)
 	assert_nil(t, err, "Failed to get VPC subnet %s", vpcInfo.ID)
 }
+
+func sharedSubnetSet(t *testing.T) {
+	// 1. Check whether default-vm-subnetset and default-pod-subnetset are created.
+	err := testData.waitForCRReadyOrDeleted(defaultTimeout, SubnetSetCRType, E2ENamespaceTarget, common.DefaultVMSubnetSet, Ready)
+	assert_nil(t, err)
+	err = testData.waitForCRReadyOrDeleted(defaultTimeout, SubnetSetCRType, E2ENamespaceTarget, common.DefaultPodSubnetSet, Ready)
+	assert_nil(t, err)
+
+	// 2. Check `Ipv4SubnetSize` and `AccessMode` should be same with related fields in VPCNetworkConfig.
+	assert_true(t, verifySubnetSetCR(common.DefaultVMSubnetSet))
+	assert_true(t, verifySubnetSetCR(common.DefaultPodSubnetSet))
+
+	portPath, _ := filepath.Abs("./manifest/testSubnet/subnetport_3.yaml")
+	err = applyYAML(portPath, E2ENamespaceShared)
+	time.Sleep(30 * time.Second)
+	assert_nil(t, err)
+	defer deleteYAML(portPath, E2ENamespaceShared)
+
+	// 3. Check SubnetSet CR status should be updated with NSX subnet info.
+	subnetSet, err := testData.crdClientset.NsxV1alpha1().SubnetSets(E2ENamespaceTarget).Get(context.TODO(), common.DefaultVMSubnetSet, v1.GetOptions{})
+	assert_nil(t, err)
+	assert.NotEmpty(t, subnetSet.Status.Subnets, "No Subnet info in SubnetSet")
+
+	// 4. Check IP address is allocated to SubnetPort.
+	port, err := testData.crdClientset.NsxV1alpha1().SubnetPorts(E2ENamespaceShared).Get(context.TODO(), "port-3", v1.GetOptions{})
+	assert_nil(t, err)
+	assert.NotEmpty(t, port.Status.IPAddresses, "No IP address in SubnetPort")
+
+	// 5. Check Subnet CIDR contains SubnetPort IP.
+	portIP := net.ParseIP(port.Status.IPAddresses[0].IP)
+	_, subnetCIDR, err := net.ParseCIDR(subnetSet.Status.Subnets[0].IPAddresses[0])
+	assert_nil(t, err)
+	assert_true(t, subnetCIDR.Contains(portIP))
+}