package router import ( "fmt" "math" "net" "os" "time" "github.com/weaveworks/mesh" "github.com/rajch/weave/common" "github.com/rajch/weave/db" weavenet "github.com/rajch/weave/net" ) const ( ChannelSize = 16 MaxUDPPacketSize = 65535 FastHeartbeat = 500 * time.Millisecond SlowHeartbeat = 10 * time.Second MaxMissedHeartbeats = 6 HeartbeatTimeout = MaxMissedHeartbeats * SlowHeartbeat MaxDuration = time.Duration(math.MaxInt64) NameSize = mesh.NameSize // should be greater than typical ARP cache expiries, i.e. > 3/2 * // /proc/sys/net/ipv4_neigh/*/base_reachable_time_ms on Linux macMaxAge = 10 * time.Minute ) var ( log = common.Log checkFatal = common.CheckFatal checkWarn = common.CheckWarn ) type NetworkConfig struct { BufSz int PacketLogging PacketLogging InjectorConsumer InjectorConsumer } type PacketLogging interface { LogPacket(string, PacketKey) LogForwardPacket(string, ForwardPacketKey) } type NetworkRouter struct { *mesh.Router NetworkConfig weavenet.BridgeConfig Macs *MacCache db db.DB } func NewNetworkRouter(config mesh.Config, networkConfig NetworkConfig, bridgeConfig weavenet.BridgeConfig, name mesh.PeerName, nickName string, overlay NetworkOverlay, db db.DB) (*NetworkRouter, error) { if overlay == nil { overlay = NullNetworkOverlay{} } if networkConfig.InjectorConsumer == nil { networkConfig.InjectorConsumer = NullInjectorConsumer{} } meshRouter, err := mesh.NewRouter(config, name, nickName, overlay, common.LogLogger()) if err != nil { return nil, err } router := &NetworkRouter{Router: meshRouter, NetworkConfig: networkConfig, BridgeConfig: bridgeConfig, db: db} router.Peers.OnInvalidateShortIDs(overlay.InvalidateShortIDs) router.Routes.OnChange(overlay.InvalidateRoutes) router.Macs = NewMacCache(macMaxAge, func(mac net.HardwareAddr, peer *mesh.Peer) { log.Debugln("Expired MAC", mac, "at", peer) }) router.Peers.OnGC(func(peer *mesh.Peer) { router.Macs.Delete(peer) }) return router, nil } // Start listening for TCP connections, locally captured packets, and // forwarded packets. func (router *NetworkRouter) Start() { log.Println("Sniffing traffic on", router.InjectorConsumer) checkFatal(router.InjectorConsumer.StartConsumingPackets(router.handleCapturedPacket)) checkFatal(router.Overlay.(NetworkOverlay).StartConsumingPackets(router.Ourself.Peer, router.Peers, router.handleForwardedPacket)) router.Router.Start() } func (router *NetworkRouter) handleCapturedPacket(key PacketKey) FlowOp { router.PacketLogging.LogPacket("Captured", key) srcMac := net.HardwareAddr(key.SrcMAC[:]) dstMac := net.HardwareAddr(key.DstMAC[:]) switch newSrcMac, conflictPeer := router.Macs.Add(srcMac, router.Ourself.Peer); { case newSrcMac: log.Debugln("Discovered local MAC", srcMac) case conflictPeer != nil: // The MAC cache has an entry for the source MAC // associated with another peer. This probably means // we are seeing a frame we injected ourself. That // shouldn't happen, but discard it just in case. log.Error("Captured frame from MAC (", srcMac, ") to (", dstMac, ") associated with another peer ", conflictPeer) return DiscardingFlowOp{} } // Discard STP broadcasts if key.DstMAC == [...]byte{0x01, 0x80, 0xC2, 0x00, 0x00, 0x00} { return DiscardingFlowOp{} } switch dstPeer := router.Macs.Lookup(dstMac); dstPeer { case router.Ourself.Peer: // The packet is destined for a local MAC. The bridge // won't normally send us such packets, and if it does // it's likely to be broadcasting the packet to all // ports. So if it happens, just drop the packet to // avoid warnings if we try to forward it. return DiscardingFlowOp{} case nil: // If we don't know which peer corresponds to the dest // MAC, broadcast it. router.PacketLogging.LogPacket("Broadcasting", key) return router.relayBroadcast(router.Ourself.Peer, key) default: router.PacketLogging.LogPacket("Forwarding", key) return router.relay(ForwardPacketKey{ PacketKey: key, SrcPeer: router.Ourself.Peer, DstPeer: dstPeer}) } } func (router *NetworkRouter) handleForwardedPacket(key ForwardPacketKey) FlowOp { srcMac := net.HardwareAddr(key.SrcMAC[:]) dstMac := net.HardwareAddr(key.DstMAC[:]) if key.SrcPeer == router.Ourself.Peer { log.Warn("Received own packet to peer ", key.DstPeer, " from MAC (", srcMac, ") to (", dstMac, ")") } if key.DstPeer != router.Ourself.Peer { // it's not for us, we're just relaying it router.PacketLogging.LogForwardPacket("Relaying", key) return router.relay(key) } // At this point, it's either unicast to us, or a broadcast // (because the DstPeer on a forwarded broadcast packet is // always set to the peer being forwarded to) switch newSrcMac, conflictPeer := router.Macs.AddForced(srcMac, key.SrcPeer); { case newSrcMac: log.Print("Discovered remote MAC ", srcMac, " at ", key.SrcPeer) case conflictPeer != nil: log.Print("Discovered remote MAC ", srcMac, " at ", key.SrcPeer, " (was at ", conflictPeer, ")") // We need to clear out any flows destined to the MAC // that forward to the old peer. router.Overlay.(NetworkOverlay).InvalidateRoutes() } router.PacketLogging.LogForwardPacket("Injecting", key) injectFop := router.InjectorConsumer.InjectPacket(key.PacketKey) dstPeer := router.Macs.Lookup(dstMac) if dstPeer == router.Ourself.Peer { return injectFop } router.PacketLogging.LogForwardPacket("Relaying broadcast", key) relayFop := router.relayBroadcast(key.SrcPeer, key.PacketKey) switch { case injectFop == nil: return relayFop case relayFop == nil: return injectFop default: mfop := NewMultiFlowOp(false) mfop.Add(injectFop) mfop.Add(relayFop) return mfop } } // Routing func (router *NetworkRouter) relay(key ForwardPacketKey) FlowOp { relayPeerName, found := router.Routes.Unicast(key.DstPeer.Name) if !found { // Not necessarily an error as there could be a race with the // dst disappearing whilst the frame is in flight log.Println("Received packet for unknown destination:", key.DstPeer) return DiscardingFlowOp{} } conn, found := router.Ourself.ConnectionTo(relayPeerName) if !found { // Again, could just be a race, not necessarily an error log.Println("Unable to find connection to relay peer", relayPeerName) return DiscardingFlowOp{} } return conn.(*mesh.LocalConnection).OverlayConn.(OverlayForwarder).Forward(key) } func (router *NetworkRouter) relayBroadcast(srcPeer *mesh.Peer, key PacketKey) FlowOp { nextHops := router.Routes.Broadcast(srcPeer.Name) if len(nextHops) == 0 { return DiscardingFlowOp{} } op := NewMultiFlowOp(true) for _, conn := range router.Ourself.ConnectionsTo(nextHops) { op.Add(conn.(*mesh.LocalConnection).OverlayConn.(OverlayForwarder).Forward(ForwardPacketKey{ PacketKey: key, SrcPeer: srcPeer, DstPeer: conn.Remote()})) } return op } // Persisting the set of peers we are supposed to connect to const peersIdent = "directPeers" func (router *NetworkRouter) persistPeers() { if err := router.db.Save(peersIdent, router.ConnectionMaker.Targets(false)); err != nil { log.Errorf("Error persisting peers: %s", err) return } } func (router *NetworkRouter) InitiateConnections(peers []string, replace bool) []error { errors := router.ConnectionMaker.InitiateConnections(peers, replace) router.persistPeers() return errors } func (router *NetworkRouter) ForgetConnections(peers []string) { router.ConnectionMaker.ForgetConnections(peers) router.persistPeers() } func (router *NetworkRouter) InitialPeers(resume bool, peers []string) ([]string, error) { if _, err := os.Stat("restart.sentinel"); err == nil || resume { var storedPeers []string if _, err := router.db.Load(peersIdent, &storedPeers); err != nil { return nil, err } log.Println("Restart/resume detected - using persisted peer list:", storedPeers) return storedPeers, nil } log.Println("Launch detected - using supplied peer list:", peers) return peers, nil } func (router *NetworkRouter) CreateRestartSentinel() error { sentinel, err := os.Create("restart.sentinel") if err != nil { return fmt.Errorf("error creating sentinel: %v", err) } sentinel.Close() return nil }