Skip to content

Commit ffbf970

Browse files
leoromanovskyleoromanovsky
committed
Address agentless poller review feedback
1 parent 7aa2181 commit ffbf970

3 files changed

Lines changed: 169 additions & 21 deletions

File tree

‎products/feature-flagging/feature-flagging-api/README.md‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,12 @@ OTEL_EXPORTER_OTLP_PROTOCOL=grpc
8989
backend. Set `DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL` to a
9090
different HTTP backend while keeping agentless delivery semantics. A bare
9191
host uses the standard server-distribution path; a URL with a path is used as
92-
the exact UFC endpoint. `remote_config` uses the existing Agent Remote
92+
the exact UFC endpoint. Configured URLs are opaque: the SDK does not add the
93+
Datadog-managed `dd_env` query parameter, so custom backends must include any
94+
required tenant or environment scope in the configured URL. The derived
95+
Datadog-managed endpoint is intended for supported commercial sites; use an
96+
explicit base URL elsewhere. Agentless responses do not have an SDK-imposed
97+
payload-size limit. `remote_config` uses the existing Agent Remote
9398
Configuration path. `offline` is reserved for startup-provided UFC bytes;
9499
until those bytes are implemented, no network source starts and evaluations
95100
use defaults.

‎products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessConfigurationSource.java‎

Lines changed: 19 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ final class AgentlessConfigurationSource implements ConfigurationSourceService {
4040
private static final String DATADOG_API_SERVER_DISTRIBUTION_PATH =
4141
"/api/v2/feature-flagging/config/server-distribution";
4242
private static final int MAX_ATTEMPTS = 3;
43-
private static final int MINUTES_BETWEEN_AUTH_WARNINGS = 5;
43+
private static final int MINUTES_BETWEEN_WARNINGS = 5;
4444
private static final long FIRST_RETRY_MIN_MILLIS = 2_000;
4545
private static final long FIRST_RETRY_MAX_MILLIS = 10_000;
4646
private static final long SECOND_RETRY_MIN_MILLIS = 5_000;
@@ -78,7 +78,7 @@ private AgentlessConfigurationSource(final Config config, final HttpUrl endpoint
7878
new AgentThreadFactory(FEATURE_FLAG_CONFIGURATION_POLLER)),
7979
TimeUnit.MILLISECONDS::sleep,
8080
() -> ThreadLocalRandom.current().nextDouble(1 - RETRY_JITTER, 1 + RETRY_JITTER),
81-
new RatelimitedLogger(LOGGER, MINUTES_BETWEEN_AUTH_WARNINGS, TimeUnit.MINUTES));
81+
new RatelimitedLogger(LOGGER, MINUTES_BETWEEN_WARNINGS, TimeUnit.MINUTES));
8282
}
8383

8484
AgentlessConfigurationSource(
@@ -95,7 +95,7 @@ private AgentlessConfigurationSource(final Config config, final HttpUrl endpoint
9595
executor,
9696
TimeUnit.MILLISECONDS::sleep,
9797
() -> 1.0,
98-
new RatelimitedLogger(LOGGER, MINUTES_BETWEEN_AUTH_WARNINGS, TimeUnit.MINUTES));
98+
new RatelimitedLogger(LOGGER, MINUTES_BETWEEN_WARNINGS, TimeUnit.MINUTES));
9999
}
100100

101101
AgentlessConfigurationSource(
@@ -114,7 +114,7 @@ private AgentlessConfigurationSource(final Config config, final HttpUrl endpoint
114114
executor,
115115
retrySleeper,
116116
jitter,
117-
new RatelimitedLogger(LOGGER, MINUTES_BETWEEN_AUTH_WARNINGS, TimeUnit.MINUTES));
117+
new RatelimitedLogger(LOGGER, MINUTES_BETWEEN_WARNINGS, TimeUnit.MINUTES));
118118
}
119119

