diff --git a/api.go b/api.go index 4c8221170..b6ceadbb2 100644 --- a/api.go +++ b/api.go @@ -214,7 +214,6 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) { - fmt.Println("--- DEBUG: forward to coordinator") if err := api.server.defaultClient.CreateIndex(ctx, indexName, options); err != nil { return nil, errors.Wrap(err, "forwarding CreateIndex to coordinator") } diff --git a/cluster.go b/cluster.go index 51aff4e4d..0bb39e80f 100644 --- a/cluster.go +++ b/cluster.go @@ -105,7 +105,7 @@ type cluster struct { // nolint: maligned sharder disco.Sharder // Required for cluster Resize. - Static bool // Static is primarily used for testing in a non-gossip environment. + Static bool // Static is primarily used for testing. holder *Holder broadcaster broadcaster diff --git a/cmd/pilosa-fsck/Makefile b/cmd/pilosa-fsck/Makefile deleted file mode 100644 index 1b1dcf14c..000000000 --- a/cmd/pilosa-fsck/Makefile +++ /dev/null @@ -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 diff --git a/cmd/pilosa-fsck/fsck.go b/cmd/pilosa-fsck/fsck.go deleted file mode 100644 index fc98fe574..000000000 --- a/cmd/pilosa-fsck/fsck.go +++ /dev/null @@ -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 -} diff --git a/cmd/pilosa-fsck/fsck_test.go b/cmd/pilosa-fsck/fsck_test.go deleted file mode 100644 index 2555215cc..000000000 --- a/cmd/pilosa-fsck/fsck_test.go +++ /dev/null @@ -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) -} diff --git a/cmd/pilosa-fsck/release-pilosa-fsck/.gitignore b/cmd/pilosa-fsck/release-pilosa-fsck/.gitignore deleted file mode 100644 index a08586f1c..000000000 --- a/cmd/pilosa-fsck/release-pilosa-fsck/.gitignore +++ /dev/null @@ -1 +0,0 @@ -pilosa-fsck diff --git a/cmd/pilosa-fsck/release-pilosa-fsck/DESIGN.md b/cmd/pilosa-fsck/release-pilosa-fsck/DESIGN.md deleted file mode 100644 index 598896c8d..000000000 --- a/cmd/pilosa-fsck/release-pilosa-fsck/DESIGN.md +++ /dev/null @@ -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. - -~~~ diff --git a/cmd/pilosa-fsck/release-pilosa-fsck/backups.tar.gz b/cmd/pilosa-fsck/release-pilosa-fsck/backups.tar.gz deleted file mode 100644 index 28b08adbd..000000000 Binary files a/cmd/pilosa-fsck/release-pilosa-fsck/backups.tar.gz and /dev/null differ diff --git a/cmd/pilosa-fsck/release-pilosa-fsck/example.sh b/cmd/pilosa-fsck/release-pilosa-fsck/example.sh deleted file mode 100755 index 79fd4cf11..000000000 --- a/cmd/pilosa-fsck/release-pilosa-fsck/example.sh +++ /dev/null @@ -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 diff --git a/cmd/pilosa-fsck/vprint.go b/cmd/pilosa-fsck/vprint.go deleted file mode 100644 index 83b1681f7..000000000 --- a/cmd/pilosa-fsck/vprint.go +++ /dev/null @@ -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 diff --git a/cmd/server_test.go b/cmd/server_test.go index b99d88ab9..f698b9e20 100644 --- a/cmd/server_test.go +++ b/cmd/server_test.go @@ -201,8 +201,6 @@ func TestServerConfig_DeprecateLongQueryTime(t *testing.T) { bind = ` + nextPort() + ` bind-grpc = ` + nextPort() + ` data-dir = "` + actualDataDir + `" - [gossip] - port = "14321" `, validation: func() error { v := validator{} @@ -218,8 +216,6 @@ func TestServerConfig_DeprecateLongQueryTime(t *testing.T) { cfgFileContent: ` bind = ` + nextPort() + ` bind-grpc = ` + nextPort() + ` - [gossip] - port = "14321" `, validation: func() error { v := validator{} @@ -235,8 +231,6 @@ func TestServerConfig_DeprecateLongQueryTime(t *testing.T) { cfgFileContent: ` bind = ` + nextPort() + ` bind-grpc = ` + nextPort() + ` - [gossip] - port = "14321" `, validation: func() error { v := validator{} diff --git a/gossip/gossip.go b/gossip/gossip.go index 2d3b413d0..10b3e2d0e 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -15,556 +15,9 @@ package gossip import ( - "bytes" - "context" - "fmt" - "io" - "io/ioutil" - "log" - "net" - "os" - "strconv" - "strings" - "sync" - "time" - - "github.com/hashicorp/memberlist" - "github.com/pilosa/pilosa/v2" - "github.com/pilosa/pilosa/v2/logger" - pnet "github.com/pilosa/pilosa/v2/net" - "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/toml" - "github.com/pilosa/pilosa/v2/topology" - "github.com/pkg/errors" ) -// Ensure GossipMemberSet implements interfaces. -var _ memberlist.Delegate = &memberSet{} - -// memberSet represents a gossip implementation of MemberSet using memberlist. -type memberSet struct { - mu sync.RWMutex - memberlist *memberlist.Memberlist - - broadcasts *memberlist.TransmitLimitedQueue - - papi *pilosa.API - config *config - - Logger logger.Logger - - // stdLogger is only used when passed into memberlist library things that take a std library logger rather than an interface. - stdLogger *log.Logger - // logOutput is similar to stdLogger in that it's passed to memberlist things which can't take a pilosa Logger. - logOutput io.Writer - - transport *Transport - - eventReceiver *eventReceiver -} - -// Open implements the MemberSet interface to start network activity. -func (g *memberSet) Open() (err error) { - g.mu.Lock() - defer g.mu.Unlock() - - g.memberlist, err = memberlist.Create(g.config.memberlistConfig) - - if err != nil { - return errors.Wrap(err, "creating memberlist") - } - - g.broadcasts = &memberlist.TransmitLimitedQueue{ - NumNodes: func() int { - g.mu.RLock() - defer g.mu.RUnlock() - return g.memberlist.NumMembers() - }, - RetransmitMult: 3, - } - - var uris = make([]*pnet.URI, len(g.config.gossipSeeds)) - for i, addr := range g.config.gossipSeeds { - uris[i], err = pnet.NewURIFromAddress(addr) - if err != nil { - return fmt.Errorf("new uri from address: %s", err) - } - } - - var nodes = make([]*topology.Node, len(uris)) - for i, uri := range uris { - nodes[i] = &topology.Node{URI: *uri} - } - - err = g.joinWithRetry(pnet.URIs(topology.Nodes(nodes).URIs()).HostPortStrings()) - if err != nil { - return errors.Wrap(err, "joinWithRetry") - } - return nil -} - -// Close attempts to gracefully leave the cluster, and finally calls shutdown -// after (at most) a timeout period. -func (g *memberSet) Close() error { - defer g.eventReceiver.Close() - - leaveErr := g.memberlist.Leave(5 * time.Second) - shutdownErr := g.memberlist.Shutdown() - if leaveErr != nil || shutdownErr != nil { - return fmt.Errorf("leaving: '%v', shutting down: '%v'", leaveErr, shutdownErr) - } - return nil -} - -// joinWithRetry wraps the standard memberlist Join function in a retry. -func (g *memberSet) joinWithRetry(hosts []string) error { - err := retry(60, 2*time.Second, func() error { - _, err := g.memberlist.Join(hosts) - return err - }) - return err -} - -// retry periodically retries function fn a specified number of attempts. -func retry(attempts int, sleep time.Duration, fn func() error) (err error) { // nolint: unparam - for i := 0; ; i++ { - err = fn() - if err == nil { - return - } - if i >= (attempts - 1) { - break - } - time.Sleep(sleep) - log.Println("retrying after error:", err) - } - return fmt.Errorf("after %d attempts, last error: %s", attempts, err) -} - -//////////////////////////////////////////////////////////////// - -type config struct { - gossipSeeds []string - memberlistConfig *memberlist.Config -} - -// memberSetOption describes a functional option for GossipMemberSet. -type memberSetOption func(*memberSet) error - -// WithTransport is a functional option for providing a transport to NewMemberSet. -func WithTransport(transport *Transport) memberSetOption { - return func(g *memberSet) error { - g.transport = transport - return nil - } -} - -// WithLogger is a functional option for providing a Go logger to NewMemberSet. -// If the memberSet's transport is nil, this logger will be used when creating -// one. If WithLogOutput is not used, this logger will be passed to memberlist -// for it to use internally. This logger is not used for logging by code in this -// (gossip) package - for that, use the WithPilosaLogger option. -func WithLogger(logger *log.Logger) memberSetOption { - return func(g *memberSet) error { - g.stdLogger = logger - return nil - } -} - -// WithLogOutput allows one to pass a Writer which will in turn be passed to -// memberlist for use in logging. -func WithLogOutput(o io.Writer) memberSetOption { - return func(g *memberSet) error { - g.logOutput = o - return nil - } -} - -// WithPilosaLogger allows one to configure a memberSet with a logger of their -// choice which satisfies the pilosa logger interface. -func WithPilosaLogger(l logger.Logger) memberSetOption { - return func(g *memberSet) error { - g.Logger = l - return nil - } -} - -// NewMemberSet returns a new instance of GossipMemberSet based on options. The -// logging options which can be passed to NewMemberSet are complicated for -// historical reasons - please pass WithPilosaLogger, and either WithLogOutput -// or WithLogger. If you pass WithLogOutput, be sure to also pass in a Transport -// using WithTransport. -func NewMemberSet(cfg Config, api *pilosa.API, options ...memberSetOption) (*memberSet, error) { - host := api.Node().URI.Host - g := &memberSet{ - papi: api, - Logger: logger.NopLogger, - } - - // options - for _, opt := range options { - if err := opt(g); err != nil { - return nil, errors.Wrap(err, "executing option") - } - } - - ger := newEventReceiver(g.Logger, api) - g.eventReceiver = ger - - if g.transport == nil { - port, err := strconv.Atoi(cfg.Port) - if err != nil { - return nil, fmt.Errorf("convert port: %s", err) - } - - if g.stdLogger == nil { - if g.logOutput != nil { - g.stdLogger = logger.NewStandardLogger(g.logOutput).Logger() - } else { - g.stdLogger = log.New(os.Stderr, "", log.LstdFlags) - } - } - - // Set up the transport. - transport, err := NewTransport(host, port, g.stdLogger) - if err != nil { - return nil, fmt.Errorf("new tranport: %s", err) - } - - g.transport = transport - } - - port := g.transport.net.GetAutoBindPort() - - var gossipKey []byte - var err error - if cfg.Key != "" { - gossipKey, err = ioutil.ReadFile(cfg.Key) - if err != nil { - return nil, fmt.Errorf("reading gossip key: %s", err) - } - } - - //////////////////// - // memberlist config - conf := memberlist.DefaultWANConfig() - conf.Transport = g.transport.net - conf.Name = api.Node().ID - conf.BindAddr = api.Node().URI.Host - conf.BindPort = port - // AdvertisePort - if cfg.AdvertisePort != "" { - if p, err := strconv.Atoi(cfg.Port); err != nil { - return nil, fmt.Errorf("convert advertise port: %s", err) - } else { - conf.AdvertisePort = p - } - } else { - conf.AdvertisePort = port - } - // AdvertiseHost - if cfg.AdvertiseHost != "" { - conf.AdvertiseAddr = cfg.AdvertiseHost - } else { - conf.AdvertiseAddr = hostToIP(api.Node().URI.Host) - } - // - conf.TCPTimeout = time.Duration(cfg.StreamTimeout) - conf.SuspicionMult = cfg.SuspicionMult - conf.PushPullInterval = time.Duration(cfg.PushPullInterval) - conf.ProbeTimeout = time.Duration(cfg.ProbeTimeout) - conf.ProbeInterval = time.Duration(cfg.ProbeInterval) - conf.GossipNodes = cfg.Nodes - conf.GossipInterval = time.Duration(cfg.Interval) - conf.GossipToTheDeadTime = time.Duration(cfg.ToTheDeadTime) - // - conf.Delegate = g - conf.SecretKey = gossipKey - conf.Events = ger - if g.logOutput != nil { - conf.LogOutput = g.logOutput - } else { - conf.Logger = g.stdLogger - } - - g.config = &config{ - memberlistConfig: conf, - gossipSeeds: cfg.Seeds, - } - - return g, nil -} - -// NodeMeta implementation of the memberlist.Delegate interface. -func (g *memberSet) NodeMeta(limit int) []byte { - buf, err := g.papi.Serializer.Marshal(g.papi.Node()) - if err != nil { - g.Logger.Printf("marshal message error: %s", err) - return []byte{} - } - return buf -} - -// NotifyMsg implementation of the memberlist.Delegate interface -// called when a user-data message is received. -func (g *memberSet) NotifyMsg(b []byte) { - err := g.papi.ClusterMessage(context.Background(), bytes.NewBuffer(b)) - if err != nil { - g.Logger.Printf("cluster message error: %s", err) - } -} - -// GetBroadcasts implementation of the memberlist.Delegate interface -// called when user data messages can be broadcast. -func (g *memberSet) GetBroadcasts(overhead, limit int) [][]byte { - return g.broadcasts.GetBroadcasts(overhead, limit) - -} - -// LocalState implementation of the memberlist.Delegate interface -// sends this Node's state data. -func (g *memberSet) LocalState(join bool) []byte { - schema, err := g.papi.Schema(context.Background()) - if err != nil { - // just panic, this code will be removed soon - panic(err) - } - m := &pilosa.NodeStatus{ - Node: g.papi.Node(), - Schema: &pilosa.Schema{Indexes: schema}, - } - for _, idx := range m.Schema.Indexes { - is := &pilosa.IndexStatus{Name: idx.Name, CreatedAt: idx.CreatedAt} - - for _, f := range idx.Fields { - availableShards := roaring.NewBitmap() - if field, _ := g.papi.Field(context.Background(), idx.Name, f.Name); field != nil { - availableShards = field.AvailableShards(false) - } - - fs := &pilosa.FieldStatus{ - Name: f.Name, - CreatedAt: f.CreatedAt, - AvailableShards: availableShards, - } - is.Fields = append(is.Fields, fs) - } - m.Indexes = append(m.Indexes, is) - } - - // Marshal nodestate data to bytes. - buf, err := pilosa.MarshalInternalMessage(m, g.papi.Serializer) - if err != nil { - g.Logger.Printf("error marshalling nodestate data, err=%s", err) - return []byte{} - } - return buf -} - -// MergeRemoteState implementation of the memberlist.Delegate interface -// receive and process the remote side's LocalState. -func (g *memberSet) MergeRemoteState(buf []byte, join bool) { - err := g.papi.ClusterMessage(context.Background(), bytes.NewBuffer(buf)) - if err != nil { - g.Logger.Printf("merge state error: %s", err) - } -} - -// eventReceiver is used to enable an application to receive -// events about joins and leaves over a channel. -// -// Care must be taken that events are processed in a timely manner from -// the channel, since this delegate will block until an event can be sent. -type eventReceiver struct { - ch chan memberlist.NodeEvent - closed chan struct{} - papi *pilosa.API - - logger logger.Logger -} - -// newEventReceiver returns a new instance of GossipEventReceiver. -func newEventReceiver(logger logger.Logger, papi *pilosa.API) *eventReceiver { - ger := &eventReceiver{ - ch: make(chan memberlist.NodeEvent, 1), - closed: make(chan struct{}), - logger: logger, - papi: papi, - } - go ger.listen() - return ger -} - -func (g *eventReceiver) NotifyJoin(n *memberlist.Node) { - // copy node to avoid data race - n2 := *n - n2.Meta = make([]byte, len(n.Meta)) - copy(n2.Meta, n.Meta) - - select { - case g.ch <- memberlist.NodeEvent{Event: memberlist.NodeJoin, Node: &n2}: - case <-g.closed: - } -} - -func (g *eventReceiver) NotifyLeave(n *memberlist.Node) { - // copy node to avoid data race - n2 := *n - n2.Meta = make([]byte, len(n.Meta)) - copy(n2.Meta, n.Meta) - - select { - case g.ch <- memberlist.NodeEvent{Event: memberlist.NodeLeave, Node: &n2}: - case <-g.closed: - } -} - -func (g *eventReceiver) NotifyUpdate(n *memberlist.Node) { - // copy node to avoid data race - n2 := *n - n2.Meta = make([]byte, len(n.Meta)) - copy(n2.Meta, n.Meta) - - select { - case g.ch <- memberlist.NodeEvent{Event: memberlist.NodeUpdate, Node: &n2}: - case <-g.closed: - } -} - -func (g *eventReceiver) Close() { - // TODO workaround to make tests pass. We are going to delete this code anyways. - select { - case <-g.closed: - return - default: - close(g.closed) - } -} - -func (g *eventReceiver) listen() { - var nodeEventType pilosa.NodeEventType - for { - var e memberlist.NodeEvent - select { - case <-g.closed: - return - case e = <-g.ch: - } - switch e.Event { - case memberlist.NodeJoin: - nodeEventType = pilosa.NodeJoin - case memberlist.NodeLeave: - nodeEventType = pilosa.NodeLeave - case memberlist.NodeUpdate: - nodeEventType = pilosa.NodeUpdate - default: - continue - } - - // Get the node from the event.Node meta data. - var n topology.Node - if err := g.papi.Serializer.Unmarshal(e.Node.Meta, &n); err != nil { - panic("failed to unmarshal event node meta into node") - } - - ne := &pilosa.NodeEvent{ - Event: nodeEventType, - Node: &n, - } - buf, err := pilosa.MarshalInternalMessage(ne, g.papi.Serializer) - if err != nil { - panic(err) - } - if err := g.papi.ClusterMessage(context.Background(), bytes.NewBuffer(buf)); err != nil { - g.logger.Printf("receive event error: %s", err) - } - } -} - -// Transport is a gossip transport for binding to a port. -type Transport struct { - //memberlist.Transport - net *memberlist.NetTransport - URI *pnet.URI -} - -// NewTransport returns a NetTransport based on the given host and port. -// It will dynamically bind to a port if port is 0. -// This is useful for test cases where specifying a port is not reasonable. -//func NewTransport(host string, port int) (*memberlist.NetTransport, error) { -func NewTransport(host string, port int, logger *log.Logger) (*Transport, error) { - // memberlist config - conf := memberlist.DefaultWANConfig() - conf.BindAddr = host - conf.BindPort = port - conf.AdvertisePort = port - conf.Logger = logger - - net, err := newTransport(conf) - if err != nil { - return nil, fmt.Errorf("new transport: %s", err) - } - - uri, err := pnet.NewURIFromHostPort(host, uint16(net.GetAutoBindPort())) - if err != nil { - return nil, fmt.Errorf("new uri from host port: %s", err) - } - - return &Transport{ - net: net, - URI: uri, - }, nil -} - -// newTransport returns a NetTransport based on the memberlist configuration. -// It will dynamically bind to a port if conf.BindPort is 0. -func newTransport(conf *memberlist.Config) (*memberlist.NetTransport, error) { - nc := &memberlist.NetTransportConfig{ - BindAddrs: []string{conf.BindAddr}, - BindPort: conf.BindPort, - Logger: conf.Logger, - } - - if conf.BindPort == 0 { - panic("TODO: remove this. problem: gossip conf.BindPort was 0!") - } - - // See comment below for details about the retry in here. - makeNetRetry := func(limit int) (*memberlist.NetTransport, error) { - var err error - for try := 0; try < limit; try++ { - var nt *memberlist.NetTransport - if nt, err = memberlist.NewNetTransport(nc); err == nil { - return nt, nil - } - if strings.Contains(err.Error(), "address already in use") { - conf.Logger.Printf("[DEBUG] Got bind error: %v", err) - continue - } - } - - return nil, fmt.Errorf("failed to obtain an address: %v", err) - } - - // The dynamic bind port operation is inherently racy because - // even though we are using the kernel to find a port for us, we - // are attempting to bind multiple protocols (and potentially - // multiple addresses) with the same port number. We build in a - // few retries here since this often gets transient errors in - // busy unit tests. - limit := 1 - if conf.BindPort == 0 { - limit = 10 - } - - nt, err := makeNetRetry(limit) - if err != nil { - return nil, errors.Wrap(err, "could not set up network transport") - } - - return nt, nil -} - // Config holds toml-friendly memberlist configuration. type Config struct { // Port indicates the port to which pilosa should bind for internal state sharing. @@ -638,21 +91,3 @@ type Config struct { Nodes int `toml:"nodes"` ToTheDeadTime toml.Duration `toml:"to-the-dead-time"` } - -// hostToIP converts host to an IP4 address based on net.LookupIP(). -func hostToIP(host string) string { - // if host is not an IP addr, check net.LookupIP() - if net.ParseIP(host) == nil { - hosts, err := net.LookupIP(host) - if err != nil { - return host - } - for _, h := range hosts { - // this restricts pilosa to IP4 - if h.To4() != nil { - return h.String() - } - } - } - return host -} diff --git a/server/cluster_test.go b/server/cluster_test.go index 23668c64f..cbadc6b65 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -29,7 +29,6 @@ import ( "github.com/pilosa/pilosa/v2/server" "github.com/pilosa/pilosa/v2/test" "github.com/pilosa/pilosa/v2/test/port" - "golang.org/x/sync/errgroup" ) // Ensure program can send/receive broadcast messages. @@ -173,8 +172,6 @@ func TestClusterResize_AddNode(t *testing.T) { m0 := test.MustRunCluster(t, 1).GetNode(0) defer m0.Close() - seed := "" - // Create a client for each node. client0 := m0.Client() @@ -188,19 +185,16 @@ func TestClusterResize_AddNode(t *testing.T) { // Configure node1 m1 := test.NewCommandNode(t) - m1.Config.Gossip.Seeds = []string{seed} - if err := port.GetListeners(func(lsns []*net.TCPListener) error { portsCfg := test.GenPortsConfig(test.NewPorts(lsns)) - m1.Config.Gossip.Port = portsCfg[0].Gossip.Port m1.Config.Etcd = portsCfg[0].Etcd m1.Config.Name = portsCfg[0].Name m1.Config.Cluster.Name = portsCfg[0].Cluster.Name m1.Config.BindGRPC = portsCfg[0].BindGRPC return m1.Start() - }, 4, 10); err != nil { + }, 3, 10); err != nil { t.Fatalf("starting second main: %v", err) } defer m1.Close() @@ -219,8 +213,6 @@ func TestClusterResize_AddNode(t *testing.T) { m0 := test.MustRunCluster(t, 1).GetNode(0) defer m0.Close() - seed := "" - // Create a client for each node. client0 := m0.Client() @@ -249,19 +241,16 @@ func TestClusterResize_AddNode(t *testing.T) { // Configure node1 m1 := test.NewCommandNode(t) - m1.Config.Gossip.Seeds = []string{seed} - if err := port.GetListeners(func(lsns []*net.TCPListener) error { portsCfg := test.GenPortsConfig(test.NewPorts(lsns)) - m1.Config.Gossip.Port = portsCfg[0].Gossip.Port m1.Config.Etcd = portsCfg[0].Etcd m1.Config.Name = portsCfg[0].Name m1.Config.Cluster.Name = portsCfg[0].Cluster.Name m1.Config.BindGRPC = portsCfg[0].BindGRPC return m1.Start() - }, 4, 10); err != nil { + }, 3, 10); err != nil { t.Fatalf("starting second main: %v", err) } defer m1.Close() @@ -283,8 +272,6 @@ func TestClusterResize_AddNode(t *testing.T) { m0 := test.MustRunCluster(t, 1).GetNode(0) defer m0.Close() - seed := "" - // Create a client for each node. client0 := m0.Client() @@ -309,19 +296,17 @@ func TestClusterResize_AddNode(t *testing.T) { // Configure node1 m1 := test.NewCommandNode(t) - m1.Config.Gossip.Seeds = []string{seed} if err := port.GetListeners(func(lsns []*net.TCPListener) error { portsCfg := test.GenPortsConfig(test.NewPorts(lsns)) - m1.Config.Gossip.Port = portsCfg[0].Gossip.Port m1.Config.Etcd = portsCfg[0].Etcd m1.Config.Name = portsCfg[0].Name m1.Config.Cluster.Name = portsCfg[0].Cluster.Name m1.Config.BindGRPC = portsCfg[0].BindGRPC return m1.Start() - }, 4, 10); err != nil { + }, 3, 10); err != nil { t.Fatalf("starting second main: %v", err) } defer m1.Close() @@ -345,8 +330,6 @@ func TestClusterResize_AddNode(t *testing.T) { m0 := test.MustRunCluster(t, 1).GetNode(0) defer m0.Close() - seed := "" - // Create a client for each node. client0 := m0.Client() @@ -375,19 +358,17 @@ func TestClusterResize_AddNode(t *testing.T) { // Configure node1 m1 := test.NewCommandNode(t) - m1.Config.Gossip.Seeds = []string{seed} if err := port.GetListeners(func(lsns []*net.TCPListener) error { portsCfg := test.GenPortsConfig(test.NewPorts(lsns)) - m1.Config.Gossip.Port = portsCfg[0].Gossip.Port m1.Config.Etcd = portsCfg[0].Etcd m1.Config.Name = portsCfg[0].Name m1.Config.Cluster.Name = portsCfg[0].Cluster.Name m1.Config.BindGRPC = portsCfg[0].BindGRPC return m1.Start() - }, 4, 10); err != nil { + }, 3, 10); err != nil { t.Fatalf("starting second main: %v", err) } @@ -416,8 +397,6 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { m0 := test.MustRunCluster(t, 1).GetNode(0) defer m0.Close() - seed := "" - // Create a client for each node. client0 := m0.Client() @@ -436,17 +415,15 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { // Configure node1 m1 := test.NewCommandNode(t) - m1.Config.Gossip.Seeds = []string{seed} if err := port.GetListeners(func(lsns []*net.TCPListener) error { portsCfg := test.GenPortsConfig(test.NewPorts(lsns)) - m1.Config.Gossip.Port = portsCfg[0].Gossip.Port m1.Config.Etcd = portsCfg[0].Etcd m1.Config.Name = portsCfg[0].Name m1.Config.Cluster.Name = portsCfg[0].Cluster.Name m1.Config.BindGRPC = portsCfg[0].BindGRPC return m1.Start() - }, 4, 10); err != nil { + }, 3, 10); err != nil { t.Fatalf("starting second main: %v", err) } defer m1.Close() @@ -468,8 +445,6 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { m0 := test.MustRunCluster(t, 1).GetNode(0) defer m0.Close() - seed := "" - // Create a client for each node. client0 := m0.Client() @@ -498,17 +473,15 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { // Configure node1 m1 := test.NewCommandNode(t) - m1.Config.Gossip.Seeds = []string{seed} if err := port.GetListeners(func(lsns []*net.TCPListener) error { portsCfg := test.GenPortsConfig(test.NewPorts(lsns)) - m1.Config.Gossip.Port = portsCfg[0].Gossip.Port m1.Config.Etcd = portsCfg[0].Etcd m1.Config.Name = portsCfg[0].Name m1.Config.Cluster.Name = portsCfg[0].Cluster.Name m1.Config.BindGRPC = portsCfg[0].BindGRPC return m1.Start() - }, 4, 10); err != nil { + }, 3, 10); err != nil { t.Fatalf("starting second main: %v", err) } errc := make(chan error, 1) @@ -536,8 +509,6 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { m0 := test.MustRunCluster(t, 1).GetNode(0) defer m0.Close() - seed := "" - // Create a client for each node. client0 := m0.Client() @@ -566,11 +537,9 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { // Configure node1 m1 := test.NewCommandNode(t) - m1.Config.Gossip.Seeds = []string{seed} if err := port.GetListeners(func(lsns []*net.TCPListener) error { portsCfg := test.GenPortsConfig(test.NewPorts(lsns)) - m1.Config.Gossip.Port = portsCfg[0].Gossip.Port m1.Config.Etcd = portsCfg[0].Etcd m1.Config.Name = portsCfg[0].Name m1.Config.Cluster.Name = portsCfg[0].Cluster.Name @@ -582,7 +551,7 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { errc <- err }() return m1.Start() - }, 4, 10); err != nil { + }, 3, 10); err != nil { t.Fatalf("starting second main: %v", err) } defer m1.Close() @@ -604,8 +573,6 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { m0 := test.MustRunCluster(t, 1).GetNode(0) defer m0.Close() - seed := "" - // Create a client for each node. client0 := m0.Client() @@ -632,11 +599,9 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { // Configure node1 m1 := test.NewCommandNode(t) - m1.Config.Gossip.Seeds = []string{seed} if err := port.GetListeners(func(lsns []*net.TCPListener) error { portsCfg := test.GenPortsConfig(test.NewPorts(lsns)) - m1.Config.Gossip.Port = portsCfg[0].Gossip.Port m1.Config.Etcd = portsCfg[0].Etcd m1.Config.Name = portsCfg[0].Name m1.Config.Cluster.Name = portsCfg[0].Cluster.Name @@ -648,7 +613,7 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { errc <- err }() return m1.Start() - }, 4, 10); err != nil { + }, 3, 10); err != nil { t.Fatalf("starting second main: %v", err) } @@ -664,74 +629,6 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { }) } -// Ensure that redundant gossip seeds are used -func TestCluster_GossipMembership(t *testing.T) { - t.Skip("skipping gossip test") - t.Run("Node0Down", func(t *testing.T) { - // Configure node0 - m0 := test.MustRunCluster(t, 1).GetNode(0) - defer m0.Close() - - seed := "" - - var eg errgroup.Group - - // Configure node1 - m1 := test.NewCommandNode(t) - defer m1.Close() - eg.Go(func() error { - // Pass invalid seed as first in list - m1.Config.Gossip.Seeds = []string{"http://localhost:8765", seed} - if err := port.GetPort(func(p int) error { - m1.Config.Gossip.Port = fmt.Sprintf("%d", p) - return m1.Start() - }, 10); err != nil { - t.Fatalf("starting second main: %v", err) - } - - return nil - }) - - // Configure node1 - m2 := test.NewCommandNode(t) - defer m2.Close() - eg.Go(func() error { - // Pass invalid seed as first in list - m2.Config.Gossip.Seeds = []string{seed, "http://localhost:8765"} - err := port.GetPort(func(p int) error { - m2.Config.Gossip.Port = fmt.Sprintf("%d", p) - return m2.Start() - }, 10) - - if err != nil { - t.Fatalf("starting second main: %v", err) - } - defer m2.Close() - return nil - }) - - if err := eg.Wait(); err != nil { - t.Fatal(err) - } - - state0, err0 := m0.API.State() - state1, err1 := m1.API.State() - state2, err2 := m2.API.State() - if err0 != nil || !test.CheckClusterState(m0, string(pilosa.ClusterStateNormal), 1000) { - t.Fatalf("unexpected node0 cluster state: %s, error: %v", state0, err0) - } else if err1 != nil || !test.CheckClusterState(m1, string(pilosa.ClusterStateNormal), 1000) { - t.Fatalf("unexpected node1 cluster state: %s, error: %v", state1, err1) - } else if err2 != nil || !test.CheckClusterState(m2, string(pilosa.ClusterStateNormal), 1000) { - t.Fatalf("unexpected node2 cluster state: %s, error: %v", state2, err2) - } - - numNodes := len(m0.API.Hosts(context.Background())) - if numNodes != 3 { - t.Fatalf("Expected 3 nodes, got %d", numNodes) - } - }) -} - func TestClusterResize_RemoveNode(t *testing.T) { cluster := test.MustRunCluster(t, 3) defer cluster.Close() diff --git a/server/handler_test.go b/server/handler_test.go index 460a4eff3..b0c6cf325 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -40,7 +40,6 @@ import ( pb "github.com/pilosa/pilosa/v2/proto" "github.com/pilosa/pilosa/v2/server" "github.com/pilosa/pilosa/v2/test" - "github.com/pilosa/pilosa/v2/test/port" ) func TestHandler_PostSchemaCluster(t *testing.T) { @@ -1405,10 +1404,7 @@ func TestCluster_TranslateStore(t *testing.T) { ), ) - if err := port.GetPort(func(p int) error { - cluster.GetIdleNode(0).Config.Gossip.Port = fmt.Sprintf("%d", p) - return cluster.GetIdleNode(0).Start() - }, 10); err != nil { + if err := cluster.GetIdleNode(0).Start(); err != nil { t.Fatalf("starting node 0: %v", err) } defer cluster.GetIdleNode(0).Close() diff --git a/server/server.go b/server/server.go index 00ab4d547..37b1b17c6 100644 --- a/server/server.go +++ b/server/server.go @@ -20,7 +20,6 @@ package server import ( - "bytes" "context" "crypto/tls" "io" @@ -47,7 +46,6 @@ import ( petcd "github.com/pilosa/pilosa/v2/etcd" "github.com/pilosa/pilosa/v2/gcnotify" "github.com/pilosa/pilosa/v2/gopsutil" - "github.com/pilosa/pilosa/v2/gossip" "github.com/pilosa/pilosa/v2/http" "github.com/pilosa/pilosa/v2/logger" pnet "github.com/pilosa/pilosa/v2/net" @@ -72,10 +70,6 @@ type Command struct { // Configuration. Config *Config - // Gossip transport - gossipTransport *gossip.Transport - gossipMemberSet io.Closer - // Standard input/output *pilosa.CmdIO @@ -84,7 +78,6 @@ type Command struct { // done will be closed when Command.Close() is called done chan struct{} - // Passed to the Gossip implementation. logOutput io.Writer logger loggerLogger @@ -233,11 +226,6 @@ func (m *Command) UpAndDown() (err error) { return errors.Wrap(err, "setting up server") } - // SetupNetworking (so we'll have profiling) - err = m.setupNetworking() - if err != nil { - return errors.Wrap(err, "setting up networking") - } go func() { err := m.Handler.Serve() if err != nil { @@ -469,35 +457,6 @@ func (m *Command) SetupServer() error { return errors.Wrap(err, "new handler") } -// setupNetworking sets up internode communication based on the configuration. -func (m *Command) setupNetworking() error { - gossipPort, err := strconv.Atoi(m.Config.Gossip.Port) - if err != nil { - return errors.Wrap(err, "parsing port") - } - - // get the host portion of addr to use for binding - gossipHost := m.listenURI.Host - m.gossipTransport, err = gossip.NewTransport(gossipHost, gossipPort, m.logger.Logger()) - if err != nil { - return errors.Wrap(err, "getting transport") - } - - gossipMemberSet, err := gossip.NewMemberSet( - m.Config.Gossip, - m.API, - gossip.WithLogOutput(&filteredWriter{logOutput: m.logOutput, v: m.Config.Verbose}), - gossip.WithPilosaLogger(m.logger), - gossip.WithTransport(m.gossipTransport), - ) - if err != nil { - return errors.Wrap(err, "getting memberset") - } - m.gossipMemberSet = gossipMemberSet - - return errors.Wrap(gossipMemberSet.Open(), "opening gossip memberset") -} - // setupLogger sets up the logger based on the configuration. func (m *Command) setupLogger() error { var f *logger.FileWriter @@ -539,13 +498,6 @@ func (m *Command) setupLogger() error { return nil } -// GossipTransport allows a caller to return the gossip transport created when -// setting up the GossipMemberSet. This is useful if one needs to determine the -// allocated ephemeral port programmatically. (usually used in tests) -func (m *Command) GossipTransport() *gossip.Transport { - return m.gossipTransport -} - // Close shuts down the server. func (m *Command) Close() error { select { @@ -558,9 +510,6 @@ func (m *Command) Close() error { eg.Go(m.Server.Close) eg.Go(m.API.Close) eg.Go(m.pgserver.Close) - if m.gossipMemberSet != nil { - eg.Go(m.gossipMemberSet.Close) - } if closer, ok := m.logOutput.(io.Closer); ok { // If closer is os.Stdout or os.Stderr, don't close it. if closer != os.Stdout && closer != os.Stderr { @@ -617,27 +566,6 @@ func getListener(uri pnet.URI, tlsconf *tls.Config) (ln net.Listener, err error) return ln, nil } -type filteredWriter struct { - v bool - logOutput io.Writer -} - -// Write forwards the write to logOutput if verbose is true, or it doesn't -// contain [DEBUG] or [INFO]. This implementation isn't technically correct -// since Write could be called with only part of a log line, but I don't think -// that actually happens, so until it becomes a problem, I don't think it's -// worth dealing with the extra complexity. (jaffee) -func (f *filteredWriter) Write(p []byte) (n int, err error) { - if bytes.Contains(p, []byte("[DEBUG]")) || bytes.Contains(p, []byte("[INFO]")) { - if f.v { - return f.logOutput.Write(p) - } - } else { - return f.logOutput.Write(p) - } - return len(p), nil -} - // ParseConfig parses s into a Config. func ParseConfig(s string) (Config, error) { var c Config diff --git a/server/server_test.go b/server/server_test.go index 64f92ecac..ea1b08e70 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -26,7 +26,6 @@ import ( "os" "reflect" "sort" - "strconv" "strings" "testing" "time" @@ -977,8 +976,6 @@ func TestClusterQueriesAfterRestart(t *testing.T) { config := cmd1.Command.Config config.Bind = cmd1.API.Node().URI.HostPort() - // this isn't necessary, but makes the test run way faster - config.Gossip.Port = strconv.Itoa(int(cmd1.Command.GossipTransport().URI.Port)) cmd1.Command = server.NewCommand(cmd1.Stdin, cmd1.Stdout, cmd1.Stderr, server.OptCommandServerOptions(pilosa.OptServerOpenTranslateStore(pilosa.OpenInMemTranslateStore))) cmd1.Command.Config = config err = cmd1.Start() diff --git a/test/cluster.go b/test/cluster.go index 74b2d0988..26e4a4985 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -401,22 +401,6 @@ func (c *Cluster) Start() error { }() portsCfg := GenPortsConfig(sliceOfPorts) - var gossipSeeds []string - for i, cc := range c.Nodes { - i := i - // get the bind uri to use as the host portion of the gossip seed. - uri, err := pilosa.AddressWithDefaults(cc.Config.Bind) - if err != nil { - return errors.Wrap(err, "processing bind address") - } - - cc.Config.Gossip.Port = portsCfg[i].Gossip.Port - gossipHost := uri.Host - gossipPort := cc.Config.Gossip.Port - - gossipSeeds = append(gossipSeeds, fmt.Sprintf("%s:%s", gossipHost, gossipPort)) - } - for i, cc := range c.Nodes { cc := cc cc.Config.Etcd = portsCfg[i].Etcd @@ -425,14 +409,12 @@ func (c *Cluster) Start() error { cc.Config.BindGRPC = portsCfg[i].BindGRPC eg.Go(func() error { - cc.Config.Gossip.Seeds = gossipSeeds - return cc.Start() }) } return eg.Wait() - }, 4*len(c.Nodes), 10) + }, 3*len(c.Nodes), 10) if err != nil { return err diff --git a/test/disco.go b/test/disco.go index 46328a24e..a903774c1 100644 --- a/test/disco.go +++ b/test/disco.go @@ -22,7 +22,6 @@ import ( "time" "github.com/pilosa/pilosa/v2/etcd" - "github.com/pilosa/pilosa/v2/gossip" "github.com/pilosa/pilosa/v2/server" ) @@ -33,8 +32,7 @@ type Ports struct { LsnP *net.TCPListener PortP int - Grpc int - Gossip int //TODO remove + Grpc int } func (ports *Ports) Close() error { @@ -65,10 +63,7 @@ func GenPortsConfig(ports []Ports) []*server.Config { } cfgs[i] = &server.Config{ - Name: name, - Gossip: gossip.Config{ - Port: fmt.Sprint(ports[i].Gossip), - }, + Name: name, BindGRPC: fmt.Sprintf(":%d", ports[i].Grpc), Etcd: etcd.Options{ Dir: discoDir, @@ -101,20 +96,18 @@ func NewPorts(lsn []*net.TCPListener) []Ports { ports[i] = lsn[i].Addr().(*net.TCPAddr).Port } - for i := 0; i < n; i = i + 4 { + for i := 0; i < n; i = i + 3 { out = append(out, Ports{ LsnC: lsn[i], PortC: ports[i], LsnP: lsn[i+1], PortP: ports[i+1], - Grpc: ports[i+2], - Gossip: ports[i+3], + Grpc: ports[i+2], }) - // make Grpc and Gossip ports available to + // make Grpc port available to // be rebound. lsn[i+2].Close() - lsn[i+3].Close() } return out diff --git a/translator_test.go b/translator_test.go index ddaff794b..518da91ff 100644 --- a/translator_test.go +++ b/translator_test.go @@ -263,17 +263,12 @@ func TestTranslation_Reset(t *testing.T) { if err := node0.SoftOpen(); err != nil { t.Fatal(err) } - gossipSeeds := []string{} - - node1.Config.Gossip.Seeds = gossipSeeds if err := node1.SoftOpen(); err != nil { t.Fatal(err) } - node2.Config.Gossip.Seeds = gossipSeeds if err := node2.SoftOpen(); err != nil { t.Fatal(err) } - node3.Config.Gossip.Seeds = gossipSeeds if err := node3.SoftOpen(); err != nil { t.Fatal(err) }