Merge pull request #608 from magicsong/wsnp

🌟 add support of workspace networkpolicy

pkg/controller/network/testing/controller.go

@@ -0,0 +1,38 @@
+package testing
+
+import (
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	kubeinformers "k8s.io/client-go/informers"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
+	"kubesphere.io/kubesphere/pkg/client/clientset/versioned/fake"
+	informers "kubesphere.io/kubesphere/pkg/client/informers/externalversions"
+)
+
+var (
+	AlwaysReady      = func() bool { return true }
+	ResyncPeriodFunc = func() time.Duration { return 1 * time.Second }
+)
+
+type FakeControllerBuilder struct {
+	KsClient    *fake.Clientset
+	KubeClient  *k8sfake.Clientset
+	Kubeobjects []runtime.Object
+	CRDObjects  []runtime.Object
+}
+
+func NewFakeControllerBuilder() *FakeControllerBuilder {
+	return &FakeControllerBuilder{
+		Kubeobjects: make([]runtime.Object, 0),
+		CRDObjects:  make([]runtime.Object, 0),
+	}
+}
+
+func (f *FakeControllerBuilder) NewControllerInformer() (informers.SharedInformerFactory, kubeinformers.SharedInformerFactory) {
+	f.KsClient = fake.NewSimpleClientset(f.CRDObjects...)
+	f.KubeClient = k8sfake.NewSimpleClientset(f.Kubeobjects...)
+	i := informers.NewSharedInformerFactory(f.KsClient, ResyncPeriodFunc())
+	k8sI := kubeinformers.NewSharedInformerFactory(f.KubeClient, ResyncPeriodFunc())
+	return i, k8sI
+}

pkg/controller/network/testing/util.go

@@ -0,0 +1,12 @@
+package testing
+
+import (
+	"strings"
+
+	"k8s.io/apimachinery/pkg/util/yaml"
+)
+
+func StringToObject(data string, obj interface{}) error {
+	reader := strings.NewReader(data)
+	return yaml.NewYAMLOrJSONDecoder(reader, 10).Decode(obj)
+}

pkg/controller/network/wsnetworkpolicy/controller.go

