Skip to content

Commit f93014e

Browse files
liran2000liran2000
committed
adding http connector - cont.
Signed-off-by: liran2000 <[email protected]>
1 parent b4ae05c commit f93014e

File tree

6 files changed

+254
-46
lines changed

6 files changed

+254
-46
lines changed

‎providers/flagd/README.md‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,39 @@ The value is updated with every (re)connection to the sync implementation.
5454
This can be used to enrich evaluations with such data.
5555
If the `in-process` mode is not used, and before the provider is ready, the `getSyncMetadata` returns an empty map.
5656

57+
#### Http Connector
58+
HttpConnector is responsible for polling data from a specified URL at regular intervals.
59+
The implementation is using Java HttpClient.
60+
61+
##### What happens if the Http source is down when application is starting ?
62+
63+
It supports optional fail-safe initialization via cache, such that on initial fetch error following by
64+
source downtime window, initial payload is taken from cache to avoid starting with default values until
65+
the source is back up. Therefore, the cache ttl expected to be higher than the expected source
66+
down-time to recover from during initialization.
67+
68+
##### Sample flow
69+
Sample flow can use:
70+
- Github as the flags payload source.
71+
- Redis cache as a fail-safe initialization cache.
72+
73+
Sample flow of initialization during Github down-time window, showing that application can still use flags
74+
values as fetched from cache.
75+
```mermaid
76+
sequenceDiagram
77+
participant Provider
78+
participant Github
79+
participant Redis
80+
81+
break source downtime
82+
Provider->>Github: initialize
83+
Github->>Provider: failure
84+
end
85+
Provider->>Redis: fetch
86+
Redis->>Provider: last payload
87+
88+
```
89+
5790
### Offline mode (File resolver)
5891

5992
In-process resolvers can also work in an offline mode.

‎providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/HttpConnector.java‎

Lines changed: 92 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -30,18 +30,22 @@
3030
import static java.net.http.HttpClient.Builder.NO_PROXY;
3131

