local_manager.go 12 KB

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