Integrate flaky, slow, stop and partition test cases (#2751)

* Must wait for the cluster to converge before starting tests.
* Blockade tool works nicely. Tested 9 combinations of partitions, all work OK.
* Testing all 4 edge cases: Flaky, slow, stop, partition

Author: manishrjain

contrib/blockade/blockade.yml

@@ -17,33 +17,33 @@ containers:
     expose:
       - 5080
       - 6080
-    command: /gobin/dgraph zero --my=zero1:5080 --replicas 3 --idx 1 --bindall --expose_trace --profile_mode block --block_rate 10 --logtostderr -v=2
+    command: /gobin/dgraph zero --my=zero1:5080 --replicas 3 --idx 1 --bindall --expose_trace --logtostderr -v=2
     volumes: {"/home/mrjn/go/bin": "/gobin"}
 
   zero2:
     image: dgraph/dgraph:latest
     hostname: "zero2"
     container_name: "zero2"
     ports:
-      - 5081
-      - 6081
+      - 5082
+      - 6082
     expose:
-      - 5081
-      - 6081
-    command: /gobin/dgraph zero -o 1 --my=zero2:5081 --replicas 3 --peer=zero1:5080 --idx 2 --bindall --expose_trace --profile_mode block --block_rate 10 --logtostderr -v=2
+      - 5082
+      - 6082
+    command: /gobin/dgraph zero -o 2 --my=zero2:5082 --replicas 3 --peer=zero1:5080 --idx 2 --bindall --expose_trace --logtostderr -v=2
     volumes: {"/home/mrjn/go/bin": "/gobin"}
 
   zero3:
     image: dgraph/dgraph:latest
     hostname: "zero3"
     container_name: "zero3"
     ports:
-      - 5082
-      - 6082
+      - 5083
+      - 6083
     expose:
-      - 5082
-      - 6082
-    command: /gobin/dgraph zero -o 2 --my=zero3:5082 --replicas 3 --peer=zero1:5080 --idx 3 --bindall --expose_trace --profile_mode block --block_rate 10 --logtostderr -v=2
+      - 5083
+      - 6083
+    command: /gobin/dgraph zero -o 3 --my=zero3:5083 --replicas 3 --peer=zero1:5080 --idx 3 --bindall --expose_trace --logtostderr -v=2
     volumes:
       "/home/mrjn/go/bin": "/gobin"
 

contrib/blockade/main.go

@@ -3,9 +3,11 @@ package main
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"log"
 	"math/rand"
+	"os"
 	"os/exec"
 	"strings"
 	"time"
@@ -20,10 +22,10 @@ func run(ctx context.Context, command string) error {
 	cmd.Stdout = &out
 	cmd.Stderr = &out
 	if err := cmd.Run(); err != nil {
-		fmt.Printf("ERROR. Command %s. Error: %v. Output:\n%s\n", command, err, out.String())
+		fmt.Printf("ERROR. Command %q. Error: %v. Output:\n%s\n", command, err, out.String())
 		return err
 	}
-	fmt.Printf("Command %s. Output:\n%s\n", command, out.String())
+	fmt.Printf("Command %q. Output:\n%s\n", command, out.String())
 	return nil
 }
 
@@ -49,65 +51,138 @@ func increment(atLeast int) error {
 			return err
 		}
 	}
-	fmt.Printf("Time taken to converge %d: %s\n",
-		atLeast, time.Since(start).Round(time.Millisecond))
+	dur := time.Since(start).Round(time.Millisecond)
+	fmt.Printf("\n===> TIME taken to converge %d alphas: %s\n\n", atLeast, dur)
 	return nil
 }
 
-func testPartitions() error {
+func getStatus(zero string) error {
+	cmd := exec.Command("http", "GET", fmt.Sprintf("%s/state", zero))
+	var out bytes.Buffer
+	cmd.Stdout = &out
+	cmd.Stderr = &out
+	if err := cmd.Run(); err != nil {
+		fmt.Printf("ERROR. Status at %s. Error: %v. Output:\n%s\n", zero, err, out.String())
+		return err
+	}
+	output := out.String()
+	if strings.Contains(output, "errors") {
+		fmt.Printf("ERROR. Status at %s. Output:\n%s\n", zero, output)
+		return fmt.Errorf(output)
+	}
+	var m map[string]interface{}
+	if err := json.Unmarshal([]byte(output), &m); err != nil {
+		return err
+	}
+	pretty, err := json.MarshalIndent(m, "", "  ")
+	if err != nil {
+		return err
+	}
+	fmt.Printf("Status at %s:\n%s\n", zero, pretty)
+	return nil
+}
+
+func testCommon(remove, join string, minAlphasUp int) error {
 	var nodes []string
 	for i := 1; i <= 3; i++ {
 		for j := 1; j <= 3; j++ {
 			nodes = append(nodes, fmt.Sprintf("zero%d dg%d", i, j))
 		}
 	}
 
-	fmt.Printf("Nodes: %v\n", nodes)
+	fmt.Printf("Nodes: %+v\n", nodes)
 	for _, node := range nodes {
-		// First partition.
-		if err := run(ctxb, "http GET localhost:6080/state"); err != nil {
+		if err := getStatus("localhost:6080"); err != nil {
 			return err
 		}
-		fmt.Printf("\n==> Partitioning NODE: %s\n", node)
-		if err := partition(node); err != nil {
+		fmt.Printf("\n==> Remove cmd %q on NODES: %s\n", remove, node)
+		if err := run(ctxb, remove+" "+node); err != nil {
 			return err
 		}
 		if err := run(ctxb, "blockade status"); err != nil {
 			return err
 		}
-		if err := increment(2); err != nil {
+		if err := increment(minAlphasUp); err != nil {
 			return err
 		}
 		// Then join.
-		if err := run(ctxb, "blockade join"); err != nil {
+		if err := run(ctxb, join); err != nil {
 			return err
 		}
 		if err := increment(3); err != nil {
 			return err
 		}
 	}
-	fmt.Println("testPartitions: OK")
 	return nil
 }
 
-func main() {
-	rand.Seed(time.Now().UnixNano())
-	fmt.Println("Starting blockade")
-	if err := run(ctxb, "blockade up"); err != nil {
-		log.Fatal(err)
+func waitForHealthy() error {
+	for _, zero := range []string{"localhost:6080", "localhost:6082", "localhost:6083"} {
+		if err := getStatus(zero); err != nil {
+			return err
+		}
+	}
+	for _, alpha := range []string{"localhost:9180", "localhost:9182", "localhost:9183"} {
+		if err := run(ctxb, "increment --addr="+alpha); err != nil {
+			return err
+		}
 	}
+	return nil
+}
 
+func runTests() error {
 	defer func() {
 		if err := run(ctxb, "blockade destroy"); err != nil {
 			log.Fatalf("While destroying: %v", err)
 		}
 	}()
-	if err := run(ctxb,
-		"increment --addr=localhost:9180"); err != nil {
-		fmt.Printf("Error during increment: %v\n", err)
+
+	for {
+		if err := waitForHealthy(); err != nil {
+			fmt.Printf("Error while waitForHealthy: %v\n.", err)
+			time.Sleep(5 * time.Second)
+			fmt.Println("Retrying...")
+		} else {
+			break
+		}
+	}
+
+	// Setting flaky --all just does not converge. Too many network interruptions.
+	if err := testCommon("blockade flaky", "blockade fast --all", 3); err != nil {
+		fmt.Printf("Error testFlaky: %v\n", err)
+		return err
+	}
+	fmt.Println("===> Flaky TEST: OK")
+
+	if err := testCommon("blockade slow", "blockade fast --all", 3); err != nil {
+		fmt.Printf("Error testSlow: %v\n", err)
+		return err
+	}
+	fmt.Println("===> Slow TEST: OK")
+
+	if err := testCommon("blockade stop", "blockade start --all", 2); err != nil {
+		fmt.Printf("Error testRestart: %v\n", err)
+		return err
 	}
+	fmt.Println("===> Restart TEST: OK")
 
-	if err := testPartitions(); err != nil {
+	if err := testCommon("blockade partition", "blockade join", 2); err != nil {
 		fmt.Printf("Error testPartitions: %v\n", err)
+		return err
+	}
+	fmt.Println("===> Partition TEST: OK")
+
+	return nil
+}
+
+func main() {
+	rand.Seed(time.Now().UnixNano())
+	fmt.Println("Starting blockade")
+	if err := run(ctxb, "blockade up"); err != nil {
+		log.Fatal(err)
+	}
+	if err := runTests(); err != nil {
+		os.Exit(1)
 	}
+	fmt.Println("Blockade tests: OK")
 }

contrib/integration/increment/main.go

@@ -117,13 +117,13 @@ func main() {
 
 	for *num > 0 {
 		cnt, err := process(dg, *ro)
+		now := time.Now().UTC().Format("0102 03:04:05.999")
 		if err != nil {
-			fmt.Printf("While trying to process counter: %v. Retrying...\n", err)
+			fmt.Printf("%-17s While trying to process counter: %v. Retrying...\n", now, err)
 			time.Sleep(time.Second)
 			continue
 		}
-		fmt.Printf("%-17s Counter VAL: %d   [ Ts: %d ]\n",
-			time.Now().UTC().Format("0102 03:04:05.999"), cnt.Val, cnt.startTs)
+		fmt.Printf("%-17s Counter VAL: %d   [ Ts: %d ]\n", now, cnt.Val, cnt.startTs)
 		*num -= 1
 		time.Sleep(waitDur)
 	}