CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: muting tests for udp_listener (#9578)

This commit is contained in:
Sebastian Spaink 2021-08-05 10:59:51 -05:00 committed by GitHub
parent 249bcd25d5
commit fa77d2fb07
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 169 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,213 +1,201 @@
package udp_listener package udp_listener
import ( // This plugin will become officially deprecated in 2.0
"bufio" // These tests have been randomly failing the nightly tests, can't remove plugin until breaking changes are allowed to be merged
"bytes" // See this issue for more information: https://github.com/influxdata/telegraf/issues/9478
"fmt"
"io/ioutil"
"log"
"net"
"os"
"strings"
"testing"
"github.com/influxdata/telegraf/plugins/parsers" // const (
"github.com/influxdata/telegraf/testutil" // testMsg = "cpu_load_short,host=server01 value=12.0 1422568543702900257\n"
"github.com/stretchr/testify/require"
)
const ( // testMsgs = `
testMsg = "cpu_load_short,host=server01 value=12.0 1422568543702900257\n" // cpu_load_short,host=server02 value=12.0 1422568543702900257
// cpu_load_short,host=server03 value=12.0 1422568543702900257
// cpu_load_short,host=server04 value=12.0 1422568543702900257
// cpu_load_short,host=server05 value=12.0 1422568543702900257
// cpu_load_short,host=server06 value=12.0 1422568543702900257
// `
// )
testMsgs = ` // func newTestUDPListener() (*UDPListener, chan []byte) {
cpu_load_short,host=server02 value=12.0 1422568543702900257 // in := make(chan []byte, 1500)
cpu_load_short,host=server03 value=12.0 1422568543702900257 // listener := &UDPListener{
cpu_load_short,host=server04 value=12.0 1422568543702900257 // Log: testutil.Logger{},
cpu_load_short,host=server05 value=12.0 1422568543702900257 // ServiceAddress: ":8125",
cpu_load_short,host=server06 value=12.0 1422568543702900257 // AllowedPendingMessages: 10000,
` // in: in,
) // done: make(chan struct{}),
func newTestUDPListener() (*UDPListener, chan []byte) {
in := make(chan []byte, 1500)
listener := &UDPListener{
Log: testutil.Logger{},
ServiceAddress: ":8125",
AllowedPendingMessages: 10000,
in: in,
done: make(chan struct{}),
}
return listener, in
}
// func TestHighTrafficUDP(t *testing.T) {
// listener := UDPListener{
// ServiceAddress: ":8126",
// AllowedPendingMessages: 100000,
// } // }
// var err error // return listener, in
// listener.parser, err = parsers.NewInfluxParser() // }
// require.NoError(t, err)
// // func TestHighTrafficUDP(t *testing.T) {
// // listener := UDPListener{
// // ServiceAddress: ":8126",
// // AllowedPendingMessages: 100000,
// // }
// // var err error
// // listener.parser, err = parsers.NewInfluxParser()
// // require.NoError(t, err)
// // acc := &testutil.Accumulator{}
// // // send multiple messages to socket
// // err = listener.Start(acc)
// // require.NoError(t, err)
// // conn, err := net.Dial("udp", "127.0.0.1:8126")
// // require.NoError(t, err)
// // mlen := int64(len(testMsgs))
// // var sent int64
// // for i := 0; i < 20000; i++ {
// // for sent > listener.BytesRecv.Get()+32000 {
// // // more than 32kb sitting in OS buffer, let it drain
// // runtime.Gosched()
// // }
// // conn.Write([]byte(testMsgs))
// // sent += mlen
// // }
// // for sent > listener.BytesRecv.Get() {
// // runtime.Gosched()
// // }
// // for len(listener.in) > 0 {
// // runtime.Gosched()
// // }
// // listener.Stop()
// // assert.Equal(t, uint64(100000), acc.NMetrics())
// // }
// func TestConnectUDP(t *testing.T) {
// listener := UDPListener{
// Log: testutil.Logger{},
// ServiceAddress: ":8127",
// AllowedPendingMessages: 10000,
// }
// listener.parser, _ = parsers.NewInfluxParser()
// acc := &testutil.Accumulator{} // acc := &testutil.Accumulator{}
// require.NoError(t, listener.Start(acc))
// defer listener.Stop()
// conn, err := net.Dial("udp", "127.0.0.1:8127")
// require.NoError(t, err)
// // send single message to socket
// _, err = fmt.Fprint(conn, testMsg)
// require.NoError(t, err)
// acc.Wait(1)
// acc.AssertContainsTaggedFields(t, "cpu_load_short",
// map[string]interface{}{"value": float64(12)},
// map[string]string{"host": "server01"},
// )
// // send multiple messages to socket // // send multiple messages to socket
// err = listener.Start(acc) // _, err = fmt.Fprint(conn, testMsgs)
// require.NoError(t, err) // require.NoError(t, err)
// acc.Wait(6)
// conn, err := net.Dial("udp", "127.0.0.1:8126") // hostTags := []string{"server02", "server03",
// require.NoError(t, err) // "server04", "server05", "server06"}
// mlen := int64(len(testMsgs)) // for _, hostTag := range hostTags {
// var sent int64 // acc.AssertContainsTaggedFields(t, "cpu_load_short",
// for i := 0; i < 20000; i++ { // map[string]interface{}{"value": float64(12)},
// for sent > listener.BytesRecv.Get()+32000 { // map[string]string{"host": hostTag},
// // more than 32kb sitting in OS buffer, let it drain // )
// runtime.Gosched()
// } // }
// conn.Write([]byte(testMsgs))
// sent += mlen
// }
// for sent > listener.BytesRecv.Get() {
// runtime.Gosched()
// }
// for len(listener.in) > 0 {
// runtime.Gosched()
// }
// listener.Stop()
// assert.Equal(t, uint64(100000), acc.NMetrics())
// } // }
func TestConnectUDP(t *testing.T) { // func TestRunParser(t *testing.T) {
listener := UDPListener{ // log.SetOutput(ioutil.Discard)
Log: testutil.Logger{}, // var testmsg = []byte("cpu_load_short,host=server01 value=12.0 1422568543702900257\n")
ServiceAddress: ":8127",
AllowedPendingMessages: 10000,
}
listener.parser, _ = parsers.NewInfluxParser()
acc := &testutil.Accumulator{} // listener, in := newTestUDPListener()
require.NoError(t, listener.Start(acc)) // acc := testutil.Accumulator{}
defer listener.Stop() // listener.acc = &acc
// defer close(listener.done)
conn, err := net.Dial("udp", "127.0.0.1:8127") // listener.parser, _ = parsers.NewInfluxParser()
require.NoError(t, err) // listener.wg.Add(1)
// go listener.udpParser()
// send single message to socket // in <- testmsg
_, err = fmt.Fprint(conn, testMsg) // require.NoError(t, listener.Gather(&acc))
require.NoError(t, err)
acc.Wait(1)
acc.AssertContainsTaggedFields(t, "cpu_load_short",
map[string]interface{}{"value": float64(12)},
map[string]string{"host": "server01"},
)
// send multiple messages to socket // acc.Wait(1)
_, err = fmt.Fprint(conn, testMsgs) // acc.AssertContainsTaggedFields(t, "cpu_load_short",
require.NoError(t, err) // map[string]interface{}{"value": float64(12)},
acc.Wait(6) // map[string]string{"host": "server01"},
hostTags := []string{"server02", "server03", // )
"server04", "server05", "server06"} // }
for _, hostTag := range hostTags {
acc.AssertContainsTaggedFields(t, "cpu_load_short",
map[string]interface{}{"value": float64(12)},
map[string]string{"host": hostTag},
)
}
}
func TestRunParser(t *testing.T) { // func TestRunParserInvalidMsg(_ *testing.T) {
log.SetOutput(ioutil.Discard) // log.SetOutput(ioutil.Discard)
var testmsg = []byte("cpu_load_short,host=server01 value=12.0 1422568543702900257\n") // var testmsg = []byte("cpu_load_short")
listener, in := newTestUDPListener() // listener, in := newTestUDPListener()
acc := testutil.Accumulator{} // acc := testutil.Accumulator{}
listener.acc = &acc // listener.acc = &acc
defer close(listener.done) // defer close(listener.done)
listener.parser, _ = parsers.NewInfluxParser() // listener.parser, _ = parsers.NewInfluxParser()
listener.wg.Add(1) // listener.wg.Add(1)
go listener.udpParser() // go listener.udpParser()
in <- testmsg // buf := bytes.NewBuffer(nil)
require.NoError(t, listener.Gather(&acc)) // log.SetOutput(buf)
// defer log.SetOutput(os.Stderr)
// in <- testmsg
acc.Wait(1) // scnr := bufio.NewScanner(buf)
acc.AssertContainsTaggedFields(t, "cpu_load_short", // for scnr.Scan() {
map[string]interface{}{"value": float64(12)}, // if strings.Contains(scnr.Text(), fmt.Sprintf(malformedwarn, 1)) {
map[string]string{"host": "server01"}, // break
) // }
} // }
// }
func TestRunParserInvalidMsg(_ *testing.T) { // func TestRunParserGraphiteMsg(t *testing.T) {
log.SetOutput(ioutil.Discard) // log.SetOutput(ioutil.Discard)
var testmsg = []byte("cpu_load_short") // var testmsg = []byte("cpu.load.graphite 12 1454780029")
listener, in := newTestUDPListener() // listener, in := newTestUDPListener()
acc := testutil.Accumulator{} // acc := testutil.Accumulator{}
listener.acc = &acc // listener.acc = &acc
defer close(listener.done) // defer close(listener.done)
listener.parser, _ = parsers.NewInfluxParser() // listener.parser, _ = parsers.NewGraphiteParser("_", []string{}, nil)
listener.wg.Add(1) // listener.wg.Add(1)
go listener.udpParser() // go listener.udpParser()
buf := bytes.NewBuffer(nil) // in <- testmsg
log.SetOutput(buf) // require.NoError(t, listener.Gather(&acc))
defer log.SetOutput(os.Stderr)
in <- testmsg
scnr := bufio.NewScanner(buf) // acc.Wait(1)
for scnr.Scan() { // acc.AssertContainsFields(t, "cpu_load_graphite",
if strings.Contains(scnr.Text(), fmt.Sprintf(malformedwarn, 1)) { // map[string]interface{}{"value": float64(12)})
break // }
}
}
}
func TestRunParserGraphiteMsg(t *testing.T) { // func TestRunParserJSONMsg(t *testing.T) {
log.SetOutput(ioutil.Discard) // log.SetOutput(ioutil.Discard)
var testmsg = []byte("cpu.load.graphite 12 1454780029") // var testmsg = []byte("{\"a\": 5, \"b\": {\"c\": 6}}\n")
listener, in := newTestUDPListener() // listener, in := newTestUDPListener()
acc := testutil.Accumulator{} // acc := testutil.Accumulator{}
listener.acc = &acc // listener.acc = &acc
defer close(listener.done) // defer close(listener.done)
listener.parser, _ = parsers.NewGraphiteParser("_", []string{}, nil) // listener.parser, _ = parsers.NewParser(&parsers.Config{
listener.wg.Add(1) // DataFormat: "json",
go listener.udpParser() // MetricName: "udp_json_test",
// })
// listener.wg.Add(1)
// go listener.udpParser()
in <- testmsg // in <- testmsg
require.NoError(t, listener.Gather(&acc)) // require.NoError(t, listener.Gather(&acc))
acc.Wait(1) // acc.Wait(1)
acc.AssertContainsFields(t, "cpu_load_graphite", // acc.AssertContainsFields(t, "udp_json_test",
map[string]interface{}{"value": float64(12)}) // map[string]interface{}{
} // "a": float64(5),
// "b_c": float64(6),
func TestRunParserJSONMsg(t *testing.T) { // })
log.SetOutput(ioutil.Discard) // }
var testmsg = []byte("{\"a\": 5, \"b\": {\"c\": 6}}\n")
listener, in := newTestUDPListener()
acc := testutil.Accumulator{}
listener.acc = &acc
defer close(listener.done)
listener.parser, _ = parsers.NewParser(&parsers.Config{
DataFormat: "json",
MetricName: "udp_json_test",
})
listener.wg.Add(1)
go listener.udpParser()
in <- testmsg
require.NoError(t, listener.Gather(&acc))
acc.Wait(1)
acc.AssertContainsFields(t, "udp_json_test",
map[string]interface{}{
"a": float64(5),
"b_c": float64(6),
})
}