Skip to content

Commit

Permalink
Merge pull request #2325 from influxdb/cluster-open
Browse files Browse the repository at this point in the history
Cluster open fixes
  • Loading branch information
jwilder committed Apr 17, 2015
2 parents 8ee8218 + 16f3cee commit 7e1303f
Show file tree
Hide file tree
Showing 4 changed files with 19 additions and 14 deletions.
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@
- [#2242](https://github.com/influxdb/influxdb/pull/2242): Distributed Query should balance requests
- [#2243](https://github.com/influxdb/influxdb/pull/2243): Use Limit Reader instead of fixed 1MB/1GB slice for DQ
- [#2190](https://github.com/influxdb/influxdb/pull/2190): Implement failover to other data nodes for distributed queries
- [#2324](https://github.com/influxdb/influxdb/issues/2324): Race in Broker.Close()/Broker.RunContinousQueryProcessing()
- [#2325](https://github.com/influxdb/influxdb/pull/2325): Cluster open fixes

## v0.9.0-rc25 [2015-04-15]

Expand Down
10 changes: 10 additions & 0 deletions broker.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"math/rand"
"net/http"
"net/url"
"sync"
"time"

"github.com/influxdb/influxdb/messaging"
Expand All @@ -27,6 +28,7 @@ const (

// Broker represents an InfluxDB specific messaging broker.
type Broker struct {
mu sync.RWMutex
*messaging.Broker
client *http.Client

Expand Down Expand Up @@ -54,12 +56,18 @@ func NewBroker() *Broker {

// RunContinuousQueryLoop starts running continuous queries on a background goroutine.
func (b *Broker) RunContinuousQueryLoop() {
b.mu.Lock()
defer b.mu.Unlock()

b.done = make(chan struct{})
go b.continuousQueryLoop(b.done)
}

// Close closes the broker.
func (b *Broker) Close() error {
b.mu.Lock()
defer b.mu.Unlock()

if b.done != nil {
close(b.done)
b.done = nil
Expand Down Expand Up @@ -121,6 +129,8 @@ func (b *Broker) runContinuousQueries() {
}

func (b *Broker) requestContinuousQueryProcessing(cqURL url.URL) error {
b.mu.RLock()
defer b.mu.RUnlock()
// Send request.
cqURL.Path = "/data/process_continuous_queries"
cqURL.Scheme = "http"
Expand Down
2 changes: 1 addition & 1 deletion cmd/influxd/run.go
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,7 @@ func (s *Node) openListener(desc, addr string, h http.Handler) (net.Listener, er

// The listener was closed so exit
// See https://github.com/golang/go/issues/4373
if operr, ok := err.(*net.OpError); ok && strings.Contains(operr.Err.Error(), "closed network connection") {
if strings.Contains(err.Error(), "closed") {
return
}
if err != nil {
Expand Down
19 changes: 6 additions & 13 deletions httpd/handler_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -541,8 +541,7 @@ func TestHandler_GzipEnabled(t *testing.T) {
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept-Encoding", "gzip")

client := &http.Client{}
resp, err := client.Do(req)
resp, err := http.DefaultClient.Do(req)
if err != nil {
panic(err)
}
Expand All @@ -567,8 +566,7 @@ func TestHandler_GzipDisabled(t *testing.T) {
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept-Encoding", "")

client := &http.Client{}
resp, err := client.Do(req)
resp, err := http.DefaultClient.Do(req)
if err != nil {
panic(err)
}
Expand Down Expand Up @@ -1234,8 +1232,6 @@ func TestHandler_serveWriteSeries_errorHasJsonContentType(t *testing.T) {
s := NewAPIServer(srvr)
defer s.Close()

client := &http.Client{}

req, err := http.NewRequest("POST", s.URL+`/write`, bytes.NewBufferString("{}"))
if err != nil {
panic(err)
Expand All @@ -1244,7 +1240,7 @@ func TestHandler_serveWriteSeries_errorHasJsonContentType(t *testing.T) {
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept-Encoding", "gzip")

resp, err := client.Do(req)
resp, err := http.DefaultClient.Do(req)
if err != nil {
panic(err)
}
Expand Down Expand Up @@ -1273,8 +1269,6 @@ func TestHandler_serveWriteSeries_queryHasJsonContentType(t *testing.T) {

srvr.Restart() // Ensure data is queryable across restarts.

client := &http.Client{}

params := url.Values{}
params.Add("db", "foo")
params.Add("q", "select * from cpu")
Expand All @@ -1285,7 +1279,7 @@ func TestHandler_serveWriteSeries_queryHasJsonContentType(t *testing.T) {

req.Header.Set("Accept-Encoding", "gzip")

resp, err := client.Do(req)
resp, err := http.DefaultClient.Do(req)
if err != nil {
panic(err)
}
Expand All @@ -1304,7 +1298,7 @@ func TestHandler_serveWriteSeries_queryHasJsonContentType(t *testing.T) {

req_error.Header.Set("Accept-Encoding", "gzip")

resp_error, err := client.Do(req_error)
resp_error, err := http.DefaultClient.Do(req_error)
if err != nil {
panic(err)
}
Expand Down Expand Up @@ -1726,8 +1720,7 @@ func MustHTTP(verb, path string, params, headers map[string]string, body string)
}
req.Header.Set("Content-Type", "application/json")

client := &http.Client{}
resp, err := client.Do(req)
resp, err := http.DefaultClient.Do(req)
if err != nil {
panic(err)
}
Expand Down

0 comments on commit 7e1303f

Please sign in to comment.