scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 00:55:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1294 from travisturner/disco-cleanup

Browse source 
Disco cleanup
This commit is contained in:
Travis Turner 2021-01-07 14:08:55 -06:00 • committed by GitHub
commit 4eb36a34f6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 140 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
boltdb/translate.go
View file

@ -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

5
boltdb/translate_test.go
View file

@ -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)
}

109
cluster.go
View file

@ -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
@ -76,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
@ -136,9 +135,9 @@ type cluster struct { // nolint: maligned
// newCluster returns a new instance of Cluster with defaults.
func newCluster() *cluster {
return &cluster{
c := &cluster{
Hasher: &Jmphasher{},
partitionN: DefaultPartitionN,
partitionN: topology.DefaultPartitionN,
ReplicaN: 1,
joiningLeavingNodes: make(chan nodeAction, 10), // buffered channel
@ -155,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.
@ -902,10 +903,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 {
@ -1024,31 +1025,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)
}
}
@ -1056,7 +1057,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
@ -1064,12 +1065,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 {
@ -1898,6 +1899,58 @@ 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) {}
///////////////////////////////////////////
///////////////////////////////////////////
// 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
}
@ -2612,9 +2665,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)

12
cmd/pilosa-fsck/fsck.go
View file

@ -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"
)
@ -464,7 +465,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 +601,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 {
@ -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)
@ -937,7 +941,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
}

15
fragment.go
View file

@ -3555,8 +3555,11 @@ 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.
snap := topology.NewClusterSnapshot(s.Cluster.noder, 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 +3675,12 @@ func (s *fragmentSyncer) syncBlockFromPrimary(id int) error {
f := s.Fragment
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(s.Cluster.noder, 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 +3727,13 @@ func (s *fragmentSyncer) syncBlock(id int) error {
f := s.Fragment
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(s.Cluster.noder, 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
}

2
holder.go
View file

@ -220,7 +220,7 @@ type HolderConfig struct {
func DefaultHolderConfig() *HolderConfig {
return &HolderConfig{
PartitionN: DefaultPartitionN,
PartitionN: topology.DefaultPartitionN,
OpenTranslateStore: OpenInMemTranslateStore,
OpenTranslateReader: nil,
OpenTransactionStore: OpenInMemTransactionStore,

5
http/client_test.go
View file

@ -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)
}

16
index.go
View file

@ -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 {

7
translator_test.go
View file

@ -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 {

2
utils_internal_test.go
View file

@ -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