diff --git a/.gitignore b/.gitignore index 7a41479b9..bee19753e 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ vendor build *~ lattice +release-pilosa-fsck.*.*.tar.gz diff --git a/Makefile b/Makefile index 52d246c64..085cef0b8 100644 --- a/Makefile +++ b/Makefile @@ -55,7 +55,7 @@ testv-race: topt-race testvsub-race # find which test is hung/deadlocked. # testvsub: - set -e; for i in ctl http pg pql rbf roaring server sql txkey; do \ + set -e; for i in boltdb ctl http pg pql rbf roaring server sql txkey; do \ echo; echo "___ testing subpkg $$i"; \ cd $$i; pwd; \ go test -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) $(NOCHECKPTR) -v -timeout 60m || break; \ @@ -64,7 +64,7 @@ testvsub: done testvsub-race: - set -e; for i in ctl http pg pql rbf roaring server sql txkey; do \ + set -e; for i in boltdb ctl http pg pql rbf roaring server sql txkey; do \ echo; echo "___ testing subpkg $$i -race"; \ cd $$i; pwd; \ go test -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) $(NOCHECKPTR) -v -race -timeout 60m || break; \ @@ -201,6 +201,9 @@ pilosa-keydump: pilosa-chk: go install -tags='$(BUILD_TAGS)' -ldflags $(LDFLAGS) $(FLAGS) ./cmd/pilosa-chk +pilosa-fsck: + cd ./cmd/pilosa-fsck && make install && make release + # Run Pilosa tests inside Docker container docker-test: docker run --rm -v $(PWD):/go/src/$(CLONE_URL) -w /go/src/$(CLONE_URL) golang:$(GO_VERSION) go test -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) ./... diff --git a/boltdb/translate.go b/boltdb/translate.go index 78e81ed7a..34d1d270d 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -17,9 +17,12 @@ package boltdb import ( "bytes" "context" + "fmt" "io" + "io/ioutil" "os" "path/filepath" + "sort" "sync" "time" @@ -27,8 +30,13 @@ import ( "github.com/pilosa/pilosa/v2" "github.com/pkg/errors" "github.com/zeebo/blake3" + + "runtime/pprof" ) +var _ = ioutil.TempFile +var _ = pprof.StartCPUProfile + var ( // ErrTranslateStoreClosed is returned when reading from an TranslateEntryReader // and the underlying store is closed. @@ -90,6 +98,10 @@ 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{ @@ -485,7 +497,7 @@ func findKeyByID(bkt *bolt.Bucket, id uint64) string { return string(boltKey) } -func (s *TranslateStore) ComputeTranslatorSummary() (sum *pilosa.TranslatorSummary, err error) { +func (s *TranslateStore) ComputeTranslatorSummaryRows() (sum *pilosa.TranslatorSummary, err error) { sum = &pilosa.TranslatorSummary{} hasher := blake3.New() @@ -524,3 +536,787 @@ func (s *TranslateStore) ComputeTranslatorSummary() (sum *pilosa.TranslatorSumma 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)) + } + firstPrimary := topo.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 := topo.GetPrimaryForColKeyTranslation(s.index, ks) + if firstPrimary < 0 { + firstPrimary = primary + } else { + if primary != firstPrimary { + panic(fmt.Sprintf("primary (%v) != firstPrimary (%v); key='%v', id=%v, shard=%v; partitionID=%v", primary, firstPrimary, ks, id, shard, partitionID)) + } + } + + // Verify the invariant that the primaries agree. Just a sanity check. + primaryForShard := topo.GetPrimaryForShardReplication(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)) + } + } + + for key2, id2 := range fwd2 { + //vv("makeStringKeyChanges on fwd2, key2='%v', id2=%x", key2, id2) + isPrimary := false + if topo != nil { + primary := topo.GetPrimaryForColKeyTranslation(s.index, key2) + 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 8c3915f00..1e0803b1b 100644 --- a/boltdb/translate_test.go +++ b/boltdb/translate_test.go @@ -28,6 +28,8 @@ import ( "github.com/pilosa/pilosa/v2/boltdb" ) +//var vv = pilosa.VV + func TestTranslateStore_TranslateKey(t *testing.T) { s := MustOpenNewTranslateStore() defer MustCloseTranslateStore(s) @@ -533,8 +535,7 @@ func TestCryptoHashPerKey(t *testing.T) { } // done with setup - - sum, err := s.ComputeTranslatorSummary() + sum, err := s.ComputeTranslatorSummaryCols(0, pilosa.NewTopology(&pilosa.Jmphasher{}, pilosa.DefaultPartitionN, 1, nil)) if err != nil { panic(err) } @@ -558,3 +559,123 @@ func TestCryptoHashPerKey(t *testing.T) { } } + +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 10bc2b72d..9059c8732 100644 --- a/cluster.go +++ b/cluster.go @@ -260,7 +260,7 @@ type cluster struct { // nolint: maligned // newCluster returns a new instance of Cluster with defaults. func newCluster() *cluster { return &cluster{ - Hasher: &jmphasher{}, + Hasher: &Jmphasher{}, partitionN: DefaultPartitionN, ReplicaN: 1, @@ -978,12 +978,13 @@ func (c *cluster) translationNodes(to *cluster) (map[string][]*translationResize return m, nil } -// shardPartition returns the partition that a shard belongs to. -func (c *cluster) shardPartition(index string, shard uint64) int { - return shardPartition(index, shard, c.partitionN) +// 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 shardPartition(index string, shard uint64, partitionN int) int { +func shardToShardPartition(index string, shard uint64, partitionN int) int { var buf [8]byte binary.BigEndian.PutUint64(buf[:], shard) @@ -994,12 +995,13 @@ func shardPartition(index string, shard uint64, partitionN int) int { return int(h.Sum64() % uint64(partitionN)) } -// keyPartition returns the partition that a key belongs to. -func (c *cluster) keyPartition(index, key string) int { - return keyPartition(index, key, c.partitionN) +// keyPartition returns the key-partition that a key belongs to. +// NOTE: the key-partition is DIFFERENT from the shard-partition. +func (topo *Topology) KeyPartition(index, key string) int { + return keyToKeyPartition(index, key, topo.PartitionN) } -func keyPartition(index, key string, partitionN int) int { +func keyToKeyPartition(index, key string, partitionN int) int { // Hash the bytes and mod by partition count. h := fnv.New64a() _, _ = h.Write([]byte(index)) @@ -1009,7 +1011,7 @@ func keyPartition(index, key string, partitionN int) int { // idPartition returns the partition that an id belongs to. func (c *cluster) idPartition(index string, id uint64) int { - return shardPartition(index, id/ShardWidth, c.partitionN) + return shardToShardPartition(index, id/ShardWidth, c.partitionN) } // ShardNodes returns a list of nodes that own a fragment. Safe for concurrent use. @@ -1021,7 +1023,7 @@ func (c *cluster) ShardNodes(index string, shard uint64) []*Node { // shardNodes returns a list of nodes that own a fragment. unprotected func (c *cluster) shardNodes(index string, shard uint64) []*Node { - return c.partitionNodes(c.shardPartition(index, shard)) + return c.partitionNodes(c.shardToShardPartition(index, shard)) } // KeyNodes returns a list of nodes that own a fragment. Safe for concurrent use. @@ -1033,7 +1035,7 @@ func (c *cluster) KeyNodes(index, key string) []*Node { // keyNodes returns a list of nodes that own a key. unprotected func (c *cluster) keyNodes(index, key string) []*Node { - return c.partitionNodes(c.keyPartition(index, key)) + return c.partitionNodes(c.Topology.KeyPartition(index, key)) } // ownsShard returns true if a host owns a fragment. @@ -1077,8 +1079,14 @@ func (c *cluster) partitionNodes(partitionID int) []*Node { } // Determine primary owner node. - nodeIndex := c.Hasher.Hash(uint64(partitionID), nodeN) - + 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([]*Node, 0, replicaN) for i := 0; i < replicaN; i++ { @@ -1109,11 +1117,66 @@ func (c *cluster) unprotectedPrimaryPartitionNode(partition int) *Node { return nil } +func (topo *Topology) IsPrimary(nodeID string, partitionID int) bool { + primary := topo.PrimaryNodeIndex(partitionID) + return nodeID == topo.nodeIDs[primary] +} + +func (topo *Topology) PrimaryNodeIndex(partitionID int) (nodeIndex int) { + n := len(topo.nodeIDs) + if n == 0 { + if topo.cluster != nil { + n = len(topo.cluster.nodes) + } + } + nodeIndex = topo.Hasher.Hash(uint64(partitionID), n) + return +} + +func (topo *Topology) GetNonPrimaryReplicas(partitionID int) (nonPrimaryReplicas []string) { + + primary := topo.PrimaryNodeIndex(partitionID) + nodeN := len(topo.nodeIDs) + + // Collect nodes around the ring. + for i := 1; i < nodeN; i++ { + nodeID := topo.nodeIDs[(primary+i)%nodeN] + if i < topo.ReplicaN { + nonPrimaryReplicas = append(nonPrimaryReplicas, nodeID) + } + } + return +} + +// the map replicaNodeIDs[nodeID] will have a true value for the primary nodeID, and false for others. +func (topo *Topology) GetReplicasForPrimary(primary int) (replicaNodeIDs, nonReplicas map[string]bool) { + if primary < 0 { + // no nodes anyway + return + } + replicaNodeIDs = make(map[string]bool) + nonReplicas = make(map[string]bool) + + nodeN := len(topo.nodeIDs) + + // Collect nodes around the ring. + for i := 0; i < nodeN; i++ { + nodeID := topo.nodeIDs[(primary+i)%nodeN] + if i < topo.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 *Node) []uint64 { var shards []uint64 _ = availableShards.ForEach(func(i uint64) error { - p := c.shardPartition(index, i) + p := c.shardToShardPartition(index, i) // Determine the nodes for partition. nodes := c.partitionNodes(p) for _, n := range nodes { @@ -1133,10 +1196,10 @@ type Hasher interface { } // jmphasher represents an implementation of jmphash. Implements Hasher. -type jmphasher struct{} +type Jmphasher struct{} // Hash returns the integer hash for the given key. -func (h *jmphasher) Hash(key uint64, n int) int { +func (h *Jmphasher) Hash(key uint64, n int) int { b, j := int64(-1), int64(0) for j < int64(n) { b = j @@ -1202,7 +1265,7 @@ func (c *cluster) waitForStarted() error { c.logger.Printf("%v wait for joining to complete", c.Node.ID) <-c.joining - c.logger.Printf("joining has completed") + c.logger.Printf("joining has completed. I am NodeID '%v'", c.Node.ID) } return nil } @@ -1853,6 +1916,8 @@ func (n nodeIDs) ContainsID(id string) bool { } // 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 @@ -1862,14 +1927,51 @@ type Topology struct { // 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 Hasher + // The number of partitions in the cluster. + PartitionN int + // The number of replicas a partition has. + ReplicaN int + + // can be nil + cluster *cluster } -func newTopology() *Topology { +// 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 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) 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() @@ -1933,7 +2035,7 @@ func (t *Topology) encode() *internal.Topology { func (c *cluster) loadTopology() error { buf, err := ioutil.ReadFile(filepath.Join(c.Path, ".topology")) if os.IsNotExist(err) { - c.Topology = newTopology() + c.Topology = NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c) return nil } else if err != nil { return errors.Wrap(err, "reading file") @@ -1943,7 +2045,7 @@ func (c *cluster) loadTopology() error { if err := proto.Unmarshal(buf, &pb); err != nil { return errors.Wrap(err, "unmarshalling") } - top, err := decodeTopology(&pb) + top, err := DecodeTopology(&pb, c.Hasher, c.partitionN, c.ReplicaN, c) if err != nil { return errors.Wrap(err, "decoding") } @@ -1954,7 +2056,6 @@ func (c *cluster) loadTopology() error { // saveTopology writes the current topology to disk. unprotected. func (c *cluster) saveTopology() error { - if err := os.MkdirAll(c.Path, 0777); err != nil { return errors.Wrap(err, "creating directory") } @@ -2461,6 +2562,27 @@ 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 (topo *Topology) GetPrimaryForColKeyTranslation(index, key string) (primary int) { + partitionID := topo.KeyPartition(index, key) + return topo.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) @@ -2472,7 +2594,7 @@ func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, ke // Split keys by partition. keysByPartition := make(map[int][]string, c.partitionN) for key := range keySet { - partitionID := c.keyPartition(indexName, key) + partitionID := c.Topology.KeyPartition(indexName, key) keysByPartition[partitionID] = append(keysByPartition[partitionID], key) } @@ -2651,12 +2773,13 @@ func encodeTopology(topology *Topology) *internal.Topology { } } -func decodeTopology(topology *internal.Topology) (*Topology, error) { +// the cluster c is optional but give it if you have it. +func DecodeTopology(topology *internal.Topology, hasher Hasher, partitionN, replicaN int, c *cluster) (*Topology, error) { if topology == nil { return nil, nil } - t := newTopology() + t := NewTopology(hasher, partitionN, replicaN, c) t.clusterID = topology.ClusterID t.nodeIDs = topology.NodeIDs sort.Slice(t.nodeIDs, diff --git a/cluster_internal_test.go b/cluster_internal_test.go index 003e7f668..a9ec930be 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -381,7 +381,7 @@ func TestCluster_Partition(t *testing.T) { c := newCluster() c.partitionN = partitionN - partitionID := c.shardPartition(index, shard) + partitionID := c.shardToShardPartition(index, shard) if partitionID < 0 || partitionID >= partitionN { t.Errorf("partition out of range: shard=%d, p=%d, n=%d", shard, partitionID, partitionN) } @@ -411,7 +411,7 @@ func TestHasher(t *testing.T) { {0x0ddc0ffeebadf00d, []int{0, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 15, 15, 15, 15}}, } { for i, v := range tt.bucket { - hasher := &jmphasher{} + hasher := &Jmphasher{} if got := hasher.Hash(tt.key, i+1); got != v { t.Errorf("hash(%v,%v)=%v, want %v", tt.key, i+1, got, v) } @@ -1053,3 +1053,27 @@ func TestCluster_confirmNodeDownDown(t *testing.T) { t.Errorf("expected node to be down") } } + +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.nodes = append(c.nodes, &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 index f0d5d9258..7e315700c 100644 --- a/cmd/pilosa-chk/chk.go +++ b/cmd/pilosa-chk/chk.go @@ -76,8 +76,10 @@ func main() { final := pilosa.NewAllTranslatorSummary() const verbose = true + const checkKeys = false + const applyKeyRepairs = false for _, idx := range holder.Indexes() { - asum, err := idx.ComputeTranslatorSummary(verbose) + asum, err := idx.ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs, nil, "fake-nodeID") if err != nil { log.Fatal(err) } @@ -103,7 +105,7 @@ func main() { fmt.Printf("==============================\n") fmt.Printf("index: %v\n", idx.Name()) fmt.Printf("==============================\n") - idx.WriteFragmentChecksums(os.Stdout, showBits, showOpsLog) + idx.WriteFragmentChecksums(os.Stdout, showBits, showOpsLog, nil, verbose) } } } diff --git a/cmd/pilosa-fsck/Makefile b/cmd/pilosa-fsck/Makefile new file mode 100644 index 000000000..1b1dcf14c --- /dev/null +++ b/cmd/pilosa-fsck/Makefile @@ -0,0 +1,36 @@ +.PHONY: install build release + +CLONE_URL=github.com/pilosa/pilosa +VERSION := $(shell git describe --tags 2> /dev/null || echo unknown) +LATTICE_COMMIT := $(shell git -C lattice rev-parse --short HEAD 2>/dev/null) +VARIANT = Molecula +VERSION_ID = $(VERSION)-$(GOOS)-$(GOARCH) +BRANCH := $(if $(TRAVIS_BRANCH),$(TRAVIS_BRANCH),$(if $(CIRCLE_BRANCH),$(CIRCLE_BRANCH),$(shell git rev-parse --abbrev-ref HEAD))) +BRANCH_ID := $(BRANCH)-$(GOOS)-$(GOARCH) +BUILD_TIME := $(shell date -u +%FT%T%z) +SHARD_WIDTH = 20 +COMMIT := $(shell git describe --exact-match >/dev/null 2>&1 || git rev-parse --short HEAD) +LDFLAGS="-X github.com/pilosa/pilosa/v2.Version=$(VERSION) -X github.com/pilosa/pilosa/v2.BuildTime=$(BUILD_TIME) -X github.com/pilosa/pilosa/v2.Variant=$(VARIANT) -X github.com/pilosa/pilosa/v2.Commit=$(COMMIT) -X github.com/pilosa/pilosa/v2.LatticeCommit=$(LATTICE_COMMIT)" +GOOS = $(shell go env GOOS) + +# Install pilosa-fsck +install: + go install -tags='$(BUILD_TAGS)' -ldflags $(LDFLAGS) $(FLAGS) + +# Compile pilosa-fsck +build: + go build -tags='$(BUILD_TAGS)' -ldflags $(LDFLAGS) $(FLAGS) + +REL = release-pilosa-fsck.$(COMMIT).$(GOOS) + +release: + mkdir $(REL) + cd release-pilosa-fsck; tar cf - . |(cd ../$(REL); tar xf - ) + go build -tags='$(BUILD_TAGS)' -ldflags $(LDFLAGS) $(FLAGS) -o $(REL)/pilosa-fsck + tar cf - $(REL) | gzip > $(REL).tar.gz + rm -rf $(REL) + mv $(REL).tar.gz ../.. + +clean: + find . -name pilosa-fsck | xargs rm -f + rm -f release-pilosa-fsck*.tar.gz diff --git a/cmd/pilosa-fsck/fsck.go b/cmd/pilosa-fsck/fsck.go new file mode 100644 index 000000000..daf5b7b69 --- /dev/null +++ b/cmd/pilosa-fsck/fsck.go @@ -0,0 +1,952 @@ +// 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" + "io" + "io/ioutil" + "log" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "github.com/dustin/go-humanize" + "github.com/gogo/protobuf/proto" + "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/boltdb" + "github.com/pilosa/pilosa/v2/internal" + "github.com/pilosa/pilosa/v2/server" + "github.com/pkg/errors" + "github.com/zeebo/blake3" +) + +// pilosa-fsck : +// an external customer tool (originally for Q2) to do 2 jobs: +// Given a set of cluster backups (and their .id and .topology files) +// mounted on the same file system, we can: +// 1) scan for fragment differences between the primary and its replicas (default); or +// 2) repair those differences by overwriting the replcas with the primary fragments (if -fix is given). +// +// 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. + +// FsckConfig configures the dumpcols() and/or read() runs. +type FsckConfig struct { + Fix bool // -fix + FixCol bool // -fixcol + + Colkeydump bool // -col + + // -col column key dump only options: + Dir string + Index string + PartitionID int + ShowHeader bool + ShowKey bool + ShowID bool + + // not flags, just the Args() left after all other flags. Should be the list + // of pilosa (holder) directories for the cluster. + Dirs []string + + Verbose bool // -v + Quiet bool // -q + + // manual workaround for not having PilosaConfigPath, if really need be. + ReplicaN int // -replicas + PilosaConfigPath string // -config + + topo *pilosa.Topology +} + +// call DefineFlags before myflags.Parse() +func (cfg *FsckConfig) DefineFlags(fs *flag.FlagSet) { + fs.BoolVar(&cfg.Fix, "fix", false, "(warning: alters the backed-up node images on disk) copy primary data to replicas to create a consistent cluster. Implies -fixcol") + fs.BoolVar(&cfg.FixCol, "fixcol", false, "(warning: alters the backed-up node images on disk) repair string key translation tables. Skip repair of index data.") + fs.BoolVar(&cfg.Verbose, "v", false, "be very verbose during analysis") + fs.BoolVar(&cfg.Quiet, "q", false, "be very quiet") + + fs.IntVar(&cfg.ReplicaN, "replicas", 0, "(required) manually entered replicaN; the number of replicas maintained in the cluster. Must be the same as the [cluster] 'replicas = R' entry in the pilosa.conf file for the cluster.") + + fs.StringVar(&cfg.PilosaConfigPath, "config", "", "(required: -replicas or -config, with -config preferred) path to the pilosa.conf for the cluster (e.g. /etc/pilosa.conf)") + + fs.Usage = func() { + fmt.Fprintf(os.Stderr, "pilosa-fsck version: %v\n\n", pilosa.VersionInfo()) + fmt.Fprintf(os.Stderr, `Use: pilosa-fsck -replicas R {-fix} {-fixcol} {-q} {-v} /backup/1/.pilosa /backup/2/.pilosa ... /backup/N/.pilosa + + -fix + (warning: alters the backed-up node images on disk) copy primary data to replicas to create a consistent cluster. Implies -fixcol + + -fixcol + (warning: alters the backed-up node images on disk) repair string key translation tables. Skip repair of index data. + + -replicas R + (required) R is a positive integer, giving the replicaN or replicator factor for the cluster. This is + the number of replicas maintained in the cluster. Must be the same as the + [cluster] 'replicas = R' entry shared across all the pilosa.conf files on each node. + + -v + be very verbose during analysis + -q + be very quiet during analysis and repair + +`) + /* + key translation dump options, usually only used by developers debugging key translation: + + -col + (optional) display column keys (very long output) + -dir string + (optional; requires -col), one pilosa data dir to read (default "/home/ubuntu/.pilosa") + -header + (optional; requires -col), display header + -id + (optional; requires -col), dump reverse mapping id->key + -index string + (optional; requires -col), index name (default "i") + -key + (optional; requires -col), dump forward mapping key->id + -partition int + (optional; requires -col), partition id to dump + + */ + fmt.Fprintf(os.Stderr, ` +Welcome to pilosa-fsck. This is a scan and repair +tool that is modeled after the classic unix file +system utility fsck. + +WARNING: DO NOT RUN ON A LIVE SYSTEM. + +The most important point to remember is that analysis +and repair must be done *offline*. + +Just as fsck must be run on an unmounted disk, +pilosa-fsck must be run on a backup. It must +not be run on the directories where a live Pilosa system +is serving queries. Instead, take a backup first. +A backup is a set of N cluster-node directories that have been +copied from your live system. They must all +be visible and mounted on one filesystem together. + +pilosa-fsck can be run in scan-mode (without -fix or -fixcol), +or in repair-mode with -fix (or -fixcol). The console output +supplies a shell script documenting the analysis +and showing what data changes would be made. If a +fix has been requested, those fixes will have +been applied during the run. The output then serves +as documentation of what has been updated. If +a fix has not been requested (in other words, neither +-fix nor -fixcol was given) then no changes will +have been made to the backups. The fragment level +sync can be completed next by running the script +if you wish. The -fixcol fixes can only be +applied by doing a -fixcol run of pilosa-fsck. + +REQUIRED COMMAND LINE ARGUMENTS + +The paths to all the top-level pilosa +directories in a cluster must be given on the command +line. The -replicas R flag is also always required. It +must be correct for your cluser. Here R is the same as +the [cluster] stanza "replicas = R" line from your +pilosa.conf. + +Example: + +Suppose you are ready to run pilosa-fsck: +you have taken a backup of your four node Pilosa +cluster and stored it all on one filesystem with +all nodes visible and uncompressed. This +is a pre-requisite to running pilosa-fsck. +Let's suppose we have replication R = 3 set. +In this example, have stored our backed-up directories in + +/backup/molecula + +and the four node backups are in +subdirectories node1/ node2/ node3/ node4/ under this: + +/backup/molecula/node1/ +/backup/molecula/node1/.pilosa/.id +/backup/molecula/node1/.pilosa/.topology +/backup/molecula/node1/.pilosa/myindex + +/backup/molecula/node2/ +/backup/molecula/node2/.pilosa/.id +/backup/molecula/node2/.pilosa/.topology +/backup/molecula/node2/.pilosa/myindex + +/backup/molecula/node3/ +/backup/molecula/node3/.pilosa/.id +/backup/molecula/node3/.pilosa/.topology +/backup/molecula/node3/.pilosa/myindex + +/backup/molecula/node4/ +/backup/molecula/node4/.pilosa/.id +/backup/molecula/node4/.pilosa/.topology +/backup/molecula/node4/.pilosa/myindex + +NOTE: your .pilosa directories need not be named .pilosa. They can +be something else, such as when the -d flag to pilosa server was used. +The .id and .topology and index directories must be found directly underneath. + +Then a typical invocation to scan a cluster backup for issues: + +$ cd /backup/molecula/ +$ pilosa-fsck -replicas 3 node1/.pilosa node2/.pilosa node3/.pilosa node4/.pilosa + +A typical invocation to repair the replication in the same backup: + +$ pilosa-fsck -replicas 3 -fix node1/.pilosa node2/.pilosa node3/.pilosa node4/.pilosa + +In both cases, the .id and .topology files must +be present in the backups. + +KEY REPAIR NOTE + +While the output of pilosa-fsck wihtout -fix or -fixcol +gives a script showing the index fragment repair operations +that can be applied by (cp/rm) shell commands subsequently, +this script alone is an incomplete repair. It does not +address string key tranlsation repairs. For a complete repair, +a run of pilosa-fsck with the -fixcol or -fix flags +will be required. + +A note about the -fixcol key translation repairs: these are fine +grained operations on the internal databases that do not have +corresponding (cp/rm) shell commands. Therefore a run of pilosa-fsck +with the -fix or -fixcol flag is required to repair these. +`) + } +} + +// call c.ValidateConfig() after myflags.Parse() +func (c *FsckConfig) ValidateConfig() error { + if c.Fix { + c.FixCol = true + } + if c.ReplicaN == 0 && c.PilosaConfigPath == "" { + return fmt.Errorf("must supply -replicas with the replica count from your pilosa.conf (positive integer count)") + } + + if c.ReplicaN == 0 && c.PilosaConfigPath != "" { + + if !FileExists(c.PilosaConfigPath) { + return fmt.Errorf(" -config path '%v' does not exist", c.PilosaConfigPath) + } + by, err := ioutil.ReadFile(c.PilosaConfigPath) + if err != nil { + return fmt.Errorf("error: could not read the -config path '%v': '%v'", c.PilosaConfigPath, err) + } + srvcfg, err := server.ParseConfig(string(by)) + if err != nil { + //vv("warning: -config path '%v' problem, could not parse toml: '%v'", c.PilosaConfigPath, err) + + // fall back to manual parsing of config + lines := strings.Split(string(by), "\n") + clusterStart := -1 + for i, line := range lines { + if strings.Contains(line, `[cluster]`) { + clusterStart = i + } + if i > clusterStart { + if strings.Contains(line, "replicas") { + split := strings.Split(line, "=") + ns := strings.TrimSpace(split[1]) + n, err := strconv.Atoi(ns) + if err != nil { + return fmt.Errorf("error: could not parse the replicaN from line %v in -config path '%v' (%v): '%v'", i+1, c.PilosaConfigPath, line, err) + } + c.ReplicaN = n + } + } + } + } else { + c.ReplicaN = srvcfg.Cluster.ReplicaN + } + if c.ReplicaN == 0 { + return fmt.Errorf("error: -config path '%v' did not list the Replica count: cannot be 0. See the [cluster] section, the 'replicas = R' line.", c.PilosaConfigPath) + } + //vv("c.ReplicaN = %v", c.ReplicaN) + } + return nil +} + +var ProgramName = "pilosa-fsck" + +func main() { + + myflags := flag.NewFlagSet(ProgramName, flag.ContinueOnError) + cfg := &FsckConfig{} + cfg.DefineFlags(myflags) + + err := myflags.Parse(os.Args[1:]) + if err != nil { + fmt.Fprintf(os.Stderr, "\n%v\n", err.Error()) + os.Exit(1) + } + err = cfg.ValidateConfig() + if err != nil { + fmt.Fprintf(os.Stderr, "%s error: %s\n", ProgramName, err) + os.Exit(1) + } + dirs := myflags.Args() + nDir := len(dirs) + if nDir <= 0 && !cfg.Colkeydump { + fmt.Fprintf(os.Stderr, "error: %v command line arguments missing error: provide all of the top-level pilosa directories for the cluster as command line arguments.\n", ProgramName) + os.Exit(1) + } + + cmdline := strings.Join(os.Args, " ") + + // make sure all the dir are distinct + dup := make(map[string]bool) + for _, dir := range dirs { + if dup[dir] { + fmt.Fprintf(os.Stderr, "%v error: duplicate data directory '%v' given in command line '%v'. Each backup directory must be distinct.\n", ProgramName, dir, cmdline) + os.Exit(1) + } else { + dup[dir] = true + } + } + + fmt.Fprintf(os.Stdout, "#!/bin/bash\n\n# pilosa-fsck version: %v\n", pilosa.VersionInfo()) + cwd, err := os.Getwd() + if err != nil { + fmt.Fprintf(os.Stderr, "error: could not read current dir: '%v'\n", err) + os.Exit(1) + } + fmt.Fprintf(os.Stdout, "# cwd: %v\n", cwd) + fmt.Fprintf(os.Stdout, "# command line: %v\n", cmdline) + t0 := time.Now() + fmt.Fprintf(os.Stdout, "# started at %v\n\n", t0.Format(RFC3339MsecTz0)) + defer func() { + fmt.Fprintf(os.Stdout, "# finished at %v (elapsed %v)\n\n", time.Now().Format(RFC3339MsecTz0), time.Since(t0)) + }() + cfg.Dirs = dirs + + _, err = cfg.Run() + if err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } +} + +func (cfg *FsckConfig) Run() (fixNeeded bool, err error) { + + if cfg.Colkeydump { + cfg.dumpcols() + } + + perNodeIndexMaps, clusterNodes, ats, err := cfg.read() + if err != nil { + return false, err + } + + if cfg.FixCol { + err := cfg.RepairTranslationStores(ats) + if err != nil { + return false, fmt.Errorf("error fixing key translation stores with cfg.RepairTranslationStores(): '%v'\n", err) + } + } + + //vv("perNodeIndexMaps='%#v', clusterNodes='%#v'", perNodeIndexMaps, clusterNodes) + + fixme, reports, err := cfg.analyze(clusterNodes, perNodeIndexMaps, ats) + if err != nil { + return false, fmt.Errorf("error in FsckConfig.analyze(): '%v'", err) + } + fixNeeded = ats.RepairNeeded || fixme + for _, report := range reports { + fmt.Printf("%v\n", report) + } + if len(reports) == 0 { + fmt.Fprintf(os.Stderr, "pilosa-fsck: no index found to analyze. cmdline was: %v\n", strings.Join(os.Args, " ")) + } + return +} + +var _ = (&FsckConfig{}).dumpAts + +func (cfg *FsckConfig) dumpAts(ats *pilosa.AllTranslatorSummary) { + fmt.Printf("# dumpAts: RepairNeeded=%v\n", ats.RepairNeeded) + for _, sum := range ats.Sums { + fmt.Printf("# sum = '%#v'\n", sum) + } + +} + +type group struct { + elem []*pilosa.TranslatorSummary + partitionID int +} + +func (g *group) String() (s string) { + for i, e := range g.elem { + s += fmt.Sprintf("partition %v, group elem [%v] out of %v: %v\n", g.partitionID, i, len(g.elem), e.String()) + } + return +} + +func (cfg *FsckConfig) RepairTranslationStores(ats *pilosa.AllTranslatorSummary) (err error) { + + verbose := cfg.Verbose + + m := make(map[int]*group) + for _, sum := range ats.Sums { + if !sum.IsColKey { + continue + } + grp := m[sum.PartitionID] + if grp == nil { + grp = &group{ + partitionID: sum.PartitionID, + } + m[sum.PartitionID] = grp + } + grp.elem = append(grp.elem, sum) + } + + for partitionID, group := range m { + _ = partitionID + prim := -1 + keyCount := 0 + for k, e := range group.elem { + if e.IsPrimary { + prim = k + } + keyCount += e.KeyCount + } + if prim == -1 { + panic(fmt.Sprintf("no primary found for group '%v'", group.String())) + } + + primary := group.elem[prim] + primaryChecksum := primary.Checksum + for _, e := range group.elem { + if e.IsPrimary { + continue + } + // is e a replica? not necessarily! have to check. + if !e.IsReplica { + //if verbose { + // since this will happen even on a fix point, where it is already empty, + // we don't report it again. + //fmt.Printf("# non-replica should have no data: creating an empty translation store here at '%v'\n", e.StorePath) + //} + err := os.RemoveAll(e.StorePath) + if err != nil { + return errors.Wrap(err, fmt.Sprintf("RepairTranslationStores() os.RemoveAll(e.StorePath='%v')", e.StorePath)) + } + store, err := boltdb.OpenTranslateStore(e.StorePath, e.Index, e.Field, e.PartitionID, pilosa.DefaultPartitionN) + if err != nil { + return errors.Wrap(err, fmt.Sprintf("RepairTranslationStores() create empty boldtdb: boltdb.OpenTranslateStore e.StorePath='%v'", e.StorePath)) + } + err = store.Close() + if err != nil { + return errors.Wrap(err, fmt.Sprintf("RepairTranslationStores() closing empty boltdb at path '%v'", e.StorePath)) + } + continue + } + // INVAR: e is a replica for this paritionID. + // Copy from primary if checksums are different. + if e.Checksum != primaryChecksum { + from := group.elem[prim].StorePath + dest := e.StorePath + if verbose { + fmt.Printf("# e.Checksum '%v' != primaryChecksum '%v': copying from primary translation store '%v' -> '%v'\n", e.Checksum, primaryChecksum, from, dest) + } + err := cp(from, dest) + if err != nil { + return fmt.Errorf("error: could not copy from primary '%v' to replica translation store '%v': '%v' ... try to keep going...\n", from, dest, err) + } + } + } + } + return nil +} + +func (cfg *FsckConfig) dumpcols() { + + verbose := cfg.Verbose + quiet := cfg.Quiet + _, _ = verbose, quiet + + dir := cfg.Dir + index := cfg.Index + partitionID := cfg.PartitionID + showKey := cfg.ShowKey + showID := cfg.ShowID + + if !quiet { + fmt.Printf("# dumpcols: opening dir '%v'... this may take a few minutes...\n", dir) + } + holder := pilosa.NewHolder(dir, nil) + holder.OpenTranslateStore = boltdb.OpenTranslateStore + err := holder.Open() + if err != nil { + log.Fatal(err) + } + if cfg.ShowHeader { + fmt.Println("# columnKey columId") + } + id_key := make(map[uint64]string) + key_id := make(map[string]uint64) + for _, idx := range holder.Indexes() { + fmt.Printf("# Looking '%v'\n", idx.Name()) + if idx.Name() == index { + store := idx.TranslateStore(partitionID) + fmt.Printf("# Key By ID partitionID = %v\n", partitionID) + err := store.KeyWalker(func(key string, col uint64) { + key_id[key] = col + if showKey { + fmt.Printf("# '%v' %v shard: %v partition: %v\n", key, col, col/pilosa.ShardWidth, partitionID) + } + }) + panicOn(err) + } + } + for _, idx := range holder.Indexes() { + if idx.Name() == index { + store := idx.TranslateStore(partitionID) + //fmt.Printf("# ID ByKey\n") + err := store.IDWalker(func(key string, col uint64) { + id_key[col] = key + if showID { + fmt.Printf("# '%v' %v\n", key, col) + } + }) + panicOn(err) + } + } + fmt.Printf("# k: %d i: %d\n", len(key_id), len(id_key)) + fmt.Println("id_key") + for k, v := range id_key { + l, ok := key_id[v] + if ok { + if k != l { + fmt.Printf("# X: %v %v %v\n", k, l, v) + } + } else { + fmt.Printf("# key not in id %v\n", v) + } + } + fmt.Println("key_id") + for k, v := range key_id { + l, ok := id_key[v] + if ok { + if k != l { + fmt.Printf("# T: %v %v %v\n", k, l, v) + } + } else { + fmt.Printf("# id not in key %v\n", v) + } + } +} + +func (cfg *FsckConfig) read() (perNodeIndexMaps []map[string]*pilosa.IndexFragmentSummary, clusterNodes []string, final *pilosa.AllTranslatorSummary, err error) { + + final = pilosa.NewAllTranslatorSummary() + + dirs := cfg.Dirs + for _, dir := range dirs { + idx2frag, nodeID, atsNode, err := cfg.readOneDir(dir) + if err != nil { + return nil, nil, nil, err + } + final.Append(atsNode) + clusterNodes = append(clusterNodes, nodeID) + perNodeIndexMaps = append(perNodeIndexMaps, idx2frag) + } + return +} + +func (cfg *FsckConfig) readOneDir(dir string) (idx2frag map[string]*pilosa.IndexFragmentSummary, nodeID string, atsNode *pilosa.AllTranslatorSummary, err error) { + + verbose := cfg.Verbose + quiet := cfg.Quiet + + if !quiet { + fmt.Printf("# opening dir '%v'... this may take a few minutes... fixcol=%v\n\n", dir, cfg.FixCol) + } + + jmphasher := &pilosa.Jmphasher{} + partitionN := pilosa.DefaultPartitionN + replicaN := cfg.ReplicaN + topo, err := loadTopology(dir, jmphasher, partitionN, replicaN) + if err != nil { + return nil, "", nil, err + } + cfg.topo = topo + //vv("topo = '%#v'", topo) + nodeIDs := topo.GetNodeIDs() + //vv("nodeIDs = '%#v'", nodeIDs) + nNodes := len(nodeIDs) + nDir := len(cfg.Dirs) + if nDir != nNodes { + return nil, "", nil, fmt.Errorf("command line had %v directories (%#v) but the .topology had %v nodes (%#v)", nDir, cfg.Dirs, nNodes, nodeIDs) + } + + holder := pilosa.NewHolder(dir, nil) + holder.OpenTranslateStore = boltdb.OpenTranslateStore + + nodeID, err = holder.LoadNodeID() + panicOn(err) + //vv("nodeID = '%v'", nodeID) + err = holder.Open() + + if err != nil { + log.Fatal(err) + } + + if !quiet { + fmt.Printf("\n# calculating hashes of row and column key translation maps on data from dir '%v'...\n", dir) + } + var indexes []*pilosa.Index + + const checkKeys = true + atsNode = pilosa.NewAllTranslatorSummary() + for _, idx := range holder.Indexes() { + asum, err := idx.ComputeTranslatorSummary(verbose, checkKeys, cfg.FixCol, topo, nodeID) + if err != nil { + log.Fatal(err) + } + atsNode.Append(asum) + indexes = append(indexes, idx) + } + atsNode.Sort() + + hasher := blake3.New() + if !quiet { + fmt.Printf("\n# summary of col/row translations in dir: %v:\n", dir) + } + for _, sum := range atsNode.Sums { + if !quiet { + fmt.Printf("# index: %v partitionID: %v blake3-%v 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:]) + + if !quiet { + fmt.Printf("# all-checksum = blake3-%x\n", buf) + } + + // fragment analysis + + showBits := false + showOpsLog := false + idx2frag = make(map[string]*pilosa.IndexFragmentSummary) // on this node. + for _, idx := range indexes { + if verbose { + fmt.Printf("# ==============================\n") + fmt.Printf("# index: %v\n", idx.Name()) + fmt.Printf("# ==============================\n") + } + frgsum := idx.WriteFragmentChecksums(os.Stdout, showBits, showOpsLog, topo, verbose) + frgsum.Dir = dir + frgsum.NodeID = nodeID + idx2frag[idx.Name()] = frgsum + } + + _ = holder.Close() + + //vv("idx2frag = '%v'", idx2frag) // tons of output. see 1234.out.full for examaple. + + return +} + +// from cluster.go:1924 +func loadTopology(holderDir string, hasher pilosa.Hasher, partitionN, replicaN int) (*pilosa.Topology, error) { + + buf, err := ioutil.ReadFile(filepath.Join(holderDir, ".topology")) + if err != nil { + return nil, err + } + + var pb internal.Topology + err = proto.Unmarshal(buf, &pb) + if err != nil { + return nil, err + } + + return pilosa.DecodeTopology(&pb, hasher, partitionN, replicaN, nil) +} + +func (cfg *FsckConfig) analyze(clusterNodes []string, perNodeIndexMaps []map[string]*pilosa.IndexFragmentSummary, ats *pilosa.AllTranslatorSummary) (fixNeeded bool, reports []string, err error) { + + verbose := cfg.Verbose + quiet := cfg.Quiet + _, _ = verbose, quiet + + allIndex := make(map[string]bool) + for _, mp := range perNodeIndexMaps { + for index := range mp { + allIndex[index] = true + } + } + if !quiet { + vv("allIndex = '%#v'", allIndex) + } + for index := range allIndex { + if !quiet { + vv("on index '%v'", index) + } + nodes2fragsum := make(map[string]*pilosa.IndexFragmentSummary) + for _, mp := range perNodeIndexMaps { + sum := mp[index] + if sum == nil { + continue + } + nodes2fragsum[sum.NodeID] = sum + } + fixme, report, err := cfg.analyzeThisIndex(index, nodes2fragsum, ats) + if err != nil { + return false, reports, fmt.Errorf("error in analyze of index '%v': '%v'", index, err) + } + fixNeeded = fixNeeded || fixme + reports = append(reports, report) + } + return fixNeeded, reports, nil +} + +func (cfg *FsckConfig) analyzeThisIndex( + index string, + nodes2fragsum map[string]*pilosa.IndexFragmentSummary, + ats *pilosa.AllTranslatorSummary, +) (fixNeeded bool, report string, err error) { + + verbose := cfg.Verbose + quiet := cfg.Quiet + _, _ = verbose, quiet + + var removedBytes int64 + var copiedBytes int64 + var changedFiles int64 + var totalFiles int64 + var overwrittenBytes int64 + var totalBytes int64 + + if !quiet { + vv("top of analyzeThisIndex(index='%v'); len of nodes2fragsum = %v; nodes2fragsum='%#v'", + index, len(nodes2fragsum), nodes2fragsum) + } + + for node, sum := range nodes2fragsum { + if !quiet { + fmt.Printf("# on node '%v'\n", node) + } + // do they disagree on who is the primary? + // for each fragment, do they disagree on the checksum? + + // Q: which nodes are supposed to have data, and which + // nodes are not supposed to have data? + + // loopFragSum: + for relpath, fragsum := range sum.RelPath2fsum { + fragsum.NodeID = node + totalFiles++ + //vv("checking %v on node %v", relpath, node) + + replicas, nonReplicas := cfg.topo.GetReplicasForPrimary(fragsum.Primary) + _, _ = replicas, nonReplicas + //vv("replicas = '%#v'", replicas) + //vv("nonReplicas = '%#v'", nonReplicas) + + err := cfg.verifyReplicasAvailable(replicas, nonReplicas, nodes2fragsum, fragsum) + if err != nil { + return fixNeeded, "", err + } + + // find the primary's checksum + primaryChecksum := "" + var primaryFragSum *pilosa.FragSum + for node, isPrimary := range replicas { + if isPrimary { + primarySum := nodes2fragsum[node] + primaryFragSum = primarySum.RelPath2fsum[relpath] + if primaryFragSum == nil { + + // This seems clear indication that we have the topology wrong. + // When the topology is right, there are NO errors of this kind. + // + msg := fmt.Sprintf("# ugh. BAD. Stopping because any fix will be wrong. We see wrong -replica %v param, OR the .id files are mis-assigned with respect to the topology file. Could not find primary FragSum for relpath = '%v'. replicas = '%#v', nonReplicas = '%#v'\n", cfg.ReplicaN, relpath, replicas, nonReplicas) + vv(msg) + fmt.Fprintf(os.Stderr, "%v\n", msg) + panic(msg) // stop. the fixes are going to be wrong. + } else { + primaryChecksum = primaryFragSum.Checksum + primaryFragSum.NodeID = node + primaryFragSum.ScanDone = true + } + break + } + } + if primaryChecksum == "" { + return fixNeeded, "", fmt.Errorf("could not find primary replica??? replicas='%#v', nodes2fragsum='%v'; for fragsum='%#v'", replicas, nodes2fragsum, fragsum) + } + + // is this a non-replica? + _, isNon := nonReplicas[fragsum.NodeID] + if isNon { + removedBytes += FileSize(fragsum.AbsPath) + changedFiles++ + + //vv("yes, is nonReplica: fragsum.NodeID='%v'", fragsum.NodeID) + if !quiet { + fmt.Printf("rm %v #### REPAIR REMOVE data from non-replica at node '%v' (fragsum='%#v') vs. primary (%#v)\n\n", fragsum.AbsPath, node, fragsum, primaryFragSum) + } + if cfg.Fix { + err := os.Remove(fragsum.AbsPath) + if err != nil { + return fixNeeded, "", fmt.Errorf("error removing non-replica extra fragment '%v': '%v'", fragsum.AbsPath, err) + } + } + } else { + presz := FileSize(fragsum.AbsPath) + totalBytes += presz + + checksum := fragsum.Checksum + if checksum != primaryChecksum { + copiedBytes += FileSize(primaryFragSum.AbsPath) + changedFiles++ + overwrittenBytes += presz + + if !quiet { + fmt.Printf("cp %v %v #### REPAIR OVERWRITE replica at node '%v' (%#v) from primary '%v' (%#v)\n", primaryFragSum.AbsPath, fragsum.AbsPath, node, fragsum, primaryFragSum.NodeID, primaryFragSum) + } + if cfg.Fix { + err := cp(primaryFragSum.AbsPath, fragsum.AbsPath) + if err != nil { + return fixNeeded, "", fmt.Errorf("error copying from '%v' to '%v': '%v'", + primaryFragSum.AbsPath, fragsum.AbsPath, err) + } + } + } + } + fragsum.ScanDone = true + } + } + nDir := len(nodes2fragsum) + + keyCount, idCount := cfg.getKeyIDCounts(ats) + + fixNeeded = changedFiles > 0 || ats.RepairNeeded + var actionTaken string + var wouldBe string + if cfg.Fix || cfg.FixCol { + if fixNeeded { + actionTaken = "*REPAIRS WERE MADE TO THE BACKUPS*" + wouldBe = "sync repairs made:" + } else { + wouldBe = "" + actionTaken = "NO REPAIR NEEDED." + } + } else { + if fixNeeded { + wouldBe = "sync actions that would be taken under -fix:" + actionTaken = "*REPAIRS NEEDED BUT WERE NOT APPLIED* ; pilosa-fsck -fix and -fixcol were omitted." + } else { + wouldBe = "" + actionTaken = "NO REPAIR NEEDED." + } + } + var fragUpdate string + if changedFiles > 0 { + fragUpdate = fmt.Sprintf(` +# %v +# copied bytes: %v +# file bytes overwritten: %v +# new bytes added: %v +# new bytes is %0.01f%% of %v total bytes +# removed %v bytes from non-replicas +# changed file count %v (%0.01f%%; total files=%v) +# +`, wouldBe, humanize.Comma(copiedBytes), humanize.Comma(overwrittenBytes), humanize.Comma(copiedBytes-overwrittenBytes), 100*float64(copiedBytes-overwrittenBytes)/float64(totalBytes), humanize.Comma(totalBytes), humanize.Comma(removedBytes), changedFiles, 100*float64(changedFiles)/float64(totalFiles), humanize.Comma(totalFiles)) + } + + report = fmt.Sprintf(` +# ======================================================== +# pilosa-fsck final report +# +# run with -fix: %v +# -fixcol: %v +# +# index examined: '%v' +# +# nodes examined: %v +# -replicas %v replication factor used +# +# feature data examined: %v bytes +# feature files examined: %v files +# +# key-translation-stores examined: %v +# key-count: %v over all replicas +# id-count: %v over all replicas +# +# %v +# %v +# ======================================================== +`, + cfg.Fix, cfg.FixCol, index, nDir, cfg.ReplicaN, humanize.Comma(totalBytes), humanize.Comma(totalFiles), humanize.Comma(int64(nDir*pilosa.DefaultPartitionN)), humanize.Comma(int64(keyCount)), humanize.Comma(int64(idCount)), actionTaken, fragUpdate) + return +} + +func (cfg *FsckConfig) verifyReplicasAvailable(replicas, nonReplicas map[string]bool, nodes2fragsum map[string]*pilosa.IndexFragmentSummary, fragsum *pilosa.FragSum) error { + for node := range replicas { + if nodes2fragsum[node] == nil { + return fmt.Errorf("error: node '%v' needed for a replica set was not availabe. Did you give ALL the directories for your cluster on the command line at once? In nodes2fragsum '%#v' (replicas: '%#v'; non-replicas '%#v') for fragsum '%v'", node, nodes2fragsum, replicas, nonReplicas, fragsum) + } + } + return nil +} + +func cp(fromPath, toPath string) (err error) { + tmpTo := toPath + ".fsck.tmp" + toFd, err := os.Create(tmpTo) + if err != nil { + return err + } + defer toFd.Close() + fromFd, err := os.Open(fromPath) + if err != nil { + return err + } + defer fromFd.Close() + + _, err = io.Copy(toFd, fromFd) + if err != nil { + return err + } + err = toFd.Close() + if err != nil { + return err + } + return os.Rename(tmpTo, toPath) +} + +func (cfg *FsckConfig) getKeyIDCounts(ats *pilosa.AllTranslatorSummary) (keyCount, idCount int) { + for _, sum := range ats.Sums { + keyCount += sum.KeyCount + idCount += sum.IDCount + } + return +} diff --git a/cmd/pilosa-fsck/fsck_test.go b/cmd/pilosa-fsck/fsck_test.go new file mode 100644 index 000000000..b8002f78f --- /dev/null +++ b/cmd/pilosa-fsck/fsck_test.go @@ -0,0 +1,365 @@ +// 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 ( + "context" + "fmt" + "io/ioutil" + "reflect" + "strconv" + "testing" + "time" + + "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/boltdb" + "github.com/pilosa/pilosa/v2/hash" + "github.com/pilosa/pilosa/v2/http" + "github.com/pilosa/pilosa/v2/server" + "github.com/pilosa/pilosa/v2/test" +) + +func Test_Repair(t *testing.T) { + + // a) setup 1 primary + 3 replicas of disagree-ing cluster dirs. + + nNodes := 4 + nReplicas := 3 + + name := t.Name() + var nodeid []string + for i := 0; i < nNodes; i++ { + // work around a bug in the test.MustRunCluster that corrupts + // the .topology file if we only join name with one "_" underscore. + nodeid = append(nodeid, name+"__"+strconv.Itoa(i)) + } + + c := test.MustRunCluster(t, nNodes, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID(nodeid[0]), + pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + pilosa.OptServerReplicaN(nReplicas), + )}, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID(nodeid[1]), + pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + pilosa.OptServerReplicaN(nReplicas), + )}, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID(nodeid[2]), + pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + pilosa.OptServerReplicaN(nReplicas), + )}, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID(nodeid[3]), + pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + pilosa.OptServerReplicaN(nReplicas), + )}, + ) + // note: do not defer c.Close() here. We manually close below. + + var nodes []*test.Command + var dirs []string + for i := 0; i < nNodes; i++ { + nd := c.GetNode(i) + nodes = append(nodes, nd) + dirs = append(dirs, nd.Server.Holder().Path()) + } + + ctx := context.Background() + + index := "rick" + fieldName := "f" + + idx, err := nodes[0].API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}) + if err != nil { + t.Fatalf("creating index: %v", err) + } + if idx.CreatedAt() == 0 { + t.Fatal("index createdAt is empty") + } + + field, err := nodes[0].API.CreateField(ctx, index, fieldName, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100)) + if err != nil { + t.Fatalf("creating field: %v", err) + } + if field.CreatedAt() == 0 { + t.Fatal("field createdAt is empty") + } + + rowID := uint64(1) + timestamp := int64(0) + + // Generate some keyed records. + rowIDs := []uint64{} + timestamps := []int64{} + N := 10 + for i := 1; i <= N; i++ { + rowIDs = append(rowIDs, rowID) + timestamps = append(timestamps, timestamp) + } + + // Keys are sharded so ordering is not guaranteed. + colKeys := []string{"col10", "col8", "col9", "col6", "col7", "col4", "col5", "col2", "col3", "col1"} + + colKeys = colKeys[:N] + + // Import data with keys to the coordinator (node0) and verify that it gets + // translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher) + req := &pilosa.ImportRequest{ + Index: index, + IndexCreatedAt: idx.CreatedAt(), + Field: fieldName, + FieldCreatedAt: field.CreatedAt(), + + // even though this says Shard: 0, that won't matter. The column keys + // get hashed and that decides the actual shard. + Shard: 0, + RowIDs: rowIDs, + ColumnKeys: colKeys, + Timestamps: timestamps, + } + + qcx := nodes[0].API.Txf().NewQcx() + + if err := nodes[0].API.Import(ctx, qcx, req); err != nil { + t.Fatal(err) + } + panicOn(qcx.Finish()) + + pql := fmt.Sprintf("Row(%s=%d)", fieldName, rowID) + + // Query node0. + if res, err := nodes[0].API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil { + t.Fatal(err) + } else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) { + t.Fatalf("expected colKeys='%#v'; observed column keys: %#v", colKeys, keys) + } + + // Query node1. + if err := test.RetryUntil(5*time.Second, func() error { + if res, err := nodes[1].API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil { + return err + } else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) { + return fmt.Errorf("unexpected column keys: %#v", keys) + } + return nil + }); err != nil { + t.Fatal(err) + } + + // end of setup. + + // partitionID in use: 6, 31, 57, 133, 185, 235 + targetPartition := 31 // which partitionID we mess with. + targetNode := nodes[0] // this is the first replica. + // 0 first replica + // 1 second replica + // 2 -- not a replica + // 3 primary + + cfg := &FsckConfig{ + Fix: false, + FixCol: false, + Quiet: true, + //Verbose: true, + ReplicaN: nReplicas, + Dirs: dirs, + } + panicOn(cfg.ValidateConfig()) + + // for this test, mess up a replica that is not the primary. + + h := targetNode.API.Holder() + idx = h.Index(index) + store := idx.TranslateStore(targetPartition) + fwd, rev := getFwdRev(store, targetPartition) + //vv("targetPartition=%v, store.PartitionID=%v, before corruption, fwd='%#v', rev='%#v'", targetPartition, store.PartitionID, fwd, rev) + + // # fsck_test.go:288 2020-10-01T13:39:57.718995-05:00 partition 31, key 'col5' -> db00001 + presz := len(rev) + delete(rev, fwd["col5"]) + postsz := len(rev) + + if postsz == presz { + panic("did not delete any key!") + } + + bolt := store.(*boltdb.TranslateStore) + //vv("pre corruption, bolt = '%v'", fileChecksum(bolt.Path)) + //bolt.DumpBolt("pre-corruption") + + if err := bolt.SetFwdRevMaps(nil, fwd, rev); err != nil { + t.Fatal(err) + } + //vv("post corruption, bolt = '%v'", fileChecksum(bolt.Path)) + //bolt.DumpBolt("post-corruption") + + //fwd3, rev3 := getFwdRev(store, targetPartition) + //vv("after corruption, fwd='%#v', rev='%#v'", fwd3, rev3) + + for _, nd := range nodes { + nd.Command.Close() + } + //panicOn(bolt.Open()) + //bolt.DumpBolt("post-corruption, after Close. bolt:") + //bolt.Close() + + //chksums := getChecksums(dirs, cfg, targetPartition) + //vv("post corruption, pre repair chksums = '%#v'", chksums) + + // first we check that the corruption can be detected + // by our test with the checksums. + + chk, err := check(dirs, cfg, targetPartition) + _ = chk + //vv("pre-fix, chk='%v'; err='%v'", chk, err) + + if err == nil { + panic("expected to see checksums not match! but no corruption detected.") + } + + // b) running in reporting mode only should report that a fix is needed. + fixNeeded, err := cfg.Run() + panicOn(err) + if !fixNeeded { + panic("fix should be needed now, before repair") + } + + // c) run the fix. + cfg.Fix = true + cfg.FixCol = true + fixNeeded, err = cfg.Run() + panicOn(err) + if !fixNeeded { + panic("fix should be marked needed if repair was made") + } + + // d) check that the replicas all look like the primary. + + //chksums = getChecksums(dirs, cfg, targetPartition) + //vv("after repair chksums = '%#v'", chksums) + + chk, err = check(dirs, cfg, targetPartition) + _ = chk + //vv("chk = '%v' after repair; err='%v'", chk, err) + panicOn(err) + + // e) run again, should see no fix needed. + fixNeeded, err = cfg.Run() + panicOn(err) + if fixNeeded { + panic("should see no fix needed after the prior repair") + } + +} + +func getFwdRev(store pilosa.TranslateStore, partitionID int) (fwd map[string]uint64, rev map[uint64]string) { + fwd = make(map[string]uint64) + rev = make(map[uint64]string) + _ = store.KeyWalker(func(key string, col uint64) { + //vv("partition %v, key '%v' -> %x", partitionID, key, col) + fwd[key] = col + }) + _ = store.IDWalker(func(key string, col uint64) { + //vv("partition %v, id %x -> '%v'", partitionID, col, key) + rev[col] = key + }) + return +} + +func check(dirs []string, cfg *FsckConfig, targetPartition int) (chksum string, err error) { + + firstChecksum := "" + firstDir := "" + quiet := cfg.Quiet + defer func() { + cfg.Quiet = quiet + }() + cfg.Quiet = true + for i := range dirs { + dir := dirs[i] + _, _, ats, err := cfg.readOneDir(dir) + panicOn(err) + + for _, s := range ats.Sums { + if s.PartitionID != targetPartition { + continue + } + if s.IsPrimary || s.IsReplica { + chksum := s.Checksum + if firstChecksum == "" { + firstChecksum = chksum + firstDir = dir + } else { + if chksum != firstChecksum { + return chksum, fmt.Errorf("bolt chksum on node %v '%v' disagrees with '%v' on '%v'", dir, chksum, firstChecksum, firstDir) + } + } + } + } + } + return firstChecksum, nil +} + +var _ = getChecksums + +func getChecksums(dirs []string, cfg *FsckConfig, targetPartition int) (chksum []string) { + + for i := range dirs { + dir := dirs[i] + _, _, ats, err := cfg.readOneDir(dir) + panicOn(err) + + for _, s := range ats.Sums { + if s.PartitionID != targetPartition { + continue + } + chksum = append(chksum, s.Checksum) + } + } + return +} + +/* on shardwidth 20 +# fsck_test.go:211 2020-09-30T17:19:05.823278-05:00 partition 6, key 'col2' -> dc00001 +# fsck_test.go:214 2020-09-30T17:19:05.823309-05:00 partition 6, id dc00001 -> 'col2' +# fsck_test.go:211 2020-09-30T17:19:05.823430-05:00 partition 31, key 'col5' -> db00001 +# fsck_test.go:214 2020-09-30T17:19:05.823447-05:00 partition 31, id db00001 -> 'col5' +# fsck_test.go:211 2020-09-30T17:19:05.823970-05:00 partition 57, key 'col10' -> 5d00001 +# fsck_test.go:214 2020-09-30T17:19:05.823998-05:00 partition 57, id 5d00001 -> 'col10' +# fsck_test.go:211 2020-09-30T17:19:05.827007-05:00 partition 133, key 'col7' -> d900001 +# fsck_test.go:214 2020-09-30T17:19:05.827071-05:00 partition 133, id d900001 -> 'col7' +# fsck_test.go:211 2020-09-30T17:19:05.827549-05:00 partition 185, key 'col3' -> dd00001 +# fsck_test.go:214 2020-09-30T17:19:05.827573-05:00 partition 185, id dd00001 -> 'col3' +# fsck_test.go:211 2020-09-30T17:19:05.827792-05:00 partition 235, key 'col9' -> d700001 +# fsck_test.go:214 2020-09-30T17:19:05.827809-05:00 partition 235, id d700001 -> 'col9' +*/ + +var _ = fileChecksum + +func fileChecksum(path string) string { + by, err := ioutil.ReadFile(path) + panicOn(err) + return hash.Blake3sum16(by) +} diff --git a/cmd/pilosa-fsck/release-pilosa-fsck/.gitignore b/cmd/pilosa-fsck/release-pilosa-fsck/.gitignore new file mode 100644 index 000000000..a08586f1c --- /dev/null +++ b/cmd/pilosa-fsck/release-pilosa-fsck/.gitignore @@ -0,0 +1 @@ +pilosa-fsck diff --git a/cmd/pilosa-fsck/release-pilosa-fsck/backups.tar.gz b/cmd/pilosa-fsck/release-pilosa-fsck/backups.tar.gz new file mode 100644 index 000000000..28b08adbd Binary files /dev/null and b/cmd/pilosa-fsck/release-pilosa-fsck/backups.tar.gz differ diff --git a/cmd/pilosa-fsck/release-pilosa-fsck/example.sh b/cmd/pilosa-fsck/release-pilosa-fsck/example.sh new file mode 100755 index 000000000..79fd4cf11 --- /dev/null +++ b/cmd/pilosa-fsck/release-pilosa-fsck/example.sh @@ -0,0 +1,21 @@ +#!/bin/bash + +set +x +export PATH=.:${PATH} + +# unpack the sample Molecula Pilosa cluster. +tar xf backups.tar.gz + + +# check if repair is needed. +pilosa-fsck -replicas 3 backups/node0/pilosa backups/node1/pilosa backups/node2/pilosa backups/node3/pilosa + + +# yes, so do the repairs. This can be done first (only) as well. +# +pilosa-fsck -fix -replicas 3 backups/node0/pilosa backups/node1/pilosa backups/node2/pilosa backups/node3/pilosa + + +# check again if you like +# +pilosa-fsck -replicas 3 backups/node0/pilosa backups/node1/pilosa backups/node2/pilosa backups/node3/pilosa diff --git a/cmd/pilosa-fsck/vprint.go b/cmd/pilosa-fsck/vprint.go new file mode 100644 index 000000000..83b1681f7 --- /dev/null +++ b/cmd/pilosa-fsck/vprint.go @@ -0,0 +1,177 @@ +// home: https://github.com/glycerine/vprint +// Copyright 2019 Jason E. Aten, Ph.D. All rights reserved. +// License: MIT +// +// MIT License +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +package main + +import ( + "fmt" + "io" + "os" + "path" + "runtime" + "runtime/debug" + "sync" + "time" +) + +const RFC3339MsecTz0 = "2006-01-02T15:04:05.000Z07:00" +const RFC3339UsecTz0 = "2006-01-02T15:04:05.000000Z07:00" + +// for tons of debug output +var VerboseVerbose bool = false + +// convience functions for . import +var pp = PP +var vv = VV + +var panicOn = PanicOn + +func init() { + // keeper linter happy + _ = pp + _ = vv +} + +func PanicOn(err error) { + if err != nil { + panic(err) + } +} + +func PP(format string, a ...interface{}) { + if VerboseVerbose { + TSPrintf(format, a...) + } +} + +func VV(format string, a ...interface{}) { + TSPrintf(format, a...) +} + +func AlwaysPrintf(format string, a ...interface{}) { + TSPrintf(format, a...) +} + +var tsPrintfMut sync.Mutex + +// time-stamped printf +func TSPrintf(format string, a ...interface{}) { + tsPrintfMut.Lock() + Printf("# %s %s ", FileLine(3), ts()) + Printf(format+"\n", a...) + tsPrintfMut.Unlock() +} + +// get timestamp for logging purposes +func ts() string { + return time.Now().Format(RFC3339UsecTz0) +} + +// so we can multi write easily, use our own printf +var OurStdout io.Writer = os.Stdout + +// Printf formats according to a format specifier and writes to standard output. +// It returns the number of bytes written and any write error encountered. +func Printf(format string, a ...interface{}) (n int, err error) { + return fmt.Fprintf(OurStdout, format, a...) +} + +func FileLine(depth int) string { + _, fileName, fileLine, ok := runtime.Caller(depth) + var s string + if ok { + s = fmt.Sprintf("%s:%d", path.Base(fileName), fileLine) + } else { + s = "" + } + return s +} + +func stack() string { + return string(debug.Stack()) +} + +func FileExists(name string) bool { + fi, err := os.Stat(name) + if err != nil { + return false + } + if fi.IsDir() { + return false + } + return true +} + +func DirExists(name string) bool { + fi, err := os.Stat(name) + if err != nil { + return false + } + if fi.IsDir() { + return true + } + return false +} + +func FileSize(name string) int64 { + fi, err := os.Stat(name) + if err != nil { + return 0 + } + return fi.Size() +} + +// Caller returns the name of the calling function. +func Caller(upStack int) string { + // elide ourself and runtime.Callers + target := upStack + 2 + + pc := make([]uintptr, target+2) + n := runtime.Callers(0, pc) + + f := runtime.Frame{Function: "unknown"} + if n > 0 { + frames := runtime.CallersFrames(pc[:n]) + for i := 0; i <= target; i++ { + contender, more := frames.Next() + if i == target { + f = contender + } + if !more { + break + } + } + } + return f.Function +} + +// happy linter: +var _ = DirExists +var _ = FileExists +var _ = Caller +var _ = stack +var _ = RFC3339MsecTz0 +var _ = RFC3339UsecTz0 +var _ = AlwaysPrintf +var _ = FileSize diff --git a/go.mod b/go.mod index 33b1d4d59..0425a1b03 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd // indirect github.com/davecgh/go-spew v1.1.1 github.com/dgraph-io/badger/v2 v2.0.1-rc1.0.20200709123515-8e896a7af361 + github.com/dustin/go-humanize v1.0.0 github.com/glycerine/lmdb-go v1.9.32 github.com/go-ole/go-ole v1.2.4 // indirect github.com/gogo/protobuf v1.2.1 diff --git a/holder.go b/holder.go index 5d8691a92..bece8baf2 100644 --- a/holder.go +++ b/holder.go @@ -1218,7 +1218,7 @@ func (h *Holder) setFileLimit() { } } -func (h *Holder) loadNodeID() (string, error) { +func (h *Holder) LoadNodeID() (string, error) { idPath := path.Join(h.path, ".id") h.Logger.Printf("load NodeID: %s", idPath) if err := os.MkdirAll(h.path, 0777); err != nil { @@ -1227,7 +1227,9 @@ func (h *Holder) loadNodeID() (string, error) { nodeIDBytes, err := ioutil.ReadFile(idPath) if err == nil { - return strings.TrimSpace(string(nodeIDBytes)), nil + nodeid := strings.TrimSpace(string(nodeIDBytes)) + h.Logger.Printf("I am NodeID: %s", nodeid) + return nodeid, nil } if !os.IsNotExist(err) { return "", errors.Wrap(err, "reading file") @@ -1237,6 +1239,7 @@ func (h *Holder) loadNodeID() (string, error) { if err != nil { return "", errors.Wrap(err, "writing file") } + h.Logger.Printf("I am NodeID: %s", nodeID) return nodeID, nil } @@ -1728,7 +1731,7 @@ func (s *holderSyncer) readIndexTranslateReader(rd TranslateEntryReader) { } // Apply replication to store. - store := idx.TranslateStore(s.Cluster.keyPartition(entry.Index, entry.Key)) + store := idx.TranslateStore(s.Cluster.Topology.KeyPartition(entry.Index, entry.Key)) if err := store.ForceSet(entry.ID, entry.Key); err != nil { s.Holder.Logger.Printf("cannot force set index translation data: %d=%q", entry.ID, entry.Key) return diff --git a/index.go b/index.go index 4e7d3459a..f2c045726 100644 --- a/index.go +++ b/index.go @@ -33,6 +33,7 @@ import ( "github.com/pilosa/pilosa/v2/stats" "github.com/pilosa/pilosa/v2/testhook" "github.com/pkg/errors" + "github.com/zeebo/blake3" "golang.org/x/sync/errgroup" ) @@ -734,6 +735,19 @@ func (idx *Index) SliceOfShards(field, view, viewPath string) (sliceOfShards []u 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 { @@ -741,6 +755,7 @@ func NewAllTranslatorSummary() *AllTranslatorSummary { } func (ats *AllTranslatorSummary) Append(b *AllTranslatorSummary) { ats.Sums = append(ats.Sums, b.Sums...) + ats.RepairNeeded = ats.RepairNeeded || b.RepairNeeded } func (ats *AllTranslatorSummary) Sort() { @@ -761,57 +776,158 @@ func (ats *AllTranslatorSummary) Sort() { if a.PartitionID > b.PartitionID { return false } - return a.Field < b.Field + 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 (i *Index) ComputeTranslatorSummary(verbose bool) (ats *AllTranslatorSummary, err error) { +func (i *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs bool, topo *Topology, nodeID string) (ats *AllTranslatorSummary, err error) { i.mu.RLock() defer i.mu.RUnlock() ats = &AllTranslatorSummary{} - fmt.Printf("\nindex: %v\n=================\n", i.name) + if verbose { + fmt.Printf("\n# index: %v\n# =================\n", i.name) + } for _, fld := range i.fields { - sum, err := fld.translateStore.ComputeTranslatorSummary() + sum, err := fld.translateStore.ComputeTranslatorSummaryRows() if err != nil { return ats, err } sum.Field = fld.name sum.Index = i.Name() sum.Checksum = hash.Blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, fld.name, i.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) + fmt.Printf("# row blake3-%v keyN: %5v idN: %5v field: '%v'\n", sum.Checksum, sum.KeyCount, sum.IDCount, fld.name) } ats.Sums = append(ats.Sums, sum) } - - fmt.Printf("====================\n") + if verbose { + fmt.Printf("# ====================\n") + } for partitionID, store := range i.translateStores { - sum, err := store.ComputeTranslatorSummary() + if checkKeys { + prim := topo.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 repair of keys on nodeID '%v', and primID '%v'\n", nodeID, primID) + } + changed, err := store.RepairKeys(topo, verbose, applyKeyRepairs) + if err != nil { + return nil, errors.Wrap(err, "ComputeTranslatorSummary() call to store.Repair()") + } + if changed { + ats.RepairNeeded = true + } + } + + // key repair has to be above, because we compute the checksum below. + + sum, err := store.ComputeTranslatorSummaryCols(partitionID, topo) if err != nil { return ats, err } if sum == nil { - // probably one of the Noop stores + // probably one of the Noop stores from the tests. continue } + sum.IsColKey = true sum.PartitionID = partitionID sum.Index = i.Name() + sum.StorePath = store.GetStorePath() + sum.NodeID = nodeID + sum.IsPrimary = topo.IsPrimary(nodeID, partitionID) + + replicas := topo.GetNonPrimaryReplicas(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, i.Name()))) if verbose { - fmt.Printf("col blake3-%v keyN: %10v idN: %10v paritionID: %03v \n", sum.Checksum, sum.KeyCount, sum.IDCount, partitionID) + // 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) } + ats.Sums = append(ats.Sums, sum) } return } -func (idx *Index) WriteFragmentChecksums(w io.Writer, showBits, showOps bool) { +// 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), + } paths, err := listFilesUnderDir(idx.path, false, "", true) panicOn(err) index := idx.name @@ -822,14 +938,37 @@ func (idx *Index) WriteFragmentChecksums(w io.Writer, showBits, showOps bool) { continue // ignore .meta paths } abspath := idx.path + sep + relpath + primary := topo.GetPrimaryForShardReplication(index, shard) checksum, hotbits := RoaringFragmentChecksum(abspath, index, field, view, shard) - fmt.Fprintf(w, "frg blake3-%v field: '%v' view: '%v' shard: %3v hotbits: %10v\n", checksum, field, view, shard, hotbits) + 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 { - fmt.Fprintf(w, "empty index '%v'", idx.path) + if verbose { + fmt.Fprintf(w, "empty index '%v'", idx.path) + } } + return } func (idx *Index) Txf() *TxFactory { diff --git a/license.exceptions b/license.exceptions index 5fb7846e8..2c8c4162f 100644 --- a/license.exceptions +++ b/license.exceptions @@ -21,3 +21,5 @@ ./synthload/vprint.go ./proto/vdsm/vdsm.proto ./proto/vdsm/vdsm.pb.go +./cmd/pilosa-fsck/vprint.go + diff --git a/mock/translator.go b/mock/translator.go index e577baf12..6afd0d48e 100644 --- a/mock/translator.go +++ b/mock/translator.go @@ -37,7 +37,10 @@ type TranslateStore struct { EntryReaderFunc func(ctx context.Context, offset uint64) (pilosa.TranslateEntryReader, error) } -func (s *TranslateStore) ComputeTranslatorSummary() (sum *pilosa.TranslatorSummary, err 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 } @@ -93,6 +96,14 @@ 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 { @@ -107,3 +118,10 @@ 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 a011ab0ba..be0a41aa3 100644 --- a/server.go +++ b/server.go @@ -579,7 +579,7 @@ func (s *Server) loadNodeID() string { if s.nodeID != "" { return s.nodeID } - nodeID, err := s.holder.loadNodeID() + nodeID, err := s.holder.LoadNodeID() if err != nil { s.logger.Printf("loading NodeID: %v", err) return s.nodeID diff --git a/server/server.go b/server/server.go index 077c8a55c..180fd4334 100644 --- a/server/server.go +++ b/server/server.go @@ -37,6 +37,7 @@ import ( "golang.org/x/sync/errgroup" + "github.com/pelletier/go-toml" "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/boltdb" "github.com/pilosa/pilosa/v2/encoding/proto" @@ -622,3 +623,10 @@ func (f *filteredWriter) Write(p []byte) (n int, err error) { } return len(p), nil } + +// ParseConfig parses s into a Config. +func ParseConfig(s string) (Config, error) { + var c Config + err := toml.Unmarshal([]byte(s), &c) + return c, err +} diff --git a/server/server_test.go b/server/server_test.go index ca1ea0e3e..270fdcaf5 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -31,7 +31,6 @@ import ( "testing" "time" - "github.com/pelletier/go-toml" "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/http" "github.com/pilosa/pilosa/v2/pql" @@ -338,7 +337,7 @@ func TestMain_MinMaxFloat(t *testing.T) { // Ensure the host can be parsed. func TestConfig_Parse_Host(t *testing.T) { - if c, err := ParseConfig(`bind = "local"`); err != nil { + if c, err := server.ParseConfig(`bind = "local"`); err != nil { t.Fatal(err) } else if c.Bind != "local" { t.Fatalf("unexpected host: %s", c.Bind) @@ -347,7 +346,7 @@ func TestConfig_Parse_Host(t *testing.T) { // Ensure the data directory can be parsed. func TestConfig_Parse_DataDir(t *testing.T) { - if c, err := ParseConfig(`data-dir = "/tmp/foo"`); err != nil { + if c, err := server.ParseConfig(`data-dir = "/tmp/foo"`); err != nil { t.Fatal(err) } else if c.DataDir != "/tmp/foo" { t.Fatalf("unexpected data dir: %s", c.DataDir) @@ -600,13 +599,6 @@ func GenerateSetCommands(n int, rand *rand.Rand) []SetCommand { return cmds } -// ParseConfig parses s into a Config. -func ParseConfig(s string) (server.Config, error) { - var c server.Config - err := toml.Unmarshal([]byte(s), &c) - return c, err -} - // MustMarshalJSON marshals v into a string. Panic on error. func MustMarshalJSON(v interface{}) string { buf, err := json.Marshal(v) diff --git a/test/cluster.go b/test/cluster.go index 41b1cea13..c71e3e68f 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -248,7 +248,7 @@ func newCluster(tb testing.TB, size int, opts ...[]server.CommandOption) (*Clust commandOpts = opts[i%len(opts)] } m := NewCommandNode(tb, i == 0, commandOpts...) - err := ioutil.WriteFile(path.Join(m.Config.DataDir, ".id"), []byte(name+"_"+strconv.Itoa(i)), 0600) + err := ioutil.WriteFile(path.Join(m.Config.DataDir, ".id"), []byte(name+"__"+strconv.Itoa(i)), 0600) if err != nil { return nil, errors.Wrap(err, "writing node id") } diff --git a/translate.go b/translate.go index 101a854a3..0f8169e5c 100644 --- a/translate.go +++ b/translate.go @@ -17,6 +17,7 @@ package pilosa import ( "context" "encoding/json" + "fmt" "io" "io/ioutil" "sync" @@ -90,7 +91,15 @@ type TranslateStore interface { // the read payload. ReadFrom(io.Reader) (int64, error) - ComputeTranslatorSummary() (sum *TranslatorSummary, err 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, @@ -101,6 +110,14 @@ type TranslatorSummary struct { // ParitionID is filled for column keys PartitionID int + NodeID string + StorePath string + IsPrimary bool + IsReplica bool + + // PrimaryNodeIndex indexes into the cluster []node array to find the primary + PrimaryNodeIndex int + // Field is filled for row keys Field string @@ -112,6 +129,41 @@ type TranslatorSummary struct { // IDCount has the number of ID->Key mappings IDCount int + + // false for RowIDs, true for string-Key column IDs. + IsColKey bool +} + +func (s *TranslatorSummary) String() string { + return fmt.Sprintf(` +TranslatorSummary{ + Index : %v + PartitionID: %v + NodeID : %v + StorePath : %v + IsPrimary : %v + IsReplica : %v + PrimaryNodeIndex: %v + Field : %v + Checksum: %v + KeyCount: %v + IDCount : %v + IsColKey: %v +} +`, + s.Index, + s.PartitionID, + s.NodeID, + s.StorePath, + s.IsPrimary, + s.IsReplica, + s.PrimaryNodeIndex, + s.Field, + s.Checksum, + s.KeyCount, + s.IDCount, + s.IsColKey, + ) } // OpenTranslateStoreFunc represents a function for instantiating and opening a TranslateStore. @@ -127,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 shardPartition(index, id/ShardWidth, partitionN) == partitionID { + if shardToShardPartition(index, id/ShardWidth, partitionN) == partitionID { return id } } @@ -304,6 +356,30 @@ 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. @@ -312,9 +388,18 @@ func OpenInMemTranslateStore(rawurl, index, field string, partitionID, partition return NewInMemTranslateStore(index, field, partitionID, partitionN), nil } -func (s *InMemTranslateStore) ComputeTranslatorSummary() (sum *TranslatorSummary, err error) { +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 91f2c0745..195c53cef 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -70,7 +70,7 @@ func NewTestCluster(tb testing.TB, n int) *cluster { c.ReplicaN = 1 c.Hasher = NewTestModHasher() c.Path = path - c.Topology = newTopology() + c.Topology = NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c) for i := 0; i < n; i++ { c.nodes = append(c.nodes, &Node{ @@ -137,6 +137,15 @@ func (t *ClusterCluster) CreateIndex(name string) error { 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{}) @@ -272,7 +281,8 @@ func (t *ClusterCluster) addCluster(i int, saveTopology bool) (*cluster, error) c.ReplicaN = 1 c.Hasher = NewTestModHasher() c.Path = path - c.Topology = newTopology() + c.partitionN = 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 @@ -515,3 +525,47 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *ResizeInstruction) error node := instr.Coordinator return bcast{t: t}.SendTo(node, complete) } + +var _ = NewTestClusterWithReplication // happy linter + +func NewTestClusterWithReplication(tb testing.TB, nNodes, nReplicas, partitionN int) (c *cluster, cleaner func()) { + path, err := testhook.TempDir(tb, "pilosa-cluster-") + if err != nil { + panic(err) + } + + // holder + h := NewHolder(path, nil) + + // cluster + availableShardFileFlushDuration.Set(100 * time.Millisecond) + c = newCluster() + c.holder = h + c.ReplicaN = nReplicas + c.Hasher = &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) + c.nodes = append(c.nodes, &Node{ + ID: nodeID, + URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)), + }) + c.Topology.addID(nodeID) + } + + c.Node = c.nodes[0] + c.Coordinator = c.nodes[0].ID + c.SetState(ClusterStateNormal) + + if err := c.holder.Open(); err != nil { + panic(err) + } + + return c, func() { + c.holder.Close() + c.close() + } +}