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

Merge pull request #1394 from travisturner/disco-clean-topology

Browse source 
Remove pilosa-chk, Topology and ClusterCluster
This commit is contained in:
Travis Turner 2021-02-06 16:18:00 -06:00 • committed by GitHub
commit 968b1ade50
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 37 additions and 2740 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
api.go
View file

@ -1859,7 +1859,7 @@ func (api *API) Info() serverInfo {
StorageBackend: api.holder.txf.TxType(),
ReplicaN: api.cluster.ReplicaN,
ShardHash: api.cluster.Hasher.Name(),
KeyHash: api.cluster.Topology.Hasher.Name(),
KeyHash: api.cluster.Hasher.Name(),
}
}

841
boltdb/translate.go
View file

@ -22,14 +22,11 @@ import (
"io/ioutil"
"os"
"path/filepath"
"sort"
"sync"
"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"
"runtime/pprof"
@ -99,10 +96,6 @@ type TranslateStore struct {
Path string
}
func (s *TranslateStore) GetStorePath() string {
return s.Path
}
// NewTranslateStore returns a new instance of TranslateStore.
func NewTranslateStore(index, field string, partitionID, partitionN int) *TranslateStore {
return &TranslateStore{
@ -579,837 +572,3 @@ func findKeyByID(bkt *bolt.Bucket, id uint64) string {
}
return string(boltKey)
}
func (s *TranslateStore) ComputeTranslatorSummaryRows() (sum *pilosa.TranslatorSummary, err error) {
sum = &pilosa.TranslatorSummary{}
hasher := blake3.New()
err = s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketKeys)
if bkt == nil {
panic("bucketKeys not found")
}
cur := bkt.Cursor()
for k, v := cur.First(); k != nil; k, v = cur.Next() {
input := append(k, v...)
_, _ = hasher.Write(input)
sum.KeyCount++
}
bkt = tx.Bucket(bucketIDs)
if bkt == nil {
panic("bucketIDs not found")
}
cur = bkt.Cursor()
for k, v := cur.First(); k != nil; k, v = cur.Next() {
input := append(k, v...)
_, _ = hasher.Write(input)
sum.IDCount++
}
return nil
})
if err != nil {
return nil, err
}
var buf [16]byte
_, _ = hasher.Digest().Read(buf[0:])
sum.Checksum = string(buf[:])
return sum, nil
}
func (s *TranslateStore) ComputeTranslatorSummaryCols(partitionID int, topo *pilosa.Topology) (sum *pilosa.TranslatorSummary, err error) {
sum = &pilosa.TranslatorSummary{}
hasher := blake3.New()
if partitionID != s.partitionID {
panic(fmt.Sprintf("inconsistent partitionID arg %v with TranslateStore.paritionID %v", partitionID, s.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 {
bkt := tx.Bucket(bucketKeys) // key -> id
if bkt == nil {
panic("bucketKeys not found")
}
cur := bkt.Cursor()
for k, v := cur.First(); k != nil; k, v = cur.Next() {
input := append(k, v...)
//vv("55555 ComputeTranslatorSummaryCols(partitionID=%v, path='%v'), k='%v', v=%x", partitionID, s.Path, string(k), v)
_, _ = hasher.Write(input)
sum.KeyCount++
}
bkt = tx.Bucket(bucketIDs) // id -> key
if bkt == nil {
panic("bucketIDs not found")
}
cur = bkt.Cursor()
for k, v := cur.First(); k != nil; k, v = cur.Next() {
// should the primary be the same for each key in this partition?
id := btou64(k)
shard := id / pilosa.ShardWidth
ks := string(v)
primary := snap.PrimaryForColKeyTranslation(s.index, ks)
if firstPrimary < 0 {
firstPrimary = primary
} else {
if primary != firstPrimary {
panic(fmt.Sprintf("s.index='%v' primary (%v) != firstPrimary (%v); key='%v', id=%v, shard=%v; partitionID=%v; topo='%v'", s.index, primary, firstPrimary, ks, id, shard, partitionID, topo.String()))
}
}
// Verify the invariant that the primaries agree. Just a sanity check.
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))
}
input := append(k, v...)
//vv("55555 ComputeTranslatorSummaryCols(partitionID=%v, path='%v'), idBucket id=%x key='%v'", partitionID, s.Path, id, ks)
_, _ = hasher.Write(input)
sum.IDCount++
}
return nil
})
if err != nil {
return nil, err
}
sum.PrimaryNodeIndex = firstPrimary
var buf [16]byte
_, _ = hasher.Digest().Read(buf[0:])
sum.Checksum = string(buf[:])
return sum, nil
}
func (s *TranslateStore) KeyWalker(walk func(key string, col uint64)) error {
return s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketKeys)
if bkt == nil {
panic("bucketKeys not found")
}
cur := bkt.Cursor()
for k, v := cur.First(); k != nil; k, v = cur.Next() {
walk(string(k), btou64(v))
}
return nil
})
}
func (s *TranslateStore) IDWalker(walk func(key string, col uint64)) error {
return s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketIDs)
if bkt == nil {
panic("bucketIDs not found")
}
cur := bkt.Cursor()
for k, v := cur.First(); k != nil; k, v = cur.Next() {
walk(string(v), btou64(k))
}
return nil
})
}
// call s.notifyWrite() when done
func (s *TranslateStore) SetFwdRevMaps(tx *bolt.Tx, fwd map[string]uint64, rev map[uint64]string) (err error) {
localTx := false
if tx == nil {
localTx = true
tx, err = s.db.Begin(true)
if err != nil {
return err
}
defer func() {
_ = tx.Rollback()
}()
}
// reinitialize buckets
err = tx.DeleteBucket(bucketKeys)
if err != nil {
return err
}
err = tx.DeleteBucket(bucketIDs)
if err != nil {
return err
}
if _, err := tx.CreateBucketIfNotExists(bucketKeys); err != nil {
return err
} else if _, err := tx.CreateBucketIfNotExists(bucketIDs); err != nil {
return err
}
key2id := tx.Bucket(bucketKeys)
for k, v := range fwd {
err := key2id.Put([]byte(k), u64tob(v))
if err != nil {
return err
}
}
id2key := tx.Bucket(bucketIDs)
for k, v := range rev {
err := id2key.Put(u64tob(k), []byte(v))
if err != nil {
return err
}
}
if localTx {
return tx.Commit()
}
return nil
}
func (s *TranslateStore) GetFwdRevMaps(tx *bolt.Tx) (fwd map[string]uint64, rev map[uint64]string, err error) {
fwd = make(map[string]uint64)
rev = make(map[uint64]string)
key2id := tx.Bucket(bucketKeys)
err = key2id.ForEach(func(k, v []byte) error {
fwd[string(k)] = btou64(v)
return nil
})
if err != nil {
return
}
id2key := tx.Bucket(bucketIDs)
err = id2key.ForEach(func(k, v []byte) error {
rev[btou64(k)] = string(v)
return nil
})
return
}
//var vv = pilosa.VV
// helpers for repair
// muint64 holds multiple unit64
type muint64 struct {
slc []uint64
}
func (m *muint64) String() (s string) {
for _, e := range m.slc {
s += fmt.Sprintf("%x, ", e)
}
return
}
// mstring holds multiple strings
type mstring struct {
slc []string
}
func (m *mstring) String() (s string) {
for _, e := range m.slc {
s += e + ","
}
return
}
func addToProblemKeys(problemKeys map[string]*muint64, k string, v uint64, noValue bool) {
mu, already := problemKeys[k]
if !already {
mu = &muint64{}
problemKeys[k] = mu
}
if !noValue {
mu.slc = append(mu.slc, v)
}
}
func addToProblemIDs(problemIDs map[uint64]*mstring, k uint64, v string, noValue bool) {
mu, already := problemIDs[k]
if !already {
mu = &mstring{}
problemIDs[k] = mu
}
if !noValue {
mu.slc = append(mu.slc, v)
}
}
// only actually apply the fixes if applyKeyRepairs is true.
// if anything changed, return changed == true.
func (s *TranslateStore) RepairKeys(topo *pilosa.Topology, verbose, applyKeyRepairs bool) (changed bool, err error) {
// strategy: get the full set of keys; the domain keys from
// the forward key->id mapping, and the range keys from the reverse id->key mapping.
// Then march through them and make sure they are mapped correctly.
// At the moment we do try to reuse dangling IDs instead of making
// new ones. This might not always be possible, but we hope for
// now that it suffices b/c it minimizes the amount of fragment
// re-write we may have to do.
/*
// ============ profiling ===============
fd, err := ioutil.TempFile(".", "cpu.prof")
if err != nil {
panic(err)
}
_ = pprof.StartCPUProfile(fd)
defer func() {
pprof.StopCPUProfile()
fd.Close()
}()
// ============ end profiling ===============
*/
tx, err := s.db.Begin(true)
if err != nil {
return false, err
}
defer func() {
_ = tx.Rollback()
}()
fwd, rev, err := s.GetFwdRevMaps(tx)
if err != nil {
return false, err
}
// place to store the correct stuff.
//
// fwd2, rev2: new, repaired versions.
// INVAR: they only contain (correct) invertible mappings.
fwd2 := make(map[string]uint64)
rev2 := make(map[uint64]string)
// and a place to store the problems.
problemKeys := make(map[string]*muint64)
problemIDs := make(map[uint64]*mstring)
fwdscan:
for k, v := range fwd {
_, already := fwd2[k]
if already {
// k has already been repaired. don't worry about further.
continue fwdscan
} else {
// if its already invertible, then just keep it, no need to repair it
// INVAR: k is not in fwd2 (at least not yet).
rkey, ok := rev[v]
if !ok {
// k -> v -> X
addToProblemIDs(problemIDs, v, k, false)
addToProblemKeys(problemKeys, k, v, false)
continue fwdscan
}
if rkey == k {
// yay. a good, invertible, mapping. no repair needed.
if k == "" {
panic("bad empty key")
}
fwd2[k] = v
rev2[v] = k
continue fwdscan
}
// some kind of problem.
// what kind?
// Define problemKey as: 2nd key mapping to id already in fwd2.
// Define problemID as: 2nd ID mapping to key already in fwd2.
// k -> v -> rkey, and rkey != k.
v2, ok := fwd[rkey]
if ok {
// k -> v -> rkey -> v2, where v2 ?= v
if v2 == v {
// k -> v -> rkey -> v
// just have a problemKey in k.
if rkey == "" {
panic("bad empty rkey")
}
fwd2[rkey] = v
rev2[v] = rkey
addToProblemKeys(problemKeys, k, v, true)
continue fwdscan
}
// this is i = 1 test case. :)
// k -> v -> rkey -> v2, where v != v2, and k != rkey.
addToProblemKeys(problemKeys, k, v, false)
addToProblemIDs(problemIDs, v, rkey, false)
continue fwdscan
} else {
// k -> v -> rkey -> X(nil), and rkey != k.
addToProblemKeys(problemKeys, k, v, false)
addToProblemKeys(problemKeys, rkey, 0, true)
addToProblemIDs(problemIDs, v, rkey, false)
}
}
}
revscan:
for id, key := range rev {
k1, already := rev2[id]
_ = k1
if already {
// fine, already there.
continue
}
// if its already invertible, keep it.
rid, ok := fwd[key]
if !ok {
// id -> key -> X
addToProblemKeys(problemKeys, key, 0, true)
addToProblemIDs(problemIDs, id, key, false)
continue revscan
}
if rid == id {
// id -> key -> id. good. but should have been added to fwd2/rev2 above.
panic("should have been added to fwd2/rev2 above!")
} else {
// id -> key -> rid, where id != rid
// so rid -> ?
keyr, ok := rev[rid]
if !ok {
// id -> key -> rid -> X, where id != rid
addToProblemKeys(problemKeys, key, rid, false)
addToProblemKeys(problemKeys, key, id, false)
addToProblemIDs(problemIDs, rid, key, false)
continue revscan
}
if keyr == key {
// id -> key -> rid -> key. So rid is correct and id is dangling.
//
// Heuristic: ASSUME here, that the 2 consistent links key->rid->key are correct,
// and that the single id -> key is in the wrong. This DOESN'T HAVE
// TO BE THE CASE.
if key == "" {
panic("bad empty key")
}
rev2[rid] = key
fwd2[key] = rid
addToProblemIDs(problemIDs, id, "", true)
} else {
// this is test case i = 0. Must handle it.
// id -> key -> rid -> keyr, id != rid, keyr != key.
id2, ok := fwd[keyr]
if ok && id2 == rid {
// id -> key -> rid -> keyr -> rid, id != rid, keyr != key.
// so rid -> keyr -> rid is good.
if keyr == "" {
panic("bad empty keyr")
}
fwd2[keyr] = rid
rev2[rid] = keyr
// and id -> key -> rid is bad, b/c id != rid.
addToProblemKeys(problemKeys, key, rid, false)
addToProblemIDs(problemIDs, id, key, false)
continue revscan
}
// one of these 3 cases holds. all have the same treatment.
// 1) id2 == id: id -> key -> rid -> keyr -> id2; id != rid, rid != id2, keyr != key.
// 2) id2 != id: id -> key -> rid -> keyr -> id2; id != rid, rid != id2, id2 != id, keyr != key.
// 3) !ok: id -> key -> rid -> keyr -> X, id != rid, keyr != key.
addToProblemIDs(problemIDs, id, key, false)
addToProblemKeys(problemKeys, key, rid, false)
addToProblemIDs(problemIDs, rid, keyr, false)
}
}
}
//vv("problemKeys = '%v'", problemKeys)
//vv("problemIDs = '%v'", problemIDs)
newIDs := make(map[uint64]bool)
// assign new IDs to any problemKeys; but first
// try to reuse already allocated IDs that are just dangling.
loopProblemKeys:
for key, ids := range problemKeys {
// first try a minor repair, maybe it was just mssing from rev
// and we can avoid allocate another id.
// sanity check
v2, already := fwd2[key]
if already {
panic(fmt.Sprintf("should not get here since fwd2 is only correct invertibles: key='%v', v2='%x'", key, v2))
}
// INVAR: we have no correct mapping for key in fwd2.
// treat the danglers as "suggestions" for the correction.
for k, id := range ids.slc {
_ = k
_, already = rev2[id]
if !already {
// is this correct?
// id is not in rev2, and key is not in fwd2.
// therefore, we can add them both and maintain consistency.
//vv("add %v to fwd2", key)
if key == "" {
panic("bad empty key")
}
fwd2[key] = id
rev2[id] = key
continue loopProblemKeys
}
}
// INVAR: key -> ? don't know. We didn't find a usable suggestion for the id.
// yes, we get here. We have key. We are looking for a suitable id for it.
// can we get a usable id from the problemIDs?
found := false
suggestions:
for idp, mkeyp := range problemIDs {
for _, candk := range mkeyp.slc {
//vv("checking problemIDs, ipd=%x, candk='%v'; candk==key is %v", idp, candk, candk == key)
if candk == key {
// we have a suggestion from problemIDs that idp might work, doing key -> idp.
// Validate that this is possible.
k2, already := rev2[idp]
_ = k2
if already {
//vv("idp is already in rev2: idp=%v, k2=%v", idp, k2)
continue suggestions
}
// idp works. put it in the correct set.
if key == "" {
panic("bad empty key")
}
rev2[idp] = key
fwd2[key] = idp
found = true
break suggestions
}
}
}
if !found {
id2 := pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
//vv("could not minor repair, allocating new id2 = %v instead", id2)
newIDs[id2] = true
if key == "" {
panic("bad empty key")
}
fwd2[key] = id2
rev2[id2] = key
}
} // end problemKeys
//for id, keys := range problemIDs {
//}
if verbose {
reportIfGainedOrLostIDs(s, fwd, fwd2, rev, rev2, newIDs)
reportIfGainedOrLostKeys(s, fwd, fwd2, rev, rev2)
}
adds, changes, changeIDs, err := makeStringKeyChanges(verbose, applyKeyRepairs, tx, s, topo, fwd, fwd2, rev, rev2, newIDs)
if err != nil {
return false, err
}
_, _, _ = adds, changes, changeIDs
//vv("changedIDs = '%#v'", changeIDs)
if len(adds) > 0 || len(changes) > 0 || len(changeIDs) > 0 || len(newIDs) > 0 {
changed = true
}
//vv("newIDs = '%#v'", newIDs)
//vv("fwd2 = '%#v'", fwd2)
//vv("rev2 = '%#v'", rev2)
err = tx.Commit()
if err == nil {
s.notifyWrite()
}
return changed, err
}
func reportIfGainedOrLostIDs(s *TranslateStore, fwd, fwd2 map[string]uint64, rev, rev2 map[uint64]string, newIDs map[uint64]bool) {
// get all IDs ever mentioned
before := make(map[uint64]bool)
after := make(map[uint64]bool)
for _, id := range fwd {
before[id] = true
}
for _, id := range fwd2 {
if !newIDs[id] {
after[id] = true
}
}
for id := range rev {
before[id] = true
}
for id := range rev2 {
if !newIDs[id] {
after[id] = true
}
}
nb := len(before)
na := len(after)
if nb != na {
fmt.Printf("# needs-repair: Num ID before %v != Num ID after %v, for boltdb = '%v'. before counts(fwd/rev) = %v/%v. after repair counts(fwd2/rev2) = %v/%v\n", nb, na, s.Path, len(fwd), len(rev), len(fwd2), len(rev2))
}
if len(newIDs) > 0 {
fmt.Printf("# needs-repair: adding newIDs '%#v', for boltdb = '%v'. before counts(fwd/rev) = %v/%v. after repair counts(fwd2/rev2) = %v/%v\n", newIDs, s.Path, len(fwd), len(rev), len(fwd2), len(rev2))
}
}
func reportIfGainedOrLostKeys(s *TranslateStore, fwd, fwd2 map[string]uint64, rev, rev2 map[uint64]string) {
// get all IDs ever mentioned
nb := len(fwd)
na := len(fwd2)
if nb != na {
diffAB := mapDiffStrings(fwd, fwd2)
diffBA := mapDiffStrings(fwd2, fwd)
fmt.Printf("# needs-repair: Num Keys before != Num Keys after, for boltdb = '%v'. before counts(fwd/rev) = %v/%v. after repair counts(fwd2/rev2) = %v/%v. fwd - fwd2 = '%#v'; fwd2-fwd = '%#v'\n", s.Path, len(fwd), len(rev), len(fwd2), len(rev2), diffAB, diffBA)
}
}
// return A - B
func mapDiffStrings(mapA, mapB map[string]uint64) (r []string) {
for a := range mapA {
_, ok := mapB[a]
if !ok {
r = append(r, a)
}
}
sort.Strings(r)
return
}
type BeforeAfterKeyChange struct {
BeforeID uint64
AfterID uint64
}
type BeforeAfterIDChange struct {
IsDelete bool
IsAdd bool
BeforeString string
AfterString string
}
// do the minimal state update.
// fwd2 is the "after" map, all string keys repaired.
func makeStringKeyChanges(
verbose bool,
applyKeyRepairs bool,
tx *bolt.Tx,
s *TranslateStore,
topo *pilosa.Topology,
fwd, fwd2 map[string]uint64,
rev, rev2 map[uint64]string,
newIDs map[uint64]bool,
) (
adds map[string]uint64,
changeKeys map[string]*BeforeAfterKeyChange,
changeIDs map[uint64]*BeforeAfterIDChange,
err error,
) {
//vv("makeStringKeyChanges called")
//vv("fwd2 = '%#v'", fwd2)
//vv("rev2 = '%#v'", rev2)
//vv("fwd = '%#v'", fwd)
//vv("rev = '%#v'", rev)
var action string
if applyKeyRepairs {
action = "applying "
}
// addition of string key
adds = make(map[string]uint64)
// change of the mapping of key -> id.
changeKeys = make(map[string]*BeforeAfterKeyChange)
// changes to bucketIDs
changeIDs = make(map[uint64]*BeforeAfterIDChange)
localTx := false
if applyKeyRepairs && tx == nil {
localTx = true
tx, err = s.db.Begin(true)
if err != nil {
return
}
defer func() {
//vv("tx.Rollback happening")
_ = tx.Rollback()
}()
}
key2id := tx.Bucket(bucketKeys)
id2key := tx.Bucket(bucketIDs)
// make a copy of rev2 that we can delete from, to see if
// any additions left in rev2 need to be added after all of
// rev is analyzed.
rev2cp := make(map[uint64]string)
for id, k := range rev2 {
rev2cp[id] = k
}
// first we clean up any stale IDs from id2key. Then the fwd2 pass
// that follows will write to both key2id and id2key.
for id, key := range rev {
//vv("makeStringKeyChanges on rev2: id=%x -> key='%v'", id, key)
key2, ok := rev2[id]
if !ok {
changeIDs[id] = &BeforeAfterIDChange{IsDelete: true}
if verbose {
fmt.Printf("# %vkey-translation-delete-id: (id %x -> %v). Remaining for that key: ('%v' -> %x)\n", action, id, key, key, fwd2[key])
}
if applyKeyRepairs {
u := u64tob(id)
err = id2key.Delete(u)
if err != nil {
return
}
}
continue
}
delete(rev2cp, id)
if key2 != key {
u := u64tob(id)
k := []byte(key2)
changeIDs[id] = &BeforeAfterIDChange{
BeforeString: key,
AfterString: key2,
}
if verbose {
fmt.Printf("# %vkey-translation-update-id: (id %x -> %v). fwd2 for that key: ('%v' -> %x)\n", action, id, key2, key2, fwd2[key2])
}
if applyKeyRepairs {
err = id2key.Put(u, k)
if err != nil {
return
}
}
}
}
// anything leftover in rev2cp is stuff that is new, only
// in rev2 and not in rev. It needs to be added.
for id, key2 := range rev2cp {
u := u64tob(id)
k := []byte(key2)
changeIDs[id] = &BeforeAfterIDChange{
IsAdd: true,
//BeforeString: left empty
AfterString: key2,
}
if verbose {
fmt.Printf("# %vkey-translation-add-id: (id %x -> %v). Fwd for that key: ('%v' -> %x)\n", action, id, key2, key2, fwd2[key2])
}
if applyKeyRepairs {
err = id2key.Put(u, k)
if err != nil {
return
}
}
}
// We assume here that fwd2 is a super-set of fwd. No string keys
// should be deleted in the repair. Confirm that.
for key, id := range fwd {
_, ok := fwd2[key]
if !ok {
panic(fmt.Sprintf("fwd2 is missing a string key from fwd. key='%v' -> id='%x'", key, id))
}
}
// 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 := snap.PrimaryForColKeyTranslation(s.index, key2)
isPrimary = s.partitionID == primary
}
_ = isPrimary
id, ok := fwd[key2]
if !ok {
adds[key2] = id2
u2 := u64tob(id2)
k2 := []byte(key2)
if verbose {
fmt.Printf("# %vkey-translation-new-key: ('%v' -> %x) added: isPrimary: %v\n", action, key2, id2, isPrimary)
}
if applyKeyRepairs {
err = key2id.Put(k2, u2)
if err != nil {
return
}
err = id2key.Put(u2, k2)
if err != nil {
return
}
}
continue
}
if id != id2 {
changeKeys[key2] = &BeforeAfterKeyChange{
BeforeID: id,
AfterID: id2,
}
if verbose {
fmt.Printf("# %vkey-translation-change-id: ('%v' -> %x) changes to ('%v' -> %x); isPrimary: %v\n", action, key2, id, key2, id2, isPrimary)
}
if applyKeyRepairs {
u2 := u64tob(id2)
k2 := []byte(key2)
err = key2id.Put(k2, u2)
if err != nil {
return
}
err = id2key.Put(u2, k2)
if err != nil {
return
}
}
}
}
if localTx {
err = tx.Commit()
}
return
}
func (s *TranslateStore) DumpBolt(label string) {
fmt.Printf("dumping bolt %v : path='%v'\n", label, s.Path)
_ = s.KeyWalker(func(key string, col uint64) {
fmt.Printf("keyWalker: key '%v' -> col '%x'\n", key, col)
})
_ = s.IDWalker(func(key string, col uint64) {
fmt.Printf("idWalker: id '%x' -> key '%v'\n", col, key)
})
fmt.Printf("DONE with dumping bolt %v; path='%v'\n", label, s.Path)
}

