Adding Networkpolicy to ClusterSets (#125)

* Adding cluster set types Adds types for cluster sets, which allows constraining a few elements of clusters including: overall resource usage, and which nodes it can use. * Add networkpolicy to clustersets Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com> * Fix comments Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com> * Fix linting issues Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com> * fixing node controller logic and nit fixes Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com> * more fixes Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com> * fix main cli Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com> * Comment the resource quota for clustersets Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com> --------- Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com> Co-authored-by: Michael Bolot <michael.bolot@suse.com>
2026-08-18 20:07:06 +00:00 · 2024-10-16 00:27:42 +03:00
parent dbe6767aff
commit bf1fe2a71c
18 changed files with 801 additions and 31 deletions
@@ -21,7 +21,10 @@ func Resource(resource string) schema.GroupResource {
 func addKnownTypes(s *runtime.Scheme) error {
 	s.AddKnownTypes(SchemeGroupVersion,
 		&Cluster{},
-		&ClusterList{})
+		&ClusterList{},
+		&ClusterSet{},
+		&ClusterSetList{},
+	)
 	metav1.AddToGroupVersion(s, SchemeGroupVersion)
 	return nil
 }
@@ -0,0 +1,52 @@
+package v1alpha1
+
+import (
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// +genclient
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+// +kubebuilder:storageversion
+// +kubebuilder:subresource:status
+
+type ClusterSet struct {
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+	metav1.TypeMeta   `json:",inline"`
+
+	// Spec is the spec of the ClusterSet
+	Spec ClusterSetSpec `json:"spec"`
+	// Status is the status of the ClusterSet
+	Status ClusterSetStatus `json:"status,omitempty"`
+}
+
+type ClusterSetSpec struct {
+	// MaxLimits are the limits that apply to all clusters (server + agent) in the set
+	MaxLimits v1.ResourceList `json:"maxLimits,omitempty"`
+	// DefaultLimits are the limits used for servers/agents when a cluster in the set doesn't provide any
+	DefaultLimits *ClusterLimit `json:"defaultLimits,omitempty"`
+	// DefaultNodeSelector is the node selector that applies to all clusters (server + agent) in the set
+	DefaultNodeSelector map[string]string `json:"defaultNodeSelector,omitempty"`
+	// DisableNetworkPolicy is an option that will disable the creation of a default networkpolicy for cluster isolation
+	DisableNetworkPolicy bool `json:"disableNetworkPolicy,omitempty"`
+}
+
+type ClusterSetStatus struct {
+	// ObservedGeneration was the generation at the time the status was updated.
+	ObservedGeneration int64 `json:"observedGeneration,omitempty"`
+	// LastUpdate is the timestamp when the status was last updated
+	LastUpdate string `json:"lastUpdateTime,omitempty"`
+	// Sumamry is a summary of the status (error, ready)
+	Summary string `json:"summary,omitempty"`
+	// Conditions are the invidual conditions for the cluster set
+	Conditions []metav1.Condition `json:"conditions,omitempty" patchStrategy:"merge" patchMergeKey:"type" protobuf:"bytes,1,rep,name=conditions"`
+}
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+type ClusterSetList struct {
+	metav1.ListMeta `json:"metadata,omitempty"`
+	metav1.TypeMeta `json:",inline"`
+
+	Items []ClusterSet `json:"items"`
+}
@@ -27,6 +27,10 @@ type ClusterSpec struct {
 	// +kubebuilder:validation:XValidation:message="invalid value for agents",rule="self >= 0"
 	// Agents is the number of K3s pods to run in agent (worker) mode.
 	Agents *int32 `json:"agents"`
+	// NodeSelector is the node selector that will be applied to all server/agent pods
+	NodeSelector map[string]string `json:"nodeSelector,omitempty"`
+	// Limit is the limits that apply for the server/worker nodes.
+	Limit *ClusterLimit `json:"clusterLimit,omitempty"`
 	// +kubebuilder:validation:XValidation:message="token is immutable",rule="self == oldSelf"
 	// Token is the token used to join the worker nodes to the cluster.
 	Token string `json:"token"`
@@ -52,6 +56,7 @@ type ClusterSpec struct {
 	// Persistence contains options controlling how the etcd data of the virtual cluster is persisted. By default, no data
 	// persistence is guaranteed, so restart of a virtual cluster pod may result in data loss without this field.
 	Persistence *PersistenceConfig `json:"persistence,omitempty"`
+	// +optional
 	// Expose contains options for exposing the apiserver inside/outside of the cluster. By default, this is only exposed as a
 	// clusterIP which is relatively secure, but difficult to access outside of the cluster.
 	Expose *ExposeConfig `json:"expose,omitempty"`
@@ -87,14 +92,17 @@ type PersistenceConfig struct {
 }

 type ExposeConfig struct {
-	Ingress      *IngressConfig      `json:"ingress"`
-	LoadBalancer *LoadBalancerConfig `json:"loadbalancer"`
-	NodePort     *NodePortConfig     `json:"nodePort"`
+	// +optional
+	Ingress *IngressConfig `json:"ingress,omitempty"`
+	// +optional
+	LoadBalancer *LoadBalancerConfig `json:"loadbalancer,omitempty"`
+	// +optional
+	NodePort *NodePortConfig `json:"nodePort,omitempty"`
 }

 type IngressConfig struct {
-	Enabled          bool   `json:"enabled"`
-	IngressClassName string `json:"ingressClassName"`
+	Enabled          bool   `json:"enabled,omitempty"`
+	IngressClassName string `json:"ingressClassName,omitempty"`
 }

 type LoadBalancerConfig struct {
@@ -6,6 +6,8 @@
 package v1alpha1

 import (
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	runtime "k8s.io/apimachinery/pkg/runtime"
 )

@@ -53,6 +55,36 @@ func (in *Cluster) DeepCopyObject() runtime.Object {
 	return nil
 }

+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ClusterLimit) DeepCopyInto(out *ClusterLimit) {
+	*out = *in
+	if in.ServerLimit != nil {
+		in, out := &in.ServerLimit, &out.ServerLimit
+		*out = make(v1.ResourceList, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val.DeepCopy()
+		}
+	}
+	if in.WorkerLimit != nil {
+		in, out := &in.WorkerLimit, &out.WorkerLimit
+		*out = make(v1.ResourceList, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val.DeepCopy()
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterLimit.
+func (in *ClusterLimit) DeepCopy() *ClusterLimit {
+	if in == nil {
+		return nil
+	}
+	out := new(ClusterLimit)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ClusterList) DeepCopyInto(out *ClusterList) {
 	*out = *in
@@ -86,6 +118,125 @@ func (in *ClusterList) DeepCopyObject() runtime.Object {
 	return nil
 }

+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ClusterSet) DeepCopyInto(out *ClusterSet) {
+	*out = *in
+	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	out.TypeMeta = in.TypeMeta
+	in.Spec.DeepCopyInto(&out.Spec)
+	in.Status.DeepCopyInto(&out.Status)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterSet.
+func (in *ClusterSet) DeepCopy() *ClusterSet {
+	if in == nil {
+		return nil
+	}
+	out := new(ClusterSet)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *ClusterSet) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ClusterSetList) DeepCopyInto(out *ClusterSetList) {
+	*out = *in
+	in.ListMeta.DeepCopyInto(&out.ListMeta)
+	out.TypeMeta = in.TypeMeta
+	if in.Items != nil {
+		in, out := &in.Items, &out.Items
+		*out = make([]ClusterSet, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterSetList.
+func (in *ClusterSetList) DeepCopy() *ClusterSetList {
+	if in == nil {
+		return nil
+	}
+	out := new(ClusterSetList)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *ClusterSetList) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ClusterSetSpec) DeepCopyInto(out *ClusterSetSpec) {
+	*out = *in
+	if in.MaxLimits != nil {
+		in, out := &in.MaxLimits, &out.MaxLimits
+		*out = make(v1.ResourceList, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val.DeepCopy()
+		}
+	}
+	if in.DefaultLimits != nil {
+		in, out := &in.DefaultLimits, &out.DefaultLimits
+		*out = new(ClusterLimit)
+		(*in).DeepCopyInto(*out)
+	}
+	if in.DefaultNodeSelector != nil {
+		in, out := &in.DefaultNodeSelector, &out.DefaultNodeSelector
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterSetSpec.
+func (in *ClusterSetSpec) DeepCopy() *ClusterSetSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(ClusterSetSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ClusterSetStatus) DeepCopyInto(out *ClusterSetStatus) {
+	*out = *in
+	if in.Conditions != nil {
+		in, out := &in.Conditions, &out.Conditions
+		*out = make([]metav1.Condition, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterSetStatus.
+func (in *ClusterSetStatus) DeepCopy() *ClusterSetStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(ClusterSetStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ClusterSpec) DeepCopyInto(out *ClusterSpec) {
 	*out = *in
@@ -99,6 +250,18 @@ func (in *ClusterSpec) DeepCopyInto(out *ClusterSpec) {
 		*out = new(int32)
 		**out = **in
 	}
+	if in.NodeSelector != nil {
+		in, out := &in.NodeSelector, &out.NodeSelector
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+	if in.Limit != nil {
+		in, out := &in.Limit, &out.Limit
+		*out = new(ClusterLimit)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.ServerArgs != nil {
 		in, out := &in.ServerArgs, &out.ServerArgs
 		*out = make([]string, len(*in))
@@ -130,8 +130,12 @@ func (a *Agent) StatefulAgent(cluster *v1alpha1.Cluster) *apps.StatefulSet {

 func (a *Agent) podSpec(image, name string, args []string, statefulSet bool, affinitySelector *metav1.LabelSelector) v1.PodSpec {
 	var limit v1.ResourceList
+	if a.cluster.Spec.Limit != nil && a.cluster.Spec.Limit.ServerLimit != nil {
+		limit = a.cluster.Spec.Limit.ServerLimit
+	}
 	args = append([]string{"agent", "--config", "/opt/rancher/k3s/config.yaml"}, args...)
 	podSpec := v1.PodSpec{
+		NodeSelector: a.cluster.Spec.NodeSelector,
 		Affinity: &v1.Affinity{
 			PodAntiAffinity: &v1.PodAntiAffinity{
 				RequiredDuringSchedulingIgnoredDuringExecution: []v1.PodAffinityTerm{
@@ -33,6 +33,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )

+const (
+	podController = "k3k-pod-controller"
+)
+
 type PodReconciler struct {
 	Client ctrlruntimeclient.Client
 	Scheme *runtime.Scheme
@@ -48,7 +52,7 @@ func AddPodController(ctx context.Context, mgr manager.Manager) error {

 	return ctrl.NewControllerManagedBy(mgr).
 		Watches(&v1.Pod{}, handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &apps.StatefulSet{}, handler.OnlyControllerOwner())).
-		Named("cluster-pod-controller").
+		Named(podController).
 		WithOptions(controller.Options{
 			MaxConcurrentReconciles: maxConcurrentReconciles,
 		}).
@@ -62,6 +62,14 @@ func Generate(ctx context.Context, cluster *v1alpha1.Cluster, ip string) (*v1.Se
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      cluster.Name + "-bootstrap",
 			Namespace: util.ClusterNamespace(cluster),
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: cluster.APIVersion,
+					Kind:       cluster.Kind,
+					Name:       cluster.Name,
+					UID:        cluster.UID,
+				},
+			},
 		},
 		Data: map[string][]byte{
 			"bootstrap": bootstrapData,
@@ -41,7 +41,12 @@ func New(cluster *v1alpha1.Cluster, client client.Client) *Server {
 }

 func (s *Server) podSpec(ctx context.Context, image, name string, persistent bool, affinitySelector *metav1.LabelSelector) v1.PodSpec {
+	var limit v1.ResourceList
+	if s.cluster.Spec.Limit != nil && s.cluster.Spec.Limit.ServerLimit != nil {
+		limit = s.cluster.Spec.Limit.ServerLimit
+	}
 	podSpec := v1.PodSpec{
+		NodeSelector: s.cluster.Spec.NodeSelector,
 		Affinity: &v1.Affinity{
 			PodAntiAffinity: &v1.PodAntiAffinity{
 				RequiredDuringSchedulingIgnoredDuringExecution: []v1.PodAffinityTerm{
@@ -110,6 +115,9 @@ func (s *Server) podSpec(ctx context.Context, image, name string, persistent boo
 			{
 				Name:  name,
 				Image: image,
+				Resources: v1.ResourceRequirements{
+					Limits: limit,
+				},
 				Env: []v1.EnvVar{
 					{
 						Name: "POD_NAME",
@@ -0,0 +1,159 @@
+package clusterset
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
+	v1 "k8s.io/api/core/v1"
+	networkingv1 "k8s.io/api/networking/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
+	ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+const (
+	clusterSetController    = "k3k-clusterset-controller"
+	networkPolicyName       = "k3k-cluster-netpol"
+	allTrafficCIDR          = "0.0.0.0/0"
+	maxConcurrentReconciles = 1
+)
+
+type ClusterSetReconciler struct {
+	Client      ctrlruntimeclient.Client
+	Scheme      *runtime.Scheme
+	ClusterCIDR string
+}
+
+// Add adds a new controller to the manager
+func Add(ctx context.Context, mgr manager.Manager, clusterCIDR string) error {
+	// initialize a new Reconciler
+	reconciler := ClusterSetReconciler{
+		Client:      mgr.GetClient(),
+		Scheme:      mgr.GetScheme(),
+		ClusterCIDR: clusterCIDR,
+	}
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&v1alpha1.ClusterSet{}).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: maxConcurrentReconciles,
+		}).
+		Complete(&reconciler)
+}
+
+func (c *ClusterSetReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
+	var (
+		clusterSet v1alpha1.ClusterSet
+	)
+	if err := c.Client.Get(ctx, types.NamespacedName{Name: req.Name, Namespace: req.Namespace}, &clusterSet); err != nil {
+		return reconcile.Result{}, fmt.Errorf("unable to get the clusterset: %w", err)
+	}
+
+	if !clusterSet.Spec.DisableNetworkPolicy {
+		setNetworkPolicy, err := netpol(ctx, c.ClusterCIDR, &clusterSet, c.Client)
+		if err != nil {
+			return reconcile.Result{}, fmt.Errorf("unable to make a networkpolicy for cluster set: %w", err)
+		}
+		if err := c.Client.Create(ctx, setNetworkPolicy); err != nil {
+			if apierrors.IsAlreadyExists(err) {
+				if err := c.Client.Update(ctx, setNetworkPolicy); err != nil {
+					return reconcile.Result{}, fmt.Errorf("unable to update networkpolicy for clusterset: %w", err)
+				}
+			}
+			return reconcile.Result{}, fmt.Errorf("unable to create networkpolicy for clusterset: %w", err)
+		}
+	}
+	// TODO: Add resource quota for clustersets
+	// if clusterSet.Spec.MaxLimits != nil {
+	// 	quota := v1.ResourceQuota{
+	// 		ObjectMeta: metav1.ObjectMeta{
+	// 			Name:      "clusterset-quota",
+	// 			Namespace: clusterSet.Namespace,
+	// 			OwnerReferences: []metav1.OwnerReference{
+	// 				{
+	// 					UID:        clusterSet.UID,
+	// 					Name:       clusterSet.Name,
+	// 					APIVersion: clusterSet.APIVersion,
+	// 					Kind:       clusterSet.Kind,
+	// 				},
+	// 			},
+	// 		},
+	// 	}
+	// 	quota.Spec.Hard = clusterSet.Spec.MaxLimits
+	// 	if err := c.Client.Create(ctx, &quota); err != nil {
+	// 		return reconcile.Result{}, fmt.Errorf("unable to create resource quota from cluster set: %w", err)
+	// 	}
+	// }
+	return reconcile.Result{}, nil
+}
+
+func netpol(ctx context.Context, clusterCIDR string, clusterSet *v1alpha1.ClusterSet, client ctrlruntimeclient.Client) (*networkingv1.NetworkPolicy, error) {
+	var cidrList []string
+	if clusterCIDR == "" {
+		var nodeList v1.NodeList
+		if err := client.List(ctx, &nodeList); err != nil {
+			return nil, err
+		}
+		for _, node := range nodeList.Items {
+			cidrList = append(cidrList, node.Spec.PodCIDRs...)
+		}
+	} else {
+		cidrList = []string{clusterCIDR}
+	}
+	return &networkingv1.NetworkPolicy{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      networkPolicyName,
+			Namespace: clusterSet.Namespace,
+		},
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "NetworkPolicy",
+			APIVersion: "networking.k8s.io/v1",
+		},
+		Spec: networkingv1.NetworkPolicySpec{
+			PolicyTypes: []networkingv1.PolicyType{
+				networkingv1.PolicyTypeIngress,
+				networkingv1.PolicyTypeEgress,
+			},
+			Ingress: []networkingv1.NetworkPolicyIngressRule{
+				{},
+			},
+			Egress: []networkingv1.NetworkPolicyEgressRule{
+				{
+					To: []networkingv1.NetworkPolicyPeer{
+						{
+							IPBlock: &networkingv1.IPBlock{
+								CIDR:   allTrafficCIDR,
+								Except: cidrList,
+							},
+						},
+						{
+							NamespaceSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{
+									"kubernetes.io/metadata.name": clusterSet.Namespace,
+								},
+							},
+						},
+						{
+							NamespaceSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{
+									"kubernetes.io/metadata.name": metav1.NamespaceSystem,
+								},
+							},
+							PodSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{
+									"k8s-app": "kube-dns",
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}, nil
+}
@@ -0,0 +1,78 @@
+package clusterset
+
+import (
+	"context"
+
+	"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
+	"github.com/rancher/k3k/pkg/controller/util"
+	v1 "k8s.io/api/core/v1"
+	networkingv1 "k8s.io/api/networking/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
+	ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+const (
+	nodeController = "k3k-node-controller"
+)
+
+type NodeReconciler struct {
+	Client      ctrlruntimeclient.Client
+	Scheme      *runtime.Scheme
+	ClusterCIDR string
+}
+
+// AddNodeController adds a new controller to the manager
+func AddNodeController(ctx context.Context, mgr manager.Manager) error {
+	// initialize a new Reconciler
+	reconciler := NodeReconciler{
+		Client: mgr.GetClient(),
+		Scheme: mgr.GetScheme(),
+	}
+
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&v1.Node{}).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: maxConcurrentReconciles,
+		}).
+		Named(nodeController).
+		Complete(&reconciler)
+}
+
+func (n *NodeReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
+	var clusterSetList v1alpha1.ClusterSetList
+	if err := n.Client.List(ctx, &clusterSetList); err != nil {
+		return reconcile.Result{}, util.LogAndReturnErr("failed to list clusterSets", err)
+	}
+
+	if len(clusterSetList.Items) <= 0 {
+		return reconcile.Result{}, nil
+	}
+
+	if err := n.ensureNetworkPolicies(ctx, clusterSetList); err != nil {
+		return reconcile.Result{}, err
+	}
+
+	return reconcile.Result{}, nil
+}
+
+func (n *NodeReconciler) ensureNetworkPolicies(ctx context.Context, clusterSetList v1alpha1.ClusterSetList) error {
+	var setNetworkPolicy *networkingv1.NetworkPolicy
+	for _, cs := range clusterSetList.Items {
+		if cs.Spec.DisableNetworkPolicy {
+			continue
+		}
+		var err error
+		setNetworkPolicy, err = netpol(ctx, "", &cs, n.Client)
+		if err != nil {
+			return err
+		}
+		if err := n.Client.Update(ctx, setNetworkPolicy); err != nil {
+			return err
+		}
+	}
+	return nil
+}