Install bpf program for every pod we discover

This introduces a bpf controller that runs the lookup loop for
every cgroup we attach a program to, reads the map and executes
handlers that write the tracked data to Prometheus metrics.

The top subcommand uses the same code but implements different handlers.

Closes kinvolk-archives#4, closes kinvolk-archives#5, closes kinvolk-archives#6, closes kinvolk-archives#9

Dockerfile

@@ -5,5 +5,5 @@ RUN apk update && apk add --no-cache libc6-compat
 COPY cgnet /cgnet
 
 EXPOSE 9101
-ENTRYPOINT ["/cgnet", "serve"]
+ENTRYPOINT ["/cgnet", "export"]
 CMD ["--port", "9101"]

Gopkg.toml

@@ -1,21 +1,3 @@
-
-# Gopkg.toml example
-#
-# Refer to https://github.com/golang/dep/blob/master/docs/Gopkg.toml.md
-# for detailed Gopkg.toml documentation.
-#
-# required = ["github.com/user/thing/cmd/thing"]
-# ignored = ["github.com/user/project/pkgX", "bitbucket.org/user/project/pkgA/pkgY"]
-#
-# [[constraint]]
-#   name = "github.com/user/project"
-#   version = "1.0.0"
-#
-# [[constraint]]
-#   name = "github.com/user/project2"
-#   branch = "dev"
-#   source = "github.com/myfork/project2"
-#
-# [[override]]
-#  name = "github.com/x/y"
-#  version = "2.4.0"
+[[constraint]]
+  name = "github.com/inconshreveable/log15"
+  branch = "master"

Makefile

@@ -22,16 +22,15 @@ container: $(BIN)
 	docker build -t $(CONTAINER):latest .
 	docker push $(CONTAINER):latest
 
-manifest: $(MANIFEST)
-$(MANIFEST):
+manifest:
+	@make -C $(MANIFEST_DIR) clean
 	@make -C $(MANIFEST_DIR)
 
 clean:
 	rm -rf $(BIN)
 	@make -C bpf/ clean
 
 deploy-clean: clean
-	@make -C $(MANIFEST_DIR) clean
 	docker rmi $(CONTAINER):latest
 
 deps: build-deps

bpf/Makefile

@@ -1,7 +1,6 @@
 OUTDIR=out
 OUTFILE=$(OUTDIR)/cgnet.o
 
-
 .PHONY: all build clean
 
 all: $(OUTFILE) bindata.go
@@ -18,6 +17,5 @@ bindata.go:
 	go-bindata -pkg bpf out/
 
 clean:
-	rm -rf cgnet.o
 	rm -rf bindata.go
 	rm -rf $(OUTFILE)

bpf/bpf.go

@@ -1,103 +1,108 @@
+/*
+Copyright 2017 Kinvolk GmbH
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package bpf
 
 import (
 	"bytes"
 	"fmt"
-	"log"
 	"os"
-	"time"
 	"unsafe"
 
-	"github.com/iovisor/gobpf/elf"
+	bpf "github.com/iovisor/gobpf/elf"
 )
 
 /*
 #include <linux/bpf.h>
 */
 import "C"
 
