diff --git a/cluster.go b/cluster.go index 1fddefcf3..132c94ec7 100644 --- a/cluster.go +++ b/cluster.go @@ -159,6 +159,9 @@ type Cluster struct { // Threshold for logging long-running queries LongQueryTime time.Duration + // Maximum number of SetBit() or ClearBit() commands per request. + MaxWritesPerRequest int + // EventReceiver receives NodeEvents pertaining to node membership. EventReceiver EventReceiver @@ -199,10 +202,11 @@ type Cluster struct { // NewCluster returns a new instance of Cluster with defaults. func NewCluster() *Cluster { return &Cluster{ - Hasher: &jmphasher{}, - PartitionN: DefaultPartitionN, - ReplicaN: DefaultReplicaN, - EventReceiver: NopEventReceiver, + Hasher: &jmphasher{}, + PartitionN: DefaultPartitionN, + ReplicaN: DefaultReplicaN, + MaxWritesPerRequest: DefaultMaxWritesPerRequest, + EventReceiver: NopEventReceiver, joiningLeavingNodes: make(chan nodeAction, 10), // buffered channel jobs: make(map[int64]*ResizeJob), diff --git a/docs/configuration.md b/docs/configuration.md index 4fe15d855..d9bfd09bf 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -99,7 +99,7 @@ Any flag that has a value that is a comma separated list on the command line bec #### Gossip Port -* Description: Port to which Pilosa should bind for internal communication. +* Description: Port to which Pilosa should bind for internal communication. If more than one Pilosa server is running on the same host, the gossip port for each server must be unique. * Flag: `--gossip.port=11101` * Env: `PILOSA_GOSSIP_PORT=11101` * Config: @@ -303,7 +303,7 @@ Any flag that has a value that is a comma separated list on the command line bec ### Example Cluster Configuration -A three node cluster could be minimally configured as follows: +A three node cluster running on different hosts could be minimally configured as follows: #### Node 0 @@ -408,3 +408,64 @@ The same cluster which uses HTTPS instead of HTTP can be configured as follows. [tls] certificate = "/home/pilosa/private/server.crt" key = "/home/pilosa/private/server.key" + +### Example Cluster Configuration (HTTPS, same host) + +You can run a cluster on the same host using the configuration above with a few changes. Gossip port and bind adress should be different for each node and a data directory should be accessed only by a single node. + +#### Node 0 + + data-dir = "/home/pilosa/data0" + bind = "https://localhost:10100" + + [gossip] + port = 12000 + seed = "localhost:12000" + key = "/home/pilosa/private/gossip.key32" + + [cluster] + replicas = 1 + type = "gossip" + hosts = ["https://localhost:10100","https://localhost:10101","https://localhost:10102"] + + [tls] + certificate = "/home/pilosa/private/server.crt" + key = "/home/pilosa/private/server.key" + +#### Node 1 + + data-dir = "/home/pilosa/data1" + bind = "https://localhost:10101" + + [gossip] + port = 12001 + seed = "localhost:12000" + key = "/home/pilosa/private/gossip.key32" + + [cluster] + replicas = 1 + type = "gossip" + hosts = ["https://localhost:10100","https://localhost:10101","https://localhost:10102"] + + [tls] + certificate = "/home/pilosa/private/server.crt" + key = "/home/pilosa/private/server.key" + +#### Node 2 + + data-dir = "/home/pilosa/data2" + bind = "https://localhost:10102" + + [gossip] + port = 12002 + seed = "locahost:12000" + key = "/home/pilosa/private/gossip.key32" + + [cluster] + replicas = 1 + type = "gossip" + hosts = ["https://localhost:10100","https://localhost:10101","https://localhost:10102"] + + [tls] + certificate = "/home/pilosa/private/server.crt" + key = "/home/pilosa/private/server.key" diff --git a/docs/input-definition.md b/docs/input-definition.md index 494f4757e..7d50c6f9d 100644 --- a/docs/input-definition.md +++ b/docs/input-definition.md @@ -20,70 +20,14 @@ Before creating a schema, let's create the repository index first: ``` curl localhost:10101/index/repository -X POST ``` -Then we can send the following input definition as JSON to Pilosa. The sample input defintion schema for the "Star Trace" project is at [Pilosa Getting Started repository](https://github.com/pilosa/getting-started), `input-definition.json` file - +The sample input definition schema for the "Star Trace" project is at [Pilosa Getting Started repository](https://github.com/pilosa/getting-started) in the `input_definition.json` file. Download it using: ``` -curl localhost:10101/index/repository/input-definition/stargazer \ - -X POST \ - -d '{ - "frames": [ - { - "name": "language", - "options": { - "inverseEnabled": true, - "timeQuantum": "YMD" - } - }, - { - "name": "stargazer", - "options": { - "inverseEnabled": true, - "timeQuantum": "YMD" - } - } - ], - "fields": [ - { - "name": "repo_id", - "primaryKey": true - }, - { - "actions": [ - { - "frame": "language", - "valueDestination": "mapping", - "valueMap": { - "C": 7, - "C#": 27, - "Go": 5, - "Java": 21, - "JavaScript": 13, - "Python": 17 - } - } - ], - "name": "language_id" - }, - { - "actions": [ - { - "frame": "stargazer", - "valueDestination": "value-to-row" - } - ], - "name": "stargazer_id" - }, - { - "actions": [ - { - "frame": "stargazer", - "valueDestination": "set-timestamp" - } - ], - "name": "time_value" - } - ] - }' +curl -OL https://github.com/pilosa/getting-started/raw/master/input_definition.json +``` + +Run the following to create the input definition: +``` +curl localhost:10101/index/repository/input-definition/stargazer -d @input_definition.json ``` Instead of creating a `stargazer` frame and a `language` frame individually like in [Getting Started](../getting-started/), we can create multiple frames in one input definition. @@ -96,25 +40,14 @@ We can also set `repo_id` for multiple frames at the same time by providing fiel ### Import Data -The sample data for the "Star Trace" project is at [Pilosa Getting Started repository](https://github.com/pilosa/getting-started). - -If you import data using an input definition, download the `json-input.json` file in that repo, then run the following request using the input definition created above: - +The sample data for the input definition we created above is in the `json_input.json` file at [Pilosa Getting Started repository](https://github.com/pilosa/getting-started). Download it using: ``` -curl localhost:10101/index/repository/input/stargazer \ - -X POST \ - -d '[ - { - "repo_id": 91720568, - "language_id": "Go", - "stargazer_id": 513114, - "time_value": "2017-05-18T20:40" - }, - { - "language_id": "Python", - "repo_id": 95122322 - } - ]' +curl -OL https://github.com/pilosa/getting-started/raw/master/json_input.json +``` + +Then run the following to import it: +``` +curl localhost:10101/index/repository/input/stargazer -d @json_input.json ``` As defined in the input definition, field name `language_id` maps language to a corresponding id defined in `valueMap` and sets the appropriate bit in the `language` frame. The value corresponding to field name `stargazer_id` is added to the `stargazer` frame as rowID. diff --git a/executor.go b/executor.go index 322fc6aaf..3460d04df 100644 --- a/executor.go +++ b/executor.go @@ -800,6 +800,12 @@ func (e *Executor) executeFieldRangeSlice(ctx context.Context, index string, c * return NewBitmap(), nil } + // LT[E] and GT[E] should return all not-null if selected range fully encompases valid field range. + if (cond.Op == pql.LT && value > field.Max) || (cond.Op == pql.LTE && value >= field.Max) || + (cond.Op == pql.GT && value < field.Min) || (cond.Op == pql.GTE && value <= field.Min) { + return frag.FieldNotNull(field.BitDepth()) + } + // outOfRange for NEQ should return all not-null. if outOfRange && cond.Op == pql.NEQ { return frag.FieldNotNull(field.BitDepth()) diff --git a/executor_test.go b/executor_test.go index 67cd2c041..c98239137 100644 --- a/executor_test.go +++ b/executor_test.go @@ -741,6 +741,15 @@ func TestExecutor_Execute_FieldRange(t *testing.T) { t.Fatal(err) } + if _, err := idx.CreateFrame("edge", pilosa.FrameOptions{ + RangeEnabled: true, + Fields: []*pilosa.Field{ + {Name: "foo", Type: pilosa.FieldTypeInt, Min: -100, Max: 100}, + }, + }); err != nil { + t.Fatal(err) + } + if _, err := e.Execute(context.Background(), "i", test.MustParse(` SetBit(frame=f, rowID=0, columnID=0) SetBit(frame=f, rowID=0, columnID=`+strconv.Itoa(SliceWidth+1)+`) @@ -751,6 +760,8 @@ func TestExecutor_Execute_FieldRange(t *testing.T) { SetFieldValue(frame=f, foo=20, columnID=`+strconv.Itoa((5*SliceWidth)+100)+`) SetFieldValue(frame=f, foo=60, columnID=`+strconv.Itoa(SliceWidth+1)+`) SetFieldValue(frame=other, foo=1000, columnID=0) + SetFieldValue(frame=edge, foo=100, columnID=0) + SetFieldValue(frame=edge, foo=-100, columnID=1) `), nil, nil); err != nil { t.Fatal(err) } @@ -850,6 +861,22 @@ func TestExecutor_Execute_FieldRange(t *testing.T) { } }) + t.Run("LTAboveMax", func(t *testing.T) { + if result, err := e.Execute(context.Background(), "i", test.MustParse(`Range(frame=edge, foo < 200)`), nil, nil); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual([]uint64{0, 1}, result[0].(*pilosa.Bitmap).Bits()) { + t.Fatalf("unexpected result: %s", spew.Sdump(result[0].(*pilosa.Bitmap).Bits())) + } + }) + + t.Run("GTBelowMin", func(t *testing.T) { + if result, err := e.Execute(context.Background(), "i", test.MustParse(`Range(frame=edge, foo > -200)`), nil, nil); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual([]uint64{0, 1}, result[0].(*pilosa.Bitmap).Bits()) { + t.Fatalf("unexpected result: %s", spew.Sdump(result[0].(*pilosa.Bitmap).Bits())) + } + }) + t.Run("ErrFrameNotFound", func(t *testing.T) { if _, err := e.Execute(context.Background(), "i", test.MustParse(`Range(frame=bad_frame, foo >= 20)`), nil, nil); err != pilosa.ErrFrameNotFound { t.Fatal(err) diff --git a/fragment.go b/fragment.go index e2d964ded..3df788629 100644 --- a/fragment.go +++ b/fragment.go @@ -494,7 +494,7 @@ func (f *Fragment) FieldValue(columnID uint64, bitDepth uint) (value uint64, exi f.mu.Lock() defer f.mu.Unlock() - // If existance bit is unset then ignore remaining bits. + // If existence bit is unset then ignore remaining bits. if v, err := f.bit(uint64(bitDepth), columnID); err != nil { return 0, false, err } else if !v { @@ -588,7 +588,7 @@ func (f *Fragment) importSetFieldValue(columnID uint64, bitDepth uint, value uin // FieldSum returns the sum of a given field as well as the number of columns involved. // A bitmap can be passed in to optionally filter the computed columns. func (f *Fragment) FieldSum(filter *Bitmap, bitDepth uint) (sum, count uint64, err error) { - // Compute count based on the existance bit. + // Compute count based on the existence bit. row := f.Row(uint64(bitDepth)) if filter != nil { count = row.IntersectionCount(filter) @@ -617,6 +617,7 @@ func (f *Fragment) FieldSum(filter *Bitmap, bitDepth uint) (sum, count uint64, e return sum, count, nil } +// FieldRange returns bitmaps with a field value encoding matching the predicate. func (f *Fragment) FieldRange(op pql.Token, bitDepth uint, predicate uint64) (*Bitmap, error) { switch op { case pql.EQ: @@ -755,6 +756,7 @@ func (f *Fragment) FieldNotNull(bitDepth uint) (*Bitmap, error) { return f.Row(uint64(bitDepth)), nil } +// FieldRangeBetween returns bitmaps with a field value encoding matching any value between predicateMin and predicateMax. func (f *Fragment) FieldRangeBetween(bitDepth uint, predicateMin, predicateMax uint64) (*Bitmap, error) { b := f.Row(uint64(bitDepth)) keep1 := NewBitmap() // GTE @@ -1825,36 +1827,43 @@ func (s *FragmentSyncer) syncBlock(id int) error { // Write updates to remote blocks. for i := 0; i < len(clients); i++ { set, clear := sets[i], clears[i] + count := 0 // Ignore if there are no differences. if len(set.ColumnIDs) == 0 && len(clear.ColumnIDs) == 0 { continue } - // Generate query with sets & clears. - var buf bytes.Buffer + // Generate query with sets & clears, and group the requests to not exceed MaxWritesPerRequest. + total := len(set.ColumnIDs) + len(clear.ColumnIDs) + buffers := make([]bytes.Buffer, int(math.Ceil(float64(total)/float64(s.Cluster.MaxWritesPerRequest)))) // Only sync the standard block. for j := 0; j < len(set.ColumnIDs); j++ { - fmt.Fprintf(&buf, "SetBit(frame=%q, rowID=%d, columnID=%d)\n", f.Frame(), set.RowIDs[j], (f.Slice()*SliceWidth)+set.ColumnIDs[j]) + fmt.Fprintf(&(buffers[count/s.Cluster.MaxWritesPerRequest]), "SetBit(frame=%q, rowID=%d, columnID=%d)\n", f.Frame(), set.RowIDs[j], (f.Slice()*SliceWidth)+set.ColumnIDs[j]) + count++ } for j := 0; j < len(clear.ColumnIDs); j++ { - fmt.Fprintf(&buf, "ClearBit(frame=%q, rowID=%d, columnID=%d)\n", f.Frame(), clear.RowIDs[j], (f.Slice()*SliceWidth)+clear.ColumnIDs[j]) + fmt.Fprintf(&(buffers[count/s.Cluster.MaxWritesPerRequest]), "ClearBit(frame=%q, rowID=%d, columnID=%d)\n", f.Frame(), clear.RowIDs[j], (f.Slice()*SliceWidth)+clear.ColumnIDs[j]) + count++ } - // Verify sync is not prematurely closing. - if s.isClosing() { - return nil - } + // Iterate over the buffers. + for k := 0; k < len(buffers); k++ { + // Verify sync is not prematurely closing. + if s.isClosing() { + return nil + } - // Execute query. - queryRequest := &internal.QueryRequest{ - Query: buf.String(), - Remote: true, - } - _, err := clients[i].ExecuteQuery(context.Background(), f.Index(), queryRequest) - if err != nil { - return err + // Execute query. + queryRequest := &internal.QueryRequest{ + Query: buffers[k].String(), + Remote: true, + } + _, err := clients[i].ExecuteQuery(context.Background(), f.Index(), queryRequest) + if err != nil { + return err + } } } diff --git a/frame.go b/frame.go index 2e18a784d..03a533dbb 100644 --- a/frame.go +++ b/frame.go @@ -1087,7 +1087,7 @@ func (f *Field) BitDepth() uint { // BaseValue adjusts the value to align with the range for Field for a certain // operation type. -// TODO: there is an edge case for GT and LT where this returns a baseValue +// Note: There is an edge case for GT and LT where this returns a baseValue // that does not fully encompass the range. // ex: Field.Min = 0, Field.Max = 1023 // BaseValue(LT, 2000) returns 1023, which will perform "LT 1023" and effectively @@ -1095,6 +1095,8 @@ func (f *Field) BitDepth() uint { // Note that in this case (because the range uses the full BitDepth 0 to 1023), // we can't simply return 1024. // In order to make this work, we effectively need to change the operator to LTE. +// Executor.executeFieldRangeSlice() takes this into account and returns +// `frag.FieldNotNull(field.BitDepth())` in such instances. func (f *Field) BaseValue(op pql.Token, value int64) (baseValue uint64, outOfRange bool) { if op == pql.GT || op == pql.GTE { if value > f.Max { diff --git a/gossip/gossip.go b/gossip/gossip.go index 7e89e17c1..5300d2e1e 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -18,6 +18,8 @@ import ( "fmt" "io" "log" + "os" + "strings" "time" "golang.org/x/sync/errgroup" @@ -61,6 +63,11 @@ func (g *GossipMemberSet) Start(h pilosa.BroadcastHandler) error { return nil } +// Seed returns the gossipSeed determined by the config. +func (g *GossipMemberSet) Seed() string { + return g.config.gossipSeed +} + // Open implements the MemberSet interface to start network activity. func (g *GossipMemberSet) Open() error { if g.handler == nil { @@ -132,20 +139,90 @@ 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.") + } + + 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 { +func NewGossipMemberSet(name string, gossipHost string, gossipPort int, gossipSeed string, server *pilosa.Server, secretKey []byte) (*GossipMemberSet, error) { g := &GossipMemberSet{ LogOutput: server.LogOutput, } + conf := memberlist.DefaultLocalConfig() + conf.BindPort = gossipPort + conf.AdvertisePort = gossipPort + //TODO: pull memberlist config from pilosa.cfg file g.config = &gossipConfig{ - memberlistConfig: memberlist.DefaultLocalConfig(), + memberlistConfig: conf, gossipSeed: gossipSeed, } + g.config.memberlistConfig.Name = name g.config.memberlistConfig.BindAddr = gossipHost - g.config.memberlistConfig.BindPort = gossipPort g.config.memberlistConfig.AdvertiseAddr = pilosa.HostToIP(gossipHost) g.config.memberlistConfig.AdvertisePort = gossipPort //g.config.memberlistConfig.PushPullInterval = 0 * time.Second // Default is 15s in DefaultLocalConfig. @@ -155,7 +232,19 @@ func NewGossipMemberSet(name string, gossipHost string, gossipPort int, gossipSe g.statusHandler = server - return g + // 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) + } + + return g, nil } // SendSync implementation of the Broadcaster interface. diff --git a/server.go b/server.go index bbb02e7c3..038ceab4b 100644 --- a/server.go +++ b/server.go @@ -161,6 +161,7 @@ func (s *Server) Open() error { e.URI = s.URI e.Cluster = s.Cluster e.MaxWritesPerRequest = s.MaxWritesPerRequest + s.Cluster.MaxWritesPerRequest = s.MaxWritesPerRequest // Initialize HTTP handler. s.Handler.Broadcaster = s.Broadcaster diff --git a/server/server.go b/server/server.go index e7396dfb1..bd51631f6 100644 --- a/server/server.go +++ b/server/server.go @@ -224,7 +224,10 @@ func (m *Command) SetupServer() error { // get the host portion of addr to use for binding gossipHost := uri.Host() m.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver() - gossipMemberSet := gossip.NewGossipMemberSet(uri.String(), gossipHost, gossipPort, gossipSeed, m.Server, gossipKey) + gossipMemberSet, err := gossip.NewGossipMemberSet(uri.String(), gossipHost, gossipPort, gossipSeed, m.Server, gossipKey) + if err != nil { + return err + } m.Server.Cluster.MemberSet = gossipMemberSet m.Server.Broadcaster = gossipMemberSet m.Server.BroadcastReceiver = gossipMemberSet diff --git a/server/server_test.go b/server/server_test.go index c0837a16a..d4d346716 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -22,19 +22,19 @@ import ( "io" "io/ioutil" "math/rand" - "net" "net/http" "os" "reflect" "runtime" "sort" - "strconv" "strings" "testing" "testing/quick" + "time" "github.com/BurntSushi/toml" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/gossip" "github.com/pilosa/pilosa/server" "github.com/pilosa/pilosa/test" ) @@ -384,21 +384,15 @@ func TestCountOpenFiles(t *testing.T) { } } -/* TODO: Fix this test. See #951. // Ensure program can send/receive broadcast messages. func TestMain_SendReceiveMessage(t *testing.T) { + m0 := MustRunMain() defer m0.Close() m1 := MustRunMain() defer m1.Close() - // Get available ports for internal messaging - freePorts, err := availablePorts(2) - if err != nil { - t.Fatal(err) - } - // Update cluster config m0.Server.Cluster.Nodes = []*pilosa.Node{ {URI: m0.Server.URI}, @@ -409,15 +403,9 @@ func TestMain_SendReceiveMessage(t *testing.T) { // Configure node0 // get the host portion of addr to use for binding - gossipHost, _, err := net.SplitHostPort(m0.Server.URI.HostPort()) - if err != nil { - gossipHost = m0.Server.URI.HostPort() - } - gossipPort, err := strconv.Atoi(freePorts[0]) - if err != nil { - t.Fatal(err) - } - gossipSeed := gossipHost + ":" + freePorts[0] + gossipHost := m0.Server.URI.Host() + gossipPort := 0 + gossipSeed := "" topology := &pilosa.Topology{NodeSet: []pilosa.URI{m0.Server.URI, m1.Server.URI}} @@ -425,7 +413,10 @@ func TestMain_SendReceiveMessage(t *testing.T) { m0.Server.Cluster.Topology = topology m0.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver() - gossipMemberSet0 := gossip.NewGossipMemberSet(m0.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m0.Server, nil) + 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 = gossipMemberSet0 m0.Server.Handler.Broadcaster = m0.Server.Broadcaster @@ -443,19 +434,17 @@ func TestMain_SendReceiveMessage(t *testing.T) { // Configure node1 // get the host portion of addr to use for binding - gossipHost, _, err = net.SplitHostPort(m1.Server.URI.HostPort()) - if err != nil { - gossipHost = m1.Server.URI.HostPort() - } - gossipPort, err = strconv.Atoi(freePorts[1]) - if err != nil { - t.Fatal(err) - } + gossipHost = m1.Server.URI.Host() + gossipPort = 0 + gossipSeed = gossipMemberSet0.Seed() m1.Server.Cluster.Coordinator = m0.Server.URI m1.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver() - gossipMemberSet1 := gossip.NewGossipMemberSet(m1.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m1.Server, nil) + 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 = gossipMemberSet1 m1.Server.Handler.Broadcaster = m1.Server.Broadcaster @@ -575,37 +564,6 @@ func TestMain_SendReceiveMessage(t *testing.T) { t.Fatal("frame not found") } } -*/ - -// availablePorts returns a slice of ports that can be used for testing. -func availablePorts(cnt int) ([]string, error) { - rtn := []string{} - - for i := 0; i < cnt; i++ { - port, err := getPort() - if err != nil { - return nil, err - } - rtn = append(rtn, strconv.Itoa(port)) - } - return rtn, nil -} - -// Ask the kernel for a free open port that is ready to use -func getPort() (int, error) { - addr, err := net.ResolveTCPAddr("tcp", "localhost:0") - if err != nil { - return 0, err - } - - l, err := net.ListenTCP("tcp", addr) - if err != nil { - return 0, err - } - defer l.Close() - - return l.Addr().(*net.TCPAddr).Port, nil -} // Main represents a test wrapper for main.Main. type Main struct {