@@ -0,0 +1,283 @@
+package wsnetworkpolicy
+
+import (
+	"fmt"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	k8snetwork "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	corev1informer "k8s.io/client-go/informers/core/v1"
+	k8snetworkinformer "k8s.io/client-go/informers/networking/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	corev1lister "k8s.io/client-go/listers/core/v1"
+	k8snetworklister "k8s.io/client-go/listers/networking/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog"
+	"k8s.io/klog/klogr"
+	workspaceapi "kubesphere.io/kubesphere/pkg/apis/tenant/v1alpha1"
+	kubesphereclient "kubesphere.io/kubesphere/pkg/client/clientset/versioned"
+	kubespherescheme "kubesphere.io/kubesphere/pkg/client/clientset/versioned/scheme"
+	networkinformer "kubesphere.io/kubesphere/pkg/client/informers/externalversions/network/v1alpha1"
+	workspaceinformer "kubesphere.io/kubesphere/pkg/client/informers/externalversions/tenant/v1alpha1"
+	networklister "kubesphere.io/kubesphere/pkg/client/listers/network/v1alpha1"
+	workspacelister "kubesphere.io/kubesphere/pkg/client/listers/tenant/v1alpha1"
+)
+
+const controllerAgentName = "wsnp-controller"
+
+var (
+	log      = klogr.New().WithName("Controller").WithValues(controllerAgentName)
+	errCount = 0
+)
+
+// Controller expose Run method
+type Controller interface {
+	Run(threadiness int, stopCh <-chan struct{}) error
+}
+type controller struct {
+	kubeClientset       kubernetes.Interface
+	kubesphereClientset kubesphereclient.Interface
+
+	wsnpInformer networkinformer.WorkspaceNetworkPolicyInformer
+	wsnpLister   networklister.WorkspaceNetworkPolicyLister
+	wsnpSynced   cache.InformerSynced
+
+	networkPolicyInformer k8snetworkinformer.NetworkPolicyInformer
+	networkPolicyLister   k8snetworklister.NetworkPolicyLister
+	networkPolicySynced   cache.InformerSynced
+
+	namespaceLister   corev1lister.NamespaceLister
+	namespaceInformer corev1informer.NamespaceInformer
+	namespaceSynced   cache.InformerSynced
+
+	workspaceLister   workspacelister.WorkspaceLister
+	workspaceInformer workspaceinformer.WorkspaceInformer
+	workspaceSynced   cache.InformerSynced
+	// workqueue is a rate limited work queue. This is used to queue work to be
+	// processed instead of performing it as soon as a change happens. This
+	// means we can ensure we only process a fixed amount of resources at a
+	// time, and makes it easy to ensure we are never processing the same item
+	// simultaneously in two different workers.
+	workqueue workqueue.RateLimitingInterface
+	// recorder is an event recorder for recording Event resources to the
+	// Kubernetes API.
+	recorder record.EventRecorder
+}
+
+func NewController(kubeclientset kubernetes.Interface,
+	kubesphereclientset kubesphereclient.Interface,
+	wsnpInformer networkinformer.WorkspaceNetworkPolicyInformer,
+	networkPolicyInformer k8snetworkinformer.NetworkPolicyInformer,
+	namespaceInformer corev1informer.NamespaceInformer,
+	workspaceInformer workspaceinformer.WorkspaceInformer) Controller {
+	utilruntime.Must(kubespherescheme.AddToScheme(scheme.Scheme))
+	log.V(4).Info("Creating event broadcaster")
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartLogging(klog.Infof)
+	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeclientset.CoreV1().Events("")})
+	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})
+	ctl := &controller{
+		kubeClientset:         kubeclientset,
+		kubesphereClientset:   kubesphereclientset,
+		wsnpInformer:          wsnpInformer,
+		wsnpLister:            wsnpInformer.Lister(),
+		wsnpSynced:            wsnpInformer.Informer().HasSynced,
+		networkPolicyInformer: networkPolicyInformer,
+		networkPolicyLister:   networkPolicyInformer.Lister(),
+		networkPolicySynced:   networkPolicyInformer.Informer().HasSynced,
+		namespaceInformer:     namespaceInformer,
+		namespaceLister:       namespaceInformer.Lister(),
+		namespaceSynced:       namespaceInformer.Informer().HasSynced,
+		workspaceInformer:     workspaceInformer,
+		workspaceLister:       workspaceInformer.Lister(),
+		workspaceSynced:       workspaceInformer.Informer().HasSynced,
+
+		workqueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "WorkspaceNetworkPolicies"),
+		recorder:  recorder,
+	}
+	log.Info("Setting up event handlers")
+	wsnpInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: ctl.enqueueWSNP,
+		UpdateFunc: func(old, new interface{}) {
+			ctl.enqueueWSNP(new)
+		},
+	})
+	networkPolicyInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: ctl.handleNP,
+		UpdateFunc: func(old, new interface{}) {
+			newNP := new.(*k8snetwork.NetworkPolicy)
+			oldNP := old.(*k8snetwork.NetworkPolicy)
+			if newNP.ResourceVersion == oldNP.ResourceVersion {
+				return
+			}
+			ctl.handleNP(new)
+		},
+		DeleteFunc: ctl.handleNP,
+	})
+	workspaceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: ctl.handleWS,
+		UpdateFunc: func(old, new interface{}) {
+			newNP := new.(*workspaceapi.Workspace)
+			oldNP := old.(*workspaceapi.Workspace)
+			if newNP.ResourceVersion == oldNP.ResourceVersion {
+				return
+			}
+			ctl.handleWS(new)
+		},
+		DeleteFunc: ctl.handleNP,
+	})
+	return ctl
+}
+
+func (c *controller) handleWS(obj interface{}) {
+	ws := obj.(*workspaceapi.Workspace)
+	wsnps, err := c.wsnpLister.List(labels.Everything())
+	if err != nil {
+		log.Error(err, "Failed to get WSNP when a workspace changed ")
+		return
+	}
+	for _, wsnp := range wsnps {
+		log.V(4).Info("Enqueue wsnp because a workspace being  changed", "obj", ws.Name)
+		c.enqueueWSNP(wsnp)
+	}
+	return
+}
+
+func (c *controller) Run(threadiness int, stopCh <-chan struct{}) error {
+	defer utilruntime.HandleCrash()
+	defer c.workqueue.ShutDown()
+
+	// Start the informer factories to begin populating the informer caches
+	log.Info("Starting WSNP controller")
+
+	// Wait for the caches to be synced before starting workers
+	log.Info("Waiting for informer caches to sync")
+	if ok := cache.WaitForCacheSync(stopCh, c.wsnpSynced, c.namespaceSynced, c.networkPolicySynced, c.workspaceSynced); !ok {
+		return fmt.Errorf("failed to wait for caches to sync")
+	}
+
+	log.Info("Starting workers")
+	// Launch two workers to process Foo resources
+	for i := 0; i < threadiness; i++ {
+		go wait.Until(c.runWorker, time.Second, stopCh)
+	}
+
+	klog.Info("Started workers")
+	<-stopCh
+	log.Info("Shutting down workers")
+
+	return nil
+}
+
+func (c *controller) enqueueWSNP(obj interface{}) {
+	var key string
+	var err error
+	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
+		utilruntime.HandleError(err)
+		return
+	}
+	c.workqueue.Add(key)
+}
+
+func (c *controller) handleNP(obj interface{}) {
+	var object metav1.Object
+	var ok bool
+	if object, ok = obj.(metav1.Object); !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			utilruntime.HandleError(fmt.Errorf("error decoding object, invalid type"))
+			return
+		}
+		object, ok = tombstone.Obj.(metav1.Object)
+		if !ok {
+			utilruntime.HandleError(fmt.Errorf("error decoding object tombstone, invalid type"))
+			return
+		}
+		log.V(4).Info("Recovered deleted object from tombstone", "name", object.GetName())
+	}
+	log.V(4).Info("Processing object:", "name", object.GetName())
+	if ownerRef := metav1.GetControllerOf(object); ownerRef != nil {
+		if ownerRef.Kind != "WorkspaceNetworkPol" {
+			return
+		}
+
+		wsnp, err := c.wsnpLister.Get(ownerRef.Name)
+		if err != nil {
+			log.V(4).Info("ignoring orphaned object", "link", object.GetSelfLink(), "name", ownerRef.Name)
+			return
+		}
+		c.enqueueWSNP(wsnp)
+		return
+	}
+}
+
+func (c *controller) runWorker() {
+	for c.processNextWorkItem() {
+	}
+}
+
+func (c *controller) processNextWorkItem() bool {
+	obj, shutdown := c.workqueue.Get()
+
+	if shutdown {
+		return false
+	}
+
+	// We wrap this block in a func so we can defer c.workqueue.Done.
+	err := func(obj interface{}) error {
+		// We call Done here so the workqueue knows we have finished
+		// processing this item. We also must remember to call Forget if we
+		// do not want this work item being re-queued. For example, we do
+		// not call Forget if a transient error occurs, instead the item is
+		// put back on the workqueue and attempted again after a back-off
+		// period.
+		defer c.workqueue.Done(obj)
+		var key string
+		var ok bool
+		// We expect strings to come off the workqueue. These are of the
+		// form namespace/name. We do this as the delayed nature of the
+		// workqueue means the items in the informer cache may actually be
+		// more up to date that when the item was initially put onto the
+		// workqueue.
+		if key, ok = obj.(string); !ok {
+			// As the item in the workqueue is actually invalid, we call
+			// Forget here else we'd go into a loop of attempting to
+			// process a work item that is invalid.
+			c.workqueue.Forget(obj)
+			utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
+			return nil
+		}
+		// Run the reconcile, passing it the namespace/name string of the
+		// Foo resource to be synced.
+		if err := c.reconcile(key); err != nil {
+			// Put the item back on the workqueue to handle any transient errors.
+			c.workqueue.AddRateLimited(key)
+			return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error())
+		}
+		// Finally, if no error occurs we Forget this item so it does not
+		// get queued again until another change happens.
+		c.workqueue.Forget(obj)
+		log.Info("Successfully synced", key)
+		return nil
+	}(obj)
+
+	if err != nil {
+		utilruntime.HandleError(err)
+		return true
+	}
+
+	return true
+}
+
+func (c *controller) handleError(err error) {
+	log.Error(err, "Error in handling")
+	errCount++
+}

