Disk caching for java (signalfx#59)

Author: mhennoch

android/gradle.properties

@@ -1,5 +1,5 @@
 SplunkOtelReactNative_kotlinVersion=1.7.0
-SplunkOtelReactNative_minSdkVersion=21
+SplunkOtelReactNative_minSdkVersion=26
 SplunkOtelReactNative_targetSdkVersion=31
 SplunkOtelReactNative_compileSdkVersion=31
 SplunkOtelReactNative_ndkversion=21.4.7075529

android/src/main/AndroidManifest.xml

@@ -1,5 +1,7 @@
 <manifest xmlns:android="http://schemas.android.com/apk/res/android"
           package="com.splunkotelreactnative">
+  <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
+
   <application>
     <provider
       android:name=".SplunkPerfProvider"

android/src/main/java/com/splunkotelreactnative/ConfigMapReader.java

@@ -20,6 +20,8 @@
 import com.facebook.react.bridge.ReadableMap;
 
 public class ConfigMapReader extends MapReader {
+  private static final boolean DEFAULT_ENABLE_DISK_BUFFERING = false;
+  private static final int DEFAULT_MAX_STORAGE_USE_MB = 25;
   private final ReadableMap map;
 
   public ConfigMapReader(ReadableMap map) {
@@ -34,13 +36,26 @@ public String getRumAccessToken() {
     return Keys.RUM_ACCESS_TOKEN.get(map);
   }
 
+  public boolean getEnableDiskBuffering() {
+    Boolean value = Keys.ENABLE_DISK_BUFFERING.get(map);
+    return value != null ? value : DEFAULT_ENABLE_DISK_BUFFERING;
+  }
+
+  public int getMaxStorageUseMb() {
+    Long value = Keys.MAX_STORAGE_USE_MB.getLong(map);
+    return value != null ? value.intValue() : DEFAULT_MAX_STORAGE_USE_MB;
+  }
+
   public ReadableMap getGlobalAttributes() {
     return ConfigMapReader.Keys.GLOBAL_ATTRIBUTES.getMap(map);
   }
 
   private interface Keys {
     StringKey BEACON_ENDPOINT = new StringKey("beaconEndpoint");
     StringKey RUM_ACCESS_TOKEN = new StringKey("rumAccessToken");
+    BooleanKey ENABLE_DISK_BUFFERING = new BooleanKey("enableDiskBuffering");
+
+    NumberKey MAX_STORAGE_USE_MB = new NumberKey("limitDiskUsageMegabytes");
     MapKey GLOBAL_ATTRIBUTES = new MapKey("globalAttributes");
   }
 }

android/src/main/java/com/splunkotelreactnative/CustomZipkinEncoder.java

@@ -36,7 +36,7 @@
  * <p>SplunkSpanDataModifier#SPLUNK_OPERATION_KEY}) with the span name properly cased, then
  * correcting the span name here at encoding time.
  */
-class CustomZipkinEncoder implements BytesEncoder<Span> {
+public class CustomZipkinEncoder implements BytesEncoder<Span> {
     static final AttributeKey<String> SPLUNK_OPERATION_KEY = stringKey("_splunk_operation");
     private final WriteBuffer.Writer<Span> writer = new V2SpanWriter();
 

android/src/main/java/com/splunkotelreactnative/LogConstants.java

@@ -0,0 +1,5 @@
+package com.splunkotelreactnative;
+
+public class LogConstants {
+  public static final String LOG_TAG = "SplunkRNRum";
+}

android/src/main/java/com/splunkotelreactnative/MapReader.java

@@ -37,6 +37,22 @@ public String get(ReadableMap map) {
     }
   }
 
+  protected static class BooleanKey {
+    private final String key;
+
+    public BooleanKey(String key) {
+      this.key = key;
+    }
+
+    public Boolean get(ReadableMap map) {
+      if (map.hasKey(key) && map.getType(key) == ReadableType.Boolean) {
+        return map.getBoolean(key);
+      } else {
+        return null;
+      }
+    }
+  }
+
   protected static class NumberKey {
     private final String key;
 

android/src/main/java/com/splunkotelreactnative/SplunkOtelReactNativeModule.java

@@ -17,6 +17,7 @@
 
 package com.splunkotelreactnative;
 
+import android.content.ContextWrapper;
 import android.util.Log;
 
 import androidx.annotation.NonNull;
@@ -27,14 +28,18 @@
 import com.facebook.react.bridge.ReactApplicationContext;
 import com.facebook.react.bridge.ReactContextBaseJavaModule;
 import com.facebook.react.bridge.ReactMethod;
+import com.facebook.react.bridge.ReadableArray;
 import com.facebook.react.bridge.ReadableMap;
 import com.facebook.react.bridge.WritableMap;
 import com.facebook.react.module.annotations.ReactModule;
 import com.splunkotelreactnative.crash.CrashEventAttributeExtractor;
 import com.splunkotelreactnative.crash.CrashReporter;
+import com.splunkotelreactnative.exporter.disk.DiskBufferingExporterFactory;
 
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 
 import io.opentelemetry.api.common.Attributes;
@@ -44,6 +49,7 @@
 import io.opentelemetry.api.trace.TraceFlags;
 import io.opentelemetry.api.trace.TraceState;
 import io.opentelemetry.exporter.zipkin.ZipkinSpanExporterBuilder;
+import io.opentelemetry.sdk.trace.data.SpanData;
 import io.opentelemetry.sdk.trace.data.StatusData;
 import io.opentelemetry.sdk.trace.export.SpanExporter;
 
@@ -79,10 +85,8 @@ public void initialize(ReadableMap configMap, Promise promise) {
 
     String endpointWithAuthentication = beaconEndpoint + "?auth=" + accessToken;
 
-    exporter = new CrashEventAttributeExtractor(new ZipkinSpanExporterBuilder()
-      .setEndpoint(endpointWithAuthentication)
-      .setEncoder(new CustomZipkinEncoder())
-      .build());
+    exporter = createExporter(endpointWithAuthentication, getReactApplicationContext(),
+      mapReader.getEnableDiskBuffering(), mapReader.getMaxStorageUseMb());
 
     crashReporter = new CrashReporter(exporter,
       attributesFromMap(mapReader.getGlobalAttributes()), getReactApplicationContext());
@@ -110,39 +114,47 @@ public void nativeCrash() {
   }
 
   @ReactMethod
-  public void export(ReadableMap spanMap, Promise promise) {
+  public void export(ReadableArray spanMaps, Promise promise) {
     SpanExporter currentExporter = exporter;
-    SpanMapReader mapReader = new SpanMapReader(spanMap);
 
     if (currentExporter == null) {
       reportFailure(promise, "Export: exporter not initialized");
       return;
     }
 
-    SpanContext context = contextFromMap(mapReader);
+    List<SpanData> spanDataList = new ArrayList<>();
 
-    if (!context.isValid()) {
-      reportFailure(promise, "Export: trace or span ID not provided");
-      return;
-    }
+    for (int i = 0; i < spanMaps.size(); i++) {
+      ReadableMap spanMap = spanMaps.getMap(i);
+      SpanMapReader mapReader = new SpanMapReader(spanMap);
 
-    SpanContext parentContext = parentContextFromMap(mapReader, context);
-    ReactSpanProperties spanProperties = propertiesFromMap(mapReader);
+      SpanContext context = contextFromMap(mapReader);
+      if (!context.isValid()) {
+        reportFailure(promise, "Export: trace or span ID not provided");
+        return;
+      }
 
-    if (spanProperties == null) {
-      reportFailure(promise, "Export: missing name, start or end time");
-      return;
-    }
+      SpanContext parentContext = parentContextFromMap(mapReader, context);
+      ReactSpanProperties spanProperties = propertiesFromMap(mapReader);
+
+      if (spanProperties == null) {
+        reportFailure(promise, "Export: missing name, start or end time");
+        return;
+      }
 
-    Attributes attributes = attributesFromMap(mapReader.getAttributes());
+      Attributes attributes = attributesFromMap(mapReader.getAttributes());
+      ReactSpanData spanData = new ReactSpanData(spanProperties, attributes, context, parentContext,
+        Collections.emptyList());
 
-    ReactSpanData spanData = new ReactSpanData(spanProperties, attributes, context, parentContext,
-      Collections.emptyList());
-    currentExporter.export(Collections.singleton(spanData));
+      spanDataList.add(spanData);
+    }
+
+    currentExporter.export(spanDataList);
 
     promise.resolve(null);
   }
 
+
   @ReactMethod
   public void setSessionId(String sessionId) {
     CrashReporter currentCrashReporter = crashReporter;
@@ -177,6 +189,18 @@ private SpanContext contextFromMap(SpanMapReader mapReader) {
     return SpanContext.create(traceId, spanId, traceFlags, TraceState.getDefault());
   }
 
+  @NonNull
+  private SpanExporter createExporter(String endpoint, ContextWrapper application,
+                                      boolean enableDiskBuffering, int maxStorageUseMb) {
+    if (!enableDiskBuffering) {
+      return new CrashEventAttributeExtractor(new ZipkinSpanExporterBuilder()
+        .setEndpoint(endpoint)
+        .setEncoder(new CustomZipkinEncoder())
+        .build());
+    }
+    return DiskBufferingExporterFactory.setupDiskBuffering(endpoint, application, maxStorageUseMb);
+  }
+
   @NonNull
   private SpanContext parentContextFromMap(SpanMapReader mapReader, SpanContext childContext) {
     String parentSpanId = mapReader.getParentSpanId();

android/src/main/java/com/splunkotelreactnative/exporter/disk/BandwidthTracker.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright Splunk Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.splunkotelreactnative.exporter.disk;
+
+import java.time.Clock;
+import java.util.ArrayDeque;
+import java.util.List;
+
+/**
+ * Utility class to track how much bandwidth is being used by span data. It tracks raw, uncompressed
+ * spans bytes being input to a sender and is not intended to represent actual
+ * gzipped/compressed/tls bytes on the network.
+ */
+class BandwidthTracker {
+  private static final int DATAPOINTS_TO_TRACK = 6;
+  private final Clock clock;
+  private final ArrayDeque<Long> times = new ArrayDeque<>();
+  private final ArrayDeque<Long> sizes = new ArrayDeque<>();
+
+  BandwidthTracker() {
+    this(Clock.systemUTC());
+  }
+
+  // Exists for testing
+  BandwidthTracker(Clock clock) {
+    this.clock = clock;
+  }
+
+  /** Call this method with encoded zipkin span data to have it tracked. */
+  void tick(List<byte[]> zipkinSpanData) {
+    if (times.size() > DATAPOINTS_TO_TRACK) {
+      times.removeFirst();
+    }
+    times.add(clock.millis());
+
+    if (sizes.size() > DATAPOINTS_TO_TRACK) {
+      sizes.removeFirst();
+    }
+    long currentSize =
+      zipkinSpanData.stream()
+        .map(bytes -> bytes.length)
+        .reduce(0, Integer::sum, Integer::sum);
+    sizes.add(currentSize);
+  }
+
+  /**
+   * Calculates the current average sustained throughput.
+   *
+   * @return - The currently tracked bandwidth, in bytes per second.
+   */
+  double totalSustainedRate() {
+    if (sizes.size() < 2) return 0;
+
+    // Don't count the first ingest payload
+    double total = sizes.stream().skip(1).reduce(0L, Long::sum, Long::sum);
+
+    double timeDeltaInSeconds = (times.getLast() - times.getFirst()) / 1000.0;
+    return total / timeDeltaInSeconds;
+  }
+}