rework msgpack buffers #2300

richardstartin · 2021-01-14T22:46:25Z

The main change here is that buffering is abstracted away from the format writer(s?) so that all the format writer needs to be aware of is the format. There is a new abstraction called StreamingBuffer with two flavours

FlushingBuffer - fixed size, flushes when full, this is how we have been sending traces for months. Traces are awkward because they are large and the format is length prefixed, which means we need to tell the agent how many traces to expect before sending the traces, which forbids streaming.
GrowableBuffer - resizes its buffer when necessary, should only be used when there is an implicit limit on its growth in practice, e.g. for the serialized string table in the v0.5 trace format.

dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java

richardstartin · 2021-01-14T22:51:27Z

dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/Payload.java

@@ -25,4 +29,24 @@ int traceCount() {
  abstract void writeTo(WritableByteChannel channel) throws IOException;

  abstract RequestBody toRequest();
+


the bleeding of msgpack into here seems a reasonable tradeoff for not needing to store the header in the buffer itself, which makes MsgPackWriter more generic.

Would it make more sense to have these as static methods on a msgpack specific class?

this effectively is a msgpack specific class and will be for the foreseeable future

richardstartin · 2021-01-14T22:54:40Z

dd-trace-core/src/main/java/datadog/trace/core/serialization/msgpack/MsgPackWriter.java

-    return allocationFreeUTF8Encode(s);
-  }
-
-  private int allocationFreeUTF8Encode(CharSequence s) {


UTF8BytesString obviated the need for these sorts of shenanigans, which are actually a lot slower than calling getBytes() because every charAt() and every put(byte) is bounds-checked. The more we use UTF8BytesString, the less we'll allocate in serialisation.

richardstartin · 2021-01-14T22:57:55Z

dd-trace-core/src/main/java/datadog/trace/core/serialization/WritableFormatter.java

@@ -1,150 +1,3 @@
 package datadog.trace.core.serialization;

-import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;


This existed to share flushing logic between the abandoned protobuf writer and the msgpack writer. We won't be trying protobuf again, and the flushing has been moved into the buffer implementation.

richardstartin · 2021-01-14T22:58:56Z

dd-trace-core/src/test/groovy/datadog/trace/common/writer/DDAgentApiTest.groovy

@@ -391,9 +392,8 @@ class DDAgentApiTest extends DDSpecification {
  }

  Payload prepareTraces(String agentVersion, List<List<DDSpan>> traces) {


This sort of thing is in too many places and made this change too difficult. I've patched up all of these, but this needs refactoring.

richardstartin · 2021-01-14T22:59:16Z

dd-trace-core/src/test/groovy/datadog/trace/common/writer/DDAgentWriterCombinedTest.groovy

@@ -200,7 +201,7 @@ class DDAgentWriterCombinedTest extends DDSpecification {
    when:
    def mapper = agentVersion.equals("v0.5/traces") ? new TraceMapperV0_5() : new TraceMapperV0_4()
    int traceSize = calculateSize(minimalTrace, mapper)
-    int maxedPayloadTraceCount = ((int) ((mapper.messageBufferSize() - 5) / traceSize))


5 was the space reserved for a header.

why no longer needed?

because there is no header in the buffer any more, it's added when writing the bytes out to the network

richardstartin · 2021-01-14T23:00:56Z

...ce-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV04PayloadTest.groovy

@@ -204,6 +205,13 @@ class TraceMapperV04PayloadTest extends DDSpecification {

    @Override
    int write(ByteBuffer src) {
+      if (captured.remaining() < src.remaining()) {


letting this grow allows the test to explore more cases, rather than rejecting the output, but the increase in heap usage needs to be monitored.

richardstartin · 2021-01-14T23:02:25Z

...ce-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV05PayloadTest.groovy

@@ -32,49 +33,6 @@ import static org.msgpack.core.MessageFormat.UINT8

 class TraceMapperV05PayloadTest extends DDSpecification {

-
-  def "dictionary overflow causes a flush"() {


The dictionary doesn't flush any more. Keeping two isolated callbacks triggered when fixed capacities were reached in sync made this code extremely hard to reason about.

richardstartin · 2021-01-14T23:03:01Z

...ce-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV05PayloadTest.groovy

@@ -98,7 +56,9 @@ class TraceMapperV05PayloadTest extends DDSpecification {
      UUID.randomUUID().toString(),
      false))
    int traceSize = calculateSize(repeatedTrace)
-    int tracesRequiredToOverflowBody = (traceMapper.messageBufferSize() + traceSize - 1) / traceSize
+    // 30KB body


no need to use MBs of data to check there's a flush when the message buffer is full

richardstartin · 2021-01-14T23:05:00Z

internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/UTF8BytesString.java

@@ -53,6 +53,12 @@ public void transferTo(ByteBuffer buffer) {
    buffer.put(utf8Bytes);
  }

+  /** Writes the UTF8 encoding of the wrapped {@code String}. */
+  public byte[] getUtf8Bytes() {


Ideally a naked reference to this byte[] should be avoided

dd-trace-core/src/main/java/datadog/trace/core/serialization/msgpack/MsgPackWriter.java

… the ripple effects of not having encapsulated this properly

dd-trace-core/src/main/java/datadog/trace/core/serialization/GrowableBuffer.java

tylerbenson · 2021-01-28T15:57:08Z

dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/Payload.java

@@ -25,4 +29,24 @@ int traceCount() {
  abstract void writeTo(WritableByteChannel channel) throws IOException;

  abstract RequestBody toRequest();
+


Would it make more sense to have these as static methods on a msgpack specific class?

tylerbenson · 2021-01-28T16:00:08Z

dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java


  public SerializingMetricWriter(WellKnownTags wellKnownTags, Sink sink) {
    this.wellKnownTags = wellKnownTags;
-    this.writer = new MsgPackWriter(sink, ByteBuffer.allocate(1 << 20), EnumSet.of(SINGLE_MESSAGE));
+    this.buffer = new GrowableBuffer(512 << 10);


In the doc for GrowableBuffer you say only use if bounded... what is it that makes metrics limited in size?

the metrics points are stored in a bounded LRU cache

tylerbenson · 2021-01-28T16:05:24Z

dd-trace-core/src/test/groovy/datadog/trace/common/writer/DDAgentWriterCombinedTest.groovy

@@ -200,7 +201,7 @@ class DDAgentWriterCombinedTest extends DDSpecification {
    when:
    def mapper = agentVersion.equals("v0.5/traces") ? new TraceMapperV0_5() : new TraceMapperV0_4()
    int traceSize = calculateSize(minimalTrace, mapper)
-    int maxedPayloadTraceCount = ((int) ((mapper.messageBufferSize() - 5) / traceSize))


why no longer needed?

richardstartin added the tag: do not merge Do not merge changes label Jan 14, 2021

richardstartin commented Jan 14, 2021

View reviewed changes

dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java Outdated Show resolved Hide resolved

richardstartin commented Jan 14, 2021

View reviewed changes

richardstartin force-pushed the rgs/streaming-buffers branch from 31e5ffa to 80c8826 Compare January 14, 2021 22:56

richardstartin commented Jan 14, 2021

View reviewed changes

richardstartin force-pushed the rgs/streaming-buffers branch 2 times, most recently from 4273061 to 19afa9c Compare January 15, 2021 00:06

bantonsson reviewed Jan 18, 2021

View reviewed changes

dd-trace-core/src/main/java/datadog/trace/core/serialization/msgpack/MsgPackWriter.java Show resolved Hide resolved

richardstartin force-pushed the rgs/streaming-buffers branch from 19afa9c to 192a872 Compare January 20, 2021 14:50

richardstartin mentioned this pull request Jan 21, 2021

metrics tweaks #2318

Merged

richardstartin force-pushed the rgs/streaming-buffers branch from 192a872 to 1a49110 Compare January 21, 2021 12:58

richardstartin removed the tag: do not merge Do not merge changes label Jan 21, 2021

richardstartin force-pushed the rgs/streaming-buffers branch 9 times, most recently from 88c1de3 to 9e9cd97 Compare January 25, 2021 15:50

richardstartin added 2 commits January 27, 2021 10:27

fold WritableFormatter back in to MsgPackWriter

d27f4c0

move flushing behaviour into a StreamingBuffer abstraction, deal with…

1121f48

… the ripple effects of not having encapsulated this properly

richardstartin force-pushed the rgs/streaming-buffers branch from 9e9cd97 to 40701bb Compare January 27, 2021 10:28

richardstartin marked this pull request as ready for review January 27, 2021 10:29

richardstartin requested a review from a team as a code owner January 27, 2021 10:29

richardstartin changed the title ~~streaming buffers~~ rework msgpack buffers Jan 27, 2021

richardstartin force-pushed the rgs/streaming-buffers branch from b7a3471 to 8935fe6 Compare January 27, 2021 10:43

richardstartin added 2 commits January 27, 2021 11:09

enable integration test

20f66ed

default payload to empty msgpack array

2746be4

richardstartin force-pushed the rgs/streaming-buffers branch from 8935fe6 to 22101e0 Compare January 27, 2021 11:09

bantonsson approved these changes Jan 27, 2021

View reviewed changes

dd-trace-core/src/main/java/datadog/trace/core/serialization/GrowableBuffer.java Show resolved Hide resolved

use GrowableBuffer to simplify SerializingMetricWriter

0263eb8

richardstartin force-pushed the rgs/streaming-buffers branch from 22101e0 to 0263eb8 Compare January 27, 2021 19:45

tylerbenson approved these changes Jan 28, 2021

View reviewed changes

richardstartin merged commit 2dbce2f into master Jan 28, 2021

richardstartin deleted the rgs/streaming-buffers branch January 28, 2021 16:55

github-actions bot added this to the 0.73.0 milestone Jan 28, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

rework msgpack buffers #2300

rework msgpack buffers #2300

richardstartin commented Jan 14, 2021 •

edited

richardstartin Jan 14, 2021

bantonsson Jan 18, 2021

tylerbenson Jan 28, 2021

richardstartin Jan 28, 2021

richardstartin Jan 14, 2021

richardstartin Jan 14, 2021

richardstartin Jan 14, 2021

richardstartin Jan 14, 2021

tylerbenson Jan 28, 2021

richardstartin Jan 28, 2021

richardstartin Jan 14, 2021

richardstartin Jan 14, 2021

richardstartin Jan 14, 2021

richardstartin Jan 14, 2021

tylerbenson Jan 28, 2021

tylerbenson Jan 28, 2021

richardstartin Jan 28, 2021

tylerbenson Jan 28, 2021

		@@ -25,4 +29,24 @@ int traceCount() {
		abstract void writeTo(WritableByteChannel channel) throws IOException;

		abstract RequestBody toRequest();

		@@ -1,150 +1,3 @@
		package datadog.trace.core.serialization;

		import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;

		@@ -391,9 +392,8 @@ class DDAgentApiTest extends DDSpecification {
		}

		Payload prepareTraces(String agentVersion, List<List<DDSpan>> traces) {

		@@ -32,49 +33,6 @@ import static org.msgpack.core.MessageFormat.UINT8

		class TraceMapperV05PayloadTest extends DDSpecification {


		def "dictionary overflow causes a flush"() {

rework msgpack buffers #2300

rework msgpack buffers #2300

Conversation

richardstartin commented Jan 14, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

richardstartin commented Jan 14, 2021 •

edited