Merge pull request #2840 from influxdb/parse_cluster_consistency

Add cluster consistency parsing to cluster package

cluster/points_writer.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"log"
 	"os"
+	"strings"
 	"sync"
 	"time"
 
@@ -41,8 +42,27 @@ var (
 
 	// ErrWriteFailed is returned when no writes succeeded.
 	ErrWriteFailed = errors.New("write failed")
+
+	// ErrInvalidConsistencyLevel is returned when parsing the string version
+	// of a consistency level.
+	ErrInvalidConsistencyLevel = errors.New("invalid consistency level")
 )
 
+func ParseConsistencyLevel(level string) (ConsistencyLevel, error) {
+	switch strings.ToLower(level) {
+	case "any":
+		return ConsistencyLevelAny, nil
+	case "one":
+		return ConsistencyLevelOne, nil
+	case "quorum":
+		return ConsistencyLevelQuorum, nil
+	case "all":
+		return ConsistencyLevelAll, nil
+	default:
+		return 0, ErrInvalidConsistencyLevel
+	}
+}
+
 // PointsWriter handles writes across multiple local and remote data nodes.
 type PointsWriter struct {
 	mu      sync.RWMutex

cmd/influxd/run/command.go

@@ -84,7 +84,10 @@ func (cmd *Command) Run(args ...string) error {
 	}
 
 	// Create server from config and start it.
-	s := NewServer(config)
+	s, err := NewServer(config)
+	if err != nil {
+		return fmt.Errorf("create server: %s", err)
+	}
 	if err := s.Open(); err != nil {
 		return fmt.Errorf("open server: %s", err)
 	}

cmd/influxd/run/server.go

@@ -42,7 +42,7 @@ type Server struct {
 }
 
 // NewServer returns a new instance of Server built from a config.
-func NewServer(c *Config) *Server {
+func NewServer(c *Config) (*Server, error) {
 	// Construct base meta store and data store.
 	s := &Server{
 		err:     make(chan error),
@@ -84,10 +84,12 @@ func NewServer(c *Config) *Server {
 	s.appendUDPService(c.UDP)
 	s.appendRetentionPolicyService(c.Retention)
 	for _, g := range c.Graphites {
-		s.appendGraphiteService(g)
+		if err := s.appendGraphiteService(g); err != nil {
+			return nil, err
+		}
 	}
 
-	return s
+	return s, nil
 }
 
 func (s *Server) appendClusterService(c cluster.Config) {
@@ -152,13 +154,18 @@ func (s *Server) appendOpenTSDBService(c opentsdb.Config) {
 	s.Services = append(s.Services, srv)
 }
 
-func (s *Server) appendGraphiteService(c graphite.Config) {
+func (s *Server) appendGraphiteService(c graphite.Config) error {
 	if !c.Enabled {
-		return
+		return nil
 	}
-	srv := graphite.NewService(c)
+	srv, err := graphite.NewService(c)
+	if err != nil {
+		return err
+	}
+
 	srv.PointsWriter = s.PointsWriter
 	s.Services = append(s.Services, srv)
+	return nil
 }
 
 func (s *Server) appendUDPService(c udp.Config) {

cmd/influxd/run/server_helpers_test.go

@@ -30,9 +30,9 @@ type Server struct {
 
 // NewServer returns a new instance of Server.
 func NewServer(c *run.Config) *Server {
-
+	srv, _ := run.NewServer(c)
 	s := Server{
-		Server: run.NewServer(c),
+		Server: srv,
 		Config: c,
 	}
 	// Set the logger to discard unless verbose is on

services/graphite/config.go

@@ -3,7 +3,6 @@ package graphite
 import (
 	"strings"
 
-	"github.com/influxdb/influxdb/cluster"
 	"github.com/influxdb/influxdb/toml"
 )
 
@@ -56,19 +55,3 @@ func NewConfig() Config {
 func (c *Config) LastEnabled() bool {
 	return c.NamePosition == strings.ToLower("last")
 }
-
-// ConsistencyAsEnum returns the enumerated write consistency level.
-func (c *Config) ConsistencyAsEnum() cluster.ConsistencyLevel {
-	switch strings.ToLower(c.ConsistencyLevel) {
-	case "any":
-		return cluster.ConsistencyLevelAny
-	case "one":
-		return cluster.ConsistencyLevelOne
-	case "quorum":
-		return cluster.ConsistencyLevelQuorum
-	case "all":
-		return cluster.ConsistencyLevelAll
-	default:
-		return cluster.ConsistencyLevelOne
-	}
-}

services/graphite/service.go

@@ -44,24 +44,29 @@ type Service struct {
 }
 
 // NewService returns an instance of the Graphite service.
-func NewService(c Config) *Service {
+func NewService(c Config) (*Service, error) {
 	s := Service{
-		bindAddress:      c.BindAddress,
-		database:         c.Database,
-		protocol:         c.Protocol,
-		batchSize:        c.BatchSize,
-		batchTimeout:     time.Duration(c.BatchTimeout),
-		consistencyLevel: c.ConsistencyAsEnum(),
-		logger:           log.New(os.Stderr, "[graphite] ", log.LstdFlags),
-		done:             make(chan struct{}),
+		bindAddress:  c.BindAddress,
+		database:     c.Database,
+		protocol:     c.Protocol,
+		batchSize:    c.BatchSize,
+		batchTimeout: time.Duration(c.BatchTimeout),
+		logger:       log.New(os.Stderr, "[graphite] ", log.LstdFlags),
+		done:         make(chan struct{}),
 	}
 
+	consistencyLevel, err := cluster.ParseConsistencyLevel(c.ConsistencyLevel)
+	if err != nil {
+		return nil, err
+	}
+	s.consistencyLevel = consistencyLevel
+
 	parser := NewParser()
 	parser.Separator = c.NameSeparator
 	parser.LastEnabled = c.LastEnabled()
 	s.parser = parser
 
-	return &s
+	return &s, nil
 }
 
 // Open starts the Graphite input processing data.

services/graphite/service_test.go

@@ -241,9 +241,9 @@ func Test_ServerGraphiteTCP(t *testing.T) {
 	config.BatchTimeout = toml.Duration(time.Second)
 	config.BindAddress = ":0"
 
-	service := graphite.NewService(config)
-	if service == nil {
-		t.Fatal("failed to create Graphite service")
+	service, err := graphite.NewService(config)
+	if err != nil {
+		t.Fatalf("failed to create Graphite service: %s", err.Error())
 	}
 
 	// Allow test to wait until points are written.
@@ -307,9 +307,9 @@ func Test_ServerGraphiteUDP(t *testing.T) {
 	config.BindAddress = ":10000"
 	config.Protocol = "udp"
 
-	service := graphite.NewService(config)
-	if service == nil {
-		t.Fatal("failed to create Graphite service")
+	service, err := graphite.NewService(config)
+	if err != nil {
+		t.Fatalf("failed to create Graphite service: %s", err.Error())
 	}
 
 	// Allow test to wait until points are written.