From 8da15b18d421a687dadeb786d0eea45333ff109f Mon Sep 17 00:00:00 2001 From: Jason Aten Date: Fri, 9 Oct 2020 11:14:41 -0500 Subject: [PATCH] pilosa-fsck: parallelize Index.ComputeTranslatorSummary - add -index option --- cmd/pilosa-fsck/fsck.go | 50 ++++++++++++--- index.go | 134 +++++++++++++++++++++++----------------- 2 files changed, 118 insertions(+), 66 deletions(-) diff --git a/cmd/pilosa-fsck/fsck.go b/cmd/pilosa-fsck/fsck.go index 2bb6b892e..240276268 100644 --- a/cmd/pilosa-fsck/fsck.go +++ b/cmd/pilosa-fsck/fsck.go @@ -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) { diff --git a/index.go b/index.go index 56adbd91b..b5c652c1c 100644 --- a/index.go +++ b/index.go @@ -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