Allow multiple Collectd plugin listeners

cmd/influxd/run/config.go

@@ -48,7 +48,7 @@ type Config struct {
 	Subscriber subscriber.Config `toml:"subscriber"`
 	HTTPD      httpd.Config      `toml:"http"`
 	Graphites  []graphite.Config `toml:"graphite"`
-	Collectd   collectd.Config   `toml:"collectd"`
+	Collectds  []collectd.Config `toml:"collectd"`
 	OpenTSDB   opentsdb.Config   `toml:"opentsdb"`
 	UDPs       []udp.Config      `toml:"udp"`
 
@@ -79,7 +79,6 @@ func NewConfig() *Config {
 	c.Monitor = monitor.NewConfig()
 	c.Subscriber = subscriber.NewConfig()
 	c.HTTPD = httpd.NewConfig()
-	c.Collectd = collectd.NewConfig()
 	c.OpenTSDB = opentsdb.NewConfig()
 
 	c.ContinuousQuery = continuous_querier.NewConfig()
@@ -104,6 +103,9 @@ func (c *Config) InitTableAttrs() {
 	if len(c.Graphites) == 0 {
 		c.Graphites = []graphite.Config{graphite.NewConfig()}
 	}
+	if len(c.Collectds) == 0 {
+		c.Collectds = []collectd.Config{collectd.NewConfig()}
+	}
 }
 
 // NewDemoConfig returns the config that runs when no config is specified.

cmd/influxd/run/config_test.go

@@ -35,9 +35,12 @@ protocol = "udp"
 [[graphite]]
 protocol = "tcp"
 
-[collectd]
+[[collectd]]
 bind-address = ":1000"
 
+[[collectd]]
+bind-address = ":1001"
+
 [opentsdb]
 bind-address = ":2000"
 
@@ -71,8 +74,10 @@ enabled = true
 		t.Fatalf("unexpected graphite protocol(0): %s", c.Graphites[0].Protocol)
 	} else if c.Graphites[1].Protocol != "tcp" {
 		t.Fatalf("unexpected graphite protocol(1): %s", c.Graphites[1].Protocol)
-	} else if c.Collectd.BindAddress != ":1000" {
-		t.Fatalf("unexpected collectd bind address: %s", c.Collectd.BindAddress)
+	} else if c.Collectds[0].BindAddress != ":1000" {
+		t.Fatalf("unexpected collectd bind address: %s", c.Collectds[0].BindAddress)
+	} else if c.Collectds[1].BindAddress != ":1001" {
+		t.Fatalf("unexpected collectd bind address: %s", c.Collectds[1].BindAddress)
 	} else if c.OpenTSDB.BindAddress != ":2000" {
 		t.Fatalf("unexpected opentsdb bind address: %s", c.OpenTSDB.BindAddress)
 	} else if c.UDPs[0].BindAddress != ":4444" {
@@ -111,7 +116,7 @@ protocol = "udp"
 [[graphite]]
 protocol = "tcp"
 
-[collectd]
+[[collectd]]
 bind-address = ":1000"
 
 [opentsdb]

cmd/influxd/run/server.go

@@ -231,7 +231,6 @@ func (s *Server) Open() error {
 	s.appendAdminService(s.config.Admin)
 	s.appendContinuousQueryService(s.config.ContinuousQuery)
 	s.appendHTTPDService(s.config.HTTPD)
-	s.appendCollectdService(s.config.Collectd)
 	if err := s.appendOpenTSDBService(s.config.OpenTSDB); err != nil {
 		return err
 	}
@@ -244,6 +243,9 @@ func (s *Server) Open() error {
 			return err
 		}
 	}
+	for _, g := range s.config.Collectds {
+		s.appendCollectdService(g)
+	}
 
 	s.Subscriber.MetaClient = s.MetaClient
 	s.Subscriber.MetaClient = s.MetaClient

etc/config.sample.toml

@@ -279,12 +279,12 @@ reporting-disabled = false
   # ]
 
 ###
-### [collectd]
+### [[collectd]]
 ###
 ### Controls the listener for collectd data.
 ###
 
-[collectd]
+[[collectd]]
   enabled = false
   # bind-address = ""
   # database = ""

services/collectd/README.md

@@ -22,7 +22,7 @@ Please note that UDP packets larger than the standard size of 1452 are dropped a
 ## Config Example
 
 ```
-[collectd]
+[[collectd]]
   enabled = true
   bind-address = ":25826" # the bind address
   database = "collectd" # Name of the database that will be written to

services/collectd/config.go

@@ -68,3 +68,34 @@ func NewConfig() Config {
 		TypesDB:         DefaultTypesDB,
 	}
 }
+
+// WithDefaults takes the given config and returns a new config with any required
+// default values set.
+func (c *Config) WithDefaults() *Config {
+	d := *c
+	if d.BindAddress == "" {
+		d.BindAddress = DefaultBindAddress
+	}
+	if d.Database == "" {
+		d.Database = DefaultDatabase
+	}
+	if d.RetentionPolicy == "" {
+		d.RetentionPolicy = DefaultRetentionPolicy
+	}
+	if d.BatchSize == 0 {
+		d.BatchSize = DefaultBatchSize
+	}
+	if d.BatchPending == 0 {
+		d.BatchPending = DefaultBatchPending
+	}
+	if d.BatchDuration == 0 {
+		d.BatchDuration = toml.Duration(10 * time.Second)
+	}
+	if d.TypesDB == "" {
+		d.TypesDB = DefaultTypesDB
+	}
+	if d.ReadBuffer == 0 {
+		d.ReadBuffer = DefaultReadBuffer
+	}
+	return &d
+}

services/collectd/service.go

@@ -63,8 +63,11 @@ type Service struct {
 
 // NewService returns a new instance of the collectd service.
 func NewService(c Config) *Service {
+	// Use defaults where necessary.
+	d := c.WithDefaults()
+
 	s := &Service{
-		Config: &c,
+		Config: d,
 		Logger: log.New(os.Stderr, "[collectd] ", log.LstdFlags),
 		err:    make(chan error),
 	}