networkdb_test.go 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914
  1. package networkdb
  2. import (
  3. "fmt"
  4. "log"
  5. "net"
  6. "os"
  7. "strconv"
  8. "sync/atomic"
  9. "testing"
  10. "time"
  11. "github.com/docker/docker/pkg/stringid"
  12. "github.com/docker/go-events"
  13. "github.com/hashicorp/memberlist"
  14. "github.com/sirupsen/logrus"
  15. "gotest.tools/v3/assert"
  16. is "gotest.tools/v3/assert/cmp"
  17. "gotest.tools/v3/poll"
  18. )
  19. var dbPort int32 = 10000
  20. func TestMain(m *testing.M) {
  21. os.WriteFile("/proc/sys/net/ipv6/conf/lo/disable_ipv6", []byte{'0', '\n'}, 0644)
  22. logrus.SetLevel(logrus.ErrorLevel)
  23. os.Exit(m.Run())
  24. }
  25. func launchNode(t *testing.T, conf Config) *NetworkDB {
  26. t.Helper()
  27. db, err := New(&conf)
  28. assert.NilError(t, err)
  29. return db
  30. }
  31. func createNetworkDBInstances(t *testing.T, num int, namePrefix string, conf *Config) []*NetworkDB {
  32. t.Helper()
  33. var dbs []*NetworkDB
  34. for i := 0; i < num; i++ {
  35. localConfig := *conf
  36. localConfig.Hostname = fmt.Sprintf("%s%d", namePrefix, i+1)
  37. localConfig.NodeID = stringid.TruncateID(stringid.GenerateRandomID())
  38. localConfig.BindPort = int(atomic.AddInt32(&dbPort, 1))
  39. db := launchNode(t, localConfig)
  40. if i != 0 {
  41. assert.Check(t, db.Join([]string{fmt.Sprintf("localhost:%d", db.config.BindPort-1)}))
  42. }
  43. dbs = append(dbs, db)
  44. }
  45. // Wait till the cluster creation is successful
  46. check := func(t poll.LogT) poll.Result {
  47. // Check that the cluster is properly created
  48. for i := 0; i < num; i++ {
  49. if num != len(dbs[i].ClusterPeers()) {
  50. return poll.Continue("%s:Waiting for cluser peers to be established", dbs[i].config.Hostname)
  51. }
  52. }
  53. return poll.Success()
  54. }
  55. poll.WaitOn(t, check, poll.WithDelay(2*time.Second), poll.WithTimeout(20*time.Second))
  56. return dbs
  57. }
  58. func closeNetworkDBInstances(t *testing.T, dbs []*NetworkDB) {
  59. t.Helper()
  60. log.Print("Closing DB instances...")
  61. for _, db := range dbs {
  62. db.Close()
  63. }
  64. }
  65. func (db *NetworkDB) verifyNodeExistence(t *testing.T, node string, present bool) {
  66. t.Helper()
  67. for i := 0; i < 80; i++ {
  68. db.RLock()
  69. _, ok := db.nodes[node]
  70. db.RUnlock()
  71. if present && ok {
  72. return
  73. }
  74. if !present && !ok {
  75. return
  76. }
  77. time.Sleep(50 * time.Millisecond)
  78. }
  79. t.Errorf("%v(%v): Node existence verification for node %s failed", db.config.Hostname, db.config.NodeID, node)
  80. }
  81. func (db *NetworkDB) verifyNetworkExistence(t *testing.T, node string, id string, present bool) {
  82. t.Helper()
  83. const sleepInterval = 50 * time.Millisecond
  84. var maxRetries int64
  85. if dl, ok := t.Deadline(); ok {
  86. maxRetries = int64(time.Until(dl) / sleepInterval)
  87. } else {
  88. maxRetries = 80
  89. }
  90. for i := int64(0); i < maxRetries; i++ {
  91. db.RLock()
  92. nn, nnok := db.networks[node]
  93. db.RUnlock()
  94. if nnok {
  95. n, ok := nn[id]
  96. if present && ok {
  97. return
  98. }
  99. if !present &&
  100. ((ok && n.leaving) ||
  101. !ok) {
  102. return
  103. }
  104. }
  105. time.Sleep(sleepInterval)
  106. }
  107. t.Error("Network existence verification failed")
  108. }
  109. func (db *NetworkDB) verifyEntryExistence(t *testing.T, tname, nid, key, value string, present bool) {
  110. t.Helper()
  111. n := 80
  112. for i := 0; i < n; i++ {
  113. entry, err := db.getEntry(tname, nid, key)
  114. if present && err == nil && string(entry.value) == value {
  115. return
  116. }
  117. if !present &&
  118. ((err == nil && entry.deleting) ||
  119. (err != nil)) {
  120. return
  121. }
  122. if i == n-1 && !present && err != nil {
  123. return
  124. }
  125. time.Sleep(50 * time.Millisecond)
  126. }
  127. t.Errorf("Entry existence verification test failed for %v(%v)", db.config.Hostname, db.config.NodeID)
  128. }
  129. func testWatch(t *testing.T, ch chan events.Event, ev interface{}, tname, nid, key, value string) {
  130. t.Helper()
  131. select {
  132. case rcvdEv := <-ch:
  133. assert.Check(t, is.Equal(fmt.Sprintf("%T", rcvdEv), fmt.Sprintf("%T", ev)))
  134. switch typ := rcvdEv.(type) {
  135. case CreateEvent:
  136. assert.Check(t, is.Equal(tname, typ.Table))
  137. assert.Check(t, is.Equal(nid, typ.NetworkID))
  138. assert.Check(t, is.Equal(key, typ.Key))
  139. assert.Check(t, is.Equal(value, string(typ.Value)))
  140. case UpdateEvent:
  141. assert.Check(t, is.Equal(tname, typ.Table))
  142. assert.Check(t, is.Equal(nid, typ.NetworkID))
  143. assert.Check(t, is.Equal(key, typ.Key))
  144. assert.Check(t, is.Equal(value, string(typ.Value)))
  145. case DeleteEvent:
  146. assert.Check(t, is.Equal(tname, typ.Table))
  147. assert.Check(t, is.Equal(nid, typ.NetworkID))
  148. assert.Check(t, is.Equal(key, typ.Key))
  149. }
  150. case <-time.After(time.Second):
  151. t.Fail()
  152. return
  153. }
  154. }
  155. func TestNetworkDBSimple(t *testing.T) {
  156. dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig())
  157. closeNetworkDBInstances(t, dbs)
  158. }
  159. func TestNetworkDBJoinLeaveNetwork(t *testing.T) {
  160. dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig())
  161. err := dbs[0].JoinNetwork("network1")
  162. assert.NilError(t, err)
  163. dbs[1].verifyNetworkExistence(t, dbs[0].config.NodeID, "network1", true)
  164. err = dbs[0].LeaveNetwork("network1")
  165. assert.NilError(t, err)
  166. dbs[1].verifyNetworkExistence(t, dbs[0].config.NodeID, "network1", false)
  167. closeNetworkDBInstances(t, dbs)
  168. }
  169. func TestNetworkDBJoinLeaveNetworks(t *testing.T) {
  170. dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig())
  171. n := 10
  172. for i := 1; i <= n; i++ {
  173. err := dbs[0].JoinNetwork(fmt.Sprintf("network0%d", i))
  174. assert.NilError(t, err)
  175. }
  176. for i := 1; i <= n; i++ {
  177. err := dbs[1].JoinNetwork(fmt.Sprintf("network1%d", i))
  178. assert.NilError(t, err)
  179. }
  180. for i := 1; i <= n; i++ {
  181. dbs[1].verifyNetworkExistence(t, dbs[0].config.NodeID, fmt.Sprintf("network0%d", i), true)
  182. }
  183. for i := 1; i <= n; i++ {
  184. dbs[0].verifyNetworkExistence(t, dbs[1].config.NodeID, fmt.Sprintf("network1%d", i), true)
  185. }
  186. for i := 1; i <= n; i++ {
  187. err := dbs[0].LeaveNetwork(fmt.Sprintf("network0%d", i))
  188. assert.NilError(t, err)
  189. }
  190. for i := 1; i <= n; i++ {
  191. err := dbs[1].LeaveNetwork(fmt.Sprintf("network1%d", i))
  192. assert.NilError(t, err)
  193. }
  194. for i := 1; i <= n; i++ {
  195. dbs[1].verifyNetworkExistence(t, dbs[0].config.NodeID, fmt.Sprintf("network0%d", i), false)
  196. }
  197. for i := 1; i <= n; i++ {
  198. dbs[0].verifyNetworkExistence(t, dbs[1].config.NodeID, fmt.Sprintf("network1%d", i), false)
  199. }
  200. closeNetworkDBInstances(t, dbs)
  201. }
  202. func TestNetworkDBCRUDTableEntry(t *testing.T) {
  203. dbs := createNetworkDBInstances(t, 3, "node", DefaultConfig())
  204. err := dbs[0].JoinNetwork("network1")
  205. assert.NilError(t, err)
  206. dbs[1].verifyNetworkExistence(t, dbs[0].config.NodeID, "network1", true)
  207. err = dbs[1].JoinNetwork("network1")
  208. assert.NilError(t, err)
  209. err = dbs[0].CreateEntry("test_table", "network1", "test_key", []byte("test_value"))
  210. assert.NilError(t, err)
  211. dbs[1].verifyEntryExistence(t, "test_table", "network1", "test_key", "test_value", true)
  212. dbs[2].verifyEntryExistence(t, "test_table", "network1", "test_key", "test_value", false)
  213. err = dbs[0].UpdateEntry("test_table", "network1", "test_key", []byte("test_updated_value"))
  214. assert.NilError(t, err)
  215. dbs[1].verifyEntryExistence(t, "test_table", "network1", "test_key", "test_updated_value", true)
  216. err = dbs[0].DeleteEntry("test_table", "network1", "test_key")
  217. assert.NilError(t, err)
  218. dbs[1].verifyEntryExistence(t, "test_table", "network1", "test_key", "", false)
  219. closeNetworkDBInstances(t, dbs)
  220. }
  221. func TestNetworkDBCRUDTableEntries(t *testing.T) {
  222. dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig())
  223. err := dbs[0].JoinNetwork("network1")
  224. assert.NilError(t, err)
  225. dbs[1].verifyNetworkExistence(t, dbs[0].config.NodeID, "network1", true)
  226. err = dbs[1].JoinNetwork("network1")
  227. assert.NilError(t, err)
  228. dbs[0].verifyNetworkExistence(t, dbs[1].config.NodeID, "network1", true)
  229. n := 10
  230. for i := 1; i <= n; i++ {
  231. err = dbs[0].CreateEntry("test_table", "network1",
  232. fmt.Sprintf("test_key0%d", i),
  233. []byte(fmt.Sprintf("test_value0%d", i)))
  234. assert.NilError(t, err)
  235. }
  236. for i := 1; i <= n; i++ {
  237. err = dbs[1].CreateEntry("test_table", "network1",
  238. fmt.Sprintf("test_key1%d", i),
  239. []byte(fmt.Sprintf("test_value1%d", i)))
  240. assert.NilError(t, err)
  241. }
  242. for i := 1; i <= n; i++ {
  243. dbs[0].verifyEntryExistence(t, "test_table", "network1",
  244. fmt.Sprintf("test_key1%d", i),
  245. fmt.Sprintf("test_value1%d", i), true)
  246. assert.NilError(t, err)
  247. }
  248. for i := 1; i <= n; i++ {
  249. dbs[1].verifyEntryExistence(t, "test_table", "network1",
  250. fmt.Sprintf("test_key0%d", i),
  251. fmt.Sprintf("test_value0%d", i), true)
  252. assert.NilError(t, err)
  253. }
  254. // Verify deletes
  255. for i := 1; i <= n; i++ {
  256. err = dbs[0].DeleteEntry("test_table", "network1",
  257. fmt.Sprintf("test_key0%d", i))
  258. assert.NilError(t, err)
  259. }
  260. for i := 1; i <= n; i++ {
  261. err = dbs[1].DeleteEntry("test_table", "network1",
  262. fmt.Sprintf("test_key1%d", i))
  263. assert.NilError(t, err)
  264. }
  265. for i := 1; i <= n; i++ {
  266. dbs[0].verifyEntryExistence(t, "test_table", "network1",
  267. fmt.Sprintf("test_key1%d", i), "", false)
  268. assert.NilError(t, err)
  269. }
  270. for i := 1; i <= n; i++ {
  271. dbs[1].verifyEntryExistence(t, "test_table", "network1",
  272. fmt.Sprintf("test_key0%d", i), "", false)
  273. assert.NilError(t, err)
  274. }
  275. closeNetworkDBInstances(t, dbs)
  276. }
  277. func TestNetworkDBNodeLeave(t *testing.T) {
  278. dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig())
  279. err := dbs[0].JoinNetwork("network1")
  280. assert.NilError(t, err)
  281. err = dbs[1].JoinNetwork("network1")
  282. assert.NilError(t, err)
  283. err = dbs[0].CreateEntry("test_table", "network1", "test_key", []byte("test_value"))
  284. assert.NilError(t, err)
  285. dbs[1].verifyEntryExistence(t, "test_table", "network1", "test_key", "test_value", true)
  286. dbs[0].Close()
  287. dbs[1].verifyEntryExistence(t, "test_table", "network1", "test_key", "test_value", false)
  288. dbs[1].Close()
  289. }
  290. func TestNetworkDBWatch(t *testing.T) {
  291. dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig())
  292. err := dbs[0].JoinNetwork("network1")
  293. assert.NilError(t, err)
  294. err = dbs[1].JoinNetwork("network1")
  295. assert.NilError(t, err)
  296. ch, cancel := dbs[1].Watch("", "", "")
  297. err = dbs[0].CreateEntry("test_table", "network1", "test_key", []byte("test_value"))
  298. assert.NilError(t, err)
  299. testWatch(t, ch.C, CreateEvent{}, "test_table", "network1", "test_key", "test_value")
  300. err = dbs[0].UpdateEntry("test_table", "network1", "test_key", []byte("test_updated_value"))
  301. assert.NilError(t, err)
  302. testWatch(t, ch.C, UpdateEvent{}, "test_table", "network1", "test_key", "test_updated_value")
  303. err = dbs[0].DeleteEntry("test_table", "network1", "test_key")
  304. assert.NilError(t, err)
  305. testWatch(t, ch.C, DeleteEvent{}, "test_table", "network1", "test_key", "")
  306. cancel()
  307. closeNetworkDBInstances(t, dbs)
  308. }
  309. func TestNetworkDBBulkSync(t *testing.T) {
  310. dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig())
  311. err := dbs[0].JoinNetwork("network1")
  312. assert.NilError(t, err)
  313. dbs[1].verifyNetworkExistence(t, dbs[0].config.NodeID, "network1", true)
  314. n := 1000
  315. for i := 1; i <= n; i++ {
  316. err = dbs[0].CreateEntry("test_table", "network1",
  317. fmt.Sprintf("test_key0%d", i),
  318. []byte(fmt.Sprintf("test_value0%d", i)))
  319. assert.NilError(t, err)
  320. }
  321. err = dbs[1].JoinNetwork("network1")
  322. assert.NilError(t, err)
  323. dbs[0].verifyNetworkExistence(t, dbs[1].config.NodeID, "network1", true)
  324. for i := 1; i <= n; i++ {
  325. dbs[1].verifyEntryExistence(t, "test_table", "network1",
  326. fmt.Sprintf("test_key0%d", i),
  327. fmt.Sprintf("test_value0%d", i), true)
  328. assert.NilError(t, err)
  329. }
  330. closeNetworkDBInstances(t, dbs)
  331. }
  332. func TestNetworkDBCRUDMediumCluster(t *testing.T) {
  333. n := 5
  334. dbs := createNetworkDBInstances(t, n, "node", DefaultConfig())
  335. for i := 0; i < n; i++ {
  336. for j := 0; j < n; j++ {
  337. if i == j {
  338. continue
  339. }
  340. dbs[i].verifyNodeExistence(t, dbs[j].config.NodeID, true)
  341. }
  342. }
  343. for i := 0; i < n; i++ {
  344. err := dbs[i].JoinNetwork("network1")
  345. assert.NilError(t, err)
  346. }
  347. for i := 0; i < n; i++ {
  348. for j := 0; j < n; j++ {
  349. dbs[i].verifyNetworkExistence(t, dbs[j].config.NodeID, "network1", true)
  350. }
  351. }
  352. err := dbs[0].CreateEntry("test_table", "network1", "test_key", []byte("test_value"))
  353. assert.NilError(t, err)
  354. for i := 1; i < n; i++ {
  355. dbs[i].verifyEntryExistence(t, "test_table", "network1", "test_key", "test_value", true)
  356. }
  357. err = dbs[0].UpdateEntry("test_table", "network1", "test_key", []byte("test_updated_value"))
  358. assert.NilError(t, err)
  359. for i := 1; i < n; i++ {
  360. dbs[i].verifyEntryExistence(t, "test_table", "network1", "test_key", "test_updated_value", true)
  361. }
  362. err = dbs[0].DeleteEntry("test_table", "network1", "test_key")
  363. assert.NilError(t, err)
  364. for i := 1; i < n; i++ {
  365. dbs[i].verifyEntryExistence(t, "test_table", "network1", "test_key", "", false)
  366. }
  367. for i := 1; i < n; i++ {
  368. _, err = dbs[i].GetEntry("test_table", "network1", "test_key")
  369. assert.Check(t, is.ErrorContains(err, ""))
  370. assert.Check(t, is.Contains(err.Error(), "deleted and pending garbage collection"), err)
  371. }
  372. closeNetworkDBInstances(t, dbs)
  373. }
  374. func TestNetworkDBNodeJoinLeaveIteration(t *testing.T) {
  375. dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig())
  376. // Single node Join/Leave
  377. err := dbs[0].JoinNetwork("network1")
  378. assert.NilError(t, err)
  379. if len(dbs[0].networkNodes["network1"]) != 1 {
  380. t.Fatalf("The networkNodes list has to have be 1 instead of %d", len(dbs[0].networkNodes["network1"]))
  381. }
  382. err = dbs[0].LeaveNetwork("network1")
  383. assert.NilError(t, err)
  384. if len(dbs[0].networkNodes["network1"]) != 0 {
  385. t.Fatalf("The networkNodes list has to have be 0 instead of %d", len(dbs[0].networkNodes["network1"]))
  386. }
  387. // Multiple nodes Join/Leave
  388. err = dbs[0].JoinNetwork("network1")
  389. assert.NilError(t, err)
  390. err = dbs[1].JoinNetwork("network1")
  391. assert.NilError(t, err)
  392. // Wait for the propagation on db[0]
  393. dbs[0].verifyNetworkExistence(t, dbs[1].config.NodeID, "network1", true)
  394. if len(dbs[0].networkNodes["network1"]) != 2 {
  395. t.Fatalf("The networkNodes list has to have be 2 instead of %d - %v", len(dbs[0].networkNodes["network1"]), dbs[0].networkNodes["network1"])
  396. }
  397. if n, ok := dbs[0].networks[dbs[0].config.NodeID]["network1"]; !ok || n.leaving {
  398. t.Fatalf("The network should not be marked as leaving:%t", n.leaving)
  399. }
  400. // Wait for the propagation on db[1]
  401. dbs[1].verifyNetworkExistence(t, dbs[0].config.NodeID, "network1", true)
  402. if len(dbs[1].networkNodes["network1"]) != 2 {
  403. t.Fatalf("The networkNodes list has to have be 2 instead of %d - %v", len(dbs[1].networkNodes["network1"]), dbs[1].networkNodes["network1"])
  404. }
  405. if n, ok := dbs[1].networks[dbs[1].config.NodeID]["network1"]; !ok || n.leaving {
  406. t.Fatalf("The network should not be marked as leaving:%t", n.leaving)
  407. }
  408. // Try a quick leave/join
  409. err = dbs[0].LeaveNetwork("network1")
  410. assert.NilError(t, err)
  411. err = dbs[0].JoinNetwork("network1")
  412. assert.NilError(t, err)
  413. dbs[0].verifyNetworkExistence(t, dbs[1].config.NodeID, "network1", true)
  414. if len(dbs[0].networkNodes["network1"]) != 2 {
  415. t.Fatalf("The networkNodes list has to have be 2 instead of %d - %v", len(dbs[0].networkNodes["network1"]), dbs[0].networkNodes["network1"])
  416. }
  417. dbs[1].verifyNetworkExistence(t, dbs[0].config.NodeID, "network1", true)
  418. if len(dbs[1].networkNodes["network1"]) != 2 {
  419. t.Fatalf("The networkNodes list has to have be 2 instead of %d - %v", len(dbs[1].networkNodes["network1"]), dbs[1].networkNodes["network1"])
  420. }
  421. closeNetworkDBInstances(t, dbs)
  422. }
  423. func TestNetworkDBGarbageCollection(t *testing.T) {
  424. keysWriteDelete := 5
  425. config := DefaultConfig()
  426. config.reapEntryInterval = 30 * time.Second
  427. config.StatsPrintPeriod = 15 * time.Second
  428. dbs := createNetworkDBInstances(t, 3, "node", config)
  429. // 2 Nodes join network
  430. err := dbs[0].JoinNetwork("network1")
  431. assert.NilError(t, err)
  432. err = dbs[1].JoinNetwork("network1")
  433. assert.NilError(t, err)
  434. for i := 0; i < keysWriteDelete; i++ {
  435. err = dbs[i%2].CreateEntry("testTable", "network1", "key-"+strconv.Itoa(i), []byte("value"))
  436. assert.NilError(t, err)
  437. }
  438. time.Sleep(time.Second)
  439. for i := 0; i < keysWriteDelete; i++ {
  440. err = dbs[i%2].DeleteEntry("testTable", "network1", "key-"+strconv.Itoa(i))
  441. assert.NilError(t, err)
  442. }
  443. for i := 0; i < 2; i++ {
  444. assert.Check(t, is.Equal(keysWriteDelete, dbs[i].networks[dbs[i].config.NodeID]["network1"].entriesNumber), "entries number should match")
  445. }
  446. // from this point the timer for the garbage collection started, wait 5 seconds and then join a new node
  447. time.Sleep(5 * time.Second)
  448. err = dbs[2].JoinNetwork("network1")
  449. assert.NilError(t, err)
  450. for i := 0; i < 3; i++ {
  451. assert.Check(t, is.Equal(keysWriteDelete, dbs[i].networks[dbs[i].config.NodeID]["network1"].entriesNumber), "entries number should match")
  452. }
  453. // at this point the entries should had been all deleted
  454. time.Sleep(30 * time.Second)
  455. for i := 0; i < 3; i++ {
  456. assert.Check(t, is.Equal(0, dbs[i].networks[dbs[i].config.NodeID]["network1"].entriesNumber), "entries should had been garbage collected")
  457. }
  458. // make sure that entries are not coming back
  459. time.Sleep(15 * time.Second)
  460. for i := 0; i < 3; i++ {
  461. assert.Check(t, is.Equal(0, dbs[i].networks[dbs[i].config.NodeID]["network1"].entriesNumber), "entries should had been garbage collected")
  462. }
  463. closeNetworkDBInstances(t, dbs)
  464. }
  465. func TestFindNode(t *testing.T) {
  466. dbs := createNetworkDBInstances(t, 1, "node", DefaultConfig())
  467. dbs[0].nodes["active"] = &node{Node: memberlist.Node{Name: "active"}}
  468. dbs[0].failedNodes["failed"] = &node{Node: memberlist.Node{Name: "failed"}}
  469. dbs[0].leftNodes["left"] = &node{Node: memberlist.Node{Name: "left"}}
  470. // active nodes is 2 because the testing node is in the list
  471. assert.Check(t, is.Len(dbs[0].nodes, 2))
  472. assert.Check(t, is.Len(dbs[0].failedNodes, 1))
  473. assert.Check(t, is.Len(dbs[0].leftNodes, 1))
  474. n, currState, m := dbs[0].findNode("active")
  475. assert.Check(t, n != nil)
  476. assert.Check(t, is.Equal("active", n.Name))
  477. assert.Check(t, is.Equal(nodeActiveState, currState))
  478. assert.Check(t, m != nil)
  479. // delete the entry manually
  480. delete(m, "active")
  481. // test if can be still find
  482. n, currState, m = dbs[0].findNode("active")
  483. assert.Check(t, is.Nil(n))
  484. assert.Check(t, is.Equal(nodeNotFound, currState))
  485. assert.Check(t, is.Nil(m))
  486. n, currState, m = dbs[0].findNode("failed")
  487. assert.Check(t, n != nil)
  488. assert.Check(t, is.Equal("failed", n.Name))
  489. assert.Check(t, is.Equal(nodeFailedState, currState))
  490. assert.Check(t, m != nil)
  491. // find and remove
  492. n, currState, m = dbs[0].findNode("left")
  493. assert.Check(t, n != nil)
  494. assert.Check(t, is.Equal("left", n.Name))
  495. assert.Check(t, is.Equal(nodeLeftState, currState))
  496. assert.Check(t, m != nil)
  497. delete(m, "left")
  498. n, currState, m = dbs[0].findNode("left")
  499. assert.Check(t, is.Nil(n))
  500. assert.Check(t, is.Equal(nodeNotFound, currState))
  501. assert.Check(t, is.Nil(m))
  502. closeNetworkDBInstances(t, dbs)
  503. }
  504. func TestChangeNodeState(t *testing.T) {
  505. dbs := createNetworkDBInstances(t, 1, "node", DefaultConfig())
  506. dbs[0].nodes["node1"] = &node{Node: memberlist.Node{Name: "node1"}}
  507. dbs[0].nodes["node2"] = &node{Node: memberlist.Node{Name: "node2"}}
  508. dbs[0].nodes["node3"] = &node{Node: memberlist.Node{Name: "node3"}}
  509. // active nodes is 4 because the testing node is in the list
  510. assert.Check(t, is.Len(dbs[0].nodes, 4))
  511. n, currState, m := dbs[0].findNode("node1")
  512. assert.Check(t, n != nil)
  513. assert.Check(t, is.Equal(nodeActiveState, currState))
  514. assert.Check(t, is.Equal("node1", n.Name))
  515. assert.Check(t, m != nil)
  516. // node1 to failed
  517. dbs[0].changeNodeState("node1", nodeFailedState)
  518. n, currState, m = dbs[0].findNode("node1")
  519. assert.Check(t, n != nil)
  520. assert.Check(t, is.Equal(nodeFailedState, currState))
  521. assert.Check(t, is.Equal("node1", n.Name))
  522. assert.Check(t, m != nil)
  523. assert.Check(t, time.Duration(0) != n.reapTime)
  524. // node1 back to active
  525. dbs[0].changeNodeState("node1", nodeActiveState)
  526. n, currState, m = dbs[0].findNode("node1")
  527. assert.Check(t, n != nil)
  528. assert.Check(t, is.Equal(nodeActiveState, currState))
  529. assert.Check(t, is.Equal("node1", n.Name))
  530. assert.Check(t, m != nil)
  531. assert.Check(t, is.Equal(time.Duration(0), n.reapTime))
  532. // node1 to left
  533. dbs[0].changeNodeState("node1", nodeLeftState)
  534. dbs[0].changeNodeState("node2", nodeLeftState)
  535. dbs[0].changeNodeState("node3", nodeLeftState)
  536. n, currState, m = dbs[0].findNode("node1")
  537. assert.Check(t, n != nil)
  538. assert.Check(t, is.Equal(nodeLeftState, currState))
  539. assert.Check(t, is.Equal("node1", n.Name))
  540. assert.Check(t, m != nil)
  541. assert.Check(t, time.Duration(0) != n.reapTime)
  542. n, currState, m = dbs[0].findNode("node2")
  543. assert.Check(t, n != nil)
  544. assert.Check(t, is.Equal(nodeLeftState, currState))
  545. assert.Check(t, is.Equal("node2", n.Name))
  546. assert.Check(t, m != nil)
  547. assert.Check(t, time.Duration(0) != n.reapTime)
  548. n, currState, m = dbs[0].findNode("node3")
  549. assert.Check(t, n != nil)
  550. assert.Check(t, is.Equal(nodeLeftState, currState))
  551. assert.Check(t, is.Equal("node3", n.Name))
  552. assert.Check(t, m != nil)
  553. assert.Check(t, time.Duration(0) != n.reapTime)
  554. // active nodes is 1 because the testing node is in the list
  555. assert.Check(t, is.Len(dbs[0].nodes, 1))
  556. assert.Check(t, is.Len(dbs[0].failedNodes, 0))
  557. assert.Check(t, is.Len(dbs[0].leftNodes, 3))
  558. closeNetworkDBInstances(t, dbs)
  559. }
  560. func TestNodeReincarnation(t *testing.T) {
  561. dbs := createNetworkDBInstances(t, 1, "node", DefaultConfig())
  562. dbs[0].nodes["node1"] = &node{Node: memberlist.Node{Name: "node1", Addr: net.ParseIP("192.168.1.1")}}
  563. dbs[0].leftNodes["node2"] = &node{Node: memberlist.Node{Name: "node2", Addr: net.ParseIP("192.168.1.2")}}
  564. dbs[0].failedNodes["node3"] = &node{Node: memberlist.Node{Name: "node3", Addr: net.ParseIP("192.168.1.3")}}
  565. // active nodes is 2 because the testing node is in the list
  566. assert.Check(t, is.Len(dbs[0].nodes, 2))
  567. assert.Check(t, is.Len(dbs[0].failedNodes, 1))
  568. assert.Check(t, is.Len(dbs[0].leftNodes, 1))
  569. b := dbs[0].purgeReincarnation(&memberlist.Node{Name: "node4", Addr: net.ParseIP("192.168.1.1")})
  570. assert.Check(t, b)
  571. dbs[0].nodes["node4"] = &node{Node: memberlist.Node{Name: "node4", Addr: net.ParseIP("192.168.1.1")}}
  572. b = dbs[0].purgeReincarnation(&memberlist.Node{Name: "node5", Addr: net.ParseIP("192.168.1.2")})
  573. assert.Check(t, b)
  574. dbs[0].nodes["node5"] = &node{Node: memberlist.Node{Name: "node5", Addr: net.ParseIP("192.168.1.1")}}
  575. b = dbs[0].purgeReincarnation(&memberlist.Node{Name: "node6", Addr: net.ParseIP("192.168.1.3")})
  576. assert.Check(t, b)
  577. dbs[0].nodes["node6"] = &node{Node: memberlist.Node{Name: "node6", Addr: net.ParseIP("192.168.1.1")}}
  578. b = dbs[0].purgeReincarnation(&memberlist.Node{Name: "node6", Addr: net.ParseIP("192.168.1.10")})
  579. assert.Check(t, !b)
  580. // active nodes is 1 because the testing node is in the list
  581. assert.Check(t, is.Len(dbs[0].nodes, 4))
  582. assert.Check(t, is.Len(dbs[0].failedNodes, 0))
  583. assert.Check(t, is.Len(dbs[0].leftNodes, 3))
  584. closeNetworkDBInstances(t, dbs)
  585. }
  586. func TestParallelCreate(t *testing.T) {
  587. dbs := createNetworkDBInstances(t, 1, "node", DefaultConfig())
  588. startCh := make(chan int)
  589. doneCh := make(chan error)
  590. var success int32
  591. for i := 0; i < 20; i++ {
  592. go func() {
  593. <-startCh
  594. err := dbs[0].CreateEntry("testTable", "testNetwork", "key", []byte("value"))
  595. if err == nil {
  596. atomic.AddInt32(&success, 1)
  597. }
  598. doneCh <- err
  599. }()
  600. }
  601. close(startCh)
  602. for i := 0; i < 20; i++ {
  603. <-doneCh
  604. }
  605. close(doneCh)
  606. // Only 1 write should have succeeded
  607. assert.Check(t, is.Equal(int32(1), success))
  608. closeNetworkDBInstances(t, dbs)
  609. }
  610. func TestParallelDelete(t *testing.T) {
  611. dbs := createNetworkDBInstances(t, 1, "node", DefaultConfig())
  612. err := dbs[0].CreateEntry("testTable", "testNetwork", "key", []byte("value"))
  613. assert.NilError(t, err)
  614. startCh := make(chan int)
  615. doneCh := make(chan error)
  616. var success int32
  617. for i := 0; i < 20; i++ {
  618. go func() {
  619. <-startCh
  620. err := dbs[0].DeleteEntry("testTable", "testNetwork", "key")
  621. if err == nil {
  622. atomic.AddInt32(&success, 1)
  623. }
  624. doneCh <- err
  625. }()
  626. }
  627. close(startCh)
  628. for i := 0; i < 20; i++ {
  629. <-doneCh
  630. }
  631. close(doneCh)
  632. // Only 1 write should have succeeded
  633. assert.Check(t, is.Equal(int32(1), success))
  634. closeNetworkDBInstances(t, dbs)
  635. }
  636. func TestNetworkDBIslands(t *testing.T) {
  637. pollTimeout := func() time.Duration {
  638. const defaultTimeout = 120 * time.Second
  639. dl, ok := t.Deadline()
  640. if !ok {
  641. return defaultTimeout
  642. }
  643. if d := time.Until(dl); d <= defaultTimeout {
  644. return d
  645. }
  646. return defaultTimeout
  647. }
  648. logrus.SetLevel(logrus.DebugLevel)
  649. conf := DefaultConfig()
  650. // Shorten durations to speed up test execution.
  651. conf.rejoinClusterDuration = conf.rejoinClusterDuration / 10
  652. conf.rejoinClusterInterval = conf.rejoinClusterInterval / 10
  653. dbs := createNetworkDBInstances(t, 5, "node", conf)
  654. // Get the node IP used currently
  655. node := dbs[0].nodes[dbs[0].config.NodeID]
  656. baseIPStr := node.Addr.String()
  657. // Node 0,1,2 are going to be the 3 bootstrap nodes
  658. members := []string{fmt.Sprintf("%s:%d", baseIPStr, dbs[0].config.BindPort),
  659. fmt.Sprintf("%s:%d", baseIPStr, dbs[1].config.BindPort),
  660. fmt.Sprintf("%s:%d", baseIPStr, dbs[2].config.BindPort)}
  661. // Rejoining will update the list of the bootstrap members
  662. for i := 3; i < 5; i++ {
  663. t.Logf("Re-joining: %d", i)
  664. assert.Check(t, dbs[i].Join(members))
  665. }
  666. // Now the 3 bootstrap nodes will cleanly leave, and will be properly removed from the other 2 nodes
  667. for i := 0; i < 3; i++ {
  668. logrus.Infof("node %d leaving", i)
  669. dbs[i].Close()
  670. }
  671. checkDBs := make(map[string]*NetworkDB)
  672. for i := 3; i < 5; i++ {
  673. db := dbs[i]
  674. checkDBs[db.config.Hostname] = db
  675. }
  676. // Give some time to let the system propagate the messages and free up the ports
  677. check := func(t poll.LogT) poll.Result {
  678. // Verify that the nodes are actually all gone and marked appropiately
  679. for name, db := range checkDBs {
  680. db.RLock()
  681. if (len(db.leftNodes) != 3) || (len(db.failedNodes) != 0) {
  682. for name := range db.leftNodes {
  683. t.Logf("%s: Node %s left", db.config.Hostname, name)
  684. }
  685. for name := range db.failedNodes {
  686. t.Logf("%s: Node %s failed", db.config.Hostname, name)
  687. }
  688. db.RUnlock()
  689. return poll.Continue("%s:Waiting for all nodes to cleanly leave, left: %d, failed nodes: %d", name, len(db.leftNodes), len(db.failedNodes))
  690. }
  691. db.RUnlock()
  692. t.Logf("%s: OK", name)
  693. delete(checkDBs, name)
  694. }
  695. return poll.Success()
  696. }
  697. poll.WaitOn(t, check, poll.WithDelay(time.Second), poll.WithTimeout(pollTimeout()))
  698. // Spawn again the first 3 nodes with different names but same IP:port
  699. for i := 0; i < 3; i++ {
  700. logrus.Infof("node %d coming back", i)
  701. dbs[i].config.NodeID = stringid.TruncateID(stringid.GenerateRandomID())
  702. dbs[i] = launchNode(t, *dbs[i].config)
  703. }
  704. // Give some time for the reconnect routine to run, it runs every 6s.
  705. check = func(t poll.LogT) poll.Result {
  706. // Verify that the cluster is again all connected. Note that the 3 previous node did not do any join
  707. for i := 0; i < 5; i++ {
  708. db := dbs[i]
  709. db.RLock()
  710. if len(db.nodes) != 5 {
  711. db.RUnlock()
  712. return poll.Continue("%s:Waiting to connect to all nodes", dbs[i].config.Hostname)
  713. }
  714. if len(db.failedNodes) != 0 {
  715. db.RUnlock()
  716. return poll.Continue("%s:Waiting for 0 failedNodes", dbs[i].config.Hostname)
  717. }
  718. if i < 3 {
  719. // nodes from 0 to 3 has no left nodes
  720. if len(db.leftNodes) != 0 {
  721. db.RUnlock()
  722. return poll.Continue("%s:Waiting to have no leftNodes", dbs[i].config.Hostname)
  723. }
  724. } else {
  725. // nodes from 4 to 5 has the 3 previous left nodes
  726. if len(db.leftNodes) != 3 {
  727. db.RUnlock()
  728. return poll.Continue("%s:Waiting to have 3 leftNodes", dbs[i].config.Hostname)
  729. }
  730. }
  731. db.RUnlock()
  732. }
  733. return poll.Success()
  734. }
  735. poll.WaitOn(t, check, poll.WithDelay(time.Second), poll.WithTimeout(pollTimeout()))
  736. closeNetworkDBInstances(t, dbs)
  737. }