add polling cache support

Signed-off-by: liran2000 <[email protected]>

Author: liran2000

tools/flagd-http-connector/README.md

@@ -15,6 +15,9 @@ The implementation is using Java HttpClient.
 
 ## Use cases and benefits
 * Reduce infrastructure/devops work, without additional containers needed.
+* Reduce latency, since the data is fetched in-process.
+* Reduce external network traffic from the Http source even without a flagd separate container / proxy when 
+  polling cache is used.
 * Use as an additional provider for fallback / internal backup service via multi-provider.
 
 ### What happens if the Http source is down when application is starting ?
@@ -24,25 +27,43 @@ source downtime window, initial payload  is taken from cache to avoid starting w
 the source is back up. Therefore, the cache ttl expected to be higher than the expected source
 down-time to recover from during initialization.
 
+### Polling cache
+The polling cache is used to store the payload fetched from the URL.  
+Used when usePollingCache is configured as true.
+A key advantage of this cache is that it enables a single microservice within a cluster to handle the polling of a 
+URL, effectively acting as a flagd/proxy while all other services leverage the shared cache. 
+This approach optimizes resource usage by preventing redundant polling across services.
+
 ### Sample flow
 Sample flow can use:
 - Github as the flags payload source.
-- Redis cache as a fail-safe initialization cache.
+- Redis cache as a fail-safe initialization cache and as a polling cache.
 
 Sample flow of initialization during Github down-time window, showing that application can still use flags
 values as fetched from cache.
 ```mermaid
 sequenceDiagram
-    participant Provider
-    participant Github
+    box Cluster
+    participant micro-service-1
+    participant micro-service-2
+    participant micro-service-3
     participant Redis
+    end
+    participant Github
     
     break source downtime
-        Provider->>Github: initialize
-        Github->>Provider: failure
+    micro-service-1->>Github: initialize
+    Github->>micro-service-1: failure
     end
-    Provider->>Redis: fetch
-    Redis->>Provider: last payload
+    micro-service-1->>Redis: fetch
+    Redis->>micro-service-1: failsafe payload
+    Note right of micro-service-1: polling interval passed
+    micro-service-1->>Github: fetch
+    Github->>micro-service-1: payload
+    micro-service-2->>Redis: fetch
+    Redis->>micro-service-2: payload
+    micro-service-3->>Redis: fetch
+    Redis->>micro-service-3: payload
 
 ```
 
