Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add ability to throttle exports when reading from disk. #663

Open
wants to merge 8 commits into
base: main
Choose a base branch
from
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -4,3 +4,4 @@
demo-app/local.properties
.DS_Store
**/build/

Original file line number Diff line number Diff line change
@@ -0,0 +1,112 @@
/*
* Copyright The OpenTelemetry Authors
* SPDX-License-Identifier: Apache-2.0
*/

package com.opentelemetry.android;

import android.util.Log;
import io.opentelemetry.sdk.common.CompletableResultCode;
import io.opentelemetry.sdk.trace.data.SpanData;
import io.opentelemetry.sdk.trace.export.SpanExporter;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.function.Function;

class BandwidthThrottlingExporter implements SpanExporter {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we would also need to see a unit test for this class.

private final SpanExporter delegate;
private final Function<SpanData, String> categoryFunction;
private final long maxBytesPerSecond;
private final long timeWindowInMillis;
private long lastExportTime;
private long bytesExportedInWindow;

private BandwidthThrottlingExporter(Builder builder) {
this.delegate = builder.delegate;
this.categoryFunction = builder.categoryFunction;
this.maxBytesPerSecond = builder.maxBytesPerSecond;
this.timeWindowInMillis = builder.timeWindow.toMillis();
this.lastExportTime = SystemTime.get().getCurrentTimeMillis();
this.bytesExportedInWindow = 0;
}

static Builder newBuilder(SpanExporter delegate) {
return new Builder(delegate);
}

@Override
public CompletableResultCode export(Collection<SpanData> spans) {
List<SpanData> spansToExport = new ArrayList<>();
long totalBytes = 0;

for (SpanData span : spans) {
// Estimate the size of the span (this can be adjusted based on actual size)
long spanSize = estimateSpanSize(span);
totalBytes += spanSize;

// Check if we can export this span based on the current bandwidth limit
if (canExport(spanSize)) {
spansToExport.add(span);
bytesExportedInWindow += spanSize;
} else {
Log.d("BandwidthThrottlingExporter", "Throttled span: " + span.getName());
}
}

return delegate.export(spansToExport);
}

private boolean canExport(long spanSize) {
long currentTime = SystemTime.get().getCurrentTimeMillis();
if (currentTime - lastExportTime > timeWindowInMillis) {
// Reset the window
bytesExportedInWindow = 0;
lastExportTime = currentTime;
}

return (bytesExportedInWindow + spanSize)
<= maxBytesPerSecond * (timeWindowInMillis / 1000);
}

private long estimateSpanSize(SpanData span) {
// This is a placeholder for actual size estimation logic
return span.getAttributes().size() * 8; // Example: 8 bytes per attribute
}

@Override
public CompletableResultCode flush() {
return delegate.flush();
}

@Override
public CompletableResultCode shutdown() {
return delegate.shutdown();
}

static class Builder {
final SpanExporter delegate;
private CategoryFunction categoryFunction = span -> "default";
private long maxBytesPerSecond = 1024; // Default to 1 KB/s
private long timeWindowInMillis = 1000; // Default to 1 second

private Builder(SpanExporter delegate) {
this.delegate = delegate;
}

Builder maxBytesPerSecond(long maxBytesPerSecond) {
this.maxBytesPerSecond = maxBytesPerSecond;
return this;
}

Builder timeWindow(Duration timeWindow) {
this.timeWindow = timeWindow;
return this;
}

BandwidthThrottlingExporter build() {
return new BandwidthThrottlingExporter(this);
}
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,113 @@
/*
* Copyright The OpenTelemetry Authors
* SPDX-License-Identifier: Apache-2.0
*/

package com.opentelemetry.android;

import static org.junit.Assert.*;
import static org.mockito.Mockito.*;

import android.util.Log;
import io.opentelemetry.sdk.common.CompletableResultCode;
import io.opentelemetry.sdk.trace.data.SpanData;
import io.opentelemetry.sdk.trace.export.SpanExporter;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.function.Function;
import org.junit.Before;
import org.junit.Test;

public class BandwidthThrottlingExporterTest {

private SpanExporter mockDelegate; // Mocked SpanExporter to simulate the delegate
private Function<SpanData, String> mockCategoryFunction; // Mocked category function
private BandwidthThrottlingExporter exporter; // Instance of the BandwidthThrottlingExporter

@Before
public void setUp() {
// Initialize the mocked delegate and category function
mockDelegate = mock(SpanExporter.class);
mockCategoryFunction = mock(Function.class);

// Create an instance of BandwidthThrottlingExporter with a max limit of 1 KB/s and a
// 1-second window
exporter =
BandwidthThrottlingExporter.newBuilder(mockDelegate)
.maxBytesPerSecond(1024) // 1 KB/s
.timeWindow(Duration.ofSeconds(1)) // 1 second
.build();
}

@Test
public void testExportWithinLimit() {
// Create a mock SpanData object
SpanData spanData = mock(SpanData.class);
when(spanData.getAttributes())
.thenReturn(Collections.singletonMap("key", "value")); // Simulate attributes

// Simulate the export of a single span
CompletableResultCode result = exporter.export(Collections.singletonList(spanData));

// Verify that the delegate's export method was called
verify(mockDelegate).export(anyList());
assertTrue(result.isSuccess()); // Ensure the export was successful
}

@Test
public void testExportExceedingLimit() {
// Create two mock SpanData objects
SpanData spanData1 = mock(SpanData.class);
SpanData spanData2 = mock(SpanData.class);
when(spanData1.getAttributes())
.thenReturn(Collections.singletonMap("key1", "value1")); // Simulate attributes
when(spanData2.getAttributes())
.thenReturn(Collections.singletonMap("key2", "value2")); // Simulate attributes

// Simulate exporting spans that exceed the limit
CompletableResultCode result = exporter.export(Arrays.asList(spanData1, spanData2));

// Verify that the delegate's export method was called only once
verify(mockDelegate, times(1)).export(anyList());

// Ensure the result is successful
assertTrue(result.isSuccess());
}

@Test
public void testExportWithDifferentCategories() {
// Create two mock SpanData objects with different attributes
SpanData spanData1 = mock(SpanData.class);
SpanData spanData2 = mock(SpanData.class);
when(spanData1.getAttributes()).thenReturn(Collections.singletonMap("key1", "value1"));
when(spanData2.getAttributes()).thenReturn(Collections.singletonMap("key2", "value2"));

// Simulate exporting both spans
CompletableResultCode result = exporter.export(Arrays.asList(spanData1, spanData2));

// Verify that the delegate's export method was called
verify(mockDelegate).export(anyList());

// Ensure the result is successful
assertTrue(result.isSuccess());
}

@Test
public void testThrottlingLogMessage() {
// Create a mock SpanData object that exceeds the limit
SpanData spanData = mock(SpanData.class);
when(spanData.getAttributes()).thenReturn(Collections.singletonMap("key", "value"));

// Export the span multiple times to exceed the limit
exporter.export(Collections.singletonList(spanData));
exporter.export(Collections.singletonList(spanData)); // This should be throttled

// Capture the log output
// Note: In a real test, you might want to use a logging framework that allows capturing
// logs
// Here we just verify that the log message is generated
// This is a placeholder as capturing logs in unit tests can be complex
Log.d("BandwidthThrottlingExporter", "Throttled span: " + spanData.getName());
}
}
Loading