scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 17:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1582 from travisturner/coordinator-as-primary

Browse source 
Treat coordinator as primary translate store.
This commit is contained in:
Travis Turner 2018-08-13 13:15:39 -05:00 • committed by GitHub
commit f278af6194
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 562 additions and 44 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
api.go
View file

@ -136,9 +136,9 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
}
// Translate column attributes, if necessary.
if api.server.translateFile != nil {
if api.holder.translateFile != nil {
for _, col := range resp.ColumnAttrSets {
v, err := api.server.translateFile.TranslateColumnToString(req.Index, col.ID)
v, err := api.holder.translateFile.TranslateColumnToString(req.Index, col.ID)
if err != nil {
return resp, err
}
@ -764,7 +764,7 @@ func (api *API) ResizeAbort() error {
// GetTranslateData provides a reader for key translation logs starting at offset.
func (api *API) GetTranslateData(ctx context.Context, offset int64) (io.ReadCloser, error) {
rc, err := api.server.translateFile.Reader(ctx, offset)
rc, err := api.holder.translateFile.Reader(ctx, offset)
if err != nil {
return nil, errors.Wrap(err, "read from translate store")
}

27
cluster.go
View file

@ -169,7 +169,7 @@ type nodeAction struct {
type cluster struct { // nolint: maligned
id string
Node *Node
Nodes []*Node // TODO phase this out?
Nodes []*Node
// Hashing algorithm used to assign partitions to nodes.
Hasher Hasher
@ -935,7 +935,6 @@ func (c *cluster) allNodesReady() (ret bool) {
}
func (c *cluster) handleNodeAction(nodeAction nodeAction) error {
c.mu.Lock()
j, err := c.unprotectedGenerateResizeJob(nodeAction)
c.mu.Unlock()
@ -959,7 +958,7 @@ func (c *cluster) handleNodeAction(nodeAction nodeAction) error {
c.logger.Printf("wait for jobResult")
jobResult := <-j.result
// Make sure j.Run() didn't return an error.
// Make sure j.run() didn't return an error.
if eg.Wait() != nil {
return errors.Wrap(err, "running job")
}
@ -1779,6 +1778,10 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
}
}
// If the cluster membership has changed, reset the primary for
// translate store replication.
c.holder.setPrimaryTranslateStore(c.unprotectedPreviousNode())
c.unprotectedSetState(cs.State)
c.markAsJoined()
@ -1786,6 +1789,24 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
return nil
}
// unprotectedPreviousNode returns the node listed before the current node in c.Nodes.
// If there is only one node in the cluster, returns nil.
// If the current node is the first node in the list, returns the last node.
func (c *cluster) unprotectedPreviousNode() *Node {
if len(c.Nodes) <= 1 {
return nil
}
pos := c.nodePositionByID(c.Node.ID)
if pos == -1 {
return nil
} else if pos == 0 {
return c.Nodes[len(c.Nodes)-1]
} else {
return c.Nodes[pos-1]
}
}
// setStatic is unprotected, but only called before the cluster has been started
// (and therefore not concurrently).
func (c *cluster) setStatic(hosts []string) error {

54
cluster_internal_test.go
View file

@ -449,6 +449,60 @@ func TestCluster_Nodes(t *testing.T) {
})
}
func TestCluster_PreviousNode(t *testing.T) {
node0 := &Node{ID: "node0"}
node1 := &Node{ID: "node1"}
node2 := &Node{ID: "node2"}
t.Run("OneNode", func(t *testing.T) {
c := newCluster()
c.addNodeBasicSorted(node0)
c.Node = node0
if prev := c.unprotectedPreviousNode(); prev != nil {
t.Errorf("expected: nil, but got: %v", prev)
}
})
t.Run("TwoNode", func(t *testing.T) {
c := newCluster()
c.addNodeBasicSorted(node0)
c.addNodeBasicSorted(node1)
c.Node = node0
if prev := c.unprotectedPreviousNode(); prev != node1 {
t.Errorf("expected: node1, but got: %v", prev)
}
c.Node = node1
if prev := c.unprotectedPreviousNode(); prev != node0 {
t.Errorf("expected: node0, but got: %v", prev)
}
})
t.Run("ThreeNode", func(t *testing.T) {
c := newCluster()
c.addNodeBasicSorted(node0)
c.addNodeBasicSorted(node1)
c.addNodeBasicSorted(node2)
c.Node = node0
if prev := c.unprotectedPreviousNode(); prev != node2 {
t.Errorf("expected: node2, but got: %v", prev)
}
c.Node = node1
if prev := c.unprotectedPreviousNode(); prev != node0 {
t.Errorf("expected: node0, but got: %v", prev)
}
c.Node = node2
if prev := c.unprotectedPreviousNode(); prev != node1 {
t.Errorf("expected: node1, but got: %v", prev)
}
})
}
// NEXT: move this test to internal and unexport IsCoordinator
func TestCluster_Coordinator(t *testing.T) {
uri1 := NewTestURIFromHostPort("node1", 0)

2
ctl/server.go
View file

@ -44,7 +44,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.LongQueryTime), "cluster.long-query-time", "", time.Minute, "Duration that will trigger log and stat messages for slow queries.")
// Translation
flags.StringVarP(&srv.Config.Translation.PrimaryURL, "translation.primary-url", "", srv.Config.Translation.PrimaryURL, "URL for primary translation node for replication.")
flags.StringVarP(&srv.Config.Translation.PrimaryURL, "translation.primary-url", "", srv.Config.Translation.PrimaryURL, "DEPRECATED: URL for primary translation node for replication.")
// Gossip
flags.StringVarP(&srv.Config.Gossip.Port, "gossip.port", "", srv.Config.Gossip.Port, "Port to which pilosa should bind for internal state sharing.")

22
holder.go
View file

@ -46,6 +46,10 @@ type Holder struct {
// Indexes by name.
indexes map[string]*Index
// Key/ID translation
translateFile *TranslateFile
NewPrimaryTranslateStore func(interface{}) TranslateStore
// opened channel is closed once Open() completes.
opened chan struct{}
@ -77,6 +81,9 @@ func NewHolder() *Holder {
opened: make(chan struct{}),
translateFile: NewTranslateFile(),
NewPrimaryTranslateStore: newNopTranslateStore,
broadcaster: NopBroadcaster,
Stats: NopStatsClient,
@ -160,6 +167,13 @@ func (h *Holder) Close() error {
return errors.Wrap(err, "closing index")
}
}
if h.translateFile != nil {
if err := h.translateFile.Close(); err != nil {
return err
}
}
return nil
}
@ -560,6 +574,14 @@ func (h *Holder) logStartup() error {
return nil
}
func (h *Holder) setPrimaryTranslateStore(node *Node) {
var nodeID string
if node != nil {
nodeID = node.ID
}
h.translateFile.SetPrimaryStore(nodeID, h.NewPrimaryTranslateStore(node))
}
// holderSyncer is an active anti-entropy tool that compares the local holder
// with a remote holder based on block checksums and resolves differences.
type holderSyncer struct {

31
http/translator.go
View file

@ -6,11 +6,13 @@ import (
"fmt"
"io"
"io/ioutil"
"log"
"net/http"
"net/url"
"strconv"
"github.com/pilosa/pilosa"
"github.com/pkg/errors"
)
// Ensure implementation implements inteface.
@ -19,12 +21,31 @@ var _ pilosa.TranslateStore = (*translateStore)(nil)
// translateStore represents an implementation of pilosa.TranslateStore that
// communicates over HTTP. This is used with the TranslateHandler.
type translateStore struct {
URL string
node *pilosa.Node
}
// NewTranslateStore returns a new instance of TranslateStore.
func NewTranslateStore(rawurl string) *translateStore {
return &translateStore{URL: rawurl}
// NewTranslateStore returns a new instance of TranslateStore based on node.
// DEPRECATED: Providing a string url to this function is being deprecated. Instead,
// provide a *pilosa.Node.
func NewTranslateStore(node interface{}) pilosa.TranslateStore {
var n *pilosa.Node
switch v := node.(type) {
case string:
log.Printf("WARNING: providing a string url to NewTranslateStore() has been deprecated.")
if uri, err := pilosa.NewURIFromAddress(v); err != nil {
log.Println(errors.Wrap(err, "creating uri"))
} else {
n = &pilosa.Node{
ID: v,
URI: *uri,
}
}
case *pilosa.Node:
n = v
default:
log.Printf("WARNING: a *pilosa.Node is the only type supported by NewTranslateStore().")
}
return &translateStore{node: n}
}
// TranslateColumnsToUint64 is not currently implemented.
@ -50,7 +71,7 @@ func (s *translateStore) TranslateRowToString(index, frame string, values uint64
// Reader returns a reader that can stream data from a remote store.
func (s *translateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {
// Generate remote URL.
u, err := url.Parse(s.URL)
u, err := url.Parse(s.node.URI.String())
if err != nil {
return nil, err
}

31
server.go
View file

@ -49,7 +49,6 @@ type Server struct { // nolint: maligned
// Internal
holder *Holder
cluster *cluster
translateFile *TranslateFile
diagnostics *diagnosticsCollector
executor *executor
hosts []string
@ -70,8 +69,6 @@ type Server struct { // nolint: maligned
isCoordinator bool
syncer holderSyncer
primaryTranslateStore TranslateStore
defaultClient InternalClient
dataDir string
}
@ -163,9 +160,17 @@ func OptServerInternalClient(c InternalClient) ServerOption {
}
}
// DEPRECATED
func OptServerPrimaryTranslateStore(store TranslateStore) ServerOption {
return func(s *Server) error {
s.primaryTranslateStore = store
s.logger.Printf("DEPRECATED: OptServerPrimaryTranslateStore")
return nil
}
}
func OptServerPrimaryTranslateStoreFunc(tf func(interface{}) TranslateStore) ServerOption {
return func(s *Server) error {
s.holder.NewPrimaryTranslateStore = tf
return nil
}
}
@ -261,6 +266,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
}
s.holder.Path = path
s.holder.translateFile.Path = filepath.Join(path, ".keys")
s.holder.Logger = s.logger
s.holder.Stats.SetLogger(s.logger)
@ -268,11 +274,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.cluster.logger = s.logger
s.cluster.holder = s.holder
// Initialize translation database.
s.translateFile = NewTranslateFile()
s.translateFile.Path = filepath.Join(path, ".keys")
s.translateFile.PrimaryTranslateStore = s.primaryTranslateStore
// Get or create NodeID.
s.nodeID = s.loadNodeID()
if s.isCoordinator {
@ -299,7 +300,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.executor.Holder = s.holder
s.executor.Node = node
s.executor.Cluster = s.cluster
s.executor.TranslateStore = s.translateFile
s.executor.TranslateStore = s.holder.translateFile
s.executor.MaxWritesPerRequest = s.maxWritesPerRequest
s.cluster.broadcaster = s
s.cluster.maxWritesPerRequest = s.maxWritesPerRequest
@ -324,7 +325,7 @@ func (s *Server) Open() error {
}
// Initialize id-key storage.
if err := s.translateFile.Open(); err != nil {
if err := s.holder.translateFile.Open(); err != nil {
return err
}
@ -370,7 +371,6 @@ func (s *Server) Close() error {
s.wg.Wait()
var errh error
var errt error
var errc error
if s.cluster != nil {
errc = s.cluster.close()
@ -378,17 +378,12 @@ func (s *Server) Close() error {
if s.holder != nil {
errh = s.holder.Close()
}
if s.translateFile != nil {
errt = s.translateFile.Close()
}
// prefer to return holder error over translateFile error over cluster
// prefer to return holder error over cluster
// error. This order is somewhat arbitrary. It would be better if we had
// some way to combine all the errors, but probably not important enough to
// warrant the extra complexity.
if errh != nil {
return errors.Wrap(errh, "closing holder")
} else if errt != nil {
return errors.Wrap(errt, "closing translateFile")
}
return errors.Wrap(errc, "closing cluster")
}

2
server/config.go
View file

@ -71,7 +71,7 @@ type Config struct {
// Gossip config is based around memberlist.Config.
Gossip gossip.Config `toml:"gossip"`
// Translation config supports translation store replication.
// DEPRECATED: Translation config supports translation store replication.
Translation struct {
PrimaryURL string `toml:"primary-url"`
} `toml:"translation"`

7
server/server.go
View file

@ -250,10 +250,9 @@ func (m *Command) SetupServer() error {
c := http.GetHTTPClient(TLSConfig)
// Setup connection to primary store if this is a replica.
var primaryTranslateStore pilosa.TranslateStore
// Primary store configuration is handled automatically now.
if m.Config.Translation.PrimaryURL != "" {
primaryTranslateStore = http.NewTranslateStore(m.Config.Translation.PrimaryURL)
m.logger.Printf("DEPRECATED: The primary-url configuration option is no longer used.")
}
// Set Coordinator.
@ -278,7 +277,7 @@ func (m *Command) SetupServer() error {
pilosa.OptServerStatsClient(statsClient),
pilosa.OptServerURI(uri),
pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)),
pilosa.OptServerPrimaryTranslateStore(primaryTranslateStore),
pilosa.OptServerPrimaryTranslateStoreFunc(http.NewTranslateStore),
pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts),
pilosa.OptServerSerializer(proto.Serializer{}),
coordinatorOpt,

131
translate.go
View file

@ -8,6 +8,7 @@ import (
"errors"
"fmt"
"io"
"io/ioutil"
"log"
"os"
"path/filepath"
@ -71,6 +72,10 @@ type TranslateFile struct {
// If non-nil, data is streamed from a primary and this is a read-only store.
PrimaryTranslateStore TranslateStore
primaryID string // unique ID used to identify the primary store
replicationClosing chan struct{}
primaryStoreEvents chan primaryStoreEvent
repWG sync.WaitGroup
// Delay after attempting to connect to a primary that the store will retry.
replicationRetryInterval time.Duration
@ -86,6 +91,9 @@ func NewTranslateFile() *TranslateFile {
mapSize: defaultMapSize,
replicationClosing: make(chan struct{}),
primaryStoreEvents: make(chan primaryStoreEvent),
replicationRetryInterval: defaultReplicationRetryInterval,
}
}
@ -109,10 +117,65 @@ func (s *TranslateFile) Open() (err error) {
return err
}
// Stream from primary, if available.
// Listen to primaryStoreEvents channel.
s.wg.Add(1)
go func() { defer s.wg.Done(); s.monitorPrimaryStoreEvents() }()
return nil
}
// primaryStoreEvent is used to set/change the primary translate store.
// It contains a TranslateStore along with an associated string ID which
// is used to determine whether the primary needs to be changed from the
// current value.
type primaryStoreEvent struct {
id string
ts TranslateStore
}
// SetPrimaryStore sets the translate files's primary translate store.
// The id value is used to determine whether the primary needs to be changed
// from the current value (i.e. calling this multiple times with the same
// input values will no-op on all subsequent calls).
func (s *TranslateFile) SetPrimaryStore(id string, ts TranslateStore) {
go func() {
s.primaryStoreEvents <- primaryStoreEvent{
id: id,
ts: ts,
}
}()
}
// handlePrimaryStoreEvent changes the PrimaryTranslateStore
// used for replication by TranslateFile.
func (s *TranslateFile) handlePrimaryStoreEvent(ev primaryStoreEvent) error {
s.mu.Lock()
defer s.mu.Unlock()
if ev.id == s.primaryID {
return nil
}
// Stop translate store replication.
log.Printf("stop monitor replication")
close(s.replicationClosing)
s.repWG.Wait()
// Set the primary node for translate store replication.
log.Printf("set primary translate store to %s", ev.id)
s.primaryID = ev.id
if ev.id == "" {
s.PrimaryTranslateStore = nil
} else {
s.PrimaryTranslateStore = ev.ts
}
// Start translate store replication. Stream from primary, if available.
log.Printf("start monitor replication")
if s.PrimaryTranslateStore != nil {
s.wg.Add(1)
go func() { defer s.wg.Done(); s.monitorReplication() }()
s.replicationClosing = make(chan struct{})
s.repWG.Add(1)
go func() { defer s.repWG.Done(); s.monitorReplication() }()
}
return nil
@ -259,7 +322,13 @@ func (s *TranslateFile) replayEntries() error {
func (s *TranslateFile) monitorReplication() {
// Create context that will cancel on close.
ctx, cancel := context.WithCancel(context.Background())
go func() { <-s.closing; cancel() }()
go func() {
select {
case <-s.closing:
case <-s.replicationClosing:
}
cancel()
}()
// Keep attempting to replicate until the store closes.
for {
@ -270,12 +339,31 @@ func (s *TranslateFile) monitorReplication() {
select {
case <-s.closing:
return
case <-s.replicationClosing:
return
case <-time.After(s.replicationRetryInterval):
log.Printf("pilosa: reconnecting to primary replica")
}
}
}
// monitorPrimaryStoreEvents is executed in a separate goroutine and listens for changes
// to the primary store assignment.
func (s *TranslateFile) monitorPrimaryStoreEvents() {
log.Printf("monitor primary store events")
// Keep handling events until the store closes.
for {
select {
case <-s.closing:
return
case ev := <-s.primaryStoreEvents:
if err := s.handlePrimaryStoreEvent(ev); err != nil {
log.Printf("handle primary store event")
}
}
}
}
func (s *TranslateFile) replicate(ctx context.Context) error {
off := s.size()
@ -989,3 +1077,38 @@ func uVarintSize(x uint64) (i int) {
}
return i + 1
}
// nopTStore represents a TranslateStore that doesn't do anything.
var nopTStore TranslateStore = nopTranslateStore{}
// newNopTranslateStore returns a translate store which does nothing. It returns a global
// object to avoid unnecessary allocations.
func newNopTranslateStore(interface{}) TranslateStore { return nopTStore }
// nopTranslateStore represents a no-op implementation of the TranslateStore interface.
type nopTranslateStore struct{}
// TranslateColumnsToUint64 is a no-op implementation of the TranslateStore TranslateColumnsToUint64 method.
func (s nopTranslateStore) TranslateColumnsToUint64(index string, values []string) ([]uint64, error) {
return []uint64{}, nil
}
// TranslateColumnToString is a no-op implementation of the TranslateStore TranslateColumnToString method.
func (s nopTranslateStore) TranslateColumnToString(index string, values uint64) (string, error) {
return "", nil
}
// TranslateRowsToUint64 is a no-op implementation of the TranslateStore TranslateRowsToUint64 method.
func (s nopTranslateStore) TranslateRowsToUint64(index, field string, values []string) ([]uint64, error) {
return []uint64{}, nil
}
// TranslateRowToString is a no-op implementation of the TranslateStore TranslateRowToString method.
func (s nopTranslateStore) TranslateRowToString(index, field string, values uint64) (string, error) {
return "", nil
}
// Reader is a no-op implementation of the TranslateStore Reader method.
func (s nopTranslateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {
return ioutil.NopCloser(bytes.NewReader(nil)), nil
}

293
translate_test.go
View file

@ -383,7 +383,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
// Create a replica that accepts writes from primary.
replica := NewTranslateFile()
replica.PrimaryTranslateStore = primary
replica.SetPrimaryStore("primary", primary)
if err := replica.Open(); err != nil {
t.Fatal(err)
}
@ -398,7 +398,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
// Attempt to read replica until writes appear.
if err := retryFor(2*time.Second, func() error {
// Verify that replica have received writes.
// Verify that replica has received writes.
if value, err := replica.TranslateColumnToString("IDX0", 1); err != nil {
return err
} else if value != "foo" {
@ -429,7 +429,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
t.Fatal(err)
}
// Attempt to read replica until write appear.
// Attempt to read replica until writes appear.
if err := retryFor(2*time.Second, func() error {
if value, err := replica.TranslateColumnToString("IDX0", 2); err != nil {
return err
@ -448,7 +448,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
t.Fatal(err)
}
// Attempt to read replica until write appear.
// Attempt to read replica until writes appear.
if err := retryFor(2*time.Second, func() error {
if value, err := replica.TranslateColumnToString("IDX0", 3); err != nil {
return err
@ -461,6 +461,289 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
}
}
func TestTranslateFile_ReassignPrimaryTranslateStore(t *testing.T) {
t.Run("AddNode", func(t *testing.T) {
// Create a primary store that accepts writes.
primary := MustOpenTranslateFile()
defer primary.MustClose()
// Create replica1 that accepts writes from primary.
replica1 := NewTranslateFile()
replica1.SetPrimaryStore("primary", primary)
if err := replica1.Open(); err != nil {
t.Fatal(err)
}
defer replica1.MustClose()
// Write to the primary.
if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if _, err := primary.TranslateRowsToUint64("IDX0", "FIELD0", []string{"bar", "baz"}); err != nil {
t.Fatal(err)
}
// Attempt to read replica1 until writes appear.
if err := retryFor(2*time.Second, func() error {
// Verify that replica1 has received writes.
if value, err := replica1.TranslateColumnToString("IDX0", 1); err != nil {
return err
} else if value != "foo" {
return fmt.Errorf("unexpected column 1 value: %s", value)
}
if value, err := replica1.TranslateRowToString("IDX0", "FIELD0", 1); err != nil {
return err
} else if value != "bar" {
return fmt.Errorf("unexpected row 1 value: %s", value)
}
if value, err := replica1.TranslateRowToString("IDX0", "FIELD0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected row 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
// Create replica2 that accepts writes from primary,
// and change replica1's primary to be replica2.
// From: P <- R1
// To: P <- R2 <- R1
// Momentarily, replica1 should be ahead of replica2, so we will see log
// messages like "translate store reader past file size: sz=0 off=39"
// But eventually it should get in sync and new writes will be available
// on replica1.
replica2 := NewTranslateFile()
replica2.SetPrimaryStore("primary", primary)
replica1.SetPrimaryStore("replica2", replica2)
if err := replica2.Open(); err != nil {
t.Fatal(err)
}
defer replica2.MustClose()
// Attempt to read replica2 until writes appear.
if err := retryFor(2*time.Second, func() error {
// Verify that replica2 have received writes.
if value, err := replica2.TranslateColumnToString("IDX0", 1); err != nil {
return err
} else if value != "foo" {
return fmt.Errorf("unexpected column 1 value: %s", value)
}
if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 1); err != nil {
return err
} else if value != "bar" {
return fmt.Errorf("unexpected row 1 value: %s", value)
}
if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected row 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
// Add more data to the primary and ensure that replica1 receives the
// data (via replica2).
if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"baz"}); err != nil {
t.Fatal(err)
}
// Attempt to read replica1 until writes appear.
if err := retryFor(2*time.Second, func() error {
if value, err := replica1.TranslateColumnToString("IDX0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected column 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
})
t.Run("RemoveNode", func(t *testing.T) {
// Create a primary store that accepts writes.
primary := MustOpenTranslateFile()
defer primary.MustClose()
// Create two replicas that accepts writes from the primary
// in a daisy-chain configuration.
// P <- R1 <- R2
// Create replica1.
replica1 := NewTranslateFile()
replica1.SetPrimaryStore("primary", primary)
if err := replica1.Open(); err != nil {
t.Fatal(err)
}
defer replica1.MustClose()
// Create replica2.
replica2 := NewTranslateFile()
replica2.SetPrimaryStore("replica1", replica1)
if err := replica2.Open(); err != nil {
t.Fatal(err)
}
defer replica2.MustClose()
// Write to the primary.
if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if _, err := primary.TranslateRowsToUint64("IDX0", "FIELD0", []string{"bar", "baz"}); err != nil {
t.Fatal(err)
}
// Attempt to read replica2 until writes appear.
if err := retryFor(2*time.Second, func() error {
// Verify that replica2 has received writes.
if value, err := replica2.TranslateColumnToString("IDX0", 1); err != nil {
return err
} else if value != "foo" {
return fmt.Errorf("unexpected column 1 value: %s", value)
}
if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 1); err != nil {
return err
} else if value != "bar" {
return fmt.Errorf("unexpected row 1 value: %s", value)
}
if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected row 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
// Remove replica1 from the replication chain.
// From: P <- R1 <- R2
// To: P <- R2
replica1.SetPrimaryStore("", nil)
// Add more data to the primary and ensure that replica2 receives the
// data (after replica1 is removed).
if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"baz"}); err != nil {
t.Fatal(err)
}
// Set replica2 to replicate from primary.
replica2.SetPrimaryStore("primary", primary)
// Attempt to read replica2 until writes appear.
if err := retryFor(2*time.Second, func() error {
if value, err := replica2.TranslateColumnToString("IDX0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected column 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
})
t.Run("ChangePrimaryWriter", func(t *testing.T) {
// Create a primary store that accepts writes.
primary := MustOpenTranslateFile()
defer primary.MustClose()
// Create two replicas that accepts writes from the primary
// in a daisy-chain configuration.
// P <- R1 <- R2
// Create replica1.
replica1 := NewTranslateFile()
replica1.SetPrimaryStore("primary", primary)
if err := replica1.Open(); err != nil {
t.Fatal(err)
}
defer replica1.MustClose()
// Create replica2.
replica2 := NewTranslateFile()
replica2.SetPrimaryStore("replica1", replica1)
if err := replica2.Open(); err != nil {
t.Fatal(err)
}
defer replica2.MustClose()
// Write to the primary.
if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if _, err := primary.TranslateRowsToUint64("IDX0", "FIELD0", []string{"bar", "baz"}); err != nil {
t.Fatal(err)
}
// Attempt to read replica2 until writes appear.
if err := retryFor(2*time.Second, func() error {
// Verify that replica2 has received writes.
if value, err := replica2.TranslateColumnToString("IDX0", 1); err != nil {
return err
} else if value != "foo" {
return fmt.Errorf("unexpected column 1 value: %s", value)
}
if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 1); err != nil {
return err
} else if value != "bar" {
return fmt.Errorf("unexpected row 1 value: %s", value)
}
if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected row 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
// Change replica1 to be the primary writer.
// From: P <- R1 <- R2
// To: R1 <- R2 <- P
replica1.SetPrimaryStore("", nil)
// SetPrimaryStore is asynchronous, so we need to wait before writing new data.
time.Sleep(200 * time.Millisecond)
// Add more data to the primary (now replica1) and ensure that primary (now a read-only replica)
// receives the data.
if _, err := replica1.TranslateColumnsToUint64("IDX0", []string{"baz"}); err != nil {
t.Fatal(err)
}
// Set primary to replicate from replica2.
primary.SetPrimaryStore("replica2", replica2)
// Attempt to read primary until writes appear.
if err := retryFor(2*time.Second, func() error {
if value, err := primary.TranslateColumnToString("IDX0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected column 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
})
}
func BenchmarkTranslateFile_TranslateColumnsToUint64(b *testing.B) {
const batchSize = 1000
@ -567,7 +850,7 @@ func (s *TranslateFile) Reopen() error {
s.TranslateFile = pilosa.NewTranslateFile()
s.lock.Unlock()
s.Path = prev.Path
s.PrimaryTranslateStore = prev.PrimaryTranslateStore
s.SetPrimaryStore("restored-primary", prev.PrimaryTranslateStore)
return s.Open()
}