diff --git a/npc/analyser.go b/npc/analyser.go index 9e70b33a..ccfce02a 100644 --- a/npc/analyser.go +++ b/npc/analyser.go @@ -12,12 +12,14 @@ import ( func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) ( rules map[string]*ruleSpec, - nsSelectors, podSelectors map[string]*selectorSpec, + nsSelectors, podSelectors, namespacedPodSelectors map[string]*selectorSpec, ipBlocks map[string]*ipBlockSpec, err error) { nsSelectors = make(map[string]*selectorSpec) podSelectors = make(map[string]*selectorSpec) + namespacedPodSelectors = make(map[string]*selectorSpec) + ipBlocks = make(map[string]*ipBlockSpec) rules = make(map[string]*ruleSpec) policyTypes := make([]policyType, 0) @@ -33,7 +35,7 @@ func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) ( // If empty, matches all pods in a namespace targetSelector, err := newSelectorSpec(&policy.Spec.PodSelector, nil, policyTypes, ns.name, ipset.HashIP) if err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, nil, err } // To prevent targetSelector being overwritten by a subsequent selector with @@ -52,7 +54,7 @@ func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) ( if !allPorts { if err = checkForNamedPorts(ingressRule.Ports); err != nil { - return nil, nil, nil, nil, fmt.Errorf("named ports in network policies is not supported yet. "+ + return nil, nil, nil, nil, nil, fmt.Errorf("named ports in network policies is not supported yet. "+ "Rejecting network policy: %s from further processing. "+err.Error(), policy.Name) } } @@ -73,23 +75,23 @@ func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) ( // NetworkPolicyPeer describes a peer to allow traffic from. if peer.PodSelector != nil && peer.NamespaceSelector != nil { - srcSelector, err = newSelectorSpec(peer.PodSelector, peer.NamespaceSelector, nil, ns.name, ipset.HashIP) + srcSelector, err = newSelectorSpec(peer.PodSelector, peer.NamespaceSelector, nil, "", ipset.HashIP) if err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, nil, err } - addIfNotExist(srcSelector, podSelectors) + addIfNotExist(srcSelector, namespacedPodSelectors) srcRuleHost = srcSelector } else if peer.PodSelector != nil { srcSelector, err = newSelectorSpec(peer.PodSelector, nil, nil, ns.name, ipset.HashIP) if err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, nil, err } addIfNotExist(srcSelector, podSelectors) srcRuleHost = srcSelector } else if peer.NamespaceSelector != nil { srcSelector, err = newSelectorSpec(nil, peer.NamespaceSelector, nil, "", ipset.ListSet) if err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, nil, err } nsSelectors[srcSelector.key] = srcSelector srcRuleHost = srcSelector @@ -123,7 +125,7 @@ func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) ( if !allPorts { if err = checkForNamedPorts(egressRule.Ports); err != nil { - return nil, nil, nil, nil, fmt.Errorf("named ports in network policies is not supported yet. "+ + return nil, nil, nil, nil, nil, fmt.Errorf("named ports in network policies is not supported yet. "+ "Rejecting network policy: %s from further processing. "+err.Error(), policy.Name) } } @@ -144,24 +146,24 @@ func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) ( // NetworkPolicyPeer describes a peer to allow traffic to. if peer.PodSelector != nil && peer.NamespaceSelector != nil { - dstSelector, err = newSelectorSpec(peer.PodSelector, peer.NamespaceSelector, nil, ns.name, ipset.HashIP) + dstSelector, err = newSelectorSpec(peer.PodSelector, peer.NamespaceSelector, nil, "", ipset.HashIP) if err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, nil, err } - addIfNotExist(dstSelector, podSelectors) + addIfNotExist(dstSelector, namespacedPodSelectors) dstRuleHost = dstSelector } else if peer.PodSelector != nil { dstSelector, err = newSelectorSpec(peer.PodSelector, nil, nil, ns.name, ipset.HashIP) if err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, nil, err } addIfNotExist(dstSelector, podSelectors) dstRuleHost = dstSelector } else if peer.NamespaceSelector != nil { - dstSelector, err = newSelectorSpec(peer.NamespaceSelector, nil, nil, "", ipset.ListSet) + dstSelector, err = newSelectorSpec(nil, peer.NamespaceSelector, nil, "", ipset.ListSet) if err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, nil, err } nsSelectors[dstSelector.key] = dstSelector dstRuleHost = dstSelector @@ -185,7 +187,7 @@ func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) ( } } - return rules, nsSelectors, podSelectors, ipBlocks, nil + return rules, nsSelectors, podSelectors, namespacedPodSelectors, ipBlocks, nil } func addIfNotExist(s *selectorSpec, ss map[string]*selectorSpec) { diff --git a/npc/controller.go b/npc/controller.go index 7c625d8b..45c34541 100644 --- a/npc/controller.go +++ b/npc/controller.go @@ -36,9 +36,10 @@ type controller struct { ipt iptables.Interface ips ipset.Interface - nss map[string]*ns // ns name -> ns struct - nsSelectors *selectorSet // selector string -> nsSelector - defaultEgressDrop bool // flag to track if base iptable rule to drop egress traffic is added or not + nss map[string]*ns // ns name -> ns struct + nsSelectors *selectorSet // selector string -> nsSelector + namespacedPodSelectors *selectorSet + defaultEgressDrop bool // flag to track if base iptable rule to drop egress traffic is added or not } func New(nodeName string, ipt iptables.Interface, ips ipset.Interface) NetworkPolicyController { @@ -50,7 +51,7 @@ func New(nodeName string, ipt iptables.Interface, ips ipset.Interface) NetworkPo doNothing := func(*selector, policyType) error { return nil } c.nsSelectors = newSelectorSet(ips, c.onNewNsSelector, doNothing, doNothing) - + c.namespacedPodSelectors = newSelectorSet(ips, c.onNewNamespacePodsSelector, doNothing, doNothing) return c } @@ -67,10 +68,28 @@ func (npc *controller) onNewNsSelector(selector *selector) error { return nil } +func (npc *controller) onNewNamespacePodsSelector(selector *selector) error { + for _, ns := range npc.nss { + if ns.namespace != nil && len(ns.pods) > 0 { + for _, pod := range ns.pods { + if hasIP(pod) { + if selector.matches(&pod.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels) { + if err := selector.addEntry(pod.ObjectMeta.UID, pod.Status.PodIP, podComment(pod)); err != nil { + return err + } + + } + } + } + } + } + return nil +} + func (npc *controller) withNS(name string, f func(ns *ns) error) error { ns, found := npc.nss[name] if !found { - newNs, err := newNS(name, npc.nodeName, npc.ipt, npc.ips, npc.nsSelectors) + newNs, err := newNS(name, npc.nodeName, npc.ipt, npc.ips, npc.nsSelectors, npc.namespacedPodSelectors) if err != nil { return err } diff --git a/npc/namespace.go b/npc/namespace.go index 1e51734d..6959fcd0 100644 --- a/npc/namespace.go +++ b/npc/namespace.go @@ -10,6 +10,8 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/uuid" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" "github.com/weaveworks/weave/common" "github.com/weaveworks/weave/net/ipset" @@ -37,30 +39,48 @@ type ns struct { ingressDefaultAllowIPSet ipset.Name egressDefaultAllowIPSet ipset.Name - nsSelectors *selectorSet - podSelectors *selectorSet - ipBlocks *ipBlockSet - rules *ruleSet + nsSelectors *selectorSet // reference to global selectorSet that is shared across the `ns`. Used to represent all pods in the matching namespaces + podSelectors *selectorSet // used to represent the matching pods in namespace respresented by this `ns` + namespacedPodsSelectors *selectorSet // reference to global selectorSet that is shared across the `ns`. Used to represent matching pods in matching namespace + ipBlocks *ipBlockSet + rules *ruleSet } -func newNS(name, nodeName string, ipt iptables.Interface, ips ipset.Interface, nsSelectors *selectorSet) (*ns, error) { +func newNS(name, nodeName string, ipt iptables.Interface, ips ipset.Interface, nsSelectors *selectorSet, namespacedPodsSelectors *selectorSet) (*ns, error) { allPods, err := newSelectorSpec(&metav1.LabelSelector{}, nil, nil, name, ipset.HashIP) if err != nil { return nil, err } + config, err := rest.InClusterConfig() + if err != nil { + return nil, err + } + + client, err := kubernetes.NewForConfig(config) + if err != nil { + return nil, err + } + + namespace, err := client.CoreV1().Namespaces().Get(name, metav1.GetOptions{}) + if err != nil { + return nil, err + } + ns := &ns{ - ipt: ipt, - ips: ips, - name: name, - nodeName: nodeName, - pods: make(map[types.UID]*coreapi.Pod), - policies: make(map[types.UID]interface{}), - uid: uuid.NewUUID(), - allPods: allPods, - nsSelectors: nsSelectors, - ipBlocks: newIPBlockSet(ips), - rules: newRuleSet(ipt), + ipt: ipt, + ips: ips, + name: name, + namespace: namespace, + nodeName: nodeName, + pods: make(map[types.UID]*coreapi.Pod), + policies: make(map[types.UID]interface{}), + uid: uuid.NewUUID(), + allPods: allPods, + nsSelectors: nsSelectors, + namespacedPodsSelectors: namespacedPodsSelectors, + ipBlocks: newIPBlockSet(ips), + rules: newRuleSet(ipt), } ns.podSelectors = newSelectorSet(ips, ns.onNewPodSelector, ns.onNewTargetPodSelector, ns.onDestroyTargetPodSelector) @@ -185,6 +205,11 @@ func (ns *ns) addPod(obj *coreapi.Pod) error { } } + _, _, err = ns.namespacedPodsSelectors.addToMatching(obj.ObjectMeta.UID, &obj.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels, obj.Status.PodIP, podComment(obj)) + if err != nil { + return err + } + return nil } @@ -203,6 +228,9 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error { if err := ns.ips.DelEntry(oldObj.ObjectMeta.UID, ns.egressDefaultAllowIPSet, oldObj.Status.PodIP); err != nil { return err } + if err := ns.namespacedPodsSelectors.delFromMatching(oldObj.ObjectMeta.UID, &oldObj.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels, oldObj.Status.PodIP); err != nil { + return err + } return ns.podSelectors.delFromMatching(oldObj.ObjectMeta.UID, &oldObj.ObjectMeta.Labels, nil, oldObj.Status.PodIP) } @@ -223,6 +251,10 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error { return err } } + _, _, err = ns.namespacedPodsSelectors.addToMatching(newObj.ObjectMeta.UID, &newObj.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels, newObj.Status.PodIP, podComment(newObj)) + if err != nil { + return err + } return nil } @@ -263,6 +295,23 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error { return err } } + for _, ps := range ns.namespacedPodsSelectors.entries { + oldMatch := ps.matches(&oldObj.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels) + newMatch := ps.matches(&newObj.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels) + if oldMatch == newMatch && oldObj.Status.PodIP == newObj.Status.PodIP { + continue + } + if oldMatch { + if err := ps.delEntry(oldObj.ObjectMeta.UID, oldObj.Status.PodIP); err != nil { + return err + } + } + if newMatch { + if err := ps.addEntry(newObj.ObjectMeta.UID, newObj.Status.PodIP, podComment(newObj)); err != nil { + return err + } + } + } } return nil @@ -298,14 +347,19 @@ func (ns *ns) deletePod(obj *coreapi.Pod) error { if err := ns.ips.DelEntry(obj.ObjectMeta.UID, ns.egressDefaultAllowIPSet, obj.Status.PodIP); err != nil { return err } - - return ns.podSelectors.delFromMatching(obj.ObjectMeta.UID, &obj.ObjectMeta.Labels, nil, obj.Status.PodIP) + if err := ns.podSelectors.delFromMatching(obj.ObjectMeta.UID, &obj.ObjectMeta.Labels, nil, obj.Status.PodIP); err != nil { + return err + } + if err := ns.namespacedPodsSelectors.delFromMatching(obj.ObjectMeta.UID, &obj.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels, obj.Status.PodIP); err != nil { + return err + } + return nil } func (ns *ns) addNetworkPolicy(obj interface{}) error { // Analyse policy, determine which rules and ipsets are required - uid, rules, nsSelectors, podSelectors, ipBlocks, err := ns.analyse(obj) + uid, rules, nsSelectors, podSelectors, namespacedPodsSelectors, ipBlocks, err := ns.analyse(obj) if err != nil { return err } @@ -317,6 +371,9 @@ func (ns *ns) addNetworkPolicy(obj interface{}) error { if err := ns.podSelectors.provision(uid, nil, podSelectors); err != nil { return err } + if err := ns.namespacedPodsSelectors.provision(uid, nil, namespacedPodsSelectors); err != nil { + return err + } if err := ns.ipBlocks.provision(uid, nil, ipBlocks); err != nil { return err } @@ -325,11 +382,11 @@ func (ns *ns) addNetworkPolicy(obj interface{}) error { func (ns *ns) updateNetworkPolicy(oldObj, newObj interface{}) error { // Analyse the old and the new policy so we can determine differences - oldUID, oldRules, oldNsSelectors, oldPodSelectors, oldIPBlocks, err := ns.analyse(oldObj) + oldUID, oldRules, oldNsSelectors, oldPodSelectors, oldNamespacedPodsSelectors, oldIPBlocks, err := ns.analyse(oldObj) if err != nil { return err } - newUID, newRules, newNsSelectors, newPodSelectors, newIPBlocks, err := ns.analyse(newObj) + newUID, newRules, newNsSelectors, newPodSelectors, newNamespacedPodsSelectors, newIPBlocks, err := ns.analyse(newObj) if err != nil { return err } @@ -347,6 +404,9 @@ func (ns *ns) updateNetworkPolicy(oldObj, newObj interface{}) error { if err := ns.podSelectors.deprovision(oldUID, oldPodSelectors, newPodSelectors); err != nil { return err } + if err := ns.namespacedPodsSelectors.deprovision(oldUID, oldNamespacedPodsSelectors, newNamespacedPodsSelectors); err != nil { + return err + } if err := ns.ipBlocks.deprovision(oldUID, oldIPBlocks, newIPBlocks); err != nil { return err } @@ -356,6 +416,9 @@ func (ns *ns) updateNetworkPolicy(oldObj, newObj interface{}) error { if err := ns.podSelectors.provision(oldUID, oldPodSelectors, newPodSelectors); err != nil { return err } + if err := ns.namespacedPodsSelectors.provision(oldUID, oldNamespacedPodsSelectors, newNamespacedPodsSelectors); err != nil { + return err + } if err := ns.ipBlocks.provision(oldUID, oldIPBlocks, newIPBlocks); err != nil { return err } @@ -364,7 +427,7 @@ func (ns *ns) updateNetworkPolicy(oldObj, newObj interface{}) error { func (ns *ns) deleteNetworkPolicy(obj interface{}) error { // Analyse network policy to free resources - uid, rules, nsSelectors, podSelectors, ipBlocks, err := ns.analyse(obj) + uid, rules, nsSelectors, podSelectors, namespacedPodsSelectors, ipBlocks, err := ns.analyse(obj) if err != nil { return err } @@ -381,6 +444,9 @@ func (ns *ns) deleteNetworkPolicy(obj interface{}) error { if err := ns.podSelectors.deprovision(uid, podSelectors, nil); err != nil { return err } + if err := ns.namespacedPodsSelectors.deprovision(uid, namespacedPodsSelectors, nil); err != nil { + return err + } if err := ns.ipBlocks.deprovision(uid, ipBlocks, nil); err != nil { return err } @@ -546,7 +612,7 @@ func (ns *ns) defaultAllowIPSetName(pt policyType) ipset.Name { func (ns *ns) analyse(obj interface{}) ( uid types.UID, rules map[string]*ruleSpec, - nsSelectors, podSelectors map[string]*selectorSpec, + nsSelectors, podSelectors, namespacedPodsSelectors map[string]*selectorSpec, ipBlocks map[string]*ipBlockSpec, err error) { @@ -562,7 +628,7 @@ func (ns *ns) analyse(obj interface{}) ( ns.policies[uid] = obj // Analyse policy, determine which rules and ipsets are required - rules, nsSelectors, podSelectors, ipBlocks, err = ns.analysePolicy(obj.(*networkingv1.NetworkPolicy)) + rules, nsSelectors, podSelectors, namespacedPodsSelectors, ipBlocks, err = ns.analysePolicy(obj.(*networkingv1.NetworkPolicy)) if err != nil { return }