diff --git a/src/main/java/dev/openfeature/sdk/EventProvider.java b/src/main/java/dev/openfeature/sdk/EventProvider.java index beb2e1ce3..95b8accd2 100644 --- a/src/main/java/dev/openfeature/sdk/EventProvider.java +++ b/src/main/java/dev/openfeature/sdk/EventProvider.java @@ -3,10 +3,13 @@ import dev.openfeature.sdk.internal.AutoCloseableReentrantReadWriteLock; import dev.openfeature.sdk.internal.ConfigurableThreadFactory; import dev.openfeature.sdk.internal.TriConsumer; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiConsumer; import lombok.extern.slf4j.Slf4j; /** @@ -24,6 +27,7 @@ @Slf4j public abstract class EventProvider implements FeatureProvider { private EventProviderListener eventProviderListener; + private final List> eventObservers = new CopyOnWriteArrayList<>(); private final ExecutorService emitterExecutor = Executors.newCachedThreadPool(new ConfigurableThreadFactory("openfeature-event-emitter-thread", true)); @@ -70,6 +74,31 @@ void detach() { this.attachment.set(null); } + /** + * Add a provider event observer. + * + *

Observers are invoked whenever this provider emits an event and are intended for advanced + * provider composition scenarios. + * + * @param observer observer callback + */ + public void addEventObserver(BiConsumer observer) { + if (observer != null) { + eventObservers.add(observer); + } + } + + /** + * Remove a previously registered provider event observer. + * + * @param observer observer callback + */ + public void removeEventObserver(BiConsumer observer) { + if (observer != null) { + eventObservers.remove(observer); + } + } + /** * Stop the event emitter executor and block until either termination has completed * or timeout period has elapsed. @@ -97,8 +126,9 @@ public void shutdown() { public Awaitable emit(final ProviderEvent event, final ProviderEventDetails details) { final var localEventProviderListener = this.eventProviderListener; final var localAttachment = this.attachment.get(); + final var localEventObservers = this.eventObservers; - if (localEventProviderListener == null && localAttachment == null) { + if (localEventProviderListener == null && localAttachment == null && localEventObservers.isEmpty()) { return Awaitable.FINISHED; } @@ -116,6 +146,13 @@ public Awaitable emit(final ProviderEvent event, final ProviderEventDetails deta if (localAttachment != null) { localAttachment.onEmit.accept(this, event, details); } + for (BiConsumer observer : localEventObservers) { + try { + observer.accept(event, details); + } catch (Exception e) { + log.error("Exception in provider event observer {}", observer, e); + } + } } finally { awaitable.wakeup(); } diff --git a/src/main/java/dev/openfeature/sdk/multiprovider/MultiProvider.java b/src/main/java/dev/openfeature/sdk/multiprovider/MultiProvider.java index 00611ed78..3eafeea58 100644 --- a/src/main/java/dev/openfeature/sdk/multiprovider/MultiProvider.java +++ b/src/main/java/dev/openfeature/sdk/multiprovider/MultiProvider.java @@ -1,11 +1,17 @@ package dev.openfeature.sdk.multiprovider; +import dev.openfeature.sdk.ErrorCode; import dev.openfeature.sdk.EvaluationContext; import dev.openfeature.sdk.EventProvider; import dev.openfeature.sdk.FeatureProvider; import dev.openfeature.sdk.Metadata; import dev.openfeature.sdk.ProviderEvaluation; +import dev.openfeature.sdk.ProviderEvent; +import dev.openfeature.sdk.ProviderEventDetails; +import dev.openfeature.sdk.ProviderState; +import dev.openfeature.sdk.TrackingEventDetails; import dev.openfeature.sdk.Value; +import dev.openfeature.sdk.exceptions.OpenFeatureError; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.ArrayList; import java.util.Collection; @@ -16,9 +22,11 @@ import java.util.Map; import java.util.Objects; import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.function.BiConsumer; import lombok.Getter; import lombok.extern.slf4j.Slf4j; @@ -28,6 +36,9 @@ *

This provider delegates flag evaluations to multiple underlying providers using a configurable * {@link Strategy}. It also exposes combined metadata containing the original metadata of each * underlying provider. + * + *

Child provider events are observed and aggregated into a single provider state using a + * "worst wins" precedence, and {@code track} calls are forwarded to every usable child provider. */ @Slf4j public class MultiProvider extends EventProvider { @@ -40,6 +51,10 @@ public class MultiProvider extends EventProvider { private final Map providers; private final Strategy strategy; + private final Map providerStates = new ConcurrentHashMap<>(); + private final Map> providerEventObservers = + new ConcurrentHashMap<>(); + private ProviderState aggregateState; private MultiProviderMetadata metadata; /** @@ -61,20 +76,67 @@ public MultiProvider(List providers) { public MultiProvider(List providers, Strategy strategy) { this.providers = buildProviders(providers); this.strategy = Objects.requireNonNull(strategy, "strategy must not be null"); + initializeProviderStates(); + this.aggregateState = determineAggregateState(); } + /** + * Builds the internal provider map, deriving a unique name for each provider. + * + *

Providers that share a metadata name are disambiguated with a numeric suffix + * ({@code name-1}, {@code name-2}, ...) so that no provider is silently dropped. + */ protected static Map buildProviders(List providers) { + Objects.requireNonNull(providers, "providers must not be null"); Map providersMap = new LinkedHashMap<>(providers.size()); + Map suffixesByBaseName = new HashMap<>(providers.size()); for (FeatureProvider provider : providers) { - FeatureProvider prevProvider = - providersMap.put(provider.getMetadata().getName(), provider); - if (prevProvider != null) { - log.info("duplicated provider name: {}", provider.getMetadata().getName()); + Objects.requireNonNull(provider, "provider must not be null"); + String baseName = getProviderBaseName(provider); + String resolvedName = resolveUniqueProviderName(baseName, providersMap, suffixesByBaseName); + if (!baseName.equals(resolvedName)) { + log.info("deduplicated provider name from {} to {}", baseName, resolvedName); } + providersMap.put(resolvedName, provider); } return Collections.unmodifiableMap(providersMap); } + private static String getProviderBaseName(FeatureProvider provider) { + Metadata providerMetadata = provider.getMetadata(); + if (providerMetadata == null + || providerMetadata.getName() == null + || providerMetadata.getName().isEmpty()) { + return "provider"; + } + return providerMetadata.getName(); + } + + private static String resolveUniqueProviderName( + String baseName, Map providersMap, Map suffixesByBaseName) { + if (!providersMap.containsKey(baseName)) { + suffixesByBaseName.putIfAbsent(baseName, 1); + return baseName; + } + int suffix = suffixesByBaseName.getOrDefault(baseName, 1); + String resolvedName = baseName + "-" + suffix; + while (providersMap.containsKey(resolvedName)) { + suffix++; + resolvedName = baseName + "-" + suffix; + } + suffixesByBaseName.put(baseName, suffix + 1); + return resolvedName; + } + + private void initializeProviderStates() { + providerStates.clear(); + if (!providers.isEmpty()) { + for (String providerName : providers.keySet()) { + providerStates.put(providerName, ProviderState.NOT_READY); + } + } + } + /** * {@inheritDoc} * @@ -95,7 +157,12 @@ public void initialize(EvaluationContext evaluationContext) throws Exception { @Override public void initialize(EvaluationContext evaluationContext, String domain) throws Exception { var metadataBuilder = MultiProviderMetadata.builder().name(NAME); - HashMap providersMetadata = new HashMap<>(); + Map providersMetadata = new LinkedHashMap<>(); + initializeProviderStates(); + synchronized (this) { + emitAggregateStateChange( + determineAggregateState(), ProviderEventDetails.builder().build()); + } if (providers.isEmpty()) { metadataBuilder.originalMetadata(Collections.unmodifiableMap(providersMetadata)); @@ -106,13 +173,22 @@ public void initialize(EvaluationContext evaluationContext, String domain) throw ExecutorService executorService = Executors.newFixedThreadPool(Math.min(INIT_THREADS_COUNT, providers.size())); try { Collection> tasks = new ArrayList<>(providers.size()); - for (FeatureProvider provider : providers.values()) { + for (Map.Entry entry : providers.entrySet()) { + String providerName = entry.getKey(); + FeatureProvider provider = entry.getValue(); + registerChildProviderObserver(providerName, provider); tasks.add(() -> { - provider.initialize(evaluationContext, domain); - return null; + try { + provider.initialize(evaluationContext, domain); + setProviderReadyIfStillNotReady(providerName); + return null; + } catch (Exception e) { + setProviderState(providerName, toStateFromException(e), providerErrorDetails(e)); + throw e; + } }); Metadata providerMetadata = provider.getMetadata(); - providersMetadata.put(providerMetadata.getName(), providerMetadata); + providersMetadata.put(providerName, providerMetadata); } metadataBuilder.originalMetadata(Collections.unmodifiableMap(providersMetadata)); @@ -172,18 +248,213 @@ public ProviderEvaluation getObjectEvaluation(String key, Value defaultVa return strategy.evaluate(providers, key, defaultValue, ctx, p -> p.getObjectEvaluation(key, defaultValue, ctx)); } + /** + * Forwards the tracking event to every child provider that is in a usable state. + * + *

Providers that are {@code NOT_READY} or {@code FATAL} are skipped. Errors raised by an + * individual provider are logged and do not prevent the remaining providers from being called. + */ + @Override + public void track(String eventName, EvaluationContext context, TrackingEventDetails details) { + for (Map.Entry entry : providers.entrySet()) { + String providerName = entry.getKey(); + FeatureProvider provider = entry.getValue(); + if (!shouldTrackProvider(providerName)) { + continue; + } + try { + provider.track(eventName, context, details); + } catch (Exception e) { + log.error("error forwarding track to provider {}", providerName, e); + } + } + } + @Override public void shutdown() { log.debug("shutdown begin"); - for (FeatureProvider provider : providers.values()) { + for (Map.Entry entry : providers.entrySet()) { + String providerName = entry.getKey(); + FeatureProvider provider = entry.getValue(); try { + unregisterChildProviderObserver(providerName, provider); provider.shutdown(); } catch (Exception e) { - log.error("error shutdown provider {}", provider.getMetadata().getName(), e); + log.error("error shutdown provider {}", providerName, e); } } + synchronized (this) { + initializeProviderStates(); + emitAggregateStateChange( + ProviderState.NOT_READY, ProviderEventDetails.builder().build()); + } log.debug("shutdown end"); // Important: ensure EventProvider's executor is also shut down super.shutdown(); } + + private void registerChildProviderObserver(String providerName, FeatureProvider provider) { + if (provider instanceof EventProvider) { + BiConsumer observer = + (event, details) -> onChildProviderEvent(providerName, event, details); + ((EventProvider) provider).addEventObserver(observer); + providerEventObservers.put(providerName, observer); + } + } + + private void unregisterChildProviderObserver(String providerName, FeatureProvider provider) { + if (provider instanceof EventProvider) { + BiConsumer observer = providerEventObservers.remove(providerName); + if (observer != null) { + ((EventProvider) provider).removeEventObserver(observer); + } + } + } + + private void onChildProviderEvent(String providerName, ProviderEvent event, ProviderEventDetails details) { + if (ProviderEvent.PROVIDER_CONFIGURATION_CHANGED.equals(event)) { + emitProviderConfigurationChanged(details); + return; + } + ProviderState state = toStateFromEvent(event, details); + if (state != null) { + setProviderState(providerName, state, details); + } + } + + private synchronized void setProviderState( + String providerName, ProviderState providerState, ProviderEventDetails details) { + providerStates.put(providerName, providerState); + ProviderState aggregate = determineAggregateState(); + emitAggregateStateChange(aggregate, details); + } + + private synchronized void setProviderReadyIfStillNotReady(String providerName) { + if (!ProviderState.NOT_READY.equals(providerStates.get(providerName))) { + return; + } + providerStates.put(providerName, ProviderState.READY); + ProviderState aggregate = determineAggregateState(); + emitAggregateStateChange(aggregate, ProviderEventDetails.builder().build()); + } + + private void emitAggregateStateChange(ProviderState aggregate, ProviderEventDetails details) { + ProviderState previous = aggregateState; + if (previous == aggregate) { + return; + } + aggregateState = aggregate; + switch (aggregate) { + case READY: + emitProviderReady(detailsOrEmpty(details)); + break; + case STALE: + emitProviderStale(detailsOrEmpty(details)); + break; + case ERROR: + emitProviderError(ensureErrorDetails(details, ErrorCode.GENERAL)); + break; + case FATAL: + emitProviderError(ensureErrorDetails(details, ErrorCode.PROVIDER_FATAL)); + break; + case NOT_READY: + break; + default: + break; + } + } + + /** Aggregates the child provider states using a "worst wins" precedence. */ + private ProviderState determineAggregateState() { + if (providerStates.isEmpty()) { + return ProviderState.READY; + } + ProviderState aggregate = ProviderState.READY; + for (ProviderState state : providerStates.values()) { + if (stateSeverity(state) > stateSeverity(aggregate)) { + aggregate = state; + } + } + return aggregate; + } + + private int stateSeverity(ProviderState state) { + if (state == null) { + return 0; + } + switch (state) { + case FATAL: + return 5; + case NOT_READY: + return 4; + case ERROR: + return 3; + case STALE: + return 2; + case READY: + return 1; + default: + return 0; + } + } + + private ProviderEventDetails detailsOrEmpty(ProviderEventDetails details) { + if (details == null) { + return ProviderEventDetails.builder().build(); + } + return details; + } + + private ProviderEventDetails ensureErrorDetails(ProviderEventDetails details, ErrorCode defaultErrorCode) { + if (details == null) { + return ProviderEventDetails.builder().errorCode(defaultErrorCode).build(); + } + if (details.getErrorCode() == null) { + return details.toBuilder().errorCode(defaultErrorCode).build(); + } + return details; + } + + private ProviderState toStateFromEvent(ProviderEvent event, ProviderEventDetails details) { + if (ProviderEvent.PROVIDER_READY.equals(event)) { + return ProviderState.READY; + } + if (ProviderEvent.PROVIDER_STALE.equals(event)) { + return ProviderState.STALE; + } + if (ProviderEvent.PROVIDER_ERROR.equals(event)) { + if (details != null && ErrorCode.PROVIDER_FATAL.equals(details.getErrorCode())) { + return ProviderState.FATAL; + } + return ProviderState.ERROR; + } + return null; + } + + private ProviderState toStateFromException(Exception exception) { + if (exception instanceof OpenFeatureError + && ErrorCode.PROVIDER_FATAL.equals(((OpenFeatureError) exception).getErrorCode())) { + return ProviderState.FATAL; + } + return ProviderState.ERROR; + } + + private ProviderEventDetails providerErrorDetails(Exception exception) { + if (exception instanceof OpenFeatureError) { + ErrorCode errorCode = ((OpenFeatureError) exception).getErrorCode(); + return ProviderEventDetails.builder() + .errorCode(errorCode) + .message(exception.getMessage()) + .build(); + } + return ProviderEventDetails.builder() + .errorCode(ErrorCode.GENERAL) + .message(exception.getMessage()) + .build(); + } + + private boolean shouldTrackProvider(String providerName) { + ProviderState providerState = providerStates.getOrDefault(providerName, ProviderState.READY); + return !ProviderState.NOT_READY.equals(providerState) && !ProviderState.FATAL.equals(providerState); + } } diff --git a/src/test/java/dev/openfeature/sdk/multiprovider/MultiProviderEventsAndTrackingTest.java b/src/test/java/dev/openfeature/sdk/multiprovider/MultiProviderEventsAndTrackingTest.java new file mode 100644 index 000000000..c01dd5b30 --- /dev/null +++ b/src/test/java/dev/openfeature/sdk/multiprovider/MultiProviderEventsAndTrackingTest.java @@ -0,0 +1,195 @@ +package dev.openfeature.sdk.multiprovider; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import dev.openfeature.sdk.Client; +import dev.openfeature.sdk.EvaluationContext; +import dev.openfeature.sdk.EventProvider; +import dev.openfeature.sdk.Metadata; +import dev.openfeature.sdk.OpenFeatureAPI; +import dev.openfeature.sdk.ProviderEvaluation; +import dev.openfeature.sdk.ProviderEvent; +import dev.openfeature.sdk.ProviderEventDetails; +import dev.openfeature.sdk.ProviderState; +import dev.openfeature.sdk.TrackingEventDetails; +import dev.openfeature.sdk.Value; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class MultiProviderEventsAndTrackingTest { + + @Test + void shouldAggregateChildProviderStateAndForwardConfigurationEvents() throws Exception { + TrackingProvider provider1 = new TrackingProvider("provider1"); + TrackingProvider provider2 = new TrackingProvider("provider2"); + MultiProvider multiProvider = new MultiProvider(List.of(provider1, provider2)); + + OpenFeatureAPI api = OpenFeatureAPI.createIsolated(); + try { + api.setProviderAndWait("multiProviderEvents", multiProvider); + Client client = api.getClient("multiProviderEvents"); + + await().atMost(Duration.ofSeconds(2)).until(() -> client.getProviderState() == ProviderState.READY); + + AtomicInteger configurationChangedCount = new AtomicInteger(); + client.onProviderConfigurationChanged(details -> configurationChangedCount.incrementAndGet()); + + provider1 + .emitProviderConfigurationChanged( + ProviderEventDetails.builder().message("changed").build()) + .await(); + await().atMost(Duration.ofSeconds(2)).until(() -> configurationChangedCount.get() == 1); + + provider1 + .emitProviderStale( + ProviderEventDetails.builder().message("stale").build()) + .await(); + await().atMost(Duration.ofSeconds(2)).until(() -> client.getProviderState() == ProviderState.STALE); + + provider2 + .emitProviderError(ProviderEventDetails.builder() + .errorCode(dev.openfeature.sdk.ErrorCode.GENERAL) + .build()) + .await(); + await().atMost(Duration.ofSeconds(2)).until(() -> client.getProviderState() == ProviderState.ERROR); + + provider2.emitProviderReady(ProviderEventDetails.builder().build()).await(); + await().atMost(Duration.ofSeconds(2)).until(() -> client.getProviderState() == ProviderState.STALE); + + provider1.emitProviderReady(ProviderEventDetails.builder().build()).await(); + await().atMost(Duration.ofSeconds(2)).until(() -> client.getProviderState() == ProviderState.READY); + + provider1 + .emitProviderError(ProviderEventDetails.builder() + .errorCode(dev.openfeature.sdk.ErrorCode.PROVIDER_FATAL) + .build()) + .await(); + await().atMost(Duration.ofSeconds(2)).until(() -> client.getProviderState() == ProviderState.FATAL); + } finally { + api.shutdown(); + } + } + + @Test + void shouldPreserveChildStateEmittedDuringInitialize() throws Exception { + TrackingProvider provider1 = new InitializingStateProvider("provider1", ProviderState.STALE); + TrackingProvider provider2 = new TrackingProvider("provider2"); + MultiProvider multiProvider = new MultiProvider(List.of(provider1, provider2)); + List emittedEvents = new CopyOnWriteArrayList<>(); + multiProvider.addEventObserver((event, details) -> emittedEvents.add(event)); + + multiProvider.initialize(null); + + await().atMost(Duration.ofSeconds(2)).until(() -> emittedEvents.contains(ProviderEvent.PROVIDER_STALE)); + assertEquals(List.of(ProviderEvent.PROVIDER_STALE), emittedEvents); + } + + @Test + void shouldForwardTrackToReadyProvidersAndSkipFatalProviders() throws Exception { + TrackingProvider provider1 = new TrackingProvider("provider1"); + TrackingProvider provider2 = new TrackingProvider("provider2"); + provider2.throwOnTrack = true; + + MultiProvider multiProvider = new MultiProvider(List.of(provider1, provider2)); + multiProvider.initialize(null); + + multiProvider.track("event1", null, null); + assertEquals(1, provider1.trackCount.get()); + assertEquals(1, provider2.trackCount.get()); + + provider1 + .emitProviderError(ProviderEventDetails.builder() + .errorCode(dev.openfeature.sdk.ErrorCode.PROVIDER_FATAL) + .build()) + .await(); + + multiProvider.track("event2", null, null); + assertEquals(1, provider1.trackCount.get()); + assertEquals(2, provider2.trackCount.get()); + } + + static class TrackingProvider extends EventProvider { + private final String name; + private final AtomicInteger trackCount = new AtomicInteger(); + private boolean throwOnTrack; + + TrackingProvider(String name) { + this.name = name; + } + + @Override + public Metadata getMetadata() { + return () -> name; + } + + @Override + public void track(String eventName, EvaluationContext context, TrackingEventDetails details) { + trackCount.incrementAndGet(); + if (throwOnTrack) { + throw new RuntimeException("track failure"); + } + } + + @Override + public ProviderEvaluation getBooleanEvaluation( + String key, Boolean defaultValue, EvaluationContext ctx) { + return ProviderEvaluation.builder().value(Boolean.TRUE).build(); + } + + @Override + public ProviderEvaluation getStringEvaluation(String key, String defaultValue, EvaluationContext ctx) { + return ProviderEvaluation.builder().value("value").build(); + } + + @Override + public ProviderEvaluation getIntegerEvaluation( + String key, Integer defaultValue, EvaluationContext ctx) { + return ProviderEvaluation.builder().value(1).build(); + } + + @Override + public ProviderEvaluation getDoubleEvaluation(String key, Double defaultValue, EvaluationContext ctx) { + return ProviderEvaluation.builder().value(1d).build(); + } + + @Override + public ProviderEvaluation getObjectEvaluation(String key, Value defaultValue, EvaluationContext ctx) { + return ProviderEvaluation.builder().value(new Value("value")).build(); + } + } + + static class InitializingStateProvider extends TrackingProvider { + private final ProviderState initializeState; + + InitializingStateProvider(String name, ProviderState initializeState) { + super(name); + this.initializeState = initializeState; + } + + @Override + public void initialize(EvaluationContext evaluationContext) throws Exception { + if (ProviderState.STALE.equals(initializeState)) { + emitProviderStale(ProviderEventDetails.builder() + .message("stale during init") + .build()) + .await(); + } else if (ProviderState.FATAL.equals(initializeState)) { + emitProviderError(ProviderEventDetails.builder() + .errorCode(dev.openfeature.sdk.ErrorCode.PROVIDER_FATAL) + .message("fatal during init") + .build()) + .await(); + } else if (ProviderState.ERROR.equals(initializeState)) { + emitProviderError(ProviderEventDetails.builder() + .errorCode(dev.openfeature.sdk.ErrorCode.GENERAL) + .message("error during init") + .build()) + .await(); + } + } + } +} diff --git a/src/test/java/dev/openfeature/sdk/multiprovider/MultiProviderTest.java b/src/test/java/dev/openfeature/sdk/multiprovider/MultiProviderTest.java index 4976d5276..012ff6681 100644 --- a/src/test/java/dev/openfeature/sdk/multiprovider/MultiProviderTest.java +++ b/src/test/java/dev/openfeature/sdk/multiprovider/MultiProviderTest.java @@ -145,7 +145,12 @@ void shouldHandleDuplicateProviderNames() { List providers = new ArrayList<>(2); providers.add(mockProvider1); providers.add(mockProvider2); - assertDoesNotThrow(() -> new MultiProvider(providers).initialize(null)); + MultiProvider multiProvider = new MultiProvider(providers); + assertDoesNotThrow(() -> multiProvider.initialize(null)); + MultiProviderMetadata metadata = (MultiProviderMetadata) multiProvider.getMetadata(); + assertEquals(2, metadata.getOriginalMetadata().size()); + assertNotNull(metadata.getOriginalMetadata().get("provider")); + assertNotNull(metadata.getOriginalMetadata().get("provider-1")); } @Test