170
boltdb/translate_test.go
View file

@ -628,173 +628,3 @@ func MustCloseTranslateStore(s *boltdb.TranslateStore) {
panic(err)
}
}
func TestCryptoHashPerKey(t *testing.T) {
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
// hash one translation
expect := map[int]string{
1: string([]byte{0x76, 0x48, 0x8b, 0x70, 0xe8, 0x54, 0x35, 0xc6, 0x8e, 0xa6, 0x4, 0x6c, 0xfa, 0xd2, 0x1a, 0x12}),
2: string([]byte{0x81, 0x46, 0x84, 0x37, 0x26, 0x96, 0x41, 0xf3, 0x54, 0x4e, 0x98, 0xbc, 0x48, 0xab, 0x1b, 0xf0}),
3: string([]byte{0x7f, 0xe9, 0xf, 0x6d, 0x7b, 0x14, 0x1, 0x44, 0xb2, 0x4e, 0xd0, 0x86, 0x2f, 0x62, 0x8c, 0xa9}),
}
for n := 1; n < 4; n++ {
var batch0 []string
for i := 0; i < n; i++ {
batch0 = append(batch0, fmt.Sprintf("key%d", i))
}
// Populate the store with the keys in batch0.
batch0IDs, err := s.TranslateKeys(batch0, true)
_ = batch0IDs
if err != nil {
t.Fatal(err)
}
// done with setup
sum, err := s.ComputeTranslatorSummaryCols(0, pilosa.NewTopology(&topology.Jmphasher{}, topology.DefaultPartitionN, 1, nil))
if err != nil {
panic(err)
}
nkey := sum.KeyCount
nid := sum.IDCount
observedChecksum := sum.Checksum
if nkey != n {
panic("wrong key count")
}
if nkey != nid {
panic("key count should match id count")
}
// shardwidth 22 has different hashes, of course.
if pilosa.ShardWidth == 20 {
expectedChecksum := expect[n]
if observedChecksum != expectedChecksum {
panic(fmt.Sprintf("got wrong checksum obs '%#v' vs expected '%#v'", observedChecksum, expectedChecksum))
}
}
}
}
func TestTranslateStore_RepairNonInvertibleStringKeyTranslation(t *testing.T) {
const N = 6
// before repair
var fwd [N]map[string]uint64
var rev [N]map[uint64]string
// after repair
var fwd2 [N]map[string]uint64
var rev2 [N]map[uint64]string
// case 0: forward is messed up (unlikely but check for it anyway, be sure we can repair)
// "key0" -> id 0 // correct.
// "key1" -> id 0 // wrong. after Repair, should see key1 -> 1 (0xec0002)
//
// id 0 -> "key0" // correct
// id 1 -> "key1" // correct
//
fwd[0] = map[string]uint64{"key0": 0xec00001, "key1": 0xec00001}
rev[0] = map[uint64]string{0xec00001: "key0", 0xec00002: "key1"}
fwd2[0] = map[string]uint64{"key0": 0xec00001, "key1": 0xec00002}
rev2[0] = map[uint64]string{0xec00001: "key0", 0xec00002: "key1"}
// case 1: reverse is messed up (we have seen this in the past)
// "key0" -> id 0 // correct
// "key1" -> id 1 // correct
//
// id 0 -> "key0" // correct.
// id 1 -> "key0" // wrong. after Repair, should see id 1 -> "key1"
//
fwd[1] = map[string]uint64{"key0": 0xec00001, "key1": 0xec00002}
rev[1] = map[uint64]string{0xec00001: "key0", 0xec00002: "key0"}
fwd2[1] = map[string]uint64{"key0": 0xec00001, "key1": 0xec00002}
rev2[1] = map[uint64]string{0xec00001: "key0", 0xec00002: "key1"}
// case 2: only present in reverse.
fwd[2] = map[string]uint64{}
rev[2] = map[uint64]string{0xec00001: "key0"}
fwd2[2] = map[string]uint64{"key0": 0xec00001}
rev2[2] = map[uint64]string{0xec00001: "key0"}
// case 3: same thing. with camoflage.
fwd[3] = map[string]uint64{"key1": 0xec00002}
rev[3] = map[uint64]string{0xec00001: "key0", 0xec00002: "key1"}
fwd2[3] = map[string]uint64{"key0": 0xec00001, "key1": 0xec00002}
rev2[3] = map[uint64]string{0xec00001: "key0", 0xec00002: "key1"}
// case 4: only present in forward.
fwd[4] = map[string]uint64{"key0": 0xec00001}
rev[4] = map[uint64]string{}
fwd2[4] = map[string]uint64{"key0": 0xec00001}
rev2[4] = map[uint64]string{0xec00001: "key0"}
// case 5: same thing. with camoflage.
fwd[5] = map[string]uint64{"key0": 0xec00001}
rev[5] = map[uint64]string{0xec00002: "key1"}
fwd2[5] = map[string]uint64{"key0": 0xec00001, "key1": 0xec00002}
rev2[5] = map[uint64]string{0xec00001: "key0", 0xec00002: "key1"}
// case 6: we had an id, but b/c of the fix, that id is no longer used.
// now that id might still be used in the fragment for a column,
// and so we will need to remove that id/column from the fragment.
// encapsulated: "did it affect the state of the fields?"
for i := 0; i < 5; i++ {
//println("i = ", i)
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
if err := s.SetFwdRevMaps(nil, fwd[i], rev[i]); err != nil {
t.Fatal(err)
}
if err := verifyState("setup", i, s, fwd[i], rev[i]); err != nil {
t.Fatal(err)
}
var topo *pilosa.Topology
verbose := false
applyKeyRepairs := true
changed, err := s.RepairKeys(topo, verbose, applyKeyRepairs)
if err != nil {
t.Fatal(err)
}
if !changed {
t.Fatalf("expected changes!")
}
if err := verifyState("afterRepair", i, s, fwd2[i], rev2[i]); err != nil {
t.Fatal(err)
}
}
}
func verifyState(label string, i int, s *boltdb.TranslateStore, fwd map[string]uint64, rev map[uint64]string) error {
// verify the setup
const writable = true
for key, expectID := range fwd {
id, err := s.TranslateKey(key, !writable)
if err != nil {
return err
}
if id != expectID {
return fmt.Errorf("fwd %v problem. i=%v, for key '%v', expected %x, observed %x", label, i, key, expectID, id)
}
}
for id, expectKey := range rev {
key, err := s.TranslateID(id)
if err != nil {
return err
}
if key != expectKey {
return fmt.Errorf("rev %v problem. i=%v, for id '%x', expected %v, observed %v", label, i, id, expectKey, key)
}
}
return nil
}

