Skip to content

Commit aaed277

Browse files
robbaveyrobbavey
committed
Improve memory management
This commit makes deserialization of Beats messages lazy for messages sent via the V2 protocol. Instead of deserializing in the parsing block, the ByteBuf is copied to the Batch instance, which will deserialize when the data is needed. Also: Fixes an integration test which was failing regularly on Travis Fixes #299
1 parent 2f68e50 commit aaed277

17 files changed

+503
-231
lines changed

‎lib/logstash/inputs/beats.rb

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -181,7 +181,6 @@ def create_server
181181

182182
server.enableSSL(ssl_builder)
183183
end
184-
185184
server
186185
end
187186

‎lib/logstash/inputs/beats/message_listener.rb

Lines changed: 14 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -27,29 +27,22 @@ def initialize(queue, input)
2727
end
2828

2929
def onNewMessage(ctx, message)
30-
begin
31-
hash = message.getData
32-
ip_address = ip_address(ctx)
30+
hash = message.getData
31+
ip_address = ip_address(ctx)
3332

34-
hash['@metadata']['ip_address'] = ip_address unless ip_address.nil? || hash['@metadata'].nil?
35-
target_field = extract_target_field(hash)
33+
hash['@metadata']['ip_address'] = ip_address unless ip_address.nil? || hash['@metadata'].nil?
34+
target_field = extract_target_field(hash)
3635

37-
if target_field.nil?
38-
event = LogStash::Event.new(hash)
39-
@nocodec_transformer.transform(event)
40-
@queue << event
41-
else
42-
codec(ctx).accept(CodecCallbackListener.new(target_field,
43-
hash,
44-
message.getIdentityStream(),
45-
@codec_transformer,
46-
@queue))
47-
end
48-
rescue => e
49-
logger.warn("Error handling message #{message}", e)
50-
raise e
51-
ensure
52-
message.release
36+
if target_field.nil?
37+
event = LogStash::Event.new(hash)
38+
@nocodec_transformer.transform(event)
39+
@queue << event
40+
else
41+
codec(ctx).accept(CodecCallbackListener.new(target_field,
42+
hash,
43+
message.getIdentityStream(),
44+
@codec_transformer,
45+
@queue))
5346
end
5447
end
5548

‎spec/integration/logstash_forwarder_spec.rb

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,6 @@
5555
f.write(events.join("\n") + "\n")
5656
end
5757
sleep(1) # give some time to the clients to pick up the changes
58-
stop_client
5958
end
6059

6160
after :each do

‎src/main/java/org/logstash/beats/Batch.java

Lines changed: 46 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -1,47 +1,48 @@
11
package org.logstash.beats;
22

3-
import java.util.ArrayList;
4-
import java.util.List;
5-
6-
public class Batch {
7-
private byte protocol = Protocol.VERSION_2;
8-
private int batchSize;
9-
private List<Message> messages = new ArrayList();
10-
11-
public List<Message> getMessages() {
12-
return messages;
13-
}
14-
15-
public void addMessage(Message message) {
16-
message.setBatch(this);
17-
messages.add(message);
18-
}
19-
20-
public int size() {
21-
return messages.size();
22-
}
23-
24-
public void setBatchSize(int size) {
25-
batchSize = size;
26-
}
27-
28-
public int getBatchSize() {
29-
return batchSize;
30-
}
31-
32-
public boolean isEmpty() {
33-
return 0 == messages.size();
34-
}
35-
36-
public boolean complete() {
37-
return size() == getBatchSize();
38-
}
39-
40-
public byte getProtocol() {
41-
return protocol;
42-
}
43-
44-
public void setProtocol(byte protocol) {
45-
this.protocol = protocol;
46-
}
47-
}
3+
/**
4+
* Interface representing a Batch of {@link Message}.
5+
*/
6+
public interface Batch extends Iterable<Message>{
7+
/**
8+
* Returns the protocol of the sent messages that this batch was constructed from
9+
* @return byte - either '1' or '2'
10+
*/
11+
byte getProtocol();
12+
13+
/**
14+
* Number of messages that the batch is expected to contain.
15+
* @return int - number of messages
16+
*/
17+
int getBatchSize();
18+
19+
/**
20+
* Set the number of messages that the batch is expected to contain.
21+
* @param batchSize int - number of messages
22+
*/
23+
void setBatchSize(int batchSize);
24+
25+
/**
26+
* Current number of messages in the batch
27+
* @return int
28+
*/
29+
int size();
30+
31+
/**
32+
* Is the batch currently empty?
33+
* @return boolean
34+
*/
35+
boolean isEmpty();
36+
37+
/**
38+
* Is the batch complete?
39+
* @return boolean
40+
*/
41+
boolean isComplete();
42+
43+
/**
44+
* Release the resources associated with the batch. Consumers of the batch *must* release
45+
* after use.
46+
*/
47+
void release();
48+
}