@@ -97,7 +118,6 @@ The Http Connector can be configured using the following properties in the `Http
 | payloadCacheOptions                       | PayloadCacheOptions | Options for configuring the payload cache. Default is null.                                                                                                                                                                                                                                                                                                                         |
 | payloadCache                              | PayloadCache        | The payload cache to use for caching responses. Default is null.                                                                                                                                                                                                                                                                                                                    |
 | useHttpCache                              | Boolean             | Whether to use HTTP caching for the requests. Default is false.                                                                                                                                                                                                                                                                                                                     |
+| useFailsafeCache                          | Boolean             | Whether to use a failsafe cache for initialization. Default is false.                                                                                                                                                                                                                                                                                                               |
+| usePollingCache                           | Boolean             | Whether to use a polling cache for initialization. Default is false.                                                                                                                                                                                                                                                                                                                |
 | PayloadCacheOptions.updateIntervalSeconds | Integer             | The interval, in seconds, at which the cache is updated. By default, this is set to 30 minutes. The goal is to avoid overloading fallback cache writes, since the cache serves only as a fallback mechanism. Typically, this value can be tuned to be shorter than the cache's TTL, balancing the need to minimize unnecessary updates while still handling edge cases effectively. |
-
-
-

tools/flagd-http-connector/src/main/java/dev/openfeature/contrib/tools/flagd/resolver/process/storage/connector/sync/http/PayloadCacheWrapper.java renamed to tools/flagd-http-connector/src/main/java/dev/openfeature/contrib/tools/flagd/resolver/process/storage/connector/sync/http/FailSafeCache.java

@@ -15,19 +15,21 @@
  * conditionally update the cache and {@link #get()} to retrieve the cached payload.</p>
  */
 @Slf4j
-public class PayloadCacheWrapper {
+public class FailSafeCache {
+    public static final String FAILSAFE_PAYLOAD_CACHE_KEY = FailSafeCache.class.getSimpleName()
+        + ".failsafe-payload";
     private long lastUpdateTimeMs;
     private long updateIntervalMs;
     private PayloadCache payloadCache;
 
     /**
-     * Constructor for PayloadCacheWrapper.
+     * Constructor for FailSafeCache.
      *
      * @param payloadCache the payload cache to be used
      * @param payloadCacheOptions the options for configuring the cache
      */
     @Builder
-    public PayloadCacheWrapper(PayloadCache payloadCache, PayloadCacheOptions payloadCacheOptions) {
+    public FailSafeCache(PayloadCache payloadCache, PayloadCacheOptions payloadCacheOptions) {
         if (payloadCacheOptions.getUpdateIntervalSeconds() < 1) {
             throw new IllegalArgumentException("pollIntervalSeconds must be larger than 0");
         }
@@ -48,7 +50,7 @@ public void updatePayloadIfNeeded(String payload) {
 
         try {
             log.debug("updating payload");
-            payloadCache.put(payload);
+            payloadCache.put(FAILSAFE_PAYLOAD_CACHE_KEY, payload);
             lastUpdateTimeMs = getCurrentTimeMillis();
         } catch (Exception e) {
             log.error("failed updating cache", e);
@@ -66,7 +68,7 @@ protected long getCurrentTimeMillis() {
      */
     public String get() {
         try {
-            return payloadCache.get();
+            return payloadCache.get(FAILSAFE_PAYLOAD_CACHE_KEY);
         } catch (Exception e) {
             log.error("failed getting from cache", e);
             return null;

tools/flagd-http-connector/src/main/java/dev/openfeature/contrib/tools/flagd/resolver/process/storage/connector/sync/http/HttpConnector.java

@@ -38,16 +38,19 @@
 @Slf4j
 public class HttpConnector implements QueueSource {
 
+    public static final String POLLING_PAYLOAD_CACHE_KEY = HttpConnector.class.getSimpleName() + ".polling-payload";
     private Integer pollIntervalSeconds;
     private Integer requestTimeoutSeconds;
     private BlockingQueue<QueuePayload> queue;
     private HttpClient client;
     private ExecutorService httpClientExecutor;
     private ScheduledExecutorService scheduler;
     private Map<String, String> headers;
-    private PayloadCacheWrapper payloadCacheWrapper;
+    private FailSafeCache failSafeCache;
     private PayloadCache payloadCache;
     private HttpCacheFetcher httpCacheFetcher;
+    private int payloadCachePollTtlSeconds;
+    private boolean usePollingCache;
 
     @NonNull
     private String url;
@@ -77,15 +80,17 @@ public HttpConnector(HttpConnectorOptions httpConnectorOptions) {
             .build();
         this.queue = new LinkedBlockingQueue<>(httpConnectorOptions.getLinkedBlockingQueueCapacity());
         this.payloadCache = httpConnectorOptions.getPayloadCache();
-        if (payloadCache != null) {
-            this.payloadCacheWrapper = PayloadCacheWrapper.builder()
+        if (payloadCache != null && Boolean.TRUE.equals(httpConnectorOptions.getUseFailsafeCache())) {
+            this.failSafeCache = FailSafeCache.builder()
                 .payloadCache(payloadCache)
                 .payloadCacheOptions(httpConnectorOptions.getPayloadCacheOptions())
                 .build();
         }
         if (Boolean.TRUE.equals(httpConnectorOptions.getUseHttpCache())) {
             httpCacheFetcher = new HttpCacheFetcher();
         }
+        payloadCachePollTtlSeconds = pollIntervalSeconds; // safety margin
+        this.usePollingCache = Boolean.TRUE.equals(httpConnectorOptions.getUsePollingCache());
     }
 
     @Override
@@ -98,18 +103,18 @@ public BlockingQueue<QueuePayload> getStreamQueue() {
         boolean success = fetchAndUpdate();
         if (!success) {
             log.info("failed initial fetch");
-            if (payloadCache != null) {
-                updateFromCache();
+            if (failSafeCache != null) {
+                updateFromFailsafeCache();
             }
         }
         Runnable pollTask = buildPollTask();
         scheduler.scheduleWithFixedDelay(pollTask, pollIntervalSeconds, pollIntervalSeconds, TimeUnit.SECONDS);
         return queue;
     }
 
-    private void updateFromCache() {
+    private void updateFromFailsafeCache() {
         log.info("taking initial payload from cache to avoid starting with default values");
-        String flagData = payloadCache.get();
+        String flagData = failSafeCache.get();
         if (flagData == null) {
             log.debug("got null from cache");
             return;
@@ -124,6 +129,14 @@ protected Runnable buildPollTask() {
     }
 
     private boolean fetchAndUpdate() {
+        if (payloadCache != null && usePollingCache) {
+            log.debug("checking cache for polling payload");
+            String payload = payloadCache.get(POLLING_PAYLOAD_CACHE_KEY);
+            if (payload != null) {
+                log.debug("got payload from polling cache key, skipping update");
+                return false;
+            }
+        }
         HttpRequest.Builder requestBuilder = HttpRequest.newBuilder()
             .uri(URI.create(url))
             .timeout(Duration.ofSeconds(requestTimeoutSeconds))
@@ -159,10 +172,18 @@ private boolean fetchAndUpdate() {
             log.warn("Unable to offer file content to queue: queue is full");
             return false;
         }
-        if (payloadCacheWrapper != null) {
+        if (payloadCache != null) {
             log.debug("scheduling cache update if needed");
-            scheduler.execute(() ->
-                payloadCacheWrapper.updatePayloadIfNeeded(payload)
+            scheduler.execute(() -> {
+                    if (failSafeCache != null) {
+                        log.debug("updating payload in failsafe cache if needed");
+                        failSafeCache.updatePayloadIfNeeded(payload);
+                    }
+                    if (payloadCache != null) {
+                        log.debug("updating polling payload in cache");
+                        payloadCache.put(POLLING_PAYLOAD_CACHE_KEY, payload, payloadCachePollTtlSeconds);
+                    }
+                }
             );
         }
         return true;

tools/flagd-http-connector/src/main/java/dev/openfeature/contrib/tools/flagd/resolver/process/storage/connector/sync/http/HttpConnectorOptions.java

@@ -40,6 +40,10 @@ public class HttpConnectorOptions {
     private PayloadCache payloadCache;
     @Builder.Default
     private Boolean useHttpCache;
+    @Builder.Default
+    private Boolean useFailsafeCache;
+    @Builder.Default
+    private Boolean usePollingCache;
     @NonNull
     private String url;
 
@@ -50,9 +54,11 @@ public class HttpConnectorOptions {
     public HttpConnectorOptions(Integer pollIntervalSeconds, Integer linkedBlockingQueueCapacity,
             Integer scheduledThreadPoolSize, Integer requestTimeoutSeconds, Integer connectTimeoutSeconds, String url,
             Map<String, String> headers, ExecutorService httpClientExecutor, String proxyHost, Integer proxyPort,
-            PayloadCacheOptions payloadCacheOptions, PayloadCache payloadCache, Boolean useHttpCache) {
+            PayloadCacheOptions payloadCacheOptions, PayloadCache payloadCache, Boolean useHttpCache,
+            Boolean useFailsafeCache, Boolean usePollingCache) {
         validate(url, pollIntervalSeconds, linkedBlockingQueueCapacity, scheduledThreadPoolSize, requestTimeoutSeconds,
-                connectTimeoutSeconds, proxyHost, proxyPort, payloadCacheOptions, payloadCache);
+            connectTimeoutSeconds, proxyHost, proxyPort, payloadCacheOptions, payloadCache, useFailsafeCache,
+            usePollingCache);
         if (pollIntervalSeconds != null) {
             this.pollIntervalSeconds = pollIntervalSeconds;
         }
@@ -90,13 +96,19 @@ public HttpConnectorOptions(Integer pollIntervalSeconds, Integer linkedBlockingQ
         if (useHttpCache != null) {
             this.useHttpCache = useHttpCache;
         }
+        if (useFailsafeCache != null) {
+            this.useFailsafeCache = useFailsafeCache;
+        }
+        if (usePollingCache != null) {
+            this.usePollingCache = usePollingCache;
+        }
     }
 
     @SneakyThrows
     private void validate(String url, Integer pollIntervalSeconds, Integer linkedBlockingQueueCapacity,
             Integer scheduledThreadPoolSize, Integer requestTimeoutSeconds, Integer connectTimeoutSeconds,
             String proxyHost, Integer proxyPort, PayloadCacheOptions payloadCacheOptions,
-            PayloadCache payloadCache) {
+            PayloadCache payloadCache, Boolean useFailsafeCache, Boolean usePollingCache) {
         new URL(url).toURI();
         if (linkedBlockingQueueCapacity != null
                 && (linkedBlockingQueueCapacity < 1 || linkedBlockingQueueCapacity > 1000)) {
@@ -128,5 +140,26 @@ private void validate(String url, Integer pollIntervalSeconds, Integer linkedBlo
         if (payloadCache != null && payloadCacheOptions == null) {
             throw new IllegalArgumentException("payloadCacheOptions must be set if payloadCache is set");
         }
+        if ((Boolean.TRUE.equals(useFailsafeCache) || Boolean.TRUE.equals(usePollingCache)) && payloadCache == null) {
+            throw new IllegalArgumentException(
+                "payloadCache must be set if useFailsafeCache or usePollingCache is set");
+        }
+
+        if (payloadCache != null && Boolean.TRUE.equals(usePollingCache)) {
+
+            // verify payloadCache overrides put(String key, String payload, int ttlSeconds)
+            boolean overridesTtlPutMethod = false;
+            try {
+                var method = payloadCache.getClass().getMethod("put", String.class, String.class, int.class);
+                // Check if the method is declared in the class and not inherited
+                overridesTtlPutMethod = method.getDeclaringClass() != PayloadCache.class;
+            } catch (NoSuchMethodException e) {
+                // Method does not exist
+            }
+            if (!overridesTtlPutMethod) {
+                throw new IllegalArgumentException("when usePollingCache is used, payloadCache must override "
+                    + "put(String key, String payload, int ttlSeconds)");
+            }
+        }
     }
 }

tools/flagd-http-connector/src/main/java/dev/openfeature/contrib/tools/flagd/resolver/process/storage/connector/sync/http/PayloadCache.java

@@ -5,7 +5,18 @@
  */
 public interface PayloadCache {
 
-    void put(String payload);
+    void put(String key, String payload);
 
-    String get();
+    String get(String key);
+
+    /**
+     * Put a payload into the cache with a time-to-live (TTL) value.
+     * Must implement if HttpConnectorOptions.usePollingCache is true.
+     * @param key
+     * @param payload
+     * @param ttlSeconds
+     */
+    default void put(String key, String payload, int ttlSeconds) {
+        throw new UnsupportedOperationException("put with ttl not supported");
+    }
 }