2016-02-25 15:57:28 +00:00
|
|
|
package udp_listener
|
|
|
|
|
|
|
|
import (
|
|
|
|
"log"
|
|
|
|
"net"
|
2016-03-17 16:16:12 +00:00
|
|
|
"strings"
|
2016-02-25 15:57:28 +00:00
|
|
|
"sync"
|
|
|
|
|
|
|
|
"github.com/influxdata/telegraf"
|
|
|
|
"github.com/influxdata/telegraf/plugins/inputs"
|
|
|
|
"github.com/influxdata/telegraf/plugins/parsers"
|
|
|
|
)
|
|
|
|
|
|
|
|
type UdpListener struct {
|
2016-04-07 18:06:56 +00:00
|
|
|
ServiceAddress string
|
|
|
|
// UDPPacketSize is deprecated, it's only here for legacy support
|
|
|
|
// we now always create 1 max size buffer and then copy only what we need
|
|
|
|
// into the in channel
|
|
|
|
// see https://github.com/influxdata/telegraf/pull/992
|
2016-02-25 15:57:28 +00:00
|
|
|
UDPPacketSize int `toml:"udp_packet_size"`
|
|
|
|
AllowedPendingMessages int
|
2016-03-17 16:16:12 +00:00
|
|
|
|
2016-02-25 15:57:28 +00:00
|
|
|
sync.Mutex
|
2016-03-17 16:16:12 +00:00
|
|
|
wg sync.WaitGroup
|
2016-02-25 15:57:28 +00:00
|
|
|
|
|
|
|
in chan []byte
|
|
|
|
done chan struct{}
|
|
|
|
|
|
|
|
parser parsers.Parser
|
|
|
|
|
|
|
|
// Keep the accumulator in this struct
|
|
|
|
acc telegraf.Accumulator
|
2016-03-17 16:16:12 +00:00
|
|
|
|
|
|
|
listener *net.UDPConn
|
2016-02-25 15:57:28 +00:00
|
|
|
}
|
|
|
|
|
2016-04-05 16:37:21 +00:00
|
|
|
// UDP packet limit, see
|
|
|
|
// https://en.wikipedia.org/wiki/User_Datagram_Protocol#Packet_structure
|
2016-04-07 18:06:56 +00:00
|
|
|
const UDP_MAX_PACKET_SIZE int = 64 * 1024
|
2016-02-25 15:57:28 +00:00
|
|
|
|
|
|
|
var dropwarn = "ERROR: Message queue full. Discarding line [%s] " +
|
|
|
|
"You may want to increase allowed_pending_messages in the config\n"
|
|
|
|
|
|
|
|
const sampleConfig = `
|
|
|
|
## Address and port to host UDP listener on
|
2016-03-01 14:53:55 +00:00
|
|
|
service_address = ":8092"
|
2016-02-25 15:57:28 +00:00
|
|
|
|
|
|
|
## Number of UDP messages allowed to queue up. Once filled, the
|
|
|
|
## UDP listener will start dropping packets.
|
|
|
|
allowed_pending_messages = 10000
|
|
|
|
|
2016-03-31 23:50:24 +00:00
|
|
|
## Data format to consume.
|
2016-02-25 15:57:28 +00:00
|
|
|
## Each data format has it's own unique set of configuration options, read
|
|
|
|
## more about them here:
|
|
|
|
## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
|
|
|
|
data_format = "influx"
|
|
|
|
`
|
|
|
|
|
|
|
|
func (u *UdpListener) SampleConfig() string {
|
|
|
|
return sampleConfig
|
|
|
|
}
|
|
|
|
|
|
|
|
func (u *UdpListener) Description() string {
|
|
|
|
return "Generic UDP listener"
|
|
|
|
}
|
|
|
|
|
|
|
|
// All the work is done in the Start() function, so this is just a dummy
|
|
|
|
// function.
|
|
|
|
func (u *UdpListener) Gather(_ telegraf.Accumulator) error {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (u *UdpListener) SetParser(parser parsers.Parser) {
|
|
|
|
u.parser = parser
|
|
|
|
}
|
|
|
|
|
|
|
|
func (u *UdpListener) Start(acc telegraf.Accumulator) error {
|
|
|
|
u.Lock()
|
|
|
|
defer u.Unlock()
|
|
|
|
|
|
|
|
u.acc = acc
|
|
|
|
u.in = make(chan []byte, u.AllowedPendingMessages)
|
|
|
|
u.done = make(chan struct{})
|
|
|
|
|
2016-03-17 16:16:12 +00:00
|
|
|
u.wg.Add(2)
|
2016-02-25 15:57:28 +00:00
|
|
|
go u.udpListen()
|
|
|
|
go u.udpParser()
|
|
|
|
|
|
|
|
log.Printf("Started UDP listener service on %s\n", u.ServiceAddress)
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (u *UdpListener) Stop() {
|
|
|
|
close(u.done)
|
2016-03-17 16:16:12 +00:00
|
|
|
u.listener.Close()
|
|
|
|
u.wg.Wait()
|
2016-02-25 15:57:28 +00:00
|
|
|
close(u.in)
|
|
|
|
log.Println("Stopped UDP listener service on ", u.ServiceAddress)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (u *UdpListener) udpListen() error {
|
2016-03-17 16:16:12 +00:00
|
|
|
defer u.wg.Done()
|
|
|
|
var err error
|
2016-02-25 15:57:28 +00:00
|
|
|
address, _ := net.ResolveUDPAddr("udp", u.ServiceAddress)
|
2016-03-17 16:16:12 +00:00
|
|
|
u.listener, err = net.ListenUDP("udp", address)
|
2016-02-25 15:57:28 +00:00
|
|
|
if err != nil {
|
|
|
|
log.Fatalf("ERROR: ListenUDP - %s", err)
|
|
|
|
}
|
2016-03-17 16:16:12 +00:00
|
|
|
log.Println("UDP server listening on: ", u.listener.LocalAddr().String())
|
2016-02-25 15:57:28 +00:00
|
|
|
|
2016-04-07 18:06:56 +00:00
|
|
|
buf := make([]byte, UDP_MAX_PACKET_SIZE)
|
2016-02-25 15:57:28 +00:00
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-u.done:
|
|
|
|
return nil
|
|
|
|
default:
|
2016-03-17 16:16:12 +00:00
|
|
|
n, _, err := u.listener.ReadFromUDP(buf)
|
|
|
|
if err != nil && !strings.Contains(err.Error(), "closed network") {
|
2016-02-25 15:57:28 +00:00
|
|
|
log.Printf("ERROR: %s\n", err.Error())
|
2016-03-17 16:16:12 +00:00
|
|
|
continue
|
2016-02-25 15:57:28 +00:00
|
|
|
}
|
2016-04-07 18:06:56 +00:00
|
|
|
bufCopy := make([]byte, n)
|
|
|
|
copy(bufCopy, buf[:n])
|
2016-02-25 15:57:28 +00:00
|
|
|
|
|
|
|
select {
|
2016-04-07 18:06:56 +00:00
|
|
|
case u.in <- bufCopy:
|
2016-02-25 15:57:28 +00:00
|
|
|
default:
|
2016-04-07 18:06:56 +00:00
|
|
|
log.Printf(dropwarn, string(bufCopy))
|
2016-02-25 15:57:28 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (u *UdpListener) udpParser() error {
|
2016-03-17 16:16:12 +00:00
|
|
|
defer u.wg.Done()
|
2016-04-05 16:37:21 +00:00
|
|
|
|
|
|
|
var packet []byte
|
2016-04-22 21:47:26 +00:00
|
|
|
var metrics []telegraf.Metric
|
|
|
|
var err error
|
2016-02-25 15:57:28 +00:00
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-u.done:
|
|
|
|
return nil
|
2016-04-05 16:37:21 +00:00
|
|
|
case packet = <-u.in:
|
2016-04-22 21:47:26 +00:00
|
|
|
metrics, err = u.parser.Parse(packet)
|
2016-02-25 15:57:28 +00:00
|
|
|
if err == nil {
|
|
|
|
u.storeMetrics(metrics)
|
|
|
|
} else {
|
|
|
|
log.Printf("Malformed packet: [%s], Error: %s\n", packet, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (u *UdpListener) storeMetrics(metrics []telegraf.Metric) error {
|
|
|
|
u.Lock()
|
|
|
|
defer u.Unlock()
|
|
|
|
for _, m := range metrics {
|
|
|
|
u.acc.AddFields(m.Name(), m.Fields(), m.Tags(), m.Time())
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func init() {
|
|
|
|
inputs.Add("udp_listener", func() telegraf.Input {
|
2016-04-07 18:06:56 +00:00
|
|
|
return &UdpListener{}
|
2016-02-25 15:57:28 +00:00
|
|
|
})
|
|
|
|
}
|