local_manager.go 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453
  1. // Copyright 2015 flannel authors
  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 etcdv2
  15. import (
  16. "errors"
  17. "fmt"
  18. "strconv"
  19. "time"
  20. etcd "github.com/coreos/etcd/client"
  21. "github.com/coreos/flannel/pkg/ip"
  22. . "github.com/coreos/flannel/subnet"
  23. log "github.com/golang/glog"
  24. "golang.org/x/net/context"
  25. )
  26. const (
  27. raceRetries = 10
  28. subnetTTL = 24 * time.Hour
  29. )
  30. type LocalManager struct {
  31. registry Registry
  32. }
  33. type watchCursor struct {
  34. index uint64
  35. }
  36. func isErrEtcdTestFailed(e error) bool {
  37. if e == nil {
  38. return false
  39. }
  40. etcdErr, ok := e.(etcd.Error)
  41. return ok && etcdErr.Code == etcd.ErrorCodeTestFailed
  42. }
  43. func isErrEtcdNodeExist(e error) bool {
  44. if e == nil {
  45. return false
  46. }
  47. etcdErr, ok := e.(etcd.Error)
  48. return ok || etcdErr.Code == etcd.ErrorCodeNodeExist
  49. }
  50. func isErrEtcdKeyNotFound(e error) bool {
  51. if e == nil {
  52. return false
  53. }
  54. etcdErr, ok := e.(etcd.Error)
  55. return ok || etcdErr.Code == etcd.ErrorCodeKeyNotFound
  56. }
  57. func (c watchCursor) String() string {
  58. return strconv.FormatUint(c.index, 10)
  59. }
  60. func NewLocalManager(config *EtcdConfig) (Manager, error) {
  61. r, err := newEtcdSubnetRegistry(config, nil)
  62. if err != nil {
  63. return nil, err
  64. }
  65. return newLocalManager(r), nil
  66. }
  67. func newLocalManager(r Registry) Manager {
  68. return &LocalManager{
  69. registry: r,
  70. }
  71. }
  72. func (m *LocalManager) GetNetworkConfig(ctx context.Context) (*Config, error) {
  73. cfg, err := m.registry.getNetworkConfig(ctx)
  74. if err != nil {
  75. return nil, err
  76. }
  77. return ParseConfig(cfg)
  78. }
  79. func (m *LocalManager) AcquireLease(ctx context.Context, attrs *LeaseAttrs) (*Lease, error) {
  80. config, err := m.GetNetworkConfig(ctx)
  81. if err != nil {
  82. return nil, err
  83. }
  84. for i := 0; i < raceRetries; i++ {
  85. l, err := m.tryAcquireLease(ctx, config, attrs.PublicIP, attrs)
  86. switch err {
  87. case nil:
  88. return l, nil
  89. case errTryAgain:
  90. continue
  91. default:
  92. return nil, err
  93. }
  94. }
  95. return nil, errors.New("Max retries reached trying to acquire a subnet")
  96. }
  97. func findLeaseByIP(leases []Lease, pubIP ip.IP4) *Lease {
  98. for _, l := range leases {
  99. if pubIP == l.Attrs.PublicIP {
  100. return &l
  101. }
  102. }
  103. return nil
  104. }
  105. func (m *LocalManager) tryAcquireLease(ctx context.Context, config *Config, extIaddr ip.IP4, attrs *LeaseAttrs) (*Lease, error) {
  106. leases, _, err := m.registry.getSubnets(ctx)
  107. if err != nil {
  108. return nil, err
  109. }
  110. // try to reuse a subnet if there's one that matches our IP
  111. if l := findLeaseByIP(leases, extIaddr); l != nil {
  112. // make sure the existing subnet is still within the configured network
  113. if isSubnetConfigCompat(config, l.Subnet) {
  114. log.Infof("Found lease (%v) for current IP (%v), reusing", l.Subnet, extIaddr)
  115. ttl := time.Duration(0)
  116. if !l.Expiration.IsZero() {
  117. // Not a reservation
  118. ttl = subnetTTL
  119. }
  120. exp, err := m.registry.updateSubnet(ctx, l.Subnet, attrs, ttl, 0)
  121. if err != nil {
  122. return nil, err
  123. }
  124. l.Attrs = *attrs
  125. l.Expiration = exp
  126. return l, nil
  127. } else {
  128. log.Infof("Found lease (%v) for current IP (%v) but not compatible with current config, deleting", l.Subnet, extIaddr)
  129. if err := m.registry.deleteSubnet(ctx, l.Subnet); err != nil {
  130. return nil, err
  131. }
  132. }
  133. }
  134. // no existing match, grab a new one
  135. sn, err := m.allocateSubnet(config, leases)
  136. if err != nil {
  137. return nil, err
  138. }
  139. exp, err := m.registry.createSubnet(ctx, sn, attrs, subnetTTL)
  140. switch {
  141. case err == nil:
  142. log.Infof("Allocated lease (%v) to current node (%v) ", sn, extIaddr)
  143. return &Lease{
  144. Subnet: sn,
  145. Attrs: *attrs,
  146. Expiration: exp,
  147. }, nil
  148. case isErrEtcdNodeExist(err):
  149. return nil, errTryAgain
  150. default:
  151. return nil, err
  152. }
  153. }
  154. func (m *LocalManager) allocateSubnet(config *Config, leases []Lease) (ip.IP4Net, error) {
  155. log.Infof("Picking subnet in range %s ... %s", config.SubnetMin, config.SubnetMax)
  156. var bag []ip.IP4
  157. sn := ip.IP4Net{IP: config.SubnetMin, PrefixLen: config.SubnetLen}
  158. OuterLoop:
  159. for ; sn.IP <= config.SubnetMax && len(bag) < 100; sn = sn.Next() {
  160. for _, l := range leases {
  161. if sn.Overlaps(l.Subnet) {
  162. continue OuterLoop
  163. }
  164. }
  165. bag = append(bag, sn.IP)
  166. }
  167. if len(bag) == 0 {
  168. return ip.IP4Net{}, errors.New("out of subnets")
  169. } else {
  170. i := randInt(0, len(bag))
  171. return ip.IP4Net{IP: bag[i], PrefixLen: config.SubnetLen}, nil
  172. }
  173. }
  174. func (m *LocalManager) RevokeLease(ctx context.Context, sn ip.IP4Net) error {
  175. return m.registry.deleteSubnet(ctx, sn)
  176. }
  177. func (m *LocalManager) RenewLease(ctx context.Context, lease *Lease) error {
  178. exp, err := m.registry.updateSubnet(ctx, lease.Subnet, &lease.Attrs, subnetTTL, 0)
  179. if err != nil {
  180. return err
  181. }
  182. lease.Expiration = exp
  183. return nil
  184. }
  185. func getNextIndex(cursor interface{}) (uint64, error) {
  186. nextIndex := uint64(0)
  187. if wc, ok := cursor.(watchCursor); ok {
  188. nextIndex = wc.index
  189. } else if s, ok := cursor.(string); ok {
  190. var err error
  191. nextIndex, err = strconv.ParseUint(s, 10, 64)
  192. if err != nil {
  193. return 0, fmt.Errorf("failed to parse cursor: %v", err)
  194. }
  195. } else {
  196. return 0, fmt.Errorf("internal error: watch cursor is of unknown type")
  197. }
  198. return nextIndex, nil
  199. }
  200. func (m *LocalManager) leaseWatchReset(ctx context.Context, sn ip.IP4Net) (LeaseWatchResult, error) {
  201. l, index, err := m.registry.getSubnet(ctx, sn)
  202. if err != nil {
  203. return LeaseWatchResult{}, err
  204. }
  205. return LeaseWatchResult{
  206. Snapshot: []Lease{*l},
  207. Cursor: watchCursor{index},
  208. }, nil
  209. }
  210. func (m *LocalManager) WatchLease(ctx context.Context, sn ip.IP4Net, cursor interface{}) (LeaseWatchResult, error) {
  211. if cursor == nil {
  212. return m.leaseWatchReset(ctx, sn)
  213. }
  214. nextIndex, err := getNextIndex(cursor)
  215. if err != nil {
  216. return LeaseWatchResult{}, err
  217. }
  218. evt, index, err := m.registry.watchSubnet(ctx, nextIndex, sn)
  219. switch {
  220. case err == nil:
  221. return LeaseWatchResult{
  222. Events: []Event{evt},
  223. Cursor: watchCursor{index},
  224. }, nil
  225. case isIndexTooSmall(err):
  226. log.Warning("Watch of subnet leases failed because etcd index outside history window")
  227. return m.leaseWatchReset(ctx, sn)
  228. default:
  229. return LeaseWatchResult{}, err
  230. }
  231. }
  232. func (m *LocalManager) WatchLeases(ctx context.Context, cursor interface{}) (LeaseWatchResult, error) {
  233. if cursor == nil {
  234. return m.leasesWatchReset(ctx)
  235. }
  236. nextIndex, err := getNextIndex(cursor)
  237. if err != nil {
  238. return LeaseWatchResult{}, err
  239. }
  240. evt, index, err := m.registry.watchSubnets(ctx, nextIndex)
  241. switch {
  242. case err == nil:
  243. return LeaseWatchResult{
  244. Events: []Event{evt},
  245. Cursor: watchCursor{index},
  246. }, nil
  247. case isIndexTooSmall(err):
  248. log.Warning("Watch of subnet leases failed because etcd index outside history window")
  249. return m.leasesWatchReset(ctx)
  250. default:
  251. return LeaseWatchResult{}, err
  252. }
  253. }
  254. func isIndexTooSmall(err error) bool {
  255. etcdErr, ok := err.(etcd.Error)
  256. return ok && etcdErr.Code == etcd.ErrorCodeEventIndexCleared
  257. }
  258. // leasesWatchReset is called when incremental lease watch failed and we need to grab a snapshot
  259. func (m *LocalManager) leasesWatchReset(ctx context.Context) (LeaseWatchResult, error) {
  260. wr := LeaseWatchResult{}
  261. leases, index, err := m.registry.getSubnets(ctx)
  262. if err != nil {
  263. return wr, fmt.Errorf("failed to retrieve subnet leases: %v", err)
  264. }
  265. wr.Cursor = watchCursor{index}
  266. wr.Snapshot = leases
  267. return wr, nil
  268. }
  269. func isSubnetConfigCompat(config *Config, sn ip.IP4Net) bool {
  270. if sn.IP < config.SubnetMin || sn.IP > config.SubnetMax {
  271. return false
  272. }
  273. return sn.PrefixLen == config.SubnetLen
  274. }
  275. func (m *LocalManager) tryAddReservation(ctx context.Context, r *Reservation) error {
  276. attrs := &LeaseAttrs{
  277. PublicIP: r.PublicIP,
  278. }
  279. _, err := m.registry.createSubnet(ctx, r.Subnet, attrs, 0)
  280. switch {
  281. case err == nil:
  282. return nil
  283. case !isErrEtcdNodeExist(err):
  284. return err
  285. }
  286. // This subnet or its reservation already exists.
  287. // Get what's there and
  288. // - if PublicIP matches, remove the TTL make it a reservation
  289. // - otherwise, error out
  290. sub, asof, err := m.registry.getSubnet(ctx, r.Subnet)
  291. switch {
  292. case err == nil:
  293. case isErrEtcdKeyNotFound(err):
  294. // Subnet just got expired or was deleted
  295. return errTryAgain
  296. default:
  297. return err
  298. }
  299. if sub.Attrs.PublicIP != r.PublicIP {
  300. // Subnet already taken
  301. return ErrLeaseTaken
  302. }
  303. // remove TTL
  304. _, err = m.registry.updateSubnet(ctx, r.Subnet, &sub.Attrs, 0, asof)
  305. if isErrEtcdTestFailed(err) {
  306. return errTryAgain
  307. }
  308. return err
  309. }
  310. func (m *LocalManager) AddReservation(ctx context.Context, r *Reservation) error {
  311. config, err := m.GetNetworkConfig(ctx)
  312. if err != nil {
  313. return err
  314. }
  315. if config.SubnetLen != r.Subnet.PrefixLen {
  316. return fmt.Errorf("reservation subnet has mask incompatible with network config")
  317. }
  318. if !config.Network.Overlaps(r.Subnet) {
  319. return fmt.Errorf("reservation subnet is outside of flannel network")
  320. }
  321. for i := 0; i < raceRetries; i++ {
  322. err := m.tryAddReservation(ctx, r)
  323. switch {
  324. case err == nil:
  325. return nil
  326. case err == errTryAgain:
  327. continue
  328. default:
  329. return err
  330. }
  331. }
  332. return ErrNoMoreTries
  333. }
  334. func (m *LocalManager) tryRemoveReservation(ctx context.Context, subnet ip.IP4Net) error {
  335. sub, asof, err := m.registry.getSubnet(ctx, subnet)
  336. if err != nil {
  337. return err
  338. }
  339. // add back the TTL
  340. _, err = m.registry.updateSubnet(ctx, subnet, &sub.Attrs, subnetTTL, asof)
  341. if isErrEtcdTestFailed(err) {
  342. return errTryAgain
  343. }
  344. return err
  345. }
  346. //RemoveReservation removes the subnet by setting TTL back to subnetTTL (24hours)
  347. func (m *LocalManager) RemoveReservation(ctx context.Context, subnet ip.IP4Net) error {
  348. for i := 0; i < raceRetries; i++ {
  349. err := m.tryRemoveReservation(ctx, subnet)
  350. switch {
  351. case err == nil:
  352. return nil
  353. case err == errTryAgain:
  354. continue
  355. default:
  356. return err
  357. }
  358. }
  359. return ErrNoMoreTries
  360. }
  361. func (m *LocalManager) ListReservations(ctx context.Context) ([]Reservation, error) {
  362. subnets, _, err := m.registry.getSubnets(ctx)
  363. if err != nil {
  364. return nil, err
  365. }
  366. rsvs := []Reservation{}
  367. for _, sub := range subnets {
  368. // Reservations don't have TTL and so no expiration
  369. if !sub.Expiration.IsZero() {
  370. continue
  371. }
  372. r := Reservation{
  373. Subnet: sub.Subnet,
  374. PublicIP: sub.Attrs.PublicIP,
  375. }
  376. rsvs = append(rsvs, r)
  377. }
  378. return rsvs, nil
  379. }