transport_test.go 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154
  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 rafthttp
  15. import (
  16. "net/http"
  17. "reflect"
  18. "testing"
  19. "time"
  20. "github.com/coreos/etcd/etcdserver/stats"
  21. "github.com/coreos/etcd/pkg/testutil"
  22. "github.com/coreos/etcd/pkg/types"
  23. "github.com/coreos/etcd/raft/raftpb"
  24. "github.com/xiang90/probing"
  25. )
  26. // TestTransportSend tests that transport can send messages using correct
  27. // underlying peer, and drop local or unknown-target messages.
  28. func TestTransportSend(t *testing.T) {
  29. ss := &stats.ServerStats{}
  30. ss.Initialize()
  31. peer1 := newFakePeer()
  32. peer2 := newFakePeer()
  33. tr := &Transport{
  34. ServerStats: ss,
  35. peers: map[types.ID]Peer{types.ID(1): peer1, types.ID(2): peer2},
  36. }
  37. wmsgsIgnored := []raftpb.Message{
  38. // bad local message
  39. {Type: raftpb.MsgBeat},
  40. // bad remote message
  41. {Type: raftpb.MsgProp, To: 3},
  42. }
  43. wmsgsTo1 := []raftpb.Message{
  44. // good message
  45. {Type: raftpb.MsgProp, To: 1},
  46. {Type: raftpb.MsgApp, To: 1},
  47. }
  48. wmsgsTo2 := []raftpb.Message{
  49. // good message
  50. {Type: raftpb.MsgProp, To: 2},
  51. {Type: raftpb.MsgApp, To: 2},
  52. }
  53. tr.Send(wmsgsIgnored)
  54. tr.Send(wmsgsTo1)
  55. tr.Send(wmsgsTo2)
  56. if !reflect.DeepEqual(peer1.msgs, wmsgsTo1) {
  57. t.Errorf("msgs to peer 1 = %+v, want %+v", peer1.msgs, wmsgsTo1)
  58. }
  59. if !reflect.DeepEqual(peer2.msgs, wmsgsTo2) {
  60. t.Errorf("msgs to peer 2 = %+v, want %+v", peer2.msgs, wmsgsTo2)
  61. }
  62. }
  63. func TestTransportAdd(t *testing.T) {
  64. ls := stats.NewLeaderStats("")
  65. tr := &Transport{
  66. LeaderStats: ls,
  67. streamRt: &roundTripperRecorder{},
  68. peers: make(map[types.ID]Peer),
  69. prober: probing.NewProber(nil),
  70. }
  71. tr.AddPeer(1, []string{"http://localhost:2380"})
  72. if _, ok := ls.Followers["1"]; !ok {
  73. t.Errorf("FollowerStats[1] is nil, want exists")
  74. }
  75. s, ok := tr.peers[types.ID(1)]
  76. if !ok {
  77. tr.Stop()
  78. t.Fatalf("senders[1] is nil, want exists")
  79. }
  80. // duplicate AddPeer is ignored
  81. tr.AddPeer(1, []string{"http://localhost:2380"})
  82. ns := tr.peers[types.ID(1)]
  83. if s != ns {
  84. t.Errorf("sender = %v, want %v", ns, s)
  85. }
  86. tr.Stop()
  87. }
  88. func TestTransportRemove(t *testing.T) {
  89. tr := &Transport{
  90. LeaderStats: stats.NewLeaderStats(""),
  91. streamRt: &roundTripperRecorder{},
  92. peers: make(map[types.ID]Peer),
  93. prober: probing.NewProber(nil),
  94. }
  95. tr.AddPeer(1, []string{"http://localhost:2380"})
  96. tr.RemovePeer(types.ID(1))
  97. defer tr.Stop()
  98. if _, ok := tr.peers[types.ID(1)]; ok {
  99. t.Fatalf("senders[1] exists, want removed")
  100. }
  101. }
  102. func TestTransportUpdate(t *testing.T) {
  103. peer := newFakePeer()
  104. tr := &Transport{
  105. peers: map[types.ID]Peer{types.ID(1): peer},
  106. prober: probing.NewProber(nil),
  107. }
  108. u := "http://localhost:2380"
  109. tr.UpdatePeer(types.ID(1), []string{u})
  110. wurls := types.URLs(testutil.MustNewURLs(t, []string{"http://localhost:2380"}))
  111. if !reflect.DeepEqual(peer.peerURLs, wurls) {
  112. t.Errorf("urls = %+v, want %+v", peer.peerURLs, wurls)
  113. }
  114. }
  115. func TestTransportErrorc(t *testing.T) {
  116. errorc := make(chan error, 1)
  117. tr := &Transport{
  118. Raft: &fakeRaft{},
  119. LeaderStats: stats.NewLeaderStats(""),
  120. ErrorC: errorc,
  121. streamRt: newRespRoundTripper(http.StatusForbidden, nil),
  122. pipelineRt: newRespRoundTripper(http.StatusForbidden, nil),
  123. peers: make(map[types.ID]Peer),
  124. prober: probing.NewProber(nil),
  125. }
  126. tr.AddPeer(1, []string{"http://localhost:2380"})
  127. defer tr.Stop()
  128. select {
  129. case <-errorc:
  130. t.Fatalf("received unexpected from errorc")
  131. case <-time.After(10 * time.Millisecond):
  132. }
  133. tr.peers[1].send(raftpb.Message{})
  134. select {
  135. case <-errorc:
  136. case <-time.After(1 * time.Second):
  137. t.Fatalf("cannot receive error from errorc")
  138. }
  139. }