local_manager.go 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401
  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/flannel-io/flannel/pkg/ip"
  22. . "github.com/flannel-io/flannel/subnet"
  23. "golang.org/x/net/context"
  24. log "k8s.io/klog"
  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. //TODO only vxlan backend and kube subnet manager support dual stack now.
  91. l.EnableIPv4 = true
  92. l.EnableIPv6 = false
  93. return l, nil
  94. case errTryAgain:
  95. continue
  96. default:
  97. return nil, err
  98. }
  99. }
  100. return nil, errors.New("Max retries reached trying to acquire a subnet")
  101. }
  102. func findLeaseByIP(leases []Lease, pubIP ip.IP4) *Lease {
  103. for _, l := range leases {
  104. if pubIP == l.Attrs.PublicIP {
  105. return &l
  106. }
  107. }
  108. return nil
  109. }
  110. func findLeaseBySubnet(leases []Lease, subnet ip.IP4Net) *Lease {
  111. for _, l := range leases {
  112. if subnet.Equal(l.Subnet) {
  113. return &l
  114. }
  115. }
  116. return nil
  117. }
  118. func (m *LocalManager) tryAcquireLease(ctx context.Context, config *Config, extIaddr ip.IP4, attrs *LeaseAttrs) (*Lease, error) {
  119. leases, _, err := m.registry.getSubnets(ctx)
  120. if err != nil {
  121. return nil, err
  122. }
  123. // Try to reuse a subnet if there's one that matches our IP
  124. if l := findLeaseByIP(leases, extIaddr); l != nil {
  125. // Make sure the existing subnet is still within the configured network
  126. if isSubnetConfigCompat(config, l.Subnet) {
  127. log.Infof("Found lease (%v) for current IP (%v), reusing", l.Subnet, extIaddr)
  128. ttl := time.Duration(0)
  129. if !l.Expiration.IsZero() {
  130. // Not a reservation
  131. ttl = subnetTTL
  132. }
  133. exp, err := m.registry.updateSubnet(ctx, l.Subnet, attrs, ttl, 0)
  134. if err != nil {
  135. return nil, err
  136. }
  137. l.Attrs = *attrs
  138. l.Expiration = exp
  139. return l, nil
  140. } else {
  141. log.Infof("Found lease (%v) for current IP (%v) but not compatible with current config, deleting", l.Subnet, extIaddr)
  142. if err := m.registry.deleteSubnet(ctx, l.Subnet); err != nil {
  143. return nil, err
  144. }
  145. }
  146. }
  147. // no existing match, check if there was a previous subnet to use
  148. var sn ip.IP4Net
  149. if !m.previousSubnet.Empty() {
  150. // use previous subnet
  151. if l := findLeaseBySubnet(leases, m.previousSubnet); l != nil {
  152. // Make sure the existing subnet is still within the configured network
  153. if isSubnetConfigCompat(config, l.Subnet) {
  154. log.Infof("Found lease (%v) matching previously leased subnet, reusing", l.Subnet)
  155. ttl := time.Duration(0)
  156. if !l.Expiration.IsZero() {
  157. // Not a reservation
  158. ttl = subnetTTL
  159. }
  160. exp, err := m.registry.updateSubnet(ctx, l.Subnet, attrs, ttl, 0)
  161. if err != nil {
  162. return nil, err
  163. }
  164. l.Attrs = *attrs
  165. l.Expiration = exp
  166. return l, nil
  167. } else {
  168. log.Infof("Found lease (%v) matching previously leased subnet but not compatible with current config, deleting", l.Subnet)
  169. if err := m.registry.deleteSubnet(ctx, l.Subnet); err != nil {
  170. return nil, err
  171. }
  172. }
  173. } else {
  174. // Check if the previous subnet is a part of the network and of the right subnet length
  175. if isSubnetConfigCompat(config, m.previousSubnet) {
  176. log.Infof("Found previously leased subnet (%v), reusing", m.previousSubnet)
  177. sn = m.previousSubnet
  178. } else {
  179. log.Errorf("Found previously leased subnet (%v) that is not compatible with the Etcd network config, ignoring", m.previousSubnet)
  180. }
  181. }
  182. }
  183. if sn.Empty() {
  184. // no existing match, grab a new one
  185. sn, err = m.allocateSubnet(config, leases)
  186. if err != nil {
  187. return nil, err
  188. }
  189. }
  190. exp, err := m.registry.createSubnet(ctx, sn, attrs, subnetTTL)
  191. switch {
  192. case err == nil:
  193. log.Infof("Allocated lease (%v) to current node (%v) ", sn, extIaddr)
  194. return &Lease{
  195. Subnet: sn,
  196. Attrs: *attrs,
  197. Expiration: exp,
  198. }, nil
  199. case isErrEtcdNodeExist(err):
  200. return nil, errTryAgain
  201. default:
  202. return nil, err
  203. }
  204. }
  205. func (m *LocalManager) allocateSubnet(config *Config, leases []Lease) (ip.IP4Net, error) {
  206. log.Infof("Picking subnet in range %s ... %s", config.SubnetMin, config.SubnetMax)
  207. var bag []ip.IP4
  208. sn := ip.IP4Net{IP: config.SubnetMin, PrefixLen: config.SubnetLen}
  209. OuterLoop:
  210. for ; sn.IP <= config.SubnetMax && len(bag) < 100; sn = sn.Next() {
  211. for _, l := range leases {
  212. if sn.Overlaps(l.Subnet) {
  213. continue OuterLoop
  214. }
  215. }
  216. bag = append(bag, sn.IP)
  217. }
  218. if len(bag) == 0 {
  219. return ip.IP4Net{}, errors.New("out of subnets")
  220. } else {
  221. i := randInt(0, len(bag))
  222. return ip.IP4Net{IP: bag[i], PrefixLen: config.SubnetLen}, nil
  223. }
  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. //TODO only vxlan backend and kube subnet manager support dual stack now.
  254. l.EnableIPv4 = true
  255. l.EnableIPv6 = false
  256. return LeaseWatchResult{
  257. Snapshot: []Lease{*l},
  258. Cursor: watchCursor{index},
  259. }, nil
  260. }
  261. func (m *LocalManager) WatchLease(ctx context.Context, sn ip.IP4Net, cursor interface{}) (LeaseWatchResult, error) {
  262. if cursor == nil {
  263. return m.leaseWatchReset(ctx, sn)
  264. }
  265. nextIndex, err := getNextIndex(cursor)
  266. if err != nil {
  267. return LeaseWatchResult{}, err
  268. }
  269. evt, index, err := m.registry.watchSubnet(ctx, nextIndex, sn)
  270. switch {
  271. case err == nil:
  272. //TODO only vxlan backend and kube subnet manager support dual stack now.
  273. evt.Lease.EnableIPv4 = true
  274. evt.Lease.EnableIPv6 = false
  275. return LeaseWatchResult{
  276. Events: []Event{evt},
  277. Cursor: watchCursor{index},
  278. }, nil
  279. case isIndexTooSmall(err):
  280. log.Warning("Watch of subnet leases failed because etcd index outside history window")
  281. return m.leaseWatchReset(ctx, sn)
  282. default:
  283. return LeaseWatchResult{}, err
  284. }
  285. }
  286. func (m *LocalManager) WatchLeases(ctx context.Context, cursor interface{}) (LeaseWatchResult, error) {
  287. if cursor == nil {
  288. return m.leasesWatchReset(ctx)
  289. }
  290. nextIndex, err := getNextIndex(cursor)
  291. if err != nil {
  292. return LeaseWatchResult{}, err
  293. }
  294. evt, index, err := m.registry.watchSubnets(ctx, nextIndex)
  295. switch {
  296. case err == nil:
  297. //TODO only vxlan backend and kube subnet manager support dual stack now.
  298. evt.Lease.EnableIPv4 = true
  299. evt.Lease.EnableIPv6 = false
  300. return LeaseWatchResult{
  301. Events: []Event{evt},
  302. Cursor: watchCursor{index},
  303. }, nil
  304. case isIndexTooSmall(err):
  305. log.Warning("Watch of subnet leases failed because etcd index outside history window")
  306. return m.leasesWatchReset(ctx)
  307. case index != 0:
  308. return LeaseWatchResult{Cursor: watchCursor{index}}, err
  309. default:
  310. return LeaseWatchResult{}, err
  311. }
  312. }
  313. func isIndexTooSmall(err error) bool {
  314. etcdErr, ok := err.(etcd.Error)
  315. return ok && etcdErr.Code == etcd.ErrorCodeEventIndexCleared
  316. }
  317. // leasesWatchReset is called when incremental lease watch failed and we need to grab a snapshot
  318. func (m *LocalManager) leasesWatchReset(ctx context.Context) (LeaseWatchResult, error) {
  319. wr := LeaseWatchResult{}
  320. leases, index, err := m.registry.getSubnets(ctx)
  321. if err != nil {
  322. return wr, fmt.Errorf("failed to retrieve subnet leases: %v", err)
  323. }
  324. wr.Cursor = watchCursor{index}
  325. wr.Snapshot = leases
  326. return wr, nil
  327. }
  328. func isSubnetConfigCompat(config *Config, sn ip.IP4Net) bool {
  329. if sn.IP < config.SubnetMin || sn.IP > config.SubnetMax {
  330. return false
  331. }
  332. return sn.PrefixLen == config.SubnetLen
  333. }
  334. func (m *LocalManager) Name() string {
  335. previousSubnet := m.previousSubnet.String()
  336. if m.previousSubnet.Empty() {
  337. previousSubnet = "None"
  338. }
  339. return fmt.Sprintf("Etcd Local Manager with Previous Subnet: %s", previousSubnet)
  340. }