diff --git a/router/fastdp.go b/router/fastdp.go new file mode 100644 index 00000000..c58cd43a --- /dev/null +++ b/router/fastdp.go @@ -0,0 +1,593 @@ +package router + +import ( + "encoding/binary" + "fmt" + "log" + "net" + "sync" + + "github.com/weaveworks/go-odp/odp" +) + +type thunk func() FlowOp + +// A bridgePortID is either an ODP vport or the router. We express it +// like this to use it as a map key. +type bridgePortID struct { + vport odp.VportID + router bool +} + +var routerPortID = bridgePortID{router: true} + +type FastDatapath struct { + vxlanPort int + + // The lock guards rthe FastDatapath state, and also + // synchronises use of the dpif + lock sync.Mutex + dpif *odp.Dpif + dp odp.DatapathHandle + consumingMisses bool + missHandlers map[odp.VportID]func(odp.FlowKeys) thunk + vxlanVportID odp.VportID + localPeer *Peer + peers *Peers + interHostConsumer InterHostConsumer + + // Bridge state: A bridge port is represented as a function + // that says how to send out a packet on that port. + bridgePorts map[bridgePortID]func(PacketKey) thunk + bridgeMACs map[MAC]func(PacketKey) thunk + + // Only accessed from Miss, so not locked + dec *EthernetDecoder +} + +func NewFastDatapath(dpname string, vxlanPort int) (*FastDatapath, error) { + dpif, err := odp.NewDpif() + if err != nil { + return nil, err + } + + success := false + defer func() { + if !success { + dpif.Close() + } + }() + + dp, err := dpif.LookupDatapath(dpname) + if err != nil { + return nil, err + } + + fastdp := &FastDatapath{ + vxlanPort: vxlanPort, + dpif: dpif, + dp: dp, + missHandlers: make(map[odp.VportID]func(odp.FlowKeys) thunk), + bridgePorts: make(map[bridgePortID]func(PacketKey) thunk), + bridgeMACs: make(map[MAC]func(PacketKey) thunk), + dec: NewEthernetDecoder(), + } + + if err := fastdp.deleteVxlanVports(); err != nil { + return nil, err + } + + if err := fastdp.clearFlows(); err != nil { + return nil, err + } + + fastdp.vxlanVportID, err = fastdp.dp.CreateVport( + odp.NewVxlanVportSpec("vxlan", uint16(vxlanPort))) + if err != nil { + return nil, err + } + + success = true + return fastdp, nil +} + +func (fastdp *FastDatapath) consumeMisses() error { + if !fastdp.consumingMisses { + if err := fastdp.dp.ConsumeMisses(fastdp); err != nil { + return err + } + + fastdp.consumingMisses = true + } + + return nil +} + +func (fastdp *FastDatapath) Close() error { + err := fastdp.dpif.Close() + fastdp.dpif = nil + return err +} + +func (fastdp *FastDatapath) clearFlows() error { + flows, err := fastdp.dp.EnumerateFlows() + if err != nil { + return err + } + + for _, flow := range flows { + err = fastdp.dp.DeleteFlow(flow.FlowSpec) + if err != nil && !odp.IsNoSuchFlowError(err) { + return err + } + } + + return nil +} + +func (fastdp *FastDatapath) deleteVxlanVports() error { + vports, err := fastdp.dp.EnumerateVports() + if err != nil { + return err + } + + for _, vport := range vports { + if vport.Spec.TypeName() != "vxlan" { + continue + } + + err = fastdp.dp.DeleteVport(vport.ID) + if err != nil && !odp.IsNoSuchVportError(err) { + return err + } + } + + return nil +} + +func (fastdp *FastDatapath) Error(err error, stopped bool) { + // XXX fatal if stopped + log.Println("Error while listening on datapath:", err) +} + +func (fastdp *FastDatapath) Miss(packet []byte, flowKeys odp.FlowKeys) error { + thunk := fastdp.handleMiss(flowKeys) + if thunk != nil { + fastdp.send(thunk(), packet) + } + + return nil +} + +func (fastdp *FastDatapath) handleMiss(fks odp.FlowKeys) thunk { + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + + ingress := fks[odp.OVS_KEY_ATTR_IN_PORT].(odp.InPortFlowKey).VportID() + + log.Println("Got miss", fks, "on port", ingress) + + handler := fastdp.missHandlers[ingress] + if handler == nil { + handler = fastdp.makeMissHandler(ingress) + if handler == nil { + return simpleThunk(vetoFlowCreation) + } + + fastdp.missHandlers[ingress] = handler + } + + return handler(fks) +} + +func (fastdp *FastDatapath) makeMissHandler( + ingress odp.VportID) func(odp.FlowKeys) thunk { + // Auto-add netdev and internal vports to the bridge (i.e. not + // vxlan vports) + vport, err := fastdp.dp.LookupVport(ingress) + if err != nil { + log.Println(err) + return nil + } + + typ := vport.Spec.TypeName() + if typ != "netdev" && typ != "internal" { + return nil + } + + // Construct a bridge port for the netdev, which simply + // outputs packets + bridgePort := func(key PacketKey) thunk { + return simpleThunk(fastdp.odpActions( + odp.NewOutputAction(ingress))) + } + fastdp.bridgePorts[bridgePortID{vport: ingress}] = bridgePort + + // Clear flows, in order to recalculate flows for broadcasts + // on the bridge + checkWarn(fastdp.clearFlows()) + + // Packets from the netdev are processed by the bridge + return func(fks odp.FlowKeys) thunk { + return multiThunk(fastdp.bridge(bridgePortID{vport: ingress}, + flowKeysToPacketKey(fks))) + } +} + +func flowKeysToPacketKey(fks odp.FlowKeys) PacketKey { + eth := fks[odp.OVS_KEY_ATTR_ETHERNET].(odp.EthernetFlowKey).Key() + return PacketKey{SrcMAC: eth.EthSrc, DstMAC: eth.EthDst} +} + +// Send a packet, creating a corresponding ODP flow rule if possible +func (fastdp *FastDatapath) send(fops FlowOp, frame []byte) { + // Gather the actions from actionFlowOps, execute any others + var dec *EthernetDecoder + flow := odp.NewFlowSpec() + createFlow := true + + for _, xfop := range FlattenFlowOp(fops) { + switch fop := xfop.(type) { + case interface { + updateFlowSpec(*odp.FlowSpec) + }: + fop.updateFlowSpec(&flow) + case vetoFlowCreationFlowOp: + createFlow = false + default: + // A foreign flow op, so send the packet the + // normal way, decoding the packet lazily. + if dec == nil { + dec = fastdp.dec + dec.DecodeLayers(frame) + createFlow = false + } + + if len(dec.decoded) != 0 { + fop.Send(frame, dec, false) + } + } + } + + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + + if len(flow.Actions) != 0 { + checkWarn(fastdp.dp.Execute(frame, nil, flow.Actions)) + } + + if createFlow { + log.Println("Creating flow", flow) + checkWarn(fastdp.dp.CreateFlow(flow)) + } +} + +type odpActionsFlowOp struct { + fastdp *FastDatapath + actions []odp.Action +} + +func (fastdp *FastDatapath) odpActions(actions ...odp.Action) FlowOp { + return odpActionsFlowOp{ + fastdp: fastdp, + actions: actions, + } +} + +func (fop odpActionsFlowOp) updateFlowSpec(flow *odp.FlowSpec) { + flow.AddActions(fop.actions) +} + +func (fop odpActionsFlowOp) Send(frame []byte, dec *EthernetDecoder, bc bool) { + fastdp := fop.fastdp + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + checkWarn(fastdp.dp.Execute(frame, nil, fop.actions)) +} + +type nopFlowOp struct{} + +func (nopFlowOp) Send([]byte, *EthernetDecoder, bool) { + // A nopFlowOp just provides a hint about flow creation, it + // doesn't send anything +} + +type vetoFlowCreationFlowOp struct { + nopFlowOp +} + +var vetoFlowCreation = vetoFlowCreationFlowOp{} + +type odpFlowKeyFlowOp struct { + key odp.FlowKey + nopFlowOp +} + +func odpFlowKey(key odp.FlowKey) FlowOp { + return odpFlowKeyFlowOp{key: key} +} + +func (fop odpFlowKeyFlowOp) updateFlowSpec(flow *odp.FlowSpec) { + flow.AddKey(fop.key) +} + +func odpEthernetFlowKey(key PacketKey) FlowOp { + fk := odp.NewEthernetFlowKey() + fk.SetEthSrc(key.SrcMAC) + fk.SetEthDst(key.DstMAC) + return odpFlowKeyFlowOp{key: fk} +} + +func simpleThunk(fop FlowOp) thunk { + return func() FlowOp { return fop } +} + +func callThunks(thunks []thunk) FlowOp { + if len(thunks) == 1 { + return thunks[0]() + } + + mfop := NewMultiFlowOp(false) + for _, thunk := range thunks { + mfop.Add(thunk()) + } + + return mfop +} + +func multiThunk(thunks []thunk) thunk { + return func() FlowOp { return callThunks(thunks) } +} + +// IntraHost + +func (fastdp *FastDatapath) ConsumeIntraHostPackets( + consumer IntraHostConsumer) error { + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + + if fastdp.bridgePorts[routerPortID] != nil { + panic("FastDatapath already has an IntraHostConsumer") + } + + fastdp.bridgePorts[routerPortID] = func(key PacketKey) thunk { + return func() FlowOp { + return consumer.CapturedPacket(key) + } + } + + return fastdp.consumeMisses() +} + +func (fastdp *FastDatapath) InjectPacket(key PacketKey) FlowOp { + return callThunks(func() []thunk { + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + return fastdp.bridge(routerPortID, key) + }()) +} + +// A trivial bridge implementation +func (fastdp *FastDatapath) bridge(ingress bridgePortID, + key PacketKey) []thunk { + if fastdp.bridgeMACs[key.SrcMAC] == nil { + // Learn the source MAC + fastdp.bridgeMACs[key.SrcMAC] = fastdp.bridgePorts[ingress] + } + + // If we know about the destination MAC, deliver it to the + // associated port. + dst := fastdp.bridgeMACs[key.DstMAC] + if dst != nil { + return []thunk{simpleThunk(odpEthernetFlowKey(key)), dst(key)} + } + + // Otherwise, it might be a real broadcast, or it might just + // be for a MAC we don't know about yet. Either way, we'll + // broadcast it. + var fop FlowOp + thunks := make([]thunk, 1, len(fastdp.bridgePorts)+2) + + if (key.DstMAC[0] & 1) == 0 { + // Not a real broadcast, so doon't create a flow rule. + // If we did, we'd need to clear the flows every time + // we learned a new MAC address, or have a more + // complicated selective invalidation scheme. + fop = vetoFlowCreation + } else { + // A real broadcast + fop = odpEthernetFlowKey(key) + + if !ingress.router { + // The rule depends on the ingress vport. + thunks = thunks[:2] + thunks[1] = simpleThunk(odpFlowKey(odp.NewInPortFlowKey( + ingress.vport))) + } + } + + thunks[0] = simpleThunk(fop) + + // Send to all ports except the one it came in on. + for id, dst := range fastdp.bridgePorts { + if id != ingress { + thunks = append(thunks, dst(key)) + } + } + + return thunks +} + +// InterHost + +func (fastdp *FastDatapath) ConsumeInterHostPackets( + localPeer *Peer, peers *Peers, consumer InterHostConsumer) error { + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + + if fastdp.missHandlers[fastdp.vxlanVportID] != nil { + panic("FastDatapath already has an InterHostConsumer") + } + + fastdp.localPeer = localPeer + fastdp.peers = peers + + fastdp.missHandlers[fastdp.vxlanVportID] = func(fks odp.FlowKeys) thunk { + tunnel := fks[odp.OVS_KEY_ATTR_TUNNEL].(odp.TunnelFlowKey).Key() + srcPeer, dstPeer := fastdp.extractPeers(tunnel.TunnelId) + if srcPeer == nil || dstPeer == nil { + return simpleThunk(vetoFlowCreation) + } + + key := ForwardPacketKey{ + SrcPeer: srcPeer, + DstPeer: dstPeer, + PacketKey: flowKeysToPacketKey(fks), + } + + // The resulting flow rule should be restricted to + // packets with the same tunnelID + var tunnelFlowKey odp.TunnelFlowKey + tunnelFlowKey.SetTunnelId(tunnel.TunnelId) + tunnelFlowKey.SetIpv4Src(tunnel.Ipv4Src) + tunnelFlowKey.SetIpv4Dst(tunnel.Ipv4Dst) + + return func() FlowOp { + fop := NewMultiFlowOp(false) + fop.Add(odpFlowKey(tunnelFlowKey)) + fop.Add(consumer(key)) + return fop + } + } + + return fastdp.consumeMisses() +} + +func (fastdp *FastDatapath) InvalidateRoutes() { + fmt.Println("InvalidateRoutes") + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + checkWarn(fastdp.clearFlows()) +} + +func (fastdp *FastDatapath) InvalidateShortIDs() { + fmt.Println("InvalidateShortIDs") + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + checkWarn(fastdp.clearFlows()) +} + +func (fastdp *FastDatapath) extractPeers(tunnelID [8]byte) (*Peer, *Peer) { + if fastdp.peers == nil { + return nil, nil + } + + vni := binary.BigEndian.Uint64(tunnelID[:]) + srcPeer := fastdp.peers.FetchByShortID(PeerShortID(vni & 0xfff)) + dstPeer := fastdp.peers.FetchByShortID(PeerShortID((vni >> 12) & 0xfff)) + return srcPeer, dstPeer +} + +type FastDatapathForwarder struct { + fastdp *FastDatapath + remotePeer *Peer + localIP [4]byte + sendControlMsg func([]byte) error + + lock sync.Mutex + listener InterHostForwarderListener + remoteIP [4]byte + established bool +} + +func (fastdp *FastDatapath) MakeForwarder(remotePeer *Peer, localIP net.IP, + remote *net.UDPAddr, connUID uint64, crypto InterHostCrypto, + sendControlMsg func([]byte) error) (InterHostForwarder, error) { + if len(localIP) != 4 { + return nil, fmt.Errorf("local IP address %s is not IPv4", localIP) + } + + fwd := &FastDatapathForwarder{ + fastdp: fastdp, + remotePeer: remotePeer, + sendControlMsg: sendControlMsg, + } + copy(fwd.localIP[:], localIP) + return fwd, nil +} + +func (fwd *FastDatapathForwarder) SetListener( + listener InterHostForwarderListener) { + fwd.lock.Lock() + defer fwd.lock.Unlock() + + if listener == nil { + panic("nil listener") + } + + fwd.listener = listener + if fwd.established { + listener.Established() + } + + fwd.sendControlMsg(fwd.localIP[:]) +} + +func (fwd *FastDatapathForwarder) ControlMessage(msg []byte) { + fwd.lock.Lock() + defer fwd.lock.Unlock() + + if len(msg) != 4 { + if fwd.listener != nil { + fwd.listener.Error(fmt.Errorf("FastDatapath control message wrong length %d", len(msg))) + fwd.listener = nil + } + + return + } + + if !fwd.established { + copy(fwd.remoteIP[:], msg) + fwd.established = true + if fwd.listener != nil { + fwd.listener.Established() + } + } +} + +func (fwd *FastDatapathForwarder) Forward(key ForwardPacketKey) FlowOp { + fwd.lock.Lock() + defer fwd.lock.Unlock() + if !fwd.established { + // Ideally we could just return nil. But then we + // would have to invalidate the resulting flows when + // we learn the remote IP. So for now, just prevent + // flows. + return vetoFlowCreation + } + + var sta odp.SetTunnelAction + sta.SetTunnelId(tunnelIDFor(key)) + sta.SetIpv4Src(fwd.localIP) + sta.SetIpv4Dst(fwd.remoteIP) + sta.SetTos(0) + sta.SetTtl(64) + sta.SetDf(true) + sta.SetCsum(false) + + return fwd.fastdp.odpActions(sta, + odp.NewOutputAction(fwd.fastdp.vxlanVportID)) +} + +func tunnelIDFor(key ForwardPacketKey) (tunnelID [8]byte) { + src := uint64(key.SrcPeer.ShortID) + dst := uint64(key.DstPeer.ShortID) + binary.BigEndian.PutUint64(tunnelID[:], src|dst<<12) + return +} + +func (fwd *FastDatapathForwarder) Close() { + // Ideally we would delete all the relevant flows here. But + // until we do that, it's probably not worth clearing all + // flows. +} diff --git a/router/flow.go b/router/flow.go index ebb05f32..4a23894d 100644 --- a/router/flow.go +++ b/router/flow.go @@ -45,6 +45,28 @@ func (mfop *MultiFlowOp) Add(ops ...FlowOp) { func (mfop *MultiFlowOp) Send(frame []byte, dec *EthernetDecoder, broadcast bool) { for _, op := range mfop.ops { - op.Send(frame, dec, mfop.broadcast) + if op != nil { + op.Send(frame, dec, mfop.broadcast) + } } } + +func FlattenFlowOp(fop FlowOp) []FlowOp { + return collectFlowOps(nil, fop) +} + +func collectFlowOps(into []FlowOp, fop FlowOp) []FlowOp { + if fop == nil { + return into + } + + if mfop, ok := fop.(*MultiFlowOp); ok { + for _, op := range mfop.ops { + into = collectFlowOps(into, op) + } + + return into + } + + return append(into, fop) +} diff --git a/router/interhost.go b/router/interhost.go index 1fba3fe1..79d271db 100644 --- a/router/interhost.go +++ b/router/interhost.go @@ -7,7 +7,7 @@ import ( // Interface to inter-host (i.e. overlay network) packet handling type InterHost interface { // Start consuming forwarded packets. - ConsumePackets(*Peer, *Peers, InterHostConsumer) error + ConsumeInterHostPackets(*Peer, *Peers, InterHostConsumer) error // The routes have changed, so any cached information should // be discarded. diff --git a/router/intrahost.go b/router/intrahost.go index 784bf4d6..cacb0d0d 100644 --- a/router/intrahost.go +++ b/router/intrahost.go @@ -6,7 +6,7 @@ type IntraHost interface { InjectPacket(PacketKey) FlowOp // Start consuming packets - ConsumePackets(IntraHostConsumer) error + ConsumeIntraHostPackets(IntraHostConsumer) error } type IntraHostConsumer interface { @@ -22,7 +22,7 @@ func (NullIntraHost) InjectPacket(PacketKey) FlowOp { return nil } -func (NullIntraHost) ConsumePackets(IntraHostConsumer) error { +func (NullIntraHost) ConsumeIntraHostPackets(IntraHostConsumer) error { return nil } diff --git a/router/odp.go b/router/odp.go new file mode 100644 index 00000000..a0d8b17b --- /dev/null +++ b/router/odp.go @@ -0,0 +1,60 @@ +package router + +import ( + "github.com/weaveworks/go-odp/odp" +) + +// ODP admin functionality + +func CreateDatapath(dpname string) error { + dpif, err := odp.NewDpif() + if err != nil { + return err + } + + defer dpif.Close() + + _, err = dpif.CreateDatapath(dpname) + if err != nil && !odp.IsDatapathNameAlreadyExistsError(err) { + return err + } + + return nil +} + +func DeleteDatapath(dpname string) error { + dpif, err := odp.NewDpif() + if err != nil { + return err + } + + defer dpif.Close() + + dp, err := dpif.LookupDatapath(dpname) + if err != nil { + if odp.IsNoSuchDatapathError(err) { + return nil + } else { + return err + } + } + + return dp.Delete() +} + +func AddDatapathInterface(dpname string, ifname string) error { + dpif, err := odp.NewDpif() + if err != nil { + return err + } + + defer dpif.Close() + + dp, err := dpif.LookupDatapath(dpname) + if err != nil { + return err + } + + _, err = dp.CreateVport(odp.NewNetdevVportSpec(ifname)) + return err +} diff --git a/router/pcap.go b/router/pcap.go index 547153ef..7ed5b169 100644 --- a/router/pcap.go +++ b/router/pcap.go @@ -31,7 +31,7 @@ func NewPcap(iface *net.Interface, bufSz int) (IntraHost, error) { return &Pcap{iface: iface, bufSz: bufSz, writeHandle: wh}, nil } -func (p *Pcap) ConsumePackets(consumer IntraHostConsumer) error { +func (p *Pcap) ConsumeIntraHostPackets(consumer IntraHostConsumer) error { rh, err := newPcapHandle(p.iface.Name, true, 65535, p.bufSz) if err != nil { return err diff --git a/router/router.go b/router/router.go index 38c3dd95..52b91e76 100644 --- a/router/router.go +++ b/router/router.go @@ -67,8 +67,8 @@ func NewRouter(config RouterConfig, name PeerName, nickName string) *Router { // Start listening for packets from containers, TCP connections and // packets forwarded over UDP. func (router *Router) Start() { - checkFatal(router.InterHost.ConsumePackets(router.Ourself.Peer, router.Peers, router.handleForwardedPacket)) - checkFatal(router.IntraHost.ConsumePackets(router)) + checkFatal(router.InterHost.ConsumeInterHostPackets(router.Ourself.Peer, router.Peers, router.handleForwardedPacket)) + checkFatal(router.IntraHost.ConsumeIntraHostPackets(router)) router.listenTCP(router.Port) } diff --git a/router/sleeve.go b/router/sleeve.go index 009a2d6f..693f3762 100644 --- a/router/sleeve.go +++ b/router/sleeve.go @@ -39,8 +39,8 @@ func NewSleeveInterHost(localPort int) InterHost { return &SleeveInterHost{localPort: localPort} } -func (sleeve *SleeveInterHost) ConsumePackets(localPeer *Peer, peers *Peers, - consumer InterHostConsumer) error { +func (sleeve *SleeveInterHost) ConsumeInterHostPackets(localPeer *Peer, + peers *Peers, consumer InterHostConsumer) error { localAddr, err := net.ResolveUDPAddr("udp4", fmt.Sprint(":", sleeve.localPort)) if err != nil { diff --git a/weave b/weave index d65de36b..0a200fe7 100755 --- a/weave +++ b/weave @@ -256,7 +256,8 @@ random_mac() { create_bridge() { [ ! -d /sys/class/net/$BRIDGE ] && { - ip link add name $BRIDGE type bridge + docker run --rm --privileged --net=host $IMAGE --createdp --dpname=$BRIDGE + #ip link add name $BRIDGE type bridge # Set a random MAC address on the bridge. Current Linux # kernels already do this when creating a bridge, but there # are rumours it was not always so. @@ -265,9 +266,9 @@ create_bridge() { # fails. Bridges take the lowest MTU of their interfaces. So # instead we create a temporary interface with the desired # MTU, attach that to the bridge, and then remove it again. - ip link add name v${CONTAINER_IFNAME}du mtu $MTU type dummy - ip link set dev v${CONTAINER_IFNAME}du master $BRIDGE - ip link del dev v${CONTAINER_IFNAME}du + #ip link add name v${CONTAINER_IFNAME}du mtu $MTU type dummy + #ip link set dev v${CONTAINER_IFNAME}du master $BRIDGE + #ip link del dev v${CONTAINER_IFNAME}du # Drop traffic from Docker bridge to Weave; it can break subnet isolation if [ "$DOCKER_BRIDGE" != "$BRIDGE" ] ; then # Note using -I to insert ahead of Docker's bridge rules @@ -293,7 +294,8 @@ create_bridge() { destroy_bridge() { if [ -d /sys/class/net/$BRIDGE ] ; then - ip link del dev $BRIDGE + docker run --rm --privileged --net=host $IMAGE --deletedp --dpname=$BRIDGE + #ip link del dev $BRIDGE fi if [ "$DOCKER_BRIDGE" != "$BRIDGE" ] ; then run_iptables -t filter -D FORWARD -i $DOCKER_BRIDGE -o $BRIDGE -j DROP 2>/dev/null || true @@ -385,7 +387,7 @@ connect_container_to_bridge() { if ! netnsenter ip link set $GUEST_IFNAME down || ! netnsenter ip link set $GUEST_IFNAME name $CONTAINER_IFNAME up || ! ip link set $LOCAL_IFNAME up || - ! ip link set $LOCAL_IFNAME master $BRIDGE || + ! docker run --rm --privileged --net=host $IMAGE --adddpif=$LOCAL_IFNAME --dpname=$BRIDGE || ! configure_arp_cache $CONTAINER_IFNAME "netnsenter" ; then return 1 fi @@ -815,7 +817,7 @@ case "$COMMAND" in # when launching the weave container. CONTAINER=$(docker run --privileged -d --name=$CONTAINER_NAME \ -p $PORT:$CONTAINER_PORT/tcp -p $PORT:$CONTAINER_PORT/udp -e WEAVE_PASSWORD -v /var/run/docker.sock:/var/run/docker.sock --net=host \ - $WEAVE_DOCKER_ARGS $IMAGE -iface $CONTAINER_IFNAME -port $CONTAINER_PORT -name "$PEERNAME" -nickname "$(hostname)" $IPRANGE "$@") + $WEAVE_DOCKER_ARGS $IMAGE -dpname $BRIDGE -port $CONTAINER_PORT -name "$PEERNAME" -nickname "$(hostname)" $IPRANGE "$@") with_container_netns $CONTAINER launch >/dev/null wait_for_status $CONTAINER_NAME $HTTP_PORT diff --git a/weaver/main.go b/weaver/main.go index 3f1c94d2..28e6fa89 100644 --- a/weaver/main.go +++ b/weaver/main.go @@ -34,26 +34,36 @@ func main() { runtime.GOMAXPROCS(procs) var ( - config weave.RouterConfig - justVersion bool - ifaceName string - routerName string - nickName string - password string - wait int - debug bool - pktdebug bool - prof string - peers []string - bufSzMB int - httpAddr string - iprangeCIDR string - peerCount int - apiPath string - iface *net.Interface + // flags that cause immediate exit + justVersion bool + createDatapath bool + deleteDatapath bool + addDatapathInterface string + + config weave.RouterConfig + ifaceName string + routerName string + nickName string + password string + wait int + debug bool + pktdebug bool + prof string + peers []string + bufSzMB int + httpAddr string + iprangeCIDR string + peerCount int + apiPath string + iface *net.Interface + datapathName string ) flag.BoolVar(&justVersion, "version", false, "print version and exit") + flag.BoolVar(&createDatapath, "createdp", false, "create ODP datapath and exit") + flag.BoolVar(&deleteDatapath, "deletedp", false, "delete ODP datapath and exit") + flag.StringVar(&addDatapathInterface, "adddpif", "", "add a network interface to the ODP datapath and exit") + flag.IntVar(&config.Port, "port", weave.Port, "router port") flag.StringVar(&ifaceName, "iface", "", "name of interface to capture/inject from (disabled if blank)") flag.StringVar(&routerName, "name", "", "name of router (defaults to MAC of interface)") @@ -69,13 +79,25 @@ func main() { flag.StringVar(&iprangeCIDR, "iprange", "", "IP address range to allocate within, in CIDR notation") flag.IntVar(&peerCount, "initpeercount", 0, "number of peers in network (for IP address allocation)") flag.StringVar(&apiPath, "api", "unix:///var/run/docker.sock", "Path to Docker API socket") + flag.StringVar(&datapathName, "dpname", "", "ODP datapath name") flag.Parse() peers = flag.Args() InitDefaultLogging(debug) - if justVersion { + + switch { + case justVersion: fmt.Printf("weave router %s\n", version) os.Exit(0) + case createDatapath: + checkFatal(weave.CreateDatapath(datapathName)) + os.Exit(0) + case deleteDatapath: + checkFatal(weave.DeleteDatapath(datapathName)) + os.Exit(0) + case addDatapathInterface != "": + checkFatal(weave.AddDatapathInterface(datapathName, addDatapathInterface)) + os.Exit(0) } log.Println("Command line options:", options()) @@ -83,17 +105,32 @@ func main() { var err error + if password == "" { + password = os.Getenv("WEAVE_PASSWORD") + } + if ifaceName != "" { iface, err := weavenet.EnsureInterface(ifaceName, wait) - if err != nil { - log.Fatal(err) - } - + checkFatal(err) // bufsz flag is in MB config.IntraHost, err = weave.NewPcap(iface, bufSzMB*1024*1024) - if err != nil { - log.Fatal(err) + checkFatal(err) + config.InterHost = weave.NewSleeveInterHost(config.Port) + } + + if datapathName != "" { + if ifaceName != "" { + log.Fatal("Both -iface and -dpname supplied") } + + if password != "" { + log.Fatal("-password supplied; encryption is not suppoorted with fast datapath") + } + + fastdp, err := weave.NewFastDatapath(datapathName, config.Port) + checkFatal(err) + config.IntraHost = fastdp + config.InterHost = fastdp } if routerName == "" { @@ -104,19 +141,11 @@ func main() { } name, err := weave.PeerNameFromUserInput(routerName) - if err != nil { - log.Fatal(err) - } + checkFatal(err) if nickName == "" { nickName, err = os.Hostname() - if err != nil { - log.Fatal(err) - } - } - - if password == "" { - password = os.Getenv("WEAVE_PASSWORD") + checkFatal(err) } if password == "" { @@ -133,7 +162,6 @@ func main() { defer profile.Start(&p).Stop() } - config.InterHost = weave.NewSleeveInterHost(config.Port) config.LogFrame = logFrameFunc(pktdebug) router := weave.NewRouter(config, name, nickName) @@ -195,17 +223,13 @@ func logFrameFunc(debug bool) weave.LogFrameFunc { func initiateConnections(router *weave.Router, peers []string) { for _, peer := range peers { - if err := router.ConnectionMaker.InitiateConnection(peer); err != nil { - log.Fatal(err) - } + checkFatal(router.ConnectionMaker.InitiateConnection(peer)) } } func createAllocator(router *weave.Router, apiPath string, iprangeCIDR string, quorum uint) *ipam.Allocator { allocator, err := ipam.NewAllocator(router.Ourself.Peer.Name, router.Ourself.Peer.UID, router.Ourself.Peer.NickName, iprangeCIDR, quorum) - if err != nil { - log.Fatal(err) - } + checkFatal(err) allocator.SetInterfaces(router.NewGossip("IPallocation", allocator)) allocator.Start() err = updater.Start(apiPath, allocator) @@ -288,3 +312,9 @@ func handleHTTP(router *weave.Router, httpAddr string, allocator *ipam.Allocator log.Fatal("Unable to create http server", err) } } + +func checkFatal(e error) { + if e != nil { + log.Fatal(e) + } +}