Aurora plugin

CHANGELOG.md

@@ -67,7 +67,10 @@ plugins, not just statsd.
 - [#1973](https://github.com/influxdata/telegraf/issues/1973): Partial fix: logparser CLF pattern with IPv6 addresses.
 - [#1975](https://github.com/influxdata/telegraf/issues/1975) & [#2102](https://github.com/influxdata/telegraf/issues/2102): Fix thread-safety when using multiple instances of the statsd input plugin.
 - [#2027](https://github.com/influxdata/telegraf/issues/2027): docker input: interface conversion panic fix.
-- [#1814](https://github.com/influxdata/telegraf/issues/1814): snmp: ensure proper context is present on error messages
+- [#1814](https://github.com/influxdata/telegraf/issues/1814): snmp: ensure proper context is present on error messages.
+- [#2299](https://github.com/influxdata/telegraf/issues/2299): opentsdb: add tcp:// prefix if no scheme provided.
+- [#2297](https://github.com/influxdata/telegraf/issues/2297): influx parser: parse line-protocol without newlines.
+- [#2245](https://github.com/influxdata/telegraf/issues/2245): influxdb output: fix field type conflict blocking output buffer.
 
 ## v1.1.2 [2016-12-12]
 

Makefile

@@ -32,6 +32,9 @@ build-for-docker:
 package:
 	./scripts/build.py --package --version="$(VERSION)" --platform=linux --arch=all --upload
 
+package-wavefront:
+	./scripts/build.py --package --version="$(VERSION)" --platform=linux --arch=all
+
 # Get dependencies and use gdm to checkout changesets
 prepare:
 	go get github.com/sparrc/gdm

circle.yml

@@ -4,9 +4,9 @@ machine:
   post:
     - sudo service zookeeper stop
     - go version
-    - go version | grep 1.7.4 || sudo rm -rf /usr/local/go
-    - wget https://storage.googleapis.com/golang/go1.7.4.linux-amd64.tar.gz
-    - sudo tar -C /usr/local -xzf go1.7.4.linux-amd64.tar.gz
+    - go version | grep 1.7.5 || sudo rm -rf /usr/local/go
+    - wget https://storage.googleapis.com/golang/go1.7.5.linux-amd64.tar.gz
+    - sudo tar -C /usr/local -xzf go1.7.5.linux-amd64.tar.gz
     - go version
 
 dependencies:

internal/models/running_output.go

@@ -90,6 +90,9 @@ func NewRunningOutput(
 // AddMetric adds a metric to the output. This function can also write cached
 // points if FlushBufferWhenFull is true.
 func (ro *RunningOutput) AddMetric(m telegraf.Metric) {
+	if m == nil {
+		return
+	}
 	// Filter any tagexclude/taginclude parameters before adding metric
 	if ro.Config.Filter.IsActive() {
 		// In order to filter out tags, we need to create a new metric, since

internal/models/running_output_test.go

@@ -75,6 +75,23 @@ func BenchmarkRunningOutputAddFailWrites(b *testing.B) {
 	}
 }
 
+func TestAddingNilMetric(t *testing.T) {
+	conf := &OutputConfig{
+		Filter: Filter{},
+	}
+
+	m := &mockOutput{}
+	ro := NewRunningOutput("test", m, conf, 1000, 10000)
+
+	ro.AddMetric(nil)
+	ro.AddMetric(nil)
+	ro.AddMetric(nil)
+
+	err := ro.Write()
+	assert.NoError(t, err)
+	assert.Len(t, m.Metrics(), 0)
+}
+
 // Test that NameDrop filters ger properly applied.
 func TestRunningOutput_DropFilter(t *testing.T) {
 	conf := &OutputConfig{

metric/metric.go

@@ -263,7 +263,7 @@ func (m *metric) Fields() map[string]interface{} {
 		case '"':
 			// string field
 			fieldMap[unescape(string(m.fields[i:][0:i1]), "fieldkey")] = unescape(string(m.fields[i:][i2+1:i3-1]), "fieldval")
-		case '0', '1', '2', '3', '4', '5', '6', '7', '8', '9':
+		case '-', '0', '1', '2', '3', '4', '5', '6', '7', '8', '9':
 			// number field
 			switch m.fields[i:][i3-1] {
 			case 'i':

metric/parse_test.go

@@ -44,6 +44,9 @@ cpu,host=foo,datacenter=us-east idle=99,busy=1i,b=true,s="string"
 cpu,host=foo,datacenter=us-east idle=99,busy=1i,b=true,s="string"
 `
 
+const negMetrics = `weather,host=local temp=-99i,temp_float=-99.4 1465839830100400200
+`
+
 // some metrics are invalid
 const someInvalid = `cpu,host=foo,datacenter=us-east usage_idle=99,usage_busy=1
 cpu,host=foo,datacenter=us-east usage_idle=99,usage_busy=1
@@ -85,6 +88,26 @@ func TestParse(t *testing.T) {
 	}
 }
 
+func TestParseNegNumbers(t *testing.T) {
+	metrics, err := Parse([]byte(negMetrics))
+	assert.NoError(t, err)
+	assert.Len(t, metrics, 1)
+
+	assert.Equal(t,
+		map[string]interface{}{
+			"temp":       int64(-99),
+			"temp_float": float64(-99.4),
+		},
+		metrics[0].Fields(),
+	)
+	assert.Equal(t,
+		map[string]string{
+			"host": "local",
+		},
+		metrics[0].Tags(),
+	)
+}
+
 func TestParseErrors(t *testing.T) {
 	start := time.Now()
 	metrics, err := Parse([]byte(someInvalid))

plugins/inputs/all/all.go

@@ -80,4 +80,5 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/inputs/win_perf_counters"
 	_ "github.com/influxdata/telegraf/plugins/inputs/zfs"
 	_ "github.com/influxdata/telegraf/plugins/inputs/zookeeper"
+	_ "github.com/influxdata/telegraf/plugins/inputs/aurora"
 )

plugins/inputs/aurora/aurora.go

@@ -0,0 +1,147 @@
+package aurora
+
+import (
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
+)
+
+type Aurora struct {
+	Timeout    int
+	Master     string
+	HttpPrefix string
+	Numeric bool
+}
+
+var sampleConfig = `
+  ## Timeout, in ms.
+  timeout = 100
+  ## Aurora Master
+  master = "localhost:8081"
+  ## Http Prefix
+  prefix = "http"
+  ## Numeric values only
+  numeric = true
+`
+
+// SampleConfig returns a sample configuration block
+func (a *Aurora) SampleConfig() string {
+	return sampleConfig
+}
+
+// Description just returns a short description of the Mesos plugin
+func (a *Aurora) Description() string {
+	return "Telegraf plugin for gathering metrics from N Apache Aurora Masters"
+}
+
+func (a *Aurora) SetDefaults() {
+	if a.Timeout == 0 {
+		log.Println("I! [aurora] Missing timeout value, setting default value (100ms)")
+		a.Timeout = 1000
+	} else if a.HttpPrefix == "" {
+		log.Println("I! [aurora] Missing http prefix value, setting default value (http)")
+		a.HttpPrefix = "http"
+	}
+}
+
+func convertToNumeric(value string) (interface{}, bool) {
+	var err error
+	var val interface{}
+	if val, err = strconv.ParseFloat(value, 64); err == nil {
+		return val, true
+	}
+	if val, err = strconv.ParseBool(value); err != nil {
+		return val.(bool), false
+	}
+	return val, true
+}
+
+func isJobMetric(key string) bool {
+	// Regex for matching job specific tasks
+	re := regexp.MustCompile("^sla_(.*?)/(.*?)/.*")
+	return re.MatchString(key)
+}
+
+func parseJobSpecificMetric(key string, value interface{}) (map[string]interface{}, map[string]string) {
+	// cut off the sla_
+	key = key[4:]
+	slashSplit := strings.Split(key, "/")
+	role := slashSplit[0]
+	env := slashSplit[1]
+	underscoreIdx := strings.Index(slashSplit[2], "_")
+	job := slashSplit[2][:underscoreIdx]
+	metric := slashSplit[2][underscoreIdx+1:]
+
+	fields := make(map[string]interface{})
+	fields[metric] = value
+
+	tags := make(map[string]string)
+	tags["role"] = role
+	tags["env"] = env
+	tags["job"] = job
+	return fields, tags
+} 
+
+// Gather() metrics from given list of Mesos Masters
+func (a *Aurora) Gather(acc telegraf.Accumulator) error {
+	a.SetDefaults()
+
+	client := http.Client{
+    	Timeout: time.Duration(a.Timeout) * time.Second,
+	}
+	url := fmt.Sprintf("%s://%s/vars", a.HttpPrefix, a.Master)
+	resp, err := client.Get(url)
+	if err != nil {
+		return err
+	}
+
+	data, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	// Map for fields
+	nonJobFields := make(map[string]interface{})
+
+	lines := strings.Split(string(data), "\n")
+	for _, line := range lines {
+		splitIdx := strings.Index(line, " ")
+		if splitIdx == -1 {
+			continue
+		}
+		key := line[:splitIdx]
+		value := line[splitIdx+1:]
+		// If numeric is true and the metric is not numeric then ignore
+		numeric, isNumeric := convertToNumeric(value)
+		if a.Numeric && !isNumeric {
+			continue
+		}
+
+		// If it matches this, then we want to parse it specially because it has jobnames in it
+		if isJobMetric(key) {
+			fields, tags := parseJobSpecificMetric(key, numeric)
+			// Per job there are different tags so need to add a field per line
+			acc.AddFields("aurora", fields, tags)
+		} else {
+			// No tags for other fields so can group add them
+			nonJobFields[key] = numeric
+		}
+	}
+	acc.AddFields("aurora", nonJobFields, make(map[string]string))
+	return nil
+}
+
+func init() {
+	inputs.Add("aurora", func() telegraf.Input {
+		return &Aurora{}
+	})
+}

plugins/inputs/aurora/aurora_test.go

@@ -0,0 +1,130 @@
+package aurora
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+
+	"github.com/influxdata/telegraf/testutil"
+	"github.com/stretchr/testify/assert"
+)
+
+var masterServer *httptest.Server
+
+//sla_role/prod/jobname_
+
+func getRawMetrics() string {
+	return `assigner_launch_failures 0
+cron_job_triggers 240
+sla_cluster_mtta_ms 18
+sla_disk_small_mttr_ms 1029
+sla_cpu_small_mtta_ms 17
+jvm_prop_java.endorsed.dirs /usr/lib/jvm/java-8-openjdk-amd64/jre/lib/endorsed`
+}
+
+func TestMain(m *testing.M) {
+	metrics := getRawMetrics()
+
+	masterRouter := http.NewServeMux()
+	masterRouter.HandleFunc("/vars", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprintf(w, metrics)
+	})
+	masterServer = httptest.NewServer(masterRouter)
+
+	rc := m.Run()
+
+	masterServer.Close()
+	os.Exit(rc)
+}
+
+func TestConvertToNumeric(t *testing.T) {
+	if _, isNumeric := convertToNumeric("0.000"); !isNumeric {
+		t.Fatalf("0.000 should have been numeric")
+	}
+	if _, isNumeric := convertToNumeric("7"); !isNumeric {
+		t.Fatalf("7 should have been numeric")
+	}
+	if boolVal, isNumeric := convertToNumeric("true"); !isNumeric {
+		if val := boolVal.(int); val != 1 {
+			t.Fatalf("true should have been converted to a 1")
+		}
+		t.Fatalf("true should have been numeric")
+	}
+	if boolVal, isNumeric := convertToNumeric("false"); !isNumeric {
+		if val := boolVal.(int); val != 0 {
+			t.Fatalf("true should have been converted to a 0")
+		}
+		t.Fatalf("true should have been numeric")
+	}
+	if _, isNumeric := convertToNumeric("&"); isNumeric {
+		t.Fatalf("& should not be numeric")
+	}
+}
+
+func TestIsJobMetric(t *testing.T) {
+	var notJobMetrics = []string{
+		"assigner_launch_failures",
+		"cron_job_triggers",
+		"sla_cluster_mtta_ms",
+		"sla_disk_small_mttr_ms",
+		"sla_cpu_small_mtta_ms",
+	}
+	for _, metric := range notJobMetrics {
+		if isJobMetric(metric) {
+			t.Fatalf("%s should not be a job metric", metric)
+		}
+	}
+	var isJobMetrics = []string{
+		"sla_role2/prod2/jobname2_job_uptime_50.00_sec",
+	}
+	for _, metric := range isJobMetrics {
+		if !isJobMetric(metric) {
+			t.Fatalf("%s should be a job metric", metric)
+		}
+	}
+}
+
+func TestParseJobSpecificMetric(t *testing.T) {
+	var expectedFields = map[string]interface{}{
+		"job_uptime_50.00_sec": 0,
+	} 
+	var expectedTags = map[string]string {
+		"role": "role2",
+		"env": "prod2",
+		"job": "jobname2",
+	}
+	key := "sla_role2/prod2/jobname2_job_uptime_50.00_sec"
+	value := 0
+	fields, tags := parseJobSpecificMetric(key, value)
+	assert.Equal(t, fields, expectedFields)
+	assert.Equal(t, tags, expectedTags)
+}
+
+func TestAuroraMaster(t *testing.T) {
+	var acc testutil.Accumulator
+
+	m := Aurora{
+		Master: masterServer.Listener.Addr().String(),
+		Timeout: 10,
+		HttpPrefix: "http",
+		Numeric: true,
+	}
+
+	err := m.Gather(&acc)
+	if err != nil {
+		t.Errorf(err.Error())
+	}
+
+	var referenceMetrics = map[string]interface{} {
+		"assigner_launch_failures": 0.0,
+		"cron_job_triggers": 240.0,
+		"sla_cluster_mtta_ms": 18.0,
+		"sla_disk_small_mttr_ms": 1029.0,
+		"sla_cpu_small_mtta_ms": 17.0,
+	}
+
+	acc.AssertContainsFields(t, "aurora", referenceMetrics)
+}