pkg/controller/network/wsnetworkpolicy/reconcile.go

@@ -0,0 +1,203 @@
+package wsnetworkpolicy
+
+import (
+	"fmt"
+	"reflect"
+	"sort"
+
+	corev1 "k8s.io/api/core/v1"
+	ks8network "k8s.io/api/networking/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	errutil "k8s.io/apimachinery/pkg/util/errors"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/retry"
+	wsnpapi "kubesphere.io/kubesphere/pkg/apis/network/v1alpha1"
+)
+
+const (
+	workspaceSelectorLabel      = "kubesphere.io/workspace"
+	workspaceNetworkPolicyLabel = "networking.kubesphere.io/wsnp"
+
+	MessageResourceExists = "Resource %q already exists and is not managed by WorkspaceNetworkPolicy"
+	ErrResourceExists     = "ErrResourceExists"
+)
+
+var everything = labels.Everything()
+var reconcileCount = 0
+
+// NetworkPolicyNameForWSNP return the name of the networkpolicy owned by this WNSP
+func NetworkPolicyNameForWSNP(wsnp string) string {
+	return wsnp + "-np"
+}
+
+func (c *controller) reconcile(key string) error {
+	reconcileCount++
+	_, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
+		return nil
+	}
+	olog := log.WithName(name)
+	olog.Info("Begin to reconcile")
+	owner, err := c.wsnpLister.Get(name)
+	if err != nil {
+		if errors.IsNotFound(err) {
+			utilruntime.HandleError(fmt.Errorf("WSNP '%s' in work queue no longer exists", key))
+			return nil
+		}
+		return err
+	}
+	namespaces, err := c.listNamespacesInWorkspace(owner.Spec.Workspace)
+	if err != nil {
+		return err
+	}
+	var errs []error
+	for _, ns := range namespaces {
+		err = c.reconcileNamespace(ns.Name, owner)
+		if err != nil {
+			errs = append(errs, err)
+		}
+	}
+	if len(errs) == 0 {
+		return nil
+	}
+	return errutil.NewAggregate(errs)
+}
+
+func (c *controller) reconcileNamespace(name string, wsnp *wsnpapi.WorkspaceNetworkPolicy) error {
+	npname := NetworkPolicyNameForWSNP(wsnp.Name)
+	np, err := c.generateNPForNamesapce(name, wsnp)
+	if err != nil {
+		log.Error(nil, "Failed to generate NetworkPolicy", "wsnp", wsnp, "namespace", name)
+		return err
+	}
+	old, err := c.networkPolicyLister.NetworkPolicies(name).Get(npname)
+	if errors.IsNotFound(err) {
+		_, err = c.kubeClientset.NetworkingV1().NetworkPolicies(name).Create(np)
+		if err != nil {
+			log.Error(err, "cannot create networkpolicy of this wsnp", wsnp)
+			return err
+		}
+		return nil
+	}
+	if err != nil {
+		log.Error(err, "Failed to get networkPolicy")
+		return err
+	}
+	if !metav1.IsControlledBy(old, wsnp) {
+		msg := fmt.Sprintf(MessageResourceExists, old.Name)
+		c.recorder.Event(wsnp, corev1.EventTypeWarning, ErrResourceExists, msg)
+		return fmt.Errorf(msg)
+	}
+	if !reflect.DeepEqual(old.Spec, np.Spec) {
+		log.V(2).Info("Detect network policy changed, updating network policy", "the old one", old.Spec, "the new one", np.Spec)
+		err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+			_, err = c.kubeClientset.NetworkingV1().NetworkPolicies(name).Update(np)
+			return err
+		})
+		if err != nil {
+			log.Error(err, "Failed to update wsnp")
+			return err
+		}
+		log.V(2).Info("updating completed")
+	}
+	return nil
+}
+
+func (c *controller) generateNPForNamesapce(ns string, wsnp *wsnpapi.WorkspaceNetworkPolicy) (*ks8network.NetworkPolicy, error) {
+	np := &ks8network.NetworkPolicy{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      NetworkPolicyNameForWSNP(wsnp.Name),
+			Namespace: ns,
+			Labels:    map[string]string{workspaceNetworkPolicyLabel: wsnp.Name},
+			OwnerReferences: []metav1.OwnerReference{
+				*metav1.NewControllerRef(wsnp, wsnpapi.SchemeGroupVersion.WithKind("WorkspaceNetworkPolicy")),
+			},
+		},
+		Spec: ks8network.NetworkPolicySpec{
+			PolicyTypes: wsnp.Spec.PolicyTypes,
+		},
+	}
+
+	if wsnp.Spec.Ingress != nil {
+		np.Spec.Ingress = make([]ks8network.NetworkPolicyIngressRule, len(wsnp.Spec.Ingress))
+		for index, ing := range wsnp.Spec.Ingress {
+			ingRule, err := c.transformWSNPIngressToK8sIngress(ing)
+			if err != nil {
+				return nil, err
+			}
+			np.Spec.Ingress[index] = *ingRule
+		}
+	}
+	return np, nil
+}
+
+func (c *controller) transformWSNPIngressToK8sIngress(rule wsnpapi.WorkspaceNetworkPolicyIngressRule) (*ks8network.NetworkPolicyIngressRule, error) {
+	k8srule := &ks8network.NetworkPolicyIngressRule{
+		Ports: rule.Ports,
+		From:  make([]ks8network.NetworkPolicyPeer, len(rule.From)),
+	}
+	for index, f := range rule.From {
+		k8srule.From[index] = f.NetworkPolicyPeer
+		if f.WorkspaceSelector != nil {
+			if f.WorkspaceSelector.Size() == 0 {
+				k8srule.From[index].NamespaceSelector = &metav1.LabelSelector{}
+			} else {
+				selector, err := metav1.LabelSelectorAsSelector(f.WorkspaceSelector)
+				if err != nil {
+					log.Error(err, "Failed to convert label selectors")
+					return nil, err
+				}
+				ws, err := c.workspaceLister.List(selector)
+				if err != nil {
+					log.Error(err, "Failed to list workspaces")
+					return nil, err
+				}
+				if len(ws) == 0 {
+					log.Info("ws selector doesnot match anything")
+					continue
+				}
+				if k8srule.From[index].NamespaceSelector == nil {
+					k8srule.From[index].NamespaceSelector = &metav1.LabelSelector{}
+				}
+				if len(ws) == 1 {
+					if k8srule.From[index].NamespaceSelector.MatchLabels == nil {
+						k8srule.From[index].NamespaceSelector.MatchLabels = make(map[string]string)
+					}
+					k8srule.From[index].NamespaceSelector.MatchLabels[workspaceSelectorLabel] = ws[0].Name
+				} else {
+					if k8srule.From[index].NamespaceSelector.MatchExpressions == nil {
+						k8srule.From[index].NamespaceSelector.MatchExpressions = make([]metav1.LabelSelectorRequirement, 0)
+					}
+					re := metav1.LabelSelectorRequirement{
+						Key:      workspaceSelectorLabel,
+						Operator: metav1.LabelSelectorOpIn,
+						Values:   make([]string, len(ws)),
+					}
+					for index, w := range ws {
+						re.Values[index] = w.Name
+					}
+					sort.Strings(re.Values)
+					k8srule.From[index].NamespaceSelector.MatchExpressions = append(k8srule.From[index].NamespaceSelector.MatchExpressions, re)
+				}
+			}
+		}
+	}
+	return k8srule, nil
+}
+func (c *controller) listNamespacesInWorkspace(workspace string) ([]*corev1.Namespace, error) {
+	selector, err := labels.Parse(workspaceSelectorLabel + "==" + workspace)
+	if err != nil {
+		log.Error(err, "Failed to parse label selector")
+		return nil, err
+	}
+	namespaces, err := c.namespaceLister.List(selector)
+	if err != nil {
+		log.Error(err, "Failed to list namespaces in this workspace")
+		return nil, err
+	}
+	return namespaces, nil
+}

