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())