From 295876e1d113017fe3d37d4e50aa56a530a03327 Mon Sep 17 00:00:00 2001 From: David Wragg Date: Wed, 9 Sep 2015 11:55:39 +0100 Subject: [PATCH] Fast datapath --- prog/weaver/main.go | 93 +++- router/consts.go | 20 +- router/fastdp.go | 1115 +++++++++++++++++++++++++++++++++++++++++++ router/flow.go | 32 +- router/odp.go | 60 +++ router/sleeve.go | 20 +- 6 files changed, 1303 insertions(+), 37 deletions(-) create mode 100644 router/fastdp.go create mode 100644 router/odp.go diff --git a/prog/weaver/main.go b/prog/weaver/main.go index eb02e1ea..5599728d 100644 --- a/prog/weaver/main.go +++ b/prog/weaver/main.go @@ -13,6 +13,7 @@ import ( "github.com/davecheney/profile" "github.com/docker/docker/pkg/mflag" "github.com/gorilla/mux" + "github.com/weaveworks/go-odp/odp" . "github.com/weaveworks/weave/common" "github.com/weaveworks/weave/common/docker" @@ -35,8 +36,13 @@ func main() { runtime.GOMAXPROCS(procs) var ( + // flags that cause immediate exit + justVersion bool + createDatapath bool + deleteDatapath bool + addDatapathInterface string + config weave.Config - justVersion bool protocolMinVersion int ifaceName string routerName string @@ -60,9 +66,14 @@ func main() { dnsClientTimeout time.Duration dnsEffectiveListenAddress string iface *net.Interface + datapathName string ) mflag.BoolVar(&justVersion, []string{"#version", "-version"}, false, "print version and exit") + mflag.BoolVar(&createDatapath, []string{"-create-datapath"}, false, "create ODP datapath and exit") + mflag.BoolVar(&deleteDatapath, []string{"-delete-datapath"}, false, "delete ODP datapath and exit") + mflag.StringVar(&addDatapathInterface, []string{"-add-datapath-iface"}, "", "add a network interface to the ODP datapath and exit") + mflag.IntVar(&config.Port, []string{"#port", "-port"}, weave.Port, "router port") mflag.IntVar(&protocolMinVersion, []string{"-min-protocol-version"}, weave.ProtocolMinVersion, "minimum weave protocol version") mflag.StringVar(&ifaceName, []string{"#iface", "-iface"}, "", "name of interface to capture/inject from (disabled if blank)") @@ -86,6 +97,7 @@ func main() { mflag.IntVar(&dnsTTL, []string{"-dns-ttl"}, nameserver.DefaultTTL, "TTL for DNS request from our domain") mflag.DurationVar(&dnsClientTimeout, []string{"-dns-fallback-timeout"}, nameserver.DefaultClientTimeout, "timeout for fallback DNS requests") mflag.StringVar(&dnsEffectiveListenAddress, []string{"-dns-effective-listen-address"}, "", "address DNS will actually be listening, after Docker port mapping") + mflag.StringVar(&datapathName, []string{"-datapath"}, "", "ODP datapath name") // crude way of detecting that we probably have been started in a // container, with `weave launch` --> suppress misleading paths in @@ -94,14 +106,37 @@ func main() { os.Args[0] = "weave" mflag.CommandLine.Init("weave", mflag.ExitOnError) } + mflag.Parse() peers = mflag.Args() SetLogLevel(logLevel) - if justVersion { + + switch { + case justVersion: fmt.Printf("weave router %s\n", version) os.Exit(0) + + case createDatapath: + err := weave.CreateDatapath(datapathName) + if odp.IsKernelLacksODPError(err) { + // When the kernel lacks ODP support, exit + // with a special status to distinguish it for + // the weave script. + os.Exit(17) + } + + checkFatal(err) + 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()) @@ -112,9 +147,29 @@ func main() { } config.ProtocolMinVersion = byte(protocolMinVersion) - var err error + var fastDPOverlay weave.Overlay + if datapathName != "" { + // A datapath name implies that "Bridge" and "Overlay" + // packet handling use fast datapath, although other + // options can override that below. Even if both + // things are overridden, we might need bridging on + // the datapath. + fastdp, err := weave.NewFastDatapath(weave.FastDatapathConfig{ + DatapathName: datapathName, + Port: config.Port, + }) + + checkFatal(err) + config.Bridge = fastdp.Bridge() + fastDPOverlay = fastdp.Overlay() + } if ifaceName != "" { + // -iface can coexist with -datapath, because + // pcap-based packet capture is a bit more efficient + // than capture via ODP misses, even when using an + // ODP-based bridge. So when using weave encyption, + // it's preferable to use -iface. iface, err := weavenet.EnsureInterface(ifaceName) checkFatal(err) @@ -123,6 +178,27 @@ func main() { checkFatal(err) } + if password == "" { + password = os.Getenv("WEAVE_PASSWORD") + } + + if password == "" { + Log.Println("Communication between peers is unencrypted.") + } else { + config.Password = []byte(password) + Log.Println("Communication between peers is encrypted.") + + // fastdp doesn't support encryption + fastDPOverlay = nil + } + + overlays := weave.NewOverlaySwitch() + if fastDPOverlay != nil { + overlays.Add("fastdp", fastDPOverlay) + } + overlays.Add("sleeve", weave.NewSleeveOverlay(config.Port)) + config.Overlay = overlays + if routerName == "" { if iface == nil { Log.Fatal("Either an interface must be specified with --iface or a name with -name") @@ -138,16 +214,6 @@ func main() { checkFatal(err) } - if password == "" { - password = os.Getenv("WEAVE_PASSWORD") - } - if password == "" { - Log.Println("Communication between peers is unencrypted.") - } else { - config.Password = []byte(password) - Log.Println("Communication between peers is encrypted.") - } - if prof != "" { p := *profile.CPUProfile p.ProfilePath = prof @@ -156,7 +222,6 @@ func main() { } config.PeerDiscovery = !noDiscovery - config.Overlay = weave.NewSleeveOverlay(config.Port) if pktdebug { config.PacketLogging = packetLogging{} diff --git a/router/consts.go b/router/consts.go index f8b718c2..6fd11588 100644 --- a/router/consts.go +++ b/router/consts.go @@ -6,12 +6,16 @@ import ( ) const ( - Port = 6783 - HTTPPort = Port + 1 - MaxUDPPacketSize = 65535 - ChannelSize = 16 - TCPHeartbeat = 30 * time.Second - GossipInterval = 30 * time.Second - MaxDuration = time.Duration(math.MaxInt64) - MaxTCPMsgSize = 10 * 1024 * 1024 + Port = 6783 + HTTPPort = Port + 1 + MaxUDPPacketSize = 65535 + ChannelSize = 16 + TCPHeartbeat = 30 * time.Second + GossipInterval = 30 * time.Second + MaxDuration = time.Duration(math.MaxInt64) + MaxTCPMsgSize = 10 * 1024 * 1024 + FastHeartbeat = 500 * time.Millisecond + SlowHeartbeat = 10 * time.Second + MaxMissedHeartbeats = 6 + HeartbeatTimeout = MaxMissedHeartbeats * SlowHeartbeat ) diff --git a/router/fastdp.go b/router/fastdp.go new file mode 100644 index 00000000..7208094a --- /dev/null +++ b/router/fastdp.go @@ -0,0 +1,1115 @@ +package router + +import ( + "encoding/binary" + "fmt" + "net" + "sync" + "time" + + "github.com/weaveworks/go-odp/odp" +) + +// The virtual bridge accepts packets from ODP vports and the router +// port (i.e. InjectPacket). We need a map key to index those +// possibilities: +type bridgePortID struct { + vport odp.VportID + router bool +} + +// A bridgeSender sends out a packet from the virtual bridge +type bridgeSender func(key PacketKey, lock *fastDatapathLock) FlowOp + +// A missHandler handles an ODP miss +type missHandler func(fks odp.FlowKeys, lock *fastDatapathLock) FlowOp + +type FastDatapath struct { + dpname string + + // The mtu from the datapath netdev (which should match the + // mtus on all the veths hooked up to the datapath). We + // validate that we are able to support that mtu. + mtu int + + // The lock guards the FastDatapath state, and also + // synchronizes use of the dpif + lock sync.Mutex + dpif *odp.Dpif + dp odp.DatapathHandle + deleteFlowsCount uint64 + missHandlers map[odp.VportID]missHandler + localPeer *Peer + peers *Peers + overlayConsumer OverlayConsumer + + // Bridge state: How to send to the given bridge port + sendToPort map[bridgePortID]bridgeSender + + // How to send to a given destination MAC + sendToMAC map[MAC]bridgeSender + + // MACs seen on the bridge recently + seenMACs map[MAC]struct{} + + // vxlan vports associated with the given UDP ports + vxlanVportIDs map[int]odp.VportID + mainVxlanVportID odp.VportID + + // A singleton pool for the occasions when we need to decode + // the packet. + dec *EthernetDecoder + + // forwarders by remote peer + forwarders map[PeerName]*fastDatapathForwarder +} + +type FastDatapathConfig struct { + DatapathName string + Port int + ExpireFlowsInterval time.Duration + ExpireMACsInterval time.Duration +} + +func NewFastDatapath(config FastDatapathConfig) (*FastDatapath, error) { + dpif, err := odp.NewDpif() + if err != nil { + return nil, err + } + + success := false + defer func() { + if !success { + dpif.Close() + } + }() + + dp, err := dpif.LookupDatapath(config.DatapathName) + if err != nil { + return nil, err + } + + iface, err := net.InterfaceByName(config.DatapathName) + if err != nil { + return nil, err + } + + fastdp := &FastDatapath{ + dpname: config.DatapathName, + mtu: iface.MTU, + dpif: dpif, + dp: dp, + missHandlers: make(map[odp.VportID]missHandler), + sendToPort: nil, + sendToMAC: make(map[MAC]bridgeSender), + seenMACs: make(map[MAC]struct{}), + vxlanVportIDs: make(map[int]odp.VportID), + forwarders: make(map[PeerName]*fastDatapathForwarder), + } + + if err := fastdp.deleteVxlanVports(); err != nil { + return nil, err + } + + if err := fastdp.deleteFlows(); err != nil { + return nil, err + } + + // We use the weave port number plus 1 for vxlan. Hard-coding + // this relationship may seem dubious, but there is no moral + // difference between this and requiring that the sleeve UDP + // port number is the same as the TCP port number. The hard + // part would be not adding weaver flags to allow the port + // numbers to be independent, but working out how to specify + // them on the connecting side. So we can wait to find out if + // anyone wants that. + fastdp.mainVxlanVportID, err = fastdp.getVxlanVportID(config.Port + 1) + if err != nil { + return nil, err + } + + // need to lock before we might receive events + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + + if _, err := dp.ConsumeMisses(fastdp); err != nil { + return nil, err + } + + if _, err := dp.ConsumeVportEvents(fastdp); err != nil { + return nil, err + } + + vports, err := dp.EnumerateVports() + if err != nil { + return nil, err + } + + for _, vport := range vports { + fastdp.makeBridgeVport(vport) + } + + success = true + go fastdp.run() + return fastdp, nil +} + +func (fastdp *FastDatapath) Close() error { + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + err := fastdp.dpif.Close() + fastdp.dpif = nil + return err +} + +// While processing a packet, we can potentially acquire and drop the +// FastDatapath lock many times (acquiring it to acceess FastDatapath +// state, and invoke ODP operations; dropping it to invoke callbacks +// that may re-enter the FastDatapath). A fastDatapathLock +// coordinates this process. +type fastDatapathLock struct { + fastdp *FastDatapath + locked bool + + // While the lock is dropped, deleteFlows could be called. We + // need to detect when this happens and avoid creating flows, + // because they may be based on stale information. + deleteFlowsCount uint64 +} + +func (fastdp *FastDatapath) startLock() fastDatapathLock { + fastdp.lock.Lock() + return fastDatapathLock{ + fastdp: fastdp, + locked: true, + deleteFlowsCount: fastdp.deleteFlowsCount, + } +} + +func (lock *fastDatapathLock) unlock() { + if lock.locked { + lock.fastdp.lock.Unlock() + lock.locked = false + } +} + +func (lock *fastDatapathLock) relock() { + if !lock.locked { + lock.fastdp.lock.Lock() + lock.locked = true + } +} + +// Bridge bits + +type fastDatapathBridge struct { + *FastDatapath +} + +func (fastdp *FastDatapath) Bridge() Bridge { + return fastDatapathBridge{fastdp} +} + +func (fastdp fastDatapathBridge) String() string { + return fmt.Sprint(fastdp.dpname, " (via ODP)") +} + +func (fastDatapathBridge) Stats() map[string]int { + return nil +} + +var routerBridgePortID = bridgePortID{router: true} + +func (fastdp fastDatapathBridge) StartConsumingPackets(consumer BridgeConsumer) error { + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + + if fastdp.sendToPort[routerBridgePortID] != nil { + return fmt.Errorf("FastDatapath already has a BridgeConsumer") + } + + // set up delivery to the weave router port on the bridge + fastdp.addSendToPort(routerBridgePortID, + func(key PacketKey, lock *fastDatapathLock) FlowOp { + // drop the FastDatapath lock in order to call + // the consumer + lock.unlock() + return consumer(key) + }) + return nil +} + +func (fastdp fastDatapathBridge) InjectPacket(key PacketKey) FlowOp { + lock := fastdp.startLock() + defer lock.unlock() + return fastdp.bridge(routerBridgePortID, key, &lock) +} + +// Ethernet bridge implementation + +func (fastdp *FastDatapath) bridge(ingress bridgePortID, + key PacketKey, lock *fastDatapathLock) FlowOp { + lock.relock() + if fastdp.sendToMAC[key.SrcMAC] == nil { + // Learn the source MAC + fastdp.sendToMAC[key.SrcMAC] = fastdp.sendToPort[ingress] + fastdp.seenMACs[key.SrcMAC] = struct{}{} + } + + // If we know about the destination MAC, deliver it to the + // associated port. + if sender := fastdp.sendToMAC[key.DstMAC]; sender != nil { + return NewMultiFlowOp(false, odpEthernetFlowKey(key), + sender(key, lock)) + } + + // Otherwise, it might be a real broadcast, or it might + // be for a MAC we don't know about yet. Either way, we'll + // broadcast it. + mfop := NewMultiFlowOp(false) + + if (key.DstMAC[0] & 1) == 0 { + // Not a real broadcast, so don't create a flow rule. + // If we did, we'd need to delete the flows every time + // we learned a new MAC address, or have a more + // complicated selective invalidation scheme. + mfop.Add(vetoFlowCreationFlowOp{}) + } else { + // A real broadcast + mfop.Add(odpEthernetFlowKey(key)) + } + + // Send to all ports except the one it came in on. The + // sendToPort map is immutable, so it is safe to iterate over + // it even though the sender functions can drop the + // fastDatapathLock + for id, sender := range fastdp.sendToPort { + if id != ingress { + mfop.Add(sender(key, lock)) + } + } + + return mfop +} + +// Overlay bits + +type fastDatapathOverlay struct { + *FastDatapath +} + +func (fastdp *FastDatapath) Overlay() Overlay { + return fastDatapathOverlay{fastdp} +} + +func (fastdp fastDatapathOverlay) InvalidateRoutes() { + log.Debug("InvalidateRoutes") + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + checkWarn(fastdp.deleteFlows()) +} + +func (fastdp fastDatapathOverlay) InvalidateShortIDs() { + log.Debug("InvalidateShortIDs") + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + checkWarn(fastdp.deleteFlows()) +} + +func (fastDatapathOverlay) AddFeaturesTo(features map[string]string) { + // Nothing needed. Fast datapath support is indicated through + // OverlaySwitch. +} + +func (fastdp fastDatapathOverlay) StartConsumingPackets(localPeer *Peer, + peers *Peers, consumer OverlayConsumer) error { + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + + if fastdp.overlayConsumer != nil { + return fmt.Errorf("FastDatapath already has an OverlayConsumer") + } + + fastdp.localPeer = localPeer + fastdp.peers = peers + fastdp.overlayConsumer = consumer + return nil +} + +func (fastdp *FastDatapath) getVxlanVportID(udpPort int) (odp.VportID, error) { + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + + if vxlanVportID, present := fastdp.vxlanVportIDs[udpPort]; present { + return vxlanVportID, nil + } + + vxlanVportID, err := fastdp.dp.CreateVport( + odp.NewVxlanVportSpec(fmt.Sprintf("vxlan-%d", udpPort), + uint16(udpPort))) + if err != nil { + return 0, err + } + + fastdp.vxlanVportIDs[udpPort] = vxlanVportID + fastdp.missHandlers[vxlanVportID] = func(fks odp.FlowKeys, lock *fastDatapathLock) FlowOp { + tunnel := fks[odp.OVS_KEY_ATTR_TUNNEL].(odp.TunnelFlowKey) + tunKey := tunnel.Key() + + lock.relock() + consumer := fastdp.overlayConsumer + if consumer == nil { + return vetoFlowCreationFlowOp{} + } + + srcPeer, dstPeer := fastdp.extractPeers(tunKey.TunnelId) + if srcPeer == nil || dstPeer == nil { + return vetoFlowCreationFlowOp{} + } + + lock.unlock() + pk := flowKeysToPacketKey(fks) + var zeroMAC MAC + if pk.SrcMAC == zeroMAC && pk.DstMAC == zeroMAC { + return vxlanSpecialPacketFlowOp{fastdp, srcPeer, + &net.UDPAddr{ + IP: net.IP(tunKey.Ipv4Src[:]), + Port: udpPort, + }, + } + } + + key := ForwardPacketKey{ + SrcPeer: srcPeer, + DstPeer: dstPeer, + PacketKey: pk, + } + + var tunnelFlowKey odp.TunnelFlowKey + tunnelFlowKey.SetTunnelId(tunKey.TunnelId) + tunnelFlowKey.SetIpv4Src(tunKey.Ipv4Src) + tunnelFlowKey.SetIpv4Dst(tunKey.Ipv4Dst) + + return NewMultiFlowOp(false, odpFlowKey(tunnelFlowKey), + consumer(key)) + } + + return vxlanVportID, nil +} + +func (fastdp *FastDatapath) extractPeers(tunnelID [8]byte) (*Peer, *Peer) { + vni := binary.BigEndian.Uint64(tunnelID[:]) + srcPeer := fastdp.peers.FetchByShortID(PeerShortID(vni & 0xfff)) + dstPeer := fastdp.peers.FetchByShortID(PeerShortID((vni >> 12) & 0xfff)) + return srcPeer, dstPeer +} + +type vxlanSpecialPacketFlowOp struct { + fastdp *FastDatapath + srcPeer *Peer + sender *net.UDPAddr +} + +func (op vxlanSpecialPacketFlowOp) Process(frame []byte, dec *EthernetDecoder, + broadcast bool) { + op.fastdp.lock.Lock() + fwd := op.fastdp.forwarders[op.srcPeer.Name] + op.fastdp.lock.Unlock() + + if fwd != nil && dec.IsSpecial() { + fwd.handleVxlanSpecialPacket(frame, op.sender) + } +} + +type fastDatapathForwarder struct { + fastdp *FastDatapath + remotePeer *Peer + localIP [4]byte + sendControlMsg func(byte, []byte) error + connUID uint64 + vxlanVportID odp.VportID + + lock sync.RWMutex + confirmed bool + remoteAddr *net.UDPAddr + heartbeatInterval time.Duration + heartbeatTimer *time.Timer + heartbeatTimeout *time.Timer + ackedHeartbeat bool + stopChan chan struct{} + stopped bool + + establishedChan chan struct{} + errorChan chan error +} + +func (fastdp fastDatapathOverlay) MakeForwarder( + params ForwarderParams) (OverlayForwarder, error) { + if params.Crypto != nil { + // No encryption suport in fastdp. The weaver main.go + // is responsible for ensuring this doesn't happen. + log.Fatal("Attempt to use FastDatapath with encryption") + } + + vxlanVportID := fastdp.mainVxlanVportID + remoteAddr := params.RemoteAddr + if remoteAddr != nil { + // The provided address contains the main weave port + // number to connect to. We need to derive the vxlan + // port number from that. + vxlanRemoteAddr := *params.RemoteAddr + vxlanRemoteAddr.Port++ + remoteAddr = &vxlanRemoteAddr + + var err error + vxlanVportID, err = fastdp.getVxlanVportID(remoteAddr.Port) + if err != nil { + return nil, err + } + } + + localIP, err := ipv4Bytes(params.LocalIP) + if err != nil { + return nil, err + } + + fwd := &fastDatapathForwarder{ + fastdp: fastdp.FastDatapath, + remotePeer: params.RemotePeer, + localIP: localIP, + sendControlMsg: params.SendControlMessage, + connUID: params.ConnUID, + vxlanVportID: vxlanVportID, + + remoteAddr: remoteAddr, + heartbeatInterval: FastHeartbeat, + stopChan: make(chan struct{}), + + establishedChan: make(chan struct{}), + errorChan: make(chan error, 1), + } + + return fwd, err +} + +func ipv4Bytes(ip net.IP) (res [4]byte, err error) { + ipv4 := ip.To4() + if ipv4 != nil { + copy(res[:], ipv4) + } else { + err = fmt.Errorf("IP address %s is not IPv4", ip) + } + return +} + +func (fwd *fastDatapathForwarder) logPrefix() string { + return fmt.Sprintf("fastdp ->[%s|%s]: ", fwd.remoteAddr, fwd.remotePeer) +} + +func (fwd *fastDatapathForwarder) Confirm() { + fwd.lock.Lock() + defer fwd.lock.Unlock() + + if fwd.confirmed { + log.Fatal(fwd.logPrefix(), "already confirmed") + } + + log.Debug(fwd.logPrefix(), "confirmed") + fwd.fastdp.addForwarder(fwd.remotePeer.Name, fwd) + fwd.confirmed = true + + if fwd.remoteAddr != nil { + // have the goroutine send a heartbeat straight away + fwd.heartbeatTimer = time.NewTimer(0) + } else { + // we'll reset the timer when we learn the remote ip + fwd.heartbeatTimer = time.NewTimer(MaxDuration) + } + + fwd.heartbeatTimeout = time.NewTimer(HeartbeatTimeout) + go fwd.doHeartbeats() +} + +func (fwd *fastDatapathForwarder) EstablishedChannel() <-chan struct{} { + return fwd.establishedChan +} + +func (fwd *fastDatapathForwarder) ErrorChannel() <-chan error { + return fwd.errorChan +} + +func (fwd *fastDatapathForwarder) doHeartbeats() { + var err error + + for err == nil { + select { + case <-fwd.heartbeatTimer.C: + if fwd.confirmed { + fwd.sendHeartbeat() + } + fwd.heartbeatTimer.Reset(fwd.heartbeatInterval) + + case <-fwd.heartbeatTimeout.C: + err = fmt.Errorf("timed out waiting for vxlan heartbeat") + + case <-fwd.stopChan: + return + } + } + + fwd.lock.Lock() + defer fwd.lock.Unlock() + fwd.handleError(err) +} + +// Handle an error which leads to notifying the listener and +// termination of the forwarder +func (fwd *fastDatapathForwarder) handleError(err error) { + if err == nil { + return + } + + select { + case fwd.errorChan <- err: + default: + } + + // stop the heartbeat goroutine + if !fwd.stopped { + fwd.stopped = true + close(fwd.stopChan) + } +} + +func (fwd *fastDatapathForwarder) sendHeartbeat() { + fwd.lock.RLock() + log.Debug(fwd.logPrefix(), "sendHeartbeat") + + // the heartbeat payload consists of the 64-bit connection uid + // followed by the 16-bit packet size. + buf := make([]byte, EthernetOverhead+fwd.fastdp.mtu) + binary.BigEndian.PutUint64(buf[EthernetOverhead:], fwd.connUID) + binary.BigEndian.PutUint16(buf[EthernetOverhead+8:], uint16(len(buf))) + + dec := NewEthernetDecoder() + dec.DecodeLayers(buf) + pk := ForwardPacketKey{ + PacketKey: dec.PacketKey(), + SrcPeer: fwd.fastdp.localPeer, + DstPeer: fwd.remotePeer, + } + fwd.lock.RUnlock() + fwd.Forward(pk).Process(buf, dec, false) +} + +const ( + FastDatapathHeartbeatAck = iota +) + +func (fwd *fastDatapathForwarder) handleVxlanSpecialPacket(frame []byte, + sender *net.UDPAddr) { + fwd.lock.Lock() + defer fwd.lock.Unlock() + + log.Debug(fwd.logPrefix(), "handleVxlanSpecialPacket") + + // the only special packet type is a heartbeat + if len(frame) < EthernetOverhead+10 { + log.Warning(fwd.logPrefix(), "short vxlan special packet: ", len(frame), " bytes") + return + } + + if binary.BigEndian.Uint64(frame[EthernetOverhead:]) != fwd.connUID || + uint16(len(frame)) != binary.BigEndian.Uint16(frame[EthernetOverhead+8:]) { + return + } + + if fwd.remoteAddr == nil { + fwd.remoteAddr = sender + + if fwd.confirmed { + fwd.heartbeatTimer.Reset(0) + } + } else if !udpAddrsEqual(fwd.remoteAddr, sender) { + log.Info(fwd.logPrefix(), + "Peer IP address changed to ", sender) + fwd.remoteAddr = sender + } + + if !fwd.ackedHeartbeat { + fwd.ackedHeartbeat = true + fwd.handleError(fwd.sendControlMsg(FastDatapathHeartbeatAck, nil)) + } + + // we can receive a heartbeat before Confirm() has set up + // heartbeatTimeout + if fwd.heartbeatTimeout != nil { + fwd.heartbeatTimeout.Reset(HeartbeatTimeout) + } +} + +func (fwd *fastDatapathForwarder) ControlMessage(tag byte, msg []byte) { + fwd.lock.Lock() + defer fwd.lock.Unlock() + + switch tag { + case FastDatapathHeartbeatAck: + fwd.handleHeartbeatAck() + + default: + log.Info(fwd.logPrefix(), + "Ignoring unknown control message: ", tag) + } +} + +func (fwd *fastDatapathForwarder) handleHeartbeatAck() { + log.Debug(fwd.logPrefix(), "handleHeartbeatAck") + + if fwd.heartbeatInterval != SlowHeartbeat { + close(fwd.establishedChan) + fwd.heartbeatInterval = SlowHeartbeat + if fwd.heartbeatTimer != nil { + fwd.heartbeatTimer.Reset(fwd.heartbeatInterval) + } + } +} + +func (fwd *fastDatapathForwarder) Forward(key ForwardPacketKey) FlowOp { + fwd.lock.RLock() + defer fwd.lock.RUnlock() + + if fwd.remoteAddr == nil { + // Returning nil would discard the packet, but also + // result in a flow rule, which we would have to + // invalidate when we learn the remote IP. So for + // now, just prevent flows. + return vetoFlowCreationFlowOp{} + } + + remoteIP, err := ipv4Bytes(fwd.remoteAddr.IP) + if err != nil { + log.Error(err) + return nil + } + + var sta odp.SetTunnelAction + sta.SetTunnelId(tunnelIDFor(key)) + sta.SetIpv4Src(fwd.localIP) + sta.SetIpv4Dst(remoteIP) + sta.SetTos(0) + sta.SetTtl(64) + sta.SetDf(true) + sta.SetCsum(false) + return fwd.fastdp.odpActions(sta, odp.NewOutputAction(fwd.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) Stop() { + // Might be nice to delete all the relevant flows here, but we + // can just let them expire. + fwd.fastdp.removeForwarder(fwd.remotePeer.Name, fwd) + + fwd.lock.Lock() + defer fwd.lock.Unlock() + fwd.sendControlMsg = func(byte, []byte) error { return nil } + + // stop the heartbeat goroutine + if !fwd.stopped { + fwd.stopped = true + close(fwd.stopChan) + } +} + +func (fastdp *FastDatapath) addForwarder(peer PeerName, + fwd *fastDatapathForwarder) { + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + + // We shouldn't have two confirmed forwarders to the same + // remotePeer, due to the checks in LocalPeer AddConnection. + fastdp.forwarders[peer] = fwd +} + +func (fastdp *FastDatapath) removeForwarder(peer PeerName, + fwd *fastDatapathForwarder) { + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + if fastdp.forwarders[peer] == fwd { + delete(fastdp.forwarders, peer) + } +} + +func (fastdp *FastDatapath) deleteFlows() error { + fastdp.deleteFlowsCount++ + + flows, err := fastdp.dp.EnumerateFlows() + if err != nil { + return err + } + + for _, flow := range flows { + err = fastdp.dp.DeleteFlow(flow.FlowKeys) + 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) run() { + expireMACsCh := time.Tick(10 * time.Minute) + expireFlowsCh := time.Tick(5 * time.Minute) + + for { + select { + case <-expireMACsCh: + fastdp.expireMACs() + + case <-expireFlowsCh: + fastdp.expireFlows() + } + } +} + +func (fastdp *FastDatapath) expireMACs() { + lock := fastdp.startLock() + defer lock.unlock() + + for mac := range fastdp.sendToMAC { + if _, present := fastdp.seenMACs[mac]; !present { + delete(fastdp.sendToMAC, mac) + } + } + + fastdp.seenMACs = make(map[MAC]struct{}) +} + +func (fastdp *FastDatapath) expireFlows() { + lock := fastdp.startLock() + defer lock.unlock() + + flows, err := fastdp.dp.EnumerateFlows() + if err != nil { + log.Warn(err) + } + + for _, flow := range flows { + if flow.Used == 0 { + log.Debug("Expiring flow ", flow.FlowSpec) + err = fastdp.dp.DeleteFlow(flow.FlowKeys) + } else { + fastdp.touchFlow(flow.FlowKeys, &lock) + err = fastdp.dp.ClearFlow(flow.FlowSpec) + } + + if err != nil && !odp.IsNoSuchFlowError(err) { + log.Warn(err) + } + } +} + +// The router needs to know which flows are active in order to +// maintain its MAC->peer table. We do this by querying the router +// without an actual packet being involved. Maybe it's +// worth devising a more unified approach in the future. +func (fastdp *FastDatapath) touchFlow(fks odp.FlowKeys, + lock *fastDatapathLock) { + // All the flows we create should have an ingress key, but we + // check here just in case we encounter one from somewhere + // else. + ingressKey, present := fks[odp.OVS_KEY_ATTR_IN_PORT] + if present { + ingress := ingressKey.(odp.InPortFlowKey).VportID() + handler := fastdp.getMissHandler(ingress) + if handler != nil { + handler(fks, lock) + lock.relock() + } + } +} + +func (fastdp *FastDatapath) Error(err error, stopped bool) { + if stopped { + log.Fatal("Error while listeniing on ODP datapath: ", err) + } + + log.Error("Error while listening on ODP datapath: ", err) +} + +func (fastdp *FastDatapath) Miss(packet []byte, fks odp.FlowKeys) error { + ingress := fks[odp.OVS_KEY_ATTR_IN_PORT].(odp.InPortFlowKey).VportID() + log.Debug("ODP miss ", fks, " on port ", ingress) + + lock := fastdp.startLock() + defer lock.unlock() + + handler := fastdp.getMissHandler(ingress) + if handler == nil { + return nil + } + + // Always include the ingress vport in the flow key. While + // this is not strictly necessary in some cases (e.g. for + // delivery to a local netdev based on the dest MAC), + // including the ingress in every flow makes things simpler + // in touchFlow. + mfop := NewMultiFlowOp(false, handler(fks, &lock), + odpFlowKey(odp.NewInPortFlowKey(ingress))) + fastdp.send(mfop, packet, &lock) + return nil +} + +func (fastdp *FastDatapath) getMissHandler(ingress odp.VportID) missHandler { + handler := fastdp.missHandlers[ingress] + if handler == nil { + vport, err := fastdp.dp.LookupVport(ingress) + if err != nil { + log.Error(err) + return nil + } + + fastdp.makeBridgeVport(vport) + } + + return handler +} + +func (fastdp *FastDatapath) VportCreated(dpid odp.DatapathID, vport odp.Vport) error { + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + + if _, present := fastdp.missHandlers[vport.ID]; !present { + fastdp.makeBridgeVport(vport) + } + + return nil +} + +func (fastdp *FastDatapath) VportDeleted(dpid odp.DatapathID, vport odp.Vport) error { + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + + // there might be flow rules that still refer to the id of + // this vport. But we just allow them to expire. Unless we + // want prompt migration of MAC addresses, that should be + // fine. + delete(fastdp.missHandlers, vport.ID) + fastdp.deleteSendToPort(bridgePortID{vport: vport.ID}) + return nil +} + +func (fastdp *FastDatapath) makeBridgeVport(vport odp.Vport) { + // Set up a bridge port for netdev and internal vports. vxlan + // vports are handled separately, as they do not correspond to + // bridge ports (we set up the miss handler for them in + // getVxlanVportID). + typ := vport.Spec.TypeName() + if typ != "netdev" && typ != "internal" { + return + } + + vportID := vport.ID + + // Sending to the bridge port outputs on the vport: + fastdp.addSendToPort(bridgePortID{vport: vportID}, + func(_ PacketKey, _ *fastDatapathLock) FlowOp { + return fastdp.odpActions(odp.NewOutputAction(vportID)) + }) + + // Delete flows, in order to recalculate flows for broadcasts + // on the bridge. + checkWarn(fastdp.deleteFlows()) + + // Packets coming from the netdev are processed by the bridge + fastdp.missHandlers[vportID] = func(flowKeys odp.FlowKeys, lock *fastDatapathLock) FlowOp { + return fastdp.bridge(bridgePortID{vport: vportID}, + flowKeysToPacketKey(flowKeys), lock) + } +} + +func flowKeysToPacketKey(fks odp.FlowKeys) PacketKey { + eth := fks[odp.OVS_KEY_ATTR_ETHERNET].(odp.EthernetFlowKey).Key() + return PacketKey{SrcMAC: eth.EthSrc, DstMAC: eth.EthDst} +} + +// The sendToPort map is read-only, so this method does the copy in +// order to add an entry. +func (fastdp *FastDatapath) addSendToPort(portID bridgePortID, + sender bridgeSender) { + sendToPort := map[bridgePortID]bridgeSender{portID: sender} + for id, sender := range fastdp.sendToPort { + sendToPort[id] = sender + } + fastdp.sendToPort = sendToPort +} + +func (fastdp *FastDatapath) deleteSendToPort(portID bridgePortID) { + sendToPort := make(map[bridgePortID]bridgeSender) + for id, sender := range fastdp.sendToPort { + if id != portID { + sendToPort[id] = sender + } + } + fastdp.sendToPort = sendToPort +} + +// Send a packet, creating a corresponding ODP flow rule if possible +func (fastdp *FastDatapath) send(fops FlowOp, frame []byte, + lock *fastDatapathLock) { + // 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 FlowOp (e.g. a sleeve forwarding + // FlowOp), so send the packet through the + // FlowOp interface, decoding the packet + // lazily. + if dec == nil { + dec = fastdp.takeDecoder(lock) + dec.DecodeLayers(frame) + + // If we are sending the packet + // through the FlowOp interface, we + // mustn't create a flow, as that + // could prevent the proper handling + // of similar packets in the future. + createFlow = false + } + + if len(dec.decoded) != 0 { + lock.unlock() + fop.Process(frame, dec, false) + } + } + } + + if dec != nil { + // put the decoder back + lock.relock() + fastdp.dec = dec + } + + if len(flow.Actions) != 0 { + lock.relock() + checkWarn(fastdp.dp.Execute(frame, nil, flow.Actions)) + } + + if createFlow { + lock.relock() + // if the fastdp's deleteFlowsCount changed since we + // initially locked it, then we might have created a + // flow on the basis of stale information. It's fine + // to handle one packet like that, but it would be bad + // to introduce a stale flow. + if lock.deleteFlowsCount == fastdp.deleteFlowsCount { + log.Debug("Creating ODP flow ", flow) + checkWarn(fastdp.dp.CreateFlow(flow)) + } + } +} + +// Get the EthernetDecoder from the singleton pool +func (fastdp *FastDatapath) takeDecoder(lock *fastDatapathLock) *EthernetDecoder { + lock.relock() + dec := fastdp.dec + if dec == nil { + dec = NewEthernetDecoder() + } else { + fastdp.dec = nil + } + return dec +} + +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) Process(frame []byte, dec *EthernetDecoder, + broadcast bool) { + fastdp := fop.fastdp + fastdp.lock.Lock() + defer fastdp.lock.Unlock() + checkWarn(fastdp.dp.Execute(frame, nil, fop.actions)) +} + +type nopFlowOp struct{} + +func (nopFlowOp) Process([]byte, *EthernetDecoder, bool) { + // A nopFlowOp just provides a hint about flow creation, it + // doesn't send anything +} + +// A vetoFlowCreationFlowOp flags that no flow should be created +type vetoFlowCreationFlowOp struct { + nopFlowOp +} + +// A odpFlowKeyFlowOp adds a FlowKey to the resulting flow +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} +} diff --git a/router/flow.go b/router/flow.go index 2a7ac039..f2f5c791 100644 --- a/router/flow.go +++ b/router/flow.go @@ -34,12 +34,18 @@ type MultiFlowOp struct { ops []FlowOp } -func NewMultiFlowOp(broadcast bool) *MultiFlowOp { - return &MultiFlowOp{broadcast: broadcast} +func NewMultiFlowOp(broadcast bool, ops ...FlowOp) *MultiFlowOp { + mfop := &MultiFlowOp{broadcast: broadcast} + mfop.Add(ops...) + return mfop } func (mfop *MultiFlowOp) Add(ops ...FlowOp) { - mfop.ops = append(mfop.ops, ops...) + for _, op := range ops { + if op != nil { + mfop.ops = append(mfop.ops, op) + } + } } func (mfop *MultiFlowOp) Process(frame []byte, dec *EthernetDecoder, @@ -48,3 +54,23 @@ func (mfop *MultiFlowOp) Process(frame []byte, dec *EthernetDecoder, op.Process(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/odp.go b/router/odp.go new file mode 100644 index 00000000..cea42dd5 --- /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 + } + + 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/sleeve.go b/router/sleeve.go index c60e6702..a2e2fd2e 100644 --- a/router/sleeve.go +++ b/router/sleeve.go @@ -45,18 +45,14 @@ import ( // sleeveForwarder.mtu <--------------------------> const ( - EthernetOverhead = 14 - UDPOverhead = 28 // 20 bytes for IPv4, 8 bytes for UDP - DefaultMTU = 65535 - FragTestSize = 60001 - PMTUDiscoverySize = 60000 - FastHeartbeat = 500 * time.Millisecond - SlowHeartbeat = 10 * time.Second - FragTestInterval = 5 * time.Minute - MTUVerifyAttempts = 8 - MTUVerifyTimeout = 10 * time.Millisecond // doubled with each attempt - MaxMissedHeartbeats = 6 - HeartbeatTimeout = MaxMissedHeartbeats * SlowHeartbeat + EthernetOverhead = 14 + UDPOverhead = 28 // 20 bytes for IPv4, 8 bytes for UDP + DefaultMTU = 65535 + FragTestSize = 60001 + PMTUDiscoverySize = 60000 + FragTestInterval = 5 * time.Minute + MTUVerifyAttempts = 8 + MTUVerifyTimeout = 10 * time.Millisecond // doubled with each attempt ) type SleeveOverlay struct {