registry.go 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155
  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. "fmt"
  17. "path"
  18. "sync"
  19. "time"
  20. "github.com/coreos/flannel/Godeps/_workspace/src/github.com/coreos/go-etcd/etcd"
  21. log "github.com/coreos/flannel/Godeps/_workspace/src/github.com/golang/glog"
  22. )
  23. type subnetRegistry interface {
  24. getConfig() (*etcd.Response, error)
  25. getSubnets() (*etcd.Response, error)
  26. createSubnet(sn, data string, ttl uint64) (*etcd.Response, error)
  27. updateSubnet(sn, data string, ttl uint64) (*etcd.Response, error)
  28. watchSubnets(since uint64, stop chan bool) (*etcd.Response, error)
  29. }
  30. type EtcdConfig struct {
  31. Endpoints []string
  32. Keyfile string
  33. Certfile string
  34. CAFile string
  35. Prefix string
  36. }
  37. type etcdSubnetRegistry struct {
  38. mux sync.Mutex
  39. cli *etcd.Client
  40. etcdCfg *EtcdConfig
  41. }
  42. func newEtcdClient(c *EtcdConfig) (*etcd.Client, error) {
  43. if c.Keyfile != "" || c.Certfile != "" || c.CAFile != "" {
  44. return etcd.NewTLSClient(c.Endpoints, c.Certfile, c.Keyfile, c.CAFile)
  45. } else {
  46. return etcd.NewClient(c.Endpoints), nil
  47. }
  48. }
  49. func newEtcdSubnetRegistry(config *EtcdConfig) (subnetRegistry, error) {
  50. r := &etcdSubnetRegistry{
  51. etcdCfg: config,
  52. }
  53. var err error
  54. r.cli, err = newEtcdClient(config)
  55. if err != nil {
  56. return nil, err
  57. }
  58. return r, nil
  59. }
  60. func (esr *etcdSubnetRegistry) getConfig() (*etcd.Response, error) {
  61. key := path.Join(esr.etcdCfg.Prefix, "config")
  62. resp, err := esr.client().Get(key, false, false)
  63. if err != nil {
  64. return nil, err
  65. }
  66. return resp, nil
  67. }
  68. func (esr *etcdSubnetRegistry) getSubnets() (*etcd.Response, error) {
  69. key := path.Join(esr.etcdCfg.Prefix, "subnets")
  70. return esr.client().Get(key, false, true)
  71. }
  72. func (esr *etcdSubnetRegistry) createSubnet(sn, data string, ttl uint64) (*etcd.Response, error) {
  73. key := path.Join(esr.etcdCfg.Prefix, "subnets", sn)
  74. resp, err := esr.client().Create(key, data, ttl)
  75. if err != nil {
  76. return nil, err
  77. }
  78. ensureExpiration(resp, ttl)
  79. return resp, nil
  80. }
  81. func (esr *etcdSubnetRegistry) updateSubnet(sn, data string, ttl uint64) (*etcd.Response, error) {
  82. key := path.Join(esr.etcdCfg.Prefix, "subnets", sn)
  83. resp, err := esr.client().Set(key, data, ttl)
  84. if err != nil {
  85. return nil, err
  86. }
  87. ensureExpiration(resp, ttl)
  88. return resp, nil
  89. }
  90. func (esr *etcdSubnetRegistry) watchSubnets(since uint64, stop chan bool) (*etcd.Response, error) {
  91. for {
  92. key := path.Join(esr.etcdCfg.Prefix, "subnets")
  93. resp, err := esr.client().RawWatch(key, since, true, nil, stop)
  94. if err != nil {
  95. if err == etcd.ErrWatchStoppedByUser {
  96. return nil, nil
  97. } else {
  98. return nil, err
  99. }
  100. }
  101. if len(resp.Body) == 0 {
  102. // etcd timed out, go back but recreate the client as the underlying
  103. // http transport gets hosed (http://code.google.com/p/go/issues/detail?id=8648)
  104. esr.resetClient()
  105. continue
  106. }
  107. return resp.Unmarshal()
  108. }
  109. }
  110. func (esr *etcdSubnetRegistry) client() *etcd.Client {
  111. esr.mux.Lock()
  112. defer esr.mux.Unlock()
  113. return esr.cli
  114. }
  115. func (esr *etcdSubnetRegistry) resetClient() {
  116. esr.mux.Lock()
  117. defer esr.mux.Unlock()
  118. var err error
  119. esr.cli, err = newEtcdClient(esr.etcdCfg)
  120. if err != nil {
  121. panic(fmt.Errorf("resetClient: error recreating etcd client: %v", err))
  122. }
  123. }
  124. func ensureExpiration(resp *etcd.Response, ttl uint64) {
  125. if resp.Node.Expiration == nil {
  126. // should not be but calc it ourselves in this case
  127. log.Info("Expiration field missing on etcd response, calculating locally")
  128. exp := time.Now().Add(time.Duration(ttl) * time.Second)
  129. resp.Node.Expiration = &exp
  130. }
  131. }