3232
/**
33-
* HttpConnector is responsible for managing HTTP connections and polling data from a specified URL
34-
* at regular intervals. It implements the QueueSource interface to enqueue and dequeue change messages.
33+
* HttpConnector is responsible for polling data from a specified URL at regular intervals.
34+
* Notice rate limits for polling http sources like Github.
35+
* It implements the QueueSource interface to enqueue and dequeue change messages.
3536
* The class supports configurable parameters such as poll interval, request timeout, and proxy settings.
3637
* It uses a ScheduledExecutorService to schedule polling tasks and an ExecutorService for HTTP client execution.
3738
* The class also provides methods to initialize, retrieve the stream queue, and shutdown the connector gracefully.
39+
* It supports optional fail-safe initialization via cache.
40+
*
41+
* See readme - Http Connector section.
3842
*/
3943
@Slf4j
4044
public class HttpConnector implements QueueSource {
4145

4246
private static final int DEFAULT_POLL_INTERVAL_SECONDS = 60;
4347
private static final int DEFAULT_LINKED_BLOCKING_QUEUE_CAPACITY = 100;
44-
private static final int DEFAULT_SCHEDULED_THREAD_POOL_SIZE = 1;
48+
private static final int DEFAULT_SCHEDULED_THREAD_POOL_SIZE = 2;
4549
private static final int DEFAULT_REQUEST_TIMEOUT_SECONDS = 10;
4650
private static final int DEFAULT_CONNECT_TIMEOUT_SECONDS = 10;
4751

@@ -52,19 +56,19 @@ public class HttpConnector implements QueueSource {
5256
private ExecutorService httpClientExecutor;
5357
private ScheduledExecutorService scheduler;
5458
private Map<String, String> headers;
59+
private PayloadCacheWrapper payloadCacheWrapper;
60+
private PayloadCache payloadCache;
61+
5562
@NonNull
5663
private String url;
5764

58-
// TODO init failure backup cache redis
59-
60-
// todo update provider readme
61-
6265
@Builder
6366
public HttpConnector(Integer pollIntervalSeconds, Integer linkedBlockingQueueCapacity,
6467
Integer scheduledThreadPoolSize, Integer requestTimeoutSeconds, Integer connectTimeoutSeconds, String url,
65-
Map<String, String> headers, ExecutorService httpClientExecutor, String proxyHost, Integer proxyPort) {
68+
Map<String, String> headers, ExecutorService httpClientExecutor, String proxyHost, Integer proxyPort,
69+
PayloadCacheOptions payloadCacheOptions, PayloadCache payloadCache) {
6670
validate(url, pollIntervalSeconds, linkedBlockingQueueCapacity, scheduledThreadPoolSize, requestTimeoutSeconds,
67-
connectTimeoutSeconds, proxyHost, proxyPort);
71+
connectTimeoutSeconds, proxyHost, proxyPort, payloadCacheOptions, payloadCache);
6872
this.pollIntervalSeconds = pollIntervalSeconds == null ? DEFAULT_POLL_INTERVAL_SECONDS : pollIntervalSeconds;
6973
int thisLinkedBlockingQueueCapacity = linkedBlockingQueueCapacity == null ? DEFAULT_LINKED_BLOCKING_QUEUE_CAPACITY : linkedBlockingQueueCapacity;
7074
int thisScheduledThreadPoolSize = scheduledThreadPoolSize == null ? DEFAULT_SCHEDULED_THREAD_POOL_SIZE : scheduledThreadPoolSize;
@@ -89,12 +93,20 @@ public HttpConnector(Integer pollIntervalSeconds, Integer linkedBlockingQueueCap
8993
.executor(this.httpClientExecutor)
9094
.build();
9195
this.queue = new LinkedBlockingQueue<>(thisLinkedBlockingQueueCapacity);
96+
this.payloadCache = payloadCache;
97+
if (payloadCache != null) {
98+
this.payloadCacheWrapper = PayloadCacheWrapper.builder()
99+
.payloadCache(payloadCache)
100+
.payloadCacheOptions(payloadCacheOptions)
101+
.build();
102+
}
92103
}
93104

94105
@SneakyThrows
95106
private void validate(String url, Integer pollIntervalSeconds, Integer linkedBlockingQueueCapacity,
96-
Integer scheduledThreadPoolSize, Integer requestTimeoutSeconds, Integer connectTimeoutSeconds,
97-
String proxyHost, Integer proxyPort) {
107+
Integer scheduledThreadPoolSize, Integer requestTimeoutSeconds, Integer connectTimeoutSeconds,
108+
String proxyHost, Integer proxyPort, PayloadCacheOptions payloadCacheOptions,
109+
PayloadCache payloadCache) {
98110
new URL(url).toURI();
99111
if (pollIntervalSeconds != null && (pollIntervalSeconds < 1 || pollIntervalSeconds > 600)) {
100112
throw new IllegalArgumentException("pollIntervalSeconds must be between 1 and 600");
@@ -119,6 +131,12 @@ private void validate(String url, Integer pollIntervalSeconds, Integer linkedBlo
119131
} else if (proxyHost == null && proxyPort != null) {
120132
throw new IllegalArgumentException("proxyHost must be set if proxyPort is set");
121133
}
134+
if (payloadCacheOptions != null && payloadCache == null) {
135+
throw new IllegalArgumentException("payloadCache must be set if payloadCacheOptions is set");
136+
}
137+
if (payloadCache != null && payloadCacheOptions == null) {
138+
throw new IllegalArgumentException("payloadCacheOptions must be set if payloadCache is set");
139+
}
122140
}
123141

124142
@Override
@@ -128,51 +146,79 @@ public void init() throws Exception {
128146

129147
@Override
130148
public BlockingQueue<QueuePayload> getStreamQueue() {
149+
boolean success = fetchAndUpdate();
150+
if (!success) {
151+
log.info("failed initial fetch");
152+
if (payloadCache != null) {
153+
updateFromCache();
154+
}
155+
}
131156
Runnable pollTask = buildPollTask();
132-
133-
// run first poll immediately and wait for it to finish
134-
pollTask.run();
135-
136157
scheduler.scheduleAtFixedRate(pollTask, pollIntervalSeconds, pollIntervalSeconds, TimeUnit.SECONDS);
137158
return queue;
138159
}
139160

161+
private void updateFromCache() {
162+
log.info("taking initial payload from cache to avoid starting with default values");
163+
String flagData = payloadCache.get();
164+
if (flagData == null) {
165+
log.debug("got null from cache");
166+
return;
167+
}
168+
if (!this.queue.offer(new QueuePayload(QueuePayloadType.DATA, flagData))) {
169+
log.warn("init: Unable to offer file content to queue: queue is full");
170+
}
171+
}
172+
140173
protected Runnable buildPollTask() {
141-
return () -> {
142-
HttpRequest.Builder requestBuilder = HttpRequest.newBuilder()
143-
.uri(URI.create(url))
144-
.timeout(Duration.ofSeconds(requestTimeoutSeconds))
145-
.GET();
146-
headers.forEach(requestBuilder::header);
147-
HttpRequest request = requestBuilder
148-
.build();
174+
return this::fetchAndUpdate;
175+
}
149176

150-
HttpResponse<String> response;
151-
try {
152-
log.debug("fetching response");
153-
response = execute(request);
154-
} catch (IOException e) {
155-
log.info("could not fetch", e);
156-
return;
157-
} catch (Exception e) {
158-
log.debug("exception", e);
159-
return;
160-
}
161-
log.debug("fetched response");
162-
if (response.statusCode() != 200) {
163-
log.info("received non-successful status code: {} {}", response.statusCode(), response.body());
164-
return;
165-
}
166-
if (!this.queue.offer(new QueuePayload(QueuePayloadType.DATA, response.body()))) {
167-
log.warn("Unable to offer file content to queue: queue is full");
168-
}
169-
};
177+
private boolean fetchAndUpdate() {
178+
HttpRequest.Builder requestBuilder = HttpRequest.newBuilder()
179+
.uri(URI.create(url))
180+
.timeout(Duration.ofSeconds(requestTimeoutSeconds))
181+
.GET();
182+
headers.forEach(requestBuilder::header);
183+
HttpRequest request = requestBuilder
184+
.build();
185+
186+
HttpResponse<String> response;
187+
try {
188+
log.debug("fetching response");
189+
response = execute(request);
190+
} catch (IOException e) {
191+
log.info("could not fetch", e);
192+
return false;
193+
} catch (Exception e) {
194+
log.debug("exception", e);
195+
return false;
196+
}
197+
log.debug("fetched response");
198+
String payload = response.body();
199+
if (response.statusCode() != 200) {
200+
log.info("received non-successful status code: {} {}", response.statusCode(), payload);
201+
return false;
202+
}
203+
if (payload == null) {
204+
log.debug("payload is null");
205+
return false;
206+
}
207+
if (!this.queue.offer(new QueuePayload(QueuePayloadType.DATA, payload))) {
208+
log.warn("Unable to offer file content to queue: queue is full");
209+
return false;
210+
}
211+
if (payloadCacheWrapper != null) {
212+
log.debug("scheduling cache update if needed");
213+
scheduler.execute(() ->
214+
payloadCacheWrapper.updatePayloadIfNeeded(payload)
215+
);
216+
}
217+
return payload != null;
170218
}
171219

172220
protected HttpResponse<String> execute(HttpRequest request) throws IOException, InterruptedException {
173-
HttpResponse<String> response;
174-
response = client.send(request, HttpResponse.BodyHandlers.ofString());
175-
return response;
221+
return client.send(request, HttpResponse.BodyHandlers.ofString());
176222
}
177223

178224
@Override

‎providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/PayloadCache.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
package dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.sync;
2+
3+
public interface PayloadCache {
4+
public void put(String payload);
5+
public String get();
6+
}

‎providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/PayloadCacheOptions.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
package dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.sync;
2+
3+
import lombok.Builder;
4+
import lombok.Getter;
5+
6+
/**
7+
* Represents configuration options for caching payloads.
8+
* <p>
9+
* This class provides options to configure the caching behavior,
10+
* specifically the interval at which the cache should be updated.
11+
* </p>
12+
* <p>
13+
* The default update interval is set to 30 minutes.
14+
* Change it typically to a value according to cache ttl and tradeoff with not updating it too much for
15+
* corner cases.
16+
* </p>
17+
*/
18+
@Builder
19+
@Getter
20+
public class PayloadCacheOptions {
21+
22+
@Builder.Default
23+
private int updateIntervalSeconds = 60 * 30; // 30 minutes
24+
}

‎providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/PayloadCacheWrapper.java‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
package dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.sync;
2+
3+
import lombok.Builder;
4+
import lombok.Getter;
5+
import lombok.extern.slf4j.Slf4j;
6+
7+
/**
8+
* A wrapper class for managing a payload cache with a specified update interval.
9+
* This class ensures that the cache is only updated if the specified time interval
10+
* has passed since the last update. It logs debug messages when updates are skipped
11+
* and error messages if the update process fails.
12+
* Not thread-safe.
13+
*
14+
* <p>Usage involves creating an instance with {@link PayloadCacheOptions} to set
15+
* the update interval, and then using {@link #updatePayloadIfNeeded(String)} to
16+
* conditionally update the cache and {@link #get()} to retrieve the cached payload.</p>
17+
*/
18+
@Slf4j
19+
public class PayloadCacheWrapper {
20+
private long lastUpdateTimeMs;
21+
private long updateIntervalMs;
22+
private PayloadCache payloadCache;
23+
24+
@Builder
25+
public PayloadCacheWrapper(PayloadCache payloadCache, PayloadCacheOptions payloadCacheOptions) {
26+
if (payloadCacheOptions.getUpdateIntervalSeconds() < 500) {
27+
throw new IllegalArgumentException("pollIntervalSeconds must be larger than 500");
28+
}
29+
this.updateIntervalMs = payloadCacheOptions.getUpdateIntervalSeconds() * 1000;
30+
this.payloadCache = payloadCache;
31+
}
32+
33+
public void updatePayloadIfNeeded(String payload) {
34+
if ((System.currentTimeMillis() - lastUpdateTimeMs) < updateIntervalMs) {
35+
log.debug("not updating payload, updateIntervalMs not reached");
36+
return;
37+
}
38+
39+
try {
40+
log.debug("updating payload");
41+
payloadCache.put(payload);
42+
lastUpdateTimeMs = System.currentTimeMillis();
43+
} catch (Exception e) {
44+
log.error("failed updating cache", e);
45+
}
46+
}
47+
48+
public String get() {
49+
try {
50+
return payloadCache.get();
51+
} catch (Exception e) {
52+
log.error("failed getting from cache", e);
53+
return null;
54+
}
55+
}
56+
}

‎providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/HttpConnectorTest.java‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -272,6 +272,49 @@ void testSuccessfulHttpResponseAddsDataToQueue() {
272272
assertEquals("test data", payload.getFlagData());
273273
}
274274

275+
@SneakyThrows
276+
@Test
277+
void testInitFailureUsingCache() {
278+
String testUrl = "http://example.com";
279+
HttpClient mockClient = mock(HttpClient.class);
280+
HttpResponse<String> mockResponse = mock(HttpResponse.class);
281+
when(mockResponse.statusCode()).thenReturn(200);
282+
when(mockClient.send(any(HttpRequest.class), any(HttpResponse.BodyHandler.class)))
283+
.thenThrow(new IOException("Simulated IO Exception"));
284+
285+
final String cachedData = "cached data";
286+
PayloadCache payloadCache = new PayloadCache() {
287+
@Override
288+
public void put(String payload) {
289+
// do nothing
290+
}
291+
292+
@Override
293+
public String get() {
294+
return cachedData;
295+
}
296+
};
297+
298+
HttpConnector connector = HttpConnector.builder()
299+
.url(testUrl)
300+
.httpClientExecutor(Executors.newSingleThreadExecutor())
301+
.payloadCache(payloadCache)
302+
.payloadCacheOptions(PayloadCacheOptions.builder().build())
303+
.build();
304+
305+
Field clientField = HttpConnector.class.getDeclaredField("client");
306+
clientField.setAccessible(true);
307+
clientField.set(connector, mockClient);
308+
309+
BlockingQueue<QueuePayload> queue = connector.getStreamQueue();
310+
311+
assertFalse(queue.isEmpty());
312+
QueuePayload payload = queue.poll();
313+
assertNotNull(payload);
314+
assertEquals(QueuePayloadType.DATA, payload.getType());
315+
assertEquals(cachedData, payload.getFlagData());
316+
}
317+
275318
@SneakyThrows
276319
@Test
277320
void testQueueBecomesFull() {

0 commit comments

Comments
 (0)