Merge pull request #900 from molecula/pilosa-fsck-rb

pilosa-fsck: fsck-like scan and repair of pilosa backups
This commit is contained in:
jaten-molecula 2020-10-02 16:53:57 -05:00 committed by GitHub
commit 75b8fa0aa8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
25 changed files with 2993 additions and 69 deletions

1
.gitignore vendored
View file

@ -6,3 +6,4 @@ vendor
build
*~
lattice
release-pilosa-fsck.*.*.tar.gz

View file

@ -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) ./...

View file

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

View file

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

View file

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

View file

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

View file

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

36
cmd/pilosa-fsck/Makefile Normal file
View file

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

952
cmd/pilosa-fsck/fsck.go Normal file
View file

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

View file

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

View file

@ -0,0 +1 @@
pilosa-fsck

Binary file not shown.

View file

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

177
cmd/pilosa-fsck/vprint.go Normal file
View file

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

1
go.mod
View file

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

View file

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

167
index.go
View file

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

View file

@ -21,3 +21,5 @@
./synthload/vprint.go
./proto/vdsm/vdsm.proto
./proto/vdsm/vdsm.pb.go
./cmd/pilosa-fsck/vprint.go

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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