120120
AgentlessConfigurationSource(
@@ -192,11 +192,18 @@ private boolean fetchAndApply() {
192192
if (closed) {
193193
return false;
194194
}
195-
if (isRetryableStatus(response.status) && attempt < MAX_ATTEMPTS) {
196-
if (!waitBeforeRetry(attempt)) {
197-
return false;
195+
if (isRetryableStatus(response.status)) {
196+
if (attempt < MAX_ATTEMPTS) {
197+
if (!waitBeforeRetry(attempt)) {
198+
return false;
199+
}
200+
continue;
198201
}
199-
continue;
202+
ratelimitedLogger.warn(
203+
"Feature Flagging agentless endpoint failed after {} attempts with HTTP {}",
204+
MAX_ATTEMPTS,
205+
response.status);
206+
return false;
200207
}
201208
synchronized (lifecycleLock) {
202209
return !closed && apply(response);
@@ -206,7 +213,10 @@ private boolean fetchAndApply() {
206213
return false;
207214
}
208215
if (attempt == MAX_ATTEMPTS) {
209-
LOGGER.debug("Feature Flagging HTTP configuration source request failed", e);
216+
ratelimitedLogger.warn(
217+
"Feature Flagging agentless endpoint request failed after {} attempts",
218+
MAX_ATTEMPTS,
219+
e);
210220
return false;
211221
}
212222
if (!waitBeforeRetry(attempt)) {

‎products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessConfigurationSourceTest.java‎

Lines changed: 144 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545
import org.junit.jupiter.api.Nested;
4646
import org.junit.jupiter.api.Test;
4747
import org.junit.jupiter.api.extension.ExtendWith;
48+
import org.mockito.ArgumentCaptor;
4849
import org.mockito.Mock;
4950
import org.mockito.junit.jupiter.MockitoExtension;
5051

@@ -163,6 +164,41 @@ void realHttpClientSendsAgentlessHeadersAndReadsResponse() throws Exception {
163164
}
164165
}
165166

167+
@Test
168+
void downloadsAndAppliesLargeUfcWithoutPayloadLimit() throws Exception {
169+
final int flagCount = 5_000;
170+
final String largeConfig = largeConfig(flagCount);
171+
assertTrue(largeConfig.getBytes(UTF_8).length > 500_000);
172+
173+
try (JavaTestHttpServer server =
174+
JavaTestHttpServer.httpServer(
175+
s -> s.handlers(h -> h.get(CONFIG_PATH, api -> api.getResponse().send(largeConfig))))) {
176+
final HttpUrl endpoint = HttpUrl.get(server.getAddress().resolve(CONFIG_PATH));
177+
final OkHttpClient httpClient = new OkHttpClient.Builder().build();
178+
final AgentlessConfigurationSource service =
179+
new AgentlessConfigurationSource(
180+
endpoint,
181+
config(),
182+
30_000,
183+
new AgentlessConfigurationSource.OkHttpUfcHttpClient(httpClient),
184+
Executors.newSingleThreadScheduledExecutor());
185+
final ArgumentCaptor<ServerConfiguration> configuration =
186+
ArgumentCaptor.forClass(ServerConfiguration.class);
187+
FeatureFlaggingGateway.addConfigListener(listener);
188+
189+
try {
190+
assertTrue(service.pollOnce());
191+
192+
verify(listener).accept(configuration.capture());
193+
assertEquals(flagCount, configuration.getValue().flags.size());
194+
} finally {
195+
service.close();
196+
httpClient.dispatcher().executorService().shutdownNow();
197+
httpClient.connectionPool().evictAll();
198+
}
199+
}
200+
}
201+
166202
@Test
167203
void realHttpClientAllowsMissingEtagAndEmptyResponseBody() throws Exception {
168204
try (JavaTestHttpServer server =
@@ -524,33 +560,73 @@ void retriesServerErrorThenKeepsColdStateOnNotModified() throws Exception {
524560
}
525561

526562
@Test
527-
void givesUpAfterRetryableFailuresAreExhausted() throws Exception {
563+
void warnsRateLimitedAfterRetryableFailuresAreExhausted() throws Exception {
564+
final RatelimitedLogger ratelimitedLogger = mock(RatelimitedLogger.class);
528565
final FakeClient client =
529566
new FakeClient(
530567
response(503, null, null), response(503, null, null), response(503, null, null));
531-
final AgentlessConfigurationSource service = service(client);
568+
final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
569+
final AgentlessConfigurationSource service =
570+
new AgentlessConfigurationSource(
571+
HttpUrl.get("http://localhost" + CONFIG_PATH),
572+
config(),
573+
30_000,
574+
client,
575+
executor,
576+
delay -> {},
577+
() -> 1.0,
578+
ratelimitedLogger);
532579
FeatureFlaggingGateway.addConfigListener(listener);
533580

534-
assertFalse(service.pollOnce());
581+
try {
582+
assertFalse(service.pollOnce());
535583

536-
assertEquals(3, client.calls.get());
537-
verifyNoInteractions(listener);
584+
assertEquals(3, client.calls.get());
585+
verify(ratelimitedLogger)
586+
.warn(
587+
"Feature Flagging agentless endpoint failed after {} attempts with HTTP {}", 3, 503);
588+
verifyNoInteractions(listener);
589+
} finally {
590+
service.close();
591+
}
538592
}
539593

540594
@Test
541-
void givesUpAfterIoFailuresAreExhausted() throws Exception {
595+
void warnsRateLimitedAfterIoFailuresAreExhausted() throws Exception {
596+
final RatelimitedLogger ratelimitedLogger = mock(RatelimitedLogger.class);
597+
final SocketTimeoutException finalFailure =
598+
new SocketTimeoutException("slow HTTP configuration source");
542599
final FakeClient client =
543600
new FakeClient(
544601
new SocketTimeoutException("slow HTTP configuration source"),
545602
new SocketTimeoutException("slow HTTP configuration source"),
546-
new SocketTimeoutException("slow HTTP configuration source"));
547-
final AgentlessConfigurationSource service = service(client);
603+
finalFailure);
604+
final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
605+
final AgentlessConfigurationSource service =
606+
new AgentlessConfigurationSource(
607+
HttpUrl.get("http://localhost" + CONFIG_PATH),
608+
config(),
609+
30_000,
610+
client,
611+
executor,
612+
delay -> {},
613+
() -> 1.0,
614+
ratelimitedLogger);
548615
FeatureFlaggingGateway.addConfigListener(listener);
549616

550-
assertFalse(service.pollOnce());
617+
try {
618+
assertFalse(service.pollOnce());
551619

552-
assertEquals(3, client.calls.get());
553-
verifyNoInteractions(listener);
620+
assertEquals(3, client.calls.get());
621+
verify(ratelimitedLogger)
622+
.warn(
623+
"Feature Flagging agentless endpoint request failed after {} attempts",
624+
3,
625+
finalFailure);
626+
verifyNoInteractions(listener);
627+
} finally {
628+
service.close();
629+
}
554630
}
555631

556632
@Test
@@ -638,6 +714,39 @@ void repeatedInitStartsOnlyOnePoller() throws Exception {
638714
assertEquals(1, client.calls.get());
639715
}
640716

717+
@Test
718+
void scheduledPollContinuesAfterListenerRuntimeException() throws Exception {
719+
final FakeClient client =
720+
new FakeClient(
721+
response(200, "etag-a", emptyConfig()), response(200, "etag-b", emptyConfig()));
722+
final AgentlessConfigurationSource service =
723+
new AgentlessConfigurationSource(
724+
HttpUrl.get("http://localhost" + CONFIG_PATH),
725+
config(),
726+
10,
727+
client,
728+
Executors.newSingleThreadScheduledExecutor());
729+
final AtomicInteger listenerCalls = new AtomicInteger();
730+
final FeatureFlaggingGateway.ConfigListener flakyListener =
731+
configuration -> {
732+
if (listenerCalls.incrementAndGet() == 1) {
733+
throw new IllegalStateException("listener rejected first configuration");
734+
}
735+
};
736+
FeatureFlaggingGateway.addConfigListener(flakyListener);
737+
738+
try {
739+
service.init();
740+
awaitCalls(client, 2);
741+
742+
assertEquals(2, listenerCalls.get());
743+
assertNull(client.requests.get(1).etag);
744+
} finally {
745+
service.close();
746+
FeatureFlaggingGateway.removeConfigListener(flakyListener);
747+
}
748+
}
749+
641750
@Test
642751
void closeCancelsInFlightRequestAndIgnoresLateSuccess() throws Exception {
643752
final CountDownLatch requestStarted = new CountDownLatch(1);
@@ -823,6 +932,30 @@ private static String emptyConfig() {
823932
+ "}";
824933
}
825934

935+
private static String largeConfig(final int flagCount) {
936+
final StringBuilder json =
937+
new StringBuilder(
938+
"{\"createdAt\":\"2024-04-17T19:40:53.716Z\","
939+
+ "\"format\":\"SERVER\","
940+
+ "\"environment\":{\"name\":\"Large Test\"},"
941+
+ "\"flags\":{");
942+
for (int index = 0; index < flagCount; index++) {
943+
if (index > 0) {
944+
json.append(',');
945+
}
946+
final String flagKey = "large-flag-" + index;
947+
json.append('"')
948+
.append(flagKey)
949+
.append("\":{\"key\":\"")
950+
.append(flagKey)
951+
.append(
952+
"\",\"enabled\":true,\"variationType\":\"STRING\","
953+
+ "\"variations\":{\"on\":{\"key\":\"on\",\"value\":\"on\"}},"
954+
+ "\"allocations\":[]}");
955+
}
956+
return json.append("}}").toString();
957+
}
958+
826959
private static void awaitCalls(final FakeClient client, final int count) throws Exception {
827960
for (int i = 0; i < 100; i++) {
828961
if (client.calls.get() >= count) {

0 commit comments

Comments
 (0)