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

Fix udp metric splitting (#2880)

Browse Source
This commit is contained in:
Daniel Nelson
2017-06-05 12:44:29 -07:00
committed by GitHub
parent 3103fc775d
commit c1da0002c0
6 changed files with 204 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
plugins/outputs/influxdb/client/udp.go
View File

@@ -4,6 +4,7 @@ import (
"bytes"
"fmt"
"io"
"log"
"net"
"net/url"
)
@@ -82,10 +83,28 @@ func (c *udpClient) WriteStream(r io.Reader, contentLength int) (int, error) {
if err != io.EOF && err != nil {
return totaln, err
}
nW, err := c.conn.Write(c.buffer[0:nR])
totaln += nW
if err != nil {
return totaln, err
if c.buffer[nR-1] == uint8('\n') {
nW, err := c.conn.Write(c.buffer[0:nR])
totaln += nW
if err != nil {
return totaln, err
}
} else {
log.Printf("E! Could not fit point into UDP payload; dropping")
// Scan forward until next line break to realign.
for {
nR, err := r.Read(c.buffer)
if nR == 0 {
break
}
if err != io.EOF && err != nil {
return totaln, err
}
if c.buffer[nR-1] == uint8('\n') {
break
}
}
}
}
return totaln, nil

81
plugins/outputs/influxdb/client/udp_test.go
View File

@@ -1,7 +1,6 @@
package client
import (
"bytes"
"net"
"testing"
"time"
@@ -10,6 +9,7 @@ import (
"github.com/influxdata/telegraf/metric"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestUDPClient(t *testing.T) {
@@ -72,63 +72,7 @@ func TestUDPClient_Write(t *testing.T) {
pkt := <-packets
assert.Equal(t, "cpu value=99\n", pkt)
metrics := `cpu value=99
cpu value=55
cpu value=44
cpu value=101
cpu value=91
cpu value=92
`
// test sending packet with 6 metrics in a stream.
reader := bytes.NewReader([]byte(metrics))
// contentLength is ignored:
n, err = client.WriteStream(reader, 10)
assert.Equal(t, n, len(metrics))
assert.NoError(t, err)
pkt = <-packets
assert.Equal(t, "cpu value=99\ncpu value=55\ncpu value=44\ncpu value=101\ncpu value=91\ncpu value=92\n", pkt)
//
// Test that UDP packets get broken up properly:
config2 := UDPConfig{
URL: "udp://localhost:8199",
PayloadSize: 25,
}
client2, err := NewUDP(config2)
assert.NoError(t, err)
wp := WriteParams{}
//
// Using Write():
buf := []byte(metrics)
n, err = client2.WriteWithParams(buf, wp)
assert.Equal(t, n, len(metrics))
assert.NoError(t, err)
pkt = <-packets
assert.Equal(t, "cpu value=99\ncpu value=55", pkt)
pkt = <-packets
assert.Equal(t, "\ncpu value=44\ncpu value=1", pkt)
pkt = <-packets
assert.Equal(t, "01\ncpu value=91\ncpu value", pkt)
pkt = <-packets
assert.Equal(t, "=92\n", pkt)
//
// Using WriteStream():
reader = bytes.NewReader([]byte(metrics))
n, err = client2.WriteStreamWithParams(reader, 10, wp)
assert.Equal(t, n, len(metrics))
assert.NoError(t, err)
pkt = <-packets
assert.Equal(t, "cpu value=99\ncpu value=55", pkt)
pkt = <-packets
assert.Equal(t, "\ncpu value=44\ncpu value=1", pkt)
pkt = <-packets
assert.Equal(t, "01\ncpu value=91\ncpu value", pkt)
pkt = <-packets
assert.Equal(t, "=92\n", pkt)
//
// Using WriteStream() & a metric.Reader:
config3 := UDPConfig{
@@ -159,4 +103,27 @@ cpu value=92
assert.Equal(t, "test value=1.1 1484142942000000000\n", pkt)
assert.NoError(t, client.Close())
config = UDPConfig{
URL: "udp://localhost:8199",
PayloadSize: 40,
}
client4, err := NewUDP(config)
assert.NoError(t, err)
ts := time.Unix(1484142943, 0)
m1, _ = metric.New("test", map[string]string{},
map[string]interface{}{"this_is_a_very_long_field_name": 1.1}, ts)
m2, _ = metric.New("test", map[string]string{},
map[string]interface{}{"value": 1.1}, ts)
ms = []telegraf.Metric{m1, m2}
reader := metric.NewReader(ms)
n, err = client4.WriteStream(reader, 0)
assert.NoError(t, err)
require.Equal(t, 35, n)
assert.NoError(t, err)
pkt = <-packets
assert.Equal(t, "test value=1.1 1484142943000000000\n", pkt)
assert.NoError(t, client4.Close())
}

17
plugins/outputs/influxdb/influxdb.go
View File

@@ -46,8 +46,7 @@ type InfluxDB struct {
// Precision is only here for legacy support. It will be ignored.
Precision string
clients []client.Client
splitPayload bool
clients []client.Client
}
var sampleConfig = `
@@ -115,7 +114,6 @@ func (i *InfluxDB) Connect() error {
return fmt.Errorf("Error creating UDP Client [%s]: %s", u, err)
}
i.clients = append(i.clients, c)
i.splitPayload = true
default:
// If URL doesn't start with "udp", assume HTTP client
config := client.HTTPConfig{
@@ -166,22 +164,9 @@ func (i *InfluxDB) Description() string {
return "Configuration for influxdb server to send metrics to"
}
func (i *InfluxDB) split(metrics []telegraf.Metric) []telegraf.Metric {
if !i.splitPayload {
return metrics
}
split := make([]telegraf.Metric, 0)
for _, m := range metrics {
split = append(split, m.Split(i.UDPPayload)...)
}
return split
}
// 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.
func (i *InfluxDB) Write(metrics []telegraf.Metric) error {
metrics = i.split(metrics)
bufsize := 0
for _, m := range metrics {

32
plugins/outputs/influxdb/influxdb_test.go
View File

@@ -6,10 +6,7 @@ import (
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/plugins/outputs/influxdb/client"
"github.com/influxdata/telegraf/testutil"
@@ -63,35 +60,6 @@ func TestUDPInflux(t *testing.T) {
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) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {