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

pilosa-fsck: parallelize Index.ComputeTranslatorSummary

Browse source 
- add -index option
This commit is contained in:
Jason Aten 2020-10-09 11:14:41 -05:00
parent d3ce4fa70e
commit 8da15b18d4
2 changed files with 118 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -52,15 +52,15 @@ type FsckConfig struct {
Fix bool // -fix
FixCol bool // -fixcol
Colkeydump bool // -col
Colkeydump bool // -col
JustThisIndex string // -index
// -col column key dump only options:
Dir string
Index string
PartitionID int
ShowHeader bool
ShowKey bool
ShowID bool
// Dir 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.
@ -87,6 +87,8 @@ func (cfg *FsckConfig) DefineFlags(fs *flag.FlagSet) {
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.StringVar(&cfg.JustThisIndex, "index", "", "(optional) restrict to just this index. Otherwise we default to all indexes.")
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} {-q} /backup/1/.pilosa /backup/2/.pilosa ... /backup/N/.pilosa
@ -99,6 +101,9 @@ func (cfg *FsckConfig) DefineFlags(fs *flag.FlagSet) {
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.
-index index_name
(optional) restrict to just this index. Otherwise we default to all indexes.
-q
be very quiet during analysis and repair
@ -323,9 +328,9 @@ func main() {
func (cfg *FsckConfig) Run() (fixNeeded bool, err error) {
if cfg.Colkeydump {
cfg.dumpcols()
}
// if cfg.Colkeydump {
// cfg.dumpcols()
//}
perNodeIndexMaps, clusterNodes, ats, err := cfg.read()
if err != nil {
@ -398,6 +403,10 @@ func (cfg *FsckConfig) RepairTranslationStores(ats *pilosa.AllTranslatorSummary)
for _, index := range indexes {
if !cfg.DoingIndex(index) {
continue
}
m := make(map[int]*group)
for _, sum := range ats.Sums {
@ -474,6 +483,7 @@ func (cfg *FsckConfig) RepairTranslationStores(ats *pilosa.AllTranslatorSummary)
return nil
}
/*
func (cfg *FsckConfig) dumpcols() {
verbose := cfg.Verbose
@ -551,6 +561,7 @@ func (cfg *FsckConfig) dumpcols() {
}
}
}
*/
func (cfg *FsckConfig) read() (perNodeIndexMaps []map[string]*pilosa.IndexFragmentSummary, clusterNodes []string, final *pilosa.AllTranslatorSummary, err error) {
@ -615,6 +626,13 @@ func (cfg *FsckConfig) readOneDir(dir string) (idx2frag map[string]*pilosa.Index
const checkKeys = true
atsNode = pilosa.NewAllTranslatorSummary()
for _, idx := range holder.Indexes() {
if !cfg.DoingIndex(idx.Name()) {
continue
}
//vv("calling idx.ComputeTranslatorSummary(verbose, checkKeys=%v, cfg.FixCol='%v')", checkKeys, cfg.FixCol)
asum, err := idx.ComputeTranslatorSummary(verbose, checkKeys, cfg.FixCol, topo, nodeID)
if err != nil {
log.Fatal(err)
@ -666,6 +684,18 @@ func (cfg *FsckConfig) readOneDir(dir string) (idx2frag map[string]*pilosa.Index
return
}
func (cfg *FsckConfig) DoingIndex(index string) bool {
if cfg.JustThisIndex == "" {
// scan all indexes
return true
}
if index == cfg.JustThisIndex {
// scan just this one
return true
}
return false
}
// from cluster.go:1924
func loadTopology(holderDir string, hasher pilosa.Hasher, partitionN, replicaN int) (*pilosa.Topology, error) {

134
index.go
View file

@ -792,84 +792,106 @@ func (i *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs boo
defer i.mu.RUnlock()
ats = &AllTranslatorSummary{}
var atsMu sync.Mutex
if verbose {
fmt.Printf("\n# index: %v\n# =================\n", i.name)
}
var g errgroup.Group
for _, fld := range i.fields {
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)
}
ats.Sums = append(ats.Sums, sum)
fld := fld
g.Go(func() error {
//vv("g.Go() on fld '%v'", fld.name)
sum, err := fld.translateStore.ComputeTranslatorSummaryRows()
if err != nil {
return 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)
}
atsMu.Lock()
ats.Sums = append(ats.Sums, sum)
atsMu.Unlock()
return nil
})
}
if verbose {
fmt.Printf("# ====================\n")
}
for partitionID, store := range i.translateStores {
if checkKeys {
prim := topo.PrimaryNodeIndex(partitionID)
primID := topo.nodeIDs[prim]
partitionID := partitionID
store := store
g.Go(func() error {
//vv("g.Go() running on store.Path = '%v'", store.GetStorePath())
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.
// note: we fix irrespective of nodeID == primID now, so that we
// get a fine grain report of what maps were off.
if verbose {
// This is pilosa-fsck output, not regular log.
fmt.Printf("# doing analysis of keys on nodeID '%v', and primID '%v'\n", nodeID, primID)
if verbose {
// This is pilosa-fsck output, not regular log.
fmt.Printf("# doing analysis of keys on nodeID '%v', and primID '%v'\n", nodeID, primID)
}
changed, err := store.RepairKeys(topo, verbose, applyKeyRepairs)
if err != nil {
return errors.Wrap(err, "ComputeTranslatorSummary() call to store.Repair()")
}
if changed {
atsMu.Lock()
ats.RepairNeeded = true
atsMu.Unlock()
}
}
changed, err := store.RepairKeys(topo, verbose, applyKeyRepairs)
// key repair has to be above, because we compute the checksum below.
sum, err := store.ComputeTranslatorSummaryCols(partitionID, topo)
if err != nil {
return nil, errors.Wrap(err, "ComputeTranslatorSummary() call to store.Repair()")
return err
}
if changed {
ats.RepairNeeded = true
if sum == nil {
// probably one of the Noop stores from the tests.
return nil
}
}
sum.IsColKey = true
sum.PartitionID = partitionID
sum.Index = i.Name()
sum.StorePath = store.GetStorePath()
sum.NodeID = nodeID
sum.IsPrimary = topo.IsPrimary(nodeID, partitionID)
// 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 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
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 {
// 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)
}
sum.Checksum = hash.Blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, partitionID, i.Name())))
if verbose {
// This is not regular index logging. This is output of the pilosa-fsck tool.
// So it must be printing straight to stdout.
fmt.Printf("# col blake3-%v keyN: %10v idN: %10v paritionID: %03v primary: %03v\n", sum.Checksum, sum.KeyCount, sum.IDCount, partitionID, sum.PrimaryNodeIndex)
}
atsMu.Lock()
ats.Sums = append(ats.Sums, sum)
atsMu.Unlock()
ats.Sums = append(ats.Sums, sum)
return nil
})
}
return
err = g.Wait()
return ats, err
}
// returned by WriteFragmentChecksums