From 98cd2dc441dccdd76d7c59ff36e71b8b683365a6 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Wed, 7 Oct 2020 14:10:32 -0500 Subject: [PATCH 1/5] WIP shard distribution endpoint --- api.go | 35 +++++++++++++++++++++++++ cluster.go | 70 ++++++++++++++++++++++++++++++++++++++++++++++--- http/handler.go | 10 +++++++ 3 files changed, 111 insertions(+), 4 deletions(-) diff --git a/api.go b/api.go index 3005bafbe..76a3543e4 100644 --- a/api.go +++ b/api.go @@ -1597,6 +1597,41 @@ func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64) error { return ef.Import(qcx, existenceRowIDs, columnIDs, nil) } +func (api *API) ShardDistribution(ctx context.Context) map[string]interface{} { + distByIndex := make(map[string]interface{}) + maxShards := api.MaxShards(ctx) + + for idx := range api.holder.indexes { + calculatedMaxShard := uint64(0) + if mx, ok := maxShards[idx]; ok { + calculatedMaxShard = mx + } + _, shards := api.cluster.shardDistributionByIndex(idx, calculatedMaxShard) + distByIndex[idx] = shards + } + + return distByIndex +} + +// ShardDistributionByIndex returns a slice of shards per node. +func (api *API) ShardDistributionByIndex(ctx context.Context, index string, provideMaxShard bool, maxShard uint64) ([]Node, [][]uint64) { + span, _ := tracing.StartSpanFromContext(ctx, "API.ShardDistributionByIndex") + defer span.Finish() + + calculatedMaxShard := uint64(0) + if provideMaxShard { + calculatedMaxShard = maxShard + } else { + // Get max shard from cluster. + maxShards := api.MaxShards(ctx) + if mx, ok := maxShards[index]; ok { + calculatedMaxShard = mx + } + } + + return api.cluster.shardDistributionByIndex(index, calculatedMaxShard) +} + // MaxShards returns the maximum shard number for each index in a map. // TODO (2.0): This method has been deprecated. Instead, use // AvailableShardsByIndex. diff --git a/cluster.go b/cluster.go index 23c86ac7b..cc8bcb826 100644 --- a/cluster.go +++ b/cluster.go @@ -978,12 +978,74 @@ func (c *cluster) translationNodes(to *cluster) (map[string][]*translationResize return m, nil } -// shardPartition returns the shard-partition that a shard belongs to. -// NOTE: this is DIFFERENT from the key-partition -func (c *cluster) shardToShardPartition(index string, shard uint64) int { - return shardToShardPartition(index, shard, c.partitionN) +// shardDistributionByIndex returns a slice of shards per node for an index, up to maxShard. +func (c *cluster) shardDistributionByIndex(index string, maxShard uint64) ([]Node, [][]uint64) { + m := make(map[Node][]uint64) + + for i := range c.nodes { + m[*c.nodes[i]] = []uint64{} + } + + c.mu.RLock() + defer c.mu.RUnlock() + + for shard := uint64(0); shard <= maxShard; shard++ { + for _, node := range c.shardNodes(index, shard) { + m[*node] = append(m[*node], shard) + } + } + + n := make([]Node, len(c.nodes)) + s := make([][]uint64, len(c.nodes)) + + for i := range c.nodes { + n[i] = *c.nodes[i] + s[i] = m[*c.nodes[i]] + } + + return n, s } +// For specified index, return an object like +/* +{ + "7aa98e81-0b53-43e4-9f98-c77d7fc2371a": { + "primary-shards": [], + "replica-shards": [] + }, + "d4a7b7ff-529f-4d28-8d95-48d307751775": { + "primary-shards": [], + "replica-shards": [] + } + ... +} +*/ +func (c *cluster) shardDistributionByIndex2(index string, maxShard uint64) map[string]interface{} { + dist := make(map[string]interface{}) + + c.mu.RLock() + defer c.mu.RUnlock() + + for i := range c.nodes { + nodeDist := make(map[string][]uint64) + primaries := make([]uint64) + replicas := make([]uint64) + for shard := uint64(0); shard <= maxShard; shard++ { + + } + dist[node.ID]["primary-shards"] = primaries + dist[node.ID]["replica-shards"] = replicas + } + return dist +} + +// shardPartition returns the partition that a shard belongs to. +func (c *cluster) shardPartition(index string, shard uint64) int { + return shardPartition(index, shard, c.partitionN) +} + +// shardPartition returns the shard-partition that a shard belongs to. +// NOTE: this is DIFFERENT from the key-partition func shardToShardPartition(index string, shard uint64, partitionN int) int { var buf [8]byte binary.BigEndian.PutUint64(buf[:], shard) diff --git a/http/handler.go b/http/handler.go index bb8cdf9f6..3d916fbcc 100644 --- a/http/handler.go +++ b/http/handler.go @@ -393,6 +393,7 @@ func newRouter(handler *Handler) http.Handler { router.HandleFunc("/ui/usage", handler.handleGetUsage).Methods("GET").Name("GetUsage") router.HandleFunc("/ui/transaction", handler.handleGetTransactionList).Methods("GET").Name("GetTransactionList") router.HandleFunc("/ui/transaction/", handler.handleGetTransactionList).Methods("GET").Name("GetTransactionList") + router.HandleFunc("/ui/shard-distribution", handler.handleGetShardDistribution).Methods("GET").Name("GetShardDistribution") // /internal endpoints are for internal use only; they may change at any time. // DO NOT rely on these for external applications! @@ -684,6 +685,15 @@ func (h *Handler) handleGetUsage(w http.ResponseWriter, r *http.Request) { } } +// handleGetUsage handles GET /ui/shard-distribution requests. +func (h *Handler) handleGetShardDistribution(w http.ResponseWriter, r *http.Request) { + dist := h.api.ShardDistribution(r.Context()) + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(dist); err != nil { + h.logger.Printf("write status response error: %s", err) + } +} + // handleGetStatus handles GET /status requests. func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) { if !validHeaderAcceptJSON(r.Header) { From 2a902b3a7b9d327ad490efbe99fcd4535d97d7a1 Mon Sep 17 00:00:00 2001 From: alan Date: Fri, 9 Oct 2020 02:17:06 -0500 Subject: [PATCH 2/5] Finish basic shard-distribution endpoint --- api.go | 31 ++++++------------- cluster.go | 81 ++++++++++++++++--------------------------------- http/handler.go | 8 +++++ 3 files changed, 44 insertions(+), 76 deletions(-) diff --git a/api.go b/api.go index 76a3543e4..05fbb86d7 100644 --- a/api.go +++ b/api.go @@ -1597,6 +1597,10 @@ func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64) error { return ef.Import(qcx, existenceRowIDs, columnIDs, nil) } +// ShardDistribution returns an object representing the distribution of shards +// across nodes for each index, distinguishing between primary and replica. +// The structure of this information is [indexName][nodeID][primaryOrReplica]uint64. +// This function supports a view in the UI, and func (api *API) ShardDistribution(ctx context.Context) map[string]interface{} { distByIndex := make(map[string]interface{}) maxShards := api.MaxShards(ctx) @@ -1606,32 +1610,13 @@ func (api *API) ShardDistribution(ctx context.Context) map[string]interface{} { if mx, ok := maxShards[idx]; ok { calculatedMaxShard = mx } - _, shards := api.cluster.shardDistributionByIndex(idx, calculatedMaxShard) - distByIndex[idx] = shards + dist := api.cluster.shardDistributionByIndex(idx, calculatedMaxShard) + distByIndex[idx] = dist } return distByIndex } -// ShardDistributionByIndex returns a slice of shards per node. -func (api *API) ShardDistributionByIndex(ctx context.Context, index string, provideMaxShard bool, maxShard uint64) ([]Node, [][]uint64) { - span, _ := tracing.StartSpanFromContext(ctx, "API.ShardDistributionByIndex") - defer span.Finish() - - calculatedMaxShard := uint64(0) - if provideMaxShard { - calculatedMaxShard = maxShard - } else { - // Get max shard from cluster. - maxShards := api.MaxShards(ctx) - if mx, ok := maxShards[index]; ok { - calculatedMaxShard = mx - } - } - - return api.cluster.shardDistributionByIndex(index, calculatedMaxShard) -} - // MaxShards returns the maximum shard number for each index in a map. // TODO (2.0): This method has been deprecated. Instead, use // AvailableShardsByIndex. @@ -1807,6 +1792,8 @@ func (api *API) Info() serverInfo { Memory: mem, TxSrc: api.holder.txf.TxType(), ReplicaN: api.cluster.ReplicaN, + ShardHash: api.cluster.Hasher.Name(), + KeyHash: api.cluster.Topology.Hasher.Name(), } } @@ -2054,6 +2041,8 @@ func (api *API) TranslateFieldDB(ctx context.Context, indexName, fieldName strin type serverInfo struct { ShardWidth uint64 `json:"shardWidth"` ReplicaN int `json:"replicaN"` + ShardHash string `json:"shardHash"` + KeyHash string `json:"keyHash"` Memory uint64 `json:"memory"` CPUType string `json:"cpuType"` CPUPhysicalCores int `json:"cpuPhysicalCores"` diff --git a/cluster.go b/cluster.go index cc8bcb826..dd1ae1457 100644 --- a/cluster.go +++ b/cluster.go @@ -978,74 +978,39 @@ func (c *cluster) translationNodes(to *cluster) (map[string][]*translationResize return m, nil } -// shardDistributionByIndex returns a slice of shards per node for an index, up to maxShard. -func (c *cluster) shardDistributionByIndex(index string, maxShard uint64) ([]Node, [][]uint64) { - m := make(map[Node][]uint64) +// shardDistributionByIndex returns a map of [nodeID][primaryOrReplica][]uint64, +// where the int slices are lists of shards. +func (c *cluster) shardDistributionByIndex(index string, maxShard uint64) map[string]map[string][]uint64 { + dist := make(map[string]map[string][]uint64) - for i := range c.nodes { - m[*c.nodes[i]] = []uint64{} + for _, node := range c.nodes { + nodeDist := make(map[string][]uint64) + nodeDist["primary-shards"] = make([]uint64, 0) + nodeDist["replica-shards"] = make([]uint64, 0) + dist[node.ID] = nodeDist } c.mu.RLock() defer c.mu.RUnlock() for shard := uint64(0); shard <= maxShard; shard++ { - for _, node := range c.shardNodes(index, shard) { - m[*node] = append(m[*node], shard) + p := c.shardToShardPartition(index, shard) + nodes := c.partitionNodes(p) + dist[nodes[0].ID]["primary-shards"] = append(dist[nodes[0].ID]["primary-shards"], shard) + for k := 1; k < len(nodes); k++ { + dist[nodes[k].ID]["replica-shards"] = append(dist[nodes[k].ID]["replica-shards"], shard) } } - n := make([]Node, len(c.nodes)) - s := make([][]uint64, len(c.nodes)) - - for i := range c.nodes { - n[i] = *c.nodes[i] - s[i] = m[*c.nodes[i]] - } - - return n, s -} - -// For specified index, return an object like -/* -{ - "7aa98e81-0b53-43e4-9f98-c77d7fc2371a": { - "primary-shards": [], - "replica-shards": [] - }, - "d4a7b7ff-529f-4d28-8d95-48d307751775": { - "primary-shards": [], - "replica-shards": [] - } - ... -} -*/ -func (c *cluster) shardDistributionByIndex2(index string, maxShard uint64) map[string]interface{} { - dist := make(map[string]interface{}) - - c.mu.RLock() - defer c.mu.RUnlock() - - for i := range c.nodes { - nodeDist := make(map[string][]uint64) - primaries := make([]uint64) - replicas := make([]uint64) - for shard := uint64(0); shard <= maxShard; shard++ { - - } - dist[node.ID]["primary-shards"] = primaries - dist[node.ID]["replica-shards"] = replicas - } return dist } -// shardPartition returns the partition that a shard belongs to. -func (c *cluster) shardPartition(index string, shard uint64) int { - return shardPartition(index, shard, c.partitionN) -} - // shardPartition returns the shard-partition that a shard belongs to. // NOTE: this is DIFFERENT from the key-partition +func (c *cluster) shardToShardPartition(index string, shard uint64) int { + return shardToShardPartition(index, shard, c.partitionN) +} + func shardToShardPartition(index string, shard uint64, partitionN int) int { var buf [8]byte binary.BigEndian.PutUint64(buf[:], shard) @@ -1255,9 +1220,10 @@ func (c *cluster) containsShards(index string, availableShards *roaring.Bitmap, type Hasher interface { // Hashes the key into a number between [0,N). Hash(key uint64, n int) int + Name() string } -// jmphasher represents an implementation of jmphash. Implements Hasher. +// Jmphasher represents an implementation of jmphash. Implements Hasher. type Jmphasher struct{} // Hash returns the integer hash for the given key. @@ -1271,6 +1237,11 @@ func (h *Jmphasher) Hash(key uint64, n int) int { return int(b) } +// Name returns the name of this hash. +func (h *Jmphasher) Name() string { + return "jump-hash" +} + func (c *cluster) setup() error { // Cluster always comes up in state STARTING until cluster membership is determined. c.state = ClusterStateStarting @@ -1967,7 +1938,7 @@ func (n nodeIDs) Len() int { return len(n) } func (n nodeIDs) Swap(i, j int) { n[i], n[j] = n[j], n[i] } func (n nodeIDs) Less(i, j int) bool { return n[i] < n[j] } -// ContainsID returns true if idi matches one of the nodesets's IDs. +// ContainsID returns true if id matches one of the nodesets's IDs. func (n nodeIDs) ContainsID(id string) bool { for _, nid := range n { if nid == id { diff --git a/http/handler.go b/http/handler.go index 3d916fbcc..1f30c1db2 100644 --- a/http/handler.go +++ b/http/handler.go @@ -685,6 +685,14 @@ func (h *Handler) handleGetUsage(w http.ResponseWriter, r *http.Request) { } } +type getUsageResponse struct { + Disk diskUsage `json:"bytesOnDisk"` +} +type diskUsage struct { + Total int64 `json:"total"` + Indexes map[string]int64 `json:"indexes"` +} + // handleGetUsage handles GET /ui/shard-distribution requests. func (h *Handler) handleGetShardDistribution(w http.ResponseWriter, r *http.Request) { dist := h.api.ShardDistribution(r.Context()) From 510902625ec4ad6f9a3247c64551e3a36e413ce0 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Fri, 9 Oct 2020 03:59:19 -0500 Subject: [PATCH 3/5] Update test hasher implementations --- api_test.go | 2 ++ test/cluster.go | 2 ++ utils_internal_test.go | 2 ++ 3 files changed, 6 insertions(+) diff --git a/api_test.go b/api_test.go index 016778302..3f6369476 100644 --- a/api_test.go +++ b/api_test.go @@ -490,6 +490,8 @@ func (*offsetModHasher) Hash(key uint64, n int) int { return int(key+1) % n } +func (*offsetModHasher) Name() string { return "mod" } + func TestAPI_ClearFlagForImportAndImportValues(t *testing.T) { c := test.MustRunCluster(t, 1, []server.CommandOption{ diff --git a/test/cluster.go b/test/cluster.go index c71e3e68f..5dfb4edfb 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -34,6 +34,8 @@ type ModHasher struct{} func (*ModHasher) Hash(key uint64, n int) int { return int(key) % n } +func (*ModHasher) Name() string { return "mod" } + // Cluster represents a Pilosa cluster (multiple Command instances) type Cluster struct { Nodes []*Command diff --git a/utils_internal_test.go b/utils_internal_test.go index 195c53cef..084389532 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -110,6 +110,8 @@ func NewTestModHasher() *TestModHasher { return &TestModHasher{} } func (*TestModHasher) Hash(key uint64, n int) int { return int(key) % n } +func (*TestModHasher) Name() string { return "mod" } + // ClusterCluster represents a cluster of test nodes, each of which // has a Cluster. // ClusterCluster implements Broadcaster interface. From ab502a0f8daf6e6d1314f1e7829f760917daa059 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Fri, 9 Oct 2020 16:35:18 -0500 Subject: [PATCH 4/5] Add basic test --- api.go | 2 +- server/handler_test.go | 26 ++++++++++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/api.go b/api.go index 05fbb86d7..e0d7d6891 100644 --- a/api.go +++ b/api.go @@ -1600,7 +1600,7 @@ func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64) error { // ShardDistribution returns an object representing the distribution of shards // across nodes for each index, distinguishing between primary and replica. // The structure of this information is [indexName][nodeID][primaryOrReplica]uint64. -// This function supports a view in the UI, and +// This function supports a view in the UI. func (api *API) ShardDistribution(ctx context.Context) map[string]interface{} { distByIndex := make(map[string]interface{}) maxShards := api.MaxShards(ctx) diff --git a/server/handler_test.go b/server/handler_test.go index 4096dc970..156817f2e 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -400,6 +400,32 @@ func TestHandler_Endpoints(t *testing.T) { } }) + t.Run("UI/shard-distribution", func(t *testing.T) { + // This tests the response structure, not the shard distribution. + w := httptest.NewRecorder() + h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/ui/shard-distribution", nil)) + if w.Code != gohttp.StatusOK { + t.Fatalf("unexpected status code: %d", w.Code) + } + + ret := mustJSONDecode(t, w.Body) + + for indexName := range ret { + indexData := ret[indexName].(map[string]interface{}) + for nodeName := range indexData { + nodeData := indexData[nodeName].(map[string]interface{}) + _, hasPrimary := nodeData["primary-shards"] + _, hasReplica := nodeData["replica-shards"] + + responseOK := hasPrimary && hasReplica + if !responseOK { + t.Fatalf("unexpected response structure") + } + } + + } + }) + t.Run("Metrics", func(t *testing.T) { w := httptest.NewRecorder() h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/metrics", nil)) From 94e82aea864d7493c75659683c26faefa37c8754 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Wed, 14 Oct 2020 18:45:53 -0500 Subject: [PATCH 5/5] Consider available shards --- api.go | 9 ++------- cluster.go | 9 ++++++--- http/handler.go | 8 -------- 3 files changed, 8 insertions(+), 18 deletions(-) diff --git a/api.go b/api.go index e0d7d6891..31813ae65 100644 --- a/api.go +++ b/api.go @@ -1599,18 +1599,13 @@ func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64) error { // ShardDistribution returns an object representing the distribution of shards // across nodes for each index, distinguishing between primary and replica. -// The structure of this information is [indexName][nodeID][primaryOrReplica]uint64. +// The structure of this information is [indexName][nodeID][primaryOrReplica][]uint64. // This function supports a view in the UI. func (api *API) ShardDistribution(ctx context.Context) map[string]interface{} { distByIndex := make(map[string]interface{}) - maxShards := api.MaxShards(ctx) for idx := range api.holder.indexes { - calculatedMaxShard := uint64(0) - if mx, ok := maxShards[idx]; ok { - calculatedMaxShard = mx - } - dist := api.cluster.shardDistributionByIndex(idx, calculatedMaxShard) + dist := api.cluster.shardDistributionByIndex(idx) distByIndex[idx] = dist } diff --git a/cluster.go b/cluster.go index dd1ae1457..20e10cd1e 100644 --- a/cluster.go +++ b/cluster.go @@ -980,7 +980,7 @@ func (c *cluster) translationNodes(to *cluster) (map[string][]*translationResize // shardDistributionByIndex returns a map of [nodeID][primaryOrReplica][]uint64, // where the int slices are lists of shards. -func (c *cluster) shardDistributionByIndex(index string, maxShard uint64) map[string]map[string][]uint64 { +func (c *cluster) shardDistributionByIndex(indexName string) map[string]map[string][]uint64 { dist := make(map[string]map[string][]uint64) for _, node := range c.nodes { @@ -990,11 +990,14 @@ func (c *cluster) shardDistributionByIndex(index string, maxShard uint64) map[st dist[node.ID] = nodeDist } + index := c.holder.Index(indexName) + available := index.AvailableShards(includeRemote).Slice() + c.mu.RLock() defer c.mu.RUnlock() - for shard := uint64(0); shard <= maxShard; shard++ { - p := c.shardToShardPartition(index, shard) + for _, shard := range available { + p := c.shardToShardPartition(indexName, shard) nodes := c.partitionNodes(p) dist[nodes[0].ID]["primary-shards"] = append(dist[nodes[0].ID]["primary-shards"], shard) for k := 1; k < len(nodes); k++ { diff --git a/http/handler.go b/http/handler.go index 1f30c1db2..3d916fbcc 100644 --- a/http/handler.go +++ b/http/handler.go @@ -685,14 +685,6 @@ func (h *Handler) handleGetUsage(w http.ResponseWriter, r *http.Request) { } } -type getUsageResponse struct { - Disk diskUsage `json:"bytesOnDisk"` -} -type diskUsage struct { - Total int64 `json:"total"` - Indexes map[string]int64 `json:"indexes"` -} - // handleGetUsage handles GET /ui/shard-distribution requests. func (h *Handler) handleGetShardDistribution(w http.ResponseWriter, r *http.Request) { dist := h.api.ShardDistribution(r.Context())