etcd.go 9.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389
  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 subnet
  15. import (
  16. "encoding/json"
  17. "errors"
  18. "fmt"
  19. "net"
  20. "regexp"
  21. "strconv"
  22. "time"
  23. etcd "github.com/coreos/flannel/Godeps/_workspace/src/github.com/coreos/etcd/client"
  24. log "github.com/coreos/flannel/Godeps/_workspace/src/github.com/golang/glog"
  25. "github.com/coreos/flannel/Godeps/_workspace/src/golang.org/x/net/context"
  26. "github.com/coreos/flannel/pkg/ip"
  27. )
  28. const (
  29. registerRetries = 10
  30. subnetTTL = 24 * time.Hour
  31. )
  32. type EtcdManager struct {
  33. registry Registry
  34. }
  35. var (
  36. subnetRegex *regexp.Regexp = regexp.MustCompile(`(\d+\.\d+.\d+.\d+)-(\d+)`)
  37. )
  38. type watchCursor struct {
  39. index uint64
  40. }
  41. func (c watchCursor) String() string {
  42. return strconv.FormatUint(c.index, 10)
  43. }
  44. func NewEtcdManager(config *EtcdConfig) (Manager, error) {
  45. r, err := newEtcdSubnetRegistry(config)
  46. if err != nil {
  47. return nil, err
  48. }
  49. return &EtcdManager{r}, nil
  50. }
  51. func newEtcdManager(r Registry) Manager {
  52. return &EtcdManager{r}
  53. }
  54. func (m *EtcdManager) GetNetworkConfig(ctx context.Context, network string) (*Config, error) {
  55. cfgResp, err := m.registry.getConfig(ctx, network)
  56. if err != nil {
  57. return nil, err
  58. }
  59. return ParseConfig(cfgResp.Node.Value)
  60. }
  61. func (m *EtcdManager) AcquireLease(ctx context.Context, network string, attrs *LeaseAttrs) (*Lease, error) {
  62. config, err := m.GetNetworkConfig(ctx, network)
  63. if err != nil {
  64. return nil, err
  65. }
  66. for {
  67. l, err := m.acquireLeaseOnce(ctx, network, config, attrs)
  68. switch {
  69. case err == nil:
  70. log.Info("Subnet lease acquired: ", l.Subnet)
  71. return l, nil
  72. case err == context.Canceled, err == context.DeadlineExceeded:
  73. return nil, err
  74. default:
  75. log.Error("Failed to acquire subnet: ", err)
  76. }
  77. select {
  78. case <-time.After(time.Second):
  79. case <-ctx.Done():
  80. return nil, ctx.Err()
  81. }
  82. }
  83. }
  84. func findLeaseByIP(leases []Lease, pubIP ip.IP4) *Lease {
  85. for _, l := range leases {
  86. if pubIP == l.Attrs.PublicIP {
  87. return &l
  88. }
  89. }
  90. return nil
  91. }
  92. func (m *EtcdManager) tryAcquireLease(ctx context.Context, network string, config *Config, extIaddr ip.IP4, attrs *LeaseAttrs) (*Lease, error) {
  93. var err error
  94. leases, _, err := m.getLeases(ctx, network)
  95. if err != nil {
  96. return nil, err
  97. }
  98. attrBytes, err := json.Marshal(attrs)
  99. if err != nil {
  100. return nil, err
  101. }
  102. // try to reuse a subnet if there's one that matches our IP
  103. if l := findLeaseByIP(leases, extIaddr); l != nil {
  104. // make sure the existing subnet is still within the configured network
  105. if isSubnetConfigCompat(config, l.Subnet) {
  106. log.Infof("Found lease (%v) for current IP (%v), reusing", l.Subnet, extIaddr)
  107. resp, err := m.registry.updateSubnet(ctx, network, l.Key(), string(attrBytes), subnetTTL)
  108. if err != nil {
  109. return nil, err
  110. }
  111. l.Attrs = attrs
  112. l.Expiration = *resp.Node.Expiration
  113. return l, nil
  114. } else {
  115. log.Infof("Found lease (%v) for current IP (%v) but not compatible with current config, deleting", l.Subnet, extIaddr)
  116. if _, err := m.registry.deleteSubnet(ctx, network, l.Key()); err != nil {
  117. return nil, err
  118. }
  119. }
  120. }
  121. // no existing match, grab a new one
  122. sn, err := m.allocateSubnet(config, leases)
  123. if err != nil {
  124. return nil, err
  125. }
  126. resp, err := m.registry.createSubnet(ctx, network, sn.StringSep(".", "-"), string(attrBytes), subnetTTL)
  127. if err == nil {
  128. return &Lease{
  129. Subnet: sn,
  130. Attrs: attrs,
  131. Expiration: *resp.Node.Expiration,
  132. }, nil
  133. }
  134. if etcdErr, ok := err.(etcd.Error); ok && etcdErr.Code == etcd.ErrorCodeNodeExist {
  135. // if etcd returned Key Already Exists, try again.
  136. return nil, nil
  137. }
  138. return nil, err
  139. }
  140. func (m *EtcdManager) acquireLeaseOnce(ctx context.Context, network string, config *Config, attrs *LeaseAttrs) (*Lease, error) {
  141. for i := 0; i < registerRetries; i++ {
  142. l, err := m.tryAcquireLease(ctx, network, config, attrs.PublicIP, attrs)
  143. switch {
  144. case err != nil:
  145. return nil, err
  146. case l != nil:
  147. return l, nil
  148. }
  149. // before moving on, check for cancel
  150. // TODO(eyakubovich): propogate ctx deeper into registry
  151. select {
  152. case <-ctx.Done():
  153. return nil, ctx.Err()
  154. default:
  155. }
  156. }
  157. return nil, errors.New("Max retries reached trying to acquire a subnet")
  158. }
  159. func parseSubnetKey(s string) (ip.IP4Net, error) {
  160. if parts := subnetRegex.FindStringSubmatch(s); len(parts) == 3 {
  161. snIp := net.ParseIP(parts[1]).To4()
  162. prefixLen, err := strconv.ParseUint(parts[2], 10, 5)
  163. if snIp != nil && err == nil {
  164. return ip.IP4Net{IP: ip.FromIP(snIp), PrefixLen: uint(prefixLen)}, nil
  165. }
  166. }
  167. return ip.IP4Net{}, errors.New("Error parsing IP Subnet")
  168. }
  169. func (m *EtcdManager) allocateSubnet(config *Config, leases []Lease) (ip.IP4Net, error) {
  170. log.Infof("Picking subnet in range %s ... %s", config.SubnetMin, config.SubnetMax)
  171. var bag []ip.IP4
  172. sn := ip.IP4Net{IP: config.SubnetMin, PrefixLen: config.SubnetLen}
  173. OuterLoop:
  174. for ; sn.IP <= config.SubnetMax && len(bag) < 100; sn = sn.Next() {
  175. for _, l := range leases {
  176. if sn.Overlaps(l.Subnet) {
  177. continue OuterLoop
  178. }
  179. }
  180. bag = append(bag, sn.IP)
  181. }
  182. if len(bag) == 0 {
  183. return ip.IP4Net{}, errors.New("out of subnets")
  184. } else {
  185. i := randInt(0, len(bag))
  186. return ip.IP4Net{IP: bag[i], PrefixLen: config.SubnetLen}, nil
  187. }
  188. }
  189. // getLeases queries etcd to get a list of currently allocated leases for a given network.
  190. // It returns the leases along with the "as-of" etcd-index that can be used as the starting
  191. // point for etcd watch.
  192. func (m *EtcdManager) getLeases(ctx context.Context, network string) ([]Lease, uint64, error) {
  193. resp, err := m.registry.getSubnets(ctx, network)
  194. leases := []Lease{}
  195. if err == nil {
  196. for _, node := range resp.Node.Nodes {
  197. sn, err := parseSubnetKey(node.Key)
  198. if err == nil {
  199. attrs := &LeaseAttrs{}
  200. if err = json.Unmarshal([]byte(node.Value), attrs); err == nil {
  201. exp := time.Time{}
  202. if node.Expiration != nil {
  203. exp = *node.Expiration
  204. }
  205. lease := Lease{
  206. Subnet: sn,
  207. Attrs: attrs,
  208. Expiration: exp,
  209. }
  210. leases = append(leases, lease)
  211. }
  212. }
  213. }
  214. return leases, resp.Index, nil
  215. }
  216. if etcdErr, ok := err.(etcd.Error); ok && etcdErr.Code == etcd.ErrorCodeKeyNotFound {
  217. // key not found: treat it as empty set
  218. return leases, etcdErr.Index, nil
  219. }
  220. return nil, 0, err
  221. }
  222. func (m *EtcdManager) RenewLease(ctx context.Context, network string, lease *Lease) error {
  223. attrBytes, err := json.Marshal(lease.Attrs)
  224. if err != nil {
  225. return err
  226. }
  227. // TODO(eyakubovich): propogate ctx into registry
  228. resp, err := m.registry.updateSubnet(ctx, network, lease.Key(), string(attrBytes), subnetTTL)
  229. if err != nil {
  230. return err
  231. }
  232. lease.Expiration = *resp.Node.Expiration
  233. return nil
  234. }
  235. func (m *EtcdManager) WatchLeases(ctx context.Context, network string, cursor interface{}) (WatchResult, error) {
  236. if cursor == nil {
  237. return m.watchReset(ctx, network)
  238. }
  239. nextIndex := uint64(0)
  240. if wc, ok := cursor.(watchCursor); ok {
  241. nextIndex = wc.index
  242. } else if s, ok := cursor.(string); ok {
  243. var err error
  244. nextIndex, err = strconv.ParseUint(s, 10, 64)
  245. if err != nil {
  246. return WatchResult{}, fmt.Errorf("failed to parse cursor: %v", err)
  247. }
  248. } else {
  249. return WatchResult{}, fmt.Errorf("internal error: watch cursor is of unknown type")
  250. }
  251. resp, err := m.registry.watchSubnets(ctx, network, nextIndex)
  252. switch {
  253. case err == nil:
  254. return parseSubnetWatchResponse(resp)
  255. case isIndexTooSmall(err):
  256. log.Warning("Watch of subnet leases failed because etcd index outside history window")
  257. return m.watchReset(ctx, network)
  258. default:
  259. return WatchResult{}, err
  260. }
  261. }
  262. func isIndexTooSmall(err error) bool {
  263. etcdErr, ok := err.(etcd.Error)
  264. return ok && etcdErr.Code == etcd.ErrorCodeEventIndexCleared
  265. }
  266. func parseSubnetWatchResponse(resp *etcd.Response) (WatchResult, error) {
  267. sn, err := parseSubnetKey(resp.Node.Key)
  268. if err != nil {
  269. return WatchResult{}, fmt.Errorf("error parsing subnet IP: %s", resp.Node.Key)
  270. }
  271. evt := Event{}
  272. switch resp.Action {
  273. case "delete", "expire":
  274. evt = Event{
  275. SubnetRemoved,
  276. Lease{Subnet: sn},
  277. }
  278. default:
  279. attrs := &LeaseAttrs{}
  280. err := json.Unmarshal([]byte(resp.Node.Value), attrs)
  281. if err != nil {
  282. return WatchResult{}, err
  283. }
  284. exp := time.Time{}
  285. if resp.Node.Expiration != nil {
  286. exp = *resp.Node.Expiration
  287. }
  288. evt = Event{
  289. SubnetAdded,
  290. Lease{
  291. Subnet: sn,
  292. Attrs: attrs,
  293. Expiration: exp,
  294. },
  295. }
  296. }
  297. cursor := watchCursor{resp.Node.ModifiedIndex}
  298. return WatchResult{
  299. Cursor: cursor,
  300. Events: []Event{evt},
  301. }, nil
  302. }
  303. // watchReset is called when incremental watch failed and we need to grab a snapshot
  304. func (m *EtcdManager) watchReset(ctx context.Context, network string) (WatchResult, error) {
  305. wr := WatchResult{}
  306. leases, index, err := m.getLeases(ctx, network)
  307. if err != nil {
  308. return wr, fmt.Errorf("failed to retrieve subnet leases: %v", err)
  309. }
  310. cursor := watchCursor{index}
  311. wr.Snapshot = leases
  312. wr.Cursor = cursor
  313. return wr, nil
  314. }
  315. func isSubnetConfigCompat(config *Config, sn ip.IP4Net) bool {
  316. if sn.IP < config.SubnetMin || sn.IP > config.SubnetMax {
  317. return false
  318. }
  319. return sn.PrefixLen == config.SubnetLen
  320. }