forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 0
/
sync.go
137 lines (114 loc) · 3.37 KB
/
sync.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
package logstash
import (
"errors"
"expvar"
"time"
"github.com/elastic/beats/libbeat/common"
"github.com/elastic/beats/libbeat/logp"
"github.com/elastic/beats/libbeat/outputs/transport"
)
// Metrics that can retrieved through the expvar web interface.
var (
ackedEvents = expvar.NewInt("libbeatLogstashPublishedAndAckedEvents")
eventsNotAcked = expvar.NewInt("libbeatLogstashPublishedButNotAckedEvents")
publishEventsCallCount = expvar.NewInt("libbeatLogstashPublishEventsCallCount")
)
// client implements the ProtocolClient interface to be used
// with different mode. The client implements slow start with low window sizes +
// window size backoff in case of long running transactions.
//
// it is suggested to use lumberjack in conjunction with roundRobinConnectionMode
// if logstash becomes unresponsive
type client struct {
*transport.Client
*protocol
win window
countTimeoutErr int
}
const (
minWindowSize int = 1
defaultStartMaxWindowSize int = 10
)
// errors
var (
ErrNotConnected = errors.New("lumberjack client is not connected")
)
func newLumberjackClient(
conn *transport.Client,
compressLevel int,
maxWindowSize int,
timeout time.Duration,
) (*client, error) {
p, err := newClientProcol(conn, timeout, compressLevel)
if err != nil {
return nil, err
}
c := &client{Client: conn, protocol: p}
c.win.init(defaultStartMaxWindowSize, maxWindowSize)
return c, nil
}
func (l *client) Connect(timeout time.Duration) error {
logp.Debug("logstash", "connect")
return l.Client.Connect()
}
func (l *client) Close() error {
logp.Debug("logstash", "close connection")
return l.Client.Close()
}
func (l *client) PublishEvent(event common.MapStr) error {
_, err := l.PublishEvents([]common.MapStr{event})
return err
}
// PublishEvents sends all events to logstash. On error a slice with all events
// not published or confirmed to be processed by logstash will be returned.
func (l *client) PublishEvents(
events []common.MapStr,
) ([]common.MapStr, error) {
publishEventsCallCount.Add(1)
totalNumberOfEvents := len(events)
for len(events) > 0 {
n, err := l.publishWindowed(events)
debug("%v events out of %v events sent to logstash. Continue sending",
n, len(events))
events = events[n:]
if err != nil {
l.win.shrinkWindow()
_ = l.Close()
logp.Err("Failed to publish events caused by: %v", err)
eventsNotAcked.Add(int64(len(events)))
ackedEvents.Add(int64(totalNumberOfEvents - len(events)))
return events, err
}
}
ackedEvents.Add(int64(totalNumberOfEvents))
return nil, nil
}
// publishWindowed published events with current maximum window size to logstash
// returning the total number of events sent (due to window size, or acks until
// failure).
func (l *client) publishWindowed(events []common.MapStr) (int, error) {
if len(events) == 0 {
return 0, nil
}
batchSize := len(events)
windowSize := l.win.get()
debug("Try to publish %v events to logstash with window size %v",
batchSize, windowSize)
// prepare message payload
if batchSize > windowSize {
events = events[:windowSize]
}
outEvents, err := l.sendEvents(events)
count := uint32(len(outEvents))
if err != nil {
if err == errAllEventsEncoding {
return len(events), nil
}
return 0, err
}
if seq, err := l.awaitACK(count); err != nil {
return int(seq), err
}
l.win.tryGrowWindow(batchSize)
return len(events), nil
}