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

Remove flaky test for udp_listener

This commit is contained in:
Daniel Nelson 2019-06-21 18:39:00 -07:00
parent f405bca034
commit aa84011dc3
No known key found for this signature in database
GPG Key ID: CAAD59C9444F6155
1 changed files with 33 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
plugins/inputs/udp_listener/udp_listener_test.go
View File

@ -8,14 +8,11 @@ import (
"log" "log"
"net" "net"
"os" "os"
"runtime"
"strings" "strings"
"testing" "testing"
"github.com/influxdata/telegraf/plugins/parsers" "github.com/influxdata/telegraf/plugins/parsers"
"github.com/influxdata/telegraf/testutil" "github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
@ -42,42 +39,42 @@ func newTestUdpListener() (*UdpListener, chan []byte) {
return listener, in return listener, in
} }
func TestHighTrafficUDP(t *testing.T) { // func TestHighTrafficUDP(t *testing.T) {
listener := UdpListener{ // listener := UdpListener{
ServiceAddress: ":8126", // ServiceAddress: ":8126",
AllowedPendingMessages: 100000, // AllowedPendingMessages: 100000,
} // }
var err error // var err error
listener.parser, err = parsers.NewInfluxParser() // listener.parser, err = parsers.NewInfluxParser()
require.NoError(t, err) // require.NoError(t, err)
acc := &testutil.Accumulator{} // acc := &testutil.Accumulator{}
// send multiple messages to socket // // send multiple messages to socket
err = listener.Start(acc) // err = listener.Start(acc)
require.NoError(t, err) // require.NoError(t, err)
conn, err := net.Dial("udp", "127.0.0.1:8126") // conn, err := net.Dial("udp", "127.0.0.1:8126")
require.NoError(t, err) // require.NoError(t, err)
mlen := int64(len(testMsgs)) // mlen := int64(len(testMsgs))
var sent int64 // var sent int64
for i := 0; i < 20000; i++ { // for i := 0; i < 20000; i++ {
for sent > listener.BytesRecv.Get()+32000 { // for sent > listener.BytesRecv.Get()+32000 {
// more than 32kb sitting in OS buffer, let it drain // // more than 32kb sitting in OS buffer, let it drain
runtime.Gosched() // runtime.Gosched()
} // }
conn.Write([]byte(testMsgs)) // conn.Write([]byte(testMsgs))
sent += mlen // sent += mlen
} // }
for sent > listener.BytesRecv.Get() { // for sent > listener.BytesRecv.Get() {
runtime.Gosched() // runtime.Gosched()
} // }
for len(listener.in) > 0 { // for len(listener.in) > 0 {
runtime.Gosched() // runtime.Gosched()
} // }
listener.Stop() // listener.Stop()
assert.Equal(t, uint64(100000), acc.NMetrics()) // assert.Equal(t, uint64(100000), acc.NMetrics())
} // }
func TestConnectUDP(t *testing.T) { func TestConnectUDP(t *testing.T) {
listener := UdpListener{ listener := UdpListener{