This repository has been archived by the owner on Oct 18, 2023. It is now read-only.
forked from influxdata/telegraf
-
Notifications
You must be signed in to change notification settings - Fork 0
/
udp.go
120 lines (99 loc) · 2.33 KB
/
udp.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
package influxdb
import (
"context"
"fmt"
"log"
"net"
"net/url"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/plugins/serializers/influx"
)
const (
// DefaultMaxPayloadSize is the maximum length of the UDP data payload
DefaultMaxPayloadSize = 512
)
type Dialer interface {
DialContext(ctx context.Context, network, address string) (Conn, error)
}
type Conn interface {
Write(b []byte) (int, error)
Close() error
}
type UDPConfig struct {
MaxPayloadSize int
URL *url.URL
Serializer *influx.Serializer
Dialer Dialer
}
func NewUDPClient(config *UDPConfig) (*udpClient, error) {
if config.URL == nil {
return nil, ErrMissingURL
}
size := config.MaxPayloadSize
if size == 0 {
size = DefaultMaxPayloadSize
}
serializer := config.Serializer
if serializer == nil {
s := influx.NewSerializer()
s.SetMaxLineBytes(config.MaxPayloadSize)
serializer = s
}
dialer := config.Dialer
if dialer == nil {
dialer = &netDialer{net.Dialer{}}
}
client := &udpClient{
url: config.URL,
serializer: serializer,
dialer: dialer,
}
return client, nil
}
type udpClient struct {
conn Conn
dialer Dialer
serializer *influx.Serializer
url *url.URL
}
func (c *udpClient) URL() string {
return c.url.String()
}
func (c *udpClient) Database() string {
return ""
}
func (c *udpClient) Write(ctx context.Context, metrics []telegraf.Metric) error {
if c.conn == nil {
conn, err := c.dialer.DialContext(ctx, c.url.Scheme, c.url.Host)
if err != nil {
return fmt.Errorf("error dialing address [%s]: %s", c.url, err)
}
c.conn = conn
}
for _, metric := range metrics {
octets, err := c.serializer.Serialize(metric)
if err != nil {
// Since we are serializing multiple metrics, don't fail the
// entire batch just because of one unserializable metric.
log.Printf("E! [outputs.influxdb] when writing to [%s] could not serialize metric: %v",
c.URL(), err)
continue
}
_, err = c.conn.Write(octets)
if err != nil {
c.conn.Close()
c.conn = nil
return err
}
}
return nil
}
func (c *udpClient) CreateDatabase(ctx context.Context) error {
return nil
}
type netDialer struct {
net.Dialer
}
func (d *netDialer) DialContext(ctx context.Context, network, address string) (Conn, error) {
return d.Dialer.DialContext(ctx, network, address)
}