Add release v2.1.3

Fix release script for plugin latest_release

CHANGELOG.md

@@ -1,3 +1,24 @@
+## Release 2.1.3
+
+This release fixes a race-condition in the IP reclaim code for weave-kube
+where two nodes could end up fighting over the same space and break
+connectivity #3190, #3192
+
+## Release 2.1.2
+
+This release fixes a couple of bugs discovered since the release of Weave Net 2.1.0
+
+##Bug fixes
+
+* Fix crash seen when starting 10-15 nodes simultaneously #3184,#3186
+* Fix NetworkPolicy blocking traffic if updates come out of order from Kubernetes #3177,#3181
+
+Thanks to the following contributors:
+@zignig
+
+[Full list of changes](https://github.com/weaveworks/weave/milestone/66?closed=1)
+
+
 ## Release 2.1.1
 
 As 2.1.0, but fixing a couple of installation glitches. #3175,#3176

bin/release

@@ -196,7 +196,7 @@ publish() {
 
     if [ "$RELEASE_TYPE" = 'PRERELEASE' ] ; then
         echo "== Building and pushing plugin"
-        make UPDATE_LATEST=$IS_MAINLINE SUDO=$SUDO WEAVE_VERSION=$VERSION DOCKERHUB_USER=$DOCKERHUB_USER plugin_publish
+        make UPDATE_LATEST=false SUDO=$SUDO WEAVE_VERSION=$VERSION DOCKERHUB_USER=$DOCKERHUB_USER plugin_publish
         echo "** Plugin built and pushed"
 
         echo "== Tagging and pushing images on docker hub as user $DOCKERHUB_USER"
@@ -232,7 +232,7 @@ publish() {
         echo '** Sanity checks OK for publishing tag' $LATEST_TAG as $DOCKERHUB_USER/weave:$VERSION
 
         echo "== Building and pushing plugin"
-        make UPDATE_LATEST=$IS_MAINLINE SUDO=$SUDO WEAVE_VERSION=$VERSION DOCKERHUB_USER=$DOCKERHUB_USER plugin_publish
+        make UPDATE_LATEST=true SUDO=$SUDO WEAVE_VERSION=$VERSION DOCKERHUB_USER=$DOCKERHUB_USER plugin_publish
         echo "** Plugin built and pushed"
 
         echo "== Tagging and pushing images on docker hub as user $DOCKERHUB_USER"

npc/controller.go

@@ -59,7 +59,7 @@ func (npc *controller) onNewNsSelector(selector *selector) error {
 	for _, ns := range npc.nss {
 		if ns.namespace != nil {
 			if selector.matches(ns.namespace.ObjectMeta.Labels) {
-				if err := selector.addEntry(string(ns.allPods.ipsetName), namespaceComment(ns)); err != nil {
+				if err := selector.addEntry(ns.namespace.ObjectMeta.UID, string(ns.allPods.ipsetName), namespaceComment(ns)); err != nil {
 					return err
 				}
 			}

npc/controller_test.go

@@ -10,13 +10,14 @@ import (
 	coreapi "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
 type mockSet struct {
 	name    ipset.Name
 	setType ipset.Type
-	subSets map[string]bool
+	subSets map[string]map[types.UID]bool
 }
 
 type mockIPSet struct {
@@ -24,70 +25,57 @@ type mockIPSet struct {
 }
 
 func newMockIPSet() mockIPSet {
-	i := mockIPSet{
+	return mockIPSet{sets: make(map[string]mockSet)}
-		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)}
+	i.sets[string(ipsetName)] = mockSet{name: ipsetName, setType: ipsetType, subSets: make(map[string]map[types.UID]bool)}
 	return nil
 }
 
-func (i *mockIPSet) AddEntry(ipsetName ipset.Name, entry string, comment string) error {
+func (i *mockIPSet) AddEntry(user types.UID, ipsetName ipset.Name, entry string, comment string) error {
-	return i.addEntry(ipsetName, entry, comment, true)
+	log.Printf("adding entry %s to %s for %s", entry, ipsetName, user)
-}
-
-func (i *mockIPSet) AddEntryIfNotExist(ipsetName ipset.Name, entry string, comment string) error {
-	return i.addEntry(ipsetName, entry, comment, false)
-}
-
-func (i *mockIPSet) addEntry(ipsetName ipset.Name, entry string, comment string, checkIfExists bool) error {
-	log.Printf("adding entry %s to %s", entry, ipsetName)
 	if _, ok := i.sets[string(ipsetName)]; !ok {
-		return errors.Errorf("ipset %s does not exist", entry)
+		return errors.Errorf("%s does not exist", entry)
 	}
-	if checkIfExists {
+	if i.sets[string(ipsetName)].subSets[entry] == nil {
-		if _, ok := i.sets[string(ipsetName)].subSets[entry]; ok {
+		i.sets[string(ipsetName)].subSets[entry] = make(map[types.UID]bool)
-			return errors.Errorf("ipset %s is already a member of %s", entry, ipsetName)
-		}
 	}
-	i.sets[string(ipsetName)].subSets[entry] = true
+	if _, ok := i.sets[string(ipsetName)].subSets[entry][user]; ok {
+		return errors.Errorf("user %s already owns entry %s", user, entry)
+	}
+	i.sets[string(ipsetName)].subSets[entry][user] = true
 
 	return nil
 }
 
-func (i *mockIPSet) DelEntry(ipsetName ipset.Name, entry string) error {
+func (i *mockIPSet) DelEntry(user types.UID, ipsetName ipset.Name, entry string) error {
-	return i.delEntry(ipsetName, entry, true)
+	log.Printf("deleting entry %s from %s for %s", entry, ipsetName, user)
-}
-
-func (i *mockIPSet) DelEntryIfExists(ipsetName ipset.Name, entry string) error {
-	return i.delEntry(ipsetName, entry, false)
-}
-
-func (i *mockIPSet) delEntry(ipsetName ipset.Name, entry string, checkIfExists bool) 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 checkIfExists {
+	if _, ok := i.sets[string(ipsetName)].subSets[entry][user]; !ok {
-		if _, ok := i.sets[string(ipsetName)].subSets[entry]; !ok {
+		return errors.Errorf("user %s does not own entry %s", user, entry)
-			return errors.Errorf("ipset %s is not a member of %s", entry, ipsetName)
+	}
-		}
+	delete(i.sets[string(ipsetName)].subSets[entry], user)
+
+	if len(i.sets[string(ipsetName)].subSets[entry]) == 0 {
+		delete(i.sets[string(ipsetName)].subSets, entry)
 	}
-	delete(i.sets[string(ipsetName)].subSets, entry)
 
 	return nil
 }
 
-func (i *mockIPSet) Exist(ipsetName ipset.Name, entry string) bool {
+func (i *mockIPSet) Exist(user types.UID, ipsetName ipset.Name, entry string) bool {
-	_, found := i.sets[string(ipsetName)].subSets[entry]
+	_, ok := i.sets[string(ipsetName)].subSets[entry][user]
-	return found
+	return ok
+}
+
+func (i *mockIPSet) entryExists(ipsetName ipset.Name, entry string) bool {
+	return len(i.sets[string(ipsetName)].subSets[entry]) > 0
 }
 
 func (i *mockIPSet) Flush(ipsetName ipset.Name) error {
@@ -191,7 +179,7 @@ func TestRegressionPolicyNamespaceOrdering3059(t *testing.T) {
 
 	controller.AddNetworkPolicy(networkPolicy)
 
-	require.Equal(t, true, m.sets[selectorIPSetName].subSets[sourceIPSetName])
+	require.Equal(t, true, len(m.sets[selectorIPSetName].subSets[sourceIPSetName]) > 0)
 
 	// NetworkPolicy first
 	m = newMockIPSet()
@@ -202,8 +190,7 @@ func TestRegressionPolicyNamespaceOrdering3059(t *testing.T) {
 	controller.AddNamespace(sourceNamespace)
 	controller.AddNamespace(destinationNamespace)
 
-	require.Equal(t, true, m.sets[selectorIPSetName].subSets[sourceIPSetName])
+	require.Equal(t, true, len(m.sets[selectorIPSetName].subSets[sourceIPSetName]) > 0)
-
 }
 
 // Tests default-allow ipset behavior when running in non-legacy mode.
@@ -238,7 +225,7 @@ func TestDefaultAllow(t *testing.T) {
 	controller.AddPod(podFoo)
 
 	// Should add the foo pod to default-allow
-	require.True(t, m.Exist(defaultAllowIPSetName, fooPodIP))
+	require.True(t, m.entryExists(defaultAllowIPSetName, fooPodIP))
 
 	podBar := &coreapi.Pod{
 		ObjectMeta: metav1.ObjectMeta{
@@ -252,7 +239,7 @@ func TestDefaultAllow(t *testing.T) {
 	controller.UpdatePod(podBarNoIP, podBar)
 
 	// Should add the bar pod to default-allow
-	require.True(t, m.Exist(defaultAllowIPSetName, barPodIP))
+	require.True(t, m.entryExists(defaultAllowIPSetName, barPodIP))
 
 	// Allow access from the bar pod to the foo pod
 	netpol := &networkingv1.NetworkPolicy{
@@ -274,37 +261,118 @@ func TestDefaultAllow(t *testing.T) {
 	controller.AddNetworkPolicy(netpol)
 
 	// Should remove the foo pod from default-allow as the netpol selects it
-	require.False(t, m.Exist(defaultAllowIPSetName, fooPodIP))
+	require.False(t, m.entryExists(defaultAllowIPSetName, fooPodIP))
-	require.True(t, m.Exist(defaultAllowIPSetName, barPodIP))
+	require.True(t, m.entryExists(defaultAllowIPSetName, barPodIP))
 
 	podBarWithNewIP := *podBar
 	podBarWithNewIP.Status.PodIP = barPodNewIP
 	controller.UpdatePod(podBar, &podBarWithNewIP)
 
 	// Should update IP addr of the bar pod in default-allow
-	require.False(t, m.Exist(defaultAllowIPSetName, barPodIP))
+	require.False(t, m.entryExists(defaultAllowIPSetName, barPodIP))
-	require.True(t, m.Exist(defaultAllowIPSetName, barPodNewIP))
+	require.True(t, m.entryExists(defaultAllowIPSetName, barPodNewIP))
 
 	controller.UpdatePod(&podBarWithNewIP, podBarNoIP)
 	// Should remove the bar pod from default-allow as it does not have any IP addr
-	require.False(t, m.Exist(defaultAllowIPSetName, barPodNewIP))
+	require.False(t, m.entryExists(defaultAllowIPSetName, barPodNewIP))
 
 	podFooWithNewLabel := *podFoo
 	podFooWithNewLabel.ObjectMeta.Labels = map[string]string{"run": "new-foo"}
 	controller.UpdatePod(podFoo, &podFooWithNewLabel)
 
 	// Should bring back the foo pod to default-allow as it does not match dst of any netpol
-	require.True(t, m.Exist(defaultAllowIPSetName, fooPodIP))
+	require.True(t, m.entryExists(defaultAllowIPSetName, fooPodIP))
 
 	controller.UpdatePod(&podFooWithNewLabel, podFoo)
 	// Should remove from default-allow as it matches the netpol after the update
-	require.False(t, m.Exist(defaultAllowIPSetName, fooPodIP))
+	require.False(t, m.entryExists(defaultAllowIPSetName, fooPodIP))
 
 	controller.DeleteNetworkPolicy(netpol)
 	// Should bring back the foo pod to default-allow as no netpol selects it
-	require.True(t, m.Exist(defaultAllowIPSetName, fooPodIP))
+	require.True(t, m.entryExists(defaultAllowIPSetName, fooPodIP))
 
 	controller.DeletePod(podFoo)
 	// Should remove foo pod from default-allow
-	require.False(t, m.Exist(defaultAllowIPSetName, fooPodIP))
+	require.False(t, m.entryExists(defaultAllowIPSetName, fooPodIP))
+}
+
+func TestOutOfOrderPodEvents(t *testing.T) {
+	const (
+		defaultAllowIPSetName = "weave-E.1.0W^NGSp]0_t5WwH/]gX@L"
+		runBarIPSetName       = "weave-bZ~x=yBgzH)Ht()K*Uv3z{M]Y"
+		podIP                 = "10.32.0.10"
+	)
+
+	m := newMockIPSet()
+	controller := New("qux", false, &mockIPTables{}, &m)
+
+	defaultNamespace := &coreapi.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "default",
+		},
+	}
+	controller.AddNamespace(defaultNamespace)
+
+	podFoo := &coreapi.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			UID:       "foo",
+			Namespace: "default",
+			Name:      "foo",
+			Labels:    map[string]string{"run": "foo"}},
+		Status: coreapi.PodStatus{PodIP: podIP}}
+	controller.AddPod(podFoo)
+
+	// Should be in default-allow as no netpol selects podFoo
+	require.True(t, m.entryExists(defaultAllowIPSetName, podIP))
+
+	netpol := &networkingv1.NetworkPolicy{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "allow-from-bar-to-foo",
+			Namespace: "default",
+		},
+		Spec: networkingv1.NetworkPolicySpec{
+			PodSelector: metav1.LabelSelector{MatchLabels: map[string]string{"run": "foo"}},
+			Ingress: []networkingv1.NetworkPolicyIngressRule{{
+				From: []networkingv1.NetworkPolicyPeer{{
+					PodSelector: &metav1.LabelSelector{
+						MatchLabels: map[string]string{"run": "bar"},
+					},
+				}},
+			}},
+		},
+	}
+	controller.AddNetworkPolicy(netpol)
+
+	// Shouldn't be in default-allow as netpol above selects podFoo
+	require.False(t, m.entryExists(defaultAllowIPSetName, podIP))
+
+	podBar := &coreapi.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			UID:       "bar",
+			Namespace: "default",
+			Name:      "bar",
+			Labels:    map[string]string{"run": "bar"}},
+		Status: coreapi.PodStatus{PodIP: podIP}}
+	controller.AddPod(podBar)
+
+	// Should be in default-allow as no netpol selects podBar
+	require.True(t, m.entryExists(defaultAllowIPSetName, podIP))
+	require.True(t, m.Exist(podBar.ObjectMeta.UID, defaultAllowIPSetName, podIP))
+	// Should be in run=bar ipset
+	require.True(t, m.entryExists(runBarIPSetName, podIP))
+
+	controller.DeletePod(podFoo)
+	// Multiple duplicate events should not affect npc state
+	controller.DeletePod(podFoo)
+	controller.DeletePod(podFoo)
+
+	// Should be in default-allow as no netpol selects podBar and podFoo removal
+	// should not affect podBar in default-allow
+	require.True(t, m.entryExists(defaultAllowIPSetName, podIP))
+
+	controller.DeletePod(podBar)
+
+	// Should remove from default-allow and run=bar ipsets
+	require.Equal(t, 0, len(m.sets[defaultAllowIPSetName].subSets))
+	require.False(t, m.entryExists(runBarIPSetName, podIP))
 }

npc/ipset/ipset.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 
 	"github.com/pkg/errors"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 type Name string
@@ -19,11 +20,9 @@ const (
 
 type Interface interface {
 	Create(ipsetName Name, ipsetType Type) error
-	AddEntry(ipsetName Name, entry string, comment string) error
+	AddEntry(user types.UID, ipsetName Name, entry string, comment string) error
-	AddEntryIfNotExist(ipsetName Name, entry string, comment string) error
+	DelEntry(user types.UID, ipsetName Name, entry string) error
-	DelEntry(ipsetName Name, entry string) error
+	Exist(user types.UID, ipsetName Name, entry string) bool
-	DelEntryIfExists(ipsetName Name, entry string) error
-	Exist(ipsetName Name, entry string) bool
 	Flush(ipsetName Name) error
 	Destroy(ipsetName Name) error
 
@@ -33,14 +32,27 @@ type Interface interface {
 	DestroyAll() error
 }
 
+type entryKey struct {
+	ipsetName Name
+	entry     string
+}
+
 type ipset struct {
-	refCount
 	*log.Logger
 	enableComments bool
+	// List of users per ipset entry. User is either a namespace or a pod.
+	// There might be multiple users for the same ipset & entry pair because
+	// events from k8s API server might be out of order causing duplicate IPs:
+	// https://github.com/weaveworks/weave/issues/2792.
+	users map[entryKey]map[types.UID]struct{}
 }
 
 func New(logger *log.Logger) Interface {
-	ips := &ipset{refCount: newRefCount(), Logger: logger, enableComments: true}
+	ips := &ipset{
+		Logger:         logger,
+		enableComments: true,
+		users:          make(map[entryKey]map[types.UID]struct{}),
+	}
 
 	// Check for comment support
 	testIpsetName := Name("weave-test-comment")
@@ -65,11 +77,15 @@ func (i *ipset) Create(ipsetName Name, ipsetType Type) error {
 	return doExec(args...)
 }
 
-func (i *ipset) AddEntry(ipsetName Name, entry string, comment string) error {
+func (i *ipset) AddEntry(user types.UID, ipsetName Name, entry string, comment string) error {
-	i.Logger.Printf("adding entry %s to %s", entry, ipsetName)
+	i.Logger.Printf("adding entry %s to %s of %s", entry, ipsetName, user)
-	if i.inc(ipsetName, entry) > 1 { // already in the set
+
+	if !i.addUser(user, ipsetName, entry) { // already in the set
 		return nil
 	}
+
+	i.Logger.Printf("added entry %s to %s of %s", entry, ipsetName, user)
+
 	args := []string{"add", string(ipsetName), entry}
 	if i.enableComments {
 		args = append(args, "comment", comment)
@@ -77,53 +93,39 @@ func (i *ipset) AddEntry(ipsetName Name, entry string, comment string) error {
 	return doExec(args...)
 }
 
-// AddEntryIfNotExist does the same as AddEntry but bypasses the ref counting.
+func (i *ipset) DelEntry(user types.UID, ipsetName Name, entry string) error {
-// Should be used only with "default-allow" ipsets.
+	i.Logger.Printf("deleting entry %s from %s of %s", entry, ipsetName, user)
-func (i *ipset) AddEntryIfNotExist(ipsetName Name, entry string, comment string) error {
-	if i.count(ipsetName, entry) == 1 {
-		return nil
-	}
-	return i.AddEntry(ipsetName, entry, comment)
-}
 
-func (i *ipset) DelEntry(ipsetName Name, entry string) error {
+	if !i.delUser(user, ipsetName, entry) { // still needed
-	i.Logger.Printf("deleting entry %s from %s", entry, ipsetName)
-	if i.dec(ipsetName, entry) > 0 { // still needed
 		return nil
 	}
+
+	i.Logger.Printf("deleted entry %s from %s of %s", entry, ipsetName, user)
+
 	return doExec("del", string(ipsetName), entry)
 }
 
-// DelEntryIfExists does the same as DelEntry but bypasses the ref counting.
+func (i *ipset) Exist(user types.UID, ipsetName Name, entry string) bool {
-// Should be used only with "default-allow" ipsets.
+	return i.existUser(user, ipsetName, entry)
-func (i *ipset) DelEntryIfExists(ipsetName Name, entry string) error {
-	if i.count(ipsetName, entry) == 0 {
-		return nil
-	}
-	return i.DelEntry(ipsetName, entry)
-}
-
-func (i *ipset) Exist(ipsetName Name, entry string) bool {
-	return i.count(ipsetName, entry) > 0
 }
 
 func (i *ipset) Flush(ipsetName Name) error {
-	i.removeSet(ipsetName)
+	i.removeSetFromUsers(ipsetName)
 	return doExec("flush", string(ipsetName))
 }
 
 func (i *ipset) FlushAll() error {
-	i.refCount = newRefCount()
+	i.users = make(map[entryKey]map[types.UID]struct{})
 	return doExec("flush")
 }
 
 func (i *ipset) Destroy(ipsetName Name) error {
-	i.removeSet(ipsetName)
+	i.removeSetFromUsers(ipsetName)
 	return doExec("destroy", string(ipsetName))
 }
 
 func (i *ipset) DestroyAll() error {
-	i.refCount = newRefCount()
+	i.users = make(map[entryKey]map[types.UID]struct{})
 	return doExec("destroy")
 }
 
@@ -145,48 +147,52 @@ func (i *ipset) List(prefix string) ([]Name, error) {
 	return selected, err
 }
 
+// Returns true if entry does not exist in ipset (entry has to be inserted into ipset).
+func (i *ipset) addUser(user types.UID, ipsetName Name, entry string) bool {
+	k := entryKey{ipsetName, entry}
+	add := false
+
+	if i.users[k] == nil {
+		i.users[k] = make(map[types.UID]struct{})
+	}
+	if len(i.users[k]) == 0 {
+		add = true
+	}
+	i.users[k][user] = struct{}{}
+
+	return add
+}
+
+// Returns true if user is the last owner of entry (entry has to be removed from ipset).
+func (i *ipset) delUser(user types.UID, ipsetName Name, entry string) bool {
+	k := entryKey{ipsetName, entry}
+
+	oneLeft := len(i.users[k]) == 1
+	delete(i.users[k], user)
+
+	if len(i.users[k]) == 0 {
+		delete(i.users, k)
+	}
+
+	return oneLeft && (len(i.users[k]) == 0)
+}
+
+func (i *ipset) existUser(user types.UID, ipsetName Name, entry string) bool {
+	_, ok := i.users[entryKey{ipsetName, entry}][user]
+	return ok
+}
+
+func (i *ipset) removeSetFromUsers(ipsetName Name) {
+	for k := range i.users {
+		if k.ipsetName == ipsetName {
+			delete(i.users, k)
+		}
+	}
+}
+
 func doExec(args ...string) error {
 	if output, err := exec.Command("ipset", args...).CombinedOutput(); err != nil {
 		return errors.Wrapf(err, "ipset %v failed: %s", args, output)
 	}
 	return nil
 }
-
-// Reference-counting
-type key struct {
-	ipsetName Name
-	entry     string
-}
-
-// note no locking is required as all operations are serialised in the controller
-type refCount struct {
-	ref map[key]int
-}
-
-func newRefCount() refCount {
-	return refCount{ref: make(map[key]int)}
-}
-
-func (rc *refCount) inc(ipsetName Name, entry string) int {
-	k := key{ipsetName, entry}
-	rc.ref[k]++
-	return rc.ref[k]
-}
-
-func (rc *refCount) dec(ipsetName Name, entry string) int {
-	k := key{ipsetName, entry}
-	rc.ref[k]--
-	return rc.ref[k]
-}
-
-func (rc *refCount) count(ipsetName Name, entry string) int {
-	return rc.ref[key{ipsetName, entry}]
-}
-
-func (rc *refCount) removeSet(ipsetName Name) {
-	for k := range rc.ref {
-		if k.ipsetName == ipsetName {
-			delete(rc.ref, k)
-		}
-	}
-}

npc/namespace.go

@@ -92,7 +92,7 @@ func (ns *ns) onNewPodSelector(selector *selector) error {
 	for _, pod := range ns.pods {
 		if hasIP(pod) {
 			if selector.matches(pod.ObjectMeta.Labels) {
-				if err := selector.addEntry(pod.Status.PodIP, podComment(pod)); err != nil {
+				if err := selector.addEntry(pod.ObjectMeta.UID, pod.Status.PodIP, podComment(pod)); err != nil {
 					return err
 				}
 
@@ -111,7 +111,7 @@ func (ns *ns) onNewDstPodSelector(selector *selector) error {
 		if hasIP(pod) {
 			// Remove the pod from default-allow if dst podselector matches the pod
 			if selector.matches(pod.ObjectMeta.Labels) {
-				if err := ns.ips.DelEntryIfExists(ns.defaultAllowIPSet, pod.Status.PodIP); err != nil {
+				if err := ns.ips.DelEntry(pod.ObjectMeta.UID, ns.defaultAllowIPSet, pod.Status.PodIP); err != nil {
 					return err
 				}
 			}
@@ -150,7 +150,7 @@ func (ns *ns) addToDefaultAllowIfNoMatching(pod *coreapi.Pod) error {
 		}
 	}
 	if !found {
-		if err := ns.ips.AddEntryIfNotExist(ns.defaultAllowIPSet, pod.Status.PodIP, podComment(pod)); err != nil {
+		if err := ns.ips.AddEntry(pod.ObjectMeta.UID, ns.defaultAllowIPSet, pod.Status.PodIP, podComment(pod)); err != nil {
 			return err
 		}
 	}
@@ -172,16 +172,16 @@ func (ns *ns) addPod(obj *coreapi.Pod) error {
 	}
 
 	if ns.checkLocalPod(obj) {
-		ns.ips.AddEntry(LocalIpset, obj.Status.PodIP, podComment(obj))
+		ns.ips.AddEntry(obj.ObjectMeta.UID, LocalIpset, obj.Status.PodIP, podComment(obj))
 	}
 
-	found, err := ns.podSelectors.addToMatching(obj.ObjectMeta.Labels, obj.Status.PodIP, podComment(obj))
+	found, err := ns.podSelectors.addToMatching(obj.ObjectMeta.UID, obj.ObjectMeta.Labels, obj.Status.PodIP, podComment(obj))
 	if err != nil {
 		return err
 	}
 	// If there are no matching dst selectors, add the pod to default-allow
 	if !ns.legacy && !found {
-		if err := ns.ips.AddEntryIfNotExist(ns.defaultAllowIPSet, obj.Status.PodIP, podComment(obj)); err != nil {
+		if err := ns.ips.AddEntry(obj.ObjectMeta.UID, ns.defaultAllowIPSet, obj.Status.PodIP, podComment(obj)); err != nil {
 			return err
 		}
 	}
@@ -199,29 +199,29 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error {
 
 	if hasIP(oldObj) && !hasIP(newObj) {
 		if ns.checkLocalPod(oldObj) {
-			ns.ips.DelEntry(LocalIpset, oldObj.Status.PodIP)
+			ns.ips.DelEntry(oldObj.ObjectMeta.UID, LocalIpset, oldObj.Status.PodIP)
 		}
 
 		if !ns.legacy {
-			if err := ns.ips.DelEntryIfExists(ns.defaultAllowIPSet, oldObj.Status.PodIP); err != nil {
+			if err := ns.ips.DelEntry(oldObj.ObjectMeta.UID, ns.defaultAllowIPSet, oldObj.Status.PodIP); err != nil {
 				return err
 			}
 		}
 
-		return ns.podSelectors.delFromMatching(oldObj.ObjectMeta.Labels, oldObj.Status.PodIP)
+		return ns.podSelectors.delFromMatching(oldObj.ObjectMeta.UID, oldObj.ObjectMeta.Labels, oldObj.Status.PodIP)
 	}
 
 	if !hasIP(oldObj) && hasIP(newObj) {
 		if ns.checkLocalPod(newObj) {
-			ns.ips.AddEntry(LocalIpset, newObj.Status.PodIP, podComment(newObj))
+			ns.ips.AddEntry(newObj.ObjectMeta.UID, LocalIpset, newObj.Status.PodIP, podComment(newObj))
 		}
-		found, err := ns.podSelectors.addToMatching(newObj.ObjectMeta.Labels, newObj.Status.PodIP, podComment(newObj))
+		found, err := ns.podSelectors.addToMatching(newObj.ObjectMeta.UID, newObj.ObjectMeta.Labels, newObj.Status.PodIP, podComment(newObj))
 		if err != nil {
 			return err
 		}
 
 		if !ns.legacy && !found {
-			if err := ns.ips.AddEntryIfNotExist(ns.defaultAllowIPSet, newObj.Status.PodIP, podComment(newObj)); err != nil {
+			if err := ns.ips.AddEntry(newObj.ObjectMeta.UID, ns.defaultAllowIPSet, newObj.Status.PodIP, podComment(newObj)); err != nil {
 				return err
 			}
 		}
@@ -231,12 +231,15 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error {
 
 	if !ns.legacy &&
 		oldObj.Status.PodIP != newObj.Status.PodIP &&
-		ns.ips.Exist(ns.defaultAllowIPSet, oldObj.Status.PodIP) {
+		ns.ips.Exist(oldObj.ObjectMeta.UID, ns.defaultAllowIPSet, oldObj.Status.PodIP) {
+		// Instead of iterating over all selectors we check whether old pod IP
+		// has been inserted into default-allow ipset to decide whether the IP
+		// in the ipset has to be updated.
 
-		if err := ns.ips.DelEntryIfExists(ns.defaultAllowIPSet, oldObj.Status.PodIP); err != nil {
+		if err := ns.ips.DelEntry(oldObj.ObjectMeta.UID, ns.defaultAllowIPSet, oldObj.Status.PodIP); err != nil {
 			return err
 		}
-		if err := ns.ips.AddEntryIfNotExist(ns.defaultAllowIPSet, newObj.Status.PodIP, podComment(newObj)); err != nil {
+		if err := ns.ips.AddEntry(newObj.ObjectMeta.UID, ns.defaultAllowIPSet, newObj.Status.PodIP, podComment(newObj)); err != nil {
 			return err
 		}
 	}
@@ -251,12 +254,12 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error {
 				continue
 			}
 			if oldMatch {
-				if err := ps.delEntry(oldObj.Status.PodIP); err != nil {
+				if err := ps.delEntry(oldObj.ObjectMeta.UID, oldObj.Status.PodIP); err != nil {
 					return err
 				}
 			}
 			if newMatch {
-				if err := ps.addEntry(newObj.Status.PodIP, podComment(newObj)); err != nil {
+				if err := ps.addEntry(newObj.ObjectMeta.UID, newObj.Status.PodIP, podComment(newObj)); err != nil {
 					return err
 				}
 			}
@@ -264,7 +267,7 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error {
 			if !ns.legacy && ns.podSelectors.dstSelectorExist(ps) {
 				switch {
 				case !oldMatch && newMatch:
-					if err := ns.ips.DelEntryIfExists(ns.defaultAllowIPSet, oldObj.Status.PodIP); err != nil {
+					if err := ns.ips.DelEntry(oldObj.ObjectMeta.UID, ns.defaultAllowIPSet, oldObj.Status.PodIP); err != nil {
 						return err
 					}
 				case oldMatch && !newMatch:
@@ -287,16 +290,16 @@ func (ns *ns) deletePod(obj *coreapi.Pod) error {
 	}
 
 	if ns.checkLocalPod(obj) {
-		ns.ips.DelEntry(LocalIpset, obj.Status.PodIP)
+		ns.ips.DelEntry(obj.ObjectMeta.UID, LocalIpset, obj.Status.PodIP)
 	}
 
 	if !ns.legacy {
-		if err := ns.ips.DelEntryIfExists(ns.defaultAllowIPSet, obj.Status.PodIP); err != nil {
+		if err := ns.ips.DelEntry(obj.ObjectMeta.UID, ns.defaultAllowIPSet, obj.Status.PodIP); err != nil {
 			return err
 		}
 	}
 
-	return ns.podSelectors.delFromMatching(obj.ObjectMeta.Labels, obj.Status.PodIP)
+	return ns.podSelectors.delFromMatching(obj.ObjectMeta.UID, obj.ObjectMeta.Labels, obj.Status.PodIP)
 }
 
 func (ns *ns) addNetworkPolicy(obj interface{}) error {
@@ -408,7 +411,7 @@ func (ns *ns) addNamespace(obj *coreapi.Namespace) error {
 	}
 
 	// Add namespace ipset to matching namespace selectors
-	_, err := ns.nsSelectors.addToMatching(obj.ObjectMeta.Labels, string(ns.allPods.ipsetName), namespaceComment(ns))
+	_, err := ns.nsSelectors.addToMatching(obj.ObjectMeta.UID, obj.ObjectMeta.Labels, string(ns.allPods.ipsetName), namespaceComment(ns))
 	return err
 }
 
@@ -442,12 +445,12 @@ func (ns *ns) updateNamespace(oldObj, newObj *coreapi.Namespace) error {
 				continue
 			}
 			if oldMatch {
-				if err := selector.delEntry(string(ns.allPods.ipsetName)); err != nil {
+				if err := selector.delEntry(ns.namespace.ObjectMeta.UID, string(ns.allPods.ipsetName)); err != nil {
 					return err
 				}
 			}
 			if newMatch {
-				if err := selector.addEntry(string(ns.allPods.ipsetName), namespaceComment(ns)); err != nil {
+				if err := selector.addEntry(ns.namespace.ObjectMeta.UID, string(ns.allPods.ipsetName), namespaceComment(ns)); err != nil {
 					return err
 				}
 			}
@@ -468,7 +471,7 @@ func (ns *ns) deleteNamespace(obj *coreapi.Namespace) error {
 	}
 
 	// Remove namespace ipset from any matching namespace selectors
-	return ns.nsSelectors.delFromMatching(obj.ObjectMeta.Labels, string(ns.allPods.ipsetName))
+	return ns.nsSelectors.delFromMatching(obj.ObjectMeta.UID, obj.ObjectMeta.Labels, string(ns.allPods.ipsetName))
 }
 
 func (ns *ns) isDefaultDeny(namespace *coreapi.Namespace) bool {

npc/selector.go

@@ -46,12 +46,12 @@ func (s *selector) matches(labelMap map[string]string) bool {
 	return s.spec.selector.Matches(labels.Set(labelMap))
 }
 
-func (s *selector) addEntry(entry string, comment string) error {
+func (s *selector) addEntry(user types.UID, entry string, comment string) error {
-	return s.ips.AddEntry(s.spec.ipsetName, entry, comment)
+	return s.ips.AddEntry(user, s.spec.ipsetName, entry, comment)
 }
 
-func (s *selector) delEntry(entry string) error {
+func (s *selector) delEntry(user types.UID, entry string) error {
-	return s.ips.DelEntry(s.spec.ipsetName, entry)
+	return s.ips.DelEntry(user, s.spec.ipsetName, entry)
 }
 
 type selectorFn func(selector *selector) error
@@ -85,14 +85,14 @@ func newSelectorSet(ips ipset.Interface, onNewSelector, onNewDstSelector selecto
 		dstSelectorsCount:    make(map[string]int)}
 }
 
-func (ss *selectorSet) addToMatching(labelMap map[string]string, entry string, comment string) (bool, error) {
+func (ss *selectorSet) addToMatching(user types.UID, labelMap map[string]string, entry string, comment string) (bool, error) {
 	found := false
 	for _, s := range ss.entries {
 		if s.matches(labelMap) {
 			if ss.dstSelectorExist(s) {
 				found = true
 			}
-			if err := s.addEntry(entry, comment); err != nil {
+			if err := s.addEntry(user, entry, comment); err != nil {
 				return found, err
 			}
 		}
@@ -100,10 +100,10 @@ func (ss *selectorSet) addToMatching(labelMap map[string]string, entry string, c
 	return found, nil
 }
 
-func (ss *selectorSet) delFromMatching(labelMap map[string]string, entry string) error {
+func (ss *selectorSet) delFromMatching(user types.UID, labelMap map[string]string, entry string) error {
 	for _, s := range ss.entries {
 		if s.matches(labelMap) {
-			if err := s.delEntry(entry); err != nil {
+			if err := s.delEntry(user, entry); err != nil {
 				return err
 			}
 		}

prog/kube-peers/annotations.go

@@ -73,8 +73,10 @@ const (
 	retryPeriod  = time.Second * 2
 	jitterFactor = 1.0
 
+	// Prefix all our annotation keys with this string so they don't clash with anyone else's
+	KubePeersPrefix = "kube-peers.weave.works/"
 	// KubePeersAnnotationKey is the default annotation key
-	KubePeersAnnotationKey = "kube-peers.weave.works/peers"
+	KubePeersAnnotationKey = KubePeersPrefix + "peers"
 )
 
 func (cml *configMapAnnotations) Init() error {
@@ -130,46 +132,59 @@ func (cml *configMapAnnotations) UpdatePeerList(list peerList) error {
 	return cml.UpdateAnnotation(KubePeersAnnotationKey, string(recordBytes))
 }
 
-func (cml *configMapAnnotations) UpdateAnnotation(key, value string) error {
+// Clean up a string so it meets the Kubernetes requiremements for Annotation keys:
+// name part must consist of alphanumeric characters, '-', '_' or '.', and must
+// start and end with an alphanumeric character (e.g. 'MyName', or 'my.name', or '123-abc')
+func cleanKey(key string) string {
+	buf := []byte(key)
+	for i, c := range buf {
+		if (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '-' || c == '_' || c == '.' || c == '/' {
+			continue
+		}
+		buf[i] = '_'
+	}
+	return string(buf)
+}
+
+func (cml *configMapAnnotations) GetAnnotation(key string) (string, bool) {
+	value, ok := cml.cm.Annotations[cleanKey(key)]
+	return value, ok
+}
+
+func (cml *configMapAnnotations) UpdateAnnotation(key, value string) (err error) {
 	if cml.cm == nil {
 		return errors.New("endpoint not initialized, call Init first")
 	}
-	cm := cml.cm
+	// speculatively change the state, then replace with whatever comes back
-	cm.Annotations[key] = value
+	// from Update(), which will be the latest on the server whatever happened
-	cm, err := cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
+	cml.cm.Annotations[cleanKey(key)] = value
-	if err == nil {
+	cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
-		cml.cm = cm
-	}
 	return err
 }
 
-func (cml *configMapAnnotations) RemoveAnnotation(key string) error {
+func (cml *configMapAnnotations) RemoveAnnotation(key string) (err error) {
 	if cml.cm == nil {
 		return errors.New("endpoint not initialized, call Init first")
 	}
-	cm := cml.cm
+	// speculatively change the state, then replace with whatever comes back
-	delete(cm.Annotations, key)
+	// from Update(), which will be the latest on the server whatever happened
-	cm, err := cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
+	delete(cml.cm.Annotations, cleanKey(key))
-	if err == nil {
+	cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
-		cml.cm = cm
-	}
 	return err
 }
 
-func (cml *configMapAnnotations) RemoveAnnotationsWithValue(valueToRemove string) error {
+func (cml *configMapAnnotations) RemoveAnnotationsWithValue(valueToRemove string) (err error) {
 	if cml.cm == nil {
 		return errors.New("endpoint not initialized, call Init first")
 	}
-	cm := cml.cm
+	// speculatively change the state, then replace with whatever comes back
-	for key, value := range cm.Annotations {
+	// from Update(), which will be the latest on the server whatever happened
+	for key, value := range cml.cm.Annotations {
 		if value == valueToRemove {
-			delete(cm.Annotations, key)
+			delete(cml.cm.Annotations, key) // don't need to clean this key as it came from the map
 		}
 	}
-	cm, err := cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
+	cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
-	if err == nil {
-		cml.cm = cm
-	}
 	return err
 }
 

prog/kube-peers/main.go

@@ -98,6 +98,9 @@ func checkIamInPeerList(cml *configMapAnnotations, c *kubernetes.Clientset, peer
 // Kubernetes, remove it from Weave IPAM
 func reclaimRemovedPeers(weave *weaveapi.Client, cml *configMapAnnotations, nodes []nodeInfo, myPeerName string) error {
 	for {
+		if err := cml.Init(); err != nil {
+			return err
+		}
 		// 1. Compare peers stored in the peerList against all peers reported by k8s now.
 		storedPeerList, err := cml.GetPeerList()
 		if err != nil {
@@ -119,7 +122,7 @@ func reclaimRemovedPeers(weave *weaveapi.Client, cml *configMapAnnotations, node
 			common.Log.Debugln("[kube-peers] Preparing to remove disappeared peer", peer)
 			okToRemove := false
 			// 3. Check if there is an existing annotation with key X
-			if existingAnnotation, found := cml.cm.Annotations[peer.PeerName]; found {
+			if existingAnnotation, found := cml.GetAnnotation(KubePeersPrefix + peer.PeerName); found {
 				common.Log.Debugln("[kube-peers] Existing annotation", existingAnnotation)
 				// 4.   If annotation already contains my identity, ok;
 				if existingAnnotation == myPeerName {
@@ -128,14 +131,16 @@ func reclaimRemovedPeers(weave *weaveapi.Client, cml *configMapAnnotations, node
 			} else {
 				// 5.   If non-existent, write an annotation with key X and contents "my identity"
 				common.Log.Debugln("[kube-peers] Noting I plan to remove ", peer.PeerName)
-				if err := cml.UpdateAnnotation(peer.PeerName, myPeerName); err == nil {
+				if err := cml.UpdateAnnotation(KubePeersPrefix+peer.PeerName, myPeerName); err == nil {
 					okToRemove = true
+				} else {
+					common.Log.Debugln("[kube-peers] error from UpdateAnnotation: ", err)
 				}
 			}
 			if okToRemove {
 				// 6.   If step 4 or 5 succeeded, rmpeer X
 				result, err := weave.RmPeer(peer.PeerName)
-				common.Log.Infoln("[kube-peers] rmpeer of", peer.PeerName, ":", result)
+				common.Log.Infof("[kube-peers] rmpeer of %s: %s", peer.PeerName, result)
 				if err != nil {
 					return err
 				}
@@ -146,15 +151,12 @@ func reclaimRemovedPeers(weave *weaveapi.Client, cml *configMapAnnotations, node
 					}
 					// 7a.    Remove X from peerList
 					storedPeerList.remove(peer.PeerName)
-					common.Log.Infoln("[kube-peers] Removing peer ", peer.PeerName, ". Expecting to remove linked annotation next.")
 					if err := cml.UpdatePeerList(*storedPeerList); err != nil {
 						return err
 					}
-					common.Log.Infoln("[kube-peers] Removing annotation ", peer.PeerName)
 					// 7b.    Remove annotation with key X
-					return cml.RemoveAnnotation(peer.PeerName)
+					return cml.RemoveAnnotation(KubePeersPrefix + peer.PeerName)
 				})
-				common.Log.Debugln("[kube-peers] Finished removal of ", peer.PeerName)
 			}
 			// 8.   If step 5 failed due to optimistic lock conflict, stop: someone else is handling X
 

site/kubernetes/kube-addon.md

@@ -156,7 +156,7 @@ labels. For more information on configuring network policies in
 Kubernetes see the
 [walkthrough](http://kubernetes.io/docs/getting-started-guides/network-policy/walkthrough/)
 and the [NetworkPolicy API object
-definition](http://kubernetes.io/docs/api-reference/extensions/v1beta1/definitions/#_v1beta1_networkpolicy).
+definition](https://v1-7.docs.kubernetes.io/docs/api-reference/v1.7/#networkpolicy-v1-networking)
 
 **Note:** as of version 1.9 of Weave Net, the Network Policy
   Controller allows all multicast traffic. Since a single multicast
@@ -299,6 +299,7 @@ You can customise the YAML you get from `cloud.weave.works` by passing some of W
   - `disable-npc`: boolean (`true|false`). Default: `false`.
   - `env.NAME=VALUE`: add environment variable `NAME` and set it to `VALUE`.
   - `seLinuxOptions.NAME=VALUE`: add SELinux option `NAME` and set it to `VALUE`, e.g. `seLinuxOptions.type=spc_t`
+  - `use-legacy-netpol`: use [legacy NetworkPolicy semantics](https://v1-7.docs.kubernetes.io/docs/api-reference/v1.7/#networkpolicy-v1-networking), boolean (`true|false`). Default: `true` for Kubernetes version <= 1.6, `false` for > 1.6.
 
 The list of variables you can set is: