fix: Adding more debug tools to cluster

api/v1/cluster_test.go

@@ -30,6 +30,9 @@ import (
 )
 
 func TestClusterStorage(t *testing.T) {
+	clusterQueryURL := "http://localhost" + TESTPORT + EndpointClusterQuery
+	graphURL := "http://localhost" + TESTPORT + EndpointGraph
+
 	cluster2 := createCluster(2)
 
 	oldGM := api.GM
@@ -46,6 +49,48 @@ func TestClusterStorage(t *testing.T) {
 		api.DDLog = nil
 	}()
 
+	// We should now get back a state
+
+	st, _, res := sendTestRequest(clusterQueryURL, "GET", nil)
+
+	if st != "200 OK" || res != `
+{
+  "failed": null,
+  "members": [
+    "TestClusterMember-0",
+    "localhost:9020"
+  ],
+  "replication": 1,
+  "ts": [
+    "TestClusterMember-0",
+    "1"
+  ],
+  "tsold": [
+    "",
+    "0"
+  ]
+}`[1:] {
+		t.Error("Unexpected response:", st, res)
+		return
+	}
+
+	// Insert some data
+
+	st, _, res = sendTestRequest(graphURL+"i41health/n", "POST", []byte(`
+[{
+	"key":"3",
+	"kind":"Upload",
+	"parcel": "12345"
+}]
+`[1:]))
+
+	cluster.WaitForTransfer()
+	fmt.Println(cluster.DumpMemoryClusterLayout("i41healthUpload.nodes"))
+
+	n, err := api.GM.FetchNode("i41health", "3", "Upload")
+
+	fmt.Println("res:", n, err)
+
 }
 
 func TestClusterQuery(t *testing.T) {
@@ -378,6 +423,8 @@ func createCluster(n int) []*cluster.DistributedStorage {
 	var mgs []*graphstorage.MemoryGraphStorage
 	var cs []*cluster.DistributedStorage
 
+	cluster.ClearMSMap()
+
 	for i := 0; i < n; i++ {
 		mgs = append(mgs, graphstorage.NewMemoryGraphStorage(fmt.Sprintf("mgs%v", i+1)).(*graphstorage.MemoryGraphStorage))
 	}

cluster/debug.go

@@ -0,0 +1,46 @@
+package cluster
+
+import (
+	"bytes"
+	"fmt"
+	"time"
+)
+
+/*
+msmap is a map of all know  memory-only memberStorages.
+*/
+var msmap = make(map[*DistributedStorage]*memberStorage)
+
+/*
+Clear the current map of known memory-only memberStorages.
+*/
+func ClearMSMap() {
+	msmap = make(map[*DistributedStorage]*memberStorage)
+}
+
+/*
+DumpMemoryClusterLayout returns the current storage layout in a memory-only cluster
+for a given storage manager (e.g. mainPerson.nodes for Person nodes of partition main).
+*/
+func DumpMemoryClusterLayout(smname string) string {
+	buf := new(bytes.Buffer)
+
+	for _, ms := range msmap {
+		buf.WriteString(fmt.Sprintf("MemoryStorage: %s\n", ms.gs.Name()))
+		buf.WriteString(ms.dump(smname))
+	}
+
+	return buf.String()
+}
+
+/*
+WaitForTransfer waits for the datatransfer to happen.
+*/
+func WaitForTransfer() {
+	for _, ms := range msmap {
+		ms.transferWorker()
+		for ms.transferRunning {
+			time.Sleep(time.Millisecond)
+		}
+	}
+}

cluster/distributedstorage.go

@@ -100,7 +100,12 @@ that the cluster has only one member.
 func NewDistributedStorage(gs graphstorage.Storage, config map[string]interface{},
 	stateInfo manager.StateInfo) (*DistributedStorage, error) {
 
-	ds, _, err := newDistributedAndMemberStorage(gs, config, stateInfo)
+	ds, ms, err := newDistributedAndMemberStorage(gs, config, stateInfo)
+
+	if _, ok := gs.(*graphstorage.MemoryGraphStorage); ok {
+		msmap[ds] = ms // Keep track of memory storages for debugging
+	}
+
 	return ds, err
 }