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

Merge pull request #18 from travisturner/disco-config-noder-remove-gossip

Browse source 
remove the rest of the gossip code (except config)
This commit is contained in:
Travis Turner 2021-02-04 21:35:22 -06:00 • committed by GitHub
commit e48fd9bf88
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 16 additions and 2724 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
api.go
View file

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

2
cluster.go
View file

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

36
cmd/pilosa-fsck/Makefile
View file

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

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

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

448
cmd/pilosa-fsck/fsck_test.go
View file

@ -1,448 +0,0 @@
// Copyright 2020 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package main
import (
"context"
"fmt"
"io/ioutil"
"reflect"
"strconv"
"testing"
"time"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/boltdb"
"github.com/pilosa/pilosa/v2/hash"
"github.com/pilosa/pilosa/v2/http"
"github.com/pilosa/pilosa/v2/server"
"github.com/pilosa/pilosa/v2/test"
)
func Test_Repair(t *testing.T) {
t.Skip("I don't quite understand what this test is doing and will need help adjusting it to pass again.")
// a) setup 1 primary + 3 replicas of disagree-ing cluster dirs.
nNodes := 4
nReplicas := 3
name := t.Name()
var nodeid []string
for i := 0; i < nNodes; i++ {
// work around a bug in the test.MustRunCluster that corrupts
// the .topology file if we only join name with one "_" underscore.
nodeid = append(nodeid, name+"__"+strconv.Itoa(i))
}
c := test.MustRunCluster(t, nNodes,
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID(nodeid[0]),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerReplicaN(nReplicas),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID(nodeid[1]),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerReplicaN(nReplicas),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID(nodeid[2]),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerReplicaN(nReplicas),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID(nodeid[3]),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerReplicaN(nReplicas),
)},
)
// note: do not defer c.Close() here. We manually close below.
var nodes []*test.Command
var dirs []string
for i := 0; i < nNodes; i++ {
nd := c.GetNode(i)
nodes = append(nodes, nd)
dirs = append(dirs, nd.Server.Holder().Path())
}
ctx := context.Background()
index := []string{"rick", "morty"}
fieldName := []string{"f", "flying_car"}
idx := make([]*pilosa.Index, len(index))
field := make([]*pilosa.Field, len(index))
var err error
for i := range index {
idx[i], err = nodes[0].API.CreateIndex(ctx, index[i], pilosa.IndexOptions{Keys: true, TrackExistence: true})
if err != nil {
t.Fatalf("creating index: %v", err)
}
if idx[i].CreatedAt() == 0 {
t.Fatal("index createdAt is empty")
}
field[i], err = nodes[0].API.CreateField(ctx, index[i], fieldName[i], pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100))
if err != nil {
t.Fatalf("creating field: %v", err)
}
if field[i].CreatedAt() == 0 {
t.Fatal("field createdAt is empty")
}
}
rowID := uint64(1)
timestamp := int64(0)
for i := range index {
// Generate some keyed records.
rowIDs := []uint64{}
timestamps := []int64{}
N := 10
for j := 1; j <= N; j++ {
rowIDs = append(rowIDs, rowID)
timestamps = append(timestamps, timestamp)
}
var colKeys []string
switch i {
case 0:
// Keys are sharded so ordering is not guaranteed.
colKeys = []string{"col10", "col8", "col9", "col6", "col7", "col4", "col5", "col2", "col3", "col1"}
colKeys = colKeys[:N]
case 1:
colKeys = []string{"col11", "col12"}
N = len(colKeys)
rowIDs = rowIDs[:N]
timestamps = timestamps[:N]
}
// Import data with keys to the coordinator (node0) and verify that it gets
// translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher)
req := &pilosa.ImportRequest{
Index: index[i],
IndexCreatedAt: idx[i].CreatedAt(),
Field: fieldName[i],
FieldCreatedAt: field[i].CreatedAt(),
// even though this says Shard: 0, that won't matter. The column keys
// get hashed and that decides the actual shard.
Shard: 0,
RowIDs: rowIDs,
ColumnKeys: colKeys,
Timestamps: timestamps,
}
qcx := nodes[0].API.Txf().NewQcx()
if err := nodes[0].API.Import(ctx, qcx, req); err != nil {
t.Fatal(err)
}
panicOn(qcx.Finish())
//qcx.Reset()
pql := fmt.Sprintf("Row(%s=%d)", fieldName[i], rowID)
// Query node0.
if res, err := nodes[0].API.Query(ctx, &pilosa.QueryRequest{Index: index[i], Query: pql}); err != nil {
t.Fatal(err)
} else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) {
t.Fatalf("expected colKeys='%#v'; observed column keys: %#v", colKeys, keys)
}
// Query node1.
if err := test.RetryUntil(5*time.Second, func() error {
if res, err := nodes[1].API.Query(ctx, &pilosa.QueryRequest{Index: index[i], Query: pql}); err != nil {
return err
} else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) {
return fmt.Errorf("unexpected column keys: %#v", keys)
}
return nil
}); err != nil {
t.Fatal(err)
}
}
// end of setup.
// partitionID in use: 6, 31, 57, 133, 185, 235
targetPartition := 31 // which partitionID we mess with.
targetNode := nodes[0] // this is the first replica.
targetIndex := index[0]
// 0 first replica
// 1 second replica
// 2 -- not a replica
// 3 primary
cfg := &FsckConfig{
Fix: false,
FixCol: false,
Quiet: true,
//Verbose: true,
ReplicaN: nReplicas,
Dirs: dirs,
ParallelReaders: 5,
}
panicOn(cfg.ValidateConfig())
// for this test, mess up a replica that is not the primary.
h := targetNode.API.Holder()
idx[0] = h.Index(index[0])
store := idx[0].TranslateStore(targetPartition)
fwd, rev := getFwdRev(store, targetPartition)
//vv("targetPartition=%v, store.PartitionID=%v, before corruption, fwd='%#v', rev='%#v'", targetPartition, store.PartitionID, fwd, rev)
// # fsck_test.go:288 2020-10-01T13:39:57.718995-05:00 partition 31, key 'col5' -> db00001
presz := len(rev)
delete(rev, fwd["col5"])
postsz := len(rev)
if postsz == presz {
panic("did not delete any key!")
}
bolt := store.(*boltdb.TranslateStore)
//vv("pre corruption, bolt = '%v'", fileChecksum(bolt.Path))
//bolt.DumpBolt("pre-corruption")
if err := bolt.SetFwdRevMaps(nil, fwd, rev); err != nil {
t.Fatal(err)
}
//vv("post corruption, bolt = '%v'", fileChecksum(bolt.Path))
//bolt.DumpBolt("post-corruption")
//fwd3, rev3 := getFwdRev(store, targetPartition)
//vv("after corruption, fwd='%#v', rev='%#v'", fwd3, rev3)
targetIndex1 := "morty"
targetPartition1 := 226 // for "col11"
// # fsck_test.go:248 2020-10-06T20:24:33.755576-05:00 on k=47, idx[1]: targetPartition=47, store.PartitionID=0x4abe160, before corruption, fwd1='map[string]uint64{"col12":0xcf00001}', rev1='map[uint64]string{0xcf00001:"col12"}'
//# fsck_test.go:248 2020-10-06T20:24:35.608568-05:00 on k=226, idx[1]: targetPartition=226, store.PartitionID=0x4abe160, before corruption, fwd1='map[string]uint64{"col11":0xcc00001}', rev1='map[uint64]string{0xcc00001:"col11"}'
idx[1] = h.Index(index[1])
store1 := idx[1].TranslateStore(targetPartition1)
fwd1, rev1 := getFwdRev(store1, targetPartition1)
//vv("on k=%v, idx[1]: targetPartition=%v, store.PartitionID=%v, before corruption, fwd1='%#v', rev1='%#v'", k, targetPartition1, store.PartitionID, fwd1, rev1)
presz1 := len(rev1)
delete(rev1, fwd1["col11"])
postsz1 := len(rev1)
if postsz1 == presz1 {
panic("did not delete any key!")
}
bolt1 := store1.(*boltdb.TranslateStore)
if err := bolt1.SetFwdRevMaps(nil, fwd1, rev1); err != nil {
t.Fatal(err)
}
// done corrupting.
for _, nd := range nodes {
nd.Command.Close()
}
//panicOn(bolt.Open())
//bolt.DumpBolt("post-corruption, after Close. bolt:")
//bolt.Close()
//chksums := getChecksums(dirs, cfg, targetPartition)
//vv("post corruption, pre repair chksums = '%#v'", chksums)
// first we check that the corruption can be detected
// by our test with the checksums.
chk, err := check(dirs, cfg, targetIndex, targetPartition)
_ = chk
//vv("pre-fix, chk='%v'; err='%v'", chk, err)
if err == nil {
panic("expected to see checksums not match! but no corruption detected.")
}
chk1, err := check(dirs, cfg, targetIndex1, targetPartition1)
_ = chk1
//vv("pre-fix, chk1='%v'; err='%v'", chk1, err)
if err == nil {
panic("expected to see checksums not match! but no corruption detected.")
}
// b) running in reporting mode only should report that a fix is needed.
fixNeeded, err := cfg.Run()
panicOn(err)
if !fixNeeded {
panic("fix should be needed now, before repair")
}
// c) run the fix.
cfg.Fix = true
cfg.FixCol = true
fixNeeded, err = cfg.Run()
panicOn(err)
if !fixNeeded {
panic("fix should be marked needed if repair was made")
}
// d) check that the replicas all look like the primary.
//chksums = getChecksums(dirs, cfg, targetPartition)
//vv("after repair chksums = '%#v'", chksums)
chk, err = check(dirs, cfg, targetIndex, targetPartition)
_ = chk
//vv("chk = '%v' after repair; err='%v'", chk, err)
panicOn(err)
chk1, err = check(dirs, cfg, targetIndex1, targetPartition1)
_ = chk1
//vv("chk = '%v' after repair; err='%v'", chk, err)
panicOn(err)
// e) run again, should see no fix needed.
fixNeeded, err = cfg.Run()
panicOn(err)
if fixNeeded {
panic("should see no fix needed after the prior repair")
}
}
func getFwdRev(store pilosa.TranslateStore, partitionID int) (fwd map[string]uint64, rev map[uint64]string) {
fwd = make(map[string]uint64)
rev = make(map[uint64]string)
_ = store.KeyWalker(func(key string, col uint64) {
//vv("partition %v, key '%v' -> %x", partitionID, key, col)
fwd[key] = col
})
_ = store.IDWalker(func(key string, col uint64) {
//vv("partition %v, id %x -> '%v'", partitionID, col, key)
rev[col] = key
})
return
}
func check(dirs []string, cfg *FsckConfig, targetIndex string, targetPartition int) (chksum string, err error) {
//vv("top of check, dirs = '%#v', targetIndex='%v', targetPartition='%v'", dirs, targetIndex, targetPartition)
//defer vv("returning from check()")
firstChecksum := ""
firstDir := ""
firstStorePath := ""
quiet := cfg.Quiet
defer func() {
cfg.Quiet = quiet
}()
cfg.Quiet = true
for i := range dirs {
dir := dirs[i]
_, _, ats, err := cfg.readOneDir(dir)
panicOn(err)
indexes := indexesFromAts(ats)
//vv("indexes = '%#v'", indexes)
for _, index := range indexes {
if index != targetIndex {
continue
}
for _, s := range ats.Sums {
//vv(" s= '%#v'", s)
if s.Index != index {
//vv("skipping s.Index '%v' != index '%v'", s.Index, index)
continue
}
if s.PartitionID != targetPartition {
continue
}
//vv("accepting s.PartitionID(%v) == targetPartition(%v); s.Index '%v'; "+
//"index '%v'; s.IsPrimary=%v, s.IsReplica=%v, s='%#v'; s.Checksum='%v', firstChecksum='%v'",
//s.PartitionID, targetPartition, s.Index, index,
//s.IsPrimary, s.IsReplica, s, s.Checksum, firstChecksum)
if s.IsPrimary || s.IsReplica {
chksum := s.Checksum
if firstChecksum == "" {
firstChecksum = chksum
firstDir = dir
firstStorePath = s.StorePath
} else {
//vv("targetIndex = '%v'; firstChecksum='%v', chksum='%v'", targetIndex, firstChecksum, chksum)
if chksum != firstChecksum {
return chksum, fmt.Errorf("bolt chksum on node %v '%v' disagrees with '%v' on '%v'; index='%v'; s.StorePath = '%v'; firstStorePath='%v'", dir, chksum, firstChecksum, firstDir, index, s.StorePath, firstStorePath)
}
}
}
}
}
}
return firstChecksum, nil
}
// These are here to satisfy the linter in CI while the test is being skipped.
var _ = getFwdRev
var _ = check
var _ = getChecksums
func getChecksums(dirs []string, cfg *FsckConfig, targetPartition int) (chksum []string) {
for i := range dirs {
dir := dirs[i]
_, _, ats, err := cfg.readOneDir(dir)
panicOn(err)
for _, s := range ats.Sums {
if s.PartitionID != targetPartition {
continue
}
chksum = append(chksum, s.Checksum)
}
}
return
}
/* on shardwidth 20
# fsck_test.go:211 2020-09-30T17:19:05.823278-05:00 partition 6, key 'col2' -> dc00001
# fsck_test.go:214 2020-09-30T17:19:05.823309-05:00 partition 6, id dc00001 -> 'col2'
# fsck_test.go:211 2020-09-30T17:19:05.823430-05:00 partition 31, key 'col5' -> db00001
# fsck_test.go:214 2020-09-30T17:19:05.823447-05:00 partition 31, id db00001 -> 'col5'
# fsck_test.go:211 2020-09-30T17:19:05.823970-05:00 partition 57, key 'col10' -> 5d00001
# fsck_test.go:214 2020-09-30T17:19:05.823998-05:00 partition 57, id 5d00001 -> 'col10'
# fsck_test.go:211 2020-09-30T17:19:05.827007-05:00 partition 133, key 'col7' -> d900001
# fsck_test.go:214 2020-09-30T17:19:05.827071-05:00 partition 133, id d900001 -> 'col7'
# fsck_test.go:211 2020-09-30T17:19:05.827549-05:00 partition 185, key 'col3' -> dd00001
# fsck_test.go:214 2020-09-30T17:19:05.827573-05:00 partition 185, id dd00001 -> 'col3'
# fsck_test.go:211 2020-09-30T17:19:05.827792-05:00 partition 235, key 'col9' -> d700001
# fsck_test.go:214 2020-09-30T17:19:05.827809-05:00 partition 235, id d700001 -> 'col9'
*/
var _ = fileChecksum
func fileChecksum(path string) string {
by, err := ioutil.ReadFile(path)
panicOn(err)
return hash.Blake3sum16(by)
}

1
cmd/pilosa-fsck/release-pilosa-fsck/.gitignore vendored
View file

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

252
cmd/pilosa-fsck/release-pilosa-fsck/DESIGN.md
View file

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

BIN
cmd/pilosa-fsck/release-pilosa-fsck/backups.tar.gz
View file

Binary file not shown.

21
cmd/pilosa-fsck/release-pilosa-fsck/example.sh
View file

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

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

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

6
cmd/server_test.go
View file

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

565
gossip/gossip.go
View file

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

119
server/cluster_test.go
View file

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

6
server/handler_test.go
View file

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

72
server/server.go
View file

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

3
server/server_test.go
View file

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

20
test/cluster.go
View file

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

17
test/disco.go
View file

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

5
translator_test.go
View file

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