drop comparable policies according to priority for an update

Author: jackshirazi

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

@@ -28,9 +28,9 @@ public final class PolicyStore {
    * Replaces the stored policies with a new accepted snapshot.
    *
    * <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.
+   * policies with the same identity are resolved by source kind priority. 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} after the snapshot is accepted
    */
@@ -105,7 +105,9 @@ private static List<TelemetryPolicy> deletedPoliciesFrom(
       PolicyKey policyKey = policyKey(previousPolicy);
       TelemetryPolicyIdentity identity = previousPolicy.getIdentity();
       if (!activePolicyKeys.contains(policyKey)) {
-        deletedPolicies.add(new DeletedTelemetryPolicy(identity, previousPolicy.getType()));
+        deletedPolicies.add(
+            new DeletedTelemetryPolicy(
+                identity, previousPolicy.getType(), previousPolicy.getSourceKind()));
       }
     }
     return deletedPolicies;
@@ -117,16 +119,22 @@ private static Map<PolicyKey, TelemetryPolicy> normalizedPolicyMap(
     for (TelemetryPolicy policy : policies) {
       Objects.requireNonNull(policy, "newPolicies cannot contain null elements");
       PolicyKey key = policyKey(policy);
-      if (!normalized.containsKey(key)) {
+      TelemetryPolicy existing = normalized.get(key);
+      if (existing == null || hasHigherPriority(policy, existing)) {
         normalized.put(key, policy);
       }
     }
     return normalized;
   }
 
+  private static boolean hasHigherPriority(TelemetryPolicy candidate, TelemetryPolicy existing) {
+    return candidate.getSourceKind().hasHigherPriorityThan(existing.getSourceKind());
+  }
+
   private static PolicyKey policyKey(TelemetryPolicy policy) {
     TelemetryPolicyIdentity identity =
         Objects.requireNonNull(policy.getIdentity(), "policy identity cannot be null");
+    Objects.requireNonNull(policy.getSourceKind(), "policy source kind cannot be null");
     return new PolicyKey(policy.getType(), identity.getId());
   }
 

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

@@ -14,6 +14,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import io.opentelemetry.contrib.dynamic.policy.source.SourceKind;
 import io.opentelemetry.contrib.dynamic.policy.tracesampling.TraceSamplingRatePolicy;
 import java.util.Arrays;
 import java.util.Collections;
@@ -79,6 +80,49 @@ void updatePoliciesIgnoresDuplicatePolicyIdentitiesInInput() {
     assertThat(store.getPolicies()).containsExactly(first);
   }
 
+  @Test
+  void updatePoliciesResolvesDuplicatePolicyIdentitiesBySourcePriority() {
+    PolicyStore store = new PolicyStore();
+    TelemetryPolicy filePolicy =
+        new TestTelemetryPolicy(
+            new TelemetryPolicyIdentity("test-policy", "File policy"),
+            "test-policy",
+            SourceKind.FILE);
+    TelemetryPolicy httpPolicy =
+        new TestTelemetryPolicy(
+            new TelemetryPolicyIdentity("test-policy", "HTTP policy"),
+            "test-policy",
+            SourceKind.HTTP);
+    TelemetryPolicy opampPolicy =
+        new TestTelemetryPolicy(
+            new TelemetryPolicyIdentity("test-policy", "OpAMP policy"),
+            "test-policy",
+            SourceKind.OPAMP);
+
+    assertThat(store.updatePolicies(Arrays.asList(filePolicy, opampPolicy, httpPolicy))).isTrue();
+
+    assertThat(store.getPolicies()).containsExactly(opampPolicy);
+  }
+
+  @Test
+  void updatePoliciesUsesHttpWhenDuplicatePolicyHasNoOpampSource() {
+    PolicyStore store = new PolicyStore();
+    TelemetryPolicy filePolicy =
+        new TestTelemetryPolicy(
+            new TelemetryPolicyIdentity("test-policy", "File policy"),
+            "test-policy",
+            SourceKind.FILE);
+    TelemetryPolicy httpPolicy =
+        new TestTelemetryPolicy(
+            new TelemetryPolicyIdentity("test-policy", "HTTP policy"),
+            "test-policy",
+            SourceKind.HTTP);
+
+    assertThat(store.updatePolicies(Arrays.asList(filePolicy, httpPolicy))).isTrue();
+
+    assertThat(store.getPolicies()).containsExactly(httpPolicy);
+  }
+
   @Test
   void getPoliciesReturnsEmptyWhenNeverUpdated() {
     assertThat(new PolicyStore().getPolicies()).isEqualTo(Collections.emptyList());
@@ -247,10 +291,17 @@ private static boolean containsTraceSamplingProbability(
   private static final class TestTelemetryPolicy implements TelemetryPolicy {
     private final TelemetryPolicyIdentity identity;
     private final String type;
+    private final SourceKind sourceKind;
 
     private TestTelemetryPolicy(TelemetryPolicyIdentity identity, String type) {
+      this(identity, type, SourceKind.CUSTOM);
+    }
+
+    private TestTelemetryPolicy(
+        TelemetryPolicyIdentity identity, String type, SourceKind sourceKind) {
       this.identity = identity;
       this.type = type;
+      this.sourceKind = sourceKind;
     }
 
     @Override
@@ -262,5 +313,10 @@ public TelemetryPolicyIdentity getIdentity() {
     public String getType() {
       return type;
     }
+
+    @Override
+    public SourceKind getSourceKind() {
+      return sourceKind;
+    }
   }
 }