mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
WIP: add tests for cluster resize
This commit adds support for allocating a gossip transport and a server listener prior to opening server (and cluster). Doing that allows tests to use a dynamically allocated port by supplying bind port: 0.
This commit is contained in:
Travis Turner
parent
9667a04cad
commit
0342da92bb
10 changed files with 974 additions and 242 deletions
87
cluster.go
87
cluster.go
|
|
@ -31,6 +31,8 @@ import (
|
|||
"sync"
|
||||
"time"
|
||||
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
"github.com/gogo/protobuf/proto"
|
||||
"github.com/pilosa/pilosa/internal"
|
||||
)
|
||||
|
|
@ -241,6 +243,7 @@ func (c *Cluster) SetCoordinator(oldURI, newURI URI) bool {
|
|||
// AddNode adds a node to the Cluster and updates and saves the
|
||||
// new topology.
|
||||
func (c *Cluster) AddNode(uri URI) error {
|
||||
c.logger().Printf("add node %s to cluster on %s", uri, c.URI)
|
||||
|
||||
// add to cluster
|
||||
_, added := c.addNodeBasicSorted(uri)
|
||||
|
|
@ -292,7 +295,7 @@ func (c *Cluster) setState(state string) {
|
|||
return
|
||||
}
|
||||
|
||||
c.logger().Printf("Change cluster state from %s to %s", c.State, state)
|
||||
c.logger().Printf("change cluster state from %s to %s on %s", c.State, state, c.URI)
|
||||
|
||||
var doCleanup bool
|
||||
|
||||
|
|
@ -352,8 +355,6 @@ func (c *Cluster) SetNodeState(state string) error {
|
|||
// Coordinator to keep track of, during startup, which nodes have
|
||||
// finished opening their Holder.
|
||||
func (c *Cluster) ReceiveNodeState(uri URI, state string) error {
|
||||
|
||||
c.logger().Printf("Receiving State %s (%s)", state, uri.String())
|
||||
if !c.IsCoordinator() {
|
||||
return nil
|
||||
}
|
||||
|
|
@ -364,11 +365,10 @@ func (c *Cluster) ReceiveNodeState(uri URI, state string) error {
|
|||
}
|
||||
|
||||
c.Topology.nodeStates[uri] = state
|
||||
c.logger().Printf("Receiving State %s (%s)", state, uri)
|
||||
c.logger().Printf("received state %s (%s)", state, uri)
|
||||
|
||||
// Set cluster state to NORMAL.
|
||||
if c.haveTopologyAgreement() && c.allNodesReady() {
|
||||
c.logger().Printf("Broadcasting ClusterStateNormal")
|
||||
return c.setStateAndBroadcast(ClusterStateNormal)
|
||||
}
|
||||
|
||||
|
|
@ -765,7 +765,10 @@ func (c *Cluster) Open() error {
|
|||
}
|
||||
|
||||
// Add the local node to the cluster.
|
||||
//NEXT
|
||||
//if c.URI.Port() != 0 {
|
||||
c.AddNode(c.URI)
|
||||
//}
|
||||
|
||||
// Start the EventReceiver.
|
||||
if err := c.EventReceiver.Start(c); err != nil {
|
||||
|
|
@ -781,6 +784,7 @@ func (c *Cluster) Open() error {
|
|||
if !c.IsCoordinator() {
|
||||
c.logger().Printf("wait for joining to complete")
|
||||
<-c.joining
|
||||
c.logger().Printf("joining has completed")
|
||||
}
|
||||
|
||||
return nil
|
||||
|
|
@ -794,7 +798,8 @@ func (c *Cluster) Close() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (c *Cluster) MarkAsJoined() {
|
||||
func (c *Cluster) markAsJoined() {
|
||||
c.logger().Printf("mark node as joined (received coordinator update)")
|
||||
if !c.joined {
|
||||
c.joined = true
|
||||
close(c.joining)
|
||||
|
|
@ -830,14 +835,24 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error {
|
|||
return err
|
||||
}
|
||||
|
||||
// Run the job.
|
||||
err = j.Run()
|
||||
if err != nil {
|
||||
// j.Run() runs in a goroutine because in the case where the
|
||||
// job requires no action, it immediately writes to the j.result
|
||||
// channel, which is not consumed until the code below.
|
||||
var eg errgroup.Group
|
||||
eg.Go(func() error {
|
||||
return j.Run()
|
||||
})
|
||||
|
||||
// Wait for the ResizeJob to finish or be aborted.
|
||||
c.logger().Printf("wait for jobResult")
|
||||
jobResult := <-j.result
|
||||
|
||||
// Make sure j.Run() didn't return an error.
|
||||
if eg.Wait() != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Wait for the ResizeJob to finish or be aborted.
|
||||
jobResult := <-j.result
|
||||
c.logger().Printf("received jobResult: %s", jobResult)
|
||||
switch jobResult {
|
||||
case ResizeJobStateDone:
|
||||
if err := c.CompleteCurrentJob(ResizeJobStateDone); err != nil {
|
||||
|
|
@ -860,6 +875,7 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error {
|
|||
func (c *Cluster) setStateAndBroadcast(state string) error {
|
||||
c.setState(state)
|
||||
// Broadcast cluster status changes to the cluster.
|
||||
c.logger().Printf("broadcasting ClusterStatus: %s", state)
|
||||
return c.Broadcaster.SendSync(c.Status())
|
||||
}
|
||||
|
||||
|
|
@ -996,11 +1012,12 @@ func (c *Cluster) generateResizeJobByAction(nodeAction nodeAction) (*ResizeJob,
|
|||
continue
|
||||
}
|
||||
instr := &internal.ResizeInstruction{
|
||||
JobID: j.ID,
|
||||
URI: u.Encode(),
|
||||
Coordinator: encodeURI(c.Coordinator),
|
||||
Sources: sources,
|
||||
Schema: pbSchema, // Include the schema to ensure it's in sync on the receiving node.
|
||||
JobID: j.ID,
|
||||
URI: u.Encode(),
|
||||
Coordinator: encodeURI(c.Coordinator),
|
||||
Sources: sources,
|
||||
Schema: pbSchema, // Include the schema to ensure it's in sync on the receiving node.
|
||||
ClusterStatus: c.Status(),
|
||||
}
|
||||
j.Instructions = append(j.Instructions, instr)
|
||||
}
|
||||
|
|
@ -1023,6 +1040,17 @@ func (c *Cluster) CompleteCurrentJob(state string) error {
|
|||
|
||||
// FollowResizeInstruction is run by any node that receives a ResizeInstruction.
|
||||
func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) error {
|
||||
c.logger().Printf("follow resize instruction on %s", c.URI)
|
||||
// Make sure the cluster status on this node agrees with the Coordinator
|
||||
// before attempting a resize.
|
||||
if err := c.MergeClusterStatus(instr.ClusterStatus); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c.logger().Printf("MergeClusterStatus done, start goroutine")
|
||||
|
||||
// The actual resizing runs in a goroutine because we don't want to block
|
||||
// the distribution of other ResizeInstructions to the rest of the cluster.
|
||||
go func() {
|
||||
|
||||
// Make sure the holder has opened.
|
||||
|
|
@ -1039,6 +1067,7 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err
|
|||
if err := func() error {
|
||||
|
||||
// Sync the schema received in the resize instruction.
|
||||
c.logger().Printf("Holder ApplySchema")
|
||||
if err := c.Holder.ApplySchema(instr.Schema); err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -1048,7 +1077,7 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err
|
|||
|
||||
// Request each source file in ResizeSources.
|
||||
for _, src := range instr.Sources {
|
||||
c.logger().Printf("\n**** Get slice %d for index %s from host %s ****\n\n", src.Slice, src.Index, src.URI)
|
||||
c.logger().Printf("get slice %d for index %s from host %s", src.Slice, src.Index, src.URI)
|
||||
|
||||
srcURI := decodeURI(src.URI)
|
||||
|
||||
|
|
@ -1071,8 +1100,18 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err
|
|||
}
|
||||
|
||||
// Stream slice from remote node.
|
||||
c.logger().Printf("retrieve slice %d for index %s from host %s", src.Slice, src.Index, src.URI)
|
||||
rd, err := client.RetrieveSliceFromURI(context.Background(), src.Index, src.Frame, src.View, src.Slice, srcURI)
|
||||
if err != nil {
|
||||
// For now it is an acceptable error if the fragment is not found
|
||||
// on the remote node. This occurs when a slice has been skipped and
|
||||
// therefore doesn't contain data. The coordinator correctly determined
|
||||
// the resize instruction to retrieve the slice, but it doesn't have data.
|
||||
// TODO: figure out a way to distinguish from "fragment not found" errors
|
||||
// which are true errors and which simply mean the fragment doesn't have data.
|
||||
if err == ErrFragmentNotFound {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
} else if rd == nil {
|
||||
return fmt.Errorf("slice %v doesn't exist on host: %s", src.Slice, src.URI)
|
||||
|
|
@ -1213,15 +1252,18 @@ func (j *ResizeJob) setState(state string) {
|
|||
|
||||
// Run distributes ResizeInstructions.
|
||||
func (j *ResizeJob) Run() error {
|
||||
j.logger().Printf("run ResizeJob")
|
||||
// Set job state to RUNNING.
|
||||
j.SetState(ResizeJobStateRunning)
|
||||
|
||||
// Job can be considered done in the case where it doesn't require any action.
|
||||
if !j.urisArePending() {
|
||||
j.logger().Printf("ResizeJob contains no pending tasks; mark as done")
|
||||
j.result <- ResizeJobStateDone
|
||||
return nil
|
||||
}
|
||||
|
||||
j.logger().Printf("distribute tasks for ResizeJob")
|
||||
err := j.distributeResizeInstructions()
|
||||
if err != nil {
|
||||
j.result <- ResizeJobStateAborted
|
||||
|
|
@ -1470,6 +1512,7 @@ func (c *Cluster) ReceiveEvent(e *NodeEvent) error {
|
|||
|
||||
switch e.Event {
|
||||
case NodeJoin:
|
||||
c.logger().Printf("received NodeJoin event: %v", e)
|
||||
// Ignore the event if this is not the coordinator.
|
||||
if !c.IsCoordinator() {
|
||||
return nil
|
||||
|
|
@ -1582,6 +1625,7 @@ func (c *Cluster) nodeLeave(uri URI) error {
|
|||
}
|
||||
|
||||
func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error {
|
||||
c.logger().Printf("merge cluster status: %v", cs)
|
||||
// Ignore status updates from self (coordinator).
|
||||
if c.IsCoordinator() {
|
||||
return nil
|
||||
|
|
@ -1596,8 +1640,13 @@ func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error {
|
|||
}
|
||||
}
|
||||
|
||||
// Remove any nodes not specified by the coordinator.
|
||||
// Remove any nodes not specified by the coordinator
|
||||
// except for self.
|
||||
for _, uri := range c.NodeSet() {
|
||||
// Don't remove this node.
|
||||
if uri == c.URI {
|
||||
continue
|
||||
}
|
||||
if NodeSet(officialURIs).ContainsURI(uri) {
|
||||
continue
|
||||
}
|
||||
|
|
@ -1608,5 +1657,7 @@ func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error {
|
|||
|
||||
c.setState(cs.State)
|
||||
|
||||
c.markAsJoined()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
211
gossip/gossip.go
211
gossip/gossip.go
|
|
@ -142,81 +142,26 @@ type gossipConfig struct {
|
|||
memberlistConfig *memberlist.Config
|
||||
}
|
||||
|
||||
// newTransport returns a NetTransport based on the memberlist configuration.
|
||||
// It will dynamically bind to a port if conf.BindPort is 0.
|
||||
// This is useful for test cases where specifiying a port is not reasonable.
|
||||
func newTransport(conf *memberlist.Config) (*memberlist.NetTransport, error) {
|
||||
if conf.LogOutput != nil && conf.Logger != nil {
|
||||
return nil, fmt.Errorf("Cannot specify both LogOutput and Logger. Please choose a single log configuration setting.")
|
||||
}
|
||||
// NewGossipMemberSetWithTransport returns a new instance of GossipMemberSet given a Transport.
|
||||
func NewGossipMemberSetWithTransport(name string, gossipHost string, transport *Transport, gossipSeed string, server *pilosa.Server, secretKey []byte) (*GossipMemberSet, error) {
|
||||
port := transport.Net.GetAutoBindPort()
|
||||
|
||||
logDest := conf.LogOutput
|
||||
if logDest == nil {
|
||||
logDest = os.Stderr
|
||||
}
|
||||
|
||||
logger := conf.Logger
|
||||
if logger == nil {
|
||||
logger = log.New(logDest, "", log.LstdFlags)
|
||||
}
|
||||
|
||||
nc := &memberlist.NetTransportConfig{
|
||||
BindAddrs: []string{conf.BindAddr},
|
||||
BindPort: conf.BindPort,
|
||||
Logger: logger,
|
||||
}
|
||||
|
||||
// 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") {
|
||||
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, fmt.Errorf("Could not set up network transport: %v", err)
|
||||
}
|
||||
if conf.BindPort == 0 {
|
||||
port := nt.GetAutoBindPort()
|
||||
conf.BindPort = port
|
||||
conf.AdvertisePort = port
|
||||
logger.Printf("[DEBUG] Using dynamic bind port %d", port)
|
||||
}
|
||||
|
||||
return nt, nil
|
||||
}
|
||||
|
||||
// NewGossipMemberSet returns a new instance of GossipMemberSet.
|
||||
func NewGossipMemberSet(name string, gossipHost string, gossipPort int, gossipSeed string, server *pilosa.Server, secretKey []byte) (*GossipMemberSet, error) {
|
||||
g := &GossipMemberSet{
|
||||
LogOutput: server.LogOutput,
|
||||
}
|
||||
|
||||
// memberlist config
|
||||
conf := memberlist.DefaultLocalConfig()
|
||||
conf.BindPort = gossipPort
|
||||
conf.AdvertisePort = gossipPort
|
||||
conf.Transport = transport.Net
|
||||
conf.BindPort = port
|
||||
conf.AdvertisePort = port
|
||||
conf.Name = name
|
||||
conf.BindAddr = gossipHost
|
||||
conf.AdvertiseAddr = pilosa.HostToIP(gossipHost)
|
||||
//conf.PushPullInterval = 0 * time.Second // Default is 15s in DefaultLocalConfig.
|
||||
conf.Delegate = g
|
||||
conf.SecretKey = secretKey
|
||||
conf.Events = server.Cluster.EventReceiver.(memberlist.EventDelegate)
|
||||
|
||||
//TODO: pull memberlist config from pilosa.cfg file
|
||||
g.config = &gossipConfig{
|
||||
|
|
@ -224,32 +169,27 @@ func NewGossipMemberSet(name string, gossipHost string, gossipPort int, gossipSe
|
|||
gossipSeed: gossipSeed,
|
||||
}
|
||||
|
||||
g.config.memberlistConfig.Name = name
|
||||
g.config.memberlistConfig.BindAddr = gossipHost
|
||||
g.config.memberlistConfig.AdvertiseAddr = pilosa.HostToIP(gossipHost)
|
||||
g.config.memberlistConfig.AdvertisePort = gossipPort
|
||||
//g.config.memberlistConfig.PushPullInterval = 0 * time.Second // Default is 15s in DefaultLocalConfig.
|
||||
g.config.memberlistConfig.Delegate = g
|
||||
g.config.memberlistConfig.SecretKey = secretKey
|
||||
g.config.memberlistConfig.Events = server.Cluster.EventReceiver.(memberlist.EventDelegate)
|
||||
|
||||
g.statusHandler = server
|
||||
|
||||
// set up the transport
|
||||
transport, err := newTransport(g.config.memberlistConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
g.config.memberlistConfig.Transport = transport
|
||||
|
||||
// If no gossipSeed is provided, use local host:port.
|
||||
if gossipSeed == "" {
|
||||
g.config.gossipSeed = fmt.Sprintf("%s:%d", gossipHost, g.config.memberlistConfig.BindPort)
|
||||
g.config.gossipSeed = fmt.Sprintf("%s:%d", gossipHost, port)
|
||||
}
|
||||
|
||||
return g, nil
|
||||
}
|
||||
|
||||
// NewGossipMemberSet returns a new instance of GossipMemberSet given a gossip port.
|
||||
func NewGossipMemberSet(name string, gossipHost string, gossipPort int, gossipSeed string, server *pilosa.Server, secretKey []byte) (*GossipMemberSet, error) {
|
||||
// set up the transport
|
||||
transport, err := NewTransport(gossipHost, gossipPort)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return NewGossipMemberSetWithTransport(name, gossipHost, transport, gossipSeed, server, secretKey)
|
||||
}
|
||||
|
||||
// SendSync implementation of the Broadcaster interface.
|
||||
func (g *GossipMemberSet) SendSync(pb proto.Message) error {
|
||||
msg, err := pilosa.MarshalMessage(pb)
|
||||
|
|
@ -432,3 +372,102 @@ func (b *broadcast) Finished() {
|
|||
close(b.notify)
|
||||
}
|
||||
}
|
||||
|
||||
// Transport is a gossip transport for binding to a port.
|
||||
type Transport struct {
|
||||
//memberlist.Transport
|
||||
Net *memberlist.NetTransport
|
||||
URI *pilosa.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 specifiying a port is not reasonable.
|
||||
//func NewTransport(host string, port int) (*memberlist.NetTransport, error) {
|
||||
func NewTransport(host string, port int) (*Transport, error) {
|
||||
// memberlist config
|
||||
conf := memberlist.DefaultLocalConfig()
|
||||
conf.BindAddr = host
|
||||
conf.BindPort = port
|
||||
conf.AdvertisePort = port
|
||||
|
||||
net, err := newTransport(conf)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
uri, err := pilosa.NewURIFromHostPort(host, uint16(net.GetAutoBindPort()))
|
||||
if err != nil {
|
||||
return nil, 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) {
|
||||
if conf.LogOutput != nil && conf.Logger != nil {
|
||||
return nil, fmt.Errorf("Cannot specify both LogOutput and Logger. Please choose a single log configuration setting.")
|
||||
}
|
||||
|
||||
logDest := conf.LogOutput
|
||||
if logDest == nil {
|
||||
logDest = os.Stderr
|
||||
}
|
||||
|
||||
logger := conf.Logger
|
||||
if logger == nil {
|
||||
logger = log.New(logDest, "", log.LstdFlags)
|
||||
}
|
||||
|
||||
nc := &memberlist.NetTransportConfig{
|
||||
BindAddrs: []string{conf.BindAddr},
|
||||
BindPort: conf.BindPort,
|
||||
Logger: logger,
|
||||
}
|
||||
|
||||
// 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") {
|
||||
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, fmt.Errorf("Could not set up network transport: %v", err)
|
||||
}
|
||||
if conf.BindPort == 0 {
|
||||
port := nt.GetAutoBindPort()
|
||||
conf.BindPort = port
|
||||
conf.AdvertisePort = port
|
||||
logger.Printf("[DEBUG] Using dynamic bind port %d", port)
|
||||
}
|
||||
|
||||
return nt, nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -124,6 +124,7 @@ func (h *Handler) SetRestricted() {
|
|||
}
|
||||
|
||||
func loadCommon(router *mux.Router, handler *Handler) {
|
||||
router.HandleFunc("/cluster/message", handler.handlePostClusterMessage).Methods("POST")
|
||||
router.HandleFunc("/cluster/resize/set-coordinator", handler.handlePostClusterResizeSetCoordinator).Methods("POST")
|
||||
router.HandleFunc("/schema", handler.handleGetSchema).Methods("GET")
|
||||
router.HandleFunc("/status", handler.handleGetStatus).Methods("GET")
|
||||
|
|
@ -175,7 +176,6 @@ func loadNormal(router *mux.Router, handler *Handler) {
|
|||
router.HandleFunc("/index/{index}/query", handler.handlePostQuery).Methods("POST")
|
||||
router.HandleFunc("/index/{index}/time-quantum", handler.handlePatchIndexTimeQuantum).Methods("PATCH")
|
||||
router.HandleFunc("/recalculate-caches", handler.handleRecalculateCaches).Methods("POST")
|
||||
router.HandleFunc("/cluster/message", handler.handlePostClusterMessage).Methods("POST")
|
||||
|
||||
// TODO: Apply MethodNotAllowed statuses to all endpoints.
|
||||
// Ideally this would be automatic, as described in this (wontfix) ticket:
|
||||
|
|
|
|||
|
|
@ -88,6 +88,7 @@ func NewHolder() *Holder {
|
|||
// without actually loading any data into memory.
|
||||
// HasData is returned, and h.hasData is set.
|
||||
func (h *Holder) Peek() bool {
|
||||
h.logger().Printf("peek at holder path: %s", h.Path)
|
||||
h.hasData = false
|
||||
|
||||
// Open path to read all index directories.
|
||||
|
|
@ -117,6 +118,7 @@ func (h *Holder) Peek() bool {
|
|||
func (h *Holder) Open() error {
|
||||
h.setFileLimit()
|
||||
|
||||
h.logger().Printf("open holder path: %s", h.Path)
|
||||
if err := os.MkdirAll(h.Path, 0777); err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -133,7 +135,6 @@ func (h *Holder) Open() error {
|
|||
return err
|
||||
}
|
||||
|
||||
h.logger().Printf("Holder Start")
|
||||
for _, fi := range fis {
|
||||
if !fi.IsDir() {
|
||||
continue
|
||||
|
|
@ -157,7 +158,7 @@ func (h *Holder) Open() error {
|
|||
}
|
||||
h.indexes[index.Name()] = index
|
||||
}
|
||||
h.logger().Printf("Holder Complete")
|
||||
h.logger().Printf("open holder: complete")
|
||||
|
||||
// Periodically flush cache.
|
||||
h.wg.Add(1)
|
||||
|
|
|
|||
|
|
@ -824,11 +824,12 @@ func (m *DeleteViewMessage) GetView() string {
|
|||
}
|
||||
|
||||
type ResizeInstruction struct {
|
||||
JobID int64 `protobuf:"varint,1,opt,name=JobID,proto3" json:"JobID,omitempty"`
|
||||
URI *URI `protobuf:"bytes,2,opt,name=URI" json:"URI,omitempty"`
|
||||
Coordinator *URI `protobuf:"bytes,3,opt,name=Coordinator" json:"Coordinator,omitempty"`
|
||||
Sources []*ResizeSource `protobuf:"bytes,4,rep,name=Sources" json:"Sources,omitempty"`
|
||||
Schema *Schema `protobuf:"bytes,5,opt,name=Schema" json:"Schema,omitempty"`
|
||||
JobID int64 `protobuf:"varint,1,opt,name=JobID,proto3" json:"JobID,omitempty"`
|
||||
URI *URI `protobuf:"bytes,2,opt,name=URI" json:"URI,omitempty"`
|
||||
Coordinator *URI `protobuf:"bytes,3,opt,name=Coordinator" json:"Coordinator,omitempty"`
|
||||
Sources []*ResizeSource `protobuf:"bytes,4,rep,name=Sources" json:"Sources,omitempty"`
|
||||
Schema *Schema `protobuf:"bytes,5,opt,name=Schema" json:"Schema,omitempty"`
|
||||
ClusterStatus *ClusterStatus `protobuf:"bytes,6,opt,name=ClusterStatus" json:"ClusterStatus,omitempty"`
|
||||
}
|
||||
|
||||
func (m *ResizeInstruction) Reset() { *m = ResizeInstruction{} }
|
||||
|
|
@ -871,6 +872,13 @@ func (m *ResizeInstruction) GetSchema() *Schema {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (m *ResizeInstruction) GetClusterStatus() *ClusterStatus {
|
||||
if m != nil {
|
||||
return m.ClusterStatus
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type ResizeSource struct {
|
||||
URI *URI `protobuf:"bytes,1,opt,name=URI" json:"URI,omitempty"`
|
||||
Index string `protobuf:"bytes,2,opt,name=Index,proto3" json:"Index,omitempty"`
|
||||
|
|
@ -2130,6 +2138,16 @@ func (m *ResizeInstruction) MarshalTo(dAtA []byte) (int, error) {
|
|||
}
|
||||
i += n17
|
||||
}
|
||||
if m.ClusterStatus != nil {
|
||||
dAtA[i] = 0x32
|
||||
i++
|
||||
i = encodeVarintPrivate(dAtA, i, uint64(m.ClusterStatus.Size()))
|
||||
n18, err := m.ClusterStatus.MarshalTo(dAtA[i:])
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
i += n18
|
||||
}
|
||||
return i, nil
|
||||
}
|
||||
|
||||
|
|
@ -2152,11 +2170,11 @@ func (m *ResizeSource) MarshalTo(dAtA []byte) (int, error) {
|
|||
dAtA[i] = 0xa
|
||||
i++
|
||||
i = encodeVarintPrivate(dAtA, i, uint64(m.URI.Size()))
|
||||
n18, err := m.URI.MarshalTo(dAtA[i:])
|
||||
n19, err := m.URI.MarshalTo(dAtA[i:])
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
i += n18
|
||||
i += n19
|
||||
}
|
||||
if len(m.Index) > 0 {
|
||||
dAtA[i] = 0x12
|
||||
|
|
@ -2208,11 +2226,11 @@ func (m *ResizeInstructionComplete) MarshalTo(dAtA []byte) (int, error) {
|
|||
dAtA[i] = 0x12
|
||||
i++
|
||||
i = encodeVarintPrivate(dAtA, i, uint64(m.URI.Size()))
|
||||
n19, err := m.URI.MarshalTo(dAtA[i:])
|
||||
n20, err := m.URI.MarshalTo(dAtA[i:])
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
i += n19
|
||||
i += n20
|
||||
}
|
||||
if len(m.Error) > 0 {
|
||||
dAtA[i] = 0x1a
|
||||
|
|
@ -2242,21 +2260,21 @@ func (m *SetCoordinatorMessage) MarshalTo(dAtA []byte) (int, error) {
|
|||
dAtA[i] = 0xa
|
||||
i++
|
||||
i = encodeVarintPrivate(dAtA, i, uint64(m.Old.Size()))
|
||||
n20, err := m.Old.MarshalTo(dAtA[i:])
|
||||
n21, err := m.Old.MarshalTo(dAtA[i:])
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
i += n20
|
||||
i += n21
|
||||
}
|
||||
if m.New != nil {
|
||||
dAtA[i] = 0x12
|
||||
i++
|
||||
i = encodeVarintPrivate(dAtA, i, uint64(m.New.Size()))
|
||||
n21, err := m.New.MarshalTo(dAtA[i:])
|
||||
n22, err := m.New.MarshalTo(dAtA[i:])
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
i += n21
|
||||
i += n22
|
||||
}
|
||||
return i, nil
|
||||
}
|
||||
|
|
@ -2783,6 +2801,10 @@ func (m *ResizeInstruction) Size() (n int) {
|
|||
l = m.Schema.Size()
|
||||
n += 1 + l + sovPrivate(uint64(l))
|
||||
}
|
||||
if m.ClusterStatus != nil {
|
||||
l = m.ClusterStatus.Size()
|
||||
n += 1 + l + sovPrivate(uint64(l))
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
|
|
@ -6611,6 +6633,39 @@ func (m *ResizeInstruction) Unmarshal(dAtA []byte) error {
|
|||
return err
|
||||
}
|
||||
iNdEx = postIndex
|
||||
case 6:
|
||||
if wireType != 2 {
|
||||
return fmt.Errorf("proto: wrong wireType = %d for field ClusterStatus", wireType)
|
||||
}
|
||||
var msglen int
|
||||
for shift := uint(0); ; shift += 7 {
|
||||
if shift >= 64 {
|
||||
return ErrIntOverflowPrivate
|
||||
}
|
||||
if iNdEx >= l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
b := dAtA[iNdEx]
|
||||
iNdEx++
|
||||
msglen |= (int(b) & 0x7F) << shift
|
||||
if b < 0x80 {
|
||||
break
|
||||
}
|
||||
}
|
||||
if msglen < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
postIndex := iNdEx + msglen
|
||||
if postIndex > l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
if m.ClusterStatus == nil {
|
||||
m.ClusterStatus = &ClusterStatus{}
|
||||
}
|
||||
if err := m.ClusterStatus.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
|
||||
return err
|
||||
}
|
||||
iNdEx = postIndex
|
||||
default:
|
||||
iNdEx = preIndex
|
||||
skippy, err := skipPrivate(dAtA[iNdEx:])
|
||||
|
|
@ -7257,80 +7312,81 @@ var (
|
|||
func init() { proto.RegisterFile("private.proto", fileDescriptorPrivate) }
|
||||
|
||||
var fileDescriptorPrivate = []byte{
|
||||
// 1192 bytes of a gzipped FileDescriptorProto
|
||||
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x57, 0x4d, 0x6f, 0x23, 0x45,
|
||||
0x13, 0x7e, 0xc7, 0x63, 0x3b, 0x76, 0x79, 0xbd, 0xeb, 0xf4, 0x9b, 0x8d, 0x9c, 0x28, 0xf2, 0x9a,
|
||||
0x3e, 0x90, 0xb0, 0x12, 0x01, 0x12, 0x09, 0x41, 0x10, 0x12, 0x6c, 0xec, 0xd5, 0x0e, 0x90, 0x64,
|
||||
0x69, 0x67, 0x17, 0xc1, 0x01, 0xa9, 0x63, 0x37, 0xc9, 0x28, 0xe3, 0x19, 0x33, 0xd3, 0x4e, 0xe2,
|
||||
0x3d, 0x70, 0x83, 0x03, 0xfc, 0x01, 0xee, 0xfc, 0x19, 0x8e, 0xdc, 0xb8, 0xa2, 0xf0, 0x23, 0x90,
|
||||
0xb8, 0x80, 0xba, 0xa6, 0x7b, 0x66, 0xfc, 0x15, 0x93, 0xdc, 0xa6, 0xaa, 0xab, 0xaa, 0x9f, 0x7e,
|
||||
0xea, 0xa3, 0x7b, 0xa0, 0x3a, 0x08, 0xdd, 0x0b, 0x2e, 0xc5, 0xf6, 0x20, 0x0c, 0x64, 0x40, 0x4a,
|
||||
0xae, 0x2f, 0x45, 0xe8, 0x73, 0x8f, 0x1e, 0x41, 0xd9, 0xf1, 0x7b, 0xe2, 0xea, 0x40, 0x48, 0x4e,
|
||||
0x9a, 0x50, 0xd9, 0x0f, 0xbc, 0x61, 0xdf, 0xff, 0x8c, 0x9f, 0x08, 0xaf, 0x6e, 0x35, 0xad, 0xad,
|
||||
0x32, 0xcb, 0xaa, 0x94, 0xc5, 0xb1, 0xdb, 0x17, 0x9f, 0x0f, 0xb9, 0x2f, 0x87, 0xfd, 0x7a, 0x2e,
|
||||
0xb6, 0xc8, 0xa8, 0xe8, 0xdf, 0x16, 0x94, 0x9f, 0x86, 0xbc, 0x2f, 0x30, 0xe2, 0x3a, 0x94, 0x58,
|
||||
0x70, 0x99, 0x0d, 0x97, 0xc8, 0xe4, 0x75, 0xb8, 0xef, 0xf8, 0x17, 0x22, 0x8c, 0x44, 0xdb, 0xe7,
|
||||
0x27, 0x9e, 0xe8, 0x61, 0xb8, 0x12, 0x9b, 0xd0, 0x92, 0x0d, 0x28, 0xef, 0xf3, 0xee, 0x99, 0x38,
|
||||
0x1e, 0x0d, 0x44, 0xdd, 0xc6, 0x20, 0xa9, 0x22, 0x59, 0xed, 0xb8, 0xaf, 0x44, 0x3d, 0xdf, 0xb4,
|
||||
0xb6, 0xaa, 0x2c, 0x55, 0x4c, 0xe2, 0x2d, 0x4c, 0xe1, 0x25, 0x14, 0xee, 0x31, 0xee, 0x9f, 0x26,
|
||||
0x18, 0x8a, 0x88, 0x61, 0x4c, 0x47, 0x36, 0xa1, 0xf8, 0xd4, 0x15, 0x5e, 0x2f, 0xaa, 0x2f, 0x35,
|
||||
0xed, 0xad, 0xca, 0xce, 0x83, 0x6d, 0xc3, 0xdf, 0x36, 0xea, 0x99, 0x5e, 0xa6, 0x14, 0xee, 0x3b,
|
||||
0xfd, 0x41, 0x10, 0x4a, 0x26, 0xa2, 0x41, 0xe0, 0x47, 0x82, 0xd4, 0xc0, 0x6e, 0x87, 0xa1, 0x3e,
|
||||
0xbb, 0xfa, 0xa4, 0xdf, 0x41, 0xed, 0x89, 0x17, 0x74, 0xcf, 0x5b, 0x5c, 0x72, 0x26, 0xbe, 0x1d,
|
||||
0x8a, 0x48, 0x92, 0x15, 0x28, 0x60, 0x16, 0xb4, 0x5d, 0x2c, 0x28, 0x2d, 0x32, 0xa9, 0x69, 0x8e,
|
||||
0x05, 0xa5, 0x45, 0x7f, 0xa4, 0x22, 0xcf, 0x62, 0x41, 0x69, 0x3b, 0x9e, 0xdb, 0x8d, 0x29, 0xc8,
|
||||
0xb3, 0x58, 0x20, 0x04, 0xf2, 0x2f, 0x5d, 0x71, 0xa9, 0xcf, 0x8d, 0xdf, 0xd4, 0x81, 0xe5, 0xcc,
|
||||
0xfe, 0x1a, 0xe6, 0x2a, 0x14, 0x59, 0x70, 0xe9, 0xb4, 0xa2, 0xba, 0xd5, 0xb4, 0xb7, 0xf2, 0x4c,
|
||||
0x4b, 0xc8, 0x2e, 0xa6, 0x5f, 0x2d, 0xe5, 0x70, 0x29, 0x55, 0xd0, 0x35, 0x28, 0x20, 0xd5, 0xea,
|
||||
0x94, 0xa9, 0xaf, 0xfa, 0xa4, 0xff, 0x58, 0x50, 0x3e, 0xe0, 0x57, 0x08, 0x23, 0x22, 0x1f, 0x42,
|
||||
0xa9, 0x23, 0xb9, 0xdf, 0xe3, 0x61, 0x0f, 0x8d, 0x2a, 0x3b, 0xaf, 0xa5, 0x14, 0x26, 0x66, 0xdb,
|
||||
0xc6, 0xa6, 0xed, 0xcb, 0x70, 0xc4, 0x12, 0x17, 0xb2, 0x07, 0x4b, 0xba, 0x26, 0x10, 0x43, 0x65,
|
||||
0xa7, 0x39, 0xcb, 0x3b, 0x29, 0x1b, 0xe5, 0x6c, 0x1c, 0xd6, 0x3f, 0x80, 0xea, 0x58, 0x58, 0x85,
|
||||
0xf5, 0x5c, 0x8c, 0x4c, 0x46, 0xce, 0xc5, 0x48, 0x71, 0x77, 0xc1, 0xbd, 0x61, 0xcc, 0x73, 0x9e,
|
||||
0xc5, 0xc2, 0x5e, 0xee, 0x3d, 0x6b, 0x7d, 0x0f, 0xee, 0x65, 0xa3, 0xde, 0xc6, 0x97, 0x7e, 0x0d,
|
||||
0x64, 0x3f, 0x14, 0x5c, 0x0a, 0x84, 0x77, 0x20, 0xa2, 0x88, 0x9f, 0x8a, 0xf9, 0x99, 0x8e, 0xb3,
|
||||
0x97, 0xcb, 0x66, 0x6f, 0x03, 0xca, 0x4e, 0x64, 0x0e, 0x6e, 0x63, 0x5d, 0xa6, 0x0a, 0xfa, 0x18,
|
||||
0x48, 0x4b, 0x78, 0x42, 0x0a, 0xdd, 0xbf, 0x37, 0xc4, 0xa7, 0x1d, 0x83, 0x65, 0xb1, 0x2d, 0xd9,
|
||||
0x84, 0xbc, 0x6a, 0x5d, 0x84, 0x52, 0xd9, 0xf9, 0x7f, 0xca, 0x74, 0x32, 0x27, 0x18, 0x1a, 0x50,
|
||||
0xd7, 0x04, 0xd5, 0xed, 0xbe, 0xe0, 0x80, 0x33, 0x4a, 0xd9, 0x6c, 0x65, 0x4f, 0x6e, 0x95, 0x0c,
|
||||
0x10, 0xbd, 0xd5, 0x47, 0xe6, 0xac, 0x77, 0xdd, 0x8a, 0x7e, 0xa5, 0xb5, 0xaa, 0x25, 0x0e, 0xd5,
|
||||
0x6a, 0xec, 0x83, 0xdf, 0xf3, 0x8f, 0x3c, 0x81, 0x43, 0xc5, 0x56, 0x3d, 0x14, 0xd5, 0xed, 0xa6,
|
||||
0xad, 0x62, 0xa3, 0x40, 0x77, 0xa1, 0xd8, 0xe9, 0x9e, 0x89, 0x3e, 0x27, 0x6f, 0xa8, 0x42, 0xed,
|
||||
0x89, 0x2b, 0x11, 0xe9, 0x32, 0x7f, 0x30, 0x41, 0x1f, 0x33, 0xeb, 0xf4, 0x27, 0x4b, 0xa3, 0x9f,
|
||||
0x83, 0xa8, 0x88, 0x7b, 0x47, 0xf5, 0xfc, 0xd4, 0xc4, 0x51, 0x7a, 0xa6, 0x97, 0x49, 0x1b, 0x6a,
|
||||
0x8e, 0x3f, 0x18, 0xca, 0x96, 0xf8, 0xc6, 0xf5, 0x5d, 0xe9, 0x06, 0x7e, 0x54, 0x2f, 0xa2, 0xcb,
|
||||
0x5a, 0x76, 0xeb, 0x31, 0x0b, 0x36, 0xe5, 0x42, 0x7f, 0xb0, 0xe0, 0xc1, 0x84, 0x72, 0x01, 0xae,
|
||||
0xdc, 0xcd, 0xb8, 0xde, 0x4d, 0x46, 0xa6, 0x8d, 0x86, 0x8d, 0xb9, 0x68, 0xc6, 0x27, 0xe8, 0x2f,
|
||||
0x16, 0xac, 0xcc, 0x32, 0x98, 0x89, 0xa6, 0x01, 0xf0, 0x3c, 0x74, 0xfb, 0x3c, 0x1c, 0x7d, 0x2a,
|
||||
0x46, 0xfa, 0xf6, 0xc8, 0x68, 0xc8, 0x17, 0xb0, 0x3a, 0x11, 0xeb, 0xe3, 0x6e, 0x4c, 0x51, 0x0c,
|
||||
0xea, 0xd1, 0x5c, 0x50, 0xb1, 0x1d, 0x9b, 0xe3, 0x4e, 0xff, 0xb2, 0xe0, 0xe1, 0xcc, 0xa5, 0xb4,
|
||||
0xfa, 0xac, 0x6c, 0xa1, 0x3f, 0x86, 0xda, 0x4b, 0x35, 0x18, 0x5a, 0x22, 0x92, 0xae, 0xcf, 0x95,
|
||||
0xa5, 0x2e, 0xcf, 0x29, 0x3d, 0x71, 0xa0, 0x84, 0xba, 0x03, 0x3e, 0xd0, 0x30, 0xdf, 0x5c, 0x00,
|
||||
0x73, 0xdb, 0xd8, 0xeb, 0xb9, 0x69, 0x44, 0x05, 0x06, 0xe7, 0xb8, 0xb9, 0x14, 0x50, 0x50, 0x13,
|
||||
0x71, 0xcc, 0xe1, 0x56, 0x53, 0x2d, 0x80, 0x0d, 0x33, 0x49, 0xc6, 0x90, 0xdc, 0xdc, 0x93, 0xef,
|
||||
0x03, 0xa4, 0xa6, 0xba, 0xdd, 0x6f, 0xa8, 0xcf, 0x8c, 0x31, 0x7d, 0x06, 0x1b, 0x66, 0xcc, 0xdd,
|
||||
0x62, 0x43, 0x53, 0x2d, 0xb9, 0xb4, 0x5a, 0x68, 0x1b, 0xec, 0x17, 0xcc, 0x51, 0x57, 0x1d, 0x76,
|
||||
0xab, 0x49, 0x91, 0x96, 0x94, 0xcb, 0xb3, 0x20, 0x92, 0xc6, 0x45, 0x7d, 0x2b, 0xdd, 0xf3, 0x20,
|
||||
0x94, 0x88, 0xb8, 0xca, 0xf0, 0x9b, 0x3a, 0x50, 0x3b, 0x0c, 0x7a, 0xa2, 0x23, 0xb9, 0x4c, 0x26,
|
||||
0xd1, 0x23, 0x0c, 0x8d, 0x01, 0x2b, 0x3b, 0xd5, 0xf4, 0x60, 0x2f, 0x98, 0xc3, 0x70, 0x53, 0x35,
|
||||
0xe0, 0x95, 0x83, 0x19, 0x4a, 0x28, 0xd0, 0x1f, 0x2d, 0x00, 0x13, 0x6b, 0x18, 0x2d, 0x8e, 0xf2,
|
||||
0x4e, 0xe6, 0x4e, 0x9d, 0x1e, 0x56, 0xc9, 0x12, 0xcb, 0xdc, 0xbc, 0x5b, 0x66, 0x36, 0x69, 0xd6,
|
||||
0x6b, 0xa9, 0x7d, 0xac, 0xd7, 0xe7, 0xe7, 0xf4, 0x10, 0xaa, 0xfb, 0xde, 0x30, 0x92, 0x22, 0xd4,
|
||||
0x70, 0x12, 0xcc, 0x56, 0x06, 0x33, 0xd9, 0x84, 0x25, 0x84, 0x2c, 0xa4, 0x1e, 0x01, 0x13, 0x40,
|
||||
0xcd, 0x2a, 0xed, 0x40, 0x61, 0x7e, 0xe7, 0x12, 0xc8, 0xe3, 0x73, 0x4e, 0x93, 0x8d, 0x2f, 0xb9,
|
||||
0x1a, 0xd8, 0x07, 0x6e, 0x5c, 0x1d, 0x36, 0x53, 0x9f, 0xa8, 0xe1, 0x57, 0x58, 0xbd, 0x4a, 0xc3,
|
||||
0xd5, 0x45, 0xb6, 0x1c, 0x57, 0x83, 0x9a, 0xbc, 0x77, 0xb9, 0x72, 0xcc, 0x8b, 0xc8, 0xce, 0xbc,
|
||||
0x88, 0x7e, 0xb7, 0x60, 0x99, 0x89, 0xc8, 0x7d, 0x25, 0x1c, 0x3f, 0x92, 0xe1, 0x30, 0xe9, 0xe4,
|
||||
0x4f, 0x82, 0x13, 0xa7, 0x85, 0x51, 0x6d, 0x16, 0x0b, 0x26, 0x47, 0xb9, 0xb9, 0x39, 0x7a, 0x4b,
|
||||
0xbd, 0xa1, 0x83, 0xb0, 0xa7, 0xda, 0x39, 0x08, 0x35, 0xeb, 0x13, 0x86, 0x59, 0x0b, 0xf2, 0x36,
|
||||
0x2c, 0x75, 0x82, 0x61, 0xd8, 0x4d, 0x66, 0xfd, 0x6a, 0x6a, 0x1c, 0xa3, 0x8a, 0x97, 0x99, 0x31,
|
||||
0xcb, 0xe4, 0xb4, 0xb0, 0x20, 0xa7, 0xdf, 0x5b, 0x70, 0x2f, 0x1b, 0xe3, 0x3f, 0x15, 0x6a, 0xcc,
|
||||
0x65, 0x6e, 0x26, 0x97, 0xf6, 0x2c, 0x2e, 0xf3, 0x29, 0x97, 0xe9, 0x4b, 0xa6, 0x90, 0x79, 0xc9,
|
||||
0xd0, 0x33, 0x58, 0x9b, 0x22, 0x78, 0x3f, 0xe8, 0x0f, 0x54, 0x26, 0xef, 0x4a, 0xf4, 0x0a, 0x14,
|
||||
0xda, 0x61, 0xa8, 0x29, 0x2e, 0xb3, 0x58, 0xa0, 0x5f, 0xc2, 0xc3, 0x8e, 0x90, 0x19, 0x7e, 0x33,
|
||||
0x2d, 0x7a, 0xe4, 0xf5, 0xe6, 0x9c, 0xfc, 0xc8, 0xeb, 0x29, 0x83, 0x43, 0x71, 0x39, 0x67, 0xc3,
|
||||
0x43, 0x71, 0x49, 0x77, 0xa1, 0x74, 0x1c, 0x0c, 0x02, 0x2f, 0x38, 0x1d, 0x65, 0xbb, 0xc0, 0xba,
|
||||
0xa9, 0x0b, 0x9e, 0xd4, 0x7e, 0xbd, 0x6e, 0x58, 0xbf, 0x5d, 0x37, 0xac, 0x3f, 0xae, 0x1b, 0xd6,
|
||||
0xcf, 0x7f, 0x36, 0xfe, 0x77, 0x52, 0xc4, 0x5f, 0xb0, 0xdd, 0x7f, 0x03, 0x00, 0x00, 0xff, 0xff,
|
||||
0x50, 0x2b, 0xa7, 0xda, 0x93, 0x0d, 0x00, 0x00,
|
||||
// 1208 bytes of a gzipped FileDescriptorProto
|
||||
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x57, 0xcf, 0x6f, 0x1b, 0xc5,
|
||||
0x17, 0xff, 0xae, 0xd7, 0x76, 0xec, 0xe7, 0xba, 0x75, 0xe7, 0xdb, 0x16, 0x27, 0x8a, 0x5c, 0x33,
|
||||
0x07, 0x12, 0x2a, 0x11, 0xc0, 0x91, 0x10, 0x04, 0x55, 0x82, 0xc6, 0xae, 0xba, 0x40, 0x92, 0x32,
|
||||
0x4e, 0x8b, 0xe0, 0x80, 0x34, 0xb1, 0x87, 0x64, 0x95, 0xf5, 0xae, 0xd9, 0x1d, 0x27, 0x71, 0x0f,
|
||||
0xdc, 0xe0, 0x00, 0x77, 0xc4, 0x9d, 0x7f, 0x86, 0x23, 0x7f, 0x02, 0x0a, 0x7f, 0x04, 0x12, 0x17,
|
||||
0xd0, 0xbc, 0x9d, 0xd9, 0x5d, 0xff, 0x0c, 0xc9, 0x6d, 0xdf, 0x9b, 0xcf, 0x7b, 0xf3, 0x99, 0xf7,
|
||||
0x6b, 0x66, 0xa1, 0x3a, 0x0c, 0xdd, 0x33, 0x2e, 0xc5, 0xd6, 0x30, 0x0c, 0x64, 0x40, 0x4a, 0xae,
|
||||
0x2f, 0x45, 0xe8, 0x73, 0x8f, 0x1e, 0x40, 0xd9, 0xf1, 0xfb, 0xe2, 0x62, 0x4f, 0x48, 0x4e, 0x9a,
|
||||
0x50, 0xd9, 0x0d, 0xbc, 0xd1, 0xc0, 0xff, 0x8c, 0x1f, 0x09, 0xaf, 0x6e, 0x35, 0xad, 0xcd, 0x32,
|
||||
0xcb, 0xaa, 0x14, 0xe2, 0xd0, 0x1d, 0x88, 0xcf, 0x47, 0xdc, 0x97, 0xa3, 0x41, 0x3d, 0x17, 0x23,
|
||||
0x32, 0x2a, 0xfa, 0xb7, 0x05, 0xe5, 0xa7, 0x21, 0x1f, 0x08, 0xf4, 0xb8, 0x06, 0x25, 0x16, 0x9c,
|
||||
0x67, 0xdd, 0x25, 0x32, 0x79, 0x03, 0x6e, 0x3b, 0xfe, 0x99, 0x08, 0x23, 0xd1, 0xf1, 0xf9, 0x91,
|
||||
0x27, 0xfa, 0xe8, 0xae, 0xc4, 0xa6, 0xb4, 0x64, 0x1d, 0xca, 0xbb, 0xbc, 0x77, 0x22, 0x0e, 0xc7,
|
||||
0x43, 0x51, 0xb7, 0xd1, 0x49, 0xaa, 0x48, 0x56, 0xbb, 0xee, 0x2b, 0x51, 0xcf, 0x37, 0xad, 0xcd,
|
||||
0x2a, 0x4b, 0x15, 0xd3, 0x7c, 0x0b, 0x33, 0x7c, 0x09, 0x85, 0x5b, 0x8c, 0xfb, 0xc7, 0x09, 0x87,
|
||||
0x22, 0x72, 0x98, 0xd0, 0x91, 0x0d, 0x28, 0x3e, 0x75, 0x85, 0xd7, 0x8f, 0xea, 0x2b, 0x4d, 0x7b,
|
||||
0xb3, 0xd2, 0xba, 0xb3, 0x65, 0xe2, 0xb7, 0x85, 0x7a, 0xa6, 0x97, 0x29, 0x85, 0xdb, 0xce, 0x60,
|
||||
0x18, 0x84, 0x92, 0x89, 0x68, 0x18, 0xf8, 0x91, 0x20, 0x35, 0xb0, 0x3b, 0x61, 0xa8, 0xcf, 0xae,
|
||||
0x3e, 0xe9, 0x77, 0x50, 0x7b, 0xe2, 0x05, 0xbd, 0xd3, 0x36, 0x97, 0x9c, 0x89, 0x6f, 0x47, 0x22,
|
||||
0x92, 0xe4, 0x1e, 0x14, 0x30, 0x0b, 0x1a, 0x17, 0x0b, 0x4a, 0x8b, 0x91, 0xd4, 0x61, 0x8e, 0x05,
|
||||
0xa5, 0x45, 0x7b, 0x0c, 0x45, 0x9e, 0xc5, 0x82, 0xd2, 0x76, 0x3d, 0xb7, 0x17, 0x87, 0x20, 0xcf,
|
||||
0x62, 0x81, 0x10, 0xc8, 0xbf, 0x74, 0xc5, 0xb9, 0x3e, 0x37, 0x7e, 0x53, 0x07, 0xee, 0x66, 0xf6,
|
||||
0xd7, 0x34, 0x1f, 0x40, 0x91, 0x05, 0xe7, 0x4e, 0x3b, 0xaa, 0x5b, 0x4d, 0x7b, 0x33, 0xcf, 0xb4,
|
||||
0x84, 0xd1, 0xc5, 0xf4, 0xab, 0xa5, 0x1c, 0x2e, 0xa5, 0x0a, 0xba, 0x0a, 0x05, 0x0c, 0xb5, 0x3a,
|
||||
0x65, 0x6a, 0xab, 0x3e, 0xe9, 0x3f, 0x16, 0x94, 0xf7, 0xf8, 0x05, 0xd2, 0x88, 0xc8, 0x63, 0x28,
|
||||
0x75, 0x25, 0xf7, 0xfb, 0x3c, 0xec, 0x23, 0xa8, 0xd2, 0x7a, 0x3d, 0x0d, 0x61, 0x02, 0xdb, 0x32,
|
||||
0x98, 0x8e, 0x2f, 0xc3, 0x31, 0x4b, 0x4c, 0xc8, 0x0e, 0xac, 0xe8, 0x9a, 0x40, 0x0e, 0x95, 0x56,
|
||||
0x73, 0x9e, 0x75, 0x52, 0x36, 0xca, 0xd8, 0x18, 0xac, 0x7d, 0x08, 0xd5, 0x09, 0xb7, 0x8a, 0xeb,
|
||||
0xa9, 0x18, 0x9b, 0x8c, 0x9c, 0x8a, 0xb1, 0x8a, 0xdd, 0x19, 0xf7, 0x46, 0x71, 0x9c, 0xf3, 0x2c,
|
||||
0x16, 0x76, 0x72, 0xef, 0x5b, 0x6b, 0x3b, 0x70, 0x2b, 0xeb, 0xf5, 0x3a, 0xb6, 0xf4, 0x6b, 0x20,
|
||||
0xbb, 0xa1, 0xe0, 0x52, 0x20, 0xbd, 0x3d, 0x11, 0x45, 0xfc, 0x58, 0x2c, 0xce, 0x74, 0x9c, 0xbd,
|
||||
0x5c, 0x36, 0x7b, 0xeb, 0x50, 0x76, 0x22, 0x73, 0x70, 0x1b, 0xeb, 0x32, 0x55, 0xd0, 0x47, 0x40,
|
||||
0xda, 0xc2, 0x13, 0x52, 0xe8, 0xfe, 0x5d, 0xe2, 0x9f, 0x76, 0x0d, 0x97, 0xab, 0xb1, 0x64, 0x03,
|
||||
0xf2, 0xaa, 0x75, 0x91, 0x4a, 0xa5, 0xf5, 0xff, 0x34, 0xd2, 0xc9, 0x9c, 0x60, 0x08, 0xa0, 0xae,
|
||||
0x71, 0xaa, 0xdb, 0xfd, 0x8a, 0x03, 0xce, 0x29, 0x65, 0xb3, 0x95, 0x3d, 0xbd, 0x55, 0x32, 0x40,
|
||||
0xf4, 0x56, 0x1f, 0x99, 0xb3, 0xde, 0x74, 0x2b, 0xfa, 0x95, 0xd6, 0xaa, 0x96, 0xd8, 0x57, 0xab,
|
||||
0xb1, 0x0d, 0x7e, 0x2f, 0x3e, 0xf2, 0x14, 0x0f, 0xe5, 0x5b, 0xf5, 0x50, 0x54, 0xb7, 0x9b, 0xb6,
|
||||
0xf2, 0x8d, 0x02, 0xdd, 0x86, 0x62, 0xb7, 0x77, 0x22, 0x06, 0x9c, 0xbc, 0xa9, 0x0a, 0xb5, 0x2f,
|
||||
0x2e, 0x44, 0xa4, 0xcb, 0xfc, 0xce, 0x54, 0xf8, 0x98, 0x59, 0xa7, 0x3f, 0x59, 0x9a, 0xfd, 0x02,
|
||||
0x46, 0x45, 0xdc, 0x3b, 0xaa, 0xe7, 0x67, 0x26, 0x8e, 0xd2, 0x33, 0xbd, 0x4c, 0x3a, 0x50, 0x73,
|
||||
0xfc, 0xe1, 0x48, 0xb6, 0xc5, 0x37, 0xae, 0xef, 0x4a, 0x37, 0xf0, 0xa3, 0x7a, 0x11, 0x4d, 0x56,
|
||||
0xb3, 0x5b, 0x4f, 0x20, 0xd8, 0x8c, 0x09, 0xfd, 0xc1, 0x82, 0x3b, 0x53, 0xca, 0x2b, 0x78, 0xe5,
|
||||
0x96, 0xf3, 0x7a, 0x2f, 0x19, 0x99, 0x36, 0x02, 0x1b, 0x0b, 0xd9, 0x4c, 0x4e, 0xd0, 0x5f, 0x2d,
|
||||
0xb8, 0x37, 0x0f, 0x30, 0x97, 0x4d, 0x03, 0xe0, 0x79, 0xe8, 0x0e, 0x78, 0x38, 0xfe, 0x54, 0x8c,
|
||||
0xf5, 0xed, 0x91, 0xd1, 0x90, 0x2f, 0xe0, 0xc1, 0x94, 0xaf, 0x8f, 0x7b, 0x71, 0x88, 0x62, 0x52,
|
||||
0x0f, 0x17, 0x92, 0x8a, 0x71, 0x6c, 0x81, 0x39, 0xfd, 0xcb, 0x82, 0xfb, 0x73, 0x97, 0xd2, 0xea,
|
||||
0xb3, 0xb2, 0x85, 0xfe, 0x08, 0x6a, 0x2f, 0xd5, 0x60, 0x68, 0x8b, 0x48, 0xba, 0x3e, 0x57, 0x48,
|
||||
0x5d, 0x9e, 0x33, 0x7a, 0xe2, 0x40, 0x09, 0x75, 0x7b, 0x7c, 0xa8, 0x69, 0xbe, 0x75, 0x05, 0xcd,
|
||||
0x2d, 0x83, 0xd7, 0x73, 0xd3, 0x88, 0x8a, 0x0c, 0xce, 0x71, 0x73, 0x29, 0xa0, 0xa0, 0x26, 0xe2,
|
||||
0x84, 0xc1, 0xb5, 0xa6, 0x5a, 0x00, 0xeb, 0x66, 0x92, 0x4c, 0x30, 0x59, 0xde, 0x93, 0x1f, 0x00,
|
||||
0xa4, 0x50, 0xdd, 0xee, 0x4b, 0xea, 0x33, 0x03, 0xa6, 0xcf, 0x60, 0xdd, 0x8c, 0xb9, 0x6b, 0x6c,
|
||||
0x68, 0xaa, 0x25, 0x97, 0x56, 0x0b, 0xed, 0x80, 0xfd, 0x82, 0x39, 0xea, 0xaa, 0xc3, 0x6e, 0x35,
|
||||
0x29, 0xd2, 0x92, 0x32, 0x79, 0x16, 0x44, 0xd2, 0x98, 0xa8, 0x6f, 0xa5, 0x7b, 0x1e, 0x84, 0x12,
|
||||
0x19, 0x57, 0x19, 0x7e, 0x53, 0x07, 0x6a, 0xfb, 0x41, 0x5f, 0x74, 0x25, 0x97, 0xc9, 0x24, 0x7a,
|
||||
0x88, 0xae, 0xd1, 0x61, 0xa5, 0x55, 0x4d, 0x0f, 0xf6, 0x82, 0x39, 0x0c, 0x37, 0x55, 0x03, 0x5e,
|
||||
0x19, 0x98, 0xa1, 0x84, 0x02, 0xfd, 0xd1, 0x02, 0x30, 0xbe, 0x46, 0xd1, 0xd5, 0x5e, 0xde, 0xcd,
|
||||
0xdc, 0xa9, 0xb3, 0xc3, 0x2a, 0x59, 0x62, 0x99, 0x9b, 0x77, 0xd3, 0xcc, 0x26, 0x1d, 0xf5, 0x5a,
|
||||
0x8a, 0x8f, 0xf5, 0xfa, 0xfc, 0x9c, 0xee, 0x43, 0x75, 0xd7, 0x1b, 0x45, 0x52, 0x84, 0x9a, 0x4e,
|
||||
0xc2, 0xd9, 0xca, 0x70, 0x26, 0x1b, 0xb0, 0x82, 0x94, 0x85, 0xd4, 0x23, 0x60, 0x8a, 0xa8, 0x59,
|
||||
0xa5, 0x5d, 0x28, 0x2c, 0xee, 0x5c, 0x02, 0x79, 0x7c, 0xce, 0xe9, 0x60, 0xe3, 0x4b, 0xae, 0x06,
|
||||
0xf6, 0x9e, 0x1b, 0x57, 0x87, 0xcd, 0xd4, 0x27, 0x6a, 0xf8, 0x05, 0x56, 0xaf, 0xd2, 0x70, 0x75,
|
||||
0x91, 0xdd, 0x8d, 0xab, 0x41, 0x4d, 0xde, 0x9b, 0x5c, 0x39, 0xe6, 0x45, 0x64, 0x67, 0x5e, 0x44,
|
||||
0x3f, 0xe7, 0xe0, 0x2e, 0x13, 0x91, 0xfb, 0x4a, 0x38, 0x7e, 0x24, 0xc3, 0x51, 0xd2, 0xc9, 0x9f,
|
||||
0x04, 0x47, 0x4e, 0x1b, 0xbd, 0xda, 0x2c, 0x16, 0x4c, 0x8e, 0x72, 0x0b, 0x73, 0xf4, 0xb6, 0x7a,
|
||||
0x43, 0x07, 0x61, 0x5f, 0xb5, 0x73, 0x10, 0xea, 0xa8, 0x4f, 0x01, 0xb3, 0x08, 0xf2, 0x0e, 0xac,
|
||||
0x74, 0x83, 0x51, 0xd8, 0x4b, 0x66, 0xfd, 0x83, 0x14, 0x1c, 0xb3, 0x8a, 0x97, 0x99, 0x81, 0x65,
|
||||
0x72, 0x5a, 0x58, 0x9e, 0x53, 0xf2, 0x78, 0x2a, 0xa7, 0xf8, 0xba, 0xad, 0xb4, 0x5e, 0x4b, 0x0d,
|
||||
0x26, 0x96, 0xd9, 0x24, 0x9a, 0x7e, 0x6f, 0xc1, 0xad, 0x2c, 0x85, 0xff, 0x54, 0xe7, 0x71, 0x2a,
|
||||
0x72, 0x73, 0x53, 0x61, 0xcf, 0x4b, 0x45, 0x3e, 0x4d, 0x45, 0xfa, 0x10, 0x2a, 0x64, 0x1e, 0x42,
|
||||
0xf4, 0x04, 0x56, 0x67, 0xf2, 0xb3, 0x1b, 0x0c, 0x86, 0xaa, 0x10, 0x6e, 0x9a, 0xa7, 0x7b, 0x50,
|
||||
0xe8, 0x84, 0xa1, 0xce, 0x50, 0x99, 0xc5, 0x02, 0xfd, 0x12, 0xee, 0x77, 0x85, 0xcc, 0xa4, 0x27,
|
||||
0xd3, 0xe1, 0x07, 0x5e, 0x7f, 0xc1, 0xc9, 0x0f, 0xbc, 0xbe, 0x02, 0xec, 0x8b, 0xf3, 0x05, 0x1b,
|
||||
0xee, 0x8b, 0x73, 0xba, 0x0d, 0xa5, 0xc3, 0x60, 0x18, 0x78, 0xc1, 0xf1, 0x38, 0xdb, 0x44, 0xd6,
|
||||
0xb2, 0x26, 0x7a, 0x52, 0xfb, 0xed, 0xb2, 0x61, 0xfd, 0x7e, 0xd9, 0xb0, 0xfe, 0xb8, 0x6c, 0x58,
|
||||
0xbf, 0xfc, 0xd9, 0xf8, 0xdf, 0x51, 0x11, 0xff, 0xe0, 0xb6, 0xff, 0x0d, 0x00, 0x00, 0xff, 0xff,
|
||||
0xee, 0xaa, 0xfe, 0xbd, 0xd2, 0x0d, 0x00, 0x00,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -156,6 +156,7 @@ message ResizeInstruction {
|
|||
URI Coordinator = 3;
|
||||
repeated ResizeSource Sources = 4;
|
||||
Schema Schema = 5;
|
||||
ClusterStatus ClusterStatus = 6;
|
||||
}
|
||||
|
||||
message ResizeSource {
|
||||
|
|
|
|||
87
server.go
87
server.go
|
|
@ -59,6 +59,9 @@ type Server struct {
|
|||
wg sync.WaitGroup
|
||||
closing chan struct{}
|
||||
|
||||
// Unique name identifying the server.
|
||||
Name string
|
||||
|
||||
// Data storage and HTTP interface.
|
||||
Holder *Holder
|
||||
Handler *Handler
|
||||
|
|
@ -117,31 +120,12 @@ func NewServer() *Server {
|
|||
|
||||
// Open opens and initializes the server.
|
||||
func (s *Server) Open() error {
|
||||
var ln net.Listener
|
||||
var err error
|
||||
|
||||
// If bind URI has the https scheme, enable TLS
|
||||
if s.URI.Scheme() == "https" && s.TLS != nil {
|
||||
ln, err = tls.Listen("tcp", s.URI.HostPort(), s.TLS)
|
||||
if err != nil {
|
||||
s.Logger().Printf("open server")
|
||||
// s.ln can be configured prior to Open() via s.OpenListener().
|
||||
if s.ln == nil {
|
||||
if err := s.OpenListener(); err != nil {
|
||||
return err
|
||||
}
|
||||
} else if s.URI.Scheme() == "http" {
|
||||
// Open HTTP listener to determine port (if specified as :0).
|
||||
ln, err = net.Listen(s.Network, s.URI.HostPort())
|
||||
if err != nil {
|
||||
return fmt.Errorf("net.Listen: %v", err)
|
||||
}
|
||||
} else {
|
||||
return fmt.Errorf("unsupported scheme: %s", s.URI.Scheme())
|
||||
}
|
||||
|
||||
s.ln = ln
|
||||
|
||||
if s.URI.Port() == 0 {
|
||||
// If the port is 0, it is set automatically.
|
||||
// Find out automatically set port and update the host.
|
||||
s.URI.SetPort(uint16(s.ln.Addr().(*net.TCPAddr).Port))
|
||||
}
|
||||
|
||||
// Set Cluster URI.
|
||||
|
|
@ -170,6 +154,9 @@ func (s *Server) Open() error {
|
|||
e.URI = s.URI
|
||||
e.Cluster = s.Cluster
|
||||
e.MaxWritesPerRequest = s.MaxWritesPerRequest
|
||||
|
||||
// Cluster settings.
|
||||
s.Cluster.Broadcaster = s.Broadcaster
|
||||
s.Cluster.MaxWritesPerRequest = s.MaxWritesPerRequest
|
||||
|
||||
// Initialize HTTP handler.
|
||||
|
|
@ -188,7 +175,7 @@ func (s *Server) Open() error {
|
|||
|
||||
// Serve HTTP.
|
||||
go func() {
|
||||
err := http.Serve(ln, s.Handler)
|
||||
err := http.Serve(s.ln, s.Handler)
|
||||
if err != nil {
|
||||
s.Logger().Printf("HTTP handler terminated with error: %s\n", err)
|
||||
}
|
||||
|
|
@ -199,6 +186,11 @@ func (s *Server) Open() error {
|
|||
return fmt.Errorf("starting BroadcastReceiver: %v", err)
|
||||
}
|
||||
|
||||
// If a Coordinator is not specified, then default to s.URI.
|
||||
if s.Cluster.Coordinator.Port() == 0 {
|
||||
s.Cluster.Coordinator = s.URI
|
||||
}
|
||||
|
||||
// Open Cluster management.
|
||||
if err := s.Cluster.Open(); err != nil {
|
||||
return fmt.Errorf("opening Cluster: %v", err)
|
||||
|
|
@ -228,6 +220,48 @@ func (s *Server) Open() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// OpenListener opens a listener for the Server.
|
||||
func (s *Server) OpenListener() error {
|
||||
s.Logger().Printf("open server listener: %s", s.URI)
|
||||
if s.ln != nil {
|
||||
return fmt.Errorf("a listener already exists for server: %s", s.URI)
|
||||
}
|
||||
|
||||
var ln net.Listener
|
||||
var err error
|
||||
|
||||
// If bind URI has the https scheme, enable TLS
|
||||
if s.URI.Scheme() == "https" && s.TLS != nil {
|
||||
ln, err = tls.Listen("tcp", s.URI.HostPort(), s.TLS)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else if s.URI.Scheme() == "http" {
|
||||
// Open HTTP listener to determine port (if specified as :0).
|
||||
ln, err = net.Listen(s.Network, s.URI.HostPort())
|
||||
if err != nil {
|
||||
return fmt.Errorf("net.Listen: %v", err)
|
||||
}
|
||||
} else {
|
||||
return fmt.Errorf("unsupported scheme: %s", s.URI.Scheme())
|
||||
}
|
||||
|
||||
s.ln = ln
|
||||
|
||||
if s.URI.Port() == 0 {
|
||||
// If the port is 0, it is set automatically.
|
||||
// Find out automatically set port and update the host.
|
||||
s.URI.SetPort(uint16(s.ln.Addr().(*net.TCPAddr).Port))
|
||||
}
|
||||
|
||||
// If name is not provided in the config, default to the URI.
|
||||
if s.Name == "" {
|
||||
s.Name = s.URI.String()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Close closes the server and waits for it to shutdown.
|
||||
func (s *Server) Close() error {
|
||||
// Notify goroutines to stop.
|
||||
|
|
@ -382,7 +416,6 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.Cluster.MarkAsJoined()
|
||||
case *internal.ResizeInstruction:
|
||||
err := s.Cluster.FollowResizeInstruction(obj)
|
||||
if err != nil {
|
||||
|
|
@ -409,6 +442,7 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
|
|||
func (s *Server) SendSync(pb proto.Message) error {
|
||||
var eg errgroup.Group
|
||||
for _, node := range s.Cluster.Nodes {
|
||||
s.Logger().Printf("SendSync to: %s", node.URI)
|
||||
// Don't forward the message to ourselves.
|
||||
if s.URI == node.URI {
|
||||
continue
|
||||
|
|
@ -430,7 +464,8 @@ func (s *Server) SendAsync(pb proto.Message) error {
|
|||
|
||||
// SendTo represents an implementation of Broadcaster.
|
||||
func (s *Server) SendTo(to *Node, pb proto.Message) error {
|
||||
ctx := context.WithValue(context.Background(), "uri", to.URI)
|
||||
s.Logger().Printf("SendTo: %s", to.URI)
|
||||
ctx := context.WithValue(context.Background(), "uri", &to.URI)
|
||||
return s.defaultClient.SendMessage(ctx, pb)
|
||||
}
|
||||
|
||||
|
|
|
|||
438
server/cluster_test.go
Normal file
438
server/cluster_test.go
Normal file
|
|
@ -0,0 +1,438 @@
|
|||
// Copyright 2017 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 server_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
"github.com/pilosa/pilosa"
|
||||
"github.com/pilosa/pilosa/gossip"
|
||||
)
|
||||
|
||||
// Ensure program can send/receive broadcast messages.
|
||||
func TestMain_XSendReceiveMessage(t *testing.T) {
|
||||
|
||||
m0 := MustRunMain()
|
||||
defer m0.Close()
|
||||
|
||||
m1 := MustRunMain()
|
||||
defer m1.Close()
|
||||
|
||||
// Update cluster config
|
||||
m0.Server.Cluster.Nodes = []*pilosa.Node{
|
||||
{URI: m0.Server.URI},
|
||||
{URI: m1.Server.URI},
|
||||
}
|
||||
m1.Server.Cluster.Nodes = m0.Server.Cluster.Nodes
|
||||
|
||||
// Configure node0
|
||||
|
||||
// get the host portion of addr to use for binding
|
||||
gossipHost := m0.Server.URI.Host()
|
||||
gossipPort := 0
|
||||
gossipSeed := ""
|
||||
|
||||
m0.Server.Cluster.Coordinator = m0.Server.URI
|
||||
m0.Server.Cluster.Topology = &pilosa.Topology{NodeSet: []pilosa.URI{m0.Server.URI, m1.Server.URI}}
|
||||
m0.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver()
|
||||
gossipMemberSet0, err := gossip.NewGossipMemberSet(m0.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m0.Server, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m0.Server.Cluster.MemberSet = gossipMemberSet0
|
||||
m0.Server.Broadcaster = m0.Server
|
||||
m0.Server.Gossiper = gossipMemberSet0
|
||||
m0.Server.Handler.Broadcaster = m0.Server.Broadcaster
|
||||
m0.Server.Holder.Broadcaster = m0.Server.Broadcaster
|
||||
m0.Server.BroadcastReceiver = gossipMemberSet0
|
||||
|
||||
if err := m0.Server.BroadcastReceiver.Start(m0.Server); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Open Cluster management.
|
||||
if err := m0.Server.Cluster.Open(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Configure node1
|
||||
|
||||
// get the host portion of addr to use for binding
|
||||
gossipHost = m1.Server.URI.Host()
|
||||
gossipPort = 0
|
||||
gossipSeed = gossipMemberSet0.Seed()
|
||||
|
||||
m1.Server.Cluster.Coordinator = m0.Server.URI
|
||||
m1.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver()
|
||||
gossipMemberSet1, err := gossip.NewGossipMemberSet(m1.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m1.Server, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m1.Server.Cluster.MemberSet = gossipMemberSet1
|
||||
m1.Server.Broadcaster = m1.Server
|
||||
m1.Server.Gossiper = gossipMemberSet1
|
||||
m1.Server.Handler.Broadcaster = m1.Server.Broadcaster
|
||||
m1.Server.Holder.Broadcaster = m1.Server.Broadcaster
|
||||
m1.Server.BroadcastReceiver = gossipMemberSet1
|
||||
|
||||
if err := m1.Server.BroadcastReceiver.Start(m1.Server); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Open Cluster management.
|
||||
if err := m1.Server.Cluster.Open(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
|
||||
|
||||
// Expected indexes and Frames
|
||||
expected := map[string][]string{
|
||||
"i": []string{"f"},
|
||||
}
|
||||
|
||||
// Create a client for each node.
|
||||
client0 := m0.Client()
|
||||
client1 := m1.Client()
|
||||
|
||||
// Create indexes and frames on one node.
|
||||
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
|
||||
t.Fatal(err)
|
||||
} else if err := client0.CreateFrame(context.Background(), "i", "f", pilosa.FrameOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Make sure node0 knows about the index and frame created.
|
||||
schema0, err := client0.Schema(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
received0 := map[string][]string{}
|
||||
for _, idx := range schema0 {
|
||||
received0[idx.Name] = []string{}
|
||||
for _, frame := range idx.Frames {
|
||||
received0[idx.Name] = append(received0[idx.Name], frame.Name)
|
||||
}
|
||||
}
|
||||
if !reflect.DeepEqual(received0, expected) {
|
||||
t.Fatalf("unexpected schema on node0: %s", received0)
|
||||
}
|
||||
|
||||
// Make sure node1 knows about the index and frame created.
|
||||
schema1, err := client1.Schema(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
received1 := map[string][]string{}
|
||||
for _, idx := range schema1 {
|
||||
received1[idx.Name] = []string{}
|
||||
for _, frame := range idx.Frames {
|
||||
received1[idx.Name] = append(received1[idx.Name], frame.Name)
|
||||
}
|
||||
}
|
||||
if !reflect.DeepEqual(received1, expected) {
|
||||
t.Fatalf("unexpected schema on node1: %s", received1)
|
||||
}
|
||||
|
||||
// Write data on first node.
|
||||
if _, err := m0.Query("i", "", `
|
||||
SetBit(rowID=1, frame="f", columnID=1)
|
||||
SetBit(rowID=1, frame="f", columnID=2400000)
|
||||
`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// We have to wait for the broadcast message to be sent before checking state.
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
// Make sure node0 knows about the latest MaxSlice.
|
||||
maxSlices0, err := client0.MaxSliceByIndex(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if maxSlices0["i"] != 2 {
|
||||
t.Fatalf("unexpected maxSlice on node0: %d", maxSlices0["i"])
|
||||
}
|
||||
|
||||
// Make sure node1 knows about the latest MaxSlice.
|
||||
maxSlices1, err := client1.MaxSliceByIndex(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if maxSlices1["i"] != 2 {
|
||||
t.Fatalf("unexpected maxSlice on node1: %d", maxSlices1["i"])
|
||||
}
|
||||
|
||||
// Write input definition to the first node.
|
||||
if _, err := m0.CreateDefinition("i", "test", `{
|
||||
"frames": [{"name": "event-time",
|
||||
"options": {
|
||||
"cacheType": "ranked",
|
||||
"timeQuantum": "YMD"
|
||||
}}],
|
||||
"fields": [{"name": "columnID",
|
||||
"primaryKey": true
|
||||
}]}
|
||||
`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// We have to wait for the broadcast message to be sent before checking state.
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
frame0 := m0.Server.Holder.Frame("i", "event-time")
|
||||
if frame0 == nil {
|
||||
t.Fatal("frame not found")
|
||||
}
|
||||
frame1 := m1.Server.Holder.Frame("i", "event-time")
|
||||
if frame1 == nil {
|
||||
t.Fatal("frame not found")
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure that an empty node comes up in a NORMAL state.
|
||||
func TestClusterResize_EmptyNode(t *testing.T) {
|
||||
m0 := MustRunMain()
|
||||
defer m0.Close()
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected cluster state: %s", m0.Server.Cluster.State)
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure that a cluster of empty nodes comes up in a NORMAL state.
|
||||
func TestClusterResize_EmptyNodes(t *testing.T) {
|
||||
// Configure node0
|
||||
m0 := NewMain()
|
||||
defer m0.Close()
|
||||
|
||||
gossipHost := "localhost"
|
||||
gossipPort := 0
|
||||
seed, coord, err := m0.RunWithTransport(gossipHost, gossipPort, "", nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Configure node1
|
||||
m1 := NewMain()
|
||||
defer m1.Close()
|
||||
|
||||
seed, coord, err = m1.RunWithTransport(gossipHost, gossipPort, seed, &coord)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
|
||||
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure that adding a node correctly resizes the cluster.
|
||||
func TestClusterResize_AddNode(t *testing.T) {
|
||||
t.Run("NoData", func(t *testing.T) {
|
||||
// Configure node0
|
||||
m0 := NewMain()
|
||||
defer m0.Close()
|
||||
|
||||
seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Configure node1
|
||||
m1 := NewMain()
|
||||
defer m1.Close()
|
||||
|
||||
var eg errgroup.Group
|
||||
eg.Go(func() error {
|
||||
_, _, err = m1.RunWithTransport("localhost", 0, seed, &coord)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err := eg.Wait(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
|
||||
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
|
||||
}
|
||||
})
|
||||
t.Run("WithIndex", func(t *testing.T) {
|
||||
// Configure node0
|
||||
m0 := NewMain()
|
||||
defer m0.Close()
|
||||
|
||||
seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Create a client for each node.
|
||||
client0 := m0.Client()
|
||||
|
||||
// Create indexes and frames on one node.
|
||||
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
|
||||
t.Fatal(err)
|
||||
} else if err := client0.CreateFrame(context.Background(), "i", "f", pilosa.FrameOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Configure node1
|
||||
m1 := NewMain()
|
||||
defer m1.Close()
|
||||
|
||||
var eg errgroup.Group
|
||||
eg.Go(func() error {
|
||||
_, _, err = m1.RunWithTransport("localhost", 0, seed, &coord)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err := eg.Wait(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Give the cluster time to settle.
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
|
||||
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
|
||||
}
|
||||
})
|
||||
t.Run("ContinuousSlices", func(t *testing.T) {
|
||||
|
||||
// Configure node0
|
||||
m0 := NewMain()
|
||||
defer m0.Close()
|
||||
|
||||
seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Create a client for each node.
|
||||
client0 := m0.Client()
|
||||
//client1 := m1.Client()
|
||||
|
||||
// Create indexes and frames on one node.
|
||||
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
|
||||
t.Fatal(err)
|
||||
} else if err := client0.CreateFrame(context.Background(), "i", "f", pilosa.FrameOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Write data on first node.
|
||||
if _, err := m0.Query("i", "", `
|
||||
SetBit(rowID=1, frame="f", columnID=1)
|
||||
SetBit(rowID=1, frame="f", columnID=1300000)
|
||||
`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Configure node1
|
||||
m1 := NewMain()
|
||||
defer m1.Close()
|
||||
|
||||
var eg errgroup.Group
|
||||
eg.Go(func() error {
|
||||
_, _, err = m1.RunWithTransport("localhost", 0, seed, &coord)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err := eg.Wait(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Give the cluster time to settle.
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
|
||||
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
|
||||
}
|
||||
})
|
||||
t.Run("SkippedSlice", func(t *testing.T) {
|
||||
|
||||
// Configure node0
|
||||
m0 := NewMain()
|
||||
defer m0.Close()
|
||||
|
||||
seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Create a client for each node.
|
||||
client0 := m0.Client()
|
||||
//client1 := m1.Client()
|
||||
|
||||
// Create indexes and frames on one node.
|
||||
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
|
||||
t.Fatal(err)
|
||||
} else if err := client0.CreateFrame(context.Background(), "i", "f", pilosa.FrameOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Write data on first node. Note that no data is placed on slice 1.
|
||||
if _, err := m0.Query("i", "", `
|
||||
SetBit(rowID=1, frame="f", columnID=1)
|
||||
SetBit(rowID=1, frame="f", columnID=2400000)
|
||||
`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Configure node1
|
||||
m1 := NewMain()
|
||||
defer m1.Close()
|
||||
|
||||
var eg errgroup.Group
|
||||
eg.Go(func() error {
|
||||
_, _, err = m1.RunWithTransport("localhost", 0, seed, &coord)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err := eg.Wait(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Give the cluster time to settle.
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
|
||||
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
|
@ -60,6 +60,9 @@ type Command struct {
|
|||
CPUProfile string
|
||||
CPUTime time.Duration
|
||||
|
||||
// Gossip transport
|
||||
GossipTransport *gossip.Transport
|
||||
|
||||
// Standard input/output
|
||||
*pilosa.CmdIO
|
||||
|
||||
|
|
@ -100,6 +103,12 @@ func (m *Command) Run(args ...string) (err error) {
|
|||
return err
|
||||
}
|
||||
|
||||
// SetupNetworking
|
||||
err = m.SetupNetworking()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Initialize server.
|
||||
if err = m.Server.Open(); err != nil {
|
||||
return fmt.Errorf("server.Open: %v", err)
|
||||
|
|
@ -123,6 +132,11 @@ func (m *Command) SetupServer() error {
|
|||
}
|
||||
m.Server.URI = *uri
|
||||
|
||||
// If using a dynamically allocated port, server.Name will get set later.
|
||||
if m.Config.Bind != "localhost:0" {
|
||||
m.Server.Name = m.Server.URI.String()
|
||||
}
|
||||
|
||||
cluster := pilosa.NewCluster()
|
||||
cluster.ReplicaN = m.Config.Cluster.ReplicaN
|
||||
cluster.Holder = m.Server.Holder
|
||||
|
|
@ -183,24 +197,41 @@ func (m *Command) SetupServer() error {
|
|||
m.Server.Handler.RemoteClient = c
|
||||
m.Server.Cluster.RemoteClient = c
|
||||
|
||||
// Default coordintor to port 0 when not specified so that coordinator
|
||||
// can be set to the value of server.URI after server binds to a port.
|
||||
// This would only be useful in a one-node cluster.
|
||||
coord := m.Config.Cluster.Coordinator
|
||||
if coord == "" {
|
||||
coord = ":0"
|
||||
}
|
||||
|
||||
// Set the coordinator node.
|
||||
curi, err := pilosa.AddressWithDefaults(m.Config.Cluster.Coordinator)
|
||||
curi, err := pilosa.AddressWithDefaults(coord)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.Server.Cluster.Coordinator = *curi
|
||||
|
||||
// Set internal port (string).
|
||||
gossipPortStr := pilosa.DefaultGossipPort
|
||||
// Config.GossipPort is deprecated, so Config.Gossip.Port has priority
|
||||
if m.Config.Gossip.Port != "" {
|
||||
gossipPortStr = m.Config.Gossip.Port
|
||||
} else if m.Config.GossipPort != "" {
|
||||
gossipPortStr = m.Config.GossipPort
|
||||
}
|
||||
// Set configuration options.
|
||||
m.Server.AntiEntropyInterval = time.Duration(m.Config.AntiEntropy.Interval)
|
||||
m.Server.Cluster.LongQueryTime = time.Duration(m.Config.Cluster.LongQueryTime)
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetupNetworking sets up internode communication based on the configuration.
|
||||
func (m *Command) SetupNetworking() error {
|
||||
switch m.Config.Cluster.Type {
|
||||
case pilosa.ClusterGossip:
|
||||
|
||||
// Set internal port (string).
|
||||
gossipPortStr := pilosa.DefaultGossipPort
|
||||
// Config.GossipPort is deprecated, so Config.Gossip.Port has priority
|
||||
if m.Config.Gossip.Port != "" {
|
||||
gossipPortStr = m.Config.Gossip.Port
|
||||
} else if m.Config.GossipPort != "" {
|
||||
gossipPortStr = m.Config.GossipPort
|
||||
}
|
||||
|
||||
gossipPort, err := strconv.Atoi(gossipPortStr)
|
||||
if err != nil {
|
||||
return err
|
||||
|
|
@ -222,9 +253,22 @@ func (m *Command) SetupServer() error {
|
|||
}
|
||||
|
||||
// get the host portion of addr to use for binding
|
||||
gossipHost := uri.Host()
|
||||
gossipHost := m.Server.URI.Host()
|
||||
var transport *gossip.Transport
|
||||
if m.GossipTransport != nil {
|
||||
transport = m.GossipTransport
|
||||
} else {
|
||||
transport, err = gossip.NewTransport(gossipHost, gossipPort)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
m.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver()
|
||||
gossipMemberSet, err := gossip.NewGossipMemberSet(uri.String(), gossipHost, gossipPort, gossipSeed, m.Server, gossipKey)
|
||||
if m.Server.Name == "" {
|
||||
return fmt.Errorf("must provide a valid name for gossip membership")
|
||||
}
|
||||
gossipMemberSet, err := gossip.NewGossipMemberSetWithTransport(m.Server.Name, gossipHost, transport, gossipSeed, m.Server, gossipKey)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -233,14 +277,13 @@ func (m *Command) SetupServer() error {
|
|||
m.Server.BroadcastReceiver = gossipMemberSet
|
||||
m.Server.Gossiper = gossipMemberSet
|
||||
case pilosa.ClusterStatic, pilosa.ClusterNone:
|
||||
|
||||
m.Server.Cluster.Static = true
|
||||
for _, address := range m.Config.Cluster.Hosts {
|
||||
uri, err := pilosa.NewURIFromAddress(address)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cluster.Nodes = append(cluster.Nodes, &pilosa.Node{
|
||||
m.Server.Cluster.Nodes = append(m.Server.Cluster.Nodes, &pilosa.Node{
|
||||
URI: *uri,
|
||||
})
|
||||
}
|
||||
|
|
@ -256,13 +299,6 @@ func (m *Command) SetupServer() error {
|
|||
default:
|
||||
return fmt.Errorf("'%v' is not a supported value for broadcaster type", m.Config.Cluster.Type)
|
||||
}
|
||||
|
||||
// Cluster management needs.
|
||||
m.Server.Cluster.Broadcaster = m.Server.Broadcaster
|
||||
|
||||
// Set configuration options.
|
||||
m.Server.AntiEntropyInterval = time.Duration(m.Config.AntiEntropy.Interval)
|
||||
m.Server.Cluster.LongQueryTime = time.Duration(m.Config.Cluster.LongQueryTime)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -636,6 +636,81 @@ func (m *Main) Reopen() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// RunWithTransport runs Main and returns the dynamically allocated gossip port.
|
||||
func (m *Main) RunWithTransport(host string, bindPort int, joinSeed string, coordinator *pilosa.URI) (seed string, coord pilosa.URI, err error) {
|
||||
defer close(m.Started)
|
||||
|
||||
m.Config.Cluster.Type = "gossip"
|
||||
|
||||
/*
|
||||
TEST:
|
||||
- SetupServer (just static settings from config)
|
||||
- OpenListener (sets Server.Name to use in gossip)
|
||||
- NewTransport (gossip)
|
||||
- SetupNetworking (does the gossip or static stuff) - uses Server.Name
|
||||
- Open server
|
||||
|
||||
PRODUCTION:
|
||||
- SetupServer (just static settings from config)
|
||||
- SetupNetworking (does the gossip or static stuff) - calls NewTransport
|
||||
- Open server - calls OpenListener
|
||||
*/
|
||||
|
||||
// SetupServer
|
||||
err = m.SetupServer()
|
||||
if err != nil {
|
||||
return seed, coord, err
|
||||
}
|
||||
|
||||
// Open server listener.
|
||||
// This is used to set Server.Name, which is used as the node
|
||||
// name for identifying a memberlist node.
|
||||
err = m.Server.OpenListener()
|
||||
if err != nil {
|
||||
return seed, coord, err
|
||||
}
|
||||
|
||||
// Open gossip transport to use in SetupServer.
|
||||
transport, err := gossip.NewTransport(host, bindPort)
|
||||
if err != nil {
|
||||
return seed, coord, err
|
||||
}
|
||||
m.GossipTransport = transport
|
||||
|
||||
if joinSeed != "" {
|
||||
m.Config.Gossip.Seed = joinSeed
|
||||
} else {
|
||||
m.Config.Gossip.Seed = transport.URI.String()
|
||||
}
|
||||
seed = m.Config.Gossip.Seed
|
||||
|
||||
// SetupNetworking
|
||||
err = m.SetupNetworking()
|
||||
if err != nil {
|
||||
return seed, coord, err
|
||||
}
|
||||
|
||||
if err = m.Server.BroadcastReceiver.Start(m.Server); err != nil {
|
||||
return seed, coord, err
|
||||
}
|
||||
|
||||
if coordinator != nil {
|
||||
coord = *coordinator
|
||||
} else {
|
||||
coord = m.Server.URI
|
||||
}
|
||||
m.Server.Cluster.Coordinator = coord
|
||||
m.Server.Cluster.Static = false
|
||||
|
||||
// Initialize server.
|
||||
err = m.Server.Open()
|
||||
if err != nil {
|
||||
return seed, coord, err
|
||||
}
|
||||
|
||||
return seed, coord, nil
|
||||
}
|
||||
|
||||
// URL returns the base URL string for accessing the running program.
|
||||
func (m *Main) URL() string { return "http://" + m.Server.Addr().String() }
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue