Skip to content

Commit 4baefdb

Browse files
leoromanovskyleoromanovsky
committed
Add Datadog-managed agentless UFC polling
1 parent b8bbd26 commit 4baefdb

4 files changed

Lines changed: 1166 additions & 1 deletion

File tree

‎internal-api/src/main/java/datadog/trace/util/AgentThreadFactory.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,8 @@ public enum AgentThread {
6666

6767
LLMOBS_EVALS_PROCESSOR("dd-llmobs-evals-processor"),
6868

69-
FEATURE_FLAG_EXPOSURE_PROCESSOR("dd-ffe-exposure-processor");
69+
FEATURE_FLAG_EXPOSURE_PROCESSOR("dd-ffe-exposure-processor"),
70+
FEATURE_FLAG_CONFIGURATION_POLLER("dd-feature-flagging-http-poller");
7071

7172
public final String threadName;
7273

‎products/feature-flagging/feature-flagging-lib/build.gradle.kts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ dependencies {
1818
api(libs.moshi)
1919
api(libs.jctools)
2020
api(project(":communication"))
21+
implementation(project(":internal-api"))
2122
api(project(":products:feature-flagging:feature-flagging-bootstrap"))
2223
api(project(":utils:queue-utils"))
2324

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

Lines changed: 358 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,358 @@
1+
package com.datadog.featureflag;
2+
3+
import static datadog.communication.http.OkHttpUtils.prepareRequest;
4+
import static datadog.trace.util.AgentThreadFactory.AgentThread.FEATURE_FLAG_CONFIGURATION_POLLER;
5+
6+
import datadog.communication.http.OkHttpUtils;
7+
import datadog.trace.api.Config;
8+
import datadog.trace.api.featureflag.FeatureFlaggingGateway;
9+
import datadog.trace.api.featureflag.ufc.v1.ServerConfiguration;
10+
import datadog.trace.util.AgentThreadFactory;
11+
import java.io.IOException;
12+
import java.net.HttpURLConnection;
13+
import java.net.URLEncoder;
14+
import java.util.HashMap;
15+
import java.util.Locale;
16+
import java.util.Map;
17+
import java.util.concurrent.Executors;
18+
import java.util.concurrent.ScheduledExecutorService;
19+
import java.util.concurrent.ScheduledFuture;
20+
import java.util.concurrent.ThreadLocalRandom;
21+
import java.util.concurrent.TimeUnit;
22+
import java.util.concurrent.atomic.AtomicBoolean;
23+
import java.util.concurrent.atomic.AtomicReference;
24+
import java.util.function.DoubleSupplier;
25+
import okhttp3.Call;
26+
import okhttp3.HttpUrl;
27+
import okhttp3.OkHttpClient;
28+
import okhttp3.Request;
29+
import okhttp3.Response;
30+
import okhttp3.ResponseBody;
31+
import org.slf4j.Logger;
32+
import org.slf4j.LoggerFactory;
33+
34+
final class AgentlessConfigurationSource implements ConfigurationSourceService {
35+
private static final Logger LOGGER = LoggerFactory.getLogger(AgentlessConfigurationSource.class);
36+
37+
// TODO before merge: confirm the final backend route with the server-distribution API owners.
38+
private static final String DATADOG_API_SERVER_DISTRIBUTION_PATH =
39+
"/api/v2/feature-flagging/config/server-distribution";
40+
private static final int MAX_ATTEMPTS = 3;
41+
private static final long FIRST_RETRY_MIN_MILLIS = 2_000;
42+
private static final long FIRST_RETRY_MAX_MILLIS = 10_000;
43+
private static final long SECOND_RETRY_MIN_MILLIS = 5_000;
44+
private static final long SECOND_RETRY_MAX_MILLIS = 30_000;
45+
private static final double RETRY_JITTER = 0.2;
46+
47+
private final HttpUrl endpoint;
48+
private final Config config;
49+
private final long pollIntervalMillis;
50+
private final UfcHttpClient client;
51+
private final ScheduledExecutorService executor;
52+
private final RetrySleeper retrySleeper;
53+
private final DoubleSupplier jitter;
54+
private final Object lifecycleLock = new Object();
55+
private final AtomicBoolean polling = new AtomicBoolean();
56+
private volatile boolean closed;
57+
private volatile ScheduledFuture<?> scheduledPoll;
58+
private volatile String etag;
59+
60+
AgentlessConfigurationSource(final Config config) {
61+
this(config, endpoint(config));
62+
}
63+
64+
private AgentlessConfigurationSource(final Config config, final HttpUrl endpoint) {
65+
this(
66+
endpoint,
67+
config,
68+
millis(config.getFeatureFlaggingConfigurationSourcePollIntervalSeconds()),
69+
new OkHttpUfcHttpClient(
70+
OkHttpUtils.buildHttpClient(
71+
endpoint,
72+
millis(config.getFeatureFlaggingConfigurationSourceRequestTimeoutSeconds()))),
73+
Executors.newSingleThreadScheduledExecutor(
74+
new AgentThreadFactory(FEATURE_FLAG_CONFIGURATION_POLLER)),
75+
TimeUnit.MILLISECONDS::sleep,
76+
() -> ThreadLocalRandom.current().nextDouble(1 - RETRY_JITTER, 1 + RETRY_JITTER));
77+
}
78+
79+
AgentlessConfigurationSource(
80+
final HttpUrl endpoint,
81+
final Config config,
82+
final long pollIntervalMillis,
83+
final UfcHttpClient client,
84+
final ScheduledExecutorService executor) {
85+
this(
86+
endpoint,
87+
config,
88+
pollIntervalMillis,
89+
client,
90+
executor,
91+
TimeUnit.MILLISECONDS::sleep,
92+
() -> 1.0);
93+
}
94+
95+
AgentlessConfigurationSource(
96+
final HttpUrl endpoint,
97+
final Config config,
98+
final long pollIntervalMillis,
99+
final UfcHttpClient client,
100+
final ScheduledExecutorService executor,
101+
final RetrySleeper retrySleeper,
102+
final DoubleSupplier jitter) {
103+
this.endpoint = endpoint;
104+
this.config = config;
105+
this.pollIntervalMillis = pollIntervalMillis;
106+
this.client = client;
107+
this.executor = executor;
108+
this.retrySleeper = retrySleeper;
109+
this.jitter = jitter;
110+
}
111+
112+
@Override
113+
public void init() {
114+
synchronized (lifecycleLock) {
115+
if (closed || scheduledPoll != null) {
116+
return;
117+
}
118+
scheduledPoll =
119+
executor.scheduleWithFixedDelay(
120+
this::pollOnceSafely, 0, pollIntervalMillis, TimeUnit.MILLISECONDS);
121+
}
122+
}
123+
124+
boolean pollOnce() {
125+
if (closed || !polling.compareAndSet(false, true)) {
126+
return false;
127+
}
128+
try {
129+
return fetchAndApply();
130+
} finally {
131+
polling.set(false);
132+
}
133+
}
134+
135+
@Override
136+
public void close() {
137+
final ScheduledFuture<?> poll;
138+
synchronized (lifecycleLock) {
139+
if (closed) {
140+
return;
141+
}
142+
closed = true;
143+
poll = scheduledPoll;
144+
scheduledPoll = null;
145+
}
146+
if (poll != null) {
147+
poll.cancel(true);
148+
}
149+
client.cancel();
150+
executor.shutdownNow();
151+
}
152+
153+
private void pollOnceSafely() {
154+
try {
155+
pollOnce();
156+
} catch (final RuntimeException e) {
157+
LOGGER.debug("Unexpected error while polling Feature Flagging HTTP configuration source", e);
158+
}
159+
}
160+
161+
private boolean fetchAndApply() {
162+
for (int attempt = 1; ; attempt++) {
163+
try {
164+
final UfcHttpResponse response = client.fetch(endpoint, config, etag);
165+
if (closed) {
166+
return false;
167+
}
168+
if (isRetryableStatus(response.status) && attempt < MAX_ATTEMPTS) {
169+
if (!waitBeforeRetry(attempt)) {
170+
return false;
171+
}
172+
continue;
173+
}
174+
synchronized (lifecycleLock) {
175+
return !closed && apply(response);
176+
}
177+
} catch (final IOException e) {
178+
if (closed) {
179+
return false;
180+
}
181+
if (attempt == MAX_ATTEMPTS) {
182+
LOGGER.debug("Feature Flagging HTTP configuration source request failed", e);
183+
return false;
184+
}
185+
if (!waitBeforeRetry(attempt)) {
186+
return false;
187+
}
188+
}
189+
}
190+
}
191+
192+
private boolean waitBeforeRetry(final int attempt) {
193+
if (closed) {
194+
return false;
195+
}
196+
try {
197+
retrySleeper.sleep(retryDelayMillis(pollIntervalMillis, attempt, jitter.getAsDouble()));
198+
return !closed;
199+
} catch (final InterruptedException e) {
200+
Thread.currentThread().interrupt();
201+
return false;
202+
}
203+
}
204+
205+
private boolean apply(final UfcHttpResponse response) {
206+
if (response.status == HttpURLConnection.HTTP_NOT_MODIFIED) {
207+
return true;
208+
}
209+
if (response.status == HttpURLConnection.HTTP_UNAUTHORIZED
210+
|| response.status == HttpURLConnection.HTTP_FORBIDDEN
211+
|| response.status != HttpURLConnection.HTTP_OK
212+
|| response.body == null) {
213+
return false;
214+
}
215+
final ServerConfiguration configuration;
216+
try {
217+
configuration =
218+
RemoteConfigServiceImpl.UniversalFlagConfigDeserializer.INSTANCE.deserialize(
219+
response.body);
220+
} catch (final IOException | RuntimeException e) {
221+
LOGGER.debug("Feature Flagging HTTP configuration source returned malformed UFC payload", e);
222+
return false;
223+
}
224+
if (configuration == null) {
225+
return false;
226+
}
227+
FeatureFlaggingGateway.dispatch(configuration);
228+
updateEtag(response.etag);
229+
return true;
230+
}
231+
232+
private static boolean isRetryableStatus(final int status) {
233+
return status == HttpURLConnection.HTTP_CLIENT_TIMEOUT
234+
|| status == 429
235+
|| (status >= 500 && status <= 599);
236+
}
237+
238+
private void updateEtag(final String nextEtag) {
239+
if (nextEtag != null && !nextEtag.trim().isEmpty()) {
240+
etag = nextEtag;
241+
}
242+
}
243+
244+
static HttpUrl endpoint(final Config config) {
245+
final String endpoint = datadogApiServerDistributionEndpoint(config);
246+
final HttpUrl parsed = HttpUrl.parse(endpoint);
247+
if (parsed == null) {
248+
throw new IllegalArgumentException(
249+
"Invalid Feature Flagging HTTP configuration source URL: " + endpoint);
250+
}
251+
return parsed;
252+
}
253+
254+
private static String datadogApiServerDistributionEndpoint(final Config config) {
255+
final StringBuilder endpoint =
256+
new StringBuilder("https://api.")
257+
.append(config.getSite().toLowerCase(Locale.ROOT))
258+
.append(DATADOG_API_SERVER_DISTRIBUTION_PATH);
259+
final String env = config.getEnv();
260+
if (env != null && !env.isEmpty()) {
261+
endpoint.append("?dd_env=").append(urlEncode(env));
262+
}
263+
return endpoint.toString();
264+
}
265+
266+
private static String urlEncode(final String value) {
267+
try {
268+
return URLEncoder.encode(value, "UTF-8");
269+
} catch (final IOException e) {
270+
throw new IllegalArgumentException("Unable to encode Feature Flagging environment", e);
271+
}
272+
}
273+
274+
static long millis(final int seconds) {
275+
return TimeUnit.SECONDS.toMillis(seconds);
276+
}
277+
278+
static long retryDelayMillis(
279+
final long pollIntervalMillis, final int attempt, final double jitter) {
280+
final long baseDelay;
281+
if (attempt == 1) {
282+
baseDelay = clamp(pollIntervalMillis / 6, FIRST_RETRY_MIN_MILLIS, FIRST_RETRY_MAX_MILLIS);
283+
} else if (attempt == 2) {
284+
baseDelay = clamp(pollIntervalMillis / 3, SECOND_RETRY_MIN_MILLIS, SECOND_RETRY_MAX_MILLIS);
285+
} else {
286+
throw new IllegalArgumentException("Unsupported Feature Flagging retry attempt: " + attempt);
287+
}
288+
return Math.max(1, Math.round(baseDelay * jitter));
289+
}
290+
291+
private static long clamp(final long value, final long minimum, final long maximum) {
292+
return Math.max(minimum, Math.min(maximum, value));
293+
}
294+
295+
interface UfcHttpClient {
296+
UfcHttpResponse fetch(HttpUrl endpoint, Config config, String etag) throws IOException;
297+
298+
void cancel();
299+
}
300+
301+
interface RetrySleeper {
302+
void sleep(long delayMillis) throws InterruptedException;
303+
}
304+
305+
static final class UfcHttpResponse {
306+
final int status;
307+
final String etag;
308+
final byte[] body;
309+
310+
UfcHttpResponse(final int status, final String etag, final byte[] body) {
311+
this.status = status;
312+
this.etag = etag;
313+
this.body = body;
314+
}
315+
}
316+
317+
static final class OkHttpUfcHttpClient implements UfcHttpClient {
318+
private final OkHttpClient httpClient;
319+
private final AtomicReference<Call> activeCall = new AtomicReference<>();
320+
private final AtomicBoolean cancelled = new AtomicBoolean();
321+
322+
OkHttpUfcHttpClient(final OkHttpClient httpClient) {
323+
this.httpClient = httpClient;
324+
}
325+
326+
@Override
327+
public UfcHttpResponse fetch(final HttpUrl endpoint, final Config config, final String etag)
328+
throws IOException {
329+
final Map<String, String> headers = new HashMap<>();
330+
if (etag != null) {
331+
headers.put("If-None-Match", etag);
332+
}
333+
final Request request = prepareRequest(endpoint, headers, config, true).get().build();
334+
final Call call = httpClient.newCall(request);
335+
if (!activeCall.compareAndSet(null, call)) {
336+
throw new IllegalStateException("Feature Flagging HTTP request already in flight");
337+
}
338+
if (cancelled.get()) {
339+
call.cancel();
340+
}
341+
try (Response response = call.execute()) {
342+
final ResponseBody responseBody = response.body();
343+
return new UfcHttpResponse(response.code(), response.header("ETag"), responseBody.bytes());
344+
} finally {
345+
activeCall.compareAndSet(call, null);
346+
}
347+
}
348+
349+
@Override
350+
public void cancel() {
351+
cancelled.set(true);
352+
final Call call = activeCall.get();
353+
if (call != null) {
354+
call.cancel();
355+
}
356+
}
357+
}
358+
}

0 commit comments

Comments
 (0)