[dynamic control] Register implementers

dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/PolicyStore.java

@@ -10,6 +10,7 @@
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Objects;
+import java.util.Set;
 
 /**
  * Holds the latest validated policy snapshot and reports whether an update changed effective
@@ -18,6 +19,7 @@
 public final class PolicyStore {
 
   private List<TelemetryPolicy> policies = Collections.emptyList();
+  private final List<PolicyImplementer> implementers = new ArrayList<>();
 
   /**
    * Replaces the stored policies when the new snapshot is not equal to the current one.
@@ -31,17 +33,58 @@ public final class PolicyStore {
    *
    * @return {@code true} if the store was updated, {@code false} if the snapshot was unchanged
    */
-  public synchronized boolean updatePolicies(List<TelemetryPolicy> newPolicies) {
+  public boolean updatePolicies(List<TelemetryPolicy> newPolicies) {
     Objects.requireNonNull(newPolicies, "newPolicies cannot be null");
     LinkedHashSet<TelemetryPolicy> newPolicySet = new LinkedHashSet<>(newPolicies);
-    if (new LinkedHashSet<>(policies).equals(newPolicySet)) {
-      return false;
+    List<TelemetryPolicy> policiesSnapshot;
+    List<PolicyImplementer> implementersSnapshot;
+    synchronized (this) {
+      if (new LinkedHashSet<>(policies).equals(newPolicySet)) {
+        return false;
+      }
+      policies = new ArrayList<>(newPolicySet);
+      policiesSnapshot = new ArrayList<>(policies);
+      implementersSnapshot = new ArrayList<>(implementers);
+    }
+    for (PolicyImplementer implementer : implementersSnapshot) {
+      implementer.onPoliciesChanged(relevantPoliciesFor(implementer, policiesSnapshot));
     }
-    policies = new ArrayList<>(newPolicySet);
     return true;
   }
 
+  public void registerImplementer(PolicyImplementer implementer) {
+    Objects.requireNonNull(implementer, "implementer cannot be null");
+    List<TelemetryPolicy> policiesSnapshot;
+    synchronized (this) {
+      implementers.add(implementer);
+      policiesSnapshot = new ArrayList<>(policies);
+    }
+    implementer.onPoliciesChanged(relevantPoliciesFor(implementer, policiesSnapshot));
+  }
+
   public synchronized List<TelemetryPolicy> getPolicies() {
     return Collections.unmodifiableList(new ArrayList<>(policies));
   }
+
+  public synchronized void clear() {
+    policies = Collections.emptyList();
+    implementers.clear();
+  }
+
+  private static List<TelemetryPolicy> relevantPoliciesFor(
+      PolicyImplementer implementer, List<TelemetryPolicy> policies) {
+    Set<String> supportedTypes = new LinkedHashSet<>();
+    for (PolicyValidator validator : implementer.getValidators()) {
+      if (validator != null && validator.getPolicyType() != null) {
+        supportedTypes.add(validator.getPolicyType());
+      }
+    }
+    ArrayList<TelemetryPolicy> relevant = new ArrayList<>();
+    for (TelemetryPolicy policy : policies) {
+      if (supportedTypes.contains(policy.getType())) {
+        relevant.add(policy);
+      }
+    }
+    return Collections.unmodifiableList(relevant);
+  }
 }

dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/registry/PolicyInit.java

@@ -226,8 +226,7 @@ private static void initializePolicyClass(
     try {
       PolicyImplementer implementer = policyTypeInitializer.initialize(autoConfiguration);
       initializedImplementers.put(policyClass, implementer);
-      // TODO: register implementer with policyStore, not yet implemented
-      // policyStore.registerImplementer(implementer);
+      policyStore.registerImplementer(implementer);
       logger.log(Level.INFO, "Initialized policy class ''{0}''", policyClass.getName());
     } catch (RuntimeException e) {
       throw new IllegalStateException(
@@ -318,6 +317,11 @@ private static List<PolicyValidator> createSourceValidators(PolicySourceConfig s
    * <p>Each source contributes an immutable snapshot; the store receives the flattened union. eg
    * OpAMPPolicyProvider and FilePolicyProvider both produce a new TraceSamplingRatePolicy, this
    * will combine them all into a single list.
+   *
+   * <p>TODO: implement spec-compliant merge semantics: matching policies' keep values must be
+   * combined using the most restrictive result; overlapping policy effects must be commutative,
+   * idempotent, and deterministic; duplicate policy IDs across providers must be resolved by
+   * provider priority.
    */
   private static void updatePoliciesForSource(
       PolicyProvider provider, List<TelemetryPolicy> policiesFromSource) {
@@ -374,6 +378,7 @@ public static void shutdown() {
     sourcePolicies.clear();
     sourcesActivated.set(false);
     initializedImplementers.clear();
+    policyStore.clear();
   }
 
   /**

dynamic-control/src/test/java/io/opentelemetry/contrib/dynamic/policy/PolicyStoreTest.java

@@ -7,6 +7,10 @@
 
 import static java.util.Collections.singletonList;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import io.opentelemetry.contrib.dynamic.policy.tracesampling.TraceSamplingRatePolicy;
 import java.util.Arrays;
@@ -66,4 +70,36 @@ void updatePoliciesIgnoresDuplicatePoliciesInInput() {
   void getPoliciesReturnsEmptyWhenNeverUpdated() {
     assertThat(new PolicyStore().getPolicies()).isEqualTo(Collections.emptyList());
   }
+
+  @Test
+  void registerImplementerReceivesCurrentRelevantPolicies() {
+    PolicyStore store = new PolicyStore();
+    store.updatePolicies(
+        Arrays.asList(new TraceSamplingRatePolicy(0.5), new TelemetryPolicy("other-policy")));
+
+    PolicyImplementer implementer = mock(PolicyImplementer.class);
+    PolicyValidator validator = mock(PolicyValidator.class);
+    when(validator.getPolicyType()).thenReturn(TraceSamplingRatePolicy.POLICY_TYPE);
+    when(implementer.getValidators()).thenReturn(singletonList(validator));
+
+    store.registerImplementer(implementer);
+
+    verify(implementer).onPoliciesChanged(singletonList(new TraceSamplingRatePolicy(0.5)));
+  }
+
+  @Test
+  void updatePoliciesNotifiesRegisteredImplementerWithRelevantPolicies() {
+    PolicyStore store = new PolicyStore();
+    PolicyImplementer implementer = mock(PolicyImplementer.class);
+    PolicyValidator validator = mock(PolicyValidator.class);
+    when(validator.getPolicyType()).thenReturn(TraceSamplingRatePolicy.POLICY_TYPE);
+    when(implementer.getValidators()).thenReturn(singletonList(validator));
+
+    store.registerImplementer(implementer);
+    clearInvocations(implementer);
+    store.updatePolicies(
+        Arrays.asList(new TelemetryPolicy("other-policy"), new TraceSamplingRatePolicy(0.25)));
+
+    verify(implementer).onPoliciesChanged(singletonList(new TraceSamplingRatePolicy(0.25)));
+  }
 }