udp.go 5.5 KB

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