mirror of
https://github.com/tiennm99/tjahzi.git
synced 2026-10-14 08:18:38 +00:00
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.
This commit is contained in:
1 parent
8a02d34ed4
commit
3172b3cfd2
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
|
||||
|
||||
+2
-2
@@ -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(
|
||||
|
||||
Reference in new issue
Block a user