udp.go 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236
  1. package udp
  2. import (
  3. "fmt"
  4. "encoding/json"
  5. "net"
  6. "os"
  7. "sync"
  8. "syscall"
  9. "github.com/coreos/flannel/Godeps/_workspace/src/github.com/vishvananda/netlink"
  10. log "github.com/coreos/flannel/Godeps/_workspace/src/github.com/golang/glog"
  11. "github.com/coreos/flannel/backend"
  12. "github.com/coreos/flannel/pkg/ip"
  13. "github.com/coreos/flannel/pkg/task"
  14. "github.com/coreos/flannel/subnet"
  15. )
  16. const (
  17. encapOverhead = 28 // 20 bytes IP hdr + 8 bytes UDP hdr
  18. defaultPort = 8285
  19. )
  20. type UdpBackend struct {
  21. sm *subnet.SubnetManager
  22. rawCfg json.RawMessage
  23. cfg struct {
  24. Port int
  25. }
  26. ctl *os.File
  27. ctl2 *os.File
  28. tun *os.File
  29. conn *net.UDPConn
  30. mtu int
  31. tunNet ip.IP4Net
  32. stop chan bool
  33. wg sync.WaitGroup
  34. }
  35. func New(sm *subnet.SubnetManager, config json.RawMessage) backend.Backend {
  36. be := UdpBackend{
  37. sm: sm,
  38. rawCfg: config,
  39. stop: make(chan bool),
  40. }
  41. be.cfg.Port = defaultPort
  42. return &be
  43. }
  44. func (m *UdpBackend) Init(extIface *net.Interface, extIP net.IP) (*backend.SubnetDef, error) {
  45. // Parse our configuration
  46. if len(m.rawCfg) > 0 {
  47. if err := json.Unmarshal(m.rawCfg, &m.cfg); err != nil {
  48. return nil, fmt.Errorf("error decoding UDP backend config: %v", err)
  49. }
  50. }
  51. // Acquire the lease form subnet manager
  52. attrs := subnet.LeaseAttrs{
  53. PublicIP: ip.FromIP(extIP),
  54. }
  55. sn, err := m.sm.AcquireLease(&attrs, m.stop)
  56. if err != nil {
  57. if err == task.ErrCanceled {
  58. return nil, err
  59. } else {
  60. return nil, fmt.Errorf("failed to acquire lease: %v", err)
  61. }
  62. }
  63. // Tunnel's subnet is that of the whole overlay network (e.g. /16)
  64. // and not that of the individual host (e.g. /24)
  65. m.tunNet = ip.IP4Net{
  66. IP: sn.IP,
  67. PrefixLen: m.sm.GetConfig().Network.PrefixLen,
  68. }
  69. // TUN MTU will be smaller b/c of encap (IP+UDP hdrs)
  70. m.mtu = extIface.MTU - encapOverhead
  71. if err = m.initTun(); err != nil {
  72. return nil, err
  73. }
  74. m.conn, err = net.ListenUDP("udp4", &net.UDPAddr{Port: m.cfg.Port})
  75. if err != nil {
  76. return nil, fmt.Errorf("failed to start listening on UDP socket: %v", err)
  77. }
  78. m.ctl, m.ctl2, err = newCtlSockets()
  79. if err != nil {
  80. return nil, fmt.Errorf("failed to create control socket: %v", err)
  81. }
  82. return &backend.SubnetDef{
  83. Net: sn,
  84. MTU: m.mtu,
  85. }, nil
  86. }
  87. func (m *UdpBackend) Run() {
  88. // one for each goroutine below
  89. m.wg.Add(2)
  90. go func() {
  91. runCProxy(m.tun, m.conn, m.ctl2, m.tunNet.IP, m.mtu)
  92. m.wg.Done()
  93. }()
  94. go func() {
  95. m.sm.LeaseRenewer(m.stop)
  96. m.wg.Done()
  97. }()
  98. m.monitorEvents()
  99. m.wg.Wait()
  100. }
  101. func (m *UdpBackend) Stop() {
  102. if m.ctl != nil {
  103. stopProxy(m.ctl)
  104. }
  105. close(m.stop)
  106. }
  107. func (m *UdpBackend) Name() string {
  108. return "UDP"
  109. }
  110. func newCtlSockets() (*os.File, *os.File, error) {
  111. fds, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_SEQPACKET, 0)
  112. if err != nil {
  113. return nil, nil, err
  114. }
  115. f1 := os.NewFile(uintptr(fds[0]), "ctl")
  116. f2 := os.NewFile(uintptr(fds[1]), "ctl")
  117. return f1, f2, nil
  118. }
  119. func (m *UdpBackend) initTun() error {
  120. var tunName string
  121. var err error
  122. m.tun, tunName, err = ip.OpenTun("flannel%d")
  123. if err != nil {
  124. return fmt.Errorf("Failed to open TUN device: %v", err)
  125. }
  126. err = configureIface(tunName, m.tunNet, m.mtu)
  127. if err != nil {
  128. return err
  129. }
  130. return nil
  131. }
  132. func configureIface(ifname string, ipn ip.IP4Net, mtu int) error {
  133. iface, err := netlink.LinkByName(ifname)
  134. if err != nil {
  135. return fmt.Errorf("failed to lookup interface %v", ifname)
  136. }
  137. err = netlink.AddrAdd(iface, &netlink.Addr{ipn.ToIPNet(), ""})
  138. if err != nil {
  139. return fmt.Errorf("failed to add IP address %v to %v: %v", ipn.String(), ifname, err)
  140. }
  141. err = netlink.LinkSetMTU(iface, mtu)
  142. if err != nil {
  143. return fmt.Errorf("failed to set MTU for %v: %v", ifname, err)
  144. }
  145. err = netlink.LinkSetUp(iface)
  146. if err != nil {
  147. return fmt.Errorf("failed to set interface %v to UP state: %v", ifname, err)
  148. }
  149. // explicitly add a route since there might be a route for a subnet already
  150. // installed by Docker and then it won't get auto added
  151. err = netlink.RouteAdd(&netlink.Route{
  152. LinkIndex: iface.Attrs().Index,
  153. Scope: netlink.SCOPE_UNIVERSE,
  154. Dst: ipn.Network().ToIPNet(),
  155. })
  156. if err != nil && err != syscall.EEXIST {
  157. return fmt.Errorf("Failed to add route (%v -> %v): %v", ipn.Network().String(), ifname, err)
  158. }
  159. return nil
  160. }
  161. func (m *UdpBackend) monitorEvents() {
  162. log.Info("Watching for new subnet leases")
  163. evts := make(chan subnet.EventBatch)
  164. m.wg.Add(1)
  165. go func() {
  166. m.sm.WatchLeases(evts, m.stop)
  167. m.wg.Done()
  168. }()
  169. for {
  170. select {
  171. case evtBatch := <-evts:
  172. m.processSubnetEvents(evtBatch)
  173. case <-m.stop:
  174. return
  175. }
  176. }
  177. }
  178. func (m *UdpBackend) processSubnetEvents(batch subnet.EventBatch) {
  179. for _, evt := range batch {
  180. switch evt.Type {
  181. case subnet.SubnetAdded:
  182. log.Info("Subnet added: ", evt.Lease.Network)
  183. setRoute(m.ctl, evt.Lease.Network, evt.Lease.Attrs.PublicIP, m.cfg.Port)
  184. case subnet.SubnetRemoved:
  185. log.Info("Subnet removed: ", evt.Lease.Network)
  186. removeRoute(m.ctl, evt.Lease.Network)
  187. default:
  188. log.Error("Internal error: unknown event type: ", int(evt.Type))
  189. }
  190. }
  191. }