forked from influxdata/influxdb
-
Notifications
You must be signed in to change notification settings - Fork 0
/
service.go
314 lines (267 loc) · 7.44 KB
/
service.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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
// Package udp provides the UDP input service for InfluxDB.
package udp // import "github.com/influxdata/influxdb/services/udp"
import (
"errors"
"net"
"sync"
"sync/atomic"
"time"
"github.com/influxdata/influxdb/logger"
"github.com/influxdata/influxdb/models"
"github.com/influxdata/influxdb/services/meta"
"github.com/influxdata/influxdb/tsdb"
"go.uber.org/zap"
)
const (
// Arbitrary, testing indicated that this doesn't typically get over 10
parserChanLen = 1000
// MaxUDPPayload is largest payload size the UDP service will accept.
MaxUDPPayload = 64 * 1024
)
// statistics gathered by the UDP package.
const (
statPointsReceived = "pointsRx"
statBytesReceived = "bytesRx"
statPointsParseFail = "pointsParseFail"
statReadFail = "readFail"
statBatchesTransmitted = "batchesTx"
statPointsTransmitted = "pointsTx"
statBatchesTransmitFail = "batchesTxFail"
)
// Service is a UDP service that will listen for incoming packets of line protocol.
type Service struct {
conn *net.UDPConn
addr *net.UDPAddr
wg sync.WaitGroup
mu sync.RWMutex
ready bool // Has the required database been created?
done chan struct{} // Is the service closing or closed?
parserChan chan []byte
batcher *tsdb.PointBatcher
config Config
PointsWriter interface {
WritePointsPrivileged(database, retentionPolicy string, consistencyLevel models.ConsistencyLevel, points []models.Point) error
}
MetaClient interface {
CreateDatabase(name string) (*meta.DatabaseInfo, error)
}
Logger *zap.Logger
stats *Statistics
defaultTags models.StatisticTags
}
// NewService returns a new instance of Service.
func NewService(c Config) *Service {
d := *c.WithDefaults()
return &Service{
config: d,
parserChan: make(chan []byte, parserChanLen),
Logger: zap.NewNop(),
stats: &Statistics{},
defaultTags: models.StatisticTags{"bind": d.BindAddress},
}
}
// Open starts the service.
func (s *Service) Open() (err error) {
s.mu.Lock()
defer s.mu.Unlock()
if !s.closed() {
return nil // Already open.
}
s.done = make(chan struct{})
if s.config.BindAddress == "" {
return errors.New("bind address has to be specified in config")
}
if s.config.Database == "" {
return errors.New("database has to be specified in config")
}
s.addr, err = net.ResolveUDPAddr("udp", s.config.BindAddress)
if err != nil {
s.Logger.Info("Failed to resolve UDP address",
zap.String("bind_address", s.config.BindAddress), zap.Error(err))
return err
}
s.conn, err = net.ListenUDP("udp", s.addr)
if err != nil {
s.Logger.Info("Failed to set up UDP listener",
zap.Stringer("addr", s.addr), zap.Error(err))
return err
}
if s.config.ReadBuffer != 0 {
err = s.conn.SetReadBuffer(s.config.ReadBuffer)
if err != nil {
s.Logger.Info("Failed to set UDP read buffer",
zap.Int("buffer_size", s.config.ReadBuffer), zap.Error(err))
return err
}
}
s.batcher = tsdb.NewPointBatcher(s.config.BatchSize, s.config.BatchPending, time.Duration(s.config.BatchTimeout))
s.batcher.Start()
s.Logger.Info("Started listening on UDP", zap.String("addr", s.config.BindAddress))
s.wg.Add(3)
go s.serve()
go s.parser()
go s.writer()
return nil
}
// Statistics maintains statistics for the UDP service.
type Statistics struct {
PointsReceived int64
BytesReceived int64
PointsParseFail int64
ReadFail int64
BatchesTransmitted int64
PointsTransmitted int64
BatchesTransmitFail int64
}
// Statistics returns statistics for periodic monitoring.
func (s *Service) Statistics(tags map[string]string) []models.Statistic {
return []models.Statistic{{
Name: "udp",
Tags: s.defaultTags.Merge(tags),
Values: map[string]interface{}{
statPointsReceived: atomic.LoadInt64(&s.stats.PointsReceived),
statBytesReceived: atomic.LoadInt64(&s.stats.BytesReceived),
statPointsParseFail: atomic.LoadInt64(&s.stats.PointsParseFail),
statReadFail: atomic.LoadInt64(&s.stats.ReadFail),
statBatchesTransmitted: atomic.LoadInt64(&s.stats.BatchesTransmitted),
statPointsTransmitted: atomic.LoadInt64(&s.stats.PointsTransmitted),
statBatchesTransmitFail: atomic.LoadInt64(&s.stats.BatchesTransmitFail),
},
}}
}
func (s *Service) writer() {
defer s.wg.Done()
for {
select {
case batch := <-s.batcher.Out():
// Will attempt to create database if not yet created.
if err := s.createInternalStorage(); err != nil {
s.Logger.Info("Required database does not yet exist",
logger.Database(s.config.Database), zap.Error(err))
continue
}
if err := s.PointsWriter.WritePointsPrivileged(s.config.Database, s.config.RetentionPolicy, models.ConsistencyLevelAny, batch); err == nil {
atomic.AddInt64(&s.stats.BatchesTransmitted, 1)
atomic.AddInt64(&s.stats.PointsTransmitted, int64(len(batch)))
} else {
s.Logger.Info("Failed to write point batch to database",
logger.Database(s.config.Database), zap.Error(err))
atomic.AddInt64(&s.stats.BatchesTransmitFail, 1)
}
case <-s.done:
return
}
}
}
func (s *Service) serve() {
defer s.wg.Done()
buf := make([]byte, MaxUDPPayload)
for {
select {
case <-s.done:
// We closed the connection, time to go.
return
default:
// Keep processing.
n, _, err := s.conn.ReadFromUDP(buf)
if err != nil {
atomic.AddInt64(&s.stats.ReadFail, 1)
s.Logger.Info("Failed to read UDP message", zap.Error(err))
continue
}
atomic.AddInt64(&s.stats.BytesReceived, int64(n))
bufCopy := make([]byte, n)
copy(bufCopy, buf[:n])
s.parserChan <- bufCopy
}
}
}
func (s *Service) parser() {
defer s.wg.Done()
for {
select {
case <-s.done:
return
case buf := <-s.parserChan:
points, err := models.ParsePointsWithPrecision(buf, time.Now().UTC(), s.config.Precision)
if err != nil {
atomic.AddInt64(&s.stats.PointsParseFail, 1)
s.Logger.Info("Failed to parse points", zap.Error(err))
continue
}
for _, point := range points {
s.batcher.In() <- point
}
atomic.AddInt64(&s.stats.PointsReceived, int64(len(points)))
}
}
}
// Close closes the service and the underlying listener.
func (s *Service) Close() error {
if wait := func() bool {
s.mu.Lock()
defer s.mu.Unlock()
if s.closed() {
return false // Already closed.
}
close(s.done)
if s.conn != nil {
s.conn.Close()
}
if s.batcher != nil {
s.batcher.Stop()
}
return true
}(); !wait {
return nil
}
s.wg.Wait()
// Release all remaining resources.
s.mu.Lock()
s.done = nil
s.conn = nil
s.batcher = nil
s.mu.Unlock()
s.Logger.Info("Service closed")
return nil
}
// Closed returns true if the service is currently closed.
func (s *Service) Closed() bool {
s.mu.Lock()
defer s.mu.Unlock()
return s.closed()
}
func (s *Service) closed() bool {
select {
case <-s.done:
// Service is closing.
return true
default:
}
return s.done == nil
}
// createInternalStorage ensures that the required database has been created.
func (s *Service) createInternalStorage() error {
s.mu.RLock()
ready := s.ready
s.mu.RUnlock()
if ready {
return nil
}
if _, err := s.MetaClient.CreateDatabase(s.config.Database); err != nil {
return err
}
// The service is now ready.
s.mu.Lock()
s.ready = true
s.mu.Unlock()
return nil
}
// WithLogger sets the logger on the service.
func (s *Service) WithLogger(log *zap.Logger) {
s.Logger = log.With(zap.String("service", "udp"))
}
// Addr returns the listener's address.
func (s *Service) Addr() net.Addr {
return s.addr
}