From 89c641ed24507f75248ade6d02a365ed2ceb8066 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Tue, 24 Oct 2017 16:36:39 +0300 Subject: [PATCH 1/7] Added single cluster config; implements #871 --- docs/configuration.md | 65 +++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 63 insertions(+), 2 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 10325b09f..dcda25a99 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -78,7 +78,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 there are more than one Pilosa servers are running on the same host, their gossip ports should be different. * Flag: `--gossip.port=11101` * Env: `PILOSA_GOSSIP_PORT=11101` * Config: @@ -257,7 +257,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 @@ -362,3 +362,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" From bbd9aedf0faf46dfffc4a12517ff8c8226f0821a Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Tue, 24 Oct 2017 23:06:25 +0300 Subject: [PATCH 2/7] fix gossip port text --- docs/configuration.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/configuration.md b/docs/configuration.md index dcda25a99..66a858cb9 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -78,7 +78,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. If there are more than one Pilosa servers are running on the same host, their gossip ports should be different. +* 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: From 59035883f432887ddb517977f35299222aab1fc4 Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Fri, 10 Nov 2017 15:03:23 -0600 Subject: [PATCH 3/7] group the write operations in syncBlock by MaxWritesPerRequest --- cluster.go | 10 +++++++--- fragment.go | 45 +++++++++++++++++++++++++++------------------ server.go | 1 + 3 files changed, 35 insertions(+), 21 deletions(-) diff --git a/cluster.go b/cluster.go index 820349ded..cdcf50b26 100644 --- a/cluster.go +++ b/cluster.go @@ -144,14 +144,18 @@ type Cluster struct { // Threshold for logging long-running queries LongQueryTime time.Duration + + // Maximum number of SetBit() or ClearBit() commands per request. + MaxWritesPerRequest int } // NewCluster returns a new instance of Cluster with defaults. func NewCluster() *Cluster { return &Cluster{ - Hasher: &jmphasher{}, - PartitionN: DefaultPartitionN, - ReplicaN: DefaultReplicaN, + Hasher: &jmphasher{}, + PartitionN: DefaultPartitionN, + ReplicaN: DefaultReplicaN, + MaxWritesPerRequest: DefaultMaxWritesPerRequest, } } diff --git a/fragment.go b/fragment.go index e5fbb7988..401faf17c 100644 --- a/fragment.go +++ b/fragment.go @@ -492,7 +492,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 { @@ -586,7 +586,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) @@ -615,6 +615,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: @@ -753,6 +754,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/server.go b/server.go index 0d55e7c47..e2769a0be 100644 --- a/server.go +++ b/server.go @@ -176,6 +176,7 @@ func (s *Server) Open() error { e.Host = s.URI.HostPort() e.Cluster = s.Cluster e.MaxWritesPerRequest = s.MaxWritesPerRequest + s.Cluster.MaxWritesPerRequest = s.MaxWritesPerRequest // Initialize HTTP handler. s.Handler.Broadcaster = s.Broadcaster From 450a4c0724430c7dce84914f2d7b7a132704eb6c Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Fri, 17 Nov 2017 18:27:06 +0300 Subject: [PATCH 4/7] Fixes the import data section in the input definition docs --- docs/input-definition.md | 95 ++++++---------------------------------- 1 file changed, 14 insertions(+), 81 deletions(-) diff --git a/docs/input-definition.md b/docs/input-definition.md index 494f4757e..14bb505c4 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 @json_input.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. From 5f4545eeb2ddee9a3cbe943f53e0cea7bb66fb24 Mon Sep 17 00:00:00 2001 From: Cody Soyland Date: Tue, 21 Nov 2017 13:43:44 -0600 Subject: [PATCH 5/7] Fix edge case with Range() calls outside field Min/Max. Fixes #876. --- executor.go | 6 ++++++ executor_test.go | 27 +++++++++++++++++++++++++++ frame.go | 4 +++- 3 files changed, 36 insertions(+), 1 deletion(-) diff --git a/executor.go b/executor.go index 2f5acf3c0..b165eb4da 100644 --- a/executor.go +++ b/executor.go @@ -803,6 +803,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 4eb81e492..dd8725298 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/frame.go b/frame.go index 3da55cc93..9940c120e 100644 --- a/frame.go +++ b/frame.go @@ -1108,7 +1108,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 @@ -1116,6 +1116,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 { From 390448c342252f272386802b0e708ca11f2ea651 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Wed, 22 Nov 2017 20:22:12 +0300 Subject: [PATCH 6/7] fix input definition file name --- docs/input-definition.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/input-definition.md b/docs/input-definition.md index 14bb505c4..7d50c6f9d 100644 --- a/docs/input-definition.md +++ b/docs/input-definition.md @@ -27,7 +27,7 @@ curl -OL https://github.com/pilosa/getting-started/raw/master/input_definition.j Run the following to create the input definition: ``` -curl localhost:10101/index/repository/input-definition/stargazer -d @json_input.json +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. From 9b54259cd73e67fc09f5eabed95b916a0faa0f53 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Mon, 4 Dec 2017 14:56:21 -0600 Subject: [PATCH 7/7] Resume test: TestMain_SendReceiveMessage Create a custom memberlist NetTransport (which will bind to an available port when port = 0 in the configuration). This allows us to bind to dynamic ports in tests while at the same time determining a valid seed for the cluster. --- gossip/gossip.go | 98 ++++++++++++++++++++++++++++++++++++++++--- server/server.go | 5 ++- server/server_test.go | 68 ++++++------------------------ 3 files changed, 110 insertions(+), 61 deletions(-) diff --git a/gossip/gossip.go b/gossip/gossip.go index 7085710b3..e413410d8 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" @@ -59,6 +61,11 @@ func (g *GossipNodeSet) Start(h pilosa.BroadcastHandler) error { return nil } +// Seed returns the gossipSeed determined by the config. +func (g *GossipNodeSet) Seed() string { + return g.config.gossipSeed +} + // Open implements the NodeSet interface to start network activity. func (g *GossipNodeSet) Open() error { if g.handler == nil { @@ -122,28 +129,109 @@ 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 +} + // NewGossipNodeSet returns a new instance of GossipNodeSet. -func NewGossipNodeSet(name string, gossipHost string, gossipPort int, gossipSeed string, server *pilosa.Server, secretKey []byte) *GossipNodeSet { +func NewGossipNodeSet(name string, gossipHost string, gossipPort int, gossipSeed string, server *pilosa.Server, secretKey []byte) (*GossipNodeSet, error) { g := &GossipNodeSet{ 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.Delegate = g g.config.memberlistConfig.SecretKey = secretKey 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/server.go b/server/server.go index a5f17f6d7..93b0f184f 100644 --- a/server/server.go +++ b/server/server.go @@ -221,7 +221,10 @@ func (m *Command) SetupServer() error { // get the host portion of addr to use for binding gossipHost := uri.Host() - gossipNodeSet := gossip.NewGossipNodeSet(uri.HostPort(), gossipHost, gossipPort, gossipSeed, m.Server, gossipKey) + gossipNodeSet, err := gossip.NewGossipNodeSet(uri.HostPort(), gossipHost, gossipPort, gossipSeed, m.Server, gossipKey) + if err != nil { + return err + } m.Server.Cluster.NodeSet = gossipNodeSet m.Server.Broadcaster = gossipNodeSet m.Server.BroadcastReceiver = gossipNodeSet diff --git a/server/server_test.go b/server/server_test.go index 28886bc67..ca8ac7f09 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{ {Host: m0.Server.URI.HostPort()}, @@ -409,17 +403,14 @@ 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]) + gossipHost := m0.Server.URI.Host() + gossipPort := 0 + gossipSeed := "" + + gossipNodeSet0, err := gossip.NewGossipNodeSet(m0.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m0.Server, nil) if err != nil { t.Fatal(err) } - gossipSeed := gossipHost + ":" + freePorts[0] - - gossipNodeSet0 := gossip.NewGossipNodeSet(m0.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m0.Server, nil) m0.Server.Cluster.NodeSet = gossipNodeSet0 m0.Server.Broadcaster = gossipNodeSet0 m0.Server.Handler.Broadcaster = m0.Server.Broadcaster @@ -437,16 +428,14 @@ 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]) + gossipHost = m1.Server.URI.Host() + gossipPort = 0 + gossipSeed = gossipNodeSet0.Seed() + + gossipNodeSet1, err := gossip.NewGossipNodeSet(m1.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m1.Server, nil) if err != nil { t.Fatal(err) } - - gossipNodeSet1 := gossip.NewGossipNodeSet(m1.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m1.Server, nil) m1.Server.Cluster.NodeSet = gossipNodeSet1 m1.Server.Broadcaster = gossipNodeSet1 m1.Server.Handler.Broadcaster = m1.Server.Broadcaster @@ -566,37 +555,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 {