474
cluster.go
View file

@ -16,22 +16,14 @@ package pilosa
import (
"context"
"encoding/binary"
"encoding/json"
"fmt"
"hash/fnv"
"io"
"io/ioutil"
"math/rand"
"os"
"path/filepath"
"sort"
"sync"
"time"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/v2/disco"
"github.com/pilosa/pilosa/v2/internal"
"github.com/pilosa/pilosa/v2/logger"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/topology"
@ -104,8 +96,7 @@ type cluster struct { // nolint: maligned
maxWritesPerRequest int
// Data directory path.
Path string
Topology *Topology
Path string
// Distributed Consensus
disCo disco.DisCo
@ -751,9 +742,12 @@ func (c *cluster) Nodes() []*topology.Node {
}
func (c *cluster) AllNodeStates() map[string]string {
c.mu.RLock()
defer c.mu.RUnlock()
return c.Topology.nodeStates
// TODO: is this being used by the UI?
// c.mu.RLock()
// defer c.mu.RUnlock()
// return c.Topology.nodeStates
m := make(map[string]string)
return m
}
// removeNodeBasicSorted removes a node from the cluster, maintaining the sort
@ -1058,214 +1052,6 @@ func (c *cluster) shardDistributionByIndex(indexName string) map[string]map[stri
return dist
}
// 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)
// Hash the bytes and mod by partition count.
h := fnv.New64a()
_, _ = h.Write([]byte(index))
_, _ = h.Write(buf[:])
return int(h.Sum64() % uint64(partitionN))
}
// KeyPartition returns the key-partition that a key belongs to.
// NOTE: the key-partition is DIFFERENT from the shard-partition.
func (t *Topology) KeyPartition(index, key string) int {
return keyToKeyPartition(index, key, t.PartitionN)
}
func keyToKeyPartition(index, key string, partitionN int) int {
// Hash the bytes and mod by partition count.
h := fnv.New64a()
_, _ = h.Write([]byte(index))
_, _ = h.Write([]byte(key))
return int(h.Sum64() % uint64(partitionN))
}
// ShardNodes returns a list of nodes that own a fragment. Safe for concurrent use.
func (c *cluster) ShardNodes(index string, shard uint64) []*topology.Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.shardNodes(index, shard)
}
// shardNodes returns a list of nodes that own a shard. unprotected
func (c *cluster) shardNodes(index string, shard uint64) []*topology.Node {
return c.partitionNodes(c.shardToShardPartition(index, shard))
}
// KeyNodes returns a list of nodes that own a fragment. Safe for concurrent use.
func (c *cluster) KeyNodes(index, key string) []*topology.Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.keyNodes(index, key)
}
// keyNodes returns a list of nodes that own a key. unprotected
func (c *cluster) keyNodes(index, key string) []*topology.Node {
return c.partitionNodes(c.Topology.KeyPartition(index, key))
}
// partitionNodes returns a list of nodes that own a partition. unprotected.
func (c *cluster) partitionNodes(partitionID int) []*topology.Node {
// Default replica count to between one and the number of nodes.
// The replica count can be zero if there are no nodes.
// Assume that c.nodes may be missing a node that is part of the cluster but not currently present.
// The partition calculation must use the full cluster size in BOTH cases:
// - use len(c.Topology.nodeIDs) instead of len(c.nodes),
// - collect nodes from c.Topology.nodeIDs rather than from c.nodes,
// - when the node is missing, it should be considered, found absent from c.nodes, then omitted from the return slice.
// Use c.Topology to determine cluster membership when it
// exists and contains data. Otherwise, fall back to using
// c.nodes. The only time c.Topology should be nil is in
// tests.
var useTopology bool
if c.Topology != nil && len(c.Topology.nodeIDs) > 0 {
useTopology = true
}
cNodes := c.noder.Nodes()
replicaN := c.ReplicaN
var nodeN int
if useTopology {
nodeN = len(c.Topology.nodeIDs)
} else {
nodeN = len(cNodes)
}
if replicaN > nodeN {
replicaN = nodeN
} else if replicaN == 0 {
replicaN = 1
}
// Determine primary owner node.
if c.Topology == nil {
c.Topology = NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c)
}
nodeIndex := c.Topology.PrimaryNodeIndex(partitionID)
if nodeIndex < 0 {
// no nodes anyway
return nil
}
// Collect nodes around the ring.
nodes := make([]*topology.Node, 0, replicaN)
for i := 0; i < replicaN; i++ {
if useTopology {
maybeNodeID := c.Topology.nodeIDs[(nodeIndex+i)%nodeN]
if node := topology.Nodes(cNodes).NodeByID(maybeNodeID); node != nil {
nodes = append(nodes, node)
}
} else {
nodes = append(nodes, cNodes[(nodeIndex+i)%len(cNodes)])
}
}
return nodes
}
func (t *Topology) IsPrimary(nodeID string, partitionID int) bool {
primary := t.PrimaryNodeIndex(partitionID)
return nodeID == t.nodeIDs[primary]
}
func (t *Topology) PrimaryNodeIndex(partitionID int) (nodeIndex int) {
n := len(t.nodeIDs)
if n == 0 {
if t.cluster != nil {
n = len(t.cluster.noder.Nodes())
}
}
nodeIndex = t.Hasher.Hash(uint64(partitionID), n)
return
}
func (t *Topology) GetNonPrimaryReplicas(partitionID int) (nonPrimaryReplicas []string) {
primary := t.PrimaryNodeIndex(partitionID)
nodeN := len(t.nodeIDs)
// Collect nodes around the ring.
for i := 1; i < nodeN; i++ {
nodeID := t.nodeIDs[(primary+i)%nodeN]
if i < t.ReplicaN {
nonPrimaryReplicas = append(nonPrimaryReplicas, nodeID)
}
}
return
}
// the map replicaNodeIDs[nodeID] will have a true value for the primary nodeID, and false for others.
func (t *Topology) GetReplicasForPrimary(primary int) (replicaNodeIDs, nonReplicas map[string]bool) {
if primary < 0 {
// no nodes anyway
return
}
replicaNodeIDs = make(map[string]bool)
nonReplicas = make(map[string]bool)
nodeN := len(t.nodeIDs)
// Collect nodes around the ring.
for i := 0; i < nodeN; i++ {
nodeID := t.nodeIDs[(primary+i)%nodeN]
if i < t.ReplicaN {
// mark true if primary
replicaNodeIDs[nodeID] = (i == 0)
} else {
nonReplicas[nodeID] = false
}
}
return
}
// containsShards is like OwnsShards, but it includes replicas.
func (c *cluster) containsShards(index string, availableShards *roaring.Bitmap, node *topology.Node) []uint64 {
var shards []uint64
_ = availableShards.ForEach(func(i uint64) error {
p := c.shardToShardPartition(index, i)
// Determine the nodes for partition.
nodes := c.partitionNodes(p)
for _, n := range nodes {
if n.ID == node.ID {
shards = append(shards, i)
}
}
return nil
})
return shards
}
func (c *cluster) setup() error {
// Load topology file if it exists.
if err := c.loadTopology(); err != nil {
return errors.Wrap(err, "loading topology")
}
return nil
}
// open is only used in internal tests.
func (c *cluster) open() error {
err := c.setup()
if err != nil {
return errors.Wrap(err, "setting up cluster")
}
return c.waitForStarted()
}
func (c *cluster) waitForStarted() error {
return nil
}
func (c *cluster) close() error {
// Notify goroutines of closing and wait for completion.
close(c.closing)
@ -1478,128 +1264,6 @@ func newResizeJob(existingNodes []*topology.Node, node *topology.Node, action st
}
}
type nodeIDs []string
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 id matches one of the nodesets's IDs.
func (n nodeIDs) ContainsID(id string) bool {
for _, nid := range n {
if nid == id {
return true
}
}
return false
}
// Topology represents the list of hosts in the cluster.
// Topology now encapsulates all knowledge needed to
// determine the primary node in the replication scheme.
type Topology struct {
mu sync.RWMutex
nodeIDs []string
clusterID string
// nodeStates holds the state of each node according to
// the coordinator. Used during startup and data load.
nodeStates map[string]string
// moved Hasher, PartitionN and ReplicaN
// from cluster for standalone use and comprehension:
// Hashing algorithm used to assign partitions to nodes.
Hasher topology.Hasher
// The number of partitions in the cluster.
PartitionN int
// The number of replicas a partition has.
ReplicaN int
// can be nil
cluster *cluster
}
// NewTopology creates a Topology.
//
// The arguments and members hasher, partitionN, and
// replicaN were refactored out of struct cluster
// to allow pilosa-fsck to load a Topology from
// backup and then compute primaries standalone -- without starting a cluster.
// As pilosa-fsck operates on all backups at once from
// a single cpu, starting a full cluster isn't possible.
//
// The hasher is the Hashing algorithm used to assign partitions to nodes.
// The cluster c should be provided if possible by pilosa code;
// the pilosa-fsck utility won't be able to provide it.
//
// For the cluster size N, the topology gives preference to
// len(t.nodeIDs) before falling back on len(c.nodes).
//
func NewTopology(hasher topology.Hasher, partitionN int, replicaN int, c *cluster) *Topology {
return &Topology{
Hasher: hasher,
PartitionN: partitionN,
ReplicaN: replicaN,
nodeStates: make(map[string]string),
cluster: c,
}
}
func (t *Topology) String() string {
return fmt.Sprintf(`
&pilosa.Topology{
nodeIDs: %v,
clusterID: %v,
nodeStates: %v,
PartitionN: %v,
ReplicaN: %v,
}
`,
t.nodeIDs,
t.clusterID,
t.nodeStates,
t.PartitionN,
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
}
// PrimaryNodeID implements the Noder interface.
func (t *Topology) PrimaryNodeID(topology.Hasher) string {
return ""
}
// 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
@ -1621,66 +1285,6 @@ func (c *cluster) SetNodeState(nodeID string, state string) {}
///////////////////////////////////////////
func (t *Topology) GetNodeIDs() []string {
return t.nodeIDs
}
// ContainsID returns true if id matches one of the topology's IDs.
func (t *Topology) ContainsID(id string) bool {
t.mu.RLock()
defer t.mu.RUnlock()
return t.containsID(id)
}
func (t *Topology) containsID(id string) bool {
return nodeIDs(t.nodeIDs).ContainsID(id)
}
// addID adds the node ID to the topology and returns true if added.
func (t *Topology) addID(nodeID string) bool {
t.mu.Lock()
defer t.mu.Unlock()
if t.containsID(nodeID) {
return false
}
t.nodeIDs = append(t.nodeIDs, nodeID)
sort.Slice(t.nodeIDs,
func(i, j int) bool {
return t.nodeIDs[i] < t.nodeIDs[j]
})
return true
}
// encode converts t into its internal representation.
func (t *Topology) encode() *internal.Topology {
return encodeTopology(t)
}
// loadTopology reads the topology for the node. unprotected.
func (c *cluster) loadTopology() error {
buf, err := ioutil.ReadFile(filepath.Join(c.Path, ".topology"))
if os.IsNotExist(err) {
c.Topology = NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c)
return nil
} else if err != nil {
return errors.Wrap(err, "reading file")
}
var pb internal.Topology
if err := proto.Unmarshal(buf, &pb); err != nil {
return errors.Wrap(err, "unmarshalling")
}
top, err := DecodeTopology(&pb, c.Hasher, c.partitionN, c.ReplicaN, c)
if err != nil {
return errors.Wrap(err, "decoding")
}
c.Topology = top
return nil
}
func (c *cluster) nodeStatus() *NodeStatus {
ns := &NodeStatus{
Node: c.Node,
@ -1974,27 +1578,6 @@ func (c *cluster) translateIndexKeys(ctx context.Context, indexName string, keys
return ids, nil
}
// The boltdb key translation stores are partitioned, designated by partitionIDs. These
// 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 (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)
// return Nodes(c.shardNodes(index, shard)).ContainsID(nodeID)
func (t *Topology) GetPrimaryForShardReplication(index string, shard uint64) int {
n := len(t.nodeIDs)
if n == 0 {
return -1
}
partition := uint64(shardToShardPartition(index, shard, t.PartitionN))
nodeIndex := t.Hasher.Hash(partition, n)
return nodeIndex
}
func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, keySet map[string]struct{}, writable bool) (map[string]uint64, error) {
keyMap := make(map[string]uint64)
@ -2062,18 +1645,18 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s
return nil, ErrIndexNotFound
}
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
// Split keys by partition.
keysByPartition := make(map[int][]string, c.partitionN)
for _, key := range keys {
partitionID := c.Topology.KeyPartition(indexName, key)
partitionID := snap.KeyToKeyPartition(indexName, key)
keysByPartition[partitionID] = append(keysByPartition[partitionID], key)
}
// TODO: use local replicas to short-circuit network traffic
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
// Group keys by node.
keysByNode := make(map[*topology.Node][]string)
for partitionID, keys := range keysByPartition {
@ -2171,18 +1754,18 @@ func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys ..
return nil, errors.Errorf("can't create index keys on unkeyed index %s", indexName)
}
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
// Split keys by partition.
keysByPartition := make(map[int][]string, c.partitionN)
for _, key := range keys {
partitionID := c.Topology.KeyPartition(indexName, key)
partitionID := snap.KeyToKeyPartition(indexName, key)
keysByPartition[partitionID] = append(keysByPartition[partitionID], key)
}
// TODO: use local replicas to short-circuit network traffic
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
// Group keys by node.
// Delete remote keys from the by-partition map so that it can be used for local translation.
keysByNode := make(map[*topology.Node][]string)
@ -2393,33 +1976,6 @@ type Schema struct {
Indexes []*IndexInfo `json:"indexes"`
}
func encodeTopology(topology *Topology) *internal.Topology {
if topology == nil {
return nil
}
return &internal.Topology{
ClusterID: topology.clusterID,
NodeIDs: topology.nodeIDs,
}
}
// the cluster c is optional but give it if you have it.
func DecodeTopology(topology *internal.Topology, hasher topology.Hasher, partitionN, replicaN int, c *cluster) (*Topology, error) {
if topology == nil {
return nil, nil
}
t := NewTopology(hasher, partitionN, replicaN, c)
t.clusterID = topology.ClusterID
t.nodeIDs = topology.NodeIDs
sort.Slice(t.nodeIDs,
func(i, j int) bool {
return t.nodeIDs[i] < t.nodeIDs[j]
})
return t, nil
}
// CreateShardMessage is an internal message indicating shard creation.
type CreateShardMessage struct {
Index string

402
cluster_internal_test.go
View file

@ -15,7 +15,6 @@
package pilosa
import (
"bytes"
"fmt"
"math/rand"
"net"
@ -31,7 +30,6 @@ import (
"github.com/pilosa/pilosa/v2/test/port"
"github.com/pilosa/pilosa/v2/testhook"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pkg/errors"
)
// GlobalPortMap avoids many races and port conflicts when setting
@ -423,13 +421,16 @@ func TestCluster_Owners(t *testing.T) {
cNodes := c.noder.Nodes()
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
// Verify nodes are distributed.
if a := c.partitionNodes(0); !reflect.DeepEqual(a, []*topology.Node{cNodes[0], cNodes[1]}) {
if a := snap.PartitionNodes(0); !reflect.DeepEqual(a, []*topology.Node{cNodes[0], cNodes[1]}) {
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
}
// Verify nodes go around the ring.
if a := c.partitionNodes(2); !reflect.DeepEqual(a, []*topology.Node{cNodes[2], cNodes[0]}) {
if a := snap.PartitionNodes(2); !reflect.DeepEqual(a, []*topology.Node{cNodes[2], cNodes[0]}) {
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
}
}
@ -440,7 +441,7 @@ func TestCluster_Partition(t *testing.T) {
c := newCluster()
c.partitionN = partitionN
partitionID := c.shardToShardPartition(index, shard)
partitionID := topology.ShardToShardPartition(index, shard, partitionN)
if partitionID < 0 || partitionID >= partitionN {
t.Errorf("partition out of range: shard=%d, p=%d, n=%d", shard, partitionID, partitionN)
}
@ -483,7 +484,11 @@ func TestCluster_ContainsShards(t *testing.T) {
c := NewTestCluster(t, 5)
c.ReplicaN = 3
cNodes := c.noder.Nodes()
shards := c.containsShards("test", roaring.NewBitmap(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10), cNodes[2])
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
shards := snap.ContainsShards("test", roaring.NewBitmap(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10), cNodes[2])
if !reflect.DeepEqual(shards, []uint64{0, 2, 3, 5, 6, 9, 10}) {
t.Fatalf("unexpected shars for node's index: %v", shards)
@ -646,368 +651,6 @@ func TestCluster_Coordinator(t *testing.T) {
})
}
func TestCluster_Topology(t *testing.T) {
t.Skip("these tests don't really apply anymore; they were meant to tests the cluster and adding topology nodes.")
c1 := NewTestCluster(t, 1) // automatically creates Node{ID: "node0"}
const urisCount = 4
var uris []pnet.URI
if err := port.GetPorts(func(ports []int) error {
for i := 0; i < urisCount; i++ {
uris = append(uris, NewTestURIFromHostPort(fmt.Sprintf("host%d", i), uint16(ports[i])))
}
return nil
}, urisCount, 10); err != nil {
t.Fatalf("getting ports: %v", err)
}
node0 := &topology.Node{ID: "node0", URI: uris[0]}
node1 := &topology.Node{ID: "node1", URI: uris[1]}
node2 := &topology.Node{ID: "node2", URI: uris[2]}
nodeinvalid := &topology.Node{ID: "nodeinvalid", URI: uris[3]}
t.Run("AddNode", func(t *testing.T) {
err := c1.addNode(node1.ID)
if err != nil {
t.Fatal(err)
}
// add the same host.
err = c1.addNode(node1.ID)
if err != nil {
t.Fatal(err)
}
err = c1.addNode(node2.ID)
if err != nil {
t.Fatal(err)
}
actual := c1.nodeIDs()
expected := []string{node0.ID, node1.ID, node2.ID}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
}
})
t.Run("ContainsID", func(t *testing.T) {
if !c1.Topology.ContainsID(node1.ID) {
t.Errorf("!ContainsHost error: %v", node1.ID)
} else if c1.Topology.ContainsID(nodeinvalid.ID) {
t.Errorf("ContainsHost error: %v", nodeinvalid.ID)
}
})
}
// Ensure that general cluster functionality works as expected.
func TestCluster_ResizeStates(t *testing.T) {
t.Skip("these tests don't really apply anymore; they were meant to tests the cluster startup process using memberlist and a topology file")
t.Run("Single node, no data", func(t *testing.T) {
tc := NewClusterCluster(t, 1)
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
node := tc.Clusters[0]
state, err := node.State()
if err != nil {
t.Fatal(err)
}
// Ensure that node comes up in state NORMAL.
if state != string(ClusterStateNormal) {
t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, state)
}
expectedTop := &Topology{
nodeIDs: []string{node.Node.ID},
}
// Verify topology file.
if !reflect.DeepEqual(node.Topology.nodeIDs, expectedTop.nodeIDs) {
t.Errorf("expected topology: %v, but got: %v", expectedTop.nodeIDs, node.Topology.nodeIDs)
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Single node, in topology", func(t *testing.T) {
tc := NewClusterCluster(t, 0)
if err := tc.addNode(); err != nil {
t.Fatalf("adding node: %v", err)
}
node := tc.Clusters[0]
// write topology to data file
top := &Topology{
nodeIDs: []string{node.Node.ID},
}
if err := tc.WriteTopology(node.Path, top); err != nil {
t.Fatalf("writing topology: %v", err)
}
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
state, err := node.State()
if err != nil {
t.Fatal(err)
}
// Ensure that node comes up in state NORMAL.
if state != string(ClusterStateNormal) {
t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, state)
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Single node, not in topology", func(t *testing.T) {
tc := NewClusterCluster(t, 0)
if err := tc.addNode(); err != nil {
t.Fatalf("adding node: %v", err)
}
node := tc.Clusters[0]
// write topology to data file
top := &Topology{
nodeIDs: []string{"some-other-host"},
}
if err := tc.WriteTopology(node.Path, top); err != nil {
t.Fatalf("writing topology: %v", err)
}
// Open TestCluster.
expected := "coordinator node0 is not in topology: [some-other-host]"
err := tc.Open()
if err == nil || errors.Cause(err).Error() != expected {
t.Errorf("did not receive expected error, got: %s", errors.Cause(err).Error())
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Multiple nodes, no data", func(t *testing.T) {
tc := NewClusterCluster(t, 0)
if err := tc.addNode(); err != nil {
t.Fatalf("adding node: %v", err)
}
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatalf("opening cluster: %v", err)
}
if err := tc.addNode(); err != nil {
t.Fatalf("adding node: %v", err)
}
node0 := tc.Clusters[0]
state0, err := node0.State()
if err != nil {
t.Fatal(err)
}
node1 := tc.Clusters[1]
state1, err := node1.State()
if err != nil {
t.Fatal(err)
}
// Ensure that nodes comes up in state NORMAL.
if state0 != string(ClusterStateNormal) {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, state0)
} else if state1 != string(ClusterStateNormal) {
t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, state1)
}
expectedTop := &Topology{
nodeIDs: []string{node0.Node.ID, node1.Node.ID},
}
// Verify topology file.
if !reflect.DeepEqual(node0.Topology.nodeIDs, expectedTop.nodeIDs) {
t.Errorf("expected node0 topology: %v, but got: %v", expectedTop.nodeIDs, node0.Topology.nodeIDs)
} else if !reflect.DeepEqual(node1.Topology.nodeIDs, expectedTop.nodeIDs) {
t.Errorf("expected node1 topology: %v, but got: %v", expectedTop.nodeIDs, node1.Topology.nodeIDs)
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Multiple nodes, in/not in topology", func(t *testing.T) {
tc := NewClusterCluster(t, 0)
if err := tc.addNode(); err != nil {
t.Fatalf("adding node: %v", err)
}
node0 := tc.Clusters[0]
// write topology to data file
top := &Topology{
nodeIDs: []string{"node0", "node2"},
}
if err := tc.WriteTopology(node0.Path, top); err != nil {
t.Fatalf("writing topology: %v", err)
}
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatalf("opening cluster: %v", err)
}
state0, err := node0.State()
if err != nil {
t.Fatal(err)
}
// Ensure that node is in state STARTING before the other node joins.
if state0 != string(ClusterStateStarting) {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateStarting, state0)
}
if err := tc.addNode(); err != nil {
t.Fatalf("adding node: %v", err)
}
node1 := tc.Clusters[1]
state1, err := node1.State()
if err != nil {
t.Fatal(err)
}
// Ensure that node comes up in state NORMAL.
if state0 != string(ClusterStateNormal) {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, state0)
} else if state1 != string(ClusterStateNormal) {
t.Errorf("expected node2 state: %v, but got: %v", ClusterStateNormal, state1)
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Multiple nodes, with data", func(t *testing.T) {
tc := NewClusterCluster(t, 0)
if err := tc.addNode(); err != nil {
t.Fatalf("adding node: %v", err)
}
node0 := tc.Clusters[0]
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
// Close TestCluster with defer.
defer func() {
if err := tc.Close(); err != nil {
t.Fatal(err)
}
}()
// Add Bit Data to node0.
if err := tc.CreateField("i", "f", OptFieldTypeDefault()); err != nil {
t.Fatalf("creating field: %v", err)
}
// Each tc.SetBit starts and commits its own Tx.
if err := tc.SetBit("i", "f", 1, 101, nil); err != nil {
t.Fatalf("setting bit: %v", err)
}
if err := tc.SetBit("i", "f", 1, ShardWidth+1, nil); err != nil {
t.Fatalf("setting bit: %v", err)
}
// Before starting the resize, get the CheckSum to use for
// comparison later.
node0Field := node0.holder.Field("i", "f")
node0View := node0Field.view("standard")
node0Fragment := node0View.Fragment(1)
node0Checksum, err := node0Fragment.Checksum()
if err != nil {
t.Fatal(err)
}
idx0 := node0.holder.Index("i")
if idx0 == nil {
t.Fatal(`idx0 was nil, could not retrieve Index("i")`)
}
// addNode needs to block until the resize process has completed.
if err := tc.addNode(); err != nil {
t.Fatalf("adding node: %v", err)
}
node1 := tc.Clusters[1]
state1, err := node1.State()
if err != nil {
t.Fatal(err)
}
state0, err := node0.State()
if err != nil {
t.Fatal(err)
}
// Ensure that nodes come up in state NORMAL.
if state0 != string(ClusterStateNormal) {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, state0)
} else if state1 != string(ClusterStateNormal) {
t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, state1)
}
// INVAR: after node1.State() is normal, the rebalancing should have been done.
expectedTop := &Topology{
nodeIDs: []string{node0.Node.ID, node1.Node.ID},
}
// Verify topology file.
if !reflect.DeepEqual(node0.Topology.nodeIDs, expectedTop.nodeIDs) {
t.Errorf("expected node0 topology: %v, but got: %v", expectedTop.nodeIDs, node0.Topology.nodeIDs)
} else if !reflect.DeepEqual(node1.Topology.nodeIDs, expectedTop.nodeIDs) {
t.Errorf("expected node1 topology: %v, but got: %v", expectedTop.nodeIDs, node1.Topology.nodeIDs)
}
// Bits
// Verify that node-1 contains the fragment (i/f/standard/1) transferred from node-0.
node1Field := node1.holder.Field("i", "f")
node1View := node1Field.view("standard")
node1Fragment := node1View.Fragment(1)
idx1 := node1.holder.Index("i")
if idx1 == nil {
t.Fatal(`idx1 was nil, could not retrieve Index("i")`)
}
// Ensure checksums are the same.
if chksum, err := node1Fragment.Checksum(); err != nil {
t.Fatal(err)
} else if !bytes.Equal(chksum, node0Checksum) {
t.Fatalf("expected standard view checksum to match: %x - %x", chksum, node0Checksum)
}
})
}
func TestAE(t *testing.T) {
t.Run("AbortDoesn'tBlockUninitialized", func(t *testing.T) {
c := newCluster()
@ -1067,26 +710,3 @@ func TestAE(t *testing.T) {
}
})
}
func TestCluster_GetNonPrimaryReplicas(t *testing.T) {
c := newCluster()
c.ReplicaN = 3
topo := NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c)
c.Topology = topo
nNodes := 4
for i := 0; i < nNodes; i++ {
nodeID := fmt.Sprintf("node%d", i)
c.noder.AppendNode(&topology.Node{
ID: nodeID,
URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})
c.Topology.addID(nodeID)
}
partitionID := 256
nonPrimes := topo.GetNonPrimaryReplicas(partitionID)
m := len(nonPrimes)
if m != c.ReplicaN-1 {
t.Fatalf("expected 2 non primes, got %v", m)
}
}

