Fix length calculation of split metric buffer (#2869)

(cherry picked from commit be03abd464)
This commit is contained in:
Daniel Nelson 2017-05-30 17:38:32 -07:00 committed by Daniel Nelson
parent f6c986b5ca
commit 5417a12ee3
No known key found for this signature in database
GPG Key ID: CAAD59C9444F6155
2 changed files with 74 additions and 7 deletions

View File

@ -2,7 +2,6 @@ package influxdb
import ( import (
"fmt" "fmt"
"io"
"log" "log"
"math/rand" "math/rand"
"strings" "strings"
@ -167,26 +166,29 @@ func (i *InfluxDB) Description() string {
return "Configuration for influxdb server to send metrics to" return "Configuration for influxdb server to send metrics to"
} }
func (i *InfluxDB) getReader(metrics []telegraf.Metric) io.Reader { func (i *InfluxDB) split(metrics []telegraf.Metric) []telegraf.Metric {
if !i.splitPayload { if !i.splitPayload {
return metric.NewReader(metrics) return metrics
} }
splitData := make([]telegraf.Metric, 0) split := make([]telegraf.Metric, 0)
for _, m := range metrics { for _, m := range metrics {
splitData = append(splitData, m.Split(i.UDPPayload)...) split = append(split, m.Split(i.UDPPayload)...)
} }
return metric.NewReader(splitData) return split
} }
// Write will choose a random server in the cluster to write to until a successful write // Write will choose a random server in the cluster to write to until a successful write
// occurs, logging each unsuccessful. If all servers fail, return error. // occurs, logging each unsuccessful. If all servers fail, return error.
func (i *InfluxDB) Write(metrics []telegraf.Metric) error { func (i *InfluxDB) Write(metrics []telegraf.Metric) error {
metrics = i.split(metrics)
bufsize := 0 bufsize := 0
for _, m := range metrics { for _, m := range metrics {
bufsize += m.Len() bufsize += m.Len()
} }
r := i.getReader(metrics)
r := metric.NewReader(metrics)
// This will get set to nil if a successful write occurs // This will get set to nil if a successful write occurs
err := fmt.Errorf("Could not write to any InfluxDB server in cluster") err := fmt.Errorf("Could not write to any InfluxDB server in cluster")

View File

@ -2,10 +2,15 @@ package influxdb
import ( import (
"fmt" "fmt"
"io"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"testing" "testing"
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/plugins/outputs/influxdb/client"
"github.com/influxdata/telegraf/testutil" "github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
@ -58,6 +63,35 @@ func TestUDPInflux(t *testing.T) {
require.NoError(t, i.Close()) require.NoError(t, i.Close())
} }
func TestBasicSplit(t *testing.T) {
c := &MockClient{}
i := InfluxDB{
clients: []client.Client{c},
UDPPayload: 50,
splitPayload: true,
}
// Input metrics:
// test1,tag1=value1 value1=1 value2=2 1257894000000000000\n
//
// Split metrics:
// test1,tag1=value1 value1=1 1257894000000000000\n
// test1,tag1=value1 value2=2 1257894000000000000\n
m, err := metric.New("test1",
map[string]string{"tag1": "value1"},
map[string]interface{}{"value1": 1.0, "value2": 2.0},
time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC),
)
require.NoError(t, err)
metrics := []telegraf.Metric{m}
err = i.Write(metrics)
require.Equal(t, 1, c.writeStreamCalled)
require.Equal(t, 94, c.contentLength)
require.NoError(t, err)
}
func TestHTTPInflux(t *testing.T) { func TestHTTPInflux(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path { switch r.URL.Path {
@ -199,3 +233,34 @@ func TestHTTPError_FieldTypeConflict(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, i.Close()) require.NoError(t, i.Close())
} }
type MockClient struct {
writeStreamCalled int
contentLength int
}
func (m *MockClient) Query(command string) error {
panic("not implemented")
}
func (m *MockClient) Write(b []byte) (int, error) {
panic("not implemented")
}
func (m *MockClient) WriteWithParams(b []byte, params client.WriteParams) (int, error) {
panic("not implemented")
}
func (m *MockClient) WriteStream(b io.Reader, contentLength int) (int, error) {
m.writeStreamCalled++
m.contentLength = contentLength
return 0, nil
}
func (m *MockClient) WriteStreamWithParams(b io.Reader, contentLength int, params client.WriteParams) (int, error) {
panic("not implemented")
}
func (m *MockClient) Close() error {
panic("not implemented")
}