dom
/
telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Ran go fmt on opentsdb sources.

This commit is contained in:
Eric 2016-07-22 08:29:55 -04:00
parent 71c53841c3
commit 8ab9cc29af
3 changed files with 104 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
plugins/outputs/opentsdb/opentsdb.go
View File

@ -17,7 +17,7 @@ type OpenTSDB struct {
Host string Host string
Port int Port int
UseHttp bool UseHttp bool
BatchSize int BatchSize int
Debug bool Debug bool
@ -47,6 +47,7 @@ var sampleConfig = `
## Debug true - Prints OpenTSDB communication ## Debug true - Prints OpenTSDB communication
debug = false debug = false
` `
type TagSet map[string]string type TagSet map[string]string
func (t TagSet) ToLineFormat() string { func (t TagSet) ToLineFormat() string {
@ -89,10 +90,10 @@ func (o *OpenTSDB) Write(metrics []telegraf.Metric) error {
func (o *OpenTSDB) WriteHttp(metrics []telegraf.Metric) error { func (o *OpenTSDB) WriteHttp(metrics []telegraf.Metric) error {
http := openTSDBHttp{ http := openTSDBHttp{
Host: o.Host, Host: o.Host,
Port: o.Port, Port: o.Port,
BatchSize: o.BatchSize, BatchSize: o.BatchSize,
Debug: o.Debug, Debug: o.Debug,
} }
for _, m := range metrics { for _, m := range metrics {
@ -106,21 +107,21 @@ func (o *OpenTSDB) WriteHttp(metrics []telegraf.Metric) error {
continue continue
} }
metric := &HttpMetric{ metric := &HttpMetric{
Metric: sanitizedChars.Replace(fmt.Sprintf("%s%s_%s", Metric: sanitizedChars.Replace(fmt.Sprintf("%s%s_%s",
o.Prefix, m.Name(), fieldName)), o.Prefix, m.Name(), fieldName)),
Tags: tags, Tags: tags,
Timestamp: now, Timestamp: now,
Value: metricValue, Value: metricValue,
} }
if err:= http.sendDataPoint(metric); err != nil { if err := http.sendDataPoint(metric); err != nil {
return err return err
} }
} }
} }
if err:= http.flush(); err != nil { if err := http.flush(); err != nil {
return err return err
} }
@ -149,7 +150,7 @@ func (o *OpenTSDB) WriteTelnet(metrics []telegraf.Metric) error {
} }
messageLine := fmt.Sprintf("put %s %v %s %s\n", messageLine := fmt.Sprintf("put %s %v %s %s\n",
sanitizedChars.Replace(fmt.Sprintf("%s%s_%s",o.Prefix, m.Name(), fieldName)), sanitizedChars.Replace(fmt.Sprintf("%s%s_%s", o.Prefix, m.Name(), fieldName)),
now, metricValue, tags) now, metricValue, tags)
if o.Debug { if o.Debug {

151
plugins/outputs/opentsdb/opentsdb_http.go
View File

@ -1,130 +1,129 @@
package opentsdb package opentsdb
import ( import (
"fmt" "bytes"
"encoding/json" "compress/gzip"
"encoding/json"
"fmt"
"io" "io"
"io/ioutil" "io/ioutil"
"log"
"net/http" "net/http"
"net/http/httputil" "net/http/httputil"
"net/url" "net/url"
"bytes"
"compress/gzip"
"log"
) )
type HttpMetric struct { type HttpMetric struct {
Metric string `json:"metric"` Metric string `json:"metric"`
Timestamp int64 `json:"timestamp"` Timestamp int64 `json:"timestamp"`
Value string `json:"value"` Value string `json:"value"`
Tags map[string]string `json:"tags"` Tags map[string]string `json:"tags"`
} }
type openTSDBHttp struct { type openTSDBHttp struct {
Host string Host string
Port int Port int
BatchSize int BatchSize int
Debug bool Debug bool
metricCounter int metricCounter int
body requestBody body requestBody
} }
type requestBody struct { type requestBody struct {
b bytes.Buffer b bytes.Buffer
g *gzip.Writer g *gzip.Writer
dbgB bytes.Buffer dbgB bytes.Buffer
w io.Writer w io.Writer
enc *json.Encoder enc *json.Encoder
empty bool empty bool
} }
func (r *requestBody) reset(debug bool) { func (r *requestBody) reset(debug bool) {
r.b.Reset() r.b.Reset()
r.dbgB.Reset() r.dbgB.Reset()
if r.g == nil { if r.g == nil {
r.g = gzip.NewWriter(&r.b) r.g = gzip.NewWriter(&r.b)
} else { } else {
r.g.Reset(&r.b) r.g.Reset(&r.b)
} }
if debug { if debug {
r.w = io.MultiWriter(r.g, &r.dbgB) r.w = io.MultiWriter(r.g, &r.dbgB)
} else { } else {
r.w = r.g r.w = r.g
} }
r.enc = json.NewEncoder(r.w) r.enc = json.NewEncoder(r.w)
io.WriteString(r.w, "[") io.WriteString(r.w, "[")
r.empty = true r.empty = true
} }
func (r *requestBody) addMetric(metric *HttpMetric) error { func (r *requestBody) addMetric(metric *HttpMetric) error {
if !r.empty { if !r.empty {
io.WriteString(r.w, ",") io.WriteString(r.w, ",")
} }
if err := r.enc.Encode(metric); err != nil { if err := r.enc.Encode(metric); err != nil {
return fmt.Errorf("Metric serialization error %s", err.Error()) return fmt.Errorf("Metric serialization error %s", err.Error())
} }
r.empty = false r.empty = false
return nil return nil
} }
func (r *requestBody) close() error { func (r *requestBody) close() error {
io.WriteString(r.w, "]") io.WriteString(r.w, "]")
if err := r.g.Close(); err != nil { if err := r.g.Close(); err != nil {
return fmt.Errorf("Error when closing gzip writer: %s", err.Error()) return fmt.Errorf("Error when closing gzip writer: %s", err.Error())
} }
return nil return nil
} }
func (o *openTSDBHttp) sendDataPoint(metric *HttpMetric) error { func (o *openTSDBHttp) sendDataPoint(metric *HttpMetric) error {
if o.metricCounter == 0 { if o.metricCounter == 0 {
o.body.reset(o.Debug) o.body.reset(o.Debug)
} }
if err := o.body.addMetric(metric); err != nil { if err := o.body.addMetric(metric); err != nil {
return err return err
} }
o.metricCounter++ o.metricCounter++
if o.metricCounter == o.BatchSize { if o.metricCounter == o.BatchSize {
if err := o.flush(); err != nil { if err := o.flush(); err != nil {
return err return err
} }
o.metricCounter = 0 o.metricCounter = 0
} }
return nil return nil
} }
func (o *openTSDBHttp) flush() error { func (o *openTSDBHttp) flush() error {
if o.metricCounter == 0 { if o.metricCounter == 0 {
return nil return nil
} }
o.body.close() o.body.close()
u := url.URL { u := url.URL{
Scheme: "http", Scheme: "http",
Host: fmt.Sprintf("%s:%d", o.Host, o.Port), Host: fmt.Sprintf("%s:%d", o.Host, o.Port),
Path: "/api/put", Path: "/api/put",
} }
if (o.Debug) { if o.Debug {
u.RawQuery = "details" u.RawQuery = "details"
} }
@ -135,7 +134,7 @@ func (o *openTSDBHttp) flush() error {
req.Header.Set("Content-Type", "applicaton/json") req.Header.Set("Content-Type", "applicaton/json")
req.Header.Set("Content-Encoding", "gzip") req.Header.Set("Content-Encoding", "gzip")
if (o.Debug) { if o.Debug {
dump, err := httputil.DumpRequestOut(req, false) dump, err := httputil.DumpRequestOut(req, false)
if err != nil { if err != nil {
return fmt.Errorf("Error when dumping request: %s", err.Error()) return fmt.Errorf("Error when dumping request: %s", err.Error())
@ -145,23 +144,23 @@ func (o *openTSDBHttp) flush() error {
fmt.Printf("Body:\n%s\n\n", o.body.dbgB.String()) fmt.Printf("Body:\n%s\n\n", o.body.dbgB.String())
} }
resp, err := http.DefaultClient.Do(req) resp, err := http.DefaultClient.Do(req)
if err != nil { if err != nil {
return fmt.Errorf("Error when sending metrics: %s", err.Error()) return fmt.Errorf("Error when sending metrics: %s", err.Error())
} }
defer resp.Body.Close() defer resp.Body.Close()
if o.Debug { if o.Debug {
dump, err := httputil.DumpResponse(resp, true) dump, err := httputil.DumpResponse(resp, true)
if err != nil { if err != nil {
return fmt.Errorf("Error when dumping response: %s", err.Error()) return fmt.Errorf("Error when dumping response: %s", err.Error())
} }
fmt.Printf("Received response\n%s\n\n", dump) fmt.Printf("Received response\n%s\n\n", dump)
} else { } else {
// Important so http client reuse connection for next request if need be. // Important so http client reuse connection for next request if need be.
io.Copy(ioutil.Discard, resp.Body) io.Copy(ioutil.Discard, resp.Body)
} }
if resp.StatusCode/100 != 2 { if resp.StatusCode/100 != 2 {
if resp.StatusCode/100 == 4 { if resp.StatusCode/100 == 4 {

31
plugins/outputs/opentsdb/opentsdb_test.go
View File

@ -1,14 +1,14 @@
package opentsdb package opentsdb
import ( import (
"reflect"
"testing"
"fmt" "fmt"
"net" "net"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"net/url" "net/url"
"reflect"
"strconv" "strconv"
"testing"
"github.com/influxdata/telegraf" "github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/testutil" "github.com/influxdata/telegraf/testutil"
@ -77,18 +77,17 @@ func TestBuildTagsTelnet(t *testing.T) {
func BenchmarkHttpSend(b *testing.B) { func BenchmarkHttpSend(b *testing.B) {
const BatchSize = 50 const BatchSize = 50
const MetricsCount = 4*BatchSize const MetricsCount = 4 * BatchSize
metrics := make([]telegraf.Metric, MetricsCount) metrics := make([]telegraf.Metric, MetricsCount)
for i:=0; i<MetricsCount; i++ { for i := 0; i < MetricsCount; i++ {
metrics[i] = testutil.TestMetric(1.0) metrics[i] = testutil.TestMetric(1.0)
} }
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json") w.Header().Set("Content-Type", "application/json")
fmt.Fprintln(w, "{}") fmt.Fprintln(w, "{}")
})) }))
defer ts.Close() defer ts.Close()
u, err := url.Parse(ts.URL) u, err := url.Parse(ts.URL)
if err != nil { if err != nil {
@ -103,17 +102,17 @@ func BenchmarkHttpSend(b *testing.B) {
} }
o := &OpenTSDB{ o := &OpenTSDB{
Host: host, Host: host,
Port: port, Port: port,
Prefix: "", Prefix: "",
UseHttp: true, UseHttp: true,
BatchSize: BatchSize, BatchSize: BatchSize,
} }
b.ResetTimer() b.ResetTimer()
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
o.Write(metrics) o.Write(metrics)
} }
} }
// func TestWrite(t *testing.T) { // func TestWrite(t *testing.T) {