111
cmd/pilosa-chk/chk.go
View file

@ -1,111 +0,0 @@
// Copyright 2020 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package main
import (
"flag"
"fmt"
"log"
"os"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/boltdb"
"github.com/pilosa/pilosa/v2/hash"
"github.com/zeebo/blake3"
)
// pilosa-chk : read boltdb files and print checksums and counts on the keys. With
// -v and -ops and -bits you can display every last bit if you want.
//
// pilosa-chk is deliberately NOT a part of pilosa so that it can run without
// forcing a customer to upgrade or downgrade their installed version.
func main() {
var dir string
var showOpsLog bool
var showBits bool
var showFrags bool
var dirChecksum bool
home := os.Getenv("HOME")
flag.StringVar(&dir, "dir", fmt.Sprintf("%v/.pilosa", home), "pilosa data dir to read")
flag.BoolVar(&showFrags, "v", false, "show the checksum hash for each fragment in each index. Warning: long output")
flag.BoolVar(&showOpsLog, "ops", false, "show the ops log for each fragment. Warning: very long output. Implies -v")
flag.BoolVar(&showBits, "bits", false, "show the hot bits for each fragment. Warning: very, very long output. Implies -v")
flag.BoolVar(&dirChecksum, "dirsum", false, "compute a directory hash")
flag.Parse()
if showBits {
showFrags = true
}
if showOpsLog {
showFrags = true
}
fmt.Printf("opening dir '%v'... this may take a few seconds...\n", dir)
if dirChecksum {
fmt.Printf("path '%v' has dirhash %v\n", dir, hash.HashOfDir(dir))
return
}
fmt.Printf(" the blake-3 hash includes the value of each mapping and the field or partitionID.\n")
holder := pilosa.NewHolder(dir, nil)
holder.OpenTranslateStore = boltdb.OpenTranslateStore
err := holder.Open()
if err != nil {
log.Fatal(err)
}
fmt.Printf("\ncalculating hashes of row and column key translation maps on data from dir '%v'...\n", dir)
var indexes []*pilosa.Index
final := pilosa.NewAllTranslatorSummary()
const verbose = true
const checkKeys = false
const applyKeyRepairs = false
for _, idx := range holder.Indexes() {
asum, err := idx.ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs, nil, "fake-nodeID", 10)
if err != nil {
log.Fatal(err)
}
final.Append(asum)
indexes = append(indexes, idx)
}
final.Sort()
hasher := blake3.New()
fmt.Printf("\nsummary of col/row translations%v:\n", dir)
for _, sum := range final.Sums {
//fmt.Printf("index: %v partitionID: %v blake3-%x keyCount: %v idCount: %v\n", sum.Index, sum.PartitionID, sum.Checksum, sum.KeyCount, sum.IDCount)
_, _ = hasher.Write([]byte(sum.Checksum))
}
var buf [16]byte
_, _ = hasher.Digest().Read(buf[0:])
fmt.Printf("all-checksum = blake3-%x\n", buf)
if showFrags {
for _, idx := range indexes {
fmt.Printf("==============================\n")
fmt.Printf("index: %v\n", idx.Name())
fmt.Printf("==============================\n")
idx.WriteFragmentChecksums(os.Stdout, showBits, showOpsLog, nil, verbose)
}
}
}

