feat(flagd): Support supplying providerId for in-process resolver as an option (#1259)

Signed-off-by: Maks Osowski <[email protected]>
Co-authored-by: Michael Beemer <[email protected]>
Co-authored-by: Todd Baert <[email protected]>

Author: 3 people

providers/flagd/README.md

@@ -117,6 +117,7 @@ Given below are the supported configurations:
 | streamDeadlineMs      | FLAGD_STREAM_DEADLINE_MS       | int                      | 600000    | rpc & in-process        |
 | keepAliveTime         | FLAGD_KEEP_ALIVE_TIME_MS       | long                     | 0         | rpc & in-process        |
 | selector              | FLAGD_SOURCE_SELECTOR          | String                   | null      | in-process              |
+| providerId            | FLAGD_SOURCE_PROVIDER_ID       | String                   | null      | in-process              |
 | cache                 | FLAGD_CACHE                    | String - lru, disabled   | lru       | rpc                     |
 | maxCacheSize          | FLAGD_MAX_CACHE_SIZE           | int                      | 1000      | rpc                     |
 | maxEventStreamRetries | FLAGD_MAX_EVENT_STREAM_RETRIES | int                      | 5         | rpc                     |

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/Config.java

@@ -34,6 +34,7 @@ public final class Config {
     static final String DEADLINE_MS_ENV_VAR_NAME = "FLAGD_DEADLINE_MS";
     static final String STREAM_DEADLINE_MS_ENV_VAR_NAME = "FLAGD_STREAM_DEADLINE_MS";
     static final String SOURCE_SELECTOR_ENV_VAR_NAME = "FLAGD_SOURCE_SELECTOR";
+    static final String SOURCE_PROVIDER_ID_ENV_VAR_NAME = "FLAGD_SOURCE_PROVIDER_ID";
     static final String OFFLINE_SOURCE_PATH = "FLAGD_OFFLINE_FLAG_SOURCE_PATH";
     static final String OFFLINE_POLL_MS = "FLAGD_OFFLINE_POLL_MS";
     static final String KEEP_ALIVE_MS_ENV_VAR_NAME_OLD = "FLAGD_KEEP_ALIVE_TIME";

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/FlagdOptions.java

@@ -109,6 +109,12 @@ public class FlagdOptions {
     @Builder.Default
     private String selector = fallBackToEnvOrDefault(Config.SOURCE_SELECTOR_ENV_VAR_NAME, null);
 
+    /**
+     * ProviderId to be used with flag sync gRPC contract.
+     **/
+    @Builder.Default
+    private String providerId = fallBackToEnvOrDefault(Config.SOURCE_PROVIDER_ID_ENV_VAR_NAME, null);
+
     /**
      * gRPC client KeepAlive in milliseconds. Disabled with 0.
      * Defaults to 0 (disabled).

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/grpc/GrpcStreamConnector.java

@@ -34,6 +34,7 @@ public class GrpcStreamConnector implements Connector {
     private final BlockingQueue<QueuePayload> blockingQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);
     private final int deadline;
     private final String selector;
+    private final String providerId;
     private final GrpcConnector<
                     FlagSyncServiceGrpc.FlagSyncServiceStub, FlagSyncServiceGrpc.FlagSyncServiceBlockingStub>
             grpcConnector;
@@ -45,6 +46,7 @@ public class GrpcStreamConnector implements Connector {
     public GrpcStreamConnector(final FlagdOptions options, Consumer<FlagdProviderEvent> onConnectionEvent) {
         deadline = options.getDeadline();
         selector = options.getSelector();
+        providerId = options.getProviderId();
         streamReceiver = new LinkedBlockingQueue<>(QUEUE_SIZE);
         grpcConnector = new GrpcConnector<>(
                 options,
@@ -53,11 +55,15 @@ public GrpcStreamConnector(final FlagdOptions options, Consumer<FlagdProviderEve
                 onConnectionEvent,
                 stub -> {
                     String localSelector = selector;
+                    String localProviderId = providerId;
 
                     final SyncFlagsRequest.Builder syncRequest = SyncFlagsRequest.newBuilder();
                     if (localSelector != null) {
                         syncRequest.setSelector(localSelector);
                     }
+                    if (localProviderId != null) {
+                        syncRequest.setProviderId(localProviderId);
+                    }
 
                     stub.syncFlags(syncRequest.build(), new GrpcStreamHandler(streamReceiver));
                 });

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/FlagdOptionsTest.java

@@ -10,14 +10,14 @@
 import static dev.openfeature.contrib.providers.flagd.Config.RESOLVER_ENV_VAR;
 import static dev.openfeature.contrib.providers.flagd.Config.RESOLVER_IN_PROCESS;
 import static dev.openfeature.contrib.providers.flagd.Config.RESOLVER_RPC;
-import static dev.openfeature.contrib.providers.flagd.Config.Resolver;
 import static dev.openfeature.contrib.providers.flagd.Config.TARGET_URI_ENV_VAR_NAME;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import dev.openfeature.contrib.providers.flagd.Config.Resolver;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.MockConnector;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.Connector;
 import io.grpc.ClientInterceptor;
@@ -44,6 +44,7 @@ void TestDefaults() {
         assertEquals(DEFAULT_CACHE, builder.getCacheType());
         assertEquals(DEFAULT_MAX_CACHE_SIZE, builder.getMaxCacheSize());
         assertNull(builder.getSelector());
+        assertNull(builder.getProviderId());
         assertNull(builder.getOpenTelemetry());
         assertNull(builder.getCustomConnector());
         assertNull(builder.getOfflineFlagSourcePath());
@@ -67,6 +68,7 @@ void TestBuilderOptions() {
                 .cacheType("lru")
                 .maxCacheSize(100)
                 .selector("app=weatherApp")
+                .providerId("test/provider/id_1")
                 .openTelemetry(openTelemetry)
                 .customConnector(connector)
                 .resolverType(Resolver.IN_PROCESS)
@@ -83,6 +85,7 @@ void TestBuilderOptions() {
         assertEquals("lru", flagdOptions.getCacheType());
         assertEquals(100, flagdOptions.getMaxCacheSize());
         assertEquals("app=weatherApp", flagdOptions.getSelector());
+        assertEquals("test/provider/id_1", flagdOptions.getProviderId());
         assertEquals(openTelemetry, flagdOptions.getOpenTelemetry());
         assertEquals(connector, flagdOptions.getCustomConnector());
         assertEquals(Resolver.IN_PROCESS, flagdOptions.getResolverType());