From 3f26d667b4de8285e278b64e49ae4d2fdb2095bd Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 6 Jan 2021 21:52:02 -0600 Subject: [PATCH 1/3] remove pilosa.DefaultPartitionN --- boltdb/translate_test.go | 5 +++-- cluster.go | 5 +---- cmd/pilosa-fsck/fsck.go | 6 +++--- holder.go | 2 +- http/client_test.go | 5 +++-- translator_test.go | 7 ++++--- utils_internal_test.go | 2 +- 7 files changed, 16 insertions(+), 16 deletions(-) diff --git a/boltdb/translate_test.go b/boltdb/translate_test.go index 90977ca82..eb720de18 100644 --- a/boltdb/translate_test.go +++ b/boltdb/translate_test.go @@ -26,6 +26,7 @@ import ( "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/boltdb" + "github.com/pilosa/pilosa/v2/topology" ) //var vv = pilosa.VV @@ -540,7 +541,7 @@ func MustNewTranslateStore() *boltdb.TranslateStore { panic(err) } - s := boltdb.NewTranslateStore("I", "F", 0, pilosa.DefaultPartitionN) + s := boltdb.NewTranslateStore("I", "F", 0, topology.DefaultPartitionN) s.Path = f.Name() return s } @@ -653,7 +654,7 @@ func TestCryptoHashPerKey(t *testing.T) { } // done with setup - sum, err := s.ComputeTranslatorSummaryCols(0, pilosa.NewTopology(&pilosa.Jmphasher{}, pilosa.DefaultPartitionN, 1, nil)) + sum, err := s.ComputeTranslatorSummaryCols(0, pilosa.NewTopology(&pilosa.Jmphasher{}, topology.DefaultPartitionN, 1, nil)) if err != nil { panic(err) } diff --git a/cluster.go b/cluster.go index 405b65901..b6e7e17e4 100644 --- a/cluster.go +++ b/cluster.go @@ -42,9 +42,6 @@ import ( ) const ( - // DefaultPartitionN is the default number of partitions in a cluster. - DefaultPartitionN = 256 - // ClusterState represents the state returned in the /status endpoint. ClusterStateStarting = "STARTING" ClusterStateDegraded = "DEGRADED" // cluster is running but we've lost some # of hosts >0 but < replicaN @@ -138,7 +135,7 @@ type cluster struct { // nolint: maligned func newCluster() *cluster { return &cluster{ Hasher: &Jmphasher{}, - partitionN: DefaultPartitionN, + partitionN: topology.DefaultPartitionN, ReplicaN: 1, joiningLeavingNodes: make(chan nodeAction, 10), // buffered channel diff --git a/cmd/pilosa-fsck/fsck.go b/cmd/pilosa-fsck/fsck.go index 34ed071f8..cef5583cd 100644 --- a/cmd/pilosa-fsck/fsck.go +++ b/cmd/pilosa-fsck/fsck.go @@ -464,7 +464,7 @@ func (cfg *FsckConfig) RepairTranslationStores(ats *pilosa.AllTranslatorSummary) if err != nil { return errors.Wrap(err, fmt.Sprintf("RepairTranslationStores() os.RemoveAll(e.StorePath='%v')", e.StorePath)) } - store, err := boltdb.OpenTranslateStore(e.StorePath, e.Index, e.Field, e.PartitionID, pilosa.DefaultPartitionN) + store, err := boltdb.OpenTranslateStore(e.StorePath, e.Index, e.Field, e.PartitionID, topology.DefaultPartitionN) if err != nil { return errors.Wrap(err, fmt.Sprintf("RepairTranslationStores() create empty boldtdb: boltdb.OpenTranslateStore e.StorePath='%v'", e.StorePath)) } @@ -600,7 +600,7 @@ func (cfg *FsckConfig) readOneDir(dir string) (idx2frag map[string]*pilosa.Index } jmphasher := &pilosa.Jmphasher{} - partitionN := pilosa.DefaultPartitionN + partitionN := topology.DefaultPartitionN replicaN := cfg.ReplicaN topo, err := loadTopology(dir, jmphasher, partitionN, replicaN) if err != nil { @@ -937,7 +937,7 @@ func (cfg *FsckConfig) analyzeThisIndex( # %v # ======================================================== `, - cfg.Fix, index, nDir, cfg.ReplicaN, humanize.Comma(totalBytes), humanize.Comma(totalFiles), humanize.Comma(int64(nDir*pilosa.DefaultPartitionN)), humanize.Comma(int64(keyCount)), humanize.Comma(int64(idCount)), actionTaken, fragUpdate) + cfg.Fix, index, nDir, cfg.ReplicaN, humanize.Comma(totalBytes), humanize.Comma(totalFiles), humanize.Comma(int64(nDir*topology.DefaultPartitionN)), humanize.Comma(int64(keyCount)), humanize.Comma(int64(idCount)), actionTaken, fragUpdate) return } diff --git a/holder.go b/holder.go index f5145ba31..b1da8d82c 100644 --- a/holder.go +++ b/holder.go @@ -220,7 +220,7 @@ type HolderConfig struct { func DefaultHolderConfig() *HolderConfig { return &HolderConfig{ - PartitionN: DefaultPartitionN, + PartitionN: topology.DefaultPartitionN, OpenTranslateStore: OpenInMemTranslateStore, OpenTranslateReader: nil, OpenTransactionStore: OpenInMemTransactionStore, diff --git a/http/client_test.go b/http/client_test.go index b1b154647..1c8ed525c 100644 --- a/http/client_test.go +++ b/http/client_test.go @@ -33,6 +33,7 @@ import ( "github.com/pilosa/pilosa/v2/pql" "github.com/pilosa/pilosa/v2/server" "github.com/pilosa/pilosa/v2/test" + "github.com/pilosa/pilosa/v2/topology" "github.com/pkg/errors" ) @@ -297,7 +298,7 @@ func TestClient_Export(t *testing.T) { bw := bufio.NewWriter(buf) // Send export request for every partition. - for i := 0; i < pilosa.DefaultPartitionN; i++ { + for i := 0; i < topology.DefaultPartitionN; i++ { if err := c.ExportCSV(context.Background(), "keyed", "unkeyedf", uint64(i), bw); err != nil { t.Fatal(err) } @@ -338,7 +339,7 @@ func TestClient_Export(t *testing.T) { bw := bufio.NewWriter(buf) // Send export request. - for i := 0; i < pilosa.DefaultPartitionN; i++ { + for i := 0; i < topology.DefaultPartitionN; i++ { if err := c.ExportCSV(context.Background(), "keyed", "keyedf", uint64(i), bw); err != nil { t.Fatal(err) } diff --git a/translator_test.go b/translator_test.go index b1f42d518..d8b637b1e 100644 --- a/translator_test.go +++ b/translator_test.go @@ -30,12 +30,13 @@ import ( "github.com/pilosa/pilosa/v2/mock" "github.com/pilosa/pilosa/v2/server" "github.com/pilosa/pilosa/v2/test" + "github.com/pilosa/pilosa/v2/topology" "github.com/pkg/errors" "golang.org/x/sync/errgroup" ) func TestInMemTranslateStore_TranslateKey(t *testing.T) { - s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, pilosa.DefaultPartitionN) + s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, topology.DefaultPartitionN) // Ensure initial key translates to ID 1. if id, err := s.TranslateKey("foo", true); err != nil { @@ -60,7 +61,7 @@ func TestInMemTranslateStore_TranslateKey(t *testing.T) { } func TestInMemTranslateStore_TranslateID(t *testing.T) { - s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, pilosa.DefaultPartitionN) + s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, topology.DefaultPartitionN) // Setup initial keys. if _, err := s.TranslateKey("foo", true); err != nil { @@ -425,7 +426,7 @@ func TestTranslation_KeyNotFound(t *testing.T) { } func TestInMemTranslateStore_ReadKey(t *testing.T) { - s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, pilosa.DefaultPartitionN) + s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, topology.DefaultPartitionN) id, err := s.TranslateKey("foo", false) if err != pilosa.ErrTranslatingKeyNotFound { diff --git a/utils_internal_test.go b/utils_internal_test.go index 4b5820351..959d0736b 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -285,7 +285,7 @@ func (t *ClusterCluster) addCluster(i int, saveTopology bool) (*cluster, error) c.ReplicaN = 1 c.Hasher = NewTestModHasher() c.Path = path - c.partitionN = DefaultPartitionN + c.partitionN = topology.DefaultPartitionN c.Topology = NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c) c.holder = h c.Node = node From 134abda51b1af0000b794705c2e2d232fbe7b8ed Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 6 Jan 2021 22:15:34 -0600 Subject: [PATCH 2/3] 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) --- boltdb/translate.go | 19 +++++++--- cluster.go | 77 +++++++++++++++++++++++++++++------------ cmd/pilosa-fsck/fsck.go | 6 +++- fragment.go | 18 ++++++++-- index.go | 16 ++++++--- 5 files changed, 101 insertions(+), 35 deletions(-) diff --git a/boltdb/translate.go b/boltdb/translate.go index 11a92f430..40ba72cd3 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -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 diff --git a/cluster.go b/cluster.go index b6e7e17e4..154e9f34c 100644 --- a/cluster.go +++ b/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) diff --git a/cmd/pilosa-fsck/fsck.go b/cmd/pilosa-fsck/fsck.go index cef5583cd..cf6d66617 100644 --- a/cmd/pilosa-fsck/fsck.go +++ b/cmd/pilosa-fsck/fsck.go @@ -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) diff --git a/fragment.go b/fragment.go index 20cdcf380..d87171807 100644 --- a/fragment.go +++ b/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 } diff --git a/index.go b/index.go index 6129289f5..457869d6a 100644 --- a/index.go +++ b/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 { From 8dbfae1d86f4c504194e083fccdfb0f91f351944 Mon Sep 17 00:00:00 2001 From: Travis Date: Thu, 7 Jan 2021 13:45:46 -0600 Subject: [PATCH 3/3] temporarily have cluster implement Noder --- cluster.go | 27 ++++++++++++++++++++++++++- fragment.go | 9 +++------ 2 files changed, 29 insertions(+), 7 deletions(-) diff --git a/cluster.go b/cluster.go index 154e9f34c..c6e09cee9 100644 --- a/cluster.go +++ b/cluster.go @@ -73,6 +73,8 @@ type nodeAction struct { // cluster represents a collection of nodes. type cluster struct { // nolint: maligned + noder topology.Noder + id string Node *topology.Node nodes []*topology.Node @@ -133,7 +135,7 @@ type cluster struct { // nolint: maligned // newCluster returns a new instance of Cluster with defaults. func newCluster() *cluster { - return &cluster{ + c := &cluster{ Hasher: &Jmphasher{}, partitionN: topology.DefaultPartitionN, ReplicaN: 1, @@ -152,6 +154,8 @@ func newCluster() *cluster { confirmDownRetries: defaultConfirmDownRetries, confirmDownSleep: defaultConfirmDownSleep, } + c.noder = c // TODO: this is temporary until etcd fully implements noder + return c } // initializeAntiEntropy is called by the anti entropy routine when it starts. @@ -1926,6 +1930,27 @@ func (t *Topology) SetNodeState(nodeID string, state string) {} /////////////////////////////////////////// +/////////////////////////////////////////// +// Cluster implements the Noder interface. +// This is temporary and should be removed once etcd is fully implemented as +// noder. + +// SetNodes implements the Noder interface. +func (c *cluster) SetNodes(nodes []*topology.Node) {} + +// AppendNode implements the Noder interface. +func (c *cluster) AppendNode(node *topology.Node) {} + +// RemoveNode implements the Noder interface. +func (c *cluster) RemoveNode(nodeID string) bool { + return false +} + +// SetNodeState implements the Noder interface. +func (c *cluster) SetNodeState(nodeID string, state string) {} + +/////////////////////////////////////////// + func (t *Topology) GetNodeIDs() []string { return t.nodeIDs } diff --git a/fragment.go b/fragment.go index d87171807..d70b4b442 100644 --- a/fragment.go +++ b/fragment.go @@ -3556,8 +3556,7 @@ func (s *fragmentSyncer) syncFragment() error { 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) + snap := topology.NewClusterSnapshot(s.Cluster.noder, s.Cluster.Hasher, s.Cluster.ReplicaN) // Determine replica set. nodes := snap.ShardNodes(s.Fragment.index(), s.Fragment.shard) @@ -3677,8 +3676,7 @@ 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) + snap := topology.NewClusterSnapshot(s.Cluster.noder, s.Cluster.Hasher, s.Cluster.ReplicaN) // Determine replica set. Return early if this is not // the primary node. @@ -3730,8 +3728,7 @@ 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) + snap := topology.NewClusterSnapshot(s.Cluster.noder, s.Cluster.Hasher, s.Cluster.ReplicaN) // Read pairs from each remote block. var uris []*pnet.URI