330
index.go
View file

@ -17,7 +17,6 @@ package pilosa
import (
"context"
"fmt"
"io"
"io/ioutil"
"os"
"path/filepath"
@ -27,14 +26,11 @@ import (
"time"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/v2/hash"
"github.com/pilosa/pilosa/v2/internal"
"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"
)
@ -707,330 +703,12 @@ func FormatQualifiedIndexName(index string) string {
// Dump prints to stdout the contents of the roaring Containers
// stored in idx. Mostly for debugging.
func (idx *Index) Dump(label string) {
func (i *Index) Dump(label string) {
fileline := FileLine(2)
fmt.Printf("\n%v Dump: %v\n\n", fileline, label)
idx.holder.txf.dbPerShard.DumpAll()
i.holder.txf.dbPerShard.DumpAll()
}
func (idx *Index) SliceOfShards(field, view, viewPath string) (sliceOfShards []uint64, err error) {
// SliceOfShards is based on view.openFragments()
// If we go to a database per shard then index will need this, or
// something like it, to read database files/directories
// and figure out what all the shards are so that a view
// can open its fragments.
file, err := os.Open(filepath.Join(viewPath, "fragments"))
if os.IsNotExist(err) {
return
} else if err != nil {
return nil, errors.Wrap(err, "opening fragments directory")
}
defer file.Close()
fis, err := file.Readdir(0)
if err != nil {
return nil, errors.Wrap(err, "reading fragments directory")
}
for _, fi := range fis {
if fi.IsDir() {
continue
}
// Parse filename into integer.
shard, err := strconv.ParseUint(filepath.Base(fi.Name()), 10, 64)
if err != nil {
idx.holder.Logger.Debugf("WARNING: couldn't use non-integer file as shard in index/field/view %s/%s/%s: %s", idx.name, field, view, fi.Name())
continue
}
sliceOfShards = append(sliceOfShards, shard)
}
return
}
type AllTranslatorSummary struct {
Sums []*TranslatorSummary
RepairNeeded bool
}
func (ats *AllTranslatorSummary) Checksum() string {
ats.Sort()
hasher := blake3.New()
for _, sum := range ats.Sums {
_, _ = hasher.Write([]byte(sum.Checksum))
}
var buf [16]byte
_, _ = hasher.Digest().Read(buf[0:])
return fmt.Sprintf("blake3-%x", buf)
}
func NewAllTranslatorSummary() *AllTranslatorSummary {
return &AllTranslatorSummary{}
}
func (ats *AllTranslatorSummary) Append(b *AllTranslatorSummary) {
ats.Sums = append(ats.Sums, b.Sums...)
ats.RepairNeeded = ats.RepairNeeded || b.RepairNeeded
}
func (ats *AllTranslatorSummary) Sort() {
// return sorted by index then PartitionID then Field
sort.Slice(ats.Sums, func(i, j int) bool {
a := ats.Sums[i]
b := ats.Sums[j]
if a.Index < b.Index {
return true
}
if a.Index > b.Index {
return false
}
// INVAR: a.Index == b.Index
if a.PartitionID < b.PartitionID {
return true
}
if a.PartitionID > b.PartitionID {
return false
}
if a.Field < b.Field {
return true
}
if a.Field > b.Field {
return false
}
return a.NodeID < b.NodeID
})
}
// sums is only guaranteed to be sorted by (index, PartitionID, field) iff err returns nil
func (idx *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs bool, topo *Topology, nodeID string, parallelReaders int) (ats *AllTranslatorSummary, err error) {
idx.mu.RLock()
defer idx.mu.RUnlock()
ats = &AllTranslatorSummary{}
var atsMu sync.Mutex
if verbose {
fmt.Printf("\n# index: %v\n# =================\n", idx.name)
}
pjob := newParallelJobs(parallelReaders)
floop:
for _, fld := range idx.fields {
fld := fld
fun := func(worker int) error {
//vv("ComputeTranslatorSummary() on fld '%v'", fld.name)
sum, err := fld.translateStore.ComputeTranslatorSummaryRows()
if err != nil {
return err
}
sum.Field = fld.name
sum.Index = idx.Name()
sum.Checksum = hash.Blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, fld.name, idx.Name())))
sum.IsColKey = false
if verbose {
fmt.Printf("# row blake3-%v keyN: %5v idN: %5v field: '%v'\n", sum.Checksum, sum.KeyCount, sum.IDCount, fld.name)
}
atsMu.Lock()
ats.Sums = append(ats.Sums, sum)
atsMu.Unlock()
return nil
}
if !pjob.run(fun) {
break floop
}
} // end floop
if verbose {
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
store := store
fun2 := func(worker int) error {
//vv("ComputeTranslatorSummary() running on store.Path = '%v'", store.GetStorePath())
if checkKeys {
prim := snap.PrimaryNodeIndex(partitionID)
primID := topo.nodeIDs[prim]
// note: we fix irrespective of nodeID == primID now, so that we
// get a fine grain report of what maps were off.
if verbose {
// This is pilosa-fsck output, not regular log.
fmt.Printf("# doing analysis of keys on nodeID '%v', and primID '%v'\n", nodeID, primID)
}
changed, err := store.RepairKeys(topo, verbose, applyKeyRepairs)
if err != nil {
return errors.Wrap(err, "ComputeTranslatorSummary() call to store.Repair()")
}
if changed {
atsMu.Lock()
ats.RepairNeeded = true
atsMu.Unlock()
}
}
// key repair has to be above, because we compute the checksum below.
sum, err := store.ComputeTranslatorSummaryCols(partitionID, topo)
if err != nil {
return err
}
if sum == nil {
// probably one of the Noop stores from the tests.
return nil
}
sum.IsColKey = true
sum.PartitionID = partitionID
sum.Index = idx.Name()
sum.StorePath = store.GetStorePath()
sum.NodeID = nodeID
sum.IsPrimary = snap.IsPrimary(nodeID, partitionID)
replicas := snap.NonPrimaryReplicas(partitionID)
for _, replica := range replicas {
if nodeID == replica {
sum.IsReplica = true
break
}
}
sum.Checksum = hash.Blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, partitionID, idx.Name())))
if verbose {
// This is not regular index logging. This is output of the pilosa-fsck tool.
// So it must be printing straight to stdout.
fmt.Printf("# col blake3-%v keyN: %10v idN: %10v paritionID: %03v primary: %03v\n", sum.Checksum, sum.KeyCount, sum.IDCount, partitionID, sum.PrimaryNodeIndex)
}
atsMu.Lock()
ats.Sums = append(ats.Sums, sum)
atsMu.Unlock()
return nil
}
if !pjob.run(fun2) {
break tloop
}
} // end tloop
err = pjob.waitForFinish()
return ats, err
}
// returned by WriteFragmentChecksums
type IndexFragmentSummary struct {
Dir string
NodeID string
Index string
IndexPath string
Frg []*FragSum
RelPath2fsum map[string]*FragSum
}
func (ifs *IndexFragmentSummary) String() (s string) {
s = fmt.Sprintf(`&pilosa.IndexFragmentSummary{
Dir: '%v'
NodeID: '%v'
Index: '%v'
IndexPath: '%v'
`, ifs.Dir, ifs.NodeID, ifs.Index, ifs.IndexPath)
for _, frg := range ifs.Frg {
s += frg.String() + "\n"
}
s += "}\n"
return
}
// used in IndexFragmentSummary
type FragSum struct {
AbsPath string
RelPath string
// critically, NodeID is how pilosa-fsck figures out if this
// fragment should be deleted if it is on a node it should not be.
NodeID string
Index string
Field string
View string
Shard uint64
Hotbits int
Checksum string
Primary int
ScanDone bool // pilosa-fsck will set this once done to avoid repairing multiple times.
}
func (fsum *FragSum) String() (s string) {
return fmt.Sprintf("%#v", fsum)
}
// if verbose, then print to w.
func (idx *Index) WriteFragmentChecksums(w io.Writer, showBits, showOps bool, topo *Topology, verbose bool) (sum *IndexFragmentSummary) {
sum = &IndexFragmentSummary{
Index: idx.name,
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
n := 0
for _, relpath := range paths {
field, view, shard, err := fragmentSpecFromRoaringPath(relpath)
if err != nil {
continue // ignore .meta paths
}
abspath := idx.path + sep + relpath
primary := snap.PrimaryForShardReplication(index, shard)
checksum, hotbits := RoaringFragmentChecksum(abspath, index, field, view, shard)
if verbose {
fmt.Fprintf(w, "# frg blake3-%v field: '%v' view: '%v' shard: %3v hotbits: %10v primary:%03v\n", checksum, field, view, shard, hotbits, primary)
}
fsum := &FragSum{
AbsPath: abspath,
RelPath: relpath,
Index: index,
Field: field,
View: view,
Shard: shard,
Hotbits: hotbits,
Checksum: checksum,
Primary: primary,
}
sum.Frg = append(sum.Frg, fsum)
_, already := sum.RelPath2fsum[relpath]
if already {
panic(fmt.Sprintf("relpath '%v' was already present!?!", relpath))
}
sum.RelPath2fsum[relpath] = fsum
n++
}
if n == 0 {
if verbose {
fmt.Fprintf(w, "empty index '%v'", idx.path)
}
}
return
}
func (idx *Index) Txf() *TxFactory {
return idx.holder.txf
func (i *Index) Txf() *TxFactory {
return i.holder.txf
}

22
mock/translator.go
View file

@ -37,13 +37,6 @@ type TranslateStore struct {
EntryReaderFunc func(ctx context.Context, offset uint64) (pilosa.TranslateEntryReader, error)
}
func (s *TranslateStore) ComputeTranslatorSummaryRows() (sum *pilosa.TranslatorSummary, err error) {
return
}
func (s *TranslateStore) ComputeTranslatorSummaryCols(partitionID int, topo *pilosa.Topology) (sum *pilosa.TranslatorSummary, err error) {
return
}
func (s *TranslateStore) Close() error {
return s.CloseFunc()
}
@ -104,14 +97,6 @@ func (s *TranslateStore) ReadFrom(r io.Reader) (int64, error) {
return 0, nil
}
func (s *TranslateStore) RepairKeys(topo *pilosa.Topology, verbose, applyKeyRepairs bool) (changed bool, err error) {
return
}
func (s *TranslateStore) GetStorePath() string {
return ""
}
var _ pilosa.TranslateEntryReader = (*TranslateEntryReader)(nil)
type TranslateEntryReader struct {
@ -126,10 +111,3 @@ func (r *TranslateEntryReader) Close() error {
func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error {
return r.ReadEntryFunc(entry)
}
func (s *TranslateStore) KeyWalker(walk func(key string, col uint64)) error {
panic("TODO")
}
func (s *TranslateStore) IDWalker(walk func(key string, col uint64)) error {
panic("TODO")
}

10
server.go
View file

@ -585,16 +585,6 @@ func (s *Server) Open() error {
s.syncer.Closing = s.closing
s.syncer.Stats = s.holder.Stats.WithTags("component:HolderSyncer")
err = s.cluster.setup()
if err != nil {
return errors.Wrap(err, "setting up cluster")
}
// Open Cluster management.
if err := s.cluster.waitForStarted(); err != nil {
return errors.Wrap(err, "opening Cluster")
}
// Open holder.
if err := s.holder.Open(); err != nil {
return errors.Wrap(err, "opening Holder")

12
topology/snapshot.go
View file

@ -71,13 +71,11 @@ func NewClusterSnapshot(noder Noder, hasher Hasher, replicas int) *ClusterSnapsh
// ShardToShardPartition returns the shard-partition that the given shard
// belongs to. NOTE: This is DIFFERENT from the key-partition.
func (c *ClusterSnapshot) ShardToShardPartition(index string, shard uint64) int {
return dedupShardToShardPartition(index, shard, c.PartitionN)
return ShardToShardPartition(index, shard, c.PartitionN)
}
// dedupShardToShardParition would ideally be called `shardToShardPartition`, but since
// we can't put this into it's own package yet (see the TODO below about import loops),
// that name conflicts with a function that already exists in the `pilosa` package.
func dedupShardToShardPartition(index string, shard uint64, partitionN int) int {
// ShardToShardParition ...
func ShardToShardPartition(index string, shard uint64, partitionN int) int {
var buf [8]byte
binary.BigEndian.PutUint64(buf[:], shard)
@ -238,14 +236,12 @@ func (c *ClusterSnapshot) PrimaryForColKeyTranslation(index, key string) (primar
}
// TODO: update this comment
// should match cluster.go:1033 cluster.ownsShard(nodeID, index, shard)
// return Nodes(c.shardNodes(index, shard)).ContainsID(nodeID)
func (c *ClusterSnapshot) PrimaryForShardReplication(index string, shard uint64) int {
n := len(c.Nodes)
if n == 0 {
return -1
}
partition := uint64(dedupShardToShardPartition(index, shard, c.PartitionN))
partition := uint64(ShardToShardPartition(index, shard, c.PartitionN))
nodeIndex := c.Hasher.Hash(partition, n)
return nodeIndex
}

49
translate.go
View file

@ -22,6 +22,7 @@ import (
"io/ioutil"
"sync"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pkg/errors"
)
@ -99,16 +100,6 @@ type TranslateStore interface { // TODO: refactor this interface; readonly shoul
// It should read from the reader and replace the data store with
// the read payload.
ReadFrom(io.Reader) (int64, error)
ComputeTranslatorSummaryRows() (sum *TranslatorSummary, err error)
ComputeTranslatorSummaryCols(partitionID int, topo *Topology) (sum *TranslatorSummary, err error)
KeyWalker(walk func(key string, col uint64)) error
IDWalker(walk func(key string, col uint64)) error
RepairKeys(topo *Topology, verbose, applyKeyRepairs bool) (changed bool, err error)
GetStorePath() string
}
// TranslatorSummary is returned, for example from the boltdb string key translators,
@ -188,7 +179,7 @@ func GenerateNextPartitionedID(index string, prev uint64, partitionID, partition
// Try to use the next ID if it is in the same partition.
// Otherwise find ID in next shard that has a matching partition.
for id := prev + 1; ; id += ShardWidth {
if shardToShardPartition(index, id/ShardWidth, partitionN) == partitionID {
if topology.ShardToShardPartition(index, id/ShardWidth, partitionN) == partitionID {
return id
}
}
@ -365,30 +356,6 @@ func NewInMemTranslateStore(index, field string, partitionID, partitionN int) *I
}
}
func (s *InMemTranslateStore) GetStorePath() string {
return ""
}
// KeyWalker executes walk for every pair in the database
func (s *InMemTranslateStore) KeyWalker(walk func(key string, col uint64)) error {
s.mu.RLock()
defer s.mu.RUnlock()
for id, key := range s.keysByID {
walk(key, id)
}
return nil
}
// IDWalker executes walk for every pair in the database
func (s *InMemTranslateStore) IDWalker(walk func(key string, col uint64)) error {
s.mu.RLock()
defer s.mu.RUnlock()
for key, id := range s.idsByKey {
walk(key, id)
}
return nil
}
var _ OpenTranslateStoreFunc = OpenInMemTranslateStore
// OpenInMemTranslateStore returns a new instance of InMemTranslateStore.
@ -397,18 +364,6 @@ func OpenInMemTranslateStore(rawurl, index, field string, partitionID, partition
return NewInMemTranslateStore(index, field, partitionID, partitionN), nil
}
func (s *InMemTranslateStore) ComputeTranslatorSummaryRows() (sum *TranslatorSummary, err error) {
panic("TODO")
}
func (s *InMemTranslateStore) ComputeTranslatorSummaryCols(partitionID int, topo *Topology) (sum *TranslatorSummary, err error) {
panic("TODO")
}
func (s *InMemTranslateStore) RepairKeys(topo *Topology, verbose, applyKeyRepairs bool) (changed bool, err error) {
panic("TODO")
}
func (s *InMemTranslateStore) Close() error {
return nil
}

354
utils_internal_test.go
View file

@ -17,18 +17,13 @@ package pilosa
import (
"bytes"
"fmt"
"io/ioutil"
"path/filepath"
"sync"
"testing"
"time"
"github.com/gogo/protobuf/proto"
pnet "github.com/pilosa/pilosa/v2/net"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/testhook"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pkg/errors"
)
// utilities used by tests
@ -72,7 +67,6 @@ func NewTestCluster(tb testing.TB, n int) *cluster {
c.ReplicaN = 1
c.Hasher = NewTestModHasher()
c.Path = path
c.Topology = NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c)
for i := 0; i < n; i++ {
c.noder.AppendNode(&topology.Node{
@ -113,352 +107,6 @@ func (*TestModHasher) Hash(key uint64, n int) int { return int(key) % n }
func (*TestModHasher) Name() string { return "mod" }
// ClusterCluster represents a cluster of test nodes, each of which
// has a Cluster.
// ClusterCluster implements Broadcaster interface.
type ClusterCluster struct {
Clusters []*cluster
common *commonClusterSettings
mu sync.RWMutex
resizing bool
resizeDone chan struct{}
tb testing.TB
}
type commonClusterSettings struct {
Nodes []*topology.Node
}
func (t *ClusterCluster) CreateIndex(name string) error {
for _, c := range t.Clusters {
if _, err := c.holder.CreateIndexIfNotExists(name, IndexOptions{}); err != nil {
return err
}
}
return nil
}
func (t *ClusterCluster) CreateIndexWithOpt(name string, opt IndexOptions) error {
for _, c := range t.Clusters {
if _, err := c.holder.CreateIndexIfNotExists(name, opt); err != nil {
return err
}
}
return nil
}
func (t *ClusterCluster) CreateField(index, field string, opts FieldOption) error {
for _, c := range t.Clusters {
idx, err := c.holder.CreateIndexIfNotExists(index, IndexOptions{})
if err != nil {
return err
}
if _, err := idx.CreateField(field, opts); err != nil {
return err
}
}
return nil
}
func (t *ClusterCluster) SetBit(index, field string, rowID, colID uint64, x *time.Time) error {
// Determine which node should receive the SetBit.
c0 := t.Clusters[0] // use the first node's cluster to determine shard location.
shard := colID / ShardWidth
nodes := c0.shardNodes(index, shard)
for _, node := range nodes {
c := t.clusterByID(node.ID)
if c == nil {
continue
}
f := c.holder.Field(index, field)
if f == nil {
return fmt.Errorf("index/field does not exist: %s/%s", index, field)
}
if err := func() error {
idx := c.holder.Index(f.index)
shard := colID / ShardWidth
tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
if tx != nil {
defer tx.Rollback()
}
if _, err := f.SetBit(tx, rowID, colID, x); err != nil {
return err
} else if err := tx.Commit(); err != nil {
return err
}
return nil
}(); err != nil {
return err
}
}
return nil
}
func (t *ClusterCluster) clusterByID(id string) *cluster {
for _, c := range t.Clusters {
if c.Node.ID == id {
return c
}
}
return nil
}
// addNode adds a node to the cluster and (potentially) starts a resize job.
func (t *ClusterCluster) addNode() error {
return nil
}
// WriteTopology writes the given topology to disk.
func (t *ClusterCluster) WriteTopology(path string, top *Topology) error {
if buf, err := proto.Marshal(top.encode()); err != nil {
return err
} else if err := ioutil.WriteFile(filepath.Join(path, ".topology"), buf, 0666); err != nil {
return err
}
return nil
}
func (t *ClusterCluster) addCluster(i int, saveTopology bool) (*cluster, error) {
id := fmt.Sprintf("node%d", i)
uri := NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0))
node := &topology.Node{
ID: id,
URI: uri,
}
// add URI to common
//t.common.NodeIDs = append(t.common.NodeIDs, id)
//sort.Sort(t.common.NodeIDs)
// add node to common
t.common.Nodes = append(t.common.Nodes, node)
// create node-specific temp directory
path, err := testhook.TempDirInDir(t.tb, *TempDir, fmt.Sprintf("pilosa-cluster-node-%d-", i))
if err != nil {
return nil, err
}
// holder
h := NewHolder(path, nil)
// cluster
c := newCluster()
c.ReplicaN = 1
c.Hasher = NewTestModHasher()
c.Path = path
c.partitionN = topology.DefaultPartitionN
c.Topology = NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c)
c.holder = h
c.Node = node
// c.Coordinator = t.common.Nodes[0].ID // the first node is the coordinator
c.broadcaster = t.broadcaster(c)
// add nodes
if saveTopology {
for _, n := range t.common.Nodes {
if err := c.addNode(n.ID); err != nil {
return nil, err
}
}
}
// Add this node to the ClusterCluster.
t.Clusters = append(t.Clusters, c)
return c, nil
}
// NewClusterCluster returns a new instance of test.Cluster.
func NewClusterCluster(tb testing.TB, n int) *ClusterCluster {
tc := &ClusterCluster{
common: &commonClusterSettings{},
tb: tb,
}
// add clusters
for i := 0; i < n; i++ {
_, err := tc.addCluster(i, true)
if err != nil {
panic(err)
}
}
return tc
}
// Open opens all clusters in the test cluster.
func (t *ClusterCluster) Open() error {
for _, c := range t.Clusters {
if err := c.open(); err != nil {
return err
}
if err := c.holder.Open(); err != nil {
return err
}
}
return nil
}
// Close closes all clusters in the test cluster.
func (t *ClusterCluster) Close() error {
for _, c := range t.Clusters {
err := c.close()
if err != nil {
return err
}
// Make sure open indexes get shut down too. we wouldn't do
// this normally for a cluster, but we want to for test cases.
c.holder.Close()
}
return nil
}
type bcast struct {
t *ClusterCluster
c *cluster
}
func (b bcast) SendSync(m Message) error {
switch obj := m.(type) {
case *ClusterStatus:
b.t.mu.RLock()
if obj.State == string(ClusterStateNormal) && b.t.resizing {
close(b.t.resizeDone)
}
b.t.mu.RUnlock()
}
return nil
}
func (t *ClusterCluster) broadcaster(c *cluster) broadcaster {
return bcast{
t: t,
c: c,
}
}
// SendAsync is a test implemenetation of Broadcaster SendAsync method.
func (bcast) SendAsync(Message) error {
return nil
}
// SendTo is a test implementation of Broadcaster SendTo method.
func (b bcast) SendTo(to *topology.Node, m Message) error {
switch obj := m.(type) {
case *ResizeInstruction:
err := b.t.FollowResizeInstruction(obj)
if err != nil {
return err
}
case *ClusterStatus:
b.t.mu.RLock()
if obj.State == string(ClusterStateNormal) && b.t.resizing {
close(b.t.resizeDone)
}
b.t.mu.RUnlock()
default:
panic(fmt.Sprintf("message not handled:\n%#v\n", obj))
}
return nil
}
// FollowResizeInstruction is a version of cluster.followResizeInstruction used for testing.
func (t *ClusterCluster) FollowResizeInstruction(instr *ResizeInstruction) error {
// Prepare the return message.
complete := &ResizeInstructionComplete{
JobID: instr.JobID,
Node: instr.Node,
Error: "",
}
// Stop processing on any error.
if err := func() error {
// figure out which node it was meant for, then call the operation on that cluster
// basically need to mimic this: client.RetrieveShardFromURI(context.Background(), src.Index, src.Field, src.View, src.Shard, srcURI)
instrNode := instr.Node
destCluster := t.clusterByID(instrNode.ID)
// Sync the schema received in the resize instruction.
if err := destCluster.holder.applySchema(instr.NodeStatus.Schema); err != nil {
return err
}
// Sync available shards.
for k, is := range instr.NodeStatus.Indexes {
_ = k
for _, fs := range is.Fields {
f := destCluster.holder.Field(is.Name, fs.Name)
// if we don't know about a field locally, log an error because
// fields should be created and synced prior to shard creation
if f == nil {
continue
}
if err := f.AddRemoteAvailableShards(fs.AvailableShards); err != nil {
return errors.Wrap(err, "adding remote available shards")
}
}
}
for _, src := range instr.Sources {
srcCluster := t.clusterByID(src.Node.ID)
srcFragment := srcCluster.holder.fragment(src.Index, src.Field, src.View, src.Shard)
destFragment := destCluster.holder.fragment(src.Index, src.Field, src.View, src.Shard)
if destFragment == nil {
// Create fragment on destination if it doesn't exist.
f := destCluster.holder.Field(src.Index, src.Field)
v := f.view(src.View)
var err error
destFragment, err = v.CreateFragmentIfNotExists(src.Shard)
if err != nil {
return err
}
}
// this is the *test* version of a network call, transferring fragments between
// nodes in a cluster. So it is allowed to be kind of a hack.
// there will be two -rbfdb directories/databases, we need to copy
// from src to dest the fragment. This simulates sending the fragment over the network.
srcIdx := srcCluster.holder.Index(src.Index)
srctx := srcIdx.holder.txf.NewTx(Txo{Write: !writable, Index: srcIdx, Fragment: srcFragment, Shard: srcFragment.shard})
destIdx := destCluster.holder.Index(src.Index)
desttx := destIdx.holder.txf.NewTx(Txo{Write: writable, Index: destIdx, Fragment: destFragment, Shard: destFragment.shard})
citer, _, err := srctx.ContainerIterator(src.Index, src.Field, src.View, src.Shard, 0)
panicOn(err)
d := destFragment
for citer.Next() {
ckey, c := citer.Value()
err := desttx.PutContainer(d.index(), d.field(), d.view(), d.shard, ckey, c)
panicOn(err)
}
citer.Close()
panicOn(desttx.Commit())
srctx.Rollback()
}
return nil
}(); err != nil {
complete.Error = err.Error()
}
node := instr.Primary
return bcast{t: t}.SendTo(node, complete)
}
var _ = NewTestClusterWithReplication // happy linter
func NewTestClusterWithReplication(tb testing.TB, nNodes, nReplicas, partitionN int) (c *cluster, cleaner func()) {
@ -478,7 +126,6 @@ func NewTestClusterWithReplication(tb testing.TB, nNodes, nReplicas, partitionN
c.Hasher = &topology.Jmphasher{}
c.Path = path
c.partitionN = partitionN
c.Topology = NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c)
for i := 0; i < nNodes; i++ {
nodeID := fmt.Sprintf("node%d", i)
@ -486,7 +133,6 @@ func NewTestClusterWithReplication(tb testing.TB, nNodes, nReplicas, partitionN
ID: nodeID,
URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})
c.Topology.addID(nodeID)
}
cNodes := c.noder.Nodes()