Ignore updates for pods on other nodes

npc/controller.go

@@ -29,6 +29,8 @@ type NetworkPolicyController interface {
 type controller struct {
 	sync.Mutex
 
+	nodeName string // my node name
+
 	ipt *iptables.IPTables
 	ips ipset.Interface
 
@@ -36,11 +38,12 @@ type controller struct {
 	nsSelectors *selectorSet   // selector string -> nsSelector
 }
 
-func New(ipt *iptables.IPTables, ips ipset.Interface) NetworkPolicyController {
+func New(nodeName string, ipt *iptables.IPTables, ips ipset.Interface) NetworkPolicyController {
 	c := &controller{
-		ipt: ipt,
-		ips: ips,
-		nss: make(map[string]*ns)}
+		nodeName: nodeName,
+		ipt:      ipt,
+		ips:      ips,
+		nss:      make(map[string]*ns)}
 
 	c.nsSelectors = newSelectorSet(ips, c.onNewNsSelector)
 
@@ -82,10 +85,21 @@ 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")
@@ -96,6 +110,10 @@ 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")
@@ -106,6 +124,9 @@ 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")

prog/weave-npc/main.go

@@ -28,6 +28,7 @@ var (
 	metricsAddr string
 	logLevel    string
 	allowMcast  bool
+	nodeName    string
 )
 
 func handleError(err error) { common.CheckFatal(err) }
@@ -116,7 +117,14 @@ func createBaseRules(ipt *iptables.IPTables, ips ipset.Interface) error {
 
 func root(cmd *cobra.Command, args []string) {
 	common.SetLogLevel(logLevel)
-	common.Log.Infof("Starting Weaveworks NPC %s", version)
+	if nodeName == "" {
+		// HOSTNAME is set by Kubernetes for pods in the host network namespace
+		nodeName = os.Getenv("HOSTNAME")
+	}
+	if nodeName == "" {
+		common.Log.Fatalf("Must set node name via --node-name or $HOSTNAME")
+	}
+	common.Log.Infof("Starting Weaveworks NPC %s; node name %q", version, nodeName)
 
 	if err := metrics.Start(metricsAddr); err != nil {
 		common.Log.Fatalf("Failed to start metrics: %v", err)
@@ -141,7 +149,7 @@ func root(cmd *cobra.Command, args []string) {
 	handleError(resetIPSets(ips))
 	handleError(createBaseRules(ipt, ips))
 
-	npc := npc.New(ipt, ips)
+	npc := npc.New(nodeName, ipt, ips)
 
 	nsController := makeController(client.Core().RESTClient(), "namespaces", &coreapi.Namespace{},
 		cache.ResourceEventHandlerFuncs{
@@ -221,6 +229,7 @@ func main() {
 	rootCmd.PersistentFlags().StringVar(&metricsAddr, "metrics-addr", ":6781", "metrics server bind address")
 	rootCmd.PersistentFlags().StringVar(&logLevel, "log-level", "debug", "logging level (debug, info, warning, error)")
 	rootCmd.PersistentFlags().BoolVar(&allowMcast, "allow-mcast", true, "allow all multicast traffic")
+	rootCmd.PersistentFlags().StringVar(&nodeName, "node-name", "", "only generate rules that apply to this node")
 
 	handleError(rootCmd.Execute())
 }