mirror of
https://github.com/hibiken/asynq.git
synced 2024-11-10 11:31:58 +08:00
Add ClusterNodes and ClusterKeySlot in Inspector
This commit is contained in:
parent
eb739a0258
commit
a88325cb96
27
inspector.go
27
inspector.go
@ -573,3 +573,30 @@ func (i *Inspector) UnpauseQueue(qname string) error {
|
|||||||
}
|
}
|
||||||
return i.rdb.Unpause(qname)
|
return i.rdb.Unpause(qname)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ClusterKeySlot returns an integer identifying the hash slot the given queue hashes to.
|
||||||
|
func (i *Inspector) ClusterKeySlot(qname string) (int64, error) {
|
||||||
|
return i.rdb.ClusterKeySlot(qname)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ClusterNode describes a node in redis cluster.
|
||||||
|
type ClusterNode struct {
|
||||||
|
// Node ID in the cluster.
|
||||||
|
ID string
|
||||||
|
|
||||||
|
// Address of the node.
|
||||||
|
Addr string
|
||||||
|
}
|
||||||
|
|
||||||
|
// ClusterNode returns a list of nodes the given queue belongs to.
|
||||||
|
func (i *Inspector) ClusterNodes(qname string) ([]ClusterNode, error) {
|
||||||
|
nodes, err := i.rdb.ClusterNodes(qname)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
var res []ClusterNode
|
||||||
|
for _, node := range nodes {
|
||||||
|
res = append(res, ClusterNode{ID: node.ID, Addr: node.Addr})
|
||||||
|
}
|
||||||
|
return res, nil
|
||||||
|
}
|
||||||
|
@ -830,3 +830,27 @@ func (r *RDB) Unpause(qname string) error {
|
|||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ClusterKeySlot returns an integer identifying the hash slot the given queue hashes to.
|
||||||
|
func (r *RDB) ClusterKeySlot(qname string) (int64, error) {
|
||||||
|
key := base.QueueKey(qname)
|
||||||
|
return r.client.ClusterKeySlot(key).Result()
|
||||||
|
}
|
||||||
|
|
||||||
|
// ClusterNodes returns a list of nodes the given queue belongs to.
|
||||||
|
func (r *RDB) ClusterNodes(qname string) ([]redis.ClusterNode, error) {
|
||||||
|
keyslot, err := r.ClusterKeySlot(qname)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
clusterSlots, err := r.client.ClusterSlots().Result()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
for _, slotRange := range clusterSlots {
|
||||||
|
if int64(slotRange.Start) <= keyslot && keyslot <= int64(slotRange.End) {
|
||||||
|
return slotRange.Nodes, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil, fmt.Errorf("nodes not found")
|
||||||
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user