Merge pull request #23 from weaveworks/issues/7-blocked-connection-me…

…trics

Expose blocked connection metrics

Makefile

@@ -20,7 +20,7 @@ cmd/weave-npc/weave-npc: $(DEPS)
 cmd/weave-npc/weave-npc: cmd/weave-npc/*.go
 	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags "-X main.version=$(VERSION)" -o $@ cmd/weave-npc/main.go
 
-build/.image.done: cmd/weave-npc/Dockerfile cmd/weave-npc/weave-npc
+build/.image.done: cmd/weave-npc/Dockerfile cmd/weave-npc/weave-npc cmd/weave-npc/ulogd.conf
 	mkdir -p build
 	cp $^ build
 	sudo docker build -t $(DH_ORG)/weave-npc:$(IMAGE_TAG) -f build/Dockerfile ./build

cmd/weave-npc/Dockerfile

@@ -2,6 +2,9 @@ FROM alpine
 RUN apk add --update \
     iptables \
     ipset \
-  && rm -rf /var/cache/apk/*
+	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"]

cmd/weave-npc/main.go

@@ -6,6 +6,7 @@ import (
 	"syscall"
 
 	log "github.com/Sirupsen/logrus"
+	"github.com/spf13/cobra"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/apis/extensions"
 	"k8s.io/kubernetes/pkg/client/cache"
@@ -19,10 +20,15 @@ import (
 	"k8s.io/kubernetes/pkg/util/wait"
 
 	weavenpc "github.com/weaveworks/weave-npc/pkg/controller"
+	"github.com/weaveworks/weave-npc/pkg/metrics"
+	"github.com/weaveworks/weave-npc/pkg/ulogd"
 	"github.com/weaveworks/weave-npc/pkg/util/ipset"
 )
 
-var version = "(unreleased)"
+var (
+	version     = "(unreleased)"
+	metricsAddr string
+)
 
 func handleError(err error) {
 	if err != nil {
@@ -98,9 +104,17 @@ func resetIPSets(ips ipset.Interface) error {
 	return nil
 }
 
-func main() {
+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)
+	}
+
 	client, err := unversioned.NewInCluster()
 	if err != nil {
 		log.Fatal(err)
@@ -158,3 +172,16 @@ func main() {
 	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)
+	}
+}

cmd/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

pkg/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
+}

pkg/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
+}