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

Merge branch 'master' into cluster-resize

Browse source
This commit is contained in:
Travis Turner 2017-12-08 15:22:12 -06:00
commit cf1e41faf1
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
11 changed files with 263 additions and 170 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
cluster.go
View file

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

65
docs/configuration.md
View file

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

95
docs/input-definition.md
View file

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

6
executor.go
View file

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

27
executor_test.go
View file

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

45
fragment.go
View file

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

4
frame.go
View file

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

97
gossip/gossip.go
View file

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

1
server.go
View file

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

5
server/server.go
View file

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

76
server/server_test.go
View file

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