Merge pull request #25 from tkowalcz/23-preserialize-static-lables

Re #23: Static labels are serialized into string at startup. If there…
This commit is contained in:
Tomasz Kowalczewski authored and GitHub committed 2021-01-24 13:31:45 +01:00
commit d2ebf5f531
5 files changed
+37 -24

No files matched your search

@@ -25,11 +25,9 @@ public class LogBufferAgent implements Agent, MessageHandler {
private final long batchSize;
private final long batchWaitMillis;
private final Map<String, String> staticLabels;
private final LogBufferDeserializer logBufferDeserializer = new LogBufferDeserializer();
private final LogBufferDeserializer logBufferDeserializer;
private Logproto.PushRequest.Builder request = Logproto.PushRequest.newBuilder();
private int estimatedBytesPending;
private long timeoutDeadline;
@@ -48,7 +46,7 @@ public class LogBufferAgent implements Agent, MessageHandler {
this.batchWaitMillis = batchWaitMillis;
this.httpClient = httpClient;
this.staticLabels = staticLabels;
this.logBufferDeserializer = new LogBufferDeserializer(staticLabels);
}
@Override
@@ -103,8 +101,7 @@ public class LogBufferAgent implements Agent, MessageHandler {
private void processMessage(MutableDirectBuffer buffer, int index) {
Logproto.StreamAdapter stream = logBufferDeserializer.deserializeIntoProtobuf(
buffer,
index,
staticLabels
index
);
request.addStreams(stream);
@@ -1,6 +1,7 @@
package pl.tkowalcz.tjahzi;
import com.google.protobuf.Timestamp;
import io.netty.util.internal.StringUtil;
import javolution.text.TextBuilder;
import logproto.Logproto;
import org.agrona.DirectBuffer;
@@ -11,11 +12,19 @@ import java.util.Map;
public class LogBufferDeserializer {
public Logproto.StreamAdapter deserializeIntoProtobuf(
DirectBuffer buffer,
int index,
Map<String, String> staticLabels
) {
private final Map<String, String> staticLabels;
private final String staticLabelsString;
public LogBufferDeserializer(Map<String, String> staticLabels) {
this.staticLabels = staticLabels;
this.staticLabelsString = buildLabelsStringIncludingStatic(
Map.of(),
StringUtil.EMPTY_STRING,
staticLabels
);
}
public Logproto.StreamAdapter deserializeIntoProtobuf(DirectBuffer buffer, int index) {
long timestamp = buffer.getLong(index);
index += Long.BYTES;
@@ -29,7 +38,6 @@ public class LogBufferDeserializer {
String line = buffer.getStringAscii(index);
return createGrpcStream(
staticLabels,
timestamp,
labels,
line
@@ -58,7 +66,6 @@ public class LogBufferDeserializer {
}
private Logproto.StreamAdapter createGrpcStream(
Map<String, String> staticLabels,
long timestamp,
Map<String, String> labels,
String line
@@ -68,7 +75,11 @@ public class LogBufferDeserializer {
return Logproto.StreamAdapter.newBuilder()
.setLabels(
buildLabelsStringIncludingStatic(staticLabels, labels)
buildLabelsStringIncludingStatic(
staticLabels,
staticLabelsString,
labels
)
)
.addEntries(
Logproto.EntryAdapter.newBuilder()
@@ -84,7 +95,13 @@ public class LogBufferDeserializer {
private static String buildLabelsStringIncludingStatic(
Map<String, String> staticLabels,
Map<String, String> labels) {
String staticLabelsString,
Map<String, String> labels
) {
if (labels.isEmpty()) {
return staticLabelsString;
}
TextBuilder textBuilder = TextBuilders.threadLocal();
textBuilder.append("{ ");
@@ -23,7 +23,7 @@ class LogBufferAgentTest {
private SettableClock clock;
@BeforeEach
private void setUp() {
void setUp() {
logBuffer = new ManyToOneRingBuffer(
new UnsafeBuffer(new byte[1024 * 1024 + RingBufferDescriptor.TRAILER_LENGTH])
);
@@ -19,6 +19,7 @@ import static pl.tkowalcz.tjahzi.LogBufferSerializerDeserializerTest.toPropertyS
class LogBufferDeserializerTest {
@SuppressWarnings("unused")
@ParameterizedTest(name = "{0}")
@MethodSource("pl.tkowalcz.tjahzi.LogBufferSerializerDeserializerTest#variousMessageConfigurations")
void shouldDeserializeMessageAndAddStaticLabels(
@@ -45,11 +46,10 @@ class LogBufferDeserializerTest {
staticLabels.put("bazz", "buzz");
// When
LogBufferDeserializer deserializer = new LogBufferDeserializer();
LogBufferDeserializer deserializer = new LogBufferDeserializer(staticLabels);
Logproto.StreamAdapter stream = deserializer.deserializeIntoProtobuf(
buffer,
0,
staticLabels
0
);
// Then
@@ -100,11 +100,10 @@ class LogBufferDeserializerTest {
);
// When
LogBufferDeserializer deserializer = new LogBufferDeserializer();
LogBufferDeserializer deserializer = new LogBufferDeserializer(staticLabels);
Logproto.StreamAdapter stream = deserializer.deserializeIntoProtobuf(
buffer,
0,
staticLabels
0
);
// Then
@@ -89,8 +89,8 @@ class LogBufferSerializerDeserializerTest {
);
// Then
LogBufferDeserializer deserializer = new LogBufferDeserializer();
Logproto.StreamAdapter stream = deserializer.deserializeIntoProtobuf(buffer, 0, Map.of());
LogBufferDeserializer deserializer = new LogBufferDeserializer(Map.of());
Logproto.StreamAdapter stream = deserializer.deserializeIntoProtobuf(buffer, 0);
assertThat(stream.getEntriesList()).hasSize(1);
assertThat(stream.getEntriesList().get(0).getTimestamp()).isEqualTo(