Merge pull request #3218 from influxdb/config_timeouts

Allow PointerWriter timeout to be configurable

CHANGELOG.md

@@ -6,6 +6,7 @@
 ### Bugfixes
 
 - [#3180](https://github.com/influxdb/influxdb/pull/3180): Log GOMAXPROCS, version, and commit on startup.
+- [#3218](https://github.com/influxdb/influxdb/pull/3218): Allow write timeouts to be configurable.
 
 
 ## v0.9.1 [2015-07-02]

cluster/config.go

@@ -7,18 +7,23 @@ import (
 )
 
 const (
+	// DefaultWriteTimeout is the default timeout for a complete write to succeed.
+	DefaultWriteTimeout = 5 * time.Second
+
 	// DefaultShardWriterTimeout is the default timeout set on shard writers.
 	DefaultShardWriterTimeout = 5 * time.Second
 )
 
 // Config represents the configuration for the the clustering service.
 type Config struct {
+	WriteTimeout       toml.Duration `toml:"write-timeout"`
 	ShardWriterTimeout toml.Duration `toml:"shard-writer-timeout"`
 }
 
 // NewConfig returns an instance of Config with defaults.
 func NewConfig() Config {
 	return Config{
+		WriteTimeout:       toml.Duration(DefaultWriteTimeout),
 		ShardWriterTimeout: toml.Duration(DefaultShardWriterTimeout),
 	}
 }

cluster/config_test.go

@@ -13,12 +13,15 @@ func TestConfig_Parse(t *testing.T) {
 	var c cluster.Config
 	if _, err := toml.Decode(`
 shard-writer-timeout = "10s"
+write-timeout = "20s"
 `, &c); err != nil {
 		t.Fatal(err)
 	}
 
 	// Validate configuration.
 	if time.Duration(c.ShardWriterTimeout) != 10*time.Second {
-		t.Fatalf("unexpected bind address: %s", c.ShardWriterTimeout)
+		t.Fatalf("unexpected shard-writer timeout: %s", c.ShardWriterTimeout)
+	} else if time.Duration(c.WriteTimeout) != 20*time.Second {
+		t.Fatalf("unexpected write timeout s: %s", c.WriteTimeout)
 	}
 }

cluster/points_writer.go

@@ -14,8 +14,6 @@ import (
 	"github.com/influxdb/influxdb/tsdb"
 )
 
-const DefaultWriteTimeout = 5 * time.Second
-
 // ConsistencyLevel represent a required replication criteria before a write can
 // be returned as successful
 type ConsistencyLevel int
@@ -67,9 +65,10 @@ func ParseConsistencyLevel(level string) (ConsistencyLevel, error) {
 
 // PointsWriter handles writes across multiple local and remote data nodes.
 type PointsWriter struct {
-	mu      sync.RWMutex
-	closing chan struct{}
-	Logger  *log.Logger
+	mu           sync.RWMutex
+	closing      chan struct{}
+	WriteTimeout time.Duration
+	Logger       *log.Logger
 
 	MetaStore interface {
 		NodeID() uint64
@@ -96,8 +95,9 @@ type PointsWriter struct {
 // NewPointsWriter returns a new instance of PointsWriter for a node.
 func NewPointsWriter() *PointsWriter {
 	return &PointsWriter{
-		closing: make(chan struct{}),
-		Logger:  log.New(os.Stderr, "[write] ", log.LstdFlags),
+		closing:      make(chan struct{}),
+		WriteTimeout: DefaultWriteTimeout,
+		Logger:       log.New(os.Stderr, "[write] ", log.LstdFlags),
 	}
 }
 
@@ -272,7 +272,7 @@ func (w *PointsWriter) writeToShard(shard *meta.ShardInfo, database, retentionPo
 	}
 
 	var wrote int
-	timeout := time.After(DefaultWriteTimeout)
+	timeout := time.After(w.WriteTimeout)
 	var writeError error
 	for _, nodeID := range shard.OwnerIDs {
 		select {

cmd/influxd/run/server.go

@@ -99,6 +99,7 @@ func NewServer(c *Config, version string) (*Server, error) {
 
 	// Initialize points writer.
 	s.PointsWriter = cluster.NewPointsWriter()
+	s.PointsWriter.WriteTimeout = time.Duration(c.Cluster.WriteTimeout)
 	s.PointsWriter.MetaStore = s.MetaStore
 	s.PointsWriter.TSDBStore = s.TSDBStore
 	s.PointsWriter.ShardWriter = s.ShardWriter

etc/config.sample.toml

@@ -48,7 +48,8 @@ reporting-disabled = false
 ###
 
 [cluster]
-  shard-writer-timeout = "5s"
+  shard-writer-timeout = "5s" # The time within which a shard must respond to write.
+  write-timeout = "5s" # The time within which a write operation must complete on the cluster.
 
 ###
 ### [retention]