Merge pull request #109 from influxdb/pr-107

Merge of PR #107, Allow Telegraf to output data to multiple locations beyond InfluxDB, such as Riemann or Kafka

.gitignore

@@ -1,3 +1,4 @@
 pkg/
 tivan
 .vagrant
+telegraf

CHANGELOG.md

@@ -14,6 +14,7 @@
 - [#98](https://github.com/influxdb/telegraf/pull/98): LeoFS plugin. Thanks @mocchira!
 - [#103](https://github.com/influxdb/telegraf/pull/103): Filter by metric tags. Thanks @srfraser!
 - [#106](https://github.com/influxdb/telegraf/pull/106): Options to filter plugins on startup. Thanks @zepouet!
+- [#107](https://github.com/influxdb/telegraf/pull/107): Multiple outputs beyong influxdb. Thanks @jipperinbham!
 
 ### Bugfixes
 - [#85](https://github.com/influxdb/telegraf/pull/85): Fix GetLocalHost testutil function for mac users

agent.go

@@ -3,17 +3,21 @@ package telegraf
 import (
 	"fmt"
 	"log"
-	"net/url"
 	"os"
 	"sort"
 	"strings"
 	"sync"
 	"time"
 
-	"github.com/influxdb/influxdb/client"
+	"github.com/influxdb/telegraf/outputs"
 	"github.com/influxdb/telegraf/plugins"
 )
 
+type runningOutput struct {
+	name   string
+	output outputs.Output
+}
+
 type runningPlugin struct {
 	name   string
 	plugin plugins.Plugin
@@ -32,9 +36,8 @@ type Agent struct {
 
 	Config *Config
 
+	outputs []*runningOutput
 	plugins []*runningPlugin
-
-	conn *client.Client
 }
 
 // NewAgent returns an Agent struct based off the given Config
@@ -64,38 +67,50 @@ func NewAgent(config *Config) (*Agent, error) {
 	return agent, nil
 }
 
-// Connect connects to the agent's config URL
+// Connect connects to all configured outputs
 func (a *Agent) Connect() error {
-	config := a.Config
+	for _, o := range a.outputs {
+		err := o.output.Connect()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
 
-	u, err := url.Parse(config.URL)
-	if err != nil {
-		return err
+// Close closes the connection to all configured outputs
+func (a *Agent) Close() error {
+	var err error
+	for _, o := range a.outputs {
+		err = o.output.Close()
 	}
+	return err
+}
 
-	c, err := client.NewClient(client.Config{
-		URL:       *u,
-		Username:  config.Username,
-		Password:  config.Password,
-		UserAgent: config.UserAgent,
-		Timeout:   config.Timeout.Duration,
-	})
+// LoadOutputs loads the agent's outputs
+func (a *Agent) LoadOutputs() ([]string, error) {
+	var names []string
 
-	if err != nil {
-		return err
-	}
+	for _, name := range a.Config.OutputsDeclared() {
+		creator, ok := outputs.Outputs[name]
+		if !ok {
+			return nil, fmt.Errorf("Undefined but requested output: %s", name)
+		}
 
-	_, err = c.Query(client.Query{
-		Command: fmt.Sprintf("CREATE DATABASE %s", config.Database),
-	})
+		output := creator()
 
-	if err != nil && !strings.Contains(err.Error(), "database already exists") {
-		log.Fatal(err)
+		err := a.Config.ApplyOutput(name, output)
+		if err != nil {
+			return nil, err
+		}
+
+		a.outputs = append(a.outputs, &runningOutput{name, output})
+		names = append(names, name)
 	}
 
-	a.conn = c
+	sort.Strings(names)
 
-	return nil
+	return names, nil
 }
 
 // LoadPlugins loads the agent's plugins
@@ -174,61 +189,59 @@ func (a *Agent) crankParallel() error {
 
 	close(points)
 
-	var acc BatchPoints
-	acc.Tags = a.Config.Tags
-	acc.Time = time.Now()
-	acc.Database = a.Config.Database
+	var bp BatchPoints
+	bp.Time = time.Now()
+	bp.Tags = a.Config.Tags
 
 	for sub := range points {
-		acc.Points = append(acc.Points, sub.Points...)
+		bp.Points = append(bp.Points, sub.Points...)
 	}
 
-	_, err := a.conn.Write(acc.BatchPoints)
-	return err
+	return a.flush(&bp)
 }
 
 func (a *Agent) crank() error {
-	var acc BatchPoints
+	var bp BatchPoints
 
-	acc.Debug = a.Debug
+	bp.Debug = a.Debug
 
 	for _, plugin := range a.plugins {
-		acc.Prefix = plugin.name + "_"
-		acc.Config = plugin.config
-		err := plugin.plugin.Gather(&acc)
+		bp.Prefix = plugin.name + "_"
+		bp.Config = plugin.config
+		err := plugin.plugin.Gather(&bp)
 		if err != nil {
 			return err
 		}
 	}
 
-	acc.Tags = a.Config.Tags
-	acc.Time = time.Now()
-	acc.Database = a.Config.Database
+	bp.Time = time.Now()
+	bp.Tags = a.Config.Tags
 
-	_, err := a.conn.Write(acc.BatchPoints)
-	return err
+	return a.flush(&bp)
 }
 
 func (a *Agent) crankSeparate(shutdown chan struct{}, plugin *runningPlugin) error {
 	ticker := time.NewTicker(plugin.config.Interval)
 
 	for {
-		var acc BatchPoints
+		var bp BatchPoints
 
-		acc.Debug = a.Debug
+		bp.Debug = a.Debug
 
-		acc.Prefix = plugin.name + "_"
-		acc.Config = plugin.config
-		err := plugin.plugin.Gather(&acc)
+		bp.Prefix = plugin.name + "_"
+		bp.Config = plugin.config
+		err := plugin.plugin.Gather(&bp)
 		if err != nil {
 			return err
 		}
 
-		acc.Tags = a.Config.Tags
-		acc.Time = time.Now()
-		acc.Database = a.Config.Database
+		bp.Tags = a.Config.Tags
+		bp.Time = time.Now()
 
-		a.conn.Write(acc.BatchPoints)
+		err = a.flush(&bp)
+		if err != nil {
+			return err
+		}
 
 		select {
 		case <-shutdown:
@@ -239,6 +252,22 @@ func (a *Agent) crankSeparate(shutdown chan struct{}, plugin *runningPlugin) err
 	}
 }
 
+func (a *Agent) flush(bp *BatchPoints) error {
+	var wg sync.WaitGroup
+	var outerr error
+	for _, o := range a.outputs {
+		wg.Add(1)
+		go func(ro *runningOutput) {
+			defer wg.Done()
+			outerr = ro.output.Write(bp.BatchPoints)
+		}(o)
+	}
+
+	wg.Wait()
+
+	return outerr
+}
+
 // TestAllPlugins verifies that we can 'Gather' from all plugins with the
 // default configuration
 func (a *Agent) TestAllPlugins() error {
@@ -297,13 +326,6 @@ func (a *Agent) Test() error {
 
 // Run runs the agent daemon, gathering every Interval
 func (a *Agent) Run(shutdown chan struct{}) error {
-	if a.conn == nil {
-		err := a.Connect()
-		if err != nil {
-			return err
-		}
-	}
-
 	var wg sync.WaitGroup
 
 	for _, plugin := range a.plugins {

cmd/telegraf/telegraf.go

@@ -59,6 +59,11 @@ func main() {
 		ag.Debug = true
 	}
 
+	outputs, err := ag.LoadOutputs()
+	if err != nil {
+		log.Fatal(err)
+	}
+
 	plugins, err := ag.LoadPlugins(*fPLuginsFilter)
 	if err != nil {
 		log.Fatal(err)
@@ -99,18 +104,14 @@ func main() {
 	}()
 
 	log.Printf("Starting Telegraf (version %s)\n", Version)
+	log.Printf("Loaded outputs: %s", strings.Join(outputs, " "))
 	log.Printf("Loaded plugins: %s", strings.Join(plugins, " "))
 	if ag.Debug {
 		log.Printf("Debug: enabled")
 		log.Printf("Agent Config: Interval:%s, Debug:%#v, Hostname:%#v\n",
 			ag.Interval, ag.Debug, ag.Hostname)
 	}
 
-	if config.URL != "" {
-		log.Printf("Sending metrics to: %s", config.URL)
-		log.Printf("Tags enabled: %v", config.ListTags())
-	}
-
 	if *fPidfile != "" {
 		f, err := os.Create(*fPidfile)
 		if err != nil {