Add Batching Feature / Configuration to the New Relic Telemetry SDK #278

yamnihcg · 2021-08-06T18:52:14Z

This PR adds a configuration to enable batching in the Telemetry SDK.

What is batching?

The Java Telemetry SDK sends Metric and Event data to New Relic by using the New Relic Ingest API. The Ingest API has limits on the amount of data that can be sent to New Relic in a single request. Specifically, the request payload has to be <= 1 MB. We implemented batching as a way to send reasonably large amounts of data (> 1 MB) without any user intervention. Batching takes request data and splits it into smaller payloads. Each of these payloads is <= 1 MB. So, each payload can then be sent to New Relic.

Enable batching in the Telemetry SDK

Batching can be turned on through constructors in the EventBuffer and the MetricBuffer. By default, batching is not enabled in the Telemetry SDK. Below are two snippets of code where batching is turned off / on. These snippets were taken from EventExample.java.

Batching: Off (by default)

EventBatchSenderFactory factory = EventBatchSenderFactory.fromHttpImplementation(OkHttpPoster::new);
EventBatchSender sender = EventBatchSender.create(factory.configureWith(licenseKey).useLicenseKey(true).build());
EventBuffer eventBuffer = new EventBuffer(getCommonAttributes());
.
.
sender.sendBatch(eventBuffer.createBatch());

Batching: On

EventBatchSenderFactory factory = EventBatchSenderFactory.fromHttpImplementation(OkHttpPoster::new);
EventBatchSender sender = EventBatchSender.create(factory.configureWith(licenseKey).useLicenseKey(true).build());
EventBuffer eventBuffer = new EventBuffer(getCommonAttributes(), true);
.
.
sender.sendBatch(eventBuffer.createBatch());

Where does the batching happen in the EventBuffer / MetricBuffer?

First, the createBatch() function is called on the buffer (refer to EventExample and CountExample). This function looks at the splitBatch variable to check if the user has turned batching on / off. If the user has turned batching off, a single batch is created by calling createSingleBatch(). An ArrayList (of size 1) contains this single batch. If the user has turned batching on, multiple batches are created by calling createBatches(). An interactive explanation of the batching algorithm is here.

The pseudocode for createBatches() is below.


Initialize currentUncompressedSizeCount = 0 // in bytes
Initialize maxUncompressedBatchSize = 18000000 // in bytes 

Initialize listOfBatches  
Initialize currentBatchEvents 

while (queueOfEvents is not empty):

    Pop event from queue 

    If currentUncompressedSizeCount > maxUncompressedBatchSize:

        listOfBatches.add(currentBatchEvents) 
        currentBatchEvents = [] // Reset Batch to Being Empty 
        currentUncompressedSizeCount = 0 // Reset UncompressedSize of Current Batch

    Else:

        Get Uncompressed Size of Event 
        Add it to currentUncompressedSizeCount 

Create Whatever is Remaining in currentBatchEvents, add it to listOfBatches 
Send listOfBatches

How is data sent to New Relic?

Data is sent to New Relic through the .sendBatch() method in the EventBatchSender / MetricBatchSender.

XiXiaPdx

@yamnihcg Lets remove this since it is not being used

XiXiaPdx · 2021-08-06T19:25:06Z

