Merge pull request #2585 from weaveworks/merge-weave-npc

Merge weave npc

.gitignore

@@ -46,6 +46,7 @@ prog/weaveutil/weaveutil
 prog/*/Dockerfile.weaveworks
 prog/kube-peers/kube-peers
 prog/weave-kube/kube-peers
+prog/weave-npc/weave-npc
 prog/plugin/plugin
 testing/cover/cover
 testing/runner/runner

.gitmodules

@@ -106,3 +106,15 @@
 [submodule "vendor/k8s.io/client-go"]
 	path = vendor/k8s.io/client-go
 	url = https://github.com/kubernetes/client-go
+[submodule "vendor/github.com/pkg/errors"]
+	path = vendor/github.com/pkg/errors
+	url = https://github.com/pkg/errors
+[submodule "vendor/github.com/spf13/cobra"]
+	path = vendor/github.com/spf13/cobra
+	url = https://github.com/spf13/cobra
+[submodule "vendor/github.com/spf13/plag"]
+	path = vendor/github.com/spf13/plag
+	url = https://github.com/spf13/pflag
+[submodule "vendor/github.com/spf13/pflag"]
+	path = vendor/github.com/spf13/pflag
+	url = https://github.com/spf13/pflag

Makefile

@@ -1,4 +1,4 @@
-PUBLISH=publish_weave publish_weaveexec publish_plugin publish_weave-kube
+PUBLISH=publish_weave publish_weaveexec publish_plugin publish_weave-kube publish_weave-npc
 
 .DEFAULT: all
 .PHONY: all exes testrunner update tests lint publish $(PUBLISH) clean clean-bin prerequisites build run-smoketests
@@ -17,6 +17,7 @@ WEAVER_EXE=prog/weaver/weaver
 WEAVEPROXY_EXE=prog/weaveproxy/weaveproxy
 SIGPROXY_EXE=prog/sigproxy/sigproxy
 KUBEPEERS_EXE=prog/kube-peers/kube-peers
+WEAVENPC_EXE=prog/weave-npc/weave-npc
 WEAVEWAIT_EXE=prog/weavewait/weavewait
 WEAVEWAIT_NOOP_EXE=prog/weavewait/weavewait_noop
 WEAVEWAIT_NOMCAST_EXE=prog/weavewait/weavewait_nomcast
@@ -25,25 +26,27 @@ PLUGIN_EXE=prog/plugin/plugin
 RUNNER_EXE=tools/runner/runner
 TEST_TLS_EXE=test/tls/tls
 
-EXES=$(WEAVER_EXE) $(SIGPROXY_EXE) $(KUBEPEERS_EXE) $(WEAVEPROXY_EXE) $(WEAVEWAIT_EXE) $(WEAVEWAIT_NOOP_EXE) $(WEAVEWAIT_NOMCAST_EXE) $(WEAVEUTIL_EXE) $(PLUGIN_EXE) $(TEST_TLS_EXE)
+EXES=$(WEAVER_EXE) $(SIGPROXY_EXE) $(KUBEPEERS_EXE) $(WEAVENPC_EXE) $(WEAVEPROXY_EXE) $(WEAVEWAIT_EXE) $(WEAVEWAIT_NOOP_EXE) $(WEAVEWAIT_NOMCAST_EXE) $(WEAVEUTIL_EXE) $(PLUGIN_EXE) $(TEST_TLS_EXE)
 
 BUILD_UPTODATE=.build.uptodate
 WEAVER_UPTODATE=.weaver.uptodate
 WEAVEEXEC_UPTODATE=.weaveexec.uptodate
 PLUGIN_UPTODATE=.plugin.uptodate
 WEAVEKUBE_UPTODATE=.weavekube.uptodate
+WEAVENPC_UPTODATE=.weavenpc.uptodate
 WEAVEDB_UPTODATE=.weavedb.uptodate
 
-IMAGES_UPTODATE=$(WEAVER_UPTODATE) $(WEAVEEXEC_UPTODATE) $(PLUGIN_UPTODATE) $(WEAVEKUBE_UPTODATE) $(WEAVEDB_UPTODATE)
+IMAGES_UPTODATE=$(WEAVER_UPTODATE) $(WEAVEEXEC_UPTODATE) $(PLUGIN_UPTODATE) $(WEAVEKUBE_UPTODATE) $(WEAVENPC_UPTODATE) $(WEAVEDB_UPTODATE)
 
 WEAVER_IMAGE=$(DOCKERHUB_USER)/weave
 WEAVEEXEC_IMAGE=$(DOCKERHUB_USER)/weaveexec
 PLUGIN_IMAGE=$(DOCKERHUB_USER)/plugin
 WEAVEKUBE_IMAGE=$(DOCKERHUB_USER)/weave-kube
+WEAVENPC_IMAGE=$(DOCKERHUB_USER)/weave-npc
 BUILD_IMAGE=$(DOCKERHUB_USER)/weavebuild
 WEAVEDB_IMAGE=$(DOCKERHUB_USER)/weavedb
 
-IMAGES=$(WEAVER_IMAGE) $(WEAVEEXEC_IMAGE) $(PLUGIN_IMAGE) $(WEAVEKUBE_IMAGE) $(WEAVEDB_IMAGE)
+IMAGES=$(WEAVER_IMAGE) $(WEAVEEXEC_IMAGE) $(PLUGIN_IMAGE) $(WEAVEKUBE_IMAGE) $(WEAVENPC_IMAGE) $(WEAVEDB_IMAGE)
 
 WEAVE_EXPORT=weave.tar.gz
 
@@ -71,6 +74,7 @@ $(WEAVEPROXY_EXE): proxy/*.go prog/weaveproxy/*.go
 $(WEAVEUTIL_EXE): prog/weaveutil/*.go net/*.go
 $(SIGPROXY_EXE): prog/sigproxy/*.go
 $(KUBEPEERS_EXE): prog/kube-peers/*.go
+$(WEAVENPC_EXE): prog/weave-npc/*.go npc/*.go npc/*/*.go
 $(PLUGIN_EXE): prog/plugin/*.go plugin/*/*.go api/*.go common/*.go common/docker/*.go net/*.go
 $(TEST_TLS_EXE): test/tls/*.go
 $(WEAVEWAIT_NOOP_EXE): prog/weavewait/*.go
@@ -103,7 +107,7 @@ else
 endif
 	$(NETGO_CHECK)
 
-$(WEAVEUTIL_EXE) $(KUBEPEERS_EXE):
+$(WEAVEUTIL_EXE) $(KUBEPEERS_EXE) $(WEAVENPC_EXE):
 	go build $(BUILD_FLAGS) -o $@ ./$(@D)
 	$(NETGO_CHECK)
 
@@ -160,6 +164,10 @@ $(WEAVEKUBE_UPTODATE): prog/weave-kube/Dockerfile.$(DOCKERHUB_USER) prog/weave-k
 	$(SUDO) docker build -f prog/weave-kube/Dockerfile.$(DOCKERHUB_USER) -t $(WEAVEKUBE_IMAGE) prog/weave-kube
 	touch $@
 
+$(WEAVENPC_UPTODATE): prog/weave-npc/Dockerfile $(WEAVENPC_EXE) prog/weave-npc/ulogd.conf
+	$(SUDO) docker build -f prog/weave-npc/Dockerfile -t $(WEAVENPC_IMAGE) prog/weave-npc
+	touch $@
+
 $(WEAVEDB_UPTODATE): prog/weavedb/Dockerfile
 	$(SUDO) docker build -t $(WEAVEDB_IMAGE) prog/weavedb
 	touch $@

docs/weavenpc-design.md

@@ -0,0 +1,100 @@
+# Overview
+
+# ipsets
+
+The policy controller maintains a number of ipsets which are
+subsequently referred to by the iptables rules used to effect network
+policy specifications. These ipsets are created, modified and
+destroyed automatically in response to Pod, Namespace and
+NetworkPolicy object updates from the k8s API server:
+
+* A `hash:ip` set per namespace, containing the IP addresses of all
+  pods in that namespace
+* A `list:set` per distinct (across all network policies in all
+  namespaces) namespace selector mentioned in a network policy,
+  containing the names of any of the above hash:ip sets whose
+  corresponding namespace labels match the selector
+* A `hash:ip` set for each distinct (within the scope of the
+  containing network policy's namespace) pod selector mentioned in a
+  network policy, containing the IP addresses of all pods in the
+  namespace whose labels match that selector
+
+ipset names are generated deterministically from a string
+representation of the corresponding label selector. Because ipset
+names are limited to 31 characters in length, this is done by taking a
+SHA hash of the selector string and then printing that out as a base
+85 string with a "weave-" prefix e.g.:
+
+    weave-k?Z;25^M}|1s7P3|H9i;*;MhG
+
+Because pod selectors are scoped to a namespace, we need to make sure
+that if the same selector definition is used in different namespaces
+that we maintain distinct ipsets. Consequently, for such selectors the
+namespace name is prepended to the label selector string before
+hashing to avoid clashes.
+
+# iptables chains
+
+The policy controller maintains two iptables chains in response to
+changes to pods, namespaces and network policies. One chain contains
+the ingress rules that implement the network policy specifications,
+and the other is used to bypass the ingress rules for namespaces which
+have an ingress isolation policy of `DefaultAllow`.
+
+## Dynamically maintained `WEAVE-NPC-DEFAULT` chain
+
+The policy controller maintains a rule in this chain for every
+namespace whose ingress isolation policy is `DefaultAllow`. The
+purpose of this rule is simply to ACCEPT any traffic destined for such
+namespaces before it reaches the ingress chain.
+
+```
+iptables -A WEAVE-NPC-DEFAULT -m set --match-set $NSIPSET dst -j ACCEPT
+```
+
+## Dynamically maintained `WEAVE-NPC-INGRESS` chain
+
+For each namespace network policy ingress rule peer/port combination:
+
+```
+iptables -A WEAVE-NPC-INGRESS -p $PROTO [-m set --match-set $SRCSET] -m set --match-set $DSTSET --dport $DPORT -j ACCEPT
+```
+
+## Static `WEAVE-NPC` chain
+
+Static configuration:
+
+```
+iptables -A WEAVE-NPC -m state --state RELATED,ESTABLISHED -j ACCEPT
+iptables -A WEAVE-NPC -m state --state NEW -j WEAVE-NPC-DEFAULT
+iptables -A WEAVE-NPC -m state --state NEW -j WEAVE-NPC-INGRESS
+```
+
+# Steering traffic into the policy engine
+
+To direct traffic into the policy engine:
+
+iptables -A FORWARD -o weave -m physdev ! --physdev-out vethwe-bridge -j WEAVE-NPC
+iptables -A FORWARD -o weave -m physdev ! --physdev-out vethwe-bridge -j DROP
+
+Note this only affects traffic which egresses the bridge on a physical
+port which is not the Weave Net router - in other words, it is
+destined for an application container veth. The following traffic is
+affected:
+
+* Traffic bridged between local application containers
+* Traffic bridged from the router to a local application container
+* Traffic originating from the internet destined for nodeports - this
+  is routed via the FORWARD chain to a container pod IP after DNAT
+
+The following traffic is NOT affected:
+
+* Traffic bridged from a local application container to the router
+* Traffic originating from processes in the host network namespace
+  (e.g. kubelet health checks)
+* Traffic routed from an application container to the internet
+
+See these resources for helpful context:
+
+* http://ebtables.netfilter.org/br_fw_ia/br_fw_ia.html
+* https://commons.wikimedia.org/wiki/File:Netfilter-packet-flow.svg

npc/analyser.go

@@ -0,0 +1,114 @@
+package npc
+
+import (
+	"fmt"
+
+	"k8s.io/client-go/pkg/api"
+	extnapi "k8s.io/client-go/pkg/apis/extensions/v1beta1"
+	"k8s.io/client-go/pkg/util/intstr"
+
+	"github.com/weaveworks/weave/npc/ipset"
+)
+
+func (ns *ns) analysePolicy(policy *extnapi.NetworkPolicy) (
+	rules map[string]*ruleSpec,
+	nsSelectors, podSelectors map[string]*selectorSpec,
+	err error) {
+
+	nsSelectors = make(map[string]*selectorSpec)
+	podSelectors = make(map[string]*selectorSpec)
+	rules = make(map[string]*ruleSpec)
+
+	dstSelector, err := newSelectorSpec(&policy.Spec.PodSelector, ns.name, ipset.HashIP)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	podSelectors[dstSelector.key] = dstSelector
+
+	for _, ingressRule := range policy.Spec.Ingress {
+		if ingressRule.Ports != nil && len(ingressRule.Ports) == 0 {
+			// Ports is empty, this rule matches no ports (no traffic matches).
+			continue
+		}
+
+		if ingressRule.From != nil && len(ingressRule.From) == 0 {
+			// From is empty, this rule matches no sources (no traffic matches).
+			continue
+		}
+
+		if ingressRule.From == nil {
+			// From is not provided, this rule matches all sources (traffic not restricted by source).
+			if ingressRule.Ports == nil {
+				// Ports is not provided, this rule matches all ports (traffic not restricted by port).
+				rule := newRuleSpec(nil, nil, dstSelector, nil)
+				rules[rule.key] = rule
+			} else {
+				// Ports is present and contains at least one item, then this rule allows traffic
+				// only if the traffic matches at least one port in the ports list.
+				withNormalisedProtoAndPort(ingressRule.Ports, func(proto, port string) {
+					rule := newRuleSpec(&proto, nil, dstSelector, &port)
+					rules[rule.key] = rule
+				})
+			}
+		} else {
+			// From is present and contains at least on item, this rule allows traffic only if the
+			// traffic matches at least one item in the from list.
+			for _, peer := range ingressRule.From {
+				var srcSelector *selectorSpec
+				if peer.PodSelector != nil {
+					srcSelector, err = newSelectorSpec(peer.PodSelector, ns.name, ipset.HashIP)
+					if err != nil {
+						return nil, nil, nil, err
+					}
+					podSelectors[srcSelector.key] = srcSelector
+				}
+				if peer.NamespaceSelector != nil {
+					srcSelector, err = newSelectorSpec(peer.NamespaceSelector, "", ipset.ListSet)
+					if err != nil {
+						return nil, nil, nil, err
+					}
+					nsSelectors[srcSelector.key] = srcSelector
+				}
+
+				if ingressRule.Ports == nil {
+					// Ports is not provided, this rule matches all ports (traffic not restricted by port).
+					rule := newRuleSpec(nil, srcSelector, dstSelector, nil)
+					rules[rule.key] = rule
+				} else {
+					// Ports is present and contains at least one item, then this rule allows traffic
+					// only if the traffic matches at least one port in the ports list.
+					withNormalisedProtoAndPort(ingressRule.Ports, func(proto, port string) {
+						rule := newRuleSpec(&proto, srcSelector, dstSelector, &port)
+						rules[rule.key] = rule
+					})
+				}
+			}
+		}
+	}
+
+	return rules, nsSelectors, podSelectors, nil
+}
+
+func withNormalisedProtoAndPort(npps []extnapi.NetworkPolicyPort, f func(proto, port string)) {
+	for _, npp := range npps {
+		// If no proto is specified, default to TCP
+		proto := string(api.ProtocolTCP)
+		if npp.Protocol != nil {
+			proto = string(*npp.Protocol)
+		}
+
+		// If no port is specified, match any port. Let iptables executable handle
+		// service name resolution
+		port := "0:65535"
+		if npp.Port != nil {
+			switch npp.Port.Type {
+			case intstr.Int:
+				port = fmt.Sprintf("%d", npp.Port.IntVal)
+			case intstr.String:
+				port = npp.Port.StrVal
+			}
+		}
+
+		f(proto, port)
+	}
+}

npc/betatypes.go

@@ -0,0 +1,40 @@
+package npc
+
+// These types are defined in https://github.com/kubernetes/kubernetes/blob/master/docs/proposals/network-policy.md
+// but not in the k8s API yet. Copies are included here for decoding the `net.beta.kubernetes.io/network-policy`
+// annotation specified in the above document.
+
+type IngressIsolationPolicy string
+
+const (
+	// Deny all ingress traffic to pods in this namespace. Ingress means
+	// any incoming traffic to pods, whether that be from other pods within this namespace
+	// or any source outside of this namespace.
+	DefaultDeny IngressIsolationPolicy = "DefaultDeny"
+)
+
+// Standard NamespaceSpec object, modified to include a new
+// NamespaceNetworkPolicy field.
+type NamespaceSpec struct {
+	// This is a pointer so that it can be left undefined.
+	NetworkPolicy *NamespaceNetworkPolicy `json:"networkPolicy,omitempty"`
+}
+
+type NamespaceNetworkPolicy struct {
+	// Ingress configuration for this namespace.  This config is
+	// applied to all pods within this namespace. For now, only
+	// ingress is supported.  This field is optional - if not
+	// defined, then the cluster default for ingress is applied.
+	Ingress *NamespaceIngressPolicy `json:"ingress,omitempty"`
+}
+
+// Configuration for ingress to pods within this namespace.
+// For now, this only supports specifying an isolation policy.
+type NamespaceIngressPolicy struct {
+	// The isolation policy to apply to pods in this namespace.
+	// Currently this field only supports "DefaultDeny", but could
+	// be extended to support other policies in the future.  When set to DefaultDeny,
+	// pods in this namespace are denied ingress traffic by default.  When not defined,
+	// the cluster default ingress isolation policy is applied (currently allow all).
+	Isolation *IngressIsolationPolicy `json:"isolation,omitempty"`
+}

npc/constants.go

@@ -0,0 +1,9 @@
+package npc
+
+const (
+	TableFilter = "filter"
+
+	MainChain    = "WEAVE-NPC"
+	DefaultChain = "WEAVE-NPC-DEFAULT"
+	IngressChain = "WEAVE-NPC-INGRESS"
+)

npc/controller.go

@@ -0,0 +1,173 @@
+package npc
+
+import (
+	"sync"
+
+	log "github.com/Sirupsen/logrus"
+	"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/npc/ipset"
+)
+
+type NetworkPolicyController interface {
+	AddNamespace(ns *coreapi.Namespace) error
+	UpdateNamespace(oldObj, newObj *coreapi.Namespace) error
+	DeleteNamespace(ns *coreapi.Namespace) error
+
+	AddPod(obj *coreapi.Pod) error
+	UpdatePod(oldObj, newObj *coreapi.Pod) error
+	DeletePod(obj *coreapi.Pod) error
+
+	AddNetworkPolicy(obj *extnapi.NetworkPolicy) error
+	UpdateNetworkPolicy(oldObj, newObj *extnapi.NetworkPolicy) error
+	DeleteNetworkPolicy(obj *extnapi.NetworkPolicy) error
+}
+
+type controller struct {
+	sync.Mutex
+
+	ipt *iptables.IPTables
+	ips ipset.Interface
+
+	nss         map[string]*ns // ns name -> ns struct
+	nsSelectors *selectorSet   // selector string -> nsSelector
+}
+
+func New(ipt *iptables.IPTables, ips ipset.Interface) NetworkPolicyController {
+	c := &controller{
+		ipt: ipt,
+		ips: ips,
+		nss: make(map[string]*ns)}
+
+	c.nsSelectors = newSelectorSet(ips, c.onNewNsSelector)
+
+	return c
+}
+
+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 := ns.ips.AddEntry(selector.spec.ipsetName, string(ns.allPods.ipsetName)); 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.ipt, npc.ips, npc.nsSelectors)
+		if err != nil {
+			return err
+		}
+		npc.nss[name] = newNs
+		ns = newNs
+	}
+	if err := f(ns); err != nil {
+		return err
+	}
+	if ns.empty() {
+		if err := ns.destroy(); err != nil {
+			return err
+		}
+		delete(npc.nss, name)
+	}
+	return nil
+}
+
+func (npc *controller) AddPod(obj *coreapi.Pod) error {
+	npc.Lock()
+	defer npc.Unlock()
+
+	log.Infof("EVENT AddPod %#v", obj)
+	return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error {
+		return errors.Wrap(ns.addPod(obj), "add pod")
+	})
+}
+
+func (npc *controller) UpdatePod(oldObj, newObj *coreapi.Pod) error {
+	npc.Lock()
+	defer npc.Unlock()
+
+	log.Infof("EVENT UpdatePod %#v %#v", oldObj, newObj)
+	return npc.withNS(oldObj.ObjectMeta.Namespace, func(ns *ns) error {
+		return errors.Wrap(ns.updatePod(oldObj, newObj), "update pod")
+	})
+}
+
+func (npc *controller) DeletePod(obj *coreapi.Pod) error {
+	npc.Lock()
+	defer npc.Unlock()
+
+	log.Infof("EVENT DeletePod %#v", obj)
+	return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error {
+		return errors.Wrap(ns.deletePod(obj), "delete pod")
+	})
+}
+
+func (npc *controller) AddNetworkPolicy(obj *extnapi.NetworkPolicy) error {
+	npc.Lock()
+	defer npc.Unlock()
+
+	log.Infof("EVENT AddNetworkPolicy %#v", obj)
+	return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error {
+		return errors.Wrap(ns.addNetworkPolicy(obj), "add network policy")
+	})
+}
+
+func (npc *controller) UpdateNetworkPolicy(oldObj, newObj *extnapi.NetworkPolicy) error {
+	npc.Lock()
+	defer npc.Unlock()
+
+	log.Infof("EVENT UpdateNetworkPolicy %#v %#v", oldObj, newObj)
+	return npc.withNS(oldObj.ObjectMeta.Namespace, func(ns *ns) error {
+		return errors.Wrap(ns.updateNetworkPolicy(oldObj, newObj), "update network policy")
+	})
+}
+
+func (npc *controller) DeleteNetworkPolicy(obj *extnapi.NetworkPolicy) error {
+	npc.Lock()
+	defer npc.Unlock()
+
+	log.Infof("EVENT DeleteNetworkPolicy %#v", obj)
+	return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error {
+		return errors.Wrap(ns.deleteNetworkPolicy(obj), "delete network policy")
+	})
+}
+
+func (npc *controller) AddNamespace(obj *coreapi.Namespace) error {
+	npc.Lock()
+	defer npc.Unlock()
+
+	log.Infof("EVENT AddNamespace %#v", obj)
+	return npc.withNS(obj.ObjectMeta.Name, func(ns *ns) error {
+		return errors.Wrap(ns.addNamespace(obj), "add namespace")
+	})
+}
+
+func (npc *controller) UpdateNamespace(oldObj, newObj *coreapi.Namespace) error {
+	npc.Lock()
+	defer npc.Unlock()
+
+	log.Infof("EVENT UpdateNamespace%#v %#v", oldObj, newObj)
+	return npc.withNS(oldObj.ObjectMeta.Name, func(ns *ns) error {
+		return errors.Wrap(ns.updateNamespace(oldObj, newObj), "update namespace")
+	})
+}
+
+func (npc *controller) DeleteNamespace(obj *coreapi.Namespace) error {
+	npc.Lock()
+	defer npc.Unlock()
+
+	log.Infof("EVENT DeleteNamespace %#v", obj)
+	return npc.withNS(obj.ObjectMeta.Name, func(ns *ns) error {
+		return errors.Wrap(ns.deleteNamespace(obj), "delete namespace")
+	})
+}

npc/ipset/ipset.go

@@ -0,0 +1,69 @@
+package ipset
+
+import (
+	"os/exec"
+
+	"github.com/pkg/errors"
+)
+
+type Name string
+
+type Type string
+
+const (
+	ListSet = Type("list:set")
+	HashIP  = Type("hash:ip")
+)
+
+type Interface interface {
+	Create(ipsetName Name, ipsetType Type) error
+	AddEntry(ipsetName Name, entry string) error
+	DelEntry(ipsetName Name, entry string) error
+	Flush(ipsetName Name) error
+	Destroy(ipsetName Name) error
+
+	FlushAll() error
+	DestroyAll() error
+}
+
+type ipset struct {
+}
+
+func New() Interface {
+	return &ipset{}
+}
+
+func (i *ipset) Create(ipsetName Name, ipsetType Type) error {
+	return doExec("create", string(ipsetName), string(ipsetType))
+}
+
+func (i *ipset) AddEntry(ipsetName Name, entry string) error {
+	return doExec("add", string(ipsetName), entry)
+}
+
+func (i *ipset) DelEntry(ipsetName Name, entry string) error {
+	return doExec("del", string(ipsetName), entry)
+}
+
+func (i *ipset) Flush(ipsetName Name) error {
+	return doExec("flush", string(ipsetName))
+}
+
+func (i *ipset) FlushAll() error {
+	return doExec("flush")
+}
+
+func (i *ipset) Destroy(ipsetName Name) error {
+	return doExec("destroy", string(ipsetName))
+}
+
+func (i *ipset) DestroyAll() error {
+	return doExec("destroy")
+}
+
+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
+}

npc/metrics/metrics.go

@@ -0,0 +1,78 @@
+package metrics
+
+import (
+	log "github.com/Sirupsen/logrus"
+	"net/http"
+	"os"
+	"strconv"
+
+	"github.com/google/gopacket"
+	"github.com/google/gopacket/layers"
+	"github.com/google/gopacket/pcapgo"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+var (
+	blockedConnections = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "weavenpc_blocked_connections_total",
+			Help: "Connection attempts blocked by policy controller.",
+		},
+		[]string{"protocol", "dport"},
+	)
+)
+
+func gatherMetrics() {
+	pipe, err := os.Open("/var/log/ulogd.pcap")
+	if err != nil {
+		log.Fatalf("Failed to open pcap: %v", err)
+	}
+
+	reader, err := pcapgo.NewReader(pipe)
+	if err != nil {
+		log.Fatalf("Failed to read pcap header: %v", err)
+	}
+
+	for {
+		data, _, err := reader.ReadPacketData()
+		if err != nil {
+			log.Fatalf("Failed to read pcap packet: %v", err)
+		}
+
+		packet := gopacket.NewPacket(data, layers.LayerTypeIPv4, gopacket.Default)
+
+		if tcpLayer := packet.Layer(layers.LayerTypeTCP); tcpLayer != nil {
+			tcp, _ := tcpLayer.(*layers.TCP)
+			if tcp.SYN && !tcp.ACK { // Only plain SYN constitutes a NEW TCP connection
+				blockedConnections.With(prometheus.Labels{"protocol": "tcp", "dport": strconv.Itoa(int(tcp.DstPort))}).Inc()
+				continue
+			}
+		}
+
+		if udpLayer := packet.Layer(layers.LayerTypeUDP); udpLayer != nil {
+			udp, _ := udpLayer.(*layers.UDP)
+			blockedConnections.With(prometheus.Labels{"protocol": "udp", "dport": strconv.Itoa(int(udp.DstPort))}).Inc()
+			continue
+		}
+	}
+}
+
+func Start(addr string) error {
+	if err := prometheus.Register(blockedConnections); err != nil {
+		return err
+	}
+
+	http.Handle("/metrics", promhttp.Handler())
+
+	go func() {
+		log.Infof("Serving /metrics on %s", addr)
+		if err := http.ListenAndServe(addr, nil); err != nil {
+			log.Fatalf("Failed to bind metrics server: %v", err)
+		}
+	}()
+
+	go gatherMetrics()
+
+	return nil
+}

npc/namespace.go

@@ -0,0 +1,337 @@
+package npc
+
+import (
+	"encoding/json"
+
+	log "github.com/Sirupsen/logrus"
+	"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"
+	"k8s.io/client-go/pkg/types"
+	"k8s.io/client-go/pkg/util/uuid"
+
+	"github.com/weaveworks/weave/npc/ipset"
+)
+
+type ns struct {
+	ipt *iptables.IPTables // interface to iptables
+	ips ipset.Interface    // interface to ipset
+
+	name      string                               // k8s Namespace 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
+
+	uid     types.UID     // surrogate UID to own allPods selector
+	allPods *selectorSpec // hash:ip ipset of all pod IPs in this namespace
+
+	nsSelectors  *selectorSet
+	podSelectors *selectorSet
+	rules        *ruleSet
+}
+
+func newNS(name 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
+	}
+
+	ns := &ns{
+		ipt:         ipt,
+		ips:         ips,
+		name:        name,
+		pods:        make(map[types.UID]*coreapi.Pod),
+		policies:    make(map[types.UID]*extnapi.NetworkPolicy),
+		uid:         uuid.NewUUID(),
+		allPods:     allPods,
+		nsSelectors: nsSelectors,
+		rules:       newRuleSet(ipt)}
+
+	ns.podSelectors = newSelectorSet(ips, ns.onNewPodSelector)
+
+	if err := ns.podSelectors.provision(ns.uid, nil, map[string]*selectorSpec{ns.allPods.key: ns.allPods}); err != nil {
+		return nil, err
+	}
+
+	return ns, nil
+}
+
+func (ns *ns) empty() bool {
+	return len(ns.pods) == 0 && len(ns.policies) == 0 && ns.namespace == nil
+}
+
+func (ns *ns) destroy() error {
+	if err := ns.podSelectors.deprovision(ns.uid, map[string]*selectorSpec{ns.allPods.key: ns.allPods}, nil); err != nil {
+		return err
+	}
+	return nil
+}
+
+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); err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func (ns *ns) addPod(obj *coreapi.Pod) error {
+	ns.pods[obj.ObjectMeta.UID] = obj
+
+	if !hasIP(obj) {
+		return nil
+	}
+
+	return ns.podSelectors.addToMatching(obj.ObjectMeta.Labels, obj.Status.PodIP)
+}
+
+func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error {
+	delete(ns.pods, oldObj.ObjectMeta.UID)
+	ns.pods[newObj.ObjectMeta.UID] = newObj
+
+	if !hasIP(oldObj) && !hasIP(newObj) {
+		return nil
+	}
+
+	if hasIP(oldObj) && !hasIP(newObj) {
+		return ns.podSelectors.delFromMatching(oldObj.ObjectMeta.Labels, oldObj.Status.PodIP)
+	}
+
+	if !hasIP(oldObj) && hasIP(newObj) {
+		return ns.podSelectors.addToMatching(newObj.ObjectMeta.Labels, newObj.Status.PodIP)
+	}
+
+	if !equals(oldObj.ObjectMeta.Labels, newObj.ObjectMeta.Labels) ||
+		oldObj.Status.PodIP != newObj.Status.PodIP {
+
+		for _, ps := range ns.podSelectors.entries {
+			oldMatch := ps.matches(oldObj.ObjectMeta.Labels)
+			newMatch := ps.matches(newObj.ObjectMeta.Labels)
+			if oldMatch == newMatch && oldObj.Status.PodIP == newObj.Status.PodIP {
+				continue
+			}
+			if oldMatch {
+				if err := ps.delEntry(oldObj.Status.PodIP); err != nil {
+					return err
+				}
+			}
+			if newMatch {
+				if err := ps.addEntry(newObj.Status.PodIP); err != nil {
+					return err
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func (ns *ns) deletePod(obj *coreapi.Pod) error {
+	delete(ns.pods, obj.ObjectMeta.UID)
+
+	if !hasIP(obj) {
+		return nil
+	}
+
+	return ns.podSelectors.delFromMatching(obj.ObjectMeta.Labels, obj.Status.PodIP)
+}
+
+func (ns *ns) addNetworkPolicy(obj *extnapi.NetworkPolicy) error {
+	ns.policies[obj.ObjectMeta.UID] = obj
+
+	// Analyse policy, determine which rules and ipsets are required
+	rules, nsSelectors, podSelectors, err := ns.analysePolicy(obj)
+	if err != nil {
+		return err
+	}
+
+	// Provision required resources in dependency order
+	if err := ns.nsSelectors.provision(obj.ObjectMeta.UID, nil, nsSelectors); err != nil {
+		return err
+	}
+	if err := ns.podSelectors.provision(obj.ObjectMeta.UID, nil, podSelectors); err != nil {
+		return err
+	}
+	if err := ns.rules.provision(obj.ObjectMeta.UID, nil, rules); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (ns *ns) updateNetworkPolicy(oldObj, newObj *extnapi.NetworkPolicy) error {
+	delete(ns.policies, oldObj.ObjectMeta.UID)
+	ns.policies[newObj.ObjectMeta.UID] = newObj
+
+	// Analyse the old and the new policy so we can determine differences
+	oldRules, oldNsSelectors, oldPodSelectors, err := ns.analysePolicy(oldObj)
+	if err != nil {
+		return err
+	}
+	newRules, newNsSelectors, newPodSelectors, err := ns.analysePolicy(newObj)
+	if err != nil {
+		return err
+	}
+
+	// Deprovision unused and provision newly required resources in dependency order
+	if err := ns.rules.deprovision(oldObj.ObjectMeta.UID, oldRules, newRules); err != nil {
+		return err
+	}
+	if err := ns.nsSelectors.deprovision(oldObj.ObjectMeta.UID, oldNsSelectors, newNsSelectors); err != nil {
+		return err
+	}
+	if err := ns.podSelectors.deprovision(oldObj.ObjectMeta.UID, oldPodSelectors, newPodSelectors); err != nil {
+		return err
+	}
+	if err := ns.nsSelectors.provision(oldObj.ObjectMeta.UID, oldNsSelectors, newNsSelectors); err != nil {
+		return err
+	}
+	if err := ns.podSelectors.provision(oldObj.ObjectMeta.UID, oldPodSelectors, newNsSelectors); err != nil {
+		return err
+	}
+	if err := ns.rules.provision(oldObj.ObjectMeta.UID, oldRules, newRules); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (ns *ns) deleteNetworkPolicy(obj *extnapi.NetworkPolicy) error {
+	delete(ns.policies, obj.ObjectMeta.UID)
+
+	// Analyse network policy to free resources
+	rules, nsSelectors, podSelectors, err := ns.analysePolicy(obj)
+	if err != nil {
+		return err
+	}
+
+	// Deprovision unused resources in dependency order
+	if err := ns.rules.deprovision(obj.ObjectMeta.UID, rules, nil); err != nil {
+		return err
+	}
+	if err := ns.nsSelectors.deprovision(obj.ObjectMeta.UID, nsSelectors, nil); err != nil {
+		return err
+	}
+	if err := ns.podSelectors.deprovision(obj.ObjectMeta.UID, podSelectors, nil); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func bypassRule(nsIpsetName ipset.Name) []string {
+	return []string{"-m", "set", "--match-set", string(nsIpsetName), "dst", "-j", "ACCEPT"}
+}
+
+func (ns *ns) ensureBypassRule(nsIpsetName ipset.Name) error {
+	return ns.ipt.Append(TableFilter, DefaultChain, bypassRule(ns.allPods.ipsetName)...)
+}
+
+func (ns *ns) deleteBypassRule(nsIpsetName ipset.Name) error {
+	return ns.ipt.Delete(TableFilter, DefaultChain, bypassRule(ns.allPods.ipsetName)...)
+}
+
+func (ns *ns) addNamespace(obj *coreapi.Namespace) error {
+	ns.namespace = obj
+
+	// Insert a rule to bypass policies if namespace is DefaultAllow
+	if !isDefaultDeny(obj) {
+		return ns.ensureBypassRule(ns.allPods.ipsetName)
+	}
+
+	// Add namespace ipset to matching namespace selectors
+	return ns.nsSelectors.addToMatching(obj.ObjectMeta.Labels, string(ns.allPods.ipsetName))
+}
+
+func (ns *ns) updateNamespace(oldObj, newObj *coreapi.Namespace) error {
+	ns.namespace = newObj
+
+	// Update bypass rule if ingress default has changed
+	oldDefaultDeny := isDefaultDeny(oldObj)
+	newDefaultDeny := isDefaultDeny(newObj)
+
+	if oldDefaultDeny != newDefaultDeny {
+		if oldDefaultDeny {
+			return ns.ensureBypassRule(ns.allPods.ipsetName)
+		}
+		if newDefaultDeny {
+			return ns.deleteBypassRule(ns.allPods.ipsetName)
+		}
+	}
+
+	// Re-evaluate namespace selector membership if labels have changed
+	if !equals(oldObj.ObjectMeta.Labels, newObj.ObjectMeta.Labels) {
+		for _, selector := range ns.nsSelectors.entries {
+			oldMatch := selector.matches(oldObj.ObjectMeta.Labels)
+			newMatch := selector.matches(newObj.ObjectMeta.Labels)
+			if oldMatch == newMatch {
+				continue
+			}
+			if oldMatch {
+				if err := selector.delEntry(string(ns.allPods.ipsetName)); err != nil {
+					return err
+				}
+			}
+			if newMatch {
+				if err := selector.addEntry(string(ns.allPods.ipsetName)); err != nil {
+					return err
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func (ns *ns) deleteNamespace(obj *coreapi.Namespace) error {
+	ns.namespace = nil
+
+	// Remove bypass rule
+	if !isDefaultDeny(obj) {
+		return ns.deleteBypassRule(ns.allPods.ipsetName)
+	}
+
+	// Remove namespace ipset from any matching namespace selectors
+	return ns.nsSelectors.delFromMatching(obj.ObjectMeta.Labels, string(ns.allPods.ipsetName))
+}
+
+func hasIP(pod *coreapi.Pod) bool {
+	// Ensure pod has an IP address and isn't sharing the host network namespace
+	return len(pod.Status.PodIP) > 0 && !pod.Spec.HostNetwork
+}
+
+func equals(a, b map[string]string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for ak, av := range a {
+		if b[ak] != av {
+			return false
+		}
+	}
+	return true
+}
+
+func isDefaultDeny(namespace *coreapi.Namespace) bool {
+	nnpJson, found := namespace.ObjectMeta.Annotations["net.beta.kubernetes.io/network-policy"]
+	if !found {
+		return false
+	}
+
+	var nnp NamespaceNetworkPolicy
+	if err := json.Unmarshal([]byte(nnpJson), &nnp); err != nil {
+		log.Warn("Ignoring network policy annotation: unmarshal failed:", err)
+		// If we can't understand the annotation, behave as if it isn't present
+		return false
+	}
+
+	return nnp.Ingress != nil &&
+		nnp.Ingress.Isolation != nil &&
+		*(nnp.Ingress.Isolation) == DefaultDeny
+}

npc/rule.go

@@ -0,0 +1,77 @@
+package npc
+
+import (
+	"strings"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/coreos/go-iptables/iptables"
+	"k8s.io/client-go/pkg/types"
+)
+
+type ruleSpec struct {
+	key  string
+	args []string
+}
+
+func newRuleSpec(proto *string, srcHost *selectorSpec, dstHost *selectorSpec, dstPort *string) *ruleSpec {
+	args := []string{}
+	if proto != nil {
+		args = append(args, "-p", *proto)
+	}
+	if srcHost != nil {
+		args = append(args, "-m", "set", "--match-set", string(srcHost.ipsetName), "src")
+	}
+	if dstHost != nil {
+		args = append(args, "-m", "set", "--match-set", string(dstHost.ipsetName), "dst")
+	}
+	if dstPort != nil {
+		args = append(args, "--dport", *dstPort)
+	}
+	args = append(args, "-j", "ACCEPT")
+	key := strings.Join(args, " ")
+
+	return &ruleSpec{key, args}
+}
+
+type ruleSet struct {
+	ipt   *iptables.IPTables
+	users map[string]map[types.UID]struct{}
+}
+
+func newRuleSet(ipt *iptables.IPTables) *ruleSet {
+	return &ruleSet{ipt, make(map[string]map[types.UID]struct{})}
+}
+
+func (rs *ruleSet) deprovision(user types.UID, current, desired map[string]*ruleSpec) error {
+	for key, spec := range current {
+		if _, found := desired[key]; !found {
+			delete(rs.users[key], user)
+			if len(rs.users[key]) == 0 {
+				log.Infof("deleting rule: %v", spec.args)
+				if err := rs.ipt.Delete(TableFilter, IngressChain, spec.args...); err != nil {
+					return err
+				}
+				delete(rs.users, key)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (rs *ruleSet) provision(user types.UID, current, desired map[string]*ruleSpec) error {
+	for key, spec := range desired {
+		if _, found := current[key]; !found {
+			if _, found := rs.users[key]; !found {
+				log.Infof("adding rule: %v", spec.args)
+				if err := rs.ipt.Append(TableFilter, IngressChain, spec.args...); err != nil {
+					return err
+				}
+				rs.users[key] = make(map[types.UID]struct{})
+			}
+			rs.users[key][user] = struct{}{}
+		}
+	}
+
+	return nil
+}

npc/selector.go

@@ -0,0 +1,130 @@
+package npc
+
+import (
+	log "github.com/Sirupsen/logrus"
+	"k8s.io/client-go/pkg/api/unversioned"
+	"k8s.io/client-go/pkg/labels"
+	"k8s.io/client-go/pkg/types"
+
+	"github.com/weaveworks/weave/npc/ipset"
+)
+
+type selectorSpec struct {
+	key      string          // string representation (for hash keying/equality comparison)
+	selector labels.Selector // k8s Selector object (for matching)
+
+	ipsetType ipset.Type // type of ipset to provision
+	ipsetName ipset.Name // generated ipset name
+}
+
+func newSelectorSpec(json *unversioned.LabelSelector, nsName string, ipsetType ipset.Type) (*selectorSpec, error) {
+	selector, err := unversioned.LabelSelectorAsSelector(json)
+	if err != nil {
+		return nil, err
+	}
+	key := selector.String()
+	return &selectorSpec{
+		key:      key,
+		selector: selector,
+		// We prefix the selector string with the namespace name when generating
+		// the shortname because you can specify the same selector in multiple
+		// namespaces - we need those to map to distinct ipsets
+		ipsetName: ipset.Name("weave-" + shortName(nsName+":"+key)),
+		ipsetType: ipsetType}, nil
+}
+
+type selector struct {
+	ips  ipset.Interface
+	spec *selectorSpec
+}
+
+func (s *selector) matches(labelMap map[string]string) bool {
+	return s.spec.selector.Matches(labels.Set(labelMap))
+}
+
+func (s *selector) addEntry(entry string) error {
+	log.Infof("adding entry %s to %s", entry, s.spec.ipsetName)
+	return s.ips.AddEntry(s.spec.ipsetName, entry)
+}
+
+func (s *selector) delEntry(entry string) error {
+	log.Infof("deleting entry %s from %s", entry, s.spec.ipsetName)
+	return s.ips.DelEntry(s.spec.ipsetName, entry)
+}
+
+type selectorFn func(selector *selector) error
+
+type selectorSet struct {
+	ips           ipset.Interface
+	onNewSelector selectorFn
+	users         map[string]map[types.UID]struct{} // list of users per selector
+	entries       map[string]*selector
+}
+
+func newSelectorSet(ips ipset.Interface, onNewSelector selectorFn) *selectorSet {
+	return &selectorSet{
+		ips:           ips,
+		onNewSelector: onNewSelector,
+		users:         make(map[string]map[types.UID]struct{}),
+		entries:       make(map[string]*selector)}
+}
+
+func (ss *selectorSet) addToMatching(labelMap map[string]string, entry string) error {
+	for _, s := range ss.entries {
+		if s.matches(labelMap) {
+			if err := s.addEntry(entry); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (ss *selectorSet) delFromMatching(labelMap map[string]string, entry string) error {
+	for _, s := range ss.entries {
+		if s.matches(labelMap) {
+			if err := s.delEntry(entry); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (ss *selectorSet) deprovision(user types.UID, current, desired map[string]*selectorSpec) error {
+	for key, spec := range current {
+		if _, found := desired[key]; !found {
+			delete(ss.users[key], user)
+			if len(ss.users[key]) == 0 {
+				log.Infof("destroying ipset: %#v", spec)
+				if err := ss.ips.Destroy(spec.ipsetName); err != nil {
+					return err
+				}
+				delete(ss.entries, key)
+				delete(ss.users, key)
+			}
+		}
+	}
+	return nil
+}
+
+func (ss *selectorSet) provision(user types.UID, current, desired map[string]*selectorSpec) error {
+	for key, spec := range desired {
+		if _, found := current[key]; !found {
+			if _, found := ss.users[key]; !found {
+				log.Infof("creating ipset: %#v", spec)
+				if err := ss.ips.Create(spec.ipsetName, spec.ipsetType); err != nil {
+					return err
+				}
+				selector := &selector{ss.ips, spec}
+				if err := ss.onNewSelector(selector); err != nil {
+					return err
+				}
+				ss.users[key] = make(map[types.UID]struct{})
+				ss.entries[key] = selector
+			}
+			ss.users[key][user] = struct{}{}
+		}
+	}
+	return nil
+}

npc/shortname.go

@@ -0,0 +1,36 @@
+package npc
+
+import (
+	"crypto/sha1"
+	"math/big"
+)
+
+const (
+	// This array:
+	// * Must include only printable UTF8 characters that are represented with a single byte
+	// * Must be at least of length 85 (`len("weave-") + l(2^160)/l(85)` equals 31, the maximum ipset name length)
+	// * Must not include commas as those are treated specially by `ipset add` when adding a named set to a list:set
+	// * Should not include space for readability
+	// * Should not include single quote or backslash to be nice to shell users
+	ShortNameSymbols = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789(){}[]<>_$%^&*|/?.;:@#~"
+)
+
+// sha1 hash an arbitrary string and represent it using the full range of
+// printable ascii characters
+func shortName(arbitrary string) string {
+	symbols := []byte(ShortNameSymbols)
+	sum := sha1.Sum([]byte(arbitrary))
+	i := big.NewInt(0).SetBytes(sum[:])
+	base := big.NewInt(int64(len(symbols)))
+	zero := big.NewInt(0)
+	result := make([]byte, 0)
+
+	for i.Cmp(zero) > 0 {
+		remainder := new(big.Int).Mod(i, base)
+		i.Sub(i, remainder)
+		i.Div(i, base)
+		result = append(result, symbols[remainder.Int64()])
+	}
+
+	return string(result)
+}

npc/ulogd/ulogd.go

@@ -0,0 +1,29 @@
+package ulogd
+
+import (
+	log "github.com/Sirupsen/logrus"
+	"io"
+	"os"
+	"os/exec"
+)
+
+func waitForExit(cmd *exec.Cmd) {
+	if err := cmd.Wait(); err != nil {
+		log.Fatalf("ulogd terminated: %v", err)
+	}
+	log.Fatal("ulogd terminated normally")
+}
+
+func Start() error {
+	cmd := exec.Command("/usr/sbin/ulogd", "-v")
+	stdout, err := cmd.StderrPipe()
+	if err != nil {
+		return err
+	}
+	if err := cmd.Start(); err != nil {
+		return err
+	}
+	go io.Copy(os.Stdout, stdout)
+	go waitForExit(cmd)
+	return nil
+}

prog/kube-peers/main.go

@@ -4,9 +4,9 @@ import (
 	"fmt"
 	"log"
 
-	"k8s.io/client-go/1.4/kubernetes"
-	"k8s.io/client-go/1.4/pkg/api"
-	"k8s.io/client-go/1.4/rest"
+	"k8s.io/client-go/kubernetes"
+	api "k8s.io/client-go/pkg/api/v1"
+	"k8s.io/client-go/rest"
 )
 
 func getKubePeers() ([]string, error) {

prog/weave-npc/Dockerfile

@@ -0,0 +1,10 @@
+FROM alpine
+RUN apk add --update \
+    iptables \
+    ipset \
+	ulogd \
+  && rm -rf /var/cache/apk/* \
+  && mknod /var/log/ulogd.pcap p
+COPY ./weave-npc /usr/bin/weave-npc
+COPY ./ulogd.conf /etc/ulogd.conf
+ENTRYPOINT ["/usr/bin/weave-npc"]

prog/weave-npc/main.go

@@ -0,0 +1,180 @@
+package main
+
+import (
+	"os"
+	"os/signal"
+	"syscall"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/coreos/go-iptables/iptables"
+	"github.com/spf13/cobra"
+	"k8s.io/client-go/kubernetes"
+	coreapi "k8s.io/client-go/pkg/api/v1"
+	extnapi "k8s.io/client-go/pkg/apis/extensions/v1beta1"
+	"k8s.io/client-go/pkg/fields"
+	"k8s.io/client-go/pkg/runtime"
+	"k8s.io/client-go/pkg/util/wait"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/weaveworks/weave/npc"
+	"github.com/weaveworks/weave/npc/ipset"
+	"github.com/weaveworks/weave/npc/metrics"
+	"github.com/weaveworks/weave/npc/ulogd"
+)
+
+var (
+	version     = "(unreleased)"
+	metricsAddr string
+)
+
+func handleError(err error) {
+	if err != nil {
+		log.Fatal(err)
+	}
+}
+
+func makeController(getter cache.Getter, resource string,
+	objType runtime.Object, handlers cache.ResourceEventHandlerFuncs) *cache.Controller {
+	listWatch := cache.NewListWatchFromClient(getter, resource, "", fields.Everything())
+	_, controller := cache.NewInformer(listWatch, objType, 0, handlers)
+	return controller
+}
+
+func resetIPTables(ipt *iptables.IPTables) error {
+	// Flush chains first so there are no refs to extant ipsets
+	if err := ipt.ClearChain(npc.TableFilter, npc.IngressChain); err != nil {
+		return err
+	}
+
+	if err := ipt.ClearChain(npc.TableFilter, npc.DefaultChain); err != nil {
+		return err
+	}
+
+	if err := ipt.ClearChain(npc.TableFilter, npc.MainChain); err != nil {
+		return err
+	}
+
+	// Configure main chain static rules
+	if err := ipt.Append(npc.TableFilter, npc.MainChain,
+		"-m", "state", "--state", "RELATED,ESTABLISHED", "-j", "ACCEPT"); err != nil {
+		return err
+	}
+
+	if err := ipt.Append(npc.TableFilter, npc.MainChain,
+		"-m", "state", "--state", "NEW", "-j", string(npc.DefaultChain)); err != nil {
+		return err
+	}
+
+	if err := ipt.Append(npc.TableFilter, npc.MainChain,
+		"-m", "state", "--state", "NEW", "-j", string(npc.IngressChain)); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func resetIPSets(ips ipset.Interface) error {
+	// TODO should restrict ipset operations to the `weave-` prefix:
+
+	if err := ips.FlushAll(); err != nil {
+		return err
+	}
+
+	if err := ips.DestroyAll(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func root(cmd *cobra.Command, args []string) {
+	log.Infof("Starting Weaveworks NPC %s", version)
+
+	if err := metrics.Start(metricsAddr); err != nil {
+		log.Fatalf("Failed to start metrics: %v", err)
+	}
+
+	if err := ulogd.Start(); err != nil {
+		log.Fatalf("Failed to start ulogd: %v", err)
+	}
+
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	client, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	ipt, err := iptables.New()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	ips := ipset.New()
+
+	handleError(resetIPTables(ipt))
+	handleError(resetIPSets(ips))
+
+	npc := npc.New(ipt, ips)
+
+	nsController := makeController(client.Core().RESTClient(), "namespaces", &coreapi.Namespace{},
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				handleError(npc.AddNamespace(obj.(*coreapi.Namespace)))
+			},
+			DeleteFunc: func(obj interface{}) {
+				handleError(npc.DeleteNamespace(obj.(*coreapi.Namespace)))
+			},
+			UpdateFunc: func(old, new interface{}) {
+				handleError(npc.UpdateNamespace(old.(*coreapi.Namespace), new.(*coreapi.Namespace)))
+			}})
+
+	podController := makeController(client.Core().RESTClient(), "pods", &coreapi.Pod{},
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				handleError(npc.AddPod(obj.(*coreapi.Pod)))
+			},
+			DeleteFunc: func(obj interface{}) {
+				handleError(npc.DeletePod(obj.(*coreapi.Pod)))
+			},
+			UpdateFunc: func(old, new interface{}) {
+				handleError(npc.UpdatePod(old.(*coreapi.Pod), new.(*coreapi.Pod)))
+			}})
+
+	npController := makeController(client.Extensions().RESTClient(), "networkpolicies", &extnapi.NetworkPolicy{},
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				handleError(npc.AddNetworkPolicy(obj.(*extnapi.NetworkPolicy)))
+			},
+			DeleteFunc: func(obj interface{}) {
+				handleError(npc.DeleteNetworkPolicy(obj.(*extnapi.NetworkPolicy)))
+			},
+			UpdateFunc: func(old, new interface{}) {
+				handleError(npc.UpdateNetworkPolicy(old.(*extnapi.NetworkPolicy), new.(*extnapi.NetworkPolicy)))
+			}})
+
+	go nsController.Run(wait.NeverStop)
+	go podController.Run(wait.NeverStop)
+	go npController.Run(wait.NeverStop)
+
+	signals := make(chan os.Signal, 1)
+	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
+	log.Fatalf("Exiting: %v", <-signals)
+}
+
+func main() {
+	rootCmd := &cobra.Command{
+		Use:   "weave-npc",
+		Short: "Weaveworks Kubernetes Network Policy Controller",
+		Run:   root}
+
+	rootCmd.PersistentFlags().StringVar(&metricsAddr, "metrics-addr", ":8686", "metrics server bind address")
+
+	if err := rootCmd.Execute(); err != nil {
+		log.Fatal(err)
+	}
+}

prog/weave-npc/ulogd.conf

@@ -0,0 +1,13 @@
+[global]
+logfile="/dev/null"
+plugin="/usr/lib/ulogd/ulogd_inppkt_NFLOG.so"
+plugin="/usr/lib/ulogd/ulogd_raw2packet_BASE.so"
+plugin="/usr/lib/ulogd/ulogd_output_PCAP.so"
+stack=log1:NFLOG,base1:BASE,pcap1:PCAP
+
+[log1]
+group=86
+
+[pcap1]
+file="/var/log/ulogd.pcap"
+sync=1