cluster_util.go 8.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259
  1. // Copyright 2015 The etcd 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 etcdserver
  15. import (
  16. "encoding/json"
  17. "fmt"
  18. "io/ioutil"
  19. "net/http"
  20. "sort"
  21. "time"
  22. "github.com/coreos/etcd/etcdserver/membership"
  23. "github.com/coreos/etcd/pkg/httputil"
  24. "github.com/coreos/etcd/pkg/types"
  25. "github.com/coreos/etcd/version"
  26. "github.com/coreos/go-semver/semver"
  27. )
  28. // isMemberBootstrapped tries to check if the given member has been bootstrapped
  29. // in the given cluster.
  30. func isMemberBootstrapped(cl *membership.RaftCluster, member string, rt http.RoundTripper, timeout time.Duration) bool {
  31. rcl, err := getClusterFromRemotePeers(getRemotePeerURLs(cl, member), timeout, false, rt)
  32. if err != nil {
  33. return false
  34. }
  35. id := cl.MemberByName(member).ID
  36. m := rcl.Member(id)
  37. if m == nil {
  38. return false
  39. }
  40. if len(m.ClientURLs) > 0 {
  41. return true
  42. }
  43. return false
  44. }
  45. // GetClusterFromRemotePeers takes a set of URLs representing etcd peers, and
  46. // attempts to construct a Cluster by accessing the members endpoint on one of
  47. // these URLs. The first URL to provide a response is used. If no URLs provide
  48. // a response, or a Cluster cannot be successfully created from a received
  49. // response, an error is returned.
  50. // Each request has a 10-second timeout. Because the upper limit of TTL is 5s,
  51. // 10 second is enough for building connection and finishing request.
  52. func GetClusterFromRemotePeers(urls []string, rt http.RoundTripper) (*membership.RaftCluster, error) {
  53. return getClusterFromRemotePeers(urls, 10*time.Second, true, rt)
  54. }
  55. // If logerr is true, it prints out more error messages.
  56. func getClusterFromRemotePeers(urls []string, timeout time.Duration, logerr bool, rt http.RoundTripper) (*membership.RaftCluster, error) {
  57. cc := &http.Client{
  58. Transport: rt,
  59. Timeout: timeout,
  60. }
  61. for _, u := range urls {
  62. resp, err := cc.Get(u + "/members")
  63. if err != nil {
  64. if logerr {
  65. plog.Warningf("could not get cluster response from %s: %v", u, err)
  66. }
  67. continue
  68. }
  69. b, err := ioutil.ReadAll(resp.Body)
  70. resp.Body.Close()
  71. if err != nil {
  72. if logerr {
  73. plog.Warningf("could not read the body of cluster response: %v", err)
  74. }
  75. continue
  76. }
  77. var membs []*membership.Member
  78. if err = json.Unmarshal(b, &membs); err != nil {
  79. if logerr {
  80. plog.Warningf("could not unmarshal cluster response: %v", err)
  81. }
  82. continue
  83. }
  84. id, err := types.IDFromString(resp.Header.Get("X-Etcd-Cluster-ID"))
  85. if err != nil {
  86. if logerr {
  87. plog.Warningf("could not parse the cluster ID from cluster res: %v", err)
  88. }
  89. continue
  90. }
  91. return membership.NewClusterFromMembers("", id, membs), nil
  92. }
  93. return nil, fmt.Errorf("could not retrieve cluster information from the given urls")
  94. }
  95. // getRemotePeerURLs returns peer urls of remote members in the cluster. The
  96. // returned list is sorted in ascending lexicographical order.
  97. func getRemotePeerURLs(cl *membership.RaftCluster, local string) []string {
  98. us := make([]string, 0)
  99. for _, m := range cl.Members() {
  100. if m.Name == local {
  101. continue
  102. }
  103. us = append(us, m.PeerURLs...)
  104. }
  105. sort.Strings(us)
  106. return us
  107. }
  108. // getVersions returns the versions of the members in the given cluster.
  109. // The key of the returned map is the member's ID. The value of the returned map
  110. // is the semver versions string, including server and cluster.
  111. // If it fails to get the version of a member, the key will be nil.
  112. func getVersions(cl *membership.RaftCluster, local types.ID, rt http.RoundTripper) map[string]*version.Versions {
  113. members := cl.Members()
  114. vers := make(map[string]*version.Versions)
  115. for _, m := range members {
  116. if m.ID == local {
  117. cv := "not_decided"
  118. if cl.Version() != nil {
  119. cv = cl.Version().String()
  120. }
  121. vers[m.ID.String()] = &version.Versions{Server: version.Version, Cluster: cv}
  122. continue
  123. }
  124. ver, err := getVersion(m, rt)
  125. if err != nil {
  126. plog.Warningf("cannot get the version of member %s (%v)", m.ID, err)
  127. vers[m.ID.String()] = nil
  128. } else {
  129. vers[m.ID.String()] = ver
  130. }
  131. }
  132. return vers
  133. }
  134. // decideClusterVersion decides the cluster version based on the versions map.
  135. // The returned version is the min server version in the map, or nil if the min
  136. // version in unknown.
  137. func decideClusterVersion(vers map[string]*version.Versions) *semver.Version {
  138. var cv *semver.Version
  139. lv := semver.Must(semver.NewVersion(version.Version))
  140. for mid, ver := range vers {
  141. if ver == nil {
  142. return nil
  143. }
  144. v, err := semver.NewVersion(ver.Server)
  145. if err != nil {
  146. plog.Errorf("cannot understand the version of member %s (%v)", mid, err)
  147. return nil
  148. }
  149. if lv.LessThan(*v) {
  150. plog.Warningf("the local etcd version %s is not up-to-date", lv.String())
  151. plog.Warningf("member %s has a higher version %s", mid, ver.Server)
  152. }
  153. if cv == nil {
  154. cv = v
  155. } else if v.LessThan(*cv) {
  156. cv = v
  157. }
  158. }
  159. return cv
  160. }
  161. // isCompatibleWithCluster return true if the local member has a compatible version with
  162. // the current running cluster.
  163. // The version is considered as compatible when at least one of the other members in the cluster has a
  164. // cluster version in the range of [MinClusterVersion, Version] and no known members has a cluster version
  165. // out of the range.
  166. // We set this rule since when the local member joins, another member might be offline.
  167. func isCompatibleWithCluster(cl *membership.RaftCluster, local types.ID, rt http.RoundTripper) bool {
  168. vers := getVersions(cl, local, rt)
  169. minV := semver.Must(semver.NewVersion(version.MinClusterVersion))
  170. maxV := semver.Must(semver.NewVersion(version.Version))
  171. maxV = &semver.Version{
  172. Major: maxV.Major,
  173. Minor: maxV.Minor,
  174. }
  175. return isCompatibleWithVers(vers, local, minV, maxV)
  176. }
  177. func isCompatibleWithVers(vers map[string]*version.Versions, local types.ID, minV, maxV *semver.Version) bool {
  178. var ok bool
  179. for id, v := range vers {
  180. // ignore comparison with local version
  181. if id == local.String() {
  182. continue
  183. }
  184. if v == nil {
  185. continue
  186. }
  187. clusterv, err := semver.NewVersion(v.Cluster)
  188. if err != nil {
  189. plog.Errorf("cannot understand the cluster version of member %s (%v)", id, err)
  190. continue
  191. }
  192. if clusterv.LessThan(*minV) {
  193. plog.Warningf("the running cluster version(%v) is lower than the minimal cluster version(%v) supported", clusterv.String(), minV.String())
  194. return false
  195. }
  196. if maxV.LessThan(*clusterv) {
  197. plog.Warningf("the running cluster version(%v) is higher than the maximum cluster version(%v) supported", clusterv.String(), maxV.String())
  198. return false
  199. }
  200. ok = true
  201. }
  202. return ok
  203. }
  204. // getVersion returns the Versions of the given member via its
  205. // peerURLs. Returns the last error if it fails to get the version.
  206. func getVersion(m *membership.Member, rt http.RoundTripper) (*version.Versions, error) {
  207. cc := &http.Client{
  208. Transport: rt,
  209. }
  210. var (
  211. err error
  212. resp *http.Response
  213. )
  214. for _, u := range m.PeerURLs {
  215. resp, err = cc.Get(u + "/version")
  216. if err != nil {
  217. plog.Warningf("failed to reach the peerURL(%s) of member %s (%v)", u, m.ID, err)
  218. continue
  219. }
  220. // etcd 2.0 does not have version endpoint on peer url.
  221. if resp.StatusCode == http.StatusNotFound {
  222. httputil.GracefulClose(resp)
  223. return &version.Versions{
  224. Server: "2.0.0",
  225. Cluster: "2.0.0",
  226. }, nil
  227. }
  228. var b []byte
  229. b, err = ioutil.ReadAll(resp.Body)
  230. resp.Body.Close()
  231. if err != nil {
  232. plog.Warningf("failed to read out the response body from the peerURL(%s) of member %s (%v)", u, m.ID, err)
  233. continue
  234. }
  235. var vers version.Versions
  236. if err = json.Unmarshal(b, &vers); err != nil {
  237. plog.Warningf("failed to unmarshal the response body got from the peerURL(%s) of member %s (%v)", u, m.ID, err)
  238. continue
  239. }
  240. return &vers, nil
  241. }
  242. return nil, err
  243. }