Merge pull request #2858 from influxdb/config_opentsdb_consis

Support setting openTSDB write consistency

CHANGELOG.md

@@ -12,6 +12,9 @@
 - [#2857](https://github.com/influxdb/influxdb/issues/2857) -- Fix parsing commas in string field values.
 - [#2833](https://github.com/influxdb/influxdb/pull/2833) -- Make the default config valid.
 
+### Features
+- [2858](https://github.com/influxdb/influxdb/pull/2858) - Support setting openTSDB write consistency.
+
 ## v0.9.0-rc32 [2015-06-07]
 
 ### Release Notes

cmd/influxd/run/server.go

@@ -80,7 +80,9 @@ func NewServer(c *Config) (*Server, error) {
 	s.appendContinuousQueryService(c.ContinuousQuery)
 	s.appendHTTPDService(c.HTTPD)
 	s.appendCollectdService(c.Collectd)
-	s.appendOpenTSDBService(c.OpenTSDB)
+	if err := s.appendOpenTSDBService(c.OpenTSDB); err != nil {
+		return nil, err
+	}
 	s.appendUDPService(c.UDP)
 	s.appendRetentionPolicyService(c.Retention)
 	for _, g := range c.Graphites {
@@ -145,13 +147,17 @@ func (s *Server) appendCollectdService(c collectd.Config) {
 	s.Services = append(s.Services, srv)
 }
 
-func (s *Server) appendOpenTSDBService(c opentsdb.Config) {
+func (s *Server) appendOpenTSDBService(c opentsdb.Config) error {
 	if !c.Enabled {
-		return
+		return nil
+	}
+	srv, err := opentsdb.NewService(c)
+	if err != nil {
+		return err
 	}
-	srv := opentsdb.NewService(c)
 	srv.PointsWriter = s.PointsWriter
 	s.Services = append(s.Services, srv)
+	return nil
 }
 
 func (s *Server) appendGraphiteService(c graphite.Config) error {

services/opentsdb/config.go

@@ -9,19 +9,24 @@ const (
 
 	// DefaultRetentionPolicy is the default retention policy used for writes.
 	DefaultRetentionPolicy = ""
+
+	// DefaultConsistencyLevel is the default write consistency level.
+	DefaultConsistencyLevel = "one"
 )
 
 type Config struct {
-	Enabled         bool   `toml:"enabled"`
-	BindAddress     string `toml:"bind-address"`
-	Database        string `toml:"database"`
-	RetentionPolicy string `toml:"retention-policy"`
+	Enabled          bool   `toml:"enabled"`
+	BindAddress      string `toml:"bind-address"`
+	Database         string `toml:"database"`
+	RetentionPolicy  string `toml:"retention-policy"`
+	ConsistencyLevel string `toml:"consistency-level"`
 }
 
 func NewConfig() Config {
 	return Config{
-		BindAddress:     DefaultBindAddress,
-		Database:        DefaultDatabase,
-		RetentionPolicy: DefaultRetentionPolicy,
+		BindAddress:      DefaultBindAddress,
+		Database:         DefaultDatabase,
+		RetentionPolicy:  DefaultRetentionPolicy,
+		ConsistencyLevel: DefaultConsistencyLevel,
 	}
 }

services/opentsdb/config_test.go

@@ -14,6 +14,7 @@ func TestConfig_Parse(t *testing.T) {
 enabled = true
 bind-address = ":9000"
 database = "xxx"
+consistency-level ="all"
 `, &c); err != nil {
 		t.Fatal(err)
 	}
@@ -25,5 +26,7 @@ database = "xxx"
 		t.Fatalf("unexpected bind address: %s", c.BindAddress)
 	} else if c.Database != "xxx" {
 		t.Fatalf("unexpected database: %s", c.Database)
+	} else if c.ConsistencyLevel != "all" {
+		t.Fatalf("unexpected database: %s", c.ConsistencyLevel)
 	}
 }

services/opentsdb/handler.go

@@ -17,8 +17,9 @@ import (
 )
 
 type Handler struct {
-	Database        string
-	RetentionPolicy string
+	Database         string
+	RetentionPolicy  string
+	ConsistencyLevel cluster.ConsistencyLevel
 
 	PointsWriter interface {
 		WritePoints(p *cluster.WritePointsRequest) error
@@ -115,7 +116,7 @@ func (h *Handler) servePut(w http.ResponseWriter, r *http.Request) {
 	if err := h.PointsWriter.WritePoints(&cluster.WritePointsRequest{
 		Database:         h.Database,
 		RetentionPolicy:  h.RetentionPolicy,
-		ConsistencyLevel: cluster.ConsistencyLevelOne,
+		ConsistencyLevel: h.ConsistencyLevel,
 		Points:           points,
 	}); influxdb.IsClientError(err) {
 		h.Logger.Println("write series error: ", err)

services/opentsdb/service.go

@@ -26,9 +26,10 @@ type Service struct {
 	wg  sync.WaitGroup
 	err chan error
 
-	BindAddress     string
-	Database        string
-	RetentionPolicy string
+	BindAddress      string
+	Database         string
+	RetentionPolicy  string
+	ConsistencyLevel cluster.ConsistencyLevel
 
 	PointsWriter interface {
 		WritePoints(p *cluster.WritePointsRequest) error
@@ -38,15 +39,21 @@ type Service struct {
 }
 
 // NewService returns a new instance of Service.
-func NewService(c Config) *Service {
+func NewService(c Config) (*Service, error) {
+	consistencyLevel, err := cluster.ParseConsistencyLevel(c.ConsistencyLevel)
+	if err != nil {
+		return nil, err
+	}
+
 	s := &Service{
-		err:             make(chan error),
-		BindAddress:     c.BindAddress,
-		Database:        c.Database,
-		RetentionPolicy: c.RetentionPolicy,
-		Logger:          log.New(os.Stderr, "[opentsdb] ", log.LstdFlags),
+		err:              make(chan error),
+		BindAddress:      c.BindAddress,
+		Database:         c.Database,
+		RetentionPolicy:  c.RetentionPolicy,
+		ConsistencyLevel: consistencyLevel,
+		Logger:           log.New(os.Stderr, "[opentsdb] ", log.LstdFlags),
 	}
-	return s
+	return s, nil
 }
 
 // Open starts the service
@@ -211,7 +218,7 @@ func (s *Service) handleTelnetConn(conn net.Conn) {
 		if err := s.PointsWriter.WritePoints(&cluster.WritePointsRequest{
 			Database:         s.Database,
 			RetentionPolicy:  s.RetentionPolicy,
-			ConsistencyLevel: cluster.ConsistencyLevelOne,
+			ConsistencyLevel: s.ConsistencyLevel,
 			Points:           []tsdb.Point{p},
 		}); err != nil {
 			s.Logger.Println("TSDB cannot write data: ", err)
@@ -223,10 +230,11 @@ func (s *Service) handleTelnetConn(conn net.Conn) {
 // serveHTTP handles connections in HTTP format.
 func (s *Service) serveHTTP() {
 	srv := &http.Server{Handler: &Handler{
-		Database:        s.Database,
-		RetentionPolicy: s.RetentionPolicy,
-		PointsWriter:    s.PointsWriter,
-		Logger:          s.Logger,
+		Database:         s.Database,
+		RetentionPolicy:  s.RetentionPolicy,
+		ConsistencyLevel: s.ConsistencyLevel,
+		PointsWriter:     s.PointsWriter,
+		Logger:           s.Logger,
 	}}
 	srv.Serve(s.httpln)
 }

services/opentsdb/service_test.go

@@ -129,12 +129,12 @@ type Service struct {
 
 // NewService returns a new instance of Service.
 func NewService(database string) *Service {
-	s := &Service{
-		Service: opentsdb.NewService(opentsdb.Config{
-			BindAddress: "127.0.0.1:0",
-			Database:    database,
-		}),
-	}
+	srv, _ := opentsdb.NewService(opentsdb.Config{
+		BindAddress:      "127.0.0.1:0",
+		Database:         database,
+		ConsistencyLevel: "one",
+	})
+	s := &Service{Service: srv}
 	s.Service.PointsWriter = &s.PointsWriter
 
 	if !testing.Verbose() {