forked from monicasarbu/topbeat
-
Notifications
You must be signed in to change notification settings - Fork 0
/
output.go
118 lines (100 loc) · 2.42 KB
/
output.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
package publisher
import (
"time"
"github.com/elastic/libbeat/common"
"github.com/elastic/libbeat/logp"
"github.com/elastic/libbeat/outputs"
)
type outputWorker struct {
messageWorker
out outputs.BulkOutputer
config outputs.MothershipConfig
maxBulkSize int
}
func newOutputWorker(
config outputs.MothershipConfig,
out outputs.Outputer,
ws *workerSignal,
hwm int,
) *outputWorker {
maxBulkSize := defaultBulkSize
if config.BulkMaxSize != nil {
maxBulkSize = *config.BulkMaxSize
}
o := &outputWorker{
out: outputs.CastBulkOutputer(out),
config: config,
maxBulkSize: maxBulkSize,
}
o.messageWorker.init(ws, hwm, o)
return o
}
func (o *outputWorker) onStop() {}
func (o *outputWorker) onMessage(m message) {
if m.event != nil {
o.onEvent(&m.context, m.event)
} else {
o.onBulk(&m.context, m.events)
}
}
func (o *outputWorker) onEvent(ctx *context, event common.MapStr) {
debug("output worker: publish single event")
ts := time.Time(event["@timestamp"].(common.Time)).UTC()
if !ctx.sync {
_ = o.out.PublishEvent(ctx.signal, ts, event)
return
}
signal := outputs.NewSyncSignal()
for {
o.out.PublishEvent(signal, ts, event)
if signal.Wait() {
outputs.SignalCompleted(ctx.signal)
break
}
}
}
func (o *outputWorker) onBulk(ctx *context, events []common.MapStr) {
if len(events) == 0 {
debug("output worker: no events to publish")
outputs.SignalCompleted(ctx.signal)
return
}
var sync *outputs.SyncSignal
if ctx.sync {
sync = outputs.NewSyncSignal()
}
if o.maxBulkSize < 0 || len(events) <= o.maxBulkSize {
o.sendBulk(sync, ctx, events)
return
}
// start splitting bulk request
splits := (len(events) + (o.maxBulkSize - 1)) / o.maxBulkSize
ctx.signal = outputs.NewSplitSignaler(ctx.signal, splits)
for len(events) > 0 {
sz := o.maxBulkSize
if sz > len(events) {
sz = len(events)
}
o.sendBulk(sync, ctx, events[:sz])
events = events[sz:]
}
}
func (o *outputWorker) sendBulk(
sync *outputs.SyncSignal,
ctx *context,
events []common.MapStr,
) {
debug("output worker: publish %v events", len(events))
ts := time.Time(events[0]["@timestamp"].(common.Time)).UTC()
if sync == nil {
err := o.out.BulkPublish(ctx.signal, ts, events)
if err != nil {
logp.Info("Error bulk publishing events: %s", err)
}
return
}
for done := false; !done; done = sync.Wait() {
o.out.BulkPublish(sync, ts, events)
}
outputs.SignalCompleted(ctx.signal)
}