rename nsnp controller
Signed-off-by: Duan Jiong <djduanjiong@gmail.com>
This commit is contained in:
Duan Jiong
@@ -0,0 +1,687 @@
|
||||
package nsnetworkpolicy
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
netv1 "k8s.io/api/networking/v1"
|
||||
"k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
typev1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/labels"
|
||||
"k8s.io/apimachinery/pkg/util/intstr"
|
||||
uruntime "k8s.io/apimachinery/pkg/util/runtime"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
v1 "k8s.io/client-go/informers/core/v1"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/client-go/util/workqueue"
|
||||
"k8s.io/klog"
|
||||
"kubesphere.io/kubesphere/pkg/apis/network/v1alpha1"
|
||||
workspacev1alpha1 "kubesphere.io/kubesphere/pkg/apis/tenant/v1alpha1"
|
||||
ksnetclient "kubesphere.io/kubesphere/pkg/client/clientset/versioned/typed/network/v1alpha1"
|
||||
nspolicy "kubesphere.io/kubesphere/pkg/client/informers/externalversions/network/v1alpha1"
|
||||
workspace "kubesphere.io/kubesphere/pkg/client/informers/externalversions/tenant/v1alpha1"
|
||||
"kubesphere.io/kubesphere/pkg/constants"
|
||||
"kubesphere.io/kubesphere/pkg/controller/network"
|
||||
"kubesphere.io/kubesphere/pkg/controller/network/provider"
|
||||
)
|
||||
|
||||
const (
|
||||
//TODO use set to track service:map
|
||||
//use period sync service label in NSNP
|
||||
defaultSleepDuration = 60 * time.Second
|
||||
|
||||
defaultThread = 5
|
||||
defaultSync = "5m"
|
||||
|
||||
//whether network isolate is enable in namespace
|
||||
NamespaceNPAnnotationKey = "kubesphere.io/network-isolate"
|
||||
NamespaceNPAnnotationEnabled = "enabled"
|
||||
|
||||
NodeNSNPAnnotationKey = "kubesphere.io/snat-node-ips"
|
||||
|
||||
AnnotationNPNAME = network.NSNPPrefix + "network-isolate"
|
||||
|
||||
//TODO: configure it
|
||||
DNSLocalIP = "169.254.25.10"
|
||||
DNSPort = 53
|
||||
DNSNamespace = "kube-system"
|
||||
DNSServiceName = "kube-dns"
|
||||
DNSServiceCoreDNS = "coredns"
|
||||
)
|
||||
|
||||
// namespacenpController implements the Controller interface for managing kubesphere network policies
|
||||
// and convery them to k8s NetworkPolicies, then syncing them to the provider.
|
||||
type NSNetworkPolicyController struct {
|
||||
client kubernetes.Interface
|
||||
ksclient ksnetclient.NetworkV1alpha1Interface
|
||||
informer nspolicy.NamespaceNetworkPolicyInformer
|
||||
informerSynced cache.InformerSynced
|
||||
|
||||
//TODO support service in same namespace
|
||||
serviceInformer v1.ServiceInformer
|
||||
serviceInformerSynced cache.InformerSynced
|
||||
|
||||
nodeInformer v1.NodeInformer
|
||||
nodeInformerSynced cache.InformerSynced
|
||||
|
||||
workspaceInformer workspace.WorkspaceInformer
|
||||
workspaceInformerSynced cache.InformerSynced
|
||||
|
||||
namespaceInformer v1.NamespaceInformer
|
||||
namespaceInformerSynced cache.InformerSynced
|
||||
|
||||
provider provider.NsNetworkPolicyProvider
|
||||
|
||||
nsQueue workqueue.RateLimitingInterface
|
||||
nsnpQueue workqueue.RateLimitingInterface
|
||||
}
|
||||
|
||||
func stringToCIDR(ipStr string) (string, error) {
|
||||
cidr := ""
|
||||
if ip := net.ParseIP(ipStr); ip != nil {
|
||||
if ip.To4() != nil {
|
||||
cidr = ipStr + "/32"
|
||||
} else {
|
||||
cidr = ipStr + "/128"
|
||||
}
|
||||
} else {
|
||||
return cidr, fmt.Errorf("ip string %s parse error\n", ipStr)
|
||||
}
|
||||
|
||||
return cidr, nil
|
||||
}
|
||||
|
||||
func generateDNSRule(nameServers []string) (netv1.NetworkPolicyEgressRule, error) {
|
||||
var rule netv1.NetworkPolicyEgressRule
|
||||
for _, nameServer := range nameServers {
|
||||
cidr, err := stringToCIDR(nameServer)
|
||||
if err != nil {
|
||||
return rule, err
|
||||
}
|
||||
rule.To = append(rule.To, netv1.NetworkPolicyPeer{
|
||||
IPBlock: &netv1.IPBlock{
|
||||
CIDR: cidr,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
protocolTCP := corev1.ProtocolTCP
|
||||
protocolUDP := corev1.ProtocolUDP
|
||||
dnsPort := intstr.FromInt(DNSPort)
|
||||
rule.Ports = append(rule.Ports, netv1.NetworkPolicyPort{
|
||||
Protocol: &protocolTCP,
|
||||
Port: &dnsPort,
|
||||
}, netv1.NetworkPolicyPort{
|
||||
Protocol: &protocolUDP,
|
||||
Port: &dnsPort,
|
||||
})
|
||||
|
||||
return rule, nil
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) generateDNSServiceRule() (netv1.NetworkPolicyEgressRule, error) {
|
||||
peer, ports, err := c.handlerPeerService(DNSNamespace, DNSServiceName, false)
|
||||
if err != nil {
|
||||
peer, ports, err = c.handlerPeerService(DNSNamespace, DNSServiceCoreDNS, false)
|
||||
}
|
||||
|
||||
return netv1.NetworkPolicyEgressRule{
|
||||
Ports: ports,
|
||||
To: []netv1.NetworkPolicyPeer{peer},
|
||||
}, err
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) handlerPeerService(namespace string, name string, ingress bool) (netv1.NetworkPolicyPeer, []netv1.NetworkPolicyPort, error) {
|
||||
peerNP := netv1.NetworkPolicyPeer{}
|
||||
var ports []netv1.NetworkPolicyPort
|
||||
|
||||
service, err := c.serviceInformer.Lister().Services(namespace).Get(name)
|
||||
if err != nil {
|
||||
return peerNP, nil, err
|
||||
}
|
||||
|
||||
peerNP.PodSelector = new(metav1.LabelSelector)
|
||||
peerNP.NamespaceSelector = new(metav1.LabelSelector)
|
||||
|
||||
if len(service.Spec.Selector) <= 0 {
|
||||
return peerNP, nil, fmt.Errorf("service %s/%s has no podselect", namespace, name)
|
||||
}
|
||||
|
||||
peerNP.PodSelector.MatchLabels = make(map[string]string)
|
||||
for key, value := range service.Spec.Selector {
|
||||
peerNP.PodSelector.MatchLabels[key] = value
|
||||
}
|
||||
peerNP.NamespaceSelector.MatchLabels = make(map[string]string)
|
||||
peerNP.NamespaceSelector.MatchLabels[constants.NamespaceLabelKey] = namespace
|
||||
|
||||
//only allow traffic to service exposed ports
|
||||
if !ingress {
|
||||
ports = make([]netv1.NetworkPolicyPort, 0)
|
||||
for _, port := range service.Spec.Ports {
|
||||
protocol := port.Protocol
|
||||
portIntString := intstr.FromInt(int(port.Port))
|
||||
ports = append(ports, netv1.NetworkPolicyPort{
|
||||
Protocol: &protocol,
|
||||
Port: &portIntString,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return peerNP, ports, err
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) convertPeer(peer v1alpha1.NetworkPolicyPeer, ingress bool) (netv1.NetworkPolicyPeer, []netv1.NetworkPolicyPort, error) {
|
||||
peerNP := netv1.NetworkPolicyPeer{}
|
||||
var ports []netv1.NetworkPolicyPort
|
||||
|
||||
if peer.ServiceSelector != nil {
|
||||
namespace := peer.ServiceSelector.Namespace
|
||||
name := peer.ServiceSelector.Name
|
||||
|
||||
return c.handlerPeerService(namespace, name, ingress)
|
||||
} else if peer.NamespaceSelector != nil {
|
||||
name := peer.NamespaceSelector.Name
|
||||
|
||||
peerNP.NamespaceSelector = new(metav1.LabelSelector)
|
||||
peerNP.NamespaceSelector.MatchLabels = make(map[string]string)
|
||||
peerNP.NamespaceSelector.MatchLabels[constants.NamespaceLabelKey] = name
|
||||
} else if peer.IPBlock != nil {
|
||||
peerNP.IPBlock = peer.IPBlock
|
||||
} else {
|
||||
klog.Errorf("Invalid nsnp peer %v\n", peer)
|
||||
return peerNP, nil, fmt.Errorf("Invalid nsnp peer %v\n", peer)
|
||||
}
|
||||
|
||||
return peerNP, ports, nil
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) convertToK8sNP(n *v1alpha1.NamespaceNetworkPolicy) (*netv1.NetworkPolicy, error) {
|
||||
np := &netv1.NetworkPolicy{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: network.NSNPPrefix + n.Name,
|
||||
Namespace: n.Namespace,
|
||||
},
|
||||
Spec: netv1.NetworkPolicySpec{
|
||||
PodSelector: metav1.LabelSelector{},
|
||||
PolicyTypes: make([]netv1.PolicyType, 0),
|
||||
},
|
||||
}
|
||||
|
||||
if n.Spec.Egress != nil {
|
||||
np.Spec.Egress = make([]netv1.NetworkPolicyEgressRule, 0)
|
||||
for _, egress := range n.Spec.Egress {
|
||||
tmpRule := netv1.NetworkPolicyEgressRule{}
|
||||
for _, peer := range egress.To {
|
||||
peer, ports, err := c.convertPeer(peer, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if ports != nil {
|
||||
np.Spec.Egress = append(np.Spec.Egress, netv1.NetworkPolicyEgressRule{
|
||||
Ports: ports,
|
||||
To: []netv1.NetworkPolicyPeer{peer},
|
||||
})
|
||||
continue
|
||||
}
|
||||
tmpRule.To = append(tmpRule.To, peer)
|
||||
}
|
||||
tmpRule.Ports = egress.Ports
|
||||
if tmpRule.To == nil {
|
||||
continue
|
||||
}
|
||||
np.Spec.Egress = append(np.Spec.Egress, tmpRule)
|
||||
}
|
||||
np.Spec.PolicyTypes = append(np.Spec.PolicyTypes, netv1.PolicyTypeEgress)
|
||||
}
|
||||
|
||||
if n.Spec.Ingress != nil {
|
||||
np.Spec.Ingress = make([]netv1.NetworkPolicyIngressRule, 0)
|
||||
for _, ingress := range n.Spec.Ingress {
|
||||
tmpRule := netv1.NetworkPolicyIngressRule{}
|
||||
for _, peer := range ingress.From {
|
||||
peer, ports, err := c.convertPeer(peer, true)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if ports != nil {
|
||||
np.Spec.Ingress = append(np.Spec.Ingress, netv1.NetworkPolicyIngressRule{
|
||||
Ports: ports,
|
||||
From: []netv1.NetworkPolicyPeer{peer},
|
||||
})
|
||||
}
|
||||
tmpRule.From = append(tmpRule.From, peer)
|
||||
}
|
||||
tmpRule.Ports = ingress.Ports
|
||||
np.Spec.Ingress = append(np.Spec.Ingress, tmpRule)
|
||||
}
|
||||
np.Spec.PolicyTypes = append(np.Spec.PolicyTypes, netv1.PolicyTypeIngress)
|
||||
}
|
||||
|
||||
return np, nil
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) generateNodeRule() (netv1.NetworkPolicyIngressRule, error) {
|
||||
var (
|
||||
rule netv1.NetworkPolicyIngressRule
|
||||
ips []string
|
||||
)
|
||||
|
||||
nodes, err := c.nodeInformer.Lister().List(labels.Everything())
|
||||
if err != nil {
|
||||
return rule, err
|
||||
}
|
||||
for _, node := range nodes {
|
||||
snatIPs := node.Annotations[NodeNSNPAnnotationKey]
|
||||
if snatIPs != "" {
|
||||
ips = append(ips, strings.Split(snatIPs, ";")...)
|
||||
}
|
||||
}
|
||||
|
||||
sort.Strings(ips)
|
||||
|
||||
for _, ip := range ips {
|
||||
cidr, err := stringToCIDR(ip)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
rule.From = append(rule.From, netv1.NetworkPolicyPeer{
|
||||
IPBlock: &netv1.IPBlock{
|
||||
CIDR: cidr,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
return rule, nil
|
||||
}
|
||||
|
||||
func generateNSNP(workspace string, namespace string, matchWorkspace bool) *netv1.NetworkPolicy {
|
||||
policy := &netv1.NetworkPolicy{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: AnnotationNPNAME,
|
||||
Namespace: namespace,
|
||||
},
|
||||
Spec: netv1.NetworkPolicySpec{
|
||||
PodSelector: metav1.LabelSelector{},
|
||||
PolicyTypes: make([]netv1.PolicyType, 0),
|
||||
Ingress: []netv1.NetworkPolicyIngressRule{{
|
||||
From: []netv1.NetworkPolicyPeer{{
|
||||
NamespaceSelector: &metav1.LabelSelector{
|
||||
MatchLabels: map[string]string{},
|
||||
},
|
||||
}},
|
||||
}},
|
||||
},
|
||||
}
|
||||
|
||||
policy.Spec.PolicyTypes = append(policy.Spec.PolicyTypes, netv1.PolicyTypeIngress)
|
||||
|
||||
if matchWorkspace {
|
||||
policy.Spec.Ingress[0].From[0].NamespaceSelector.MatchLabels[constants.WorkspaceLabelKey] = workspace
|
||||
} else {
|
||||
policy.Spec.Ingress[0].From[0].NamespaceSelector.MatchLabels[constants.NamespaceLabelKey] = namespace
|
||||
}
|
||||
|
||||
return policy
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) nsEnqueue(ns *corev1.Namespace) {
|
||||
key, err := cache.MetaNamespaceKeyFunc(ns)
|
||||
if err != nil {
|
||||
uruntime.HandleError(fmt.Errorf("Get namespace key %s failed", ns.Name))
|
||||
return
|
||||
}
|
||||
|
||||
workspaceName := ns.Labels[constants.WorkspaceLabelKey]
|
||||
if workspaceName == "" {
|
||||
return
|
||||
}
|
||||
|
||||
c.nsQueue.Add(key)
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) addWorkspace(newObj interface{}) {
|
||||
new := newObj.(*workspacev1alpha1.Workspace)
|
||||
|
||||
label := labels.SelectorFromSet(labels.Set{constants.WorkspaceLabelKey: new.Name})
|
||||
nsList, err := c.namespaceInformer.Lister().List(label)
|
||||
if err != nil {
|
||||
klog.Errorf("Error while list namespace by label %s", label.String())
|
||||
return
|
||||
}
|
||||
|
||||
for _, ns := range nsList {
|
||||
c.nsEnqueue(ns)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) addNode(newObj interface{}) {
|
||||
nsList, err := c.namespaceInformer.Lister().List(labels.Everything())
|
||||
if err != nil {
|
||||
klog.Errorf("Error while list namespace by label")
|
||||
return
|
||||
}
|
||||
|
||||
for _, ns := range nsList {
|
||||
c.nsEnqueue(ns)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) addNamespace(obj interface{}) {
|
||||
ns := obj.(*corev1.Namespace)
|
||||
|
||||
workspaceName := ns.Labels[constants.WorkspaceLabelKey]
|
||||
if workspaceName == "" {
|
||||
return
|
||||
}
|
||||
|
||||
c.nsEnqueue(ns)
|
||||
}
|
||||
|
||||
func isNetworkIsolateEnabled(ns *corev1.Namespace) bool {
|
||||
if ns.Annotations[NamespaceNPAnnotationKey] == NamespaceNPAnnotationEnabled {
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) syncNs(key string) error {
|
||||
_, name, err := cache.SplitMetaNamespaceKey(key)
|
||||
if err != nil {
|
||||
klog.Errorf("Not a valid controller key %s, %#v", key, err)
|
||||
return err
|
||||
}
|
||||
|
||||
ns, err := c.namespaceInformer.Lister().Get(name)
|
||||
if err != nil {
|
||||
// not found, possibly been deleted
|
||||
if errors.IsNotFound(err) {
|
||||
klog.V(2).Infof("Namespace %v has been deleted", key)
|
||||
return nil
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
workspaceName := ns.Labels[constants.WorkspaceLabelKey]
|
||||
if workspaceName == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
wksp, err := c.workspaceInformer.Lister().Get(workspaceName)
|
||||
if err != nil {
|
||||
//Should not be here
|
||||
if errors.IsNotFound(err) {
|
||||
klog.V(2).Infof("Workspace %v has been deleted", workspaceName)
|
||||
return nil
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
matchWorkspace := false
|
||||
delete := false
|
||||
nsnpList, _ := c.informer.Lister().NamespaceNetworkPolicies(ns.Name).List(labels.Everything())
|
||||
if isNetworkIsolateEnabled(ns) {
|
||||
matchWorkspace = false
|
||||
} else if wksp.Spec.NetworkIsolation {
|
||||
matchWorkspace = true
|
||||
//delete all namespace np when networkisolate not active
|
||||
if err != nil && len(nsnpList) > 0 {
|
||||
if c.ksclient.NamespaceNetworkPolicies(ns.Name).DeleteCollection(nil, typev1.ListOptions{}) != nil {
|
||||
klog.Errorf("Error when delete all nsnps in namespace %s", ns.Name)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
delete = true
|
||||
}
|
||||
|
||||
policy := generateNSNP(workspaceName, ns.Name, matchWorkspace)
|
||||
if shouldAddDNSRule(nsnpList) {
|
||||
ruleDNS, err := generateDNSRule([]string{DNSLocalIP})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
policy.Spec.Egress = append(policy.Spec.Egress, ruleDNS)
|
||||
ruleDNSService, err := c.generateDNSServiceRule()
|
||||
if err == nil {
|
||||
policy.Spec.Egress = append(policy.Spec.Egress, ruleDNSService)
|
||||
} else {
|
||||
klog.Warningf("Cannot handle service %s or %s", DNSServiceName, DNSServiceCoreDNS)
|
||||
}
|
||||
policy.Spec.PolicyTypes = append(policy.Spec.PolicyTypes, netv1.PolicyTypeEgress)
|
||||
}
|
||||
ruleNode, err := c.generateNodeRule()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(ruleNode.From) > 0 {
|
||||
policy.Spec.Ingress = append(policy.Spec.Ingress, ruleNode)
|
||||
}
|
||||
|
||||
if delete {
|
||||
c.provider.Delete(c.provider.GetKey(AnnotationNPNAME, ns.Name))
|
||||
} else {
|
||||
err = c.provider.Set(policy)
|
||||
if err != nil {
|
||||
klog.Errorf("Error while converting %#v to provider's network policy.", policy)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func shouldAddDNSRule(nsnpList []*v1alpha1.NamespaceNetworkPolicy) bool {
|
||||
for _, nsnp := range nsnpList {
|
||||
if len(nsnp.Spec.Egress) > 0 {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) nsWorker() {
|
||||
for c.processNsWorkItem() {
|
||||
}
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) processNsWorkItem() bool {
|
||||
key, quit := c.nsQueue.Get()
|
||||
if quit {
|
||||
return false
|
||||
}
|
||||
defer c.nsQueue.Done(key)
|
||||
|
||||
if err := c.syncNs(key.(string)); err != nil {
|
||||
klog.Errorf("Error when syncns %s", err)
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) nsnpEnqueue(obj interface{}) {
|
||||
nsnp := obj.(*v1alpha1.NamespaceNetworkPolicy)
|
||||
|
||||
key, err := cache.MetaNamespaceKeyFunc(nsnp)
|
||||
if err != nil {
|
||||
uruntime.HandleError(fmt.Errorf("get namespace network policy key %s failed", nsnp.Name))
|
||||
return
|
||||
}
|
||||
|
||||
c.nsnpQueue.Add(key)
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) syncNSNP(key string) error {
|
||||
namespace, name, err := cache.SplitMetaNamespaceKey(key)
|
||||
if err != nil {
|
||||
klog.Errorf("Not a valid controller key %s, %#v", key, err)
|
||||
return err
|
||||
}
|
||||
|
||||
c.nsQueue.Add(namespace)
|
||||
|
||||
nsnp, err := c.informer.Lister().NamespaceNetworkPolicies(namespace).Get(name)
|
||||
if err != nil {
|
||||
if errors.IsNotFound(err) {
|
||||
klog.V(4).Infof("NSNP %v has been deleted", key)
|
||||
c.provider.Delete(c.provider.GetKey(network.NSNPPrefix+name, namespace))
|
||||
return nil
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
np, err := c.convertToK8sNP(nsnp)
|
||||
if err != nil {
|
||||
klog.Errorf("Error while convert nsnp to k8snp: %s", err)
|
||||
return err
|
||||
}
|
||||
err = c.provider.Set(np)
|
||||
if err != nil {
|
||||
klog.Errorf("Error while set provider: %s", err)
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) nsNPWorker() {
|
||||
for c.processNSNPWorkItem() {
|
||||
}
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) processNSNPWorkItem() bool {
|
||||
key, quit := c.nsnpQueue.Get()
|
||||
if quit {
|
||||
return false
|
||||
}
|
||||
defer c.nsnpQueue.Done(key)
|
||||
|
||||
c.syncNSNP(key.(string))
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// NewnamespacenpController returns a controller which manages NSNSP objects.
|
||||
func NewNSNetworkPolicyController(
|
||||
client kubernetes.Interface,
|
||||
ksclient ksnetclient.NetworkV1alpha1Interface,
|
||||
nsnpInformer nspolicy.NamespaceNetworkPolicyInformer,
|
||||
serviceInformer v1.ServiceInformer,
|
||||
nodeInformer v1.NodeInformer,
|
||||
workspaceInformer workspace.WorkspaceInformer,
|
||||
namespaceInformer v1.NamespaceInformer,
|
||||
policyProvider provider.NsNetworkPolicyProvider) *NSNetworkPolicyController {
|
||||
|
||||
controller := &NSNetworkPolicyController{
|
||||
client: client,
|
||||
ksclient: ksclient,
|
||||
informer: nsnpInformer,
|
||||
informerSynced: nsnpInformer.Informer().HasSynced,
|
||||
serviceInformer: serviceInformer,
|
||||
serviceInformerSynced: serviceInformer.Informer().HasSynced,
|
||||
nodeInformer: nodeInformer,
|
||||
nodeInformerSynced: nodeInformer.Informer().HasSynced,
|
||||
workspaceInformer: workspaceInformer,
|
||||
workspaceInformerSynced: workspaceInformer.Informer().HasSynced,
|
||||
namespaceInformer: namespaceInformer,
|
||||
namespaceInformerSynced: namespaceInformer.Informer().HasSynced,
|
||||
provider: policyProvider,
|
||||
nsQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "namespace"),
|
||||
nsnpQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "namespacenp"),
|
||||
}
|
||||
|
||||
workspaceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: controller.addWorkspace,
|
||||
UpdateFunc: func(oldObj, newObj interface{}) {
|
||||
old := oldObj.(*workspacev1alpha1.Workspace)
|
||||
new := newObj.(*workspacev1alpha1.Workspace)
|
||||
if old.Spec.NetworkIsolation == new.Spec.NetworkIsolation {
|
||||
return
|
||||
}
|
||||
controller.addWorkspace(newObj)
|
||||
},
|
||||
})
|
||||
|
||||
nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: controller.addNode,
|
||||
UpdateFunc: func(oldObj, newObj interface{}) {
|
||||
old := oldObj.(*corev1.Node)
|
||||
new := newObj.(*corev1.Node)
|
||||
if old.Annotations[NodeNSNPAnnotationKey] == new.Annotations[NodeNSNPAnnotationKey] {
|
||||
return
|
||||
}
|
||||
controller.addNode(newObj)
|
||||
},
|
||||
})
|
||||
|
||||
namespaceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: controller.addNamespace,
|
||||
UpdateFunc: func(oldObj interface{}, newObj interface{}) {
|
||||
old := oldObj.(*corev1.Namespace)
|
||||
new := newObj.(*corev1.Namespace)
|
||||
if old.Annotations[NamespaceNPAnnotationKey] == new.Annotations[NamespaceNPAnnotationKey] {
|
||||
return
|
||||
}
|
||||
controller.addNamespace(newObj)
|
||||
},
|
||||
})
|
||||
|
||||
nsnpInformer.Informer().AddEventHandlerWithResyncPeriod(cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: func(obj interface{}) {
|
||||
controller.nsnpEnqueue(obj)
|
||||
},
|
||||
UpdateFunc: func(oldObj interface{}, newObj interface{}) {
|
||||
controller.nsnpEnqueue(newObj)
|
||||
},
|
||||
DeleteFunc: func(obj interface{}) {
|
||||
controller.nsnpEnqueue(obj)
|
||||
},
|
||||
}, defaultSleepDuration)
|
||||
|
||||
return controller
|
||||
}
|
||||
|
||||
func (c *NSNetworkPolicyController) Start(stopCh <-chan struct{}) error {
|
||||
return c.Run(defaultThread, defaultSync, stopCh)
|
||||
}
|
||||
|
||||
// Run starts the controller.
|
||||
func (c *NSNetworkPolicyController) Run(threadiness int, reconcilerPeriod string, stopCh <-chan struct{}) error {
|
||||
defer uruntime.HandleCrash()
|
||||
|
||||
defer c.nsQueue.ShutDown()
|
||||
defer c.nsnpQueue.ShutDown()
|
||||
|
||||
klog.Info("Waiting to sync with Kubernetes API (NSNP)")
|
||||
if ok := cache.WaitForCacheSync(stopCh, c.informerSynced, c.serviceInformerSynced, c.workspaceInformerSynced, c.namespaceInformerSynced, c.nodeInformerSynced); !ok {
|
||||
return fmt.Errorf("Failed to wait for caches to sync")
|
||||
}
|
||||
klog.Info("Finished syncing with Kubernetes API (NSNP)")
|
||||
|
||||
// Start a number of worker threads to read from the queue. Each worker
|
||||
// will pull keys off the resource cache event queue and sync them to the
|
||||
// K8s datastore.
|
||||
for i := 0; i < threadiness; i++ {
|
||||
go wait.Until(c.nsWorker, time.Second, stopCh)
|
||||
go wait.Until(c.nsNPWorker, time.Second, stopCh)
|
||||
}
|
||||
|
||||
//Work to sync K8s NetworkPolicy
|
||||
go c.provider.Start(stopCh)
|
||||
|
||||
klog.Info("NSNP controller is now running")
|
||||
<-stopCh
|
||||
klog.Info("Stopping NSNP controller")
|
||||
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user