Merge pull request weaveworks#3057 from weaveworks/wip-npc-race-test

Fix weave-npc race condition

npc/controller.go

@@ -3,13 +3,13 @@ package npc
 import (
 	"sync"
 
-	"github.com/coreos/go-iptables/iptables"
 	"github.com/pkg/errors"
 	coreapi "k8s.io/client-go/pkg/api/v1"
 	extnapi "k8s.io/client-go/pkg/apis/extensions/v1beta1"
 
 	"github.com/weaveworks/weave/common"
 	"github.com/weaveworks/weave/npc/ipset"
+	"github.com/weaveworks/weave/npc/iptables"
 )
 
 type NetworkPolicyController interface {
@@ -31,14 +31,14 @@ type controller struct {
 
 	nodeName string // my node name
 
-	ipt *iptables.IPTables
+	ipt iptables.Interface
 	ips ipset.Interface
 
 	nss         map[string]*ns // ns name -> ns struct
 	nsSelectors *selectorSet   // selector string -> nsSelector
 }
 
-func New(nodeName string, ipt *iptables.IPTables, ips ipset.Interface) NetworkPolicyController {
+func New(nodeName string, ipt iptables.Interface, ips ipset.Interface) NetworkPolicyController {
 	c := &controller{
 		nodeName: nodeName,
 		ipt:      ipt,

npc/controller_test.go

@@ -0,0 +1,182 @@
+package npc
+
+import (
+	"log"
+	"testing"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
+	"github.com/weaveworks/weave/npc/ipset"
+	unversionedapi "k8s.io/client-go/pkg/api/unversioned"
+	coreapi "k8s.io/client-go/pkg/api/v1"
+	extnapi "k8s.io/client-go/pkg/apis/extensions/v1beta1"
+	"k8s.io/client-go/pkg/util/intstr"
+)
+
+type mockSet struct {
+	name    ipset.Name
+	setType ipset.Type
+	subSets map[string]bool
+}
+
+type mockIPSet struct {
+	sets map[string]mockSet
+}
+
+func newMockIPSet() mockIPSet {
+	i := mockIPSet{
+		sets: make(map[string]mockSet),
+	}
+
+	return i
+}
+
+func (i *mockIPSet) Create(ipsetName ipset.Name, ipsetType ipset.Type) error {
+	if _, ok := i.sets[string(ipsetName)]; ok {
+		return errors.Errorf("ipset %s already exists", ipsetName)
+	}
+	i.sets[string(ipsetName)] = mockSet{name: ipsetName, setType: ipsetType, subSets: make(map[string]bool)}
+	return nil
+}
+
+func (i *mockIPSet) AddEntry(ipsetName ipset.Name, entry string) error {
+	log.Printf("adding entry %s to %s", entry, ipsetName)
+	if _, ok := i.sets[entry]; !ok {
+		return errors.Errorf("ipset %s does not exist", entry)
+	}
+	if _, ok := i.sets[string(ipsetName)].subSets[entry]; ok {
+		return errors.Errorf("ipset %s is already a member of %s", entry, ipsetName)
+	}
+	i.sets[string(ipsetName)].subSets[entry] = true
+
+	return nil
+}
+
+func (i *mockIPSet) DelEntry(ipsetName ipset.Name, entry string) error {
+	log.Printf("deleting entry %s from %s", entry, ipsetName)
+	if _, ok := i.sets[string(ipsetName)]; !ok {
+		return errors.Errorf("ipset %s does not exist", ipsetName)
+	}
+	if _, ok := i.sets[string(ipsetName)].subSets[entry]; !ok {
+		return errors.Errorf("ipset %s is not a member of %s", entry, ipsetName)
+	}
+	delete(i.sets[string(ipsetName)].subSets, entry)
+
+	return nil
+}
+
+func (i *mockIPSet) Flush(ipsetName ipset.Name) error {
+	return errors.New("Not Implemented")
+}
+
+func (i *mockIPSet) FlushAll() error {
+	return errors.New("Not Implemented")
+}
+
+func (i *mockIPSet) Destroy(ipsetName ipset.Name) error {
+	if _, ok := i.sets[string(ipsetName)]; !ok {
+		return errors.Errorf("ipset %s does not exist", ipsetName)
+	}
+	delete(i.sets, string(ipsetName))
+	return nil
+}
+
+func (i *mockIPSet) DestroyAll() error {
+	return errors.New("Not Implemented")
+}
+
+func (i *mockIPSet) List(prefix string) ([]ipset.Name, error) {
+	return []ipset.Name{}, errors.New("Not Implemented")
+}
+
+type mockIPTables struct {
+}
+
+func (ipt *mockIPTables) Append(table, chain string, rulespec ...string) error {
+	return nil
+}
+
+func (ipt *mockIPTables) Delete(table, chain string, rulespec ...string) error {
+	return nil
+}
+
+func (ipt *mockIPTables) Insert(table, chain string, pos int, rulespec ...string) error {
+	return nil
+}
+
+func TestRegressionPolicyNamespaceOrdering3059(t *testing.T) {
+	// Test for race condition between namespace and networkpolicy events
+	// https://github.com/weaveworks/weave/issues/3059
+
+	sourceNamespace := &coreapi.Namespace{
+		ObjectMeta: coreapi.ObjectMeta{
+			Name: "source",
+			Labels: map[string]string{
+				"app": "source",
+			},
+		},
+	}
+
+	destinationNamespace := &coreapi.Namespace{
+		ObjectMeta: coreapi.ObjectMeta{
+			Name: "destination",
+		},
+	}
+
+	port := intstr.FromInt(12345)
+
+	networkPolicy := &extnapi.NetworkPolicy{
+		ObjectMeta: coreapi.ObjectMeta{
+			Name:      "network-policy",
+			Namespace: "destination",
+		},
+		Spec: extnapi.NetworkPolicySpec{
+			Ingress: []extnapi.NetworkPolicyIngressRule{
+				{
+					From: []extnapi.NetworkPolicyPeer{
+						{
+							NamespaceSelector: &unversionedapi.LabelSelector{
+								MatchLabels: map[string]string{
+									"app": "source",
+								},
+							},
+						},
+					},
+					Ports: []extnapi.NetworkPolicyPort{
+						{
+							Port: &port,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	// Namespaces first
+	m := newMockIPSet()
+	controller := New("foo", &mockIPTables{}, &m)
+
+	const (
+		selectorIPSetName = "weave-I239Zp%sCvoVt*D6u=A!2]YEk"
+		sourceIPSetName   = "weave-HboJG1fGgG]/SR%k9H#hv5e96"
+	)
+
+	controller.AddNamespace(sourceNamespace)
+	controller.AddNamespace(destinationNamespace)
+
+	controller.AddNetworkPolicy(networkPolicy)
+
+	require.Equal(t, true, m.sets[selectorIPSetName].subSets[sourceIPSetName])
+
+	// NetworkPolicy first
+	m = newMockIPSet()
+	controller = New("foo", &mockIPTables{}, &m)
+
+	controller.AddNetworkPolicy(networkPolicy)
+
+	controller.AddNamespace(sourceNamespace)
+	controller.AddNamespace(destinationNamespace)
+
+	require.Equal(t, true, m.sets[selectorIPSetName].subSets[sourceIPSetName])
+
+}

npc/iptables/iptables.go

@@ -0,0 +1,7 @@
+package iptables
+
+type Interface interface {
+	Append(table, chain string, rulespec ...string) error
+	Delete(table, chain string, rulespec ...string) error
+	Insert(table, chain string, pos int, rulespec ...string) error
+}

npc/namespace.go

@@ -3,7 +3,6 @@ package npc
 import (
 	"encoding/json"
 
-	"github.com/coreos/go-iptables/iptables"
 	"k8s.io/client-go/pkg/api/unversioned"
 	coreapi "k8s.io/client-go/pkg/api/v1"
 	extnapi "k8s.io/client-go/pkg/apis/extensions/v1beta1"
@@ -12,10 +11,11 @@ import (
 
 	"github.com/weaveworks/weave/common"
 	"github.com/weaveworks/weave/npc/ipset"
+	"github.com/weaveworks/weave/npc/iptables"
 )
 
 type ns struct {
-	ipt *iptables.IPTables // interface to iptables
+	ipt iptables.Interface // interface to iptables
 	ips ipset.Interface    // interface to ipset
 
 	name      string                               // k8s Namespace name
@@ -32,7 +32,7 @@ type ns struct {
 	rules        *ruleSet
 }
 
-func newNS(name, nodeName string, ipt *iptables.IPTables, ips ipset.Interface, nsSelectors *selectorSet) (*ns, error) {
+func newNS(name, nodeName string, ipt iptables.Interface, ips ipset.Interface, nsSelectors *selectorSet) (*ns, error) {
 	allPods, err := newSelectorSpec(&unversioned.LabelSelector{}, name, ipset.HashIP)
 	if err != nil {
 		return nil, err
@@ -265,7 +265,9 @@ func (ns *ns) addNamespace(obj *coreapi.Namespace) error {
 
 	// Insert a rule to bypass policies if namespace is DefaultAllow
 	if !isDefaultDeny(obj) {
-		return ns.ensureBypassRule(ns.allPods.ipsetName)
+		if err := ns.ensureBypassRule(ns.allPods.ipsetName); err != nil {
+			return err
+		}
 	}
 
 	// Add namespace ipset to matching namespace selectors
@@ -282,10 +284,14 @@ func (ns *ns) updateNamespace(oldObj, newObj *coreapi.Namespace) error {
 	if oldDefaultDeny != newDefaultDeny {
 		common.Log.Infof("namespace DefaultDeny changed from %t to %t", oldDefaultDeny, newDefaultDeny)
 		if oldDefaultDeny {
-			return ns.ensureBypassRule(ns.allPods.ipsetName)
+			if err := ns.ensureBypassRule(ns.allPods.ipsetName); err != nil {
+				return err
+			}
 		}
 		if newDefaultDeny {
-			return ns.deleteBypassRule(ns.allPods.ipsetName)
+			if err := ns.deleteBypassRule(ns.allPods.ipsetName); err != nil {
+				return err
+			}
 		}
 	}
 
@@ -318,7 +324,9 @@ func (ns *ns) deleteNamespace(obj *coreapi.Namespace) error {
 
 	// Remove bypass rule
 	if !isDefaultDeny(obj) {
-		return ns.deleteBypassRule(ns.allPods.ipsetName)
+		if err := ns.deleteBypassRule(ns.allPods.ipsetName); err != nil {
+			return err
+		}
 	}
 
 	// Remove namespace ipset from any matching namespace selectors

npc/rule.go

@@ -3,10 +3,10 @@ package npc
 import (
 	"strings"
 
-	"github.com/coreos/go-iptables/iptables"
 	"k8s.io/client-go/pkg/types"
 
 	"github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/npc/iptables"
 )
 
 type ruleSpec struct {
@@ -35,11 +35,11 @@ func newRuleSpec(proto *string, srcHost *selectorSpec, dstHost *selectorSpec, ds
 }
 
 type ruleSet struct {
-	ipt   *iptables.IPTables
+	ipt   iptables.Interface
 	users map[string]map[types.UID]struct{}
 }
 
-func newRuleSet(ipt *iptables.IPTables) *ruleSet {
+func newRuleSet(ipt iptables.Interface) *ruleSet {
 	return &ruleSet{ipt, make(map[string]map[types.UID]struct{})}
 }