-const assetpath string = "out/cgnet.o"
-const mapname string = "count"
+const (
+	assetPath string = "out/cgnet.o"
+	mapName   string = "count_map"
+)
+
+func Attach(cgroupPath string) (*Controller, error) {
+	b, err := initModule()
+	if err != nil {
+		return nil, err
+	}
 
-var (
-	zero       uint32 = 0
-	packetsKey uint32 = zero
-	bytesKey   uint32 = 1
+	for prog := range b.IterCgroupProgram() {
+		if err := bpf.AttachCgroupProgram(prog, cgroupPath, bpf.EgressType); err != nil {
+			return nil, fmt.Errorf("error attaching to cgroup %s: %s", cgroupPath, err)
+		}
+	}
 
-	b *elf.Module
-)
+	if err := initMap(b); err != nil {
+		return nil, err
+	}
 
-func Setup(cgroupPath string) error {
-	log.SetFlags(log.LstdFlags | log.Lshortfile)
+	ctl := &Controller{
+		cgroup: cgroupPath,
+		module: b,
+		quit:   make(chan struct{}),
+	}
+	return ctl, nil
+}
 
-	// for quick development on the bpf program
+func initModule() (*bpf.Module, error) {
+	var b *bpf.Module
 	if path := os.Getenv("BPF_PROG_PATH"); path != "" {
-		fmt.Printf("using: %s\n", path)
-		b = elf.NewModule(path)
+		// for development
+		b = bpf.NewModule(path)
 	} else {
-		reader := bytes.NewReader(MustAsset(assetpath))
-		b = elf.NewModuleFromReader(reader)
+		// from assets
+		reader := bytes.NewReader(MustAsset(assetPath))
+		b = bpf.NewModuleFromReader(reader)
 	}
 	if b == nil {
-		return fmt.Errorf("System doesn't support BPF")
+		return nil, fmt.Errorf("system doesn't seem to support BPF")
 	}
 
 	if err := b.Load(nil); err != nil {
-		return fmt.Errorf("%s", err)
-	}
-
-	for prog := range b.IterCgroupProgram() {
-		if err := elf.AttachCgroupProgram(prog, cgroupPath, elf.EgressType); err != nil {
-			return fmt.Errorf("%s", err)
-		}
-	}
-
-	mp := b.Map(mapname)
-	if mp == nil {
-		return fmt.Errorf("Can't find map '%s'", mapname)
+		return nil, fmt.Errorf("loading module failed: %s", err)
 	}
+	return b, nil
+}
 
-	if err := b.UpdateElement(mp, unsafe.Pointer(&packetsKey), unsafe.Pointer(&zero), C.BPF_ANY); err != nil {
+func initMap(b *bpf.Module) error {
+	if err := update(b, packetsKey, 0); err != nil {
 		return fmt.Errorf("error updating map: %s", err)
 	}
-
-	if err := b.UpdateElement(mp, unsafe.Pointer(&bytesKey), unsafe.Pointer(&zero), C.BPF_EXIST); err != nil {
+	if err := update(b, bytesKey, 0); err != nil {
 		return fmt.Errorf("error updating map: %s", err)
 	}
-
-	fmt.Println("Ready.")
 	return nil
 }
 
-func UpdateLoop(quit chan struct{}) error {
-	mp := b.Map(mapname)
-	if mp == nil {
-		return fmt.Errorf("Can't find map '%s'", mapname)
-	}
+func update(b *bpf.Module, key uint32, value uint64) error {
 
-	var packets, bytes uint64
-
-	for {
-		select {
-		case <-quit:
-			return nil
-		case <-time.After(1000 * time.Millisecond):
-			if err := updateElements(mp, packets, bytes); err != nil {
-				return err
-			}
-			fmt.Printf("cgroup received %d packets (%d bytes)\n", packets, bytes)
-		}
+	mp := b.Map(mapName)
+	if err := b.UpdateElement(mp, unsafe.Pointer(&key), unsafe.Pointer(&value), C.BPF_ANY); err != nil {
+		return err
 	}
+	return nil
 }
 
-func updateElements(mp *elf.Map, packets, bytes uint64) error {
-	if err := b.LookupElement(mp, unsafe.Pointer(&packetsKey), unsafe.Pointer(&packets)); err != nil {
-		return fmt.Errorf("error looking up in map: %s", err)
+func lookup(b *bpf.Module, key uint32) (uint64, error) {
+	mp := b.Map(mapName)
+	var value uint64
+	if err := b.LookupElement(mp, unsafe.Pointer(&key), unsafe.Pointer(&value)); err != nil {
+		return 0, err
 	}
-
-	if err := b.LookupElement(mp, unsafe.Pointer(&bytesKey), unsafe.Pointer(&bytes)); err != nil {
-		return fmt.Errorf("error looking up in map: %s", err)
-	}
-
-	return nil
+	return value, nil
 }

bpf/controller.go

@@ -0,0 +1,89 @@
+/*
+Copyright 2017 Kinvolk GmbH
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package bpf
+
+import (
+	"context"
+	"log"
+	"time"
+
+	bpf "github.com/iovisor/gobpf/elf"
+)
+
+/*
+#include <linux/bpf.h>
+*/
+import "C"
+
+const (
+	packetsKey uint32 = 0
+	bytesKey   uint32 = 1
+)
+
+type Controller struct {
+	cgroup string
+	module *bpf.Module
+	quit   chan struct{}
+
+	packetsHandler func(uint64) error
+	bytesHandler   func(uint64) error
+}
+
+func (c *Controller) Stop() {
+	c.quit <- struct{}{}
+	if err := c.module.Close(); err != nil {
+		log.Printf("error closing bpf module: %s", err)
+	}
+}
+
+func (c *Controller) SetPacketsHandler(h func(uint64) error) {
+	c.packetsHandler = h
+}
+func (c *Controller) SetBytesHandler(h func(uint64) error) {
+	c.bytesHandler = h
+}
+
+func (c *Controller) Run(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-c.quit:
+			return
+		case <-time.After(1 * time.Second):
+			// TODO this needs to be solved differently
+			// Maybe sending new values on individual channels?
+			packets, err := lookup(c.module, packetsKey)
+			if err != nil {
+				log.Printf("lookup failed: %s", err)
+				continue
+			}
+			if err := c.packetsHandler(packets); err != nil {
+				log.Printf("packetsHandler failed: %s", err)
+			}
+
+			bytes, err := lookup(c.module, bytesKey)
+			if err != nil {
+				log.Printf("lookup failed: %s", err)
+				continue
+			}
+			if err := c.bytesHandler(bytes); err != nil {
+				log.Printf("bytesHandler failed: %s", err)
+			}
+		}
+	}
+}

bpf/src/cgnet.c

@@ -20,7 +20,10 @@
 #include <stddef.h>
 #include "bpf_helpers.h"
 
-struct bpf_map_def SEC("maps/count") count_map = {
+#define PACKETS_KEY 0
+#define BYTES_KEY 1
+
+struct bpf_map_def SEC("maps/count_map") count_map = {
 	.type = BPF_MAP_TYPE_ARRAY,
 	.key_size = sizeof(int),
 	.value_size = sizeof(__u64),
@@ -29,7 +32,7 @@ struct bpf_map_def SEC("maps/count") count_map = {
 
 SEC("cgroup/skb")
 int count_packets(struct __sk_buff *skb) {
-    int packets_key = 0, bytes_key = 1;
+    int packets_key = PACKETS_KEY, bytes_key = BYTES_KEY;
     __u64 *packets = 0;
     __u64 *bytes = 0;