Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,19 @@
package io.opentelemetry.exporter.internal;

import io.opentelemetry.sdk.common.export.GrpcStatusCode;
import java.time.Duration;
import java.time.Instant;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import javax.annotation.Nullable;

/**
* This class is internal and is hence not for public use. Its APIs are unstable and can change at
Expand Down Expand Up @@ -47,4 +55,37 @@ public static Set<String> retryableGrpcStatusCodes() {
public static Set<Integer> retryableHttpResponseCodes() {
return RETRYABLE_HTTP_STATUS_CODES;
}

/**
* Returns the delay specified by a {@code Retry-After} header, or empty if the value is absent or
* malformed.
*/
public static OptionalLong retryAfterNanos(@Nullable String retryAfter) {
Comment thread
ADITYA-CODE-SOURCE marked this conversation as resolved.
return retryAfterNanos(retryAfter, Instant.now());
}

static OptionalLong retryAfterNanos(@Nullable String retryAfter, Instant now) {
if (retryAfter == null) {
return OptionalLong.empty();
}

try {
long delaySeconds = Long.parseLong(retryAfter);
if (delaySeconds < 0) {
return OptionalLong.empty();
}
return OptionalLong.of(TimeUnit.SECONDS.toNanos(delaySeconds));
} catch (NumberFormatException ignored) {
// Fall through and try the HTTP-date form.
}

try {
Instant retryAfterInstant =
ZonedDateTime.parse(retryAfter, DateTimeFormatter.RFC_1123_DATE_TIME).toInstant();
Comment thread
ADITYA-CODE-SOURCE marked this conversation as resolved.
long delayNanos = Duration.between(now, retryAfterInstant).toNanos();
return OptionalLong.of(Math.max(0, delayNanos));
} catch (DateTimeParseException | ArithmeticException ignored) {
return OptionalLong.empty();
}
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
/*
* Copyright The OpenTelemetry Authors
* SPDX-License-Identifier: Apache-2.0
*/

package io.opentelemetry.exporter.internal;

import static org.assertj.core.api.Assertions.assertThat;

import java.time.Instant;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.OptionalLong;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Test;

class RetryUtilTest {

@Test
void retryAfterNull() {
OptionalLong delayNanos = RetryUtil.retryAfterNanos(null, Instant.EPOCH);

assertThat(delayNanos).isEmpty();
}

@Test
void retryAfterSeconds() {
OptionalLong delayNanos = RetryUtil.retryAfterNanos("30", Instant.EPOCH);

assertThat(delayNanos).hasValue(TimeUnit.SECONDS.toNanos(30));
}

@Test
void retryAfterNegativeSeconds() {
OptionalLong delayNanos = RetryUtil.retryAfterNanos("-1", Instant.EPOCH);

assertThat(delayNanos).isEmpty();
}

@Test
void retryAfterDate() {
Instant now = Instant.parse("2026-07-17T00:00:00Z");
String retryAfter =
ZonedDateTime.ofInstant(now.plusSeconds(45), ZoneOffset.UTC)
.format(DateTimeFormatter.RFC_1123_DATE_TIME);

OptionalLong delayNanos = RetryUtil.retryAfterNanos(retryAfter, now);

assertThat(delayNanos).hasValue(TimeUnit.SECONDS.toNanos(45));
}

@Test
void retryAfterPastDateClampsToZero() {
Instant now = Instant.parse("2026-07-17T00:00:00Z");
String retryAfter =
ZonedDateTime.ofInstant(now.minusSeconds(1), ZoneOffset.UTC)
.format(DateTimeFormatter.RFC_1123_DATE_TIME);

OptionalLong delayNanos = RetryUtil.retryAfterNanos(retryAfter, now);

assertThat(delayNanos).hasValue(0L);
}

@Test
void retryAfterMalformed() {
OptionalLong delayNanos = RetryUtil.retryAfterNanos("bad-value", Instant.EPOCH);

assertThat(delayNanos).isEmpty();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -729,6 +729,34 @@ void retryableError_tooManyAttempts() {
assertThat(attempts).hasValue(2);
}

@Test
void retryableError_retryAfterHonored() {
addHttpResponse(502, "0");

assertThat(
exporter
.export(Collections.singletonList(generateFakeTelemetry()))
.join(10, TimeUnit.SECONDS)
.isSuccess())
.isTrue();

assertThat(attempts).hasValue(2);
}

@Test
void retryableError_malformedRetryAfterFallsBack() {
addHttpResponse(503, "not-a-retry-after");

assertThat(
exporter
.export(Collections.singletonList(generateFakeTelemetry()))
.join(10, TimeUnit.SECONDS)
.isSuccess())
.isTrue();

assertThat(attempts).hasValue(2);
}

@ParameterizedTest
@SuppressLogger(HttpExporter.class)
@ValueSource(ints = {400, 401, 403, 500, 501})
Expand Down Expand Up @@ -1208,6 +1236,14 @@ private static void addHttpResponse(int code) {
httpErrors.add(HttpResponse.of(code));
}

private static void addHttpResponse(int code, String retryAfter) {
httpErrors.add(
HttpResponse.of(
ResponseHeaders.builder(HttpStatus.valueOf(code))
.add("Retry-After", retryAfter)
.build()));
}

private static void addHttpResponse(int code, AbstractMessageLite<?, ?> bodyMessage) {
httpErrors.add(
HttpResponse.of(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

package io.opentelemetry.exporter.sender.jdk.internal;

import io.opentelemetry.exporter.internal.RetryUtil;
import io.opentelemetry.sdk.common.CompletableResultCode;
import io.opentelemetry.sdk.common.export.Compressor;
import io.opentelemetry.sdk.common.export.HttpResponse;
Expand All @@ -27,6 +28,7 @@
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
Expand Down Expand Up @@ -54,7 +56,7 @@
*/
public final class JdkHttpSender implements HttpSender {

private static final Set<Integer> retryableStatusCodes = Set.of(429, 502, 503, 504);
private static final Set<Integer> retryableStatusCodes = RetryUtil.retryableHttpResponseCodes();

private static final ThreadLocal<NoCopyByteArrayOutputStream> threadLocalBaos =
ThreadLocal.withInitial(NoCopyByteArrayOutputStream::new);
Expand Down Expand Up @@ -212,21 +214,30 @@ HttpResponse sendInternal(MessageWriter requestBodyWriter) throws IOException {

// If no retry policy, short circuit
if (retryPolicy == null) {
return sendRequest(requestBuilder, byteBufferPool);
return toHttpResponse(sendRequest(requestBuilder, byteBufferPool));
}

long attempt = 0;
long nextBackoffNanos = retryPolicy.getInitialBackoff().toNanos();
HttpResponse httpResponse = null;
IOException exception = null;
OptionalLong retryDelayNanos = OptionalLong.empty();
do {
if (attempt > 0) {
long remainingNanos = timeout.toNanos() - (System.nanoTime() - startTimeNanos);
if (remainingNanos <= 0) {
break;
}
// Compute and sleep for backoff
long currentBackoffNanos =
Math.min(nextBackoffNanos, retryPolicy.getMaxBackoff().toNanos());
long backoffNanos =
(long) (ThreadLocalRandom.current().nextDouble(0.8d, 1.2d) * currentBackoffNanos);
long requestedBackoffNanos =
retryDelayNanos.isPresent()
? retryDelayNanos.getAsLong()
: (long) (ThreadLocalRandom.current().nextDouble(0.8d, 1.2d) * currentBackoffNanos);
long backoffNanos = Math.min(requestedBackoffNanos, remainingNanos);
nextBackoffNanos = (long) (currentBackoffNanos * retryPolicy.getBackoffMultiplier());
retryDelayNanos = OptionalLong.empty();
try {
TimeUnit.NANOSECONDS.sleep(backoffNanos);
} catch (InterruptedException e) {
Expand All @@ -243,8 +254,10 @@ HttpResponse sendInternal(MessageWriter requestBodyWriter) throws IOException {
exception = null;
requestBuilder.timeout(timeout.minusNanos(System.nanoTime() - startTimeNanos));
try {
httpResponse = sendRequest(requestBuilder, byteBufferPool);
boolean retryable = retryableStatusCodes.contains(httpResponse.getStatusCode());
java.net.http.HttpResponse<InputStream> rawResponse =
sendRequest(requestBuilder, byteBufferPool);
httpResponse = toHttpResponse(rawResponse);
boolean retryable = retryableStatusCodes.contains(rawResponse.statusCode());
if (logger.isLoggable(Level.FINER)) {
logger.log(
Level.FINER,
Expand All @@ -258,6 +271,7 @@ HttpResponse sendInternal(MessageWriter requestBodyWriter) throws IOException {
if (!retryable) {
return httpResponse;
}
retryDelayNanos = retryDelayNanos(rawResponse);
} catch (IOException e) {
exception = e;
boolean retryable = retryExceptionPredicate.test(exception);
Expand Down Expand Up @@ -287,12 +301,10 @@ private static String responseStringRepresentation(HttpResponse response) {
return "HttpResponse{code=" + response.getStatusCode() + "}";
}

private HttpResponse sendRequest(
private java.net.http.HttpResponse<InputStream> sendRequest(
HttpRequest.Builder requestBuilder, ByteBufferPool byteBufferPool) throws IOException {
try {
java.net.http.HttpResponse<InputStream> response =
client.send(requestBuilder.build(), BodyHandlers.ofInputStream());
return toHttpResponse(response);
return client.send(requestBuilder.build(), BodyHandlers.ofInputStream());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException(e);
Expand Down Expand Up @@ -412,6 +424,10 @@ private void resetPool() {
}
}

private static OptionalLong retryDelayNanos(java.net.http.HttpResponse<?> response) {
return RetryUtil.retryAfterNanos(response.headers().firstValue("Retry-After").orElse(null));
}

@Override
public CompletableResultCode shutdown() {
if (managedExecutor) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.OptionalLong;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
Expand Down Expand Up @@ -116,7 +117,8 @@ public OkHttpGrpcSender(
.connectTimeout(Duration.ofMillis(connectTimeoutMillis));
if (retryPolicy != null) {
clientBuilder.addInterceptor(
new RetryInterceptor(retryPolicy, OkHttpGrpcSender::isRetryable));
new RetryInterceptor(
retryPolicy, OkHttpGrpcSender::isRetryable, response -> OptionalLong.empty()));
}

boolean isPlainHttp = endpoint.startsWith("http://");
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.OptionalLong;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
Expand Down Expand Up @@ -103,7 +104,9 @@ public OkHttpHttpSender(
}

if (retryPolicy != null) {
builder.addInterceptor(new RetryInterceptor(retryPolicy, OkHttpHttpSender::isRetryable));
builder.addInterceptor(
new RetryInterceptor(
retryPolicy, OkHttpHttpSender::isRetryable, OkHttpHttpSender::retryDelayNanos));
}

boolean isPlainHttp = endpoint.getScheme().equals("http");
Expand All @@ -121,6 +124,10 @@ public OkHttpHttpSender(
this.maxResponseBodySize = maxResponseBodySize;
}

private static OptionalLong retryDelayNanos(Response response) {
return RetryUtil.retryAfterNanos(response.header("Retry-After"));
}

@Override
public void send(
MessageWriter messageWriter, Consumer<HttpResponse> onResponse, Consumer<Throwable> onError) {
Expand Down
Loading
Loading