registry.go 5.1 KB

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