reweave/weave
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3028 from weaveworks/issues/3025-track-local-pods-separately

Browse Source 
weave-npc: stop ignoring local pods
This commit is contained in:
Bryan Boreham authored and GitHub committed 2017-06-23 10:39:01 +01:00
2 parents ac0c0db942 ced84955d3
commit 3cc537b978
4 files changed
+54 -27
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

npc/controller.go
+1 -19
View File
@@ -66,7 +66,7 @@ func (npc *controller) onNewNsSelector(selector *selector) error {
func (npc *controller) withNS(name string, f func(ns *ns) error) error {
ns, found := npc.nss[name]
if !found {
newNs, err := newNS(name, npc.ipt, npc.ips, npc.nsSelectors)
newNs, err := newNS(name, npc.nodeName, npc.ipt, npc.ips, npc.nsSelectors)
if err != nil {
return err
}
@@ -85,21 +85,10 @@ func (npc *controller) withNS(name string, f func(ns *ns) error) error {
return nil
}
func (npc *controller) checkLocalPod(obj *coreapi.Pod, event string) bool {
if obj.Spec.NodeName != npc.nodeName {
common.Log.Debugf("%s ignored for pod %s/%s on node %s", event, obj.ObjectMeta.Namespace, obj.ObjectMeta.Name, obj.Spec.NodeName)
return false
}
return true
}
func (npc *controller) AddPod(obj *coreapi.Pod) error {
npc.Lock()
defer npc.Unlock()
if !npc.checkLocalPod(obj, "AddPod") {
return nil
}
common.Log.Debugf("EVENT AddPod %s", js(obj))
return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error {
return errors.Wrap(ns.addPod(obj), "add pod")
@@ -110,10 +99,6 @@ func (npc *controller) UpdatePod(oldObj, newObj *coreapi.Pod) error {
npc.Lock()
defer npc.Unlock()
// Pods don't move across nodes so we only check the new version
if !npc.checkLocalPod(newObj, "UpdatePod") {
return nil
}
common.Log.Debugf("EVENT UpdatePod %s %s", js(oldObj), js(newObj))
return npc.withNS(oldObj.ObjectMeta.Namespace, func(ns *ns) error {
return errors.Wrap(ns.updatePod(oldObj, newObj), "update pod")
@@ -124,9 +109,6 @@ func (npc *controller) DeletePod(obj *coreapi.Pod) error {
npc.Lock()
defer npc.Unlock()
if !npc.checkLocalPod(obj, "DeletePod") {
return nil
}
common.Log.Debugf("EVENT DeletePod %s", js(obj))
return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error {
return errors.Wrap(ns.deletePod(obj), "delete pod")
npc/namespace.go
+25 -5
View File
@@ -19,6 +19,7 @@ type ns struct {
ips ipset.Interface // interface to ipset
name string // k8s Namespace name
nodeName string // my node name
namespace *coreapi.Namespace // k8s Namespace object
pods map[types.UID]*coreapi.Pod // k8s Pod objects by UID
policies map[types.UID]*extnapi.NetworkPolicy // k8s NetworkPolicy objects by UID
@@ -31,7 +32,7 @@ type ns struct {
rules *ruleSet
}
func newNS(name string, ipt *iptables.IPTables, ips ipset.Interface, nsSelectors *selectorSet) (*ns, error) {
func newNS(name, nodeName string, ipt *iptables.IPTables, ips ipset.Interface, nsSelectors *selectorSet) (*ns, error) {
allPods, err := newSelectorSpec(&unversioned.LabelSelector{}, name, ipset.HashIP)
if err != nil {
return nil, err
@@ -41,6 +42,7 @@ func newNS(name string, ipt *iptables.IPTables, ips ipset.Interface, nsSelectors
ipt: ipt,
ips: ips,
name: name,
nodeName: nodeName,
pods: make(map[types.UID]*coreapi.Pod),
policies: make(map[types.UID]*extnapi.NetworkPolicy),
uid: uuid.NewUUID(),
@@ -81,6 +83,13 @@ func (ns *ns) onNewPodSelector(selector *selector) error {
return nil
}
func (ns *ns) checkLocalPod(obj *coreapi.Pod) bool {
if obj.Spec.NodeName != ns.nodeName {
return false
}
return true
}
func (ns *ns) addPod(obj *coreapi.Pod) error {
ns.pods[obj.ObjectMeta.UID] = obj
@@ -88,7 +97,9 @@ func (ns *ns) addPod(obj *coreapi.Pod) error {
return nil
}
ns.ips.AddEntry(LocalIpset, obj.Status.PodIP)
if ns.checkLocalPod(obj) {
ns.ips.AddEntry(LocalIpset, obj.Status.PodIP)
}
return ns.podSelectors.addToMatching(obj.ObjectMeta.Labels, obj.Status.PodIP)
}
@@ -101,12 +112,16 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error {
}
if hasIP(oldObj) && !hasIP(newObj) {
ns.ips.DelEntry(LocalIpset, oldObj.Status.PodIP)
if ns.checkLocalPod(oldObj) {
ns.ips.DelEntry(LocalIpset, oldObj.Status.PodIP)
}
return ns.podSelectors.delFromMatching(oldObj.ObjectMeta.Labels, oldObj.Status.PodIP)
}
if !hasIP(oldObj) && hasIP(newObj) {
ns.ips.AddEntry(LocalIpset, newObj.Status.PodIP)
if ns.checkLocalPod(newObj) {
ns.ips.AddEntry(LocalIpset, newObj.Status.PodIP)
}
return ns.podSelectors.addToMatching(newObj.ObjectMeta.Labels, newObj.Status.PodIP)
}
@@ -142,7 +157,9 @@ func (ns *ns) deletePod(obj *coreapi.Pod) error {
return nil
}
ns.ips.DelEntry(LocalIpset, obj.Status.PodIP)
if ns.checkLocalPod(obj) {
ns.ips.DelEntry(LocalIpset, obj.Status.PodIP)
}
return ns.podSelectors.delFromMatching(obj.ObjectMeta.Labels, obj.Status.PodIP)
}
@@ -234,10 +251,12 @@ func bypassRule(nsIpsetName ipset.Name) []string {
}
func (ns *ns) ensureBypassRule(nsIpsetName ipset.Name) error {
common.Log.Debugf("ensuring rule for DefaultAllow in namespace: %s, set %s", ns.name, nsIpsetName)
return ns.ipt.Append(TableFilter, DefaultChain, bypassRule(ns.allPods.ipsetName)...)
}
func (ns *ns) deleteBypassRule(nsIpsetName ipset.Name) error {
common.Log.Debugf("removing default rule in namespace: %s, set %s", ns.name, nsIpsetName)
return ns.ipt.Delete(TableFilter, DefaultChain, bypassRule(ns.allPods.ipsetName)...)
}
@@ -261,6 +280,7 @@ func (ns *ns) updateNamespace(oldObj, newObj *coreapi.Namespace) error {
newDefaultDeny := isDefaultDeny(newObj)
if oldDefaultDeny != newDefaultDeny {
common.Log.Infof("namespace DefaultDeny changed from %t to %t", oldDefaultDeny, newDefaultDeny)
if oldDefaultDeny {
return ns.ensureBypassRule(ns.allPods.ipsetName)
}
test/840_weave_kube_3_test.sh
+27 -3
View File
@@ -70,6 +70,30 @@ assert "run_on $HOST1 ps aux | grep -c '[d]efunct'" "0"
assert "run_on $HOST2 ps aux | grep -c '[d]efunct'" "0"
assert "run_on $HOST3 ps aux | grep -c '[d]efunct'" "0"
# Set up a simple network policy so all our test pods can talk to each other
run_on $HOST1 "$KUBECTL annotate ns default net.beta.kubernetes.io/network-policy='{\"ingress\":{\"isolation\":\"DefaultDeny\"}}'"
run_on $HOST1 "$KUBECTL apply -f -" <<EOF
apiVersion: extensions/v1beta1
kind: NetworkPolicy
metadata:
name: test840
spec:
ingress:
- from:
- podSelector:
matchExpressions:
- key: run
operator: In
values:
- nettest
ports:
- port: 8080
protocol: TCP
podSelector:
matchLabels:
run: nettest
EOF
check_ready() {
run_on $HOST1 "$KUBECTL get nodes | grep -c -w Ready | grep $NUM_HOSTS"
}
@@ -94,10 +118,11 @@ spec:
run: nettest
EOF
podName=$($SSH $HOST1 "$KUBECTL get pods -l run=nettest -o go-template='{{(index .items 0).metadata.name}}'")
check_all_pods_communicate() {
podIP=$($SSH $HOST1 "$KUBECTL get pods -l run=nettest -o go-template='{{(index .items 0).status.podIP}}' 2>/dev/null")
if [ -n podIP ] ; then
status=$($SSH $HOST1 "curl -s -S http://$podIP:8080/status")
status=$($SSH $HOST1 "$KUBECTL exec $podName -- curl -s -S http://127.0.0.1:8080/status")
if [ $status = "pass" ] ; then
return 0
fi
@@ -108,7 +133,6 @@ check_all_pods_communicate() {
assert_raises 'wait_for_x check_all_pods_communicate pods'
# Check that a pod can contact the outside world
podName=$($SSH $HOST1 "$KUBECTL get pods -l run=nettest -o go-template='{{(index .items 0).metadata.name}}'")
assert_raises "$SSH $HOST1 $KUBECTL exec $podName -- $PING 8.8.8.8"
tear_down_kubeadm
test/images/network-tester/Dockerfile
+1
View File
@@ -13,6 +13,7 @@
# limitations under the License.
FROM bash
RUN apk add --no-cache curl
ADD webserver webserver
EXPOSE 8080
ENTRYPOINT ["/webserver"]