discovery_test.go 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558
  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 discovery
  15. import (
  16. "context"
  17. "errors"
  18. "math"
  19. "math/rand"
  20. "net/http"
  21. "reflect"
  22. "sort"
  23. "strconv"
  24. "testing"
  25. "time"
  26. "github.com/coreos/etcd/client"
  27. "github.com/jonboulle/clockwork"
  28. )
  29. const (
  30. maxRetryInTest = 3
  31. )
  32. func TestNewProxyFuncUnset(t *testing.T) {
  33. pf, err := newProxyFunc("")
  34. if pf != nil {
  35. t.Fatal("unexpected non-nil proxyFunc")
  36. }
  37. if err != nil {
  38. t.Fatalf("unexpected non-nil err: %v", err)
  39. }
  40. }
  41. func TestNewProxyFuncBad(t *testing.T) {
  42. tests := []string{
  43. "%%",
  44. "http://foo.com/%1",
  45. }
  46. for i, in := range tests {
  47. pf, err := newProxyFunc(in)
  48. if pf != nil {
  49. t.Errorf("#%d: unexpected non-nil proxyFunc", i)
  50. }
  51. if err == nil {
  52. t.Errorf("#%d: unexpected nil err", i)
  53. }
  54. }
  55. }
  56. func TestNewProxyFunc(t *testing.T) {
  57. tests := map[string]string{
  58. "bar.com": "http://bar.com",
  59. "http://disco.foo.bar": "http://disco.foo.bar",
  60. }
  61. for in, w := range tests {
  62. pf, err := newProxyFunc(in)
  63. if pf == nil {
  64. t.Errorf("%s: unexpected nil proxyFunc", in)
  65. continue
  66. }
  67. if err != nil {
  68. t.Errorf("%s: unexpected non-nil err: %v", in, err)
  69. continue
  70. }
  71. g, err := pf(&http.Request{})
  72. if err != nil {
  73. t.Errorf("%s: unexpected non-nil err: %v", in, err)
  74. }
  75. if g.String() != w {
  76. t.Errorf("%s: proxyURL=%q, want %q", in, g, w)
  77. }
  78. }
  79. }
  80. func TestCheckCluster(t *testing.T) {
  81. cluster := "/prefix/1000"
  82. self := "/1000/1"
  83. tests := []struct {
  84. nodes []*client.Node
  85. index uint64
  86. werr error
  87. wsize int
  88. }{
  89. {
  90. // self is in the size range
  91. []*client.Node{
  92. {Key: "/1000/_config/size", Value: "3", CreatedIndex: 1},
  93. {Key: "/1000/_config/"},
  94. {Key: self, CreatedIndex: 2},
  95. {Key: "/1000/2", CreatedIndex: 3},
  96. {Key: "/1000/3", CreatedIndex: 4},
  97. {Key: "/1000/4", CreatedIndex: 5},
  98. },
  99. 5,
  100. nil,
  101. 3,
  102. },
  103. {
  104. // self is in the size range
  105. []*client.Node{
  106. {Key: "/1000/_config/size", Value: "3", CreatedIndex: 1},
  107. {Key: "/1000/_config/"},
  108. {Key: "/1000/2", CreatedIndex: 2},
  109. {Key: "/1000/3", CreatedIndex: 3},
  110. {Key: self, CreatedIndex: 4},
  111. {Key: "/1000/4", CreatedIndex: 5},
  112. },
  113. 5,
  114. nil,
  115. 3,
  116. },
  117. {
  118. // self is out of the size range
  119. []*client.Node{
  120. {Key: "/1000/_config/size", Value: "3", CreatedIndex: 1},
  121. {Key: "/1000/_config/"},
  122. {Key: "/1000/2", CreatedIndex: 2},
  123. {Key: "/1000/3", CreatedIndex: 3},
  124. {Key: "/1000/4", CreatedIndex: 4},
  125. {Key: self, CreatedIndex: 5},
  126. },
  127. 5,
  128. ErrFullCluster,
  129. 3,
  130. },
  131. {
  132. // self is not in the cluster
  133. []*client.Node{
  134. {Key: "/1000/_config/size", Value: "3", CreatedIndex: 1},
  135. {Key: "/1000/_config/"},
  136. {Key: "/1000/2", CreatedIndex: 2},
  137. {Key: "/1000/3", CreatedIndex: 3},
  138. },
  139. 3,
  140. nil,
  141. 3,
  142. },
  143. {
  144. []*client.Node{
  145. {Key: "/1000/_config/size", Value: "3", CreatedIndex: 1},
  146. {Key: "/1000/_config/"},
  147. {Key: "/1000/2", CreatedIndex: 2},
  148. {Key: "/1000/3", CreatedIndex: 3},
  149. {Key: "/1000/4", CreatedIndex: 4},
  150. },
  151. 3,
  152. ErrFullCluster,
  153. 3,
  154. },
  155. {
  156. // bad size key
  157. []*client.Node{
  158. {Key: "/1000/_config/size", Value: "bad", CreatedIndex: 1},
  159. },
  160. 0,
  161. ErrBadSizeKey,
  162. 0,
  163. },
  164. {
  165. // no size key
  166. []*client.Node{},
  167. 0,
  168. ErrSizeNotFound,
  169. 0,
  170. },
  171. }
  172. for i, tt := range tests {
  173. var rs []*client.Response
  174. if len(tt.nodes) > 0 {
  175. rs = append(rs, &client.Response{Node: tt.nodes[0], Index: tt.index})
  176. rs = append(rs, &client.Response{
  177. Node: &client.Node{
  178. Key: cluster,
  179. Nodes: tt.nodes[1:],
  180. },
  181. Index: tt.index,
  182. })
  183. }
  184. c := &clientWithResp{rs: rs}
  185. dBase := discovery{cluster: cluster, id: 1, c: c}
  186. cRetry := &clientWithRetry{failTimes: 3}
  187. cRetry.rs = rs
  188. fc := clockwork.NewFakeClock()
  189. dRetry := discovery{cluster: cluster, id: 1, c: cRetry, clock: fc}
  190. for _, d := range []discovery{dBase, dRetry} {
  191. go func() {
  192. for i := uint(1); i <= maxRetryInTest; i++ {
  193. fc.BlockUntil(1)
  194. fc.Advance(time.Second * (0x1 << i))
  195. }
  196. }()
  197. ns, size, index, err := d.checkCluster()
  198. if err != tt.werr {
  199. t.Errorf("#%d: err = %v, want %v", i, err, tt.werr)
  200. }
  201. if reflect.DeepEqual(ns, tt.nodes) {
  202. t.Errorf("#%d: nodes = %v, want %v", i, ns, tt.nodes)
  203. }
  204. if size != tt.wsize {
  205. t.Errorf("#%d: size = %v, want %d", i, size, tt.wsize)
  206. }
  207. if index != tt.index {
  208. t.Errorf("#%d: index = %v, want %d", i, index, tt.index)
  209. }
  210. }
  211. }
  212. }
  213. func TestWaitNodes(t *testing.T) {
  214. all := []*client.Node{
  215. 0: {Key: "/1000/1", CreatedIndex: 2},
  216. 1: {Key: "/1000/2", CreatedIndex: 3},
  217. 2: {Key: "/1000/3", CreatedIndex: 4},
  218. }
  219. tests := []struct {
  220. nodes []*client.Node
  221. rs []*client.Response
  222. }{
  223. {
  224. all,
  225. []*client.Response{},
  226. },
  227. {
  228. all[:1],
  229. []*client.Response{
  230. {Node: &client.Node{Key: "/1000/2", CreatedIndex: 3}},
  231. {Node: &client.Node{Key: "/1000/3", CreatedIndex: 4}},
  232. },
  233. },
  234. {
  235. all[:2],
  236. []*client.Response{
  237. {Node: &client.Node{Key: "/1000/3", CreatedIndex: 4}},
  238. },
  239. },
  240. {
  241. append(all, &client.Node{Key: "/1000/4", CreatedIndex: 5}),
  242. []*client.Response{
  243. {Node: &client.Node{Key: "/1000/3", CreatedIndex: 4}},
  244. },
  245. },
  246. }
  247. for i, tt := range tests {
  248. // Basic case
  249. c := &clientWithResp{rs: nil, w: &watcherWithResp{rs: tt.rs}}
  250. dBase := &discovery{cluster: "1000", c: c}
  251. // Retry case
  252. var retryScanResp []*client.Response
  253. if len(tt.nodes) > 0 {
  254. retryScanResp = append(retryScanResp, &client.Response{
  255. Node: &client.Node{
  256. Key: "1000",
  257. Value: strconv.Itoa(3),
  258. },
  259. })
  260. retryScanResp = append(retryScanResp, &client.Response{
  261. Node: &client.Node{
  262. Nodes: tt.nodes,
  263. },
  264. })
  265. }
  266. cRetry := &clientWithResp{
  267. rs: retryScanResp,
  268. w: &watcherWithRetry{rs: tt.rs, failTimes: 2},
  269. }
  270. fc := clockwork.NewFakeClock()
  271. dRetry := &discovery{
  272. cluster: "1000",
  273. c: cRetry,
  274. clock: fc,
  275. }
  276. for _, d := range []*discovery{dBase, dRetry} {
  277. go func() {
  278. for i := uint(1); i <= maxRetryInTest; i++ {
  279. fc.BlockUntil(1)
  280. fc.Advance(time.Second * (0x1 << i))
  281. }
  282. }()
  283. g, err := d.waitNodes(tt.nodes, 3, 0) // we do not care about index in this test
  284. if err != nil {
  285. t.Errorf("#%d: err = %v, want %v", i, err, nil)
  286. }
  287. if !reflect.DeepEqual(g, all) {
  288. t.Errorf("#%d: all = %v, want %v", i, g, all)
  289. }
  290. }
  291. }
  292. }
  293. func TestCreateSelf(t *testing.T) {
  294. rs := []*client.Response{{Node: &client.Node{Key: "1000/1", CreatedIndex: 2}}}
  295. w := &watcherWithResp{rs: rs}
  296. errw := &watcherWithErr{err: errors.New("watch err")}
  297. c := &clientWithResp{rs: rs, w: w}
  298. errc := &clientWithErr{err: errors.New("create err"), w: w}
  299. errdupc := &clientWithErr{err: client.Error{Code: client.ErrorCodeNodeExist}}
  300. errwc := &clientWithResp{rs: rs, w: errw}
  301. tests := []struct {
  302. c client.KeysAPI
  303. werr error
  304. }{
  305. // no error
  306. {c, nil},
  307. // client.create returns an error
  308. {errc, errc.err},
  309. // watcher.next returns an error
  310. {errwc, errw.err},
  311. // parse key exist error to duplicate ID error
  312. {errdupc, ErrDuplicateID},
  313. }
  314. for i, tt := range tests {
  315. d := discovery{cluster: "1000", c: tt.c}
  316. if err := d.createSelf(""); err != tt.werr {
  317. t.Errorf("#%d: err = %v, want %v", i, err, nil)
  318. }
  319. }
  320. }
  321. func TestNodesToCluster(t *testing.T) {
  322. tests := []struct {
  323. nodes []*client.Node
  324. size int
  325. wcluster string
  326. werr error
  327. }{
  328. {
  329. []*client.Node{
  330. 0: {Key: "/1000/1", Value: "1=http://1.1.1.1:2380", CreatedIndex: 1},
  331. 1: {Key: "/1000/2", Value: "2=http://2.2.2.2:2380", CreatedIndex: 2},
  332. 2: {Key: "/1000/3", Value: "3=http://3.3.3.3:2380", CreatedIndex: 3},
  333. },
  334. 3,
  335. "1=http://1.1.1.1:2380,2=http://2.2.2.2:2380,3=http://3.3.3.3:2380",
  336. nil,
  337. },
  338. {
  339. []*client.Node{
  340. 0: {Key: "/1000/1", Value: "1=http://1.1.1.1:2380", CreatedIndex: 1},
  341. 1: {Key: "/1000/2", Value: "2=http://2.2.2.2:2380", CreatedIndex: 2},
  342. 2: {Key: "/1000/3", Value: "2=http://3.3.3.3:2380", CreatedIndex: 3},
  343. },
  344. 3,
  345. "1=http://1.1.1.1:2380,2=http://2.2.2.2:2380,2=http://3.3.3.3:2380",
  346. ErrDuplicateName,
  347. },
  348. {
  349. []*client.Node{
  350. 0: {Key: "/1000/1", Value: "1=1.1.1.1:2380", CreatedIndex: 1},
  351. 1: {Key: "/1000/2", Value: "2=http://2.2.2.2:2380", CreatedIndex: 2},
  352. 2: {Key: "/1000/3", Value: "2=http://3.3.3.3:2380", CreatedIndex: 3},
  353. },
  354. 3,
  355. "1=1.1.1.1:2380,2=http://2.2.2.2:2380,2=http://3.3.3.3:2380",
  356. ErrInvalidURL,
  357. },
  358. }
  359. for i, tt := range tests {
  360. cluster, err := nodesToCluster(tt.nodes, tt.size)
  361. if err != tt.werr {
  362. t.Errorf("#%d: err = %v, want %v", i, err, tt.werr)
  363. }
  364. if !reflect.DeepEqual(cluster, tt.wcluster) {
  365. t.Errorf("#%d: cluster = %v, want %v", i, cluster, tt.wcluster)
  366. }
  367. }
  368. }
  369. func TestSortableNodes(t *testing.T) {
  370. ns := []*client.Node{
  371. 0: {CreatedIndex: 5},
  372. 1: {CreatedIndex: 1},
  373. 2: {CreatedIndex: 3},
  374. 3: {CreatedIndex: 4},
  375. }
  376. // add some randomness
  377. for i := 0; i < 10000; i++ {
  378. ns = append(ns, &client.Node{CreatedIndex: uint64(rand.Int31())})
  379. }
  380. sns := sortableNodes{ns}
  381. sort.Sort(sns)
  382. var cis []int
  383. for _, n := range sns.Nodes {
  384. cis = append(cis, int(n.CreatedIndex))
  385. }
  386. if !sort.IntsAreSorted(cis) {
  387. t.Errorf("isSorted = %v, want %v", sort.IntsAreSorted(cis), true)
  388. }
  389. cis = make([]int, 0)
  390. for _, n := range ns {
  391. cis = append(cis, int(n.CreatedIndex))
  392. }
  393. if !sort.IntsAreSorted(cis) {
  394. t.Errorf("isSorted = %v, want %v", sort.IntsAreSorted(cis), true)
  395. }
  396. }
  397. func TestRetryFailure(t *testing.T) {
  398. nRetries = maxRetryInTest
  399. defer func() { nRetries = math.MaxUint32 }()
  400. cluster := "1000"
  401. c := &clientWithRetry{failTimes: 4}
  402. fc := clockwork.NewFakeClock()
  403. d := discovery{
  404. cluster: cluster,
  405. id: 1,
  406. c: c,
  407. clock: fc,
  408. }
  409. go func() {
  410. for i := uint(1); i <= maxRetryInTest; i++ {
  411. fc.BlockUntil(1)
  412. fc.Advance(time.Second * (0x1 << i))
  413. }
  414. }()
  415. if _, _, _, err := d.checkCluster(); err != ErrTooManyRetries {
  416. t.Errorf("err = %v, want %v", err, ErrTooManyRetries)
  417. }
  418. }
  419. type clientWithResp struct {
  420. rs []*client.Response
  421. w client.Watcher
  422. client.KeysAPI
  423. }
  424. func (c *clientWithResp) Create(ctx context.Context, key string, value string) (*client.Response, error) {
  425. if len(c.rs) == 0 {
  426. return &client.Response{}, nil
  427. }
  428. r := c.rs[0]
  429. c.rs = c.rs[1:]
  430. return r, nil
  431. }
  432. func (c *clientWithResp) Get(ctx context.Context, key string, opts *client.GetOptions) (*client.Response, error) {
  433. if len(c.rs) == 0 {
  434. return &client.Response{}, &client.Error{Code: client.ErrorCodeKeyNotFound}
  435. }
  436. r := c.rs[0]
  437. c.rs = append(c.rs[1:], r)
  438. return r, nil
  439. }
  440. func (c *clientWithResp) Watcher(key string, opts *client.WatcherOptions) client.Watcher {
  441. return c.w
  442. }
  443. type clientWithErr struct {
  444. err error
  445. w client.Watcher
  446. client.KeysAPI
  447. }
  448. func (c *clientWithErr) Create(ctx context.Context, key string, value string) (*client.Response, error) {
  449. return &client.Response{}, c.err
  450. }
  451. func (c *clientWithErr) Get(ctx context.Context, key string, opts *client.GetOptions) (*client.Response, error) {
  452. return &client.Response{}, c.err
  453. }
  454. func (c *clientWithErr) Watcher(key string, opts *client.WatcherOptions) client.Watcher {
  455. return c.w
  456. }
  457. type watcherWithResp struct {
  458. client.KeysAPI
  459. rs []*client.Response
  460. }
  461. func (w *watcherWithResp) Next(context.Context) (*client.Response, error) {
  462. if len(w.rs) == 0 {
  463. return &client.Response{}, nil
  464. }
  465. r := w.rs[0]
  466. w.rs = w.rs[1:]
  467. return r, nil
  468. }
  469. type watcherWithErr struct {
  470. err error
  471. }
  472. func (w *watcherWithErr) Next(context.Context) (*client.Response, error) {
  473. return &client.Response{}, w.err
  474. }
  475. // clientWithRetry will timeout all requests up to failTimes
  476. type clientWithRetry struct {
  477. clientWithResp
  478. failCount int
  479. failTimes int
  480. }
  481. func (c *clientWithRetry) Create(ctx context.Context, key string, value string) (*client.Response, error) {
  482. if c.failCount < c.failTimes {
  483. c.failCount++
  484. return nil, &client.ClusterError{Errors: []error{context.DeadlineExceeded}}
  485. }
  486. return c.clientWithResp.Create(ctx, key, value)
  487. }
  488. func (c *clientWithRetry) Get(ctx context.Context, key string, opts *client.GetOptions) (*client.Response, error) {
  489. if c.failCount < c.failTimes {
  490. c.failCount++
  491. return nil, &client.ClusterError{Errors: []error{context.DeadlineExceeded}}
  492. }
  493. return c.clientWithResp.Get(ctx, key, opts)
  494. }
  495. // watcherWithRetry will timeout all requests up to failTimes
  496. type watcherWithRetry struct {
  497. rs []*client.Response
  498. failCount int
  499. failTimes int
  500. }
  501. func (w *watcherWithRetry) Next(context.Context) (*client.Response, error) {
  502. if w.failCount < w.failTimes {
  503. w.failCount++
  504. return nil, &client.ClusterError{Errors: []error{context.DeadlineExceeded}}
  505. }
  506. if len(w.rs) == 0 {
  507. return &client.Response{}, nil
  508. }
  509. r := w.rs[0]
  510. w.rs = w.rs[1:]
  511. return r, nil
  512. }