mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 17:15:56 +00:00
Implement snap := ClusterSnapshot()
Below is the list of instance of `ClusterSnapshot()` in the latest `with-etcd` code. Some of these may not yet exist in the `disco` branch, but this commit is implementing any that currently apply. ========================== Done: ========================== index.go 930: snap := NewClusterSnapshot(topo, topo.Hasher, topo.ReplicaN) 1072: snap := NewClusterSnapshot(topo, topo.Hasher, topo.ReplicaN) cmd/pilosa-fsck/fsck.go 786: snap := pilosa.NewClusterSnapshot(cfg.topo, cfg.topo.Hasher, cfg.topo.ReplicaN) boltdb/translate.go 558: snap := pilosa.NewClusterSnapshot(topo, topo.Hasher, topo.ReplicaN) 1264: snap := pilosa.NewClusterSnapshot(topo, topo.Hasher, topo.ReplicaN) fragment.go 3448: snap := NewClusterSnapshot(s.Cluster.noder, s.Cluster.Hasher, s.Cluster.ReplicaN) 3568: snap := NewClusterSnapshot(s.Cluster.noder, s.Cluster.Hasher, s.Cluster.ReplicaN) 3620: snap := NewClusterSnapshot(s.Cluster.noder, s.Cluster.Hasher, s.Cluster.ReplicaN) ========================== Remaining: ========================== cluster.go 371: snap := NewClusterSnapshot(NewLocalNoder(nodes), c.Hasher, c.ReplicaN) 474: snap := NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN) 639: fSnap := NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN) 640: toSnap := NewClusterSnapshot(to.noder, c.Hasher, to.ReplicaN) 703: snap := NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN) 1475: snap := NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN) 1502: snap := NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN) 1941: snap := NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN) 1986: snap := NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN) 2049: snap := NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN) 2126: snap := NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN) api.go 475: snap := NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) 604: snap := NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) 690: snap := NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) 1684: snap := NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) 1946: snap := NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) executor.go 3781: snap := NewClusterSnapshot(e.Cluster.noder, e.Cluster.Hasher, e.Cluster.ReplicaN) 4157: snap := NewClusterSnapshot(e.Cluster.noder, e.Cluster.Hasher, e.Cluster.ReplicaN) 4200: snap := NewClusterSnapshot(e.Cluster.noder, e.Cluster.Hasher, e.Cluster.ReplicaN) 4243: snap := NewClusterSnapshot(e.Cluster.noder, e.Cluster.Hasher, e.Cluster.ReplicaN) 4517: snap := NewClusterSnapshot(NewLocalNoder(e.Cluster.Nodes()), e.Cluster.Hasher, e.Cluster.ReplicaN) holder.go 1465: snap := NewClusterSnapshot(s.Cluster.noder, s.Cluster.Hasher, s.Cluster.ReplicaN) 1668: snap := NewClusterSnapshot(s.Cluster.noder, s.Cluster.Hasher, s.Cluster.ReplicaN) 1889: snap := NewClusterSnapshot(s.Cluster.noder, s.Cluster.Hasher, s.Cluster.ReplicaN) 1963: snap := NewClusterSnapshot(c.Cluster.noder, c.Cluster.Hasher, c.Cluster.ReplicaN)
This commit is contained in:
Travis
parent
3f26d667b4
commit
134abda51b
5 changed files with 101 additions and 35 deletions
|
|
@ -27,6 +27,7 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/pilosa/pilosa/v2"
|
||||
"github.com/pilosa/pilosa/v2/topology"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/zeebo/blake3"
|
||||
bolt "go.etcd.io/bbolt"
|
||||
|
|
@ -626,7 +627,11 @@ func (s *TranslateStore) ComputeTranslatorSummaryCols(partitionID int, topo *pil
|
|||
if partitionID != s.partitionID {
|
||||
panic(fmt.Sprintf("inconsistent partitionID arg %v with TranslateStore.paritionID %v", partitionID, s.partitionID))
|
||||
}
|
||||
firstPrimary := topo.PrimaryNodeIndex(partitionID)
|
||||
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
snap := topology.NewClusterSnapshot(topo, topo.Hasher, topo.ReplicaN)
|
||||
|
||||
firstPrimary := snap.PrimaryNodeIndex(partitionID)
|
||||
|
||||
err = s.db.View(func(tx *bolt.Tx) error {
|
||||
|
||||
|
|
@ -656,7 +661,7 @@ func (s *TranslateStore) ComputeTranslatorSummaryCols(partitionID int, topo *pil
|
|||
shard := id / pilosa.ShardWidth
|
||||
|
||||
ks := string(v)
|
||||
primary := topo.GetPrimaryForColKeyTranslation(s.index, ks)
|
||||
primary := snap.PrimaryForColKeyTranslation(s.index, ks)
|
||||
if firstPrimary < 0 {
|
||||
firstPrimary = primary
|
||||
} else {
|
||||
|
|
@ -666,7 +671,7 @@ func (s *TranslateStore) ComputeTranslatorSummaryCols(partitionID int, topo *pil
|
|||
}
|
||||
|
||||
// Verify the invariant that the primaries agree. Just a sanity check.
|
||||
primaryForShard := topo.GetPrimaryForShardReplication(s.index, shard)
|
||||
primaryForShard := snap.PrimaryForShardReplication(s.index, shard)
|
||||
if primaryForShard != firstPrimary {
|
||||
panic(fmt.Sprintf("primaryForShard (%v) != firstPrimary (%v); key='%v', id=%v, shard=%v; partitionID=%v", primaryForShard, firstPrimary, ks, id, shard, partitionID))
|
||||
}
|
||||
|
|
@ -1329,11 +1334,17 @@ func makeStringKeyChanges(
|
|||
}
|
||||
}
|
||||
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
var snap *topology.ClusterSnapshot
|
||||
if topo != nil {
|
||||
snap = topology.NewClusterSnapshot(topo, topo.Hasher, topo.ReplicaN)
|
||||
}
|
||||
|
||||
for key2, id2 := range fwd2 {
|
||||
//vv("makeStringKeyChanges on fwd2, key2='%v', id2=%x", key2, id2)
|
||||
isPrimary := false
|
||||
if topo != nil {
|
||||
primary := topo.GetPrimaryForColKeyTranslation(s.index, key2)
|
||||
primary := snap.PrimaryForColKeyTranslation(s.index, key2)
|
||||
isPrimary = s.partitionID == primary
|
||||
}
|
||||
_ = isPrimary
|
||||
|
|
|
|||
77
cluster.go
77
cluster.go
|
|
@ -899,10 +899,10 @@ func shardToShardPartition(index string, shard uint64, partitionN int) int {
|
|||
return int(h.Sum64() % uint64(partitionN))
|
||||
}
|
||||
|
||||
// keyPartition returns the key-partition that a key belongs to.
|
||||
// KeyPartition returns the key-partition that a key belongs to.
|
||||
// NOTE: the key-partition is DIFFERENT from the shard-partition.
|
||||
func (topo *Topology) KeyPartition(index, key string) int {
|
||||
return keyToKeyPartition(index, key, topo.PartitionN)
|
||||
func (t *Topology) KeyPartition(index, key string) int {
|
||||
return keyToKeyPartition(index, key, t.PartitionN)
|
||||
}
|
||||
|
||||
func keyToKeyPartition(index, key string, partitionN int) int {
|
||||
|
|
@ -1021,31 +1021,31 @@ func (c *cluster) unprotectedPrimaryPartitionNode(partition int) *topology.Node
|
|||
return nil
|
||||
}
|
||||
|
||||
func (topo *Topology) IsPrimary(nodeID string, partitionID int) bool {
|
||||
primary := topo.PrimaryNodeIndex(partitionID)
|
||||
return nodeID == topo.nodeIDs[primary]
|
||||
func (t *Topology) IsPrimary(nodeID string, partitionID int) bool {
|
||||
primary := t.PrimaryNodeIndex(partitionID)
|
||||
return nodeID == t.nodeIDs[primary]
|
||||
}
|
||||
|
||||
func (topo *Topology) PrimaryNodeIndex(partitionID int) (nodeIndex int) {
|
||||
n := len(topo.nodeIDs)
|
||||
func (t *Topology) PrimaryNodeIndex(partitionID int) (nodeIndex int) {
|
||||
n := len(t.nodeIDs)
|
||||
if n == 0 {
|
||||
if topo.cluster != nil {
|
||||
n = len(topo.cluster.nodes)
|
||||
if t.cluster != nil {
|
||||
n = len(t.cluster.nodes)
|
||||
}
|
||||
}
|
||||
nodeIndex = topo.Hasher.Hash(uint64(partitionID), n)
|
||||
nodeIndex = t.Hasher.Hash(uint64(partitionID), n)
|
||||
return
|
||||
}
|
||||
|
||||
func (topo *Topology) GetNonPrimaryReplicas(partitionID int) (nonPrimaryReplicas []string) {
|
||||
func (t *Topology) GetNonPrimaryReplicas(partitionID int) (nonPrimaryReplicas []string) {
|
||||
|
||||
primary := topo.PrimaryNodeIndex(partitionID)
|
||||
nodeN := len(topo.nodeIDs)
|
||||
primary := t.PrimaryNodeIndex(partitionID)
|
||||
nodeN := len(t.nodeIDs)
|
||||
|
||||
// Collect nodes around the ring.
|
||||
for i := 1; i < nodeN; i++ {
|
||||
nodeID := topo.nodeIDs[(primary+i)%nodeN]
|
||||
if i < topo.ReplicaN {
|
||||
nodeID := t.nodeIDs[(primary+i)%nodeN]
|
||||
if i < t.ReplicaN {
|
||||
nonPrimaryReplicas = append(nonPrimaryReplicas, nodeID)
|
||||
}
|
||||
}
|
||||
|
|
@ -1053,7 +1053,7 @@ func (topo *Topology) GetNonPrimaryReplicas(partitionID int) (nonPrimaryReplicas
|
|||
}
|
||||
|
||||
// the map replicaNodeIDs[nodeID] will have a true value for the primary nodeID, and false for others.
|
||||
func (topo *Topology) GetReplicasForPrimary(primary int) (replicaNodeIDs, nonReplicas map[string]bool) {
|
||||
func (t *Topology) GetReplicasForPrimary(primary int) (replicaNodeIDs, nonReplicas map[string]bool) {
|
||||
if primary < 0 {
|
||||
// no nodes anyway
|
||||
return
|
||||
|
|
@ -1061,12 +1061,12 @@ func (topo *Topology) GetReplicasForPrimary(primary int) (replicaNodeIDs, nonRep
|
|||
replicaNodeIDs = make(map[string]bool)
|
||||
nonReplicas = make(map[string]bool)
|
||||
|
||||
nodeN := len(topo.nodeIDs)
|
||||
nodeN := len(t.nodeIDs)
|
||||
|
||||
// Collect nodes around the ring.
|
||||
for i := 0; i < nodeN; i++ {
|
||||
nodeID := topo.nodeIDs[(primary+i)%nodeN]
|
||||
if i < topo.ReplicaN {
|
||||
nodeID := t.nodeIDs[(primary+i)%nodeN]
|
||||
if i < t.ReplicaN {
|
||||
// mark true if primary
|
||||
replicaNodeIDs[nodeID] = (i == 0)
|
||||
} else {
|
||||
|
|
@ -1895,6 +1895,37 @@ func (t *Topology) String() string {
|
|||
t.ReplicaN,
|
||||
)
|
||||
}
|
||||
|
||||
///////////////////////////////////////////
|
||||
// Topology implements the Noder interface.
|
||||
|
||||
// Nodes implements the Noder interface.
|
||||
func (t *Topology) Nodes() []*topology.Node {
|
||||
nodes := make([]*topology.Node, len(t.nodeIDs))
|
||||
for i, nodeID := range t.nodeIDs {
|
||||
nodes[i] = &topology.Node{
|
||||
ID: nodeID,
|
||||
}
|
||||
}
|
||||
return nodes
|
||||
}
|
||||
|
||||
// SetNodes implements the Noder interface.
|
||||
func (t *Topology) SetNodes(nodes []*topology.Node) {}
|
||||
|
||||
// AppendNode implements the Noder interface.
|
||||
func (t *Topology) AppendNode(node *topology.Node) {}
|
||||
|
||||
// RemoveNode implements the Noder interface.
|
||||
func (t *Topology) RemoveNode(nodeID string) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// SetNodeState implements the Noder interface.
|
||||
func (t *Topology) SetNodeState(nodeID string, state string) {}
|
||||
|
||||
///////////////////////////////////////////
|
||||
|
||||
func (t *Topology) GetNodeIDs() []string {
|
||||
return t.nodeIDs
|
||||
}
|
||||
|
|
@ -2609,9 +2640,9 @@ func (c *cluster) translateIndexKeys(ctx context.Context, indexName string, keys
|
|||
// are shared between replicas, and one node is the primary for
|
||||
// replication. So with 4 nodes and 3-way replication, each node has 3/4 of
|
||||
// the translation stores on it.
|
||||
func (topo *Topology) GetPrimaryForColKeyTranslation(index, key string) (primary int) {
|
||||
partitionID := topo.KeyPartition(index, key)
|
||||
return topo.PrimaryNodeIndex(partitionID)
|
||||
func (t *Topology) GetPrimaryForColKeyTranslation(index, key string) (primary int) {
|
||||
partitionID := t.KeyPartition(index, key)
|
||||
return t.PrimaryNodeIndex(partitionID)
|
||||
}
|
||||
|
||||
// should match cluster.go:1033 cluster.ownsShard(nodeID, index, shard)
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ import (
|
|||
"github.com/pilosa/pilosa/v2/boltdb"
|
||||
"github.com/pilosa/pilosa/v2/internal"
|
||||
"github.com/pilosa/pilosa/v2/server"
|
||||
"github.com/pilosa/pilosa/v2/topology"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/zeebo/blake3"
|
||||
)
|
||||
|
|
@ -782,6 +783,9 @@ func (cfg *FsckConfig) analyzeThisIndex(
|
|||
index, len(nodes2fragsum), nodes2fragsum)
|
||||
}
|
||||
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
snap := topology.NewClusterSnapshot(cfg.topo, cfg.topo.Hasher, cfg.topo.ReplicaN)
|
||||
|
||||
for node, sum := range nodes2fragsum {
|
||||
if !quiet {
|
||||
fmt.Printf("# on node '%v'\n", node)
|
||||
|
|
@ -798,7 +802,7 @@ func (cfg *FsckConfig) analyzeThisIndex(
|
|||
totalFiles++
|
||||
//vv("checking %v on node %v", relpath, node)
|
||||
|
||||
replicas, nonReplicas := cfg.topo.GetReplicasForPrimary(fragsum.Primary)
|
||||
replicas, nonReplicas := snap.ReplicasForPrimary(fragsum.Primary)
|
||||
_, _ = replicas, nonReplicas
|
||||
//vv("replicas = '%#v'", replicas)
|
||||
//vv("nonReplicas = '%#v'", nonReplicas)
|
||||
|
|
|
|||
18
fragment.go
18
fragment.go
|
|
@ -3555,8 +3555,12 @@ func (s *fragmentSyncer) syncFragment() error {
|
|||
span, ctx := tracing.StartSpanFromContext(context.Background(), "FragmentSyncer.syncFragment")
|
||||
defer span.Finish()
|
||||
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
// TODO: this needs to use Cluster.noder once that has been implemented.
|
||||
snap := topology.NewClusterSnapshot(topology.NewLocalNoder(s.Cluster.Nodes()), s.Cluster.Hasher, s.Cluster.ReplicaN)
|
||||
|
||||
// Determine replica set.
|
||||
nodes := s.Cluster.shardNodes(s.Fragment.index(), s.Fragment.shard)
|
||||
nodes := snap.ShardNodes(s.Fragment.index(), s.Fragment.shard)
|
||||
if len(nodes) == 1 {
|
||||
return nil
|
||||
}
|
||||
|
|
@ -3672,9 +3676,13 @@ func (s *fragmentSyncer) syncBlockFromPrimary(id int) error {
|
|||
|
||||
f := s.Fragment
|
||||
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
// TODO: this needs to use Cluster.noder once that has been implemented.
|
||||
snap := topology.NewClusterSnapshot(topology.NewLocalNoder(s.Cluster.Nodes()), s.Cluster.Hasher, s.Cluster.ReplicaN)
|
||||
|
||||
// Determine replica set. Return early if this is not
|
||||
// the primary node.
|
||||
nodes := s.Cluster.shardNodes(f.index(), f.shard)
|
||||
nodes := snap.ShardNodes(f.index(), f.shard)
|
||||
if s.Node.ID != nodes[0].ID {
|
||||
f.holder.Logger.Debugf("non-primary replica expecting sync from primary: %s, index=%s, field=%s, shard=%d", nodes[0].ID, f.index(), f.field(), f.shard)
|
||||
return nil
|
||||
|
|
@ -3721,10 +3729,14 @@ func (s *fragmentSyncer) syncBlock(id int) error {
|
|||
|
||||
f := s.Fragment
|
||||
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
// TODO: this needs to use Cluster.noder once that has been implemented.
|
||||
snap := topology.NewClusterSnapshot(topology.NewLocalNoder(s.Cluster.Nodes()), s.Cluster.Hasher, s.Cluster.ReplicaN)
|
||||
|
||||
// Read pairs from each remote block.
|
||||
var uris []*pnet.URI
|
||||
var pairSets []pairSet
|
||||
for _, node := range s.Cluster.shardNodes(f.index(), f.shard) {
|
||||
for _, node := range snap.ShardNodes(f.index(), f.shard) {
|
||||
if s.Node.ID == node.ID {
|
||||
continue
|
||||
}
|
||||
|
|
|
|||
16
index.go
16
index.go
|
|
@ -32,6 +32,7 @@ import (
|
|||
"github.com/pilosa/pilosa/v2/roaring"
|
||||
"github.com/pilosa/pilosa/v2/stats"
|
||||
"github.com/pilosa/pilosa/v2/testhook"
|
||||
"github.com/pilosa/pilosa/v2/topology"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/zeebo/blake3"
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
|
@ -847,6 +848,9 @@ floop:
|
|||
fmt.Printf("# ====================\n")
|
||||
}
|
||||
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
snap := topology.NewClusterSnapshot(topo, topo.Hasher, topo.ReplicaN)
|
||||
|
||||
tloop:
|
||||
for partitionID, store := range idx.translateStores {
|
||||
partitionID := partitionID
|
||||
|
|
@ -855,7 +859,7 @@ tloop:
|
|||
fun2 := func(worker int) error {
|
||||
//vv("ComputeTranslatorSummary() running on store.Path = '%v'", store.GetStorePath())
|
||||
if checkKeys {
|
||||
prim := topo.PrimaryNodeIndex(partitionID)
|
||||
prim := snap.PrimaryNodeIndex(partitionID)
|
||||
primID := topo.nodeIDs[prim]
|
||||
|
||||
// note: we fix irrespective of nodeID == primID now, so that we
|
||||
|
|
@ -891,9 +895,9 @@ tloop:
|
|||
sum.Index = idx.Name()
|
||||
sum.StorePath = store.GetStorePath()
|
||||
sum.NodeID = nodeID
|
||||
sum.IsPrimary = topo.IsPrimary(nodeID, partitionID)
|
||||
sum.IsPrimary = snap.IsPrimary(nodeID, partitionID)
|
||||
|
||||
replicas := topo.GetNonPrimaryReplicas(partitionID)
|
||||
replicas := snap.NonPrimaryReplicas(partitionID)
|
||||
for _, replica := range replicas {
|
||||
if nodeID == replica {
|
||||
sum.IsReplica = true
|
||||
|
|
@ -980,6 +984,10 @@ func (idx *Index) WriteFragmentChecksums(w io.Writer, showBits, showOps bool, to
|
|||
IndexPath: idx.path,
|
||||
RelPath2fsum: make(map[string]*FragSum),
|
||||
}
|
||||
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
snap := topology.NewClusterSnapshot(topo, topo.Hasher, topo.ReplicaN)
|
||||
|
||||
paths, err := listFilesUnderDir(idx.path, false, "", true)
|
||||
panicOn(err)
|
||||
index := idx.name
|
||||
|
|
@ -990,7 +998,7 @@ func (idx *Index) WriteFragmentChecksums(w io.Writer, showBits, showOps bool, to
|
|||
continue // ignore .meta paths
|
||||
}
|
||||
abspath := idx.path + sep + relpath
|
||||
primary := topo.GetPrimaryForShardReplication(index, shard)
|
||||
primary := snap.PrimaryForShardReplication(index, shard)
|
||||
|
||||
checksum, hotbits := RoaringFragmentChecksum(abspath, index, field, view, shard)
|
||||
if verbose {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue