// Copyright 2015 flannel authors // // 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 vxlan import ( "bytes" "encoding/json" "fmt" "net" "sync" "time" log "github.com/golang/glog" "github.com/vishvananda/netlink" "golang.org/x/net/context" "github.com/coreos/flannel/backend" "github.com/coreos/flannel/pkg/ip" "github.com/coreos/flannel/subnet" ) type network struct { backend.SimpleNetwork name string extIface *backend.ExternalInterface dev *vxlanDevice rts routes sm subnet.Manager } func newNetwork(name string, sm subnet.Manager, extIface *backend.ExternalInterface, dev *vxlanDevice, nw ip.IP4Net, l *subnet.Lease) (*network, error) { n := &network{ SimpleNetwork: backend.SimpleNetwork{ SubnetLease: l, ExtIface: extIface, }, name: name, sm: sm, dev: dev, } return n, nil } func (n *network) Run(ctx context.Context) { log.Info("Watching for L3 misses") misses := make(chan *netlink.Neigh, 100) // Unfrtunately MonitorMisses does not take a cancel channel // as there's no wait to interrupt netlink socket recv go n.dev.MonitorMisses(misses) wg := sync.WaitGroup{} log.Info("Watching for new subnet leases") evts := make(chan []subnet.Event) wg.Add(1) go func() { subnet.WatchLeases(ctx, n.sm, n.name, n.SubnetLease, evts) log.Info("WatchLeases exited") wg.Done() }() defer wg.Wait() initialEvtsBatch := <-evts for { err := n.handleInitialSubnetEvents(initialEvtsBatch) if err == nil { break } log.Error(err, " About to retry") time.Sleep(time.Second) } for { select { case miss := <-misses: n.handleMiss(miss) case evtBatch := <-evts: n.handleSubnetEvents(evtBatch) case <-ctx.Done(): return } } } func (n *network) MTU() int { return n.dev.MTU() } type vxlanLeaseAttrs struct { VtepMAC hardwareAddr } func (n *network) handleSubnetEvents(batch []subnet.Event) { for _, evt := range batch { switch evt.Type { case subnet.EventAdded: log.Info("Subnet added: ", evt.Lease.Subnet) if evt.Lease.Attrs.BackendType != "vxlan" { log.Warningf("Ignoring non-vxlan subnet: type=%v", evt.Lease.Attrs.BackendType) continue } var attrs vxlanLeaseAttrs if err := json.Unmarshal(evt.Lease.Attrs.BackendData, &attrs); err != nil { log.Error("Error decoding subnet lease JSON: ", err) continue } n.rts.set(evt.Lease.Subnet, net.HardwareAddr(attrs.VtepMAC)) n.dev.AddL2(neigh{IP: evt.Lease.Attrs.PublicIP, MAC: net.HardwareAddr(attrs.VtepMAC)}) case subnet.EventRemoved: log.Info("Subnet removed: ", evt.Lease.Subnet) if evt.Lease.Attrs.BackendType != "vxlan" { log.Warningf("Ignoring non-vxlan subnet: type=%v", evt.Lease.Attrs.BackendType) continue } var attrs vxlanLeaseAttrs if err := json.Unmarshal(evt.Lease.Attrs.BackendData, &attrs); err != nil { log.Error("Error decoding subnet lease JSON: ", err) continue } if len(attrs.VtepMAC) > 0 { n.dev.DelL2(neigh{IP: evt.Lease.Attrs.PublicIP, MAC: net.HardwareAddr(attrs.VtepMAC)}) } n.rts.remove(evt.Lease.Subnet) default: log.Error("Internal error: unknown event type: ", int(evt.Type)) } } } func (n *network) handleInitialSubnetEvents(batch []subnet.Event) error { log.Infof("Handling initial subnet events") fdbTable, err := n.dev.GetL2List() if err != nil { return fmt.Errorf("error fetching L2 table: %v", err) } for _, fdbEntry := range fdbTable { log.Infof("fdb already populated with: %s %s ", fdbEntry.IP, fdbEntry.HardwareAddr) } evtMarker := make([]bool, len(batch)) leaseAttrsList := make([]vxlanLeaseAttrs, len(batch)) fdbEntryMarker := make([]bool, len(fdbTable)) for i, evt := range batch { if evt.Lease.Attrs.BackendType != "vxlan" { log.Warningf("Ignoring non-vxlan subnet: type=%v", evt.Lease.Attrs.BackendType) evtMarker[i] = true continue } if err := json.Unmarshal(evt.Lease.Attrs.BackendData, &leaseAttrsList[i]); err != nil { log.Error("Error decoding subnet lease JSON: ", err) evtMarker[i] = true continue } for j, fdbEntry := range fdbTable { if evt.Lease.Attrs.PublicIP.ToIP().Equal(fdbEntry.IP) && bytes.Equal([]byte(leaseAttrsList[i].VtepMAC), []byte(fdbEntry.HardwareAddr)) { evtMarker[i] = true fdbEntryMarker[j] = true break } } n.rts.set(evt.Lease.Subnet, net.HardwareAddr(leaseAttrsList[i].VtepMAC)) } for j, marker := range fdbEntryMarker { if !marker && fdbTable[j].IP != nil { err := n.dev.DelL2(neigh{IP: ip.FromIP(fdbTable[j].IP), MAC: fdbTable[j].HardwareAddr}) if err != nil { log.Error("Delete L2 failed: ", err) } } } for i, marker := range evtMarker { if !marker { err := n.dev.AddL2(neigh{IP: batch[i].Lease.Attrs.PublicIP, MAC: net.HardwareAddr(leaseAttrsList[i].VtepMAC)}) if err != nil { log.Error("Add L2 failed: ", err) } } } return nil } func (n *network) handleMiss(miss *netlink.Neigh) { switch { case len(miss.IP) == 0 && len(miss.HardwareAddr) == 0: log.Info("Ignoring nil miss") case len(miss.HardwareAddr) == 0: n.handleL3Miss(miss) default: log.Infof("Ignoring not a miss: %v, %v", miss.HardwareAddr, miss.IP) } } func (n *network) handleL3Miss(miss *netlink.Neigh) { log.Infof("L3 miss: %v", miss.IP) rt := n.rts.findByNetwork(ip.FromIP(miss.IP)) if rt == nil { log.Infof("Route for %v not found", miss.IP) return } if err := n.dev.AddL3(neigh{IP: ip.FromIP(miss.IP), MAC: rt.vtepMAC}); err != nil { log.Errorf("AddL3 failed: %v", err) } else { log.Info("AddL3 succeeded") } }