Align aggregation to spec

Author: jackshirazi

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

@@ -7,17 +7,16 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
-/**
- * Holds the latest validated policy snapshot and reports whether an update changed effective
- * configuration.
- */
+/** Holds the latest validated policy snapshot and publishes accepted updates to implementers. */
 public final class PolicyStore {
   private static final Logger logger = Logger.getLogger(PolicyStore.class.getName());
 
@@ -26,30 +25,25 @@ public final class PolicyStore {
   private long policyVersion;
 
   /**
-   * Replaces the stored policies when the new snapshot is not equal to the current one.
+   * Replaces the stored policies with a new accepted snapshot.
    *
-   * <p>Input lists are normalized to a set of distinct policies using value equality: duplicates
-   * are dropped and only the first occurrence of each policy is kept (insertion order). Change
-   * detection uses set equality, so list order does not matter. That matches telemetry policy
-   * semantics where the effective result does not depend on processing order (see the telemetry
-   * policy OTEP, commutativity / no user-defined ordering between policies).
+   * <p>Input lists are normalized by policy identity, using policy type and policy id. Duplicate
+   * policies with the same identity are dropped and only the first occurrence is kept in insertion
+   * order. The store does not perform value-level no-op detection; providers can use source
+   * hashes/versions to skip unchanged snapshots, and implementers must apply policies idempotently.
    *
-   * @return {@code true} if the store was updated, {@code false} if the snapshot was unchanged
+   * @return {@code true} after the snapshot is accepted
    */
   public boolean updatePolicies(List<TelemetryPolicy> newPolicies) {
     Objects.requireNonNull(newPolicies, "newPolicies cannot be null");
-    LinkedHashSet<TelemetryPolicy> newPolicySet = new LinkedHashSet<>(newPolicies);
+    Map<PolicyKey, TelemetryPolicy> newPolicyMap = normalizedPolicyMap(newPolicies);
     List<TelemetryPolicy> policiesSnapshot;
     List<TelemetryPolicy> notificationSnapshot;
     List<RegisteredImplementer> implementersSnapshot;
     long snapshotVersion;
     synchronized (this) {
-      if (new LinkedHashSet<>(policies).equals(newPolicySet)) {
-        return false;
-      }
-      List<TelemetryPolicy> deletedPolicies =
-          deletedPoliciesFrom(policies, new ArrayList<>(newPolicySet));
-      policies = new ArrayList<>(newPolicySet);
+      List<TelemetryPolicy> deletedPolicies = deletedPoliciesFrom(policies, newPolicyMap.keySet());
+      policies = new ArrayList<>(newPolicyMap.values());
       policyVersion++;
       snapshotVersion = policyVersion;
       policiesSnapshot = new ArrayList<>(policies);
@@ -105,14 +99,10 @@ private static List<TelemetryPolicy> relevantPoliciesFor(
   }
 
   private static List<TelemetryPolicy> deletedPoliciesFrom(
-      List<TelemetryPolicy> previousPolicies, List<TelemetryPolicy> newPolicies) {
-    Set<String> activePolicyKeys = new LinkedHashSet<>();
-    for (TelemetryPolicy policy : newPolicies) {
-      activePolicyKeys.add(policyKey(policy));
-    }
+      List<TelemetryPolicy> previousPolicies, Set<PolicyKey> activePolicyKeys) {
     ArrayList<TelemetryPolicy> deletedPolicies = new ArrayList<>();
     for (TelemetryPolicy previousPolicy : previousPolicies) {
-      String policyKey = policyKey(previousPolicy);
+      PolicyKey policyKey = policyKey(previousPolicy);
       TelemetryPolicyIdentity identity = previousPolicy.getIdentity();
       if (!activePolicyKeys.contains(policyKey)) {
         deletedPolicies.add(new DeletedTelemetryPolicy(identity, previousPolicy.getType()));
@@ -121,8 +111,21 @@ private static List<TelemetryPolicy> deletedPoliciesFrom(
     return deletedPolicies;
   }
 
-  private static String policyKey(TelemetryPolicy policy) {
-    return policy.getType() + "\u0000" + policy.getIdentity().getId();
+  private static Map<PolicyKey, TelemetryPolicy> normalizedPolicyMap(
+      List<TelemetryPolicy> policies) {
+    LinkedHashMap<PolicyKey, TelemetryPolicy> normalized = new LinkedHashMap<>();
+    for (TelemetryPolicy policy : policies) {
+      Objects.requireNonNull(policy, "newPolicies cannot contain null elements");
+      PolicyKey key = policyKey(policy);
+      if (!normalized.containsKey(key)) {
+        normalized.put(key, policy);
+      }
+    }
+    return normalized;
+  }
+
+  private static PolicyKey policyKey(TelemetryPolicy policy) {
+    return new PolicyKey(policy.getType(), policy.getIdentity().getId());
   }
 
   private static void notifyImplementer(
@@ -150,4 +153,33 @@ private RegisteredImplementer(PolicyImplementer implementer) {
       this.implementer = implementer;
     }
   }
+
+  private static final class PolicyKey {
+    private final String type;
+    private final String id;
+
+    private PolicyKey(String type, String id) {
+      this.type = Objects.requireNonNull(type, "policy type cannot be null");
+      this.id = Objects.requireNonNull(id, "policy id cannot be null");
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (this == obj) {
+        return true;
+      }
+      if (!(obj instanceof PolicyKey)) {
+        return false;
+      }
+      PolicyKey that = (PolicyKey) obj;
+      return type.equals(that.type) && id.equals(that.id);
+    }
+
+    @Override
+    public int hashCode() {
+      int result = type.hashCode();
+      result = 31 * result + id.hashCode();
+      return result;
+    }
+  }
 }

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

@@ -32,40 +32,51 @@ void updatePoliciesReturnsTrueOnFirstSet() {
   }
 
   @Test
-  void updatePoliciesReturnsFalseWhenEqualContent() {
+  void updatePoliciesAcceptsRepeatedEquivalentSnapshot() {
     PolicyStore store = new PolicyStore();
     assertThat(store.updatePolicies(singletonList(new TraceSamplingRatePolicy(0.5)))).isTrue();
-    assertThat(store.updatePolicies(singletonList(new TraceSamplingRatePolicy(0.5)))).isFalse();
+    assertThat(store.updatePolicies(singletonList(new TraceSamplingRatePolicy(0.5)))).isTrue();
   }
 
   @Test
   void updatePoliciesReturnsTrueWhenProbabilityChanges() {
     PolicyStore store = new PolicyStore();
     assertThat(store.updatePolicies(singletonList(new TraceSamplingRatePolicy(0.25)))).isTrue();
     assertThat(store.updatePolicies(singletonList(new TraceSamplingRatePolicy(0.75)))).isTrue();
-    assertThat(store.getPolicies()).containsExactly(new TraceSamplingRatePolicy(0.75));
+    assertThat(store.getPolicies()).hasSize(1);
+    assertThat(((TraceSamplingRatePolicy) store.getPolicies().get(0)).getProbability())
+        .isEqualTo(0.75);
   }
 
   @Test
-  void updatePoliciesReturnsFalseWhenOnlyOrderDiffers() {
+  void updatePoliciesAcceptsReorderedSnapshot() {
     PolicyStore store = new PolicyStore();
     List<TelemetryPolicy> first =
-        Arrays.asList(new TraceSamplingRatePolicy(0.1), new TraceSamplingRatePolicy(0.2));
-    List<TelemetryPolicy> reordered =
-        Arrays.asList(new TraceSamplingRatePolicy(0.2), new TraceSamplingRatePolicy(0.1));
+        Arrays.asList(
+            new TestTelemetryPolicy(
+                new TelemetryPolicyIdentity("first-policy", "First policy"), "test-policy"),
+            new TestTelemetryPolicy(
+                new TelemetryPolicyIdentity("second-policy", "Second policy"), "test-policy"));
+    List<TelemetryPolicy> reordered = Arrays.asList(first.get(1), first.get(0));
 
     assertThat(store.updatePolicies(first)).isTrue();
-    assertThat(store.updatePolicies(reordered)).isFalse();
-    assertThat(store.getPolicies()).isEqualTo(first);
+    assertThat(store.updatePolicies(reordered)).isTrue();
+    assertThat(store.getPolicies()).containsExactly(reordered.get(0), reordered.get(1));
   }
 
   @Test
-  void updatePoliciesIgnoresDuplicatePoliciesInInput() {
+  void updatePoliciesIgnoresDuplicatePolicyIdentitiesInInput() {
     PolicyStore store = new PolicyStore();
-    TraceSamplingRatePolicy p = new TraceSamplingRatePolicy(0.5);
-    assertThat(store.updatePolicies(Arrays.asList(p, new TraceSamplingRatePolicy(0.5)))).isTrue();
-    assertThat(store.getPolicies()).containsExactly(p);
-    assertThat(store.updatePolicies(singletonList(new TraceSamplingRatePolicy(0.5)))).isFalse();
+    TelemetryPolicy first =
+        new TestTelemetryPolicy(
+            new TelemetryPolicyIdentity("test-policy", "Test policy"), "test-policy");
+    TelemetryPolicy duplicate =
+        new TestTelemetryPolicy(
+            new TelemetryPolicyIdentity("test-policy", "Updated name"), "test-policy");
+
+    assertThat(store.updatePolicies(Arrays.asList(first, duplicate))).isTrue();
+
+    assertThat(store.getPolicies()).containsExactly(first);
   }
 
   @Test
@@ -85,7 +96,8 @@ void registerImplementerReceivesCurrentRelevantPolicies() {
 
     store.registerImplementer(implementer);
 
-    verify(implementer).onPoliciesChanged(singletonList(new TraceSamplingRatePolicy(0.5)));
+    verify(implementer)
+        .onPoliciesChanged(argThat(policies -> containsTraceSamplingProbability(policies, 0.5)));
   }
 
   @Test
@@ -97,7 +109,8 @@ void updatePoliciesNotifiesRegisteredImplementerWithRelevantPolicies() {
     clearInvocations(implementer);
     store.updatePolicies(Arrays.asList(unrelatedPolicy(), new TraceSamplingRatePolicy(0.25)));
 
-    verify(implementer).onPoliciesChanged(singletonList(new TraceSamplingRatePolicy(0.25)));
+    verify(implementer)
+        .onPoliciesChanged(argThat(policies -> containsTraceSamplingProbability(policies, 0.25)));
   }
 
   @Test
@@ -145,7 +158,7 @@ void updatePoliciesNotifiesDeletedPolicyForAnyIdentityBearingPolicy() {
     PolicyImplementer implementer = implementerFor("test-policy");
     TestTelemetryPolicy removedPolicy =
         new TestTelemetryPolicy(
-            new TelemetryPolicyIdentity("test-policy-id", "Test policy"), "test-policy", "old");
+            new TelemetryPolicyIdentity("test-policy-id", "Test policy"), "test-policy");
     store.updatePolicies(singletonList(removedPolicy));
     store.registerImplementer(implementer);
     clearInvocations(implementer);
@@ -219,18 +232,25 @@ private static PolicyImplementer implementerFor(String policyType) {
 
   private static TelemetryPolicy unrelatedPolicy() {
     return new TestTelemetryPolicy(
-        new TelemetryPolicyIdentity("other-policy", "Other policy"), "other-policy", "value");
+        new TelemetryPolicyIdentity("other-policy", "Other policy"), "other-policy");
+  }
+
+  private static boolean containsTraceSamplingProbability(
+      List<TelemetryPolicy> policies, double probability) {
+    if (policies.size() != 1 || !(policies.get(0) instanceof TraceSamplingRatePolicy)) {
+      return false;
+    }
+    TraceSamplingRatePolicy policy = (TraceSamplingRatePolicy) policies.get(0);
+    return Double.compare(policy.getProbability(), probability) == 0;
   }
 
   private static final class TestTelemetryPolicy implements TelemetryPolicy {
     private final TelemetryPolicyIdentity identity;
     private final String type;
-    private final String value;
 
-    private TestTelemetryPolicy(TelemetryPolicyIdentity identity, String type, String value) {
+    private TestTelemetryPolicy(TelemetryPolicyIdentity identity, String type) {
       this.identity = identity;
       this.type = type;
-      this.value = value;
     }
 
     @Override
@@ -242,25 +262,5 @@ public TelemetryPolicyIdentity getIdentity() {
     public String getType() {
       return type;
     }
-
-    @Override
-    public boolean equals(Object obj) {
-      if (this == obj) {
-        return true;
-      }
-      if (!(obj instanceof TestTelemetryPolicy)) {
-        return false;
-      }
-      TestTelemetryPolicy that = (TestTelemetryPolicy) obj;
-      return identity.equals(that.identity) && type.equals(that.type) && value.equals(that.value);
-    }
-
-    @Override
-    public int hashCode() {
-      int result = identity.hashCode();
-      result = 31 * result + type.hashCode();
-      result = 31 * result + value.hashCode();
-      return result;
-    }
   }
 }