Merge pull request #93 from iovisor/dist_bridge

Dist bridge

examples/distributed_bridge/main.py

+#!/usr/bin/env python
+# Copyright (c) PLUMgrid, Inc.
+# Licensed under the Apache License, Version 2.0 (the "License")
+
+from pyroute2 import IPRoute, NetNS, IPDB, NSPopen
+from simulation import Simulation
+from subprocess import PIPE
+
+ipr = IPRoute()
+ipdb = IPDB(nl=ipr)
+
+num_hosts = 3
+null = open("/dev/null", "w")
+
+class TunnelSimulation(Simulation):
+    def __init__(self, ipdb):
+        super(TunnelSimulation, self).__init__(ipdb)
+
+    def start(self):
+        # each entry is tuple of ns_ipdb, out_ifc, in_ifc
+        host_info = []
+        for i in range(0, num_hosts):
+            print("Launching host %i of %i" % (i + 1, num_hosts))
+            ipaddr = "172.16.1.%d/24" % (100 + i)
+            host_info.append(self._create_ns("host%d" % i, ipaddr=ipaddr,
+                disable_ipv6=True))
+            cmd = ["python", "tunnel.py"]
+            p = NSPopen(host_info[i][0].nl.netns, cmd, stdin=PIPE)
+            self.processes.append(p)
+        with self.ipdb.create(ifname="br-fabric", kind="bridge") as br:
+            for host in host_info: br.add_port(host[1])
+            br.up()
+
+try:
+    sim = TunnelSimulation(ipdb)
+    sim.start()
+    input("Press enter to quit:")
+finally:
+    if "br-fabric" in ipdb.interfaces: ipdb.interfaces["br-fabric"].remove().commit()
+    for p in sim.processes: p.communicate(b"\n")
+    sim.release()
+    ipdb.release()
+    null.close()

examples/distributed_bridge/simulation.py

+../simulation.py
\ No newline at end of file

examples/distributed_bridge/tunnel.c

+// Copyright (c) PLUMgrid, Inc.
+// Licensed under the Apache License, Version 2.0 (the "License")
+#include <bcc/proto.h>
+
+BPF_TABLE("hash", u32, int, vni2if, 1024);
+
+struct vni_key {
+  u64 mac;
+  int ifindex;
+  int pad;
+};
+struct host {
+  u32 tunnel_id;
+  u32 remote_ipv4;
+  u64 rx_pkts;
+  u64 tx_pkts;
+};
+BPF_TABLE("hash", struct vni_key, struct host, mac2host, 10240);
+
+struct config {
+  int tunnel_ifindex;
+};
+BPF_TABLE("hash", int, struct config, conf, 1);
+
+// Handle packets from the encap device, demux into the dest tenant
+int handle_ingress(struct __sk_buff *skb) {
+  u8 *cursor = 0;
+
+  struct ethernet_t *ethernet = cursor_advance(cursor, sizeof(*ethernet));
+
+  struct bpf_tunnel_key tkey = {};
+  bpf_skb_get_tunnel_key(skb, &tkey, sizeof(tkey));
+
+  int *ifindex = vni2if.lookup(&tkey.tunnel_id);
+  if (ifindex) {
+    //bpf_trace_printk("ingress tunnel_id=%d ifindex=%d\n", tkey.tunnel_id, *ifindex);
+    struct vni_key vk = {ethernet->src, *ifindex, 0};
+    struct host *src_host = mac2host.lookup_or_init(&vk,
+        &(struct host){tkey.tunnel_id, tkey.remote_ipv4, 0, 0});
+    lock_xadd(&src_host->rx_pkts, 1);
+    bpf_clone_redirect(skb, *ifindex, 1/*ingress*/);
+  } else {
+    bpf_trace_printk("ingress invalid tunnel_id=%d\n", tkey.tunnel_id);
+  }
+
+  return 1;
+}
+
+// Handle packets from the tenant, mux into the encap device
+int handle_egress(struct __sk_buff *skb) {
+  u8 *cursor = 0;
+
+  int one = 1;
+  struct config *cfg = conf.lookup(&one);
+  if (!cfg) return 1;
+
+  struct ethernet_t *ethernet = cursor_advance(cursor, sizeof(*ethernet));
+
+  struct vni_key vk = {ethernet->dst, skb->ifindex, 0};
+  struct host *dst_host = mac2host.lookup(&vk);
+  struct bpf_tunnel_key tkey = {};
+  if (dst_host) {
+    u32 zero = 0;
+    tkey.tunnel_id = dst_host->tunnel_id;
+    tkey.remote_ipv4 = dst_host->remote_ipv4;
+    bpf_skb_set_tunnel_key(skb, &tkey, sizeof(tkey));
+    lock_xadd(&dst_host->tx_pkts, 1);
+  } else {
+    struct bpf_tunnel_key tkey = {};
+    vk.mac = 0xFFFFFFFFFFFFull;
+    dst_host = mac2host.lookup(&vk);
+    if (!dst_host)
+      return 1;
+    tkey.tunnel_id = dst_host->tunnel_id;
+    tkey.remote_ipv4 = dst_host->remote_ipv4;
+    bpf_skb_set_tunnel_key(skb, &tkey, sizeof(tkey));
+  }
+  bpf_clone_redirect(skb, cfg->tunnel_ifindex, 0/*egress*/);
+  return 1;
+}

examples/distributed_bridge/tunnel.py

