mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-12 23:51:03 +00:00
remove pilosa-fsck
This commit is contained in:
Travis
parent
a4b37273ea
commit
c8c59b649d
8 changed files with 0 additions and 1924 deletions
|
|
@ -1,36 +0,0 @@
|
|||
.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
|
||||
|
|
@ -1,989 +0,0 @@
|
|||
// Copyright 2020 Pilosa Corp.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"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/pilosa/pilosa/v2/topology"
|
||||
"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
|
||||
JustThisIndex string // -index
|
||||
|
||||
// -col column key dump only options:
|
||||
// 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.
|
||||
Dirs []string
|
||||
|
||||
Verbose bool // -v
|
||||
Quiet bool // -q
|
||||
|
||||
// manual workaround for not having PilosaConfigPath, if really need be.
|
||||
ReplicaN int // -replicas
|
||||
PilosaConfigPath string // -config
|
||||
|
||||
ParallelReaders int // -readers
|
||||
|
||||
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.IntVar(&cfg.ParallelReaders, "readers", 10, "how many parallel readers to use to scan at once. 0 means do everything possible in parallel. 1 means serialize everything through a single reader. Can be adjusted to control memory consumption.")
|
||||
|
||||
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
|
||||
|
||||
-fix
|
||||
(warning: alters the backed-up node images on disk) copy primary data to replicas to create a consistent cluster.
|
||||
|
||||
-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.
|
||||
|
||||
-index index_name
|
||||
(optional) restrict to just this index. Otherwise we default to all indexes.
|
||||
|
||||
-readers PR
|
||||
how many parallel readers to use to scan at once. PR==0 means do everything
|
||||
possible in parallel. PR==1 means serialize everything through a single reader.
|
||||
Adjust PR to control memory consumption if needed. As a practical limit, setting
|
||||
PR > 10000 will have no effect. (default is 10).
|
||||
|
||||
-q
|
||||
be very quiet during analysis and repair
|
||||
|
||||
`)
|
||||
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 Pilosa data 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 in repair-mode with -fix. The console output
|
||||
supplies a log documenting the analysis
|
||||
and showing what data changes would have been made.
|
||||
|
||||
REQUIRED COMMAND LINE ARGUMENTS
|
||||
|
||||
The paths to all the top-level Pilosa
|
||||
data 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 file, the .topology file, and the 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 &> log
|
||||
|
||||
A typical invocation to repair the replication in the same backup:
|
||||
|
||||
$ pilosa-fsck -replicas 3 -fix node1/.pilosa node2/.pilosa node3/.pilosa node4/.pilosa &> log
|
||||
|
||||
In both cases, the .id and .topology files must
|
||||
be present in the backups.
|
||||
|
||||
Without -fix, no modifications will be made to the backups. Only
|
||||
by running with -fix will repairs be made. The user can safely
|
||||
always run with -fix to repair only if needed.
|
||||
|
||||
A zero error code will be returned to the shell if no repairs were needed.
|
||||
|
||||
A zero error code will be also be returned to the shell if
|
||||
repairs were needed and they were accomplished under -fix.
|
||||
|
||||
A non-zero error code indicates that repairs were needed but
|
||||
were not made.
|
||||
`)
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
cfg.Verbose = true
|
||||
|
||||
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
|
||||
|
||||
fixNeeded, err := cfg.Run()
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stdout, "# finished at %v (elapsed %v)\n\n", time.Now().Format(RFC3339MsecTz0), time.Since(t0))
|
||||
fmt.Fprintf(os.Stderr, "error: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
if fixNeeded && !cfg.Fix {
|
||||
fmt.Fprintf(os.Stdout, "# finished at %v (elapsed %v)\n\n", time.Now().Format(RFC3339MsecTz0), time.Since(t0))
|
||||
fmt.Fprintf(os.Stderr, "# pilosa-fsck exiting with non-zero error code because a repair is needed, but -fix was not given.\n")
|
||||
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 indexesFromAts(ats *pilosa.AllTranslatorSummary) (indexes []string) {
|
||||
indexMap := make(map[string]bool)
|
||||
for _, sum := range ats.Sums {
|
||||
if !indexMap[sum.Index] {
|
||||
indexMap[sum.Index] = true
|
||||
indexes = append(indexes, sum.Index)
|
||||
}
|
||||
}
|
||||
sort.Strings(indexes)
|
||||
return
|
||||
}
|
||||
|
||||
func (cfg *FsckConfig) RepairTranslationStores(ats *pilosa.AllTranslatorSummary) (err error) {
|
||||
|
||||
verbose := cfg.Verbose
|
||||
|
||||
// group by index first. then repair.
|
||||
indexes := indexesFromAts(ats)
|
||||
|
||||
for _, index := range indexes {
|
||||
|
||||
if !cfg.DoingIndex(index) {
|
||||
continue
|
||||
}
|
||||
|
||||
m := make(map[int]*group)
|
||||
for _, sum := range ats.Sums {
|
||||
|
||||
if !sum.IsColKey || sum.Index != index {
|
||||
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, topology.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...\n\n", dir)
|
||||
}
|
||||
|
||||
jmphasher := &topology.Jmphasher{}
|
||||
partitionN := topology.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() {
|
||||
|
||||
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, cfg.ParallelReaders)
|
||||
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
|
||||
}
|
||||
|
||||
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 topology.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)
|
||||
}
|
||||
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
snap := topology.NewClusterSnapshot(cfg.topo, cfg.topo.Hasher, cfg.topo.ReplicaN)
|
||||
|
||||
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 := snap.ReplicasForPrimary(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(index, 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 was 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
|
||||
#
|
||||
# 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, index, nDir, cfg.ReplicaN, humanize.Comma(totalBytes), humanize.Comma(totalFiles), humanize.Comma(int64(nDir*topology.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(index string, ats *pilosa.AllTranslatorSummary) (keyCount, idCount int) {
|
||||
for _, sum := range ats.Sums {
|
||||
if sum.Index == index {
|
||||
keyCount += sum.KeyCount
|
||||
idCount += sum.IDCount
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
|
@ -1,448 +0,0 @@
|
|||
// Copyright 2020 Pilosa Corp.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"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) {
|
||||
t.Skip("I don't quite understand what this test is doing and will need help adjusting it to pass again.")
|
||||
// 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 := []string{"rick", "morty"}
|
||||
fieldName := []string{"f", "flying_car"}
|
||||
idx := make([]*pilosa.Index, len(index))
|
||||
field := make([]*pilosa.Field, len(index))
|
||||
var err error
|
||||
|
||||
for i := range index {
|
||||
|
||||
idx[i], err = nodes[0].API.CreateIndex(ctx, index[i], pilosa.IndexOptions{Keys: true, TrackExistence: true})
|
||||
if err != nil {
|
||||
t.Fatalf("creating index: %v", err)
|
||||
}
|
||||
if idx[i].CreatedAt() == 0 {
|
||||
t.Fatal("index createdAt is empty")
|
||||
}
|
||||
|
||||
field[i], err = nodes[0].API.CreateField(ctx, index[i], fieldName[i], pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100))
|
||||
if err != nil {
|
||||
t.Fatalf("creating field: %v", err)
|
||||
}
|
||||
if field[i].CreatedAt() == 0 {
|
||||
t.Fatal("field createdAt is empty")
|
||||
}
|
||||
}
|
||||
|
||||
rowID := uint64(1)
|
||||
timestamp := int64(0)
|
||||
|
||||
for i := range index {
|
||||
|
||||
// Generate some keyed records.
|
||||
rowIDs := []uint64{}
|
||||
timestamps := []int64{}
|
||||
N := 10
|
||||
for j := 1; j <= N; j++ {
|
||||
rowIDs = append(rowIDs, rowID)
|
||||
timestamps = append(timestamps, timestamp)
|
||||
}
|
||||
|
||||
var colKeys []string
|
||||
switch i {
|
||||
case 0:
|
||||
// Keys are sharded so ordering is not guaranteed.
|
||||
colKeys = []string{"col10", "col8", "col9", "col6", "col7", "col4", "col5", "col2", "col3", "col1"}
|
||||
colKeys = colKeys[:N]
|
||||
case 1:
|
||||
colKeys = []string{"col11", "col12"}
|
||||
N = len(colKeys)
|
||||
rowIDs = rowIDs[:N]
|
||||
timestamps = timestamps[: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[i],
|
||||
IndexCreatedAt: idx[i].CreatedAt(),
|
||||
Field: fieldName[i],
|
||||
FieldCreatedAt: field[i].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())
|
||||
//qcx.Reset()
|
||||
|
||||
pql := fmt.Sprintf("Row(%s=%d)", fieldName[i], rowID)
|
||||
|
||||
// Query node0.
|
||||
if res, err := nodes[0].API.Query(ctx, &pilosa.QueryRequest{Index: index[i], 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[i], 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.
|
||||
targetIndex := index[0]
|
||||
// 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,
|
||||
ParallelReaders: 5,
|
||||
}
|
||||
panicOn(cfg.ValidateConfig())
|
||||
|
||||
// for this test, mess up a replica that is not the primary.
|
||||
|
||||
h := targetNode.API.Holder()
|
||||
idx[0] = h.Index(index[0])
|
||||
store := idx[0].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)
|
||||
|
||||
targetIndex1 := "morty"
|
||||
targetPartition1 := 226 // for "col11"
|
||||
// # fsck_test.go:248 2020-10-06T20:24:33.755576-05:00 on k=47, idx[1]: targetPartition=47, store.PartitionID=0x4abe160, before corruption, fwd1='map[string]uint64{"col12":0xcf00001}', rev1='map[uint64]string{0xcf00001:"col12"}'
|
||||
//# fsck_test.go:248 2020-10-06T20:24:35.608568-05:00 on k=226, idx[1]: targetPartition=226, store.PartitionID=0x4abe160, before corruption, fwd1='map[string]uint64{"col11":0xcc00001}', rev1='map[uint64]string{0xcc00001:"col11"}'
|
||||
idx[1] = h.Index(index[1])
|
||||
store1 := idx[1].TranslateStore(targetPartition1)
|
||||
fwd1, rev1 := getFwdRev(store1, targetPartition1)
|
||||
//vv("on k=%v, idx[1]: targetPartition=%v, store.PartitionID=%v, before corruption, fwd1='%#v', rev1='%#v'", k, targetPartition1, store.PartitionID, fwd1, rev1)
|
||||
|
||||
presz1 := len(rev1)
|
||||
delete(rev1, fwd1["col11"])
|
||||
postsz1 := len(rev1)
|
||||
|
||||
if postsz1 == presz1 {
|
||||
panic("did not delete any key!")
|
||||
}
|
||||
bolt1 := store1.(*boltdb.TranslateStore)
|
||||
if err := bolt1.SetFwdRevMaps(nil, fwd1, rev1); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// done corrupting.
|
||||
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, targetIndex, 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.")
|
||||
}
|
||||
|
||||
chk1, err := check(dirs, cfg, targetIndex1, targetPartition1)
|
||||
_ = chk1
|
||||
//vv("pre-fix, chk1='%v'; err='%v'", chk1, 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, targetIndex, targetPartition)
|
||||
_ = chk
|
||||
//vv("chk = '%v' after repair; err='%v'", chk, err)
|
||||
panicOn(err)
|
||||
|
||||
chk1, err = check(dirs, cfg, targetIndex1, targetPartition1)
|
||||
_ = chk1
|
||||
//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, targetIndex string, targetPartition int) (chksum string, err error) {
|
||||
//vv("top of check, dirs = '%#v', targetIndex='%v', targetPartition='%v'", dirs, targetIndex, targetPartition)
|
||||
//defer vv("returning from check()")
|
||||
|
||||
firstChecksum := ""
|
||||
firstDir := ""
|
||||
firstStorePath := ""
|
||||
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)
|
||||
indexes := indexesFromAts(ats)
|
||||
//vv("indexes = '%#v'", indexes)
|
||||
|
||||
for _, index := range indexes {
|
||||
|
||||
if index != targetIndex {
|
||||
continue
|
||||
}
|
||||
for _, s := range ats.Sums {
|
||||
//vv(" s= '%#v'", s)
|
||||
if s.Index != index {
|
||||
//vv("skipping s.Index '%v' != index '%v'", s.Index, index)
|
||||
continue
|
||||
}
|
||||
if s.PartitionID != targetPartition {
|
||||
continue
|
||||
}
|
||||
//vv("accepting s.PartitionID(%v) == targetPartition(%v); s.Index '%v'; "+
|
||||
//"index '%v'; s.IsPrimary=%v, s.IsReplica=%v, s='%#v'; s.Checksum='%v', firstChecksum='%v'",
|
||||
//s.PartitionID, targetPartition, s.Index, index,
|
||||
//s.IsPrimary, s.IsReplica, s, s.Checksum, firstChecksum)
|
||||
|
||||
if s.IsPrimary || s.IsReplica {
|
||||
chksum := s.Checksum
|
||||
if firstChecksum == "" {
|
||||
|
||||
firstChecksum = chksum
|
||||
firstDir = dir
|
||||
firstStorePath = s.StorePath
|
||||
|
||||
} else {
|
||||
//vv("targetIndex = '%v'; firstChecksum='%v', chksum='%v'", targetIndex, firstChecksum, chksum)
|
||||
|
||||
if chksum != firstChecksum {
|
||||
return chksum, fmt.Errorf("bolt chksum on node %v '%v' disagrees with '%v' on '%v'; index='%v'; s.StorePath = '%v'; firstStorePath='%v'", dir, chksum, firstChecksum, firstDir, index, s.StorePath, firstStorePath)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return firstChecksum, nil
|
||||
}
|
||||
|
||||
// These are here to satisfy the linter in CI while the test is being skipped.
|
||||
var _ = getFwdRev
|
||||
var _ = check
|
||||
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)
|
||||
}
|
||||
|
|
@ -1 +0,0 @@
|
|||
pilosa-fsck
|
||||
|
|
@ -1,252 +0,0 @@
|
|||
Design for pilosa-fsck
|
||||
======================
|
||||
|
||||
Problem Background
|
||||
------------------
|
||||
|
||||
Molecula Pilosa provides replication for fault-tolerance within a Pilosa cluster.
|
||||
|
||||
Three kinds of data are replicated: Roaring bitmap data, Column-Key translation data,
|
||||
and Row-Key data are replicated. Only the first two, Roaring data and Column-Key
|
||||
data are relevant here. Broadly, the Roaring bitmap data
|
||||
forms the central features -- the bits -- of a large, sparse bitmap matrix.
|
||||
The Column-Keys are the labels for the columns at the top margin of this matrix.
|
||||
|
||||
For speed, the Roaring bitmap data is stored separately from the
|
||||
Key data. The Roaring data is stored in sharded files
|
||||
within a directory heirarchy under PILOSA-DATA-DIR/index_name/field_name/...
|
||||
The Key translation data is stored in sharded BoltDB databases within
|
||||
the PILOSA-DATA-DIR/index_name/_key directory.
|
||||
|
||||
The current approach to Roaring file replication involves an
|
||||
eventually consistent mechanism that uses an Anti-Entropy agent to
|
||||
fix partial or incomplete replication from the primary shard to all
|
||||
replica shards.
|
||||
|
||||
Unfortunately, the Anti-Entropy agent approach has proved inadequate on two
|
||||
fronts. First, it does not provide for immediately consistent reads in the
|
||||
event that the primary is lost. Second, the Anti-Entropy agent itself experienced
|
||||
out-of-memory issues that have yet to be resolved.
|
||||
|
||||
Therefore, work is now underway to replace this replication
|
||||
approach with a more consistent design.
|
||||
|
||||
However, in the meantime, for our customers in production with Molecula
|
||||
Pilosa, we wish to provide a means to re-establish correct replication.
|
||||
Thus even in the event of a node failure followed by a read from a replica, the
|
||||
returned read will be correct.
|
||||
|
||||
The pilosa-fsck tool can therefore be seen as a temporary, stop-gap
|
||||
measure to address immediate issues while the cluster replication
|
||||
mechanism is replaced.
|
||||
|
||||
The second factor motivating the creation of pilosa-fsck was the discovery
|
||||
of a bug in the Key-translation process. Unfortunately this was a hard
|
||||
to reproduce bug. It happened only on the customer's premises,
|
||||
and only after running the system for a long time, with a
|
||||
large amount of data, and with various eccentric node failures
|
||||
and recoveries.
|
||||
|
||||
However, we were able to reproduce a plausible explanation.
|
||||
Non-primary replicas were creating keys when they should have been
|
||||
forwarding the request to the primary. Correcting this bug is impetus
|
||||
for the v2.1.4 release of Molecula Pilosa.
|
||||
|
||||
A fine point here: since we were not able to precisely reproduce the customer's
|
||||
issue in the development environment, we cannot guarantee with 100%
|
||||
certainty that we have actually addressed the bug that the customer
|
||||
was seeing.
|
||||
|
||||
Therefore we also desired an additional insurance
|
||||
policy. We wished to be able to empower customers to proactively discover any
|
||||
future Key-translation issues that happen in their on-premise systems.
|
||||
|
||||
To do this, we proposed providing select customers with the pilosa-fsck
|
||||
tool which can analyze their offline backups for issues.
|
||||
|
||||
Optionally, these issues can also be repaired in-place in the
|
||||
offline backup on which pilosa-fsck is run.
|
||||
|
||||
The -fix flag repairs both kinds of replication issues.
|
||||
|
||||
Solution Approach: mechanism of action
|
||||
--------------------------------------
|
||||
|
||||
The pilosa-fsck is run offline on a full set of backups taken from
|
||||
all nodes in a Pilosa cluster. It runs on a single computer that
|
||||
must be separate from the production or staging Pilosa environments.
|
||||
|
||||
When run, pilosa-fsck analyzes the differences between the
|
||||
primary and its replicas. Both the Roaring
|
||||
files and the Key translation databases are analyzed.
|
||||
The computer running pilosa-fsck must have the same or more
|
||||
memory as the Pilosa nodes in the cluster, as it will
|
||||
"pretend" to be each Pilosa node in turn. However, as each
|
||||
node's backup is closed before the next node's backup is
|
||||
opened, we do not require substantially more memory than a single
|
||||
production node. Short Blake3 cryptographic checksums are
|
||||
computed for each Roaring fragment and each Key translation
|
||||
database. These are held in memory (and printed to the log)
|
||||
for comparing nodes. This comparison forms the heart of
|
||||
the consistency checks, and is the basis for any subsequent
|
||||
repair.
|
||||
|
||||
We recommend capturing both stdout and stderr to a log.
|
||||
Use `&> log` or `2>&1 > log` at the end of the
|
||||
pilosa-fsck invocation to save a log of the run to disk.
|
||||
|
||||
In a typical cluster, the Replication factor R may be less
|
||||
than the number of nodes N in the cluster. For example, while
|
||||
N may be 4, the R may be only 3. In this example, within
|
||||
each replicated shard, one node will be the primary for
|
||||
that shard, two nodes will be non-primary replicas, and one
|
||||
node will be a non-replica. Note that the designation
|
||||
of primary changes for different Roaring shards within an index,
|
||||
even on a single node.
|
||||
|
||||
The essence of the the -fix repair operation that pilosa-fsck
|
||||
can do is this: it will copy from the primary to the
|
||||
the non-primary replicas. Further, it will remove data from
|
||||
any non-replica node if it was mistakenly present.
|
||||
|
||||
The pilosa-fsck output log will contain
|
||||
a sequence of command line 'cp' and 'rm' commands.
|
||||
These commands are merely a record (with
|
||||
accompanying justifcation in the comment following the
|
||||
command) of what actions would be performed to repair
|
||||
the Roaring file data.
|
||||
|
||||
Only with -fix will the repair actions actually happen
|
||||
during the pilosa-fsck run.
|
||||
|
||||
|
||||
Details: running pilosa-fsck
|
||||
----------------------------
|
||||
|
||||
Errors in invocation are reported on stderr and the program will exit with a non-zero
|
||||
error code if invocation errors are present. A non-zero error code
|
||||
is returned if a repair is needed and -fix was not given.
|
||||
|
||||
A -fix run will return a zero error code to the shell if the fix was
|
||||
successfully made; or if no fix was required.
|
||||
|
||||
The log of the run is printed to stdout.
|
||||
|
||||
The -h flag to pilosa-fsck prints a summary of its operation
|
||||
and a guide to laying out the backup directories.
|
||||
|
||||
The help is reproduced below.
|
||||
|
||||
~~~
|
||||
$ pilosa-fsck version: Molecula Pilosa v2.2.1-43-g9dacbccf (Oct 5 2020 1:28PM, 9dacbccf)
|
||||
|
||||
Use: pilosa-fsck -replicas R {-fix} {-q} /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.
|
||||
|
||||
-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.
|
||||
|
||||
-q
|
||||
be very quiet during analysis and repair
|
||||
|
||||
|
||||
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 Pilosa data 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 in repair-mode with -fix. The console output
|
||||
supplies a log documenting the analysis
|
||||
and showing what data changes would have been made.
|
||||
|
||||
REQUIRED COMMAND LINE ARGUMENTS
|
||||
|
||||
The paths to all the top-level Pilosa
|
||||
data 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 file, the .topology file, and the 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 &> log
|
||||
|
||||
A typical invocation to repair the replication in the same backup:
|
||||
|
||||
$ pilosa-fsck -replicas 3 -fix node1/.pilosa node2/.pilosa node3/.pilosa node4/.pilosa &> log
|
||||
|
||||
In both cases, the .id and .topology files must
|
||||
be present in the backups.
|
||||
|
||||
Without -fix, no modifications will be made to the backups. Only
|
||||
by running with -fix will repairs be made. The user can safely
|
||||
always run with -fix to repair only if needed.
|
||||
|
||||
A zero error code will be returned to the shell if no repairs were needed.
|
||||
|
||||
A zero error code will be also be returned to the shell if
|
||||
repairs were needed and they were accomplished under -fix.
|
||||
|
||||
A non-zero error code indicates that repairs were needed but
|
||||
were not made.
|
||||
|
||||
~~~
|
||||
Binary file not shown.
|
|
@ -1,21 +0,0 @@
|
|||
#!/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
|
||||
|
|
@ -1,177 +0,0 @@
|
|||
// 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
|
||||
Loading…
Add table
Reference in a new issue