diff --git a/api.go b/api.go index a3d9107a6..b86b11344 100644 --- a/api.go +++ b/api.go @@ -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(), } } diff --git a/boltdb/translate.go b/boltdb/translate.go index 40ba72cd3..00f0e657c 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -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) - -} diff --git a/boltdb/translate_test.go b/boltdb/translate_test.go index 1da4bf0ed..6908da6ac 100644 --- a/boltdb/translate_test.go +++ b/boltdb/translate_test.go @@ -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 -} diff --git a/cluster.go b/cluster.go index 94a282d7b..039607ec0 100644 --- a/cluster.go +++ b/cluster.go @@ -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 diff --git a/cluster_internal_test.go b/cluster_internal_test.go index 68c44a7fb..938c65c0b 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -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) - } -} diff --git a/cmd/pilosa-chk/chk.go b/cmd/pilosa-chk/chk.go deleted file mode 100644 index 353b33b3b..000000000 --- a/cmd/pilosa-chk/chk.go +++ /dev/null @@ -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) - } - } -} diff --git a/index.go b/index.go index 457869d6a..c81419c46 100644 --- a/index.go +++ b/index.go @@ -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 } diff --git a/mock/translator.go b/mock/translator.go index 9be03715a..e7e88644f 100644 --- a/mock/translator.go +++ b/mock/translator.go @@ -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") -} diff --git a/server.go b/server.go index 74ebb76c2..b43e61de6 100644 --- a/server.go +++ b/server.go @@ -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") diff --git a/topology/snapshot.go b/topology/snapshot.go index fc1a2d83f..b7d865335 100644 --- a/topology/snapshot.go +++ b/topology/snapshot.go @@ -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 } diff --git a/translate.go b/translate.go index e8ed1f083..dced91507 100644 --- a/translate.go +++ b/translate.go @@ -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 } diff --git a/utils_internal_test.go b/utils_internal_test.go index 26414ad11..f0f4ef7d4 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -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()