From 3172b3cfd2dc26e6e181fcb2fc1b7718d233e71d Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Sun, 24 Jan 2021 13:13:07 +0100 Subject: [PATCH] Re #23: Static labels are serialized into string at startup. If there are no other labels to add to the stream then we will just use that string. --- .../pl/tkowalcz/tjahzi/LogBufferAgent.java | 9 ++--- .../tjahzi/LogBufferDeserializer.java | 35 ++++++++++++++----- .../tkowalcz/tjahzi/LogBufferAgentTest.java | 2 +- .../tjahzi/LogBufferDeserializerTest.java | 11 +++--- .../LogBufferSerializerDeserializerTest.java | 4 +-- 5 files changed, 37 insertions(+), 24 deletions(-) diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferAgent.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferAgent.java index 3ebea8d..2765c55 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferAgent.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferAgent.java @@ -25,11 +25,9 @@ public class LogBufferAgent implements Agent, MessageHandler { private final long batchSize; private final long batchWaitMillis; - private final Map 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); diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferDeserializer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferDeserializer.java index 6ab35b8..a444334 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferDeserializer.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferDeserializer.java @@ -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 staticLabels - ) { + private final Map staticLabels; + private final String staticLabelsString; + + public LogBufferDeserializer(Map 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 staticLabels, long timestamp, Map 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 staticLabels, - Map labels) { + String staticLabelsString, + Map labels + ) { + if (labels.isEmpty()) { + return staticLabelsString; + } + TextBuilder textBuilder = TextBuilders.threadLocal(); textBuilder.append("{ "); diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java index 6fd6ef3..bcdf971 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java @@ -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]) ); diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferDeserializerTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferDeserializerTest.java index f7c0e92..0aa6235 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferDeserializerTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferDeserializerTest.java @@ -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 diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferSerializerDeserializerTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferSerializerDeserializerTest.java index 79a64d4..15eecec 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferSerializerDeserializerTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferSerializerDeserializerTest.java @@ -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(