libnetwork/cmd/networkdb-test/dbclient/ndbClient.go
package dbclient
import (
"context"
"fmt"
"io"
"net"
"net/http"
"os"
"regexp"
"strconv"
"strings"
"time"
"github.com/containerd/log"
)
var servicePort string
const totalWrittenKeys string = "totalKeys"
type resultTuple struct {
id string
result int
}
func httpGetFatalError(ip, port, path string) {
body, err := httpGet(ip, port, path)
if err != nil || !strings.Contains(string(body), "OK") {
log.G(context.TODO()).Fatalf("[%s] error %s %s", path, err, body)
}
}
func httpGet(ip, port, path string) ([]byte, error) {
resp, err := http.Get("http://" + ip + ":" + port + path)
if err != nil {
log.G(context.TODO()).Errorf("httpGet error:%s", err)
return nil, err
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
return body, err
}
func joinCluster(ip, port string, members []string, doneCh chan resultTuple) {
httpGetFatalError(ip, port, "/join?members="+strings.Join(members, ","))
if doneCh != nil {
doneCh <- resultTuple{id: ip, result: 0}
}
}
func joinNetwork(ip, port, network string, doneCh chan resultTuple) {
httpGetFatalError(ip, port, "/joinnetwork?nid="+network)
if doneCh != nil {
doneCh <- resultTuple{id: ip, result: 0}
}
}
func leaveNetwork(ip, port, network string, doneCh chan resultTuple) {
httpGetFatalError(ip, port, "/leavenetwork?nid="+network)
if doneCh != nil {
doneCh <- resultTuple{id: ip, result: 0}
}
}
func writeTableKey(ip, port, networkName, tableName, key string) {
createPath := "/createentry?unsafe&nid=" + networkName + "&tname=" + tableName + "&value=v&key="
httpGetFatalError(ip, port, createPath+key)
}
func deleteTableKey(ip, port, networkName, tableName, key string) {
deletePath := "/deleteentry?nid=" + networkName + "&tname=" + tableName + "&key="
httpGetFatalError(ip, port, deletePath+key)
}
func clusterPeersNumber(ip, port string, doneCh chan resultTuple) {
body, err := httpGet(ip, port, "/clusterpeers")
if err != nil {
log.G(context.TODO()).Errorf("clusterPeers %s there was an error: %s", ip, err)
doneCh <- resultTuple{id: ip, result: -1}
return
}
peersRegexp := regexp.MustCompile(`total entries: ([0-9]+)`)
peersNum, _ := strconv.Atoi(peersRegexp.FindStringSubmatch(string(body))[1])
doneCh <- resultTuple{id: ip, result: peersNum}
}
func networkPeersNumber(ip, port, networkName string, doneCh chan resultTuple) {
body, err := httpGet(ip, port, "/networkpeers?nid="+networkName)
if err != nil {
log.G(context.TODO()).Errorf("networkPeersNumber %s there was an error: %s", ip, err)
doneCh <- resultTuple{id: ip, result: -1}
return
}
peersRegexp := regexp.MustCompile(`total entries: ([0-9]+)`)
peersNum, _ := strconv.Atoi(peersRegexp.FindStringSubmatch(string(body))[1])
doneCh <- resultTuple{id: ip, result: peersNum}
}
func dbTableEntriesNumber(ip, port, networkName, tableName string, doneCh chan resultTuple) {
body, err := httpGet(ip, port, "/gettable?nid="+networkName+"&tname="+tableName)
if err != nil {
log.G(context.TODO()).Errorf("tableEntriesNumber %s there was an error: %s", ip, err)
doneCh <- resultTuple{id: ip, result: -1}
return
}
elementsRegexp := regexp.MustCompile(`total entries: ([0-9]+)`)
entriesNum, _ := strconv.Atoi(elementsRegexp.FindStringSubmatch(string(body))[1])
doneCh <- resultTuple{id: ip, result: entriesNum}
}
func dbQueueLength(ip, port, networkName string, doneCh chan resultTuple) {
body, err := httpGet(ip, port, "/networkstats?nid="+networkName)
if err != nil {
log.G(context.TODO()).Errorf("queueLength %s there was an error: %s", ip, err)
doneCh <- resultTuple{id: ip, result: -1}
return
}
elementsRegexp := regexp.MustCompile(`qlen: ([0-9]+)`)
entriesNum, _ := strconv.Atoi(elementsRegexp.FindStringSubmatch(string(body))[1])
doneCh <- resultTuple{id: ip, result: entriesNum}
}
func clientWatchTable(ip, port, networkName, tableName string, doneCh chan resultTuple) {
httpGetFatalError(ip, port, "/watchtable?nid="+networkName+"&tname="+tableName)
if doneCh != nil {
doneCh <- resultTuple{id: ip, result: 0}
}
}
func clientTableEntriesNumber(ip, port, networkName, tableName string, doneCh chan resultTuple) {
body, err := httpGet(ip, port, "/watchedtableentries?nid="+networkName+"&tname="+tableName)
if err != nil {
log.G(context.TODO()).Errorf("clientTableEntriesNumber %s there was an error: %s", ip, err)
doneCh <- resultTuple{id: ip, result: -1}
return
}
elementsRegexp := regexp.MustCompile(`total elements: ([0-9]+)`)
entriesNum, _ := strconv.Atoi(elementsRegexp.FindStringSubmatch(string(body))[1])
doneCh <- resultTuple{id: ip, result: entriesNum}
}
func writeKeysNumber(ip, port, networkName, tableName, key string, number int, doneCh chan resultTuple) {
x := 0
for ; x < number; x++ {
k := key + strconv.Itoa(x)
// write key
writeTableKey(ip, port, networkName, tableName, k)
}
doneCh <- resultTuple{id: ip, result: x}
}
func deleteKeysNumber(ip, port, networkName, tableName, key string, number int, doneCh chan resultTuple) {
x := 0
for ; x < number; x++ {
k := key + strconv.Itoa(x)
// write key
deleteTableKey(ip, port, networkName, tableName, k)
}
doneCh <- resultTuple{id: ip, result: x}
}
func writeUniqueKeys(ctx context.Context, ip, port, networkName, tableName, key string, doneCh chan resultTuple) {
for x := 0; ; x++ {
select {
case <-ctx.Done():
doneCh <- resultTuple{id: ip, result: x}
return
default:
k := key + strconv.Itoa(x)
// write key
writeTableKey(ip, port, networkName, tableName, k)
// give time to send out key writes
time.Sleep(100 * time.Millisecond)
}
}
}
func writeDeleteUniqueKeys(ctx context.Context, ip, port, networkName, tableName, key string, doneCh chan resultTuple) {
for x := 0; ; x++ {
select {
case <-ctx.Done():
doneCh <- resultTuple{id: ip, result: x}
return
default:
k := key + strconv.Itoa(x)
// write key
writeTableKey(ip, port, networkName, tableName, k)
// give time to send out key writes
time.Sleep(100 * time.Millisecond)
// delete key
deleteTableKey(ip, port, networkName, tableName, k)
}
}
}
func writeDeleteLeaveJoin(ctx context.Context, ip, port, networkName, tableName, key string, doneCh chan resultTuple) {
for x := 0; ; x++ {
select {
case <-ctx.Done():
doneCh <- resultTuple{id: ip, result: x}
return
default:
k := key + strconv.Itoa(x)
// write key
writeTableKey(ip, port, networkName, tableName, k)
time.Sleep(100 * time.Millisecond)
// delete key
deleteTableKey(ip, port, networkName, tableName, k)
// give some time
time.Sleep(100 * time.Millisecond)
// leave network
leaveNetwork(ip, port, networkName, nil)
// join network
joinNetwork(ip, port, networkName, nil)
}
}
}
func ready(ip, port string, doneCh chan resultTuple) {
for {
body, err := httpGet(ip, port, "/ready")
if err != nil || !strings.Contains(string(body), "OK") {
time.Sleep(500 * time.Millisecond)
continue
}
// success
break
}
// notify the completion
doneCh <- resultTuple{id: ip, result: 0}
}
func checkTable(ctx context.Context, ips []string, port, networkName, tableName string, expectedEntries int, fn func(string, string, string, string, chan resultTuple)) (opTime time.Duration) {
startTime := time.Now().UnixNano()
var successTime int64
// Loop for 2 minutes to guarantee that the result is stable
for {
select {
case <-ctx.Done():
// Validate test success, if the time is set means that all the tables are empty
if successTime != 0 {
opTime = time.Duration(successTime-startTime) / time.Millisecond
log.G(ctx).Infof("Check table passed, the cluster converged in %d msec", opTime)
return
}
log.G(ctx).Fatal("Test failed, there is still entries in the tables of the nodes")
default:
log.G(ctx).Infof("Checking table %s expected %d", tableName, expectedEntries)
doneCh := make(chan resultTuple, len(ips))
for _, ip := range ips {
go fn(ip, servicePort, networkName, tableName, doneCh)
}
nodesWithCorrectEntriesNum := 0
for i := len(ips); i > 0; i-- {
tableEntries := <-doneCh
log.G(ctx).Infof("Node %s has %d entries", tableEntries.id, tableEntries.result)
if tableEntries.result == expectedEntries {
nodesWithCorrectEntriesNum++
}
}
close(doneCh)
if nodesWithCorrectEntriesNum == len(ips) {
if successTime == 0 {
successTime = time.Now().UnixNano()
log.G(ctx).Infof("Success after %d msec", time.Duration(successTime-startTime)/time.Millisecond)
}
} else {
successTime = 0
}
time.Sleep(10 * time.Second)
}
}
}
func waitWriters(parallelWriters int, mustWrite bool, doneCh chan resultTuple) map[string]int {
var totalKeys int
resultTable := make(map[string]int)
for i := 0; i < parallelWriters; i++ {
log.G(context.TODO()).Infof("Waiting for %d workers", parallelWriters-i)
workerReturn := <-doneCh
totalKeys += workerReturn.result
if mustWrite && workerReturn.result == 0 {
log.G(context.TODO()).Fatalf("The worker %s did not write any key %d == 0", workerReturn.id, workerReturn.result)
}
if !mustWrite && workerReturn.result != 0 {
log.G(context.TODO()).Fatalf("The worker %s was supposed to return 0 instead %d != 0", workerReturn.id, workerReturn.result)
}
if mustWrite {
resultTable[workerReturn.id] = workerReturn.result
log.G(context.TODO()).Infof("The worker %s wrote %d keys", workerReturn.id, workerReturn.result)
}
}
resultTable[totalWrittenKeys] = totalKeys
return resultTable
}
// ready
func doReady(ips []string) {
doneCh := make(chan resultTuple, len(ips))
// check all the nodes
for _, ip := range ips {
go ready(ip, servicePort, doneCh)
}
// wait for the readiness of all nodes
for i := len(ips); i > 0; i-- {
<-doneCh
}
close(doneCh)
}
// join
func doJoin(ips []string) {
doneCh := make(chan resultTuple, len(ips))
// check all the nodes
for i, ip := range ips {
members := append([]string(nil), ips[:i]...)
members = append(members, ips[i+1:]...)
go joinCluster(ip, servicePort, members, doneCh)
}
// wait for the readiness of all nodes
for i := len(ips); i > 0; i-- {
<-doneCh
}
close(doneCh)
}
// cluster-peers expectedNumberPeers maxRetry
func doClusterPeers(ips []string, args []string) {
doneCh := make(chan resultTuple, len(ips))
expectedPeers, _ := strconv.Atoi(args[0])
maxRetry, _ := strconv.Atoi(args[1])
for retry := 0; retry < maxRetry; retry++ {
// check all the nodes
for _, ip := range ips {
go clusterPeersNumber(ip, servicePort, doneCh)
}
var failed bool
// wait for the readiness of all nodes
for i := len(ips); i > 0; i-- {
node := <-doneCh
if node.result != expectedPeers {
failed = true
if retry == maxRetry-1 {
log.G(context.TODO()).Fatalf("Expected peers from %s mismatch %d != %d", node.id, expectedPeers, node.result)
} else {
log.G(context.TODO()).Warnf("Expected peers from %s mismatch %d != %d", node.id, expectedPeers, node.result)
}
time.Sleep(1 * time.Second)
}
}
// check if needs retry
if !failed {
break
}
}
close(doneCh)
}
// join-network networkName
func doJoinNetwork(ips []string, args []string) {
doneCh := make(chan resultTuple, len(ips))
// check all the nodes
for _, ip := range ips {
go joinNetwork(ip, servicePort, args[0], doneCh)
}
// wait for the readiness of all nodes
for i := len(ips); i > 0; i-- {
<-doneCh
}
close(doneCh)
}
// leave-network networkName
func doLeaveNetwork(ips []string, args []string) {
doneCh := make(chan resultTuple, len(ips))
// check all the nodes
for _, ip := range ips {
go leaveNetwork(ip, servicePort, args[0], doneCh)
}
// wait for the readiness of all nodes
for i := len(ips); i > 0; i-- {
<-doneCh
}
close(doneCh)
}
// network-peers networkName expectedNumberPeers maxRetry
func doNetworkPeers(ips []string, args []string) {
doneCh := make(chan resultTuple, len(ips))
networkName := args[0]
expectedPeers, _ := strconv.Atoi(args[1])
maxRetry, _ := strconv.Atoi(args[2])
for retry := 0; retry < maxRetry; retry++ {
// check all the nodes
for _, ip := range ips {
go networkPeersNumber(ip, servicePort, networkName, doneCh)
}
var failed bool
// wait for the readiness of all nodes
for i := len(ips); i > 0; i-- {
node := <-doneCh
if node.result != expectedPeers {
failed = true
if retry == maxRetry-1 {
log.G(context.TODO()).Fatalf("Expected peers from %s mismatch %d != %d", node.id, expectedPeers, node.result)
} else {
log.G(context.TODO()).Warnf("Expected peers from %s mismatch %d != %d", node.id, expectedPeers, node.result)
}
time.Sleep(1 * time.Second)
}
}
// check if needs retry
if !failed {
break
}
}
close(doneCh)
}
// network-stats-queue networkName <gt/lt> queueSize
func doNetworkStatsQueue(ips []string, args []string) {
doneCh := make(chan resultTuple, len(ips))
networkName := args[0]
comparison := args[1]
size, _ := strconv.Atoi(args[2])
// check all the nodes
for _, ip := range ips {
go dbQueueLength(ip, servicePort, networkName, doneCh)
}
var avgQueueSize int
// wait for the readiness of all nodes
for i := len(ips); i > 0; i-- {
node := <-doneCh
switch comparison {
case "lt":
if node.result > size {
log.G(context.TODO()).Fatalf("Expected queue size from %s to be %d < %d", node.id, node.result, size)
}
case "gt":
if node.result < size {
log.G(context.TODO()).Fatalf("Expected queue size from %s to be %d > %d", node.id, node.result, size)
}
default:
log.G(context.TODO()).Fatal("unknown comparison operator")
}
avgQueueSize += node.result
}
close(doneCh)
avgQueueSize /= len(ips)
fmt.Fprintf(os.Stderr, "doNetworkStatsQueue succeeded with avg queue:%d", avgQueueSize)
}
// write-keys networkName tableName parallelWriters numberOfKeysEach
func doWriteKeys(ips []string, args []string) {
networkName := args[0]
tableName := args[1]
parallelWriters, _ := strconv.Atoi(args[2])
numberOfKeys, _ := strconv.Atoi(args[3])
doneCh := make(chan resultTuple, parallelWriters)
// Enable watch of tables from clients
for i := 0; i < parallelWriters; i++ {
go clientWatchTable(ips[i], servicePort, networkName, tableName, doneCh)
}
waitWriters(parallelWriters, false, doneCh)
// Start parallel writers that will create and delete unique keys
defer close(doneCh)
for i := 0; i < parallelWriters; i++ {
key := "key-" + strconv.Itoa(i) + "-"
log.G(context.TODO()).Infof("Spawn worker: %d on IP:%s", i, ips[i])
go writeKeysNumber(ips[i], servicePort, networkName, tableName, key, numberOfKeys, doneCh)
}
// Sync with all the writers
keyMap := waitWriters(parallelWriters, true, doneCh)
log.G(context.TODO()).Infof("Written a total of %d keys on the cluster", keyMap[totalWrittenKeys])
// check table entries for 2 minutes
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
opTime := checkTable(ctx, ips, servicePort, networkName, tableName, keyMap[totalWrittenKeys], dbTableEntriesNumber)
cancel()
fmt.Fprintf(os.Stderr, "doWriteKeys succeeded in %d msec", opTime)
}
// delete-keys networkName tableName parallelWriters numberOfKeysEach
func doDeleteKeys(ips []string, args []string) {
networkName := args[0]
tableName := args[1]
parallelWriters, _ := strconv.Atoi(args[2])
numberOfKeys, _ := strconv.Atoi(args[3])
doneCh := make(chan resultTuple, parallelWriters)
// Enable watch of tables from clients
for i := 0; i < parallelWriters; i++ {
go clientWatchTable(ips[i], servicePort, networkName, tableName, doneCh)
}
waitWriters(parallelWriters, false, doneCh)
// Start parallel writers that will create and delete unique keys
defer close(doneCh)
for i := 0; i < parallelWriters; i++ {
key := "key-" + strconv.Itoa(i) + "-"
log.G(context.TODO()).Infof("Spawn worker: %d on IP:%s", i, ips[i])
go deleteKeysNumber(ips[i], servicePort, networkName, tableName, key, numberOfKeys, doneCh)
}
// Sync with all the writers
keyMap := waitWriters(parallelWriters, true, doneCh)
log.G(context.TODO()).Infof("Written a total of %d keys on the cluster", keyMap[totalWrittenKeys])
// check table entries for 2 minutes
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
opTime := checkTable(ctx, ips, servicePort, networkName, tableName, 0, dbTableEntriesNumber)
cancel()
fmt.Fprintf(os.Stderr, "doDeletekeys succeeded in %d msec", opTime)
}
// write-delete-unique-keys networkName tableName numParallelWriters writeTimeSec
func doWriteDeleteUniqueKeys(ips []string, args []string) {
networkName := args[0]
tableName := args[1]
parallelWriters, _ := strconv.Atoi(args[2])
writeTimeSec, _ := strconv.Atoi(args[3])
doneCh := make(chan resultTuple, parallelWriters)
// Enable watch of tables from clients
for i := 0; i < parallelWriters; i++ {
go clientWatchTable(ips[i], servicePort, networkName, tableName, doneCh)
}
waitWriters(parallelWriters, false, doneCh)
// Start parallel writers that will create and delete unique keys
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(writeTimeSec)*time.Second)
for i := 0; i < parallelWriters; i++ {
key := "key-" + strconv.Itoa(i) + "-"
log.G(ctx).Infof("Spawn worker: %d on IP:%s", i, ips[i])
go writeDeleteUniqueKeys(ctx, ips[i], servicePort, networkName, tableName, key, doneCh)
}
// Sync with all the writers
keyMap := waitWriters(parallelWriters, true, doneCh)
cancel()
log.G(ctx).Infof("Written a total of %d keys on the cluster", keyMap[totalWrittenKeys])
// check table entries for 2 minutes
ctx, cancel = context.WithTimeout(context.Background(), 2*time.Minute)
opDBTime := checkTable(ctx, ips, servicePort, networkName, tableName, 0, dbTableEntriesNumber)
cancel()
ctx, cancel = context.WithTimeout(context.Background(), 30*time.Second)
opClientTime := checkTable(ctx, ips, servicePort, networkName, tableName, 0, clientTableEntriesNumber)
cancel()
fmt.Fprintf(os.Stderr, "doWriteDeleteUniqueKeys succeeded in %d msec and client %d msec", opDBTime, opClientTime)
}
// write-unique-keys networkName tableName numParallelWriters writeTimeSec
func doWriteUniqueKeys(ips []string, args []string) {
networkName := args[0]
tableName := args[1]
parallelWriters, _ := strconv.Atoi(args[2])
writeTimeSec, _ := strconv.Atoi(args[3])
doneCh := make(chan resultTuple, parallelWriters)
// Enable watch of tables from clients
for i := 0; i < parallelWriters; i++ {
go clientWatchTable(ips[i], servicePort, networkName, tableName, doneCh)
}
waitWriters(parallelWriters, false, doneCh)
// Start parallel writers that will create and delete unique keys
defer close(doneCh)
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(writeTimeSec)*time.Second)
for i := 0; i < parallelWriters; i++ {
key := "key-" + strconv.Itoa(i) + "-"
log.G(ctx).Infof("Spawn worker: %d on IP:%s", i, ips[i])
go writeUniqueKeys(ctx, ips[i], servicePort, networkName, tableName, key, doneCh)
}
// Sync with all the writers
keyMap := waitWriters(parallelWriters, true, doneCh)
cancel()
log.G(ctx).Infof("Written a total of %d keys on the cluster", keyMap[totalWrittenKeys])
// check table entries for 2 minutes
ctx, cancel = context.WithTimeout(context.Background(), 2*time.Minute)
opTime := checkTable(ctx, ips, servicePort, networkName, tableName, keyMap[totalWrittenKeys], dbTableEntriesNumber)
cancel()
fmt.Fprintf(os.Stderr, "doWriteUniqueKeys succeeded in %d msec", opTime)
}
// write-delete-leave-join networkName tableName numParallelWriters writeTimeSec
func doWriteDeleteLeaveJoin(ips []string, args []string) {
networkName := args[0]
tableName := args[1]
parallelWriters, _ := strconv.Atoi(args[2])
writeTimeSec, _ := strconv.Atoi(args[3])
// Start parallel writers that will create and delete unique keys
doneCh := make(chan resultTuple, parallelWriters)
defer close(doneCh)
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(writeTimeSec)*time.Second)
for i := 0; i < parallelWriters; i++ {
key := "key-" + strconv.Itoa(i) + "-"
log.G(ctx).Infof("Spawn worker: %d on IP:%s", i, ips[i])
go writeDeleteLeaveJoin(ctx, ips[i], servicePort, networkName, tableName, key, doneCh)
}
// Sync with all the writers
keyMap := waitWriters(parallelWriters, true, doneCh)
cancel()
log.G(ctx).Infof("Written a total of %d keys on the cluster", keyMap["totalKeys"])
// check table entries for 2 minutes
ctx, cancel = context.WithTimeout(context.Background(), 2*time.Minute)
opTime := checkTable(ctx, ips, servicePort, networkName, tableName, 0, dbTableEntriesNumber)
cancel()
fmt.Fprintf(os.Stderr, "doWriteDeleteLeaveJoin succeeded in %d msec", opTime)
}
// write-delete-wait-leave-join networkName tableName numParallelWriters writeTimeSec
func doWriteDeleteWaitLeaveJoin(ips []string, args []string) {
networkName := args[0]
tableName := args[1]
parallelWriters, _ := strconv.Atoi(args[2])
writeTimeSec, _ := strconv.Atoi(args[3])
// Start parallel writers that will create and delete unique keys
doneCh := make(chan resultTuple, parallelWriters)
defer close(doneCh)
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(writeTimeSec)*time.Second)
for i := 0; i < parallelWriters; i++ {
key := "key-" + strconv.Itoa(i) + "-"
log.G(ctx).Infof("Spawn worker: %d on IP:%s", i, ips[i])
go writeDeleteUniqueKeys(ctx, ips[i], servicePort, networkName, tableName, key, doneCh)
}
// Sync with all the writers
keyMap := waitWriters(parallelWriters, true, doneCh)
cancel()
log.G(ctx).Infof("Written a total of %d keys on the cluster", keyMap[totalWrittenKeys])
// The writers will leave the network
for i := 0; i < parallelWriters; i++ {
log.G(ctx).Infof("worker leaveNetwork: %d on IP:%s", i, ips[i])
go leaveNetwork(ips[i], servicePort, networkName, doneCh)
}
waitWriters(parallelWriters, false, doneCh)
// Give some time
time.Sleep(100 * time.Millisecond)
// The writers will join the network
for i := 0; i < parallelWriters; i++ {
log.G(ctx).Infof("worker joinNetwork: %d on IP:%s", i, ips[i])
go joinNetwork(ips[i], servicePort, networkName, doneCh)
}
waitWriters(parallelWriters, false, doneCh)
// check table entries for 2 minutes
ctx, cancel = context.WithTimeout(context.Background(), 2*time.Minute)
opTime := checkTable(ctx, ips, servicePort, networkName, tableName, 0, dbTableEntriesNumber)
cancel()
fmt.Fprintf(os.Stderr, "doWriteDeleteWaitLeaveJoin succeeded in %d msec", opTime)
}
// write-wait-leave networkName tableName numParallelWriters writeTimeSec
func doWriteWaitLeave(ips []string, args []string) {
networkName := args[0]
tableName := args[1]
parallelWriters, _ := strconv.Atoi(args[2])
writeTimeSec, _ := strconv.Atoi(args[3])
// Start parallel writers that will create and delete unique keys
doneCh := make(chan resultTuple, parallelWriters)
defer close(doneCh)
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(writeTimeSec)*time.Second)
for i := 0; i < parallelWriters; i++ {
key := "key-" + strconv.Itoa(i) + "-"
log.G(ctx).Infof("Spawn worker: %d on IP:%s", i, ips[i])
go writeUniqueKeys(ctx, ips[i], servicePort, networkName, tableName, key, doneCh)
}
// Sync with all the writers
keyMap := waitWriters(parallelWriters, true, doneCh)
cancel()
log.G(ctx).Infof("Written a total of %d keys on the cluster", keyMap[totalWrittenKeys])
// The writers will leave the network
for i := 0; i < parallelWriters; i++ {
log.G(ctx).Infof("worker leaveNetwork: %d on IP:%s", i, ips[i])
go leaveNetwork(ips[i], servicePort, networkName, doneCh)
}
waitWriters(parallelWriters, false, doneCh)
// check table entries for 2 minutes
ctx, cancel = context.WithTimeout(context.Background(), 2*time.Minute)
opTime := checkTable(ctx, ips, servicePort, networkName, tableName, 0, dbTableEntriesNumber)
cancel()
fmt.Fprintf(os.Stderr, "doWriteLeaveJoin succeeded in %d msec", opTime)
}
// write-wait-leave-join networkName tableName numParallelWriters writeTimeSec numParallelLeaver
func doWriteWaitLeaveJoin(ips []string, args []string) {
networkName := args[0]
tableName := args[1]
parallelWriters, _ := strconv.Atoi(args[2])
writeTimeSec, _ := strconv.Atoi(args[3])
parallelLeaver, _ := strconv.Atoi(args[4])
// Start parallel writers that will create and delete unique keys
doneCh := make(chan resultTuple, parallelWriters)
defer close(doneCh)
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(writeTimeSec)*time.Second)
for i := 0; i < parallelWriters; i++ {
key := "key-" + strconv.Itoa(i) + "-"
log.G(ctx).Infof("Spawn worker: %d on IP:%s", i, ips[i])
go writeUniqueKeys(ctx, ips[i], servicePort, networkName, tableName, key, doneCh)
}
// Sync with all the writers
keyMap := waitWriters(parallelWriters, true, doneCh)
cancel()
log.G(ctx).Infof("Written a total of %d keys on the cluster", keyMap[totalWrittenKeys])
keysExpected := keyMap[totalWrittenKeys]
// The Leavers will leave the network
for i := 0; i < parallelLeaver; i++ {
log.G(ctx).Infof("worker leaveNetwork: %d on IP:%s", i, ips[i])
go leaveNetwork(ips[i], servicePort, networkName, doneCh)
// Once a node leave all the keys written previously will be deleted, so the expected keys will consider that as removed
keysExpected -= keyMap[ips[i]]
}
waitWriters(parallelLeaver, false, doneCh)
// Give some time
time.Sleep(100 * time.Millisecond)
// The writers will join the network
for i := 0; i < parallelLeaver; i++ {
log.G(ctx).Infof("worker joinNetwork: %d on IP:%s", i, ips[i])
go joinNetwork(ips[i], servicePort, networkName, doneCh)
}
waitWriters(parallelLeaver, false, doneCh)
// check table entries for 2 minutes
ctx, cancel = context.WithTimeout(context.Background(), 2*time.Minute)
opTime := checkTable(ctx, ips, servicePort, networkName, tableName, keysExpected, dbTableEntriesNumber)
cancel()
fmt.Fprintf(os.Stderr, "doWriteWaitLeaveJoin succeeded in %d msec", opTime)
}
var cmdArgChec = map[string]int{
"debug": 0,
"fail": 0,
"ready": 2,
"join": 2,
"leave": 2,
"join-network": 3,
"leave-network": 3,
"cluster-peers": 5,
"network-peers": 5,
"write-delete-unique-keys": 7,
}
// Client is a client
func Client(args []string) {
log.G(context.TODO()).Infof("[CLIENT] Starting with arguments %v", args)
command := args[0]
if len(args) < cmdArgChec[command] {
log.G(context.TODO()).Fatalf("Command %s requires %d arguments, passed %d, aborting...", command, cmdArgChec[command], len(args))
}
switch command {
case "debug":
time.Sleep(1 * time.Hour)
os.Exit(0)
case "fail":
log.G(context.TODO()).Fatalf("Test error condition with message: error error error") //nolint:dupword
}
serviceName := args[1]
ips, _ := net.LookupHost("tasks." + serviceName)
log.G(context.TODO()).Infof("got the ips %v", ips)
if len(ips) == 0 {
log.G(context.TODO()).Fatalf("Cannot resolve any IP for the service tasks.%s", serviceName)
}
servicePort = args[2]
commandArgs := args[3:]
log.G(context.TODO()).Infof("Executing %s with args:%v", command, commandArgs)
switch command {
case "ready":
doReady(ips)
case "join":
doJoin(ips)
case "leave":
case "cluster-peers":
// cluster-peers maxRetry
doClusterPeers(ips, commandArgs)
case "join-network":
// join-network networkName
doJoinNetwork(ips, commandArgs)
case "leave-network":
// leave-network networkName
doLeaveNetwork(ips, commandArgs)
case "network-peers":
// network-peers networkName expectedNumberPeers maxRetry
doNetworkPeers(ips, commandArgs)
// case "network-stats-entries":
// // network-stats-entries networkName maxRetry
// doNetworkPeers(ips, commandArgs)
case "network-stats-queue":
// network-stats-queue networkName <lt/gt> queueSize
doNetworkStatsQueue(ips, commandArgs)
case "write-keys":
// write-keys networkName tableName parallelWriters numberOfKeysEach
doWriteKeys(ips, commandArgs)
case "delete-keys":
// delete-keys networkName tableName parallelWriters numberOfKeysEach
doDeleteKeys(ips, commandArgs)
case "write-unique-keys":
// write-delete-unique-keys networkName tableName numParallelWriters writeTimeSec
doWriteUniqueKeys(ips, commandArgs)
case "write-delete-unique-keys":
// write-delete-unique-keys networkName tableName numParallelWriters writeTimeSec
doWriteDeleteUniqueKeys(ips, commandArgs)
case "write-delete-leave-join":
// write-delete-leave-join networkName tableName numParallelWriters writeTimeSec
doWriteDeleteLeaveJoin(ips, commandArgs)
case "write-delete-wait-leave-join":
// write-delete-wait-leave-join networkName tableName numParallelWriters writeTimeSec
doWriteDeleteWaitLeaveJoin(ips, commandArgs)
case "write-wait-leave":
// write-wait-leave networkName tableName numParallelWriters writeTimeSec
doWriteWaitLeave(ips, commandArgs)
case "write-wait-leave-join":
// write-wait-leave networkName tableName numParallelWriters writeTimeSec
doWriteWaitLeaveJoin(ips, commandArgs)
default:
log.G(context.TODO()).Fatalf("Command %s not recognized", command)
}
}