+#!/usr/bin/env python
+# Copyright (c) PLUMgrid, Inc.
+# Licensed under the Apache License, Version 2.0 (the "License")
+
+from bpf import BPF
+from builtins import input
+from ctypes import c_int, c_uint
+from http.server import HTTPServer, SimpleHTTPRequestHandler
+import json
+from netaddr import EUI, IPAddress
+from pyroute2 import IPRoute, NetNS, IPDB, NSPopen
+from socket import htons, AF_INET
+from threading import Thread
+
+b = BPF(src_file="tunnel.c")
+ingress_fn = b.load_func("handle_ingress", BPF.SCHED_CLS)
+egress_fn = b.load_func("handle_egress", BPF.SCHED_CLS)
+mac2host = b.get_table("mac2host")
+vni2if = b.get_table("vni2if")
+conf = b.get_table("conf")
+
+ipr = IPRoute()
+ipdb = IPDB(nl=ipr)
+
+ifc = ipdb.interfaces.eth0
+mcast = IPAddress("239.1.1.1")
+
+# ifcs to cleanup at the end
+ifc_gc = []
+
+def run():
+    ipdb.routes.add({"dst": "224.0.0.0/4", "oif": ifc.index}).commit()
+    with ipdb.create(ifname="vxlan0", kind="vxlan", vxlan_id=0,
+                     vxlan_link=ifc, vxlan_port=htons(4789),
+                     vxlan_group=str(mcast), vxlan_flowbased=True,
+                     vxlan_collect_metadata=True,
+                     vxlan_learning=False) as vx:
+        vx.up()
+        ifc_gc.append(vx.ifname)
+
+    conf[c_int(1)] = c_int(vx.index)
+
+    ipr.tc("add", "ingress", vx.index, "ffff:")
+    ipr.tc("add-filter", "bpf", vx.index, ":1", fd=ingress_fn.fd,
+           name=ingress_fn.name, parent="ffff:", action="drop", classid=1)
+
+    for i in range(0, 2):
+        vni = 10000 + i
+        with ipdb.create(ifname="br%d" % i, kind="bridge") as br:
+            v = ipdb.create(ifname="dummy%d" % i, kind="dummy").up().commit()
+            mcast_key = mac2host.Key(0xFFFFFFFFFFFF, v.index, 0)
+            mcast_leaf = mac2host.Leaf(vni, mcast.value, 0, 0)
+            mac2host[mcast_key] = mcast_leaf
+
+            ipr.tc("add", "sfq", v.index, "1:")
+            ipr.tc("add-filter", "bpf", v.index, ":1", fd=egress_fn.fd,
+                   name=egress_fn.name, parent="1:", action="drop", classid=1)
+            br.add_port(v)
+            br.up()
+            ifc_gc.append(v.ifname)
+            ifc_gc.append(br.ifname)
+            vni2if[c_uint(vni)] = c_int(v.index)
+
+try:
+    run()
+    input("")
+    print("---")
+    for k, v in mac2host.items():
+        print(EUI(k.mac), k.ifindex, IPAddress(v.remote_ipv4),
+              v.tunnel_id, v.rx_pkts, v.tx_pkts)
+finally:
+    for v in ifc_gc: ipdb.interfaces[v].remove().commit()
+    ipdb.release()

src/cc/compat/linux/bpf.h

@@ -258,6 +258,17 @@ enum bpf_func_id {
 	BPF_FUNC_get_cgroup_classid,
 	BPF_FUNC_skb_vlan_push, /* bpf_skb_vlan_push(skb, vlan_proto, vlan_tci) */
 	BPF_FUNC_skb_vlan_pop,  /* bpf_skb_vlan_pop(skb) */
+
+	/**
+	 * bpf_skb_[gs]et_tunnel_key(skb, struct bpf_tunnel_key *key, int size)
+	 * retrieve or populate tunnel metadata
+	 * @skb: pointer to skb
+	 * @key: pointer to 'struct bpf_tunnel_key'
+	 * @size: size of 'struct bpf_tunnel_key'
+	 * Retrun: 0 on success
+	 */
+	BPF_FUNC_skb_get_tunnel_key,
+	BPF_FUNC_skb_set_tunnel_key,
 	__BPF_FUNC_MAX_ID,
 };
 
@@ -280,4 +291,9 @@ struct __sk_buff {
 	__u32 cb[5];
 };
 
+struct bpf_tunnel_key {
+	__u32 tunnel_id;
+	__u32 remote_ipv4;
+};
+
 #endif /* _UAPI__LINUX_BPF_H__ */

src/cc/export/helpers.h

@@ -64,7 +64,7 @@ static int (*bpf_trace_printk_)(const char *fmt, u64 fmt_size, ...) =
 	(void *) BPF_FUNC_trace_printk;
 int bpf_trace_printk(const char *fmt, ...) asm("llvm.bpf.extra");
 #if LINUX_VERSION_CODE >= KERNEL_VERSION(4,2,0)
-static u64 (*bpf_clone_redirect)(void *ctx, u64 ifindex, u64 flags) =
+static int (*bpf_clone_redirect)(void *ctx, u64 ifindex, u64 flags) =
 	(void *) BPF_FUNC_clone_redirect;
 static u64 (*bpf_get_smp_processor_id)(void) =
 	(void *) BPF_FUNC_get_smp_processor_id;
@@ -83,6 +83,10 @@ static u64 (*bpf_skb_vlan_pop)(void *ctx) =
 static void bpf_tail_call_(u64 map_fd, void *ctx, int index) {
   ((void (*)(void *, u64, int))BPF_FUNC_tail_call)(ctx, map_fd, index);
 }
+static int (*bpf_skb_get_tunnel_key)(void *ctx, void *to, u32 size) =
+  (void *) BPF_FUNC_skb_get_tunnel_key;
+static int (*bpf_skb_set_tunnel_key)(void *ctx, void *from, u32 size) =
+  (void *) BPF_FUNC_skb_set_tunnel_key;
 #endif
 
 /* llvm builtin functions that eBPF C program may use to