telemetry-core/src/main/java/com/newrelic/telemetry/events/Event.java


 /**
 * An Event is ad hoc set of key-value pairs with an associated timestamp, recorded in the New Relic
 * Metric API.
 */
 public final class Event implements Telemetry {
+  private static final Logger logger = LoggerFactory.getLogger(Event.class);


lets remove this since it is not being used

XiXiaPdx · 2021-08-06T19:38:49Z

telemetry-core/src/main/java/com/newrelic/telemetry/events/EventBatchSender.java

+      response = sendBatch(batch);
+
+      // If there is an issue with a batch, stop sending batches
+      if ((response.getStatusCode() != 200)) {


Checking the respond code line won't be reached if there is an exception. The underlying exception from sendPayload will bubble up to the sendBatch(one batch) method here and then this whole method of sending batches will exit.

I think what we might need to do is create a pool of threads and have each batch be sent async on its own thread.

Or, we can try catch the exception and log that something went wrong. The rest of the batches in this list will still get sent (at least attempted to). I would prefer the async approach

XiXiaPdx · 2021-08-06T19:42:10Z

telemetry-core/src/main/java/com/newrelic/telemetry/events/EventBuffer.java

+      batches.add(singleEventBatch);
+      return batches;
+    } else {
+      batches = createBatches();


return createBatches(); 😄

XiXiaPdx · 2021-08-06T19:43:54Z

telemetry-core/src/main/java/com/newrelic/telemetry/events/EventBuffer.java

+    logger.debug("Creating Event batch.");
+
+    int currentUncompressedBatchSize = 0;
+


nit: line can be removed

XiXiaPdx · 2021-08-06T19:47:51Z

telemetry-core/src/main/java/com/newrelic/telemetry/metrics/MetricBatchSender.java

+    for (MetricBatch batch : metricBatches) {
+      response = sendBatch(batch);
+
+      // If there is an issue with a batch, stop sending batches


same as above, this line won't be reached when a response exception is thrown

XiXiaPdx · 2021-08-06T19:48:50Z

telemetry-core/src/main/java/com/newrelic/telemetry/metrics/MetricBuffer.java

+    logger.debug("Creating metric batch.");
+    ArrayList<MetricBatch> metricBatches = new ArrayList<>();
+    Collection<Metric> metricsInBatch = new ArrayList<>();
+


nit: extra line

XiXiaPdx · 2021-08-06T19:49:24Z

telemetry-core/src/main/java/com/newrelic/telemetry/metrics/MetricBuffer.java

+    int currentUncompressedBatchSize = 0;
+
+    // Construct JSON for common attributes and add to uncompressed batch size
+


You can place the comment directly above the line, no space needed

XiXiaPdx · 2021-08-06T20:01:08Z

telemetry-core/src/main/java/com/newrelic/telemetry/metrics/MetricBuffer.java

+    // JSON generation + calculating payload size
+
+    Metric curMetric;
+    while ((curMetric = this.metrics.poll()) != null) {


Ok, it looks like this flows this way and I think there is a problem:

get curMetric (assume if this metric gets added to the batch, it will exceed size)
-- uncompressed size is not over, skip to next if block
-- curMetric is a Count, calculate size and add it to the current uncompressized Size. The current uncompressed Size now exceeds max...but it will be too late by the time this gets checked again?
-- curMetric is added to Batch (The batch now exceeds the size)
-- Uncompressed is greater than Max...create Batch(which is too big) and add to list.

Did I follow this incorrectly?

XiXiaPdx · 2021-08-06T20:02:47Z

telemetry-core/src/main/java/com/newrelic/telemetry/transport/BatchDataSender.java

@@ -132,7 +132,8 @@ public Response send(String json, TelemetryBatch<? extends Telemetry> batch)
  private byte[] compressJson(String result) throws IOException {
    ByteArrayOutputStream compressedOutput = new ByteArrayOutputStream();
    GZIPOutputStream gzipOutputStream = new GZIPOutputStream(compressedOutput);
-    gzipOutputStream.write(result.getBytes(StandardCharsets.UTF_8));
+    gzipOutputStream.write(
+        result.getBytes(StandardCharsets.UTF_8)); // Used to get number of bytes in piece of data


remove this comment?

XiXiaPdx

This is looking good...please take a look!

yamnihcg added 8 commits August 3, 2021 13:44

All files changed so batching (split on size) feature can be added.

1b8df91

Old createBatch() method in TelemetryClientExamplee

ba272f6

Updated file with GoogleJavaFormatting

a36b17e

Updated TelemetryClientExample to ignore new MetricBatch change

07c0af8

Updated TelemetryClientExample and Javadoc in Event/Metric buffers.

880cfaa

Updated Javadoc in Event/Metric buffers.

d875cd5

Merge branch 'main' into batching-feature-branch

1995d46

Simplified .createBatches() method in the MetricBuffer

9c125e2

XiXiaPdx reviewed Aug 6, 2021

View reviewed changes

XiXiaPdx suggested changes Aug 6, 2021

View reviewed changes

kford-newrelic added the Q3 SC label Aug 31, 2021

kford-newrelic removed the Q3 SC label Sep 10, 2021

kford-newrelic marked this pull request as draft October 3, 2022 15:14

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add Batching Feature / Configuration to the New Relic Telemetry SDK #278

Add Batching Feature / Configuration to the New Relic Telemetry SDK #278

yamnihcg commented Aug 6, 2021 •

edited

Loading

XiXiaPdx left a comment

XiXiaPdx Aug 6, 2021

XiXiaPdx Aug 6, 2021

XiXiaPdx Aug 6, 2021

XiXiaPdx Aug 6, 2021

XiXiaPdx Aug 6, 2021

XiXiaPdx Aug 6, 2021

XiXiaPdx Aug 6, 2021

XiXiaPdx Aug 6, 2021 •

edited

Loading

XiXiaPdx Aug 6, 2021

XiXiaPdx left a comment

		logger.debug("Creating Event batch.");

		int currentUncompressedBatchSize = 0;

		int currentUncompressedBatchSize = 0;

		// Construct JSON for common attributes and add to uncompressed batch size

Add Batching Feature / Configuration to the New Relic Telemetry SDK #278

Are you sure you want to change the base?

Add Batching Feature / Configuration to the New Relic Telemetry SDK #278

Conversation

yamnihcg commented Aug 6, 2021 • edited Loading

XiXiaPdx left a 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

XiXiaPdx Aug 6, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

XiXiaPdx left a comment

Choose a reason for hiding this comment

yamnihcg commented Aug 6, 2021 •

edited

Loading

XiXiaPdx Aug 6, 2021 •

edited

Loading