pkg/controller/network/wsnetworkpolicy/wsnetworkpolicy_suite_test.go

@@ -0,0 +1,21 @@
+package wsnetworkpolicy
+
+import (
+	"flag"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"k8s.io/klog"
+)
+
+func TestWsnetworkpolicy(t *testing.T) {
+	klog.InitFlags(nil)
+	flag.Set("logtostderr", "false")
+	flag.Set("alsologtostderr", "false")
+	flag.Set("v", "4")
+	flag.Parse()
+	klog.SetOutput(GinkgoWriter)
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Wsnetworkpolicy Suite")
+}

pkg/controller/network/wsnetworkpolicy/wsnetworkpolicy_test.go

@@ -0,0 +1,241 @@
+package wsnetworkpolicy
+
+import (
+	"fmt"
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	. "github.com/onsi/gomega/gstruct"
+	corev1 "k8s.io/api/core/v1"
+	k8snetwork "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	netv1lister "k8s.io/client-go/listers/networking/v1"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/klog"
+	"kubesphere.io/kubesphere/pkg/apis/network/v1alpha1"
+	tenant "kubesphere.io/kubesphere/pkg/apis/tenant/v1alpha1"
+	controllertesting "kubesphere.io/kubesphere/pkg/controller/network/testing"
+)
+
+var (
+	fakeControllerBuilder *controllertesting.FakeControllerBuilder
+	c                     Controller
+	npLister              netv1lister.NetworkPolicyLister
+	stopCh                chan struct{}
+	deletePolicy          metav1.DeletionPropagation
+	testName              string
+)
+
+var _ = Describe("Wsnetworkpolicy", func() {
+	BeforeEach(func() {
+		deletePolicy = metav1.DeletePropagationBackground
+		fakeControllerBuilder = controllertesting.NewFakeControllerBuilder()
+		informer, k8sinformer := fakeControllerBuilder.NewControllerInformer()
+		stopCh = make(chan struct{})
+		c = NewController(fakeControllerBuilder.KubeClient, fakeControllerBuilder.KsClient,
+			informer.Network().V1alpha1().WorkspaceNetworkPolicies(), k8sinformer.Networking().V1().NetworkPolicies(),
+			k8sinformer.Core().V1().Namespaces(), informer.Tenant().V1alpha1().Workspaces())
+		originalController := c.(*controller)
+		go originalController.wsnpInformer.Informer().Run(stopCh)
+		go originalController.networkPolicyInformer.Informer().Run(stopCh)
+		go originalController.namespaceInformer.Informer().Run(stopCh)
+		go originalController.workspaceInformer.Informer().Run(stopCh)
+		originalController.recorder = &record.FakeRecorder{}
+		go c.Run(1, stopCh)
+		npLister = k8sinformer.Networking().V1().NetworkPolicies().Lister()
+		testName = "test"
+		ns1 := newWorkspaceNamespaces("ns1", testName)
+		ns2 := newWorkspaceNamespaces("ns2", testName)
+		_, err := fakeControllerBuilder.KubeClient.CoreV1().Namespaces().Create(ns1)
+		Expect(err).ShouldNot(HaveOccurred())
+		_, err = fakeControllerBuilder.KubeClient.CoreV1().Namespaces().Create(ns2)
+		Expect(err).ShouldNot(HaveOccurred())
+	})
+
+	AfterEach(func() {
+		close(stopCh)
+	})
+
+	It("Should proper ingress rule when using workspaceSelector", func() {
+		label := map[string]string{"workspace": "test-selector"}
+		ws := &tenant.Workspace{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:   "test",
+				Labels: label,
+			},
+		}
+		_, err := fakeControllerBuilder.KsClient.TenantV1alpha1().Workspaces().Create(ws)
+		wsnp := newWorkspaceNP(testName)
+		wsnp.Spec.PolicyTypes = []k8snetwork.PolicyType{k8snetwork.PolicyTypeIngress}
+		wsnp.Spec.Ingress = []v1alpha1.WorkspaceNetworkPolicyIngressRule{
+			v1alpha1.WorkspaceNetworkPolicyIngressRule{
+				From: []v1alpha1.WorkspaceNetworkPolicyPeer{
+					v1alpha1.WorkspaceNetworkPolicyPeer{
+						WorkspaceSelector: &metav1.LabelSelector{
+							MatchLabels: label,
+						},
+					},
+				},
+			}}
+		_, err = fakeControllerBuilder.KsClient.NetworkV1alpha1().WorkspaceNetworkPolicies().Create(wsnp)
+		Expect(err).ShouldNot(HaveOccurred())
+		expect1Json := `{
+			"apiVersion": "networking.k8s.io/v1",
+			"kind": "NetworkPolicy",
+			"metadata": {
+				"name": "test-np",
+				"namespace": "ns1",
+				"labels": {
+					"networking.kubesphere.io/wsnp": "test"
+				}
+			},
+			"spec": {
+				"policyTypes": [
+					"Ingress"
+				],
+				"ingress": [
+					{
+						"from": [
+							{
+								"namespaceSelector": { 
+									"matchLabels": {
+										"kubesphere.io/workspace": "test"
+									}
+								}
+							}
+						]
+					}
+				]
+			}
+		}`
+		expect1 := &k8snetwork.NetworkPolicy{}
+		Expect(controllertesting.StringToObject(expect1Json, expect1)).ShouldNot(HaveOccurred())
+		nps := []*k8snetwork.NetworkPolicy{}
+		Eventually(func() error {
+			selector, _ := labels.Parse(workspaceNetworkPolicyLabel + "==test")
+			nps, err = npLister.List(selector)
+			if err != nil {
+				klog.Errorf("Failed to list npmerr:%s", err.Error())
+				return err
+			}
+			if len(nps) != 2 {
+				return fmt.Errorf("Length is not right, current length :%d", len(nps))
+			}
+			return nil
+		}, time.Second*5, time.Second).ShouldNot(HaveOccurred())
+
+		for _, np := range nps {
+			Expect(np.Labels).To(Equal(expect1.Labels))
+			Expect(np.Spec).To(Equal(expect1.Spec))
+		}
+		// create a new ws will change the `From`
+		ws2 := &tenant.Workspace{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:   "test2",
+				Labels: label,
+			},
+		}
+		_, err = fakeControllerBuilder.KsClient.TenantV1alpha1().Workspaces().Create(ws2)
+		Expect(err).ShouldNot(HaveOccurred())
+		expect2Json := `{
+			"apiVersion": "networking.k8s.io/v1",
+			"kind": "NetworkPolicy",
+			"metadata": {
+				"name": "test-np",
+				"namespace": "ns1",
+				"labels": {
+					"networking.kubesphere.io/wsnp": "test"
+				}
+			},
+			"spec": {
+				"policyTypes": [
+					"Ingress"
+				],
+				"ingress": [
+					{
+						"from": [
+							{
+								"namespaceSelector": { 
+									"matchExpressions": [{
+										"key": "kubesphere.io/workspace",
+										"operator":"In",
+										"values": ["test", "test2"]
+									}]
+								}
+							}
+						]
+					}
+				]
+			}
+		}`
+		expect2 := &k8snetwork.NetworkPolicy{}
+		Expect(controllertesting.StringToObject(expect2Json, expect2)).ShouldNot(HaveOccurred())
+
+		id := func(element interface{}) string {
+			e := element.(*k8snetwork.NetworkPolicy)
+			return e.Namespace
+		}
+		Eventually(func() []*k8snetwork.NetworkPolicy {
+			selector, _ := labels.Parse(workspaceNetworkPolicyLabel + "=test")
+			nps, err := npLister.List(selector)
+			if err != nil {
+				return nil
+			}
+			if len(nps) != 2 {
+				klog.Errorf("Length is not right, current length :%d", len(nps))
+				return nil
+			}
+			return nps
+		}, time.Second*5, time.Second).Should(MatchAllElements(id, Elements{
+			"ns1": PointTo(MatchFields(IgnoreExtras, Fields{
+				"Spec": Equal(expect2.Spec),
+			})),
+			"ns2": PointTo(MatchFields(IgnoreExtras, Fields{
+				"Spec": Equal(expect2.Spec),
+			})),
+		}))
+	})
+
+	It("Should create networkpolicies", func() {
+		//create a wsnp
+		_, err := fakeControllerBuilder.KsClient.NetworkV1alpha1().WorkspaceNetworkPolicies().Create(newWorkspaceNP(testName))
+		Expect(err).ShouldNot(HaveOccurred())
+		Eventually(func() error {
+			selector, _ := labels.Parse(workspaceNetworkPolicyLabel + "=" + testName)
+			nps, err := npLister.List(selector)
+			if err != nil {
+				return err
+			}
+			if len(nps) != 2 {
+				return fmt.Errorf("Length is not right, current length :%d", len(nps))
+			}
+			return nil
+		}, time.Second*5, time.Second).ShouldNot(HaveOccurred())
+		err = fakeControllerBuilder.KsClient.NetworkV1alpha1().WorkspaceNetworkPolicies().Delete(testName, &metav1.DeleteOptions{
+			PropagationPolicy: &deletePolicy,
+		})
+		Expect(err).ShouldNot(HaveOccurred())
+	})
+})
+
+func newWorkspaceNP(name string) *v1alpha1.WorkspaceNetworkPolicy {
+	return &v1alpha1.WorkspaceNetworkPolicy{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: v1alpha1.WorkspaceNetworkPolicySpec{
+			Workspace: name,
+		},
+	}
+}
+
+func newWorkspaceNamespaces(ns, ws string) *corev1.Namespace {
+	return &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   ns,
+			Labels: map[string]string{workspaceSelectorLabel: ws},
+		},
+	}
+}