‎src/main/java/org/logstash/beats/BeatsHandler.java

Lines changed: 19 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@
99
import java.net.InetSocketAddress;
1010
import javax.net.ssl.SSLHandshakeException;
1111

12-
public class BeatsHandler extends SimpleChannelInboundHandler<NewBatch> {
12+
public class BeatsHandler extends SimpleChannelInboundHandler<Batch> {
1313
private final static Logger logger = LogManager.getLogger(BeatsHandler.class);
1414
public static AttributeKey<Boolean> PROCESSING_BATCH = AttributeKey.valueOf("processing-batch");
1515
private final IMessageListener messageListener;
@@ -45,50 +45,28 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
4545

4646
@Override
4747
public void channelRead0(ChannelHandlerContext ctx, Batch batch) throws Exception {
48-
if(logger.isTraceEnabled()) {
49-
logger.trace(format("Received a new payload"));
48+
if(logger.isDebugEnabled()) {
49+
logger.debug(format("Received a new payload"));
5050
}
5151

5252
ctx.channel().attr(BeatsHandler.PROCESSING_BATCH).set(true);
53-
for(Message message : batch.getMessages()) {
54-
if(logger.isTraceEnabled()) {
55-
logger.trace(format("Sending a new message for the listener, sequence: " + message.getSequence()));
56-
}
57-
messageListener.onNewMessage(ctx, message);
5853

59-
if(needAck(message)) {
60-
ack(ctx, message);
54+
try {
55+
for (Message message : batch) {
56+
if (logger.isDebugEnabled()) {
57+
logger.debug(format("Sending a new message for the listener, sequence: " + message.getSequence()));
58+
}
59+
messageListener.onNewMessage(ctx, message);
60+
61+
if (needAck(message)) {
62+
ack(ctx, message);
63+
}
6164
}
65+
}finally{
66+
batch.release();
67+
ctx.flush();
68+
ctx.channel().attr(PROCESSING_BATCH).set(false);
6269
}
63-
ctx.flush();
64-
ctx.channel().attr(PROCESSING_BATCH).set(false);
65-
}
66-
67-
@Override
68-
public void channelRead0(ChannelHandlerContext ctx, NewBatch batch) throws Exception {
69-
if(logger.isDebugEnabled()) {
70-
logger.debug(format("Received a new payload"));
71-
}
72-
73-
ctx.channel().attr(PROCESSING_BATCH).set(true);
74-
batch.getMessageStream().forEach(e -> {
75-
messageListener.onNewMessage(ctx, e);
76-
if (needAck(e)){
77-
ack(ctx, e);
78-
}
79-
});
80-
// for(Message message : batch.getMessages()) {
81-
// if(logger.isDebugEnabled()) {
82-
// logger.debug(format("Sending a new message for the listener, sequence: " + message.getSequence()));
83-
// }
84-
// messageListener.onNewMessage(ctx, message);
85-
//
86-
// if(needAck(message)) {
87-
// ack(ctx, message);
88-
// }
89-
// }
90-
ctx.flush();
91-
ctx.channel().attr(PROCESSING_BATCH).set(false);
9270
}
9371

9472
/*
@@ -120,11 +98,10 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
12098
}
12199

122100
private boolean needAck(Message message) {
123-
return message.getSequence() == message.getNewBatch().getBatchSize();
101+
return message.getSequence() == message.getBatch().getBatchSize();
124102
}
125103

126104
private void ack(ChannelHandlerContext ctx, Message message) {
127-
logger.warn(format("Acking message number " + message.getSequence()));
128105
if (logger.isTraceEnabled()){
129106
logger.trace(format("Acking message number " + message.getSequence()));
130107
}
@@ -155,7 +132,7 @@ private String format(String message) {
155132
remotehost = remote.getAddress().getHostAddress() + ":" + remote.getPort();
156133
} else{
157134
remotehost = "undefined";
158-
};
135+
}
159136

160137
return "[local: " + localhost + ", remote: " + remotehost + "] " + message;
161138
}

‎src/main/java/org/logstash/beats/BeatsParser.java

Lines changed: 14 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,6 @@
11
package org.logstash.beats;
22

33

4-
import com.fasterxml.jackson.databind.ObjectMapper;
5-
import com.fasterxml.jackson.module.afterburner.AfterburnerModule;
64
import io.netty.buffer.ByteBuf;
75
import io.netty.buffer.ByteBufOutputStream;
86
import io.netty.channel.ChannelHandlerContext;
@@ -20,8 +18,6 @@
2018

2119

2220
public class BeatsParser extends ByteToMessageDecoder {
23-
private static final int CHUNK_SIZE = 1024;
24-
public final static ObjectMapper MAPPER = new ObjectMapper().registerModule(new AfterburnerModule());
2521
private final static Logger logger = LogManager.getLogger(BeatsParser.class);
2622

2723
private Batch batch;
@@ -56,19 +52,17 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
5652

5753
switch (currentState) {
5854
case READ_HEADER: {
59-
if (batch == null){
60-
batch = new Batch();
61-
}
6255
logger.trace("Running: READ_HEADER");
6356

6457
byte currentVersion = in.readByte();
65-
66-
if(Protocol.isVersion2(currentVersion)) {
67-
logger.trace("Frame version 2 detected");
68-
batch.setProtocol(Protocol.VERSION_2);
69-
} else {
70-
logger.trace("Frame version 1 detected");
71-
batch.setProtocol(Protocol.VERSION_1);
58+
if (batch == null) {
59+
if (Protocol.isVersion2(currentVersion)) {
60+
batch = new V2Batch();
61+
logger.trace("Frame version 2 detected");
62+
} else {
63+
logger.trace("Frame version 1 detected");
64+
batch = new V1Batch();
65+
}
7266
}
7367
transition(States.READ_FRAME_TYPE);
7468
break;
@@ -145,15 +139,13 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
145139

146140
count++;
147141
}
148-
149142
Message message = new Message(sequence, dataMap);
150-
batch.addMessage(message);
143+
((V1Batch)batch).addMessage(message);
151144

152-
if(batch.complete()) {
145+
if (batch.isComplete()){
153146
out.add(batch);
154147
batchComplete();
155148
}
156-
157149
transition(States.READ_HEADER);
158150

159151
break;
@@ -184,9 +176,9 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
184176
ByteBuf buffer = ctx.alloc().buffer(requiredBytes);
185177
try (
186178
ByteBufOutputStream buffOutput = new ByteBufOutputStream(buffer);
187-
InflaterOutputStream inflater = new InflaterOutputStream(buffOutput, new Inflater());
179+
InflaterOutputStream inflater = new InflaterOutputStream(buffOutput, new Inflater())
188180
) {
189-
ByteBuf bytesRead = in.readBytes(inflater, requiredBytes);
181+
in.readBytes(inflater, requiredBytes);
190182
transition(States.READ_HEADER);
191183
try {
192184
while (buffer.readableBytes() > 0) {
@@ -201,13 +193,11 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
201193
}
202194
case READ_JSON: {
203195
logger.trace("Running: READ_JSON");
204-
205-
batch.addMessage(new Message(sequence, in.readBytes(requiredBytes)));
206-
if(batch.size() == batch.getBatchSize()) {
196+
((V2Batch)batch).addMessage(sequence, in, requiredBytes);
197+
if(batch.isComplete()) {
207198
if(logger.isTraceEnabled()) {
208199
logger.trace("Sending batch size: " + this.batch.size() + ", windowSize: " + batch.getBatchSize() + " , seq: " + sequence);
209200
}
210-
211201
out.add(batch);
212202
batchComplete();
213203
}
@@ -238,7 +228,6 @@ private void batchComplete() {
238228
requiredBytes = 0;
239229
sequence = 0;
240230
batch = null;
241-
logger.warn("batch compete");
242231
}
243232

244233
public class InvalidFrameProtocolException extends Exception {

‎src/main/java/org/logstash/beats/KeepAliveHandler.java

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,14 +14,13 @@ public class KeepAliveHandler extends ChannelDuplexHandler {
1414

1515
@Override
1616
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
17-
IdleStateEvent e = null;
17+
IdleStateEvent e;
1818

1919
if (evt instanceof IdleStateEvent) {
2020
e = (IdleStateEvent) evt;
2121
if (e.state() == IdleState.WRITER_IDLE) {
2222
if (isProcessing(ctx)) {
2323
ChannelFuture f = ctx.writeAndFlush(new Ack(Protocol.VERSION_2, 0));
24-
logger.warn("sending keep alive ack to libbeat");
2524
if (logger.isTraceEnabled()) {
2625
logger.trace("sending keep alive ack to libbeat");
2726
f.addListener((ChannelFutureListener) future -> {
@@ -51,7 +50,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
5150
}
5251
}
5352

54-
public boolean isProcessing(ChannelHandlerContext ctx) {
53+
public boolean isProcessing(ChannelHandlerContext ctx) {
5554
return ctx.channel().hasAttr(BeatsHandler.PROCESSING_BATCH) && ctx.channel().attr(BeatsHandler.PROCESSING_BATCH).get();
5655
}
5756
}

0 commit comments

Comments
 (0)