Period based adaptive sampler

agent-interfaces/src/main/java/com/newrelic/agent/interfaces/SamplingPriorityQueue.java

@@ -34,13 +34,7 @@ public interface SamplingPriorityQueue<E extends PriorityAware> {
 
     String getServiceName();
 
-    int getSampled();
-
-    int getDecided();
-
-    int getTarget();
-
-    int getDecidedLast();
+    int getTotalSampledPriorityEvents();
 
     int size();
 

agent-model/src/main/java/com/newrelic/agent/model/AnalyticsEvent.java

@@ -68,11 +68,6 @@ public long getTimestamp() {
         return timestamp;
     }
 
-    @Override
-    public boolean decider() {
-        return false;
-    }
-
     @Override
     public float getPriority() {
         return priority;

agent-model/src/main/java/com/newrelic/agent/model/PriorityAware.java

@@ -8,12 +8,10 @@
 package com.newrelic.agent.model;
 
 /**
- * Simple interface to grab a priority (float) value from an object and to determine if this app was the "decider".
+ * Simple interface to grab a priority (float) value from an object.
  */
 public interface PriorityAware {
 
-    boolean decider();
-
     float getPriority();
 
 }

agent-model/src/main/java/com/newrelic/agent/model/SpanEvent.java

@@ -25,13 +25,11 @@ public class SpanEvent extends AnalyticsEvent implements JSONStreamAware {
     private final String appName;
     private final Map<String, Object> intrinsics;
     private final Map<String, Object> agentAttributes;
-    private final boolean decider;
 
     private SpanEvent(Builder builder) {
         super(SPAN, builder.timestamp, builder.priority, builder.userAttributes);
         this.appName = builder.appName;
         this.agentAttributes = builder.agentAttributes;
-        this.decider = builder.decider;
         this.intrinsics = builder.intrinsics;
     }
 
@@ -51,11 +49,6 @@ public Map<String, Object> getAgentAttributes() {
         return agentAttributes;
     }
 
-    @Override
-    public boolean decider() {
-        return decider;
-    }
-
     @Override
     public void writeJSONString(Writer out) throws IOException {
         JSONArray.writeJSONString(Arrays.asList(intrinsics, getMutableUserAttributes(), getAgentAttributes()), out);
@@ -98,16 +91,15 @@ public boolean equals(Object o) {
             return false;
         }
         SpanEvent spanEvent = (SpanEvent) o;
-        return decider == spanEvent.decider &&
-                Objects.equals(appName, spanEvent.appName) &&
+        return  Objects.equals(appName, spanEvent.appName) &&
                 Objects.equals(intrinsics, spanEvent.intrinsics) &&
                 Objects.equals(agentAttributes, spanEvent.agentAttributes) &&
                 super.equals(o);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(appName, intrinsics, agentAttributes, decider);
+        return Objects.hash(appName, intrinsics, agentAttributes);
     }
 
     public static class Builder {
@@ -116,7 +108,6 @@ public static class Builder {
         private final Map<String, Object> userAttributes = new HashMap<>();
         private String appName;
         private float priority;
-        private boolean decider;
         private long timestamp;
         private Object spanKind;
 
@@ -191,11 +182,6 @@ public Object getSpanKindFromUserAttributes() {
             return result == null ? CLIENT_SPAN_KIND : result;
         }
 
-        public Builder decider(boolean decider) {
-            this.decider = decider;
-            return this;
-        }
-
         public Builder timestamp(long timestamp) {
             this.timestamp = timestamp;
             return this;

agent-model/src/test/java/com/newrelic/agent/model/AnalyticsEventTest.java

@@ -46,7 +46,6 @@ public void allGetters_returnProperly() {
         Assert.assertEquals(0.9F, event.getPriority(), .1);
         Assert.assertEquals("type", event.getType());
         Assert.assertTrue(event.isValid());
-        Assert.assertFalse(event.decider());
     }
 
     @Test

agent-model/src/test/java/com/newrelic/agent/model/SpanEventTest.java

@@ -53,10 +53,6 @@ public void testEquals() {
         assertNotEquals(span1, span5);
         assertNotEquals(span2, span5);
 
-        SpanEvent span6 = baseBuilder(now).decider(false).build();
-        assertNotEquals(span1, span6);
-        assertNotEquals(span2, span6);
-
         SpanEvent span7 = baseBuilder(now).appName("somethingDifferent").build();
         assertNotEquals(span1, span7);
         assertNotEquals(span2, span7);
@@ -128,7 +124,6 @@ public void spanEvent_getsAll_Attributes() {
         assertEquals("thud", spanEvent.getName());
         assertEquals("8675zzz", spanEvent.getTransactionId());
         assertEquals("wally", spanEvent.getAppName());
-        assertEquals(true, spanEvent.decider());
     }
 
     private SpanEvent.Builder baseBuilderExtraUser(long now, String extraUserAttr, String value) {
@@ -155,7 +150,6 @@ private SpanEvent.Builder baseBuilder(long now) {
                 .putAllUserAttributes(userAttributes)
                 .putAllAgentAttributes(agentAttributes)
                 .putAllIntrinsics(intrinsics)
-                .decider(true)
                 .appName("wally")
                 .priority(21.7f)
                 .timestamp(now);

instrumentation-test/src/test/java/com/newrelic/agent/introspec/internal/SpanEventImplTest.java

@@ -13,7 +13,7 @@ public class SpanEventImplTest {
     @Test
     public void getters_returnUnderlyingSpanEventData() throws IOException {
         SpanEvent.Builder spanEventBuilder = SpanEvent.builder();
-        SpanEvent spanEvent = spanEventBuilder.appName("appname").decider(true).priority(.1f).timestamp(100000).putIntrinsic("category", "http")
+        SpanEvent spanEvent = spanEventBuilder.appName("appname").priority(.1f).timestamp(100000).putIntrinsic("category", "http")
                 .putIntrinsic("transactionId", "transactionid").putIntrinsic("duration", 0.1f)
                 .putIntrinsic("name", "name").putIntrinsic("traceId", "traceid").putIntrinsic("guid", "guid")
                 .putIntrinsic("parentId", "parentid").putAgentAttribute("http.url", "url").putAgentAttribute("http.statusCode", 200)

newrelic-agent/src/main/java/com/newrelic/agent/HeadersUtil.java

@@ -8,6 +8,7 @@
 package com.newrelic.agent;
 
 import com.google.common.collect.ImmutableSet;
+import com.newrelic.agent.tracing.Sampled;
 import com.newrelic.api.agent.Headers;
 import com.newrelic.agent.config.DistributedTracingConfig;
 import com.newrelic.agent.tracers.DefaultTracer;
@@ -265,7 +266,7 @@ public static void parseAndAcceptDistributedTraceHeaders(Transaction tx, Inbound
                 }
                 if (w3CTracePayload.getTraceParent() != null) {
                     tx.getSpanProxy().setInitiatingW3CTraceParent(w3CTracePayload.getTraceParent());
-                    tx.applyDistributedTracingSamplerConfig(w3CTracePayload.getTraceParent());
+                    tx.assignPriorityFromRemoteParent(w3CTracePayload.getTraceParent().sampled());
                 }
                 if (w3CTracePayload.getTraceState() != null) {
                     tx.getSpanProxy().setInitiatingW3CTraceState(w3CTracePayload.getTraceState());
@@ -276,6 +277,11 @@ public static void parseAndAcceptDistributedTraceHeaders(Transaction tx, Inbound
             String tracePayload = HeadersUtil.getNewRelicTraceHeader(inboundHeaders);
             if (tracePayload != null) {
                 tx.acceptDistributedTracePayload(tracePayload);
+                DistributedTracePayloadImpl newRelicPayload = tx.getSpanProxy().getInboundDistributedTracePayload();
+                boolean samplingDecisionExists = newRelicPayload.sampled != Sampled.UNKNOWN;
+                if (samplingDecisionExists) {
+                    tx.assignPriorityFromRemoteParent(newRelicPayload.sampled.booleanValue());
+                }
             }
         }
     }

newrelic-agent/src/main/java/com/newrelic/agent/RPMService.java

@@ -14,6 +14,7 @@
 import com.newrelic.agent.config.AgentJarHelper;
 import com.newrelic.agent.config.BrowserMonitoringConfig;
 import com.newrelic.agent.config.BrowserMonitoringConfigImpl;
+import com.newrelic.agent.config.DistributedTracingConfig;
 import com.newrelic.agent.config.Hostname;
 import com.newrelic.agent.config.SystemPropertyFactory;
 import com.newrelic.agent.environment.AgentIdentity;
@@ -238,9 +239,6 @@ private Map<String, Object> getSettings(boolean sendEnvironmentInfo) {
         }
         settings.put("services", ServiceFactory.getServicesConfiguration());
 
-        // the sysprops and envvars above an in unmodifiable collections, so just change it here
-        updateSamplingTargetSettings(settings);
-
         return settings;
     }
 
@@ -973,51 +971,6 @@ private void handle503Error(Exception e) {
         }
     }
 
-    private void updateSamplingTargetSettings (Map<String, Object> settings) {
-        // the new distributed_tracing.sampler.adaptive_sampling_target is meant to be per minute
-        // but, the harvest cycle is 12 times per minute right now,
-        // so we need to divide this number by 12 until that changes
-        // there are 2 spots we need to do this:
-        //   - settings.distributed_tracing_sampler_adaptive_sampling_target
-        //   - settings.distributed_tracing.sampler.adaptive_sampling_target
-        if (settings == null) return;
-        try {
-            // local settings
-            Map<String, Object> dtConfig = (Map<String, Object>) settings.get("distributed_tracing");
-            if (dtConfig != null) {
-                Map<String, Object> samplerConfig = (Map<String, Object>) dtConfig.get("sampler");
-                if (samplerConfig != null) {
-                    Integer adaptiveSamplingTarget = toInt(samplerConfig.get("adaptive_sampling_target"));
-                    if (adaptiveSamplingTarget != null) {
-                        Integer newAdaptiveSamplingTarget = (int) Math.ceil(adaptiveSamplingTarget.doubleValue() / 12.0);
-                        NewRelic.getAgent()
-                                .getLogger()
-                                .log(Level.FINE, "Updating local setting adaptive_sampling_target from " + adaptiveSamplingTarget + " to " +
-                                        newAdaptiveSamplingTarget);
-                        samplerConfig.put("adaptive_sampling_target", newAdaptiveSamplingTarget);
-                    }
-                }
-            }
-        } catch (Exception e) {
-            NewRelic.getAgent().getLogger().log(Level.WARNING, "Unable to parse local setting adaptive_sampling_target setting: {0}", e);
-        }
-
-        try {
-            // env var
-            if (settings.containsKey("distributed_tracing_sampler_adaptive_sampling_target")) {
-                Integer adaptiveSamplingTarget = toInt(settings.get("distributed_tracing_sampler_adaptive_sampling_target"));
-                Integer newAdaptiveSamplingTarget = (int) Math.ceil(adaptiveSamplingTarget.doubleValue() / 12.0);
-                NewRelic.getAgent()
-                        .getLogger()
-                        .log(Level.FINE, "Updating environment variable adaptive_sampling_target from " + adaptiveSamplingTarget + " to " +
-                                newAdaptiveSamplingTarget);
-                settings.put("distributed_tracing_sampler_adaptive_sampling_target", newAdaptiveSamplingTarget);
-            }
-        } catch (Exception e) {
-            NewRelic.getAgent().getLogger().log(Level.WARNING, "Unable to parse environment variable adaptive_sampling_target setting: {0}", e);
-        }
-    }
-
     private static Integer toInt(Object o) {
         if (o == null) return null;
         if (o instanceof Number) {

newrelic-agent/src/main/java/com/newrelic/agent/Transaction.java

@@ -40,8 +40,6 @@
 import com.newrelic.agent.normalization.Normalizer;
 import com.newrelic.agent.service.ServiceFactory;
 import com.newrelic.agent.service.ServiceUtils;
-import com.newrelic.agent.service.analytics.DistributedSamplingPriorityQueue;
-import com.newrelic.agent.service.analytics.TransactionEvent;
 import com.newrelic.agent.sql.SlowQueryListener;
 import com.newrelic.agent.stats.AbstractMetricAggregator;
 import com.newrelic.agent.stats.StatsWorks;
@@ -59,8 +57,8 @@
 import com.newrelic.agent.tracing.DistributedTracePayloadImpl;
 import com.newrelic.agent.tracing.DistributedTraceService;
 import com.newrelic.agent.tracing.DistributedTraceServiceImpl;
+import com.newrelic.agent.tracing.Sampled;
 import com.newrelic.agent.tracing.SpanProxy;
-import com.newrelic.agent.tracing.W3CTraceParent;
 import com.newrelic.agent.transaction.PriorityTransactionName;
 import com.newrelic.agent.transaction.TransactionCache;
 import com.newrelic.agent.transaction.TransactionCounts;
@@ -283,6 +281,7 @@ public long getTransportDurationInMillis() {
 
     // WARNING: Mutates this instance by mutating the span proxy
     public DistributedTracePayloadImpl createDistributedTracePayload(String spanId) {
+        assignPriorityRootIfNotSet();
         SpanProxy spanProxy = this.spanProxy.get();
         long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - this.getTransactionTimer().getStartTimeInNanos());
         long txnStartTimeSinceEpochInMillis = System.currentTimeMillis() - elapsedMillis;
@@ -333,39 +332,72 @@ public boolean acceptDistributedTracePayload(DistributedTracePayload payload) {
         }
     }
 
-    public void applyDistributedTracingSamplerConfig(W3CTraceParent parent) {
-        if (parent != null) {
-            DistributedTracingConfig dtConfig = getAgentConfig().getDistributedTracingConfig();
-            if (parent.sampled()) { // traceparent exists and sampled is 1
-                if (DistributedTracingConfig.SAMPLE_ALWAYS_ON.equals(dtConfig.getRemoteParentSampled())) {
-                    this.setPriorityIfNotNull(2.0f);
-                } else if (DistributedTracingConfig.SAMPLE_ALWAYS_OFF.equals(dtConfig.getRemoteParentSampled())) {
-                    this.setPriorityIfNotNull(0.0f);
-                } // else leave it as it was
-            } else { // traceparent exists and sampled is 0
-                if (DistributedTracingConfig.SAMPLE_ALWAYS_ON.equals(dtConfig.getRemoteParentNotSampled())) {
-                    this.setPriorityIfNotNull(2.0f);
-                } else if (DistributedTracingConfig.SAMPLE_ALWAYS_OFF.equals(dtConfig.getRemoteParentNotSampled())) {
-                    this.setPriorityIfNotNull(0.0f);
-                } // else leave it as it was
+    /**
+     * Assigns priority to this transaction using sampling and priority data from a remote parent.
+     *
+     * There are two kinds of parent data that are used:
+     * - The remote parent sampled flag, which indicates whether the remote parent is sampled or not. This
+     * determines which parent sampler (remote_parent_sampled or remote_parent_not_sampled) to use when making the priority assignment.
+     * - Inbound priority data, which is taken from the span proxy's inbound payload if available. This
+     * will be used by the adaptive sampler if configured, and ignored by all other sampler types.
+     *
+     * @param remoteParentSampled whether the remote parent was sampled or not
+     */
+    public void assignPriorityFromRemoteParent(boolean remoteParentSampled) {
+        DistributedTraceService dtService = ServiceFactory.getDistributedTraceService();
+        float priority = dtService.calculatePriorityRemoteParent(remoteParentSampled, getPriorityFromInboundSamplingDecision());
+        this.priority.set(priority);
+    }
+
+    /**
+     * Assigns priority to this transaction (unless previously assigned) without any information from a remote parent.
+     *
+     * If Distributed Tracing is enabled, and no priority has been set on this transaction, the configured root sampler
+     * will be used to obtain a priority for this transaction. No inbound priority data is read (because if an inbound
+     * payload was processed, it should have made a priority assignment earlier).
+     *
+     * If a priority assignment has already been made, this call is ignored. This is a required check to avoid
+     * overwriting any priority decision that was made earlier, either because a remote parent was processed or a previous
+     * call to this method was made earlier in the txn's lifecycle.
+     *
+     * If Distributed Tracing is not enabled, a random priority in [0,1) is assigned.
+     */
+    public void assignPriorityRootIfNotSet(){
+        if (getAgentConfig().getDistributedTracingConfig().isEnabled()){
+            if (priority.get() == null){
+                Float samplerPriority = ServiceFactory.getDistributedTraceService().calculatePriorityRoot();
+                priority.compareAndSet(null, samplerPriority);
             }
+        } else {
+            priority.compareAndSet(null, DistributedTraceServiceImpl.nextTruncatedFloat());
         }
     }
 
-    private void checkAndSetPriority() {
-        if (getAgentConfig().getDistributedTracingConfig().isEnabled()) {
-            DistributedTraceService distributedTraceService = ServiceFactory.getDistributedTraceService();
-
-            DistributedTracePayloadImpl inboundPayload = spanProxy.get().getInboundDistributedTracePayload();
-            Float inboundPriority = inboundPayload != null ? inboundPayload.priority : null;
-
-            DistributedSamplingPriorityQueue<TransactionEvent> reservoir = ServiceFactory.getTransactionEventsService()
-                    .getOrCreateDistributedSamplingReservoir(getApplicationName());
+    /***
+     * Retrieve priority from the inbound payload (using both sampling and priority-related information).
+     *
+     * First, check to see if there is a sampling decision available on the inbound payload.
+     * - If there is a sampling decision, use the inbound priority if it exists or compute a new priority.
+     * - If there is no sampling decision, return null.
+     *
+     * In the case of W3C headers, this is distinct from the remoteParentSampled decision we get from the trace parent header.
+     * The sampling and priority values in the payload come from the trace state header (and they may be missing, even if we got a sampled
+     * flag on the trace parent header).
+     *
+     * @return a float in [0, 2) if priority-related information was found, or null if a new decision needs to be made
+     */
 
-            priority.compareAndSet(null, distributedTraceService.calculatePriority(inboundPriority, reservoir));
-        } else {
-            priority.compareAndSet(null, DistributedTraceServiceImpl.nextTruncatedFloat());
+    @VisibleForTesting
+    protected Float getPriorityFromInboundSamplingDecision(){
+        DistributedTracePayloadImpl payload = spanProxy.get().getInboundDistributedTracePayload();
+        if (payload != null && payload.sampled != Sampled.UNKNOWN) {
+            if (payload.priority != null) {
+                return payload.priority;
+            } else {
+                return (payload.sampled.booleanValue() ? 1.0f : 0.0f) + DistributedTraceServiceImpl.nextTruncatedFloat();
+            }
         }
+        return null;
     }
 
     public TransportType getTransportType() {
@@ -496,7 +528,6 @@ protected Transaction() {
     // registered.
     private void postConstruct() {
         this.initialActivity = TransactionActivity.create(this, nextActivityId.getAndIncrement());;
-        checkAndSetPriority();
     }
 
     private static long getGCTime() {
@@ -1012,7 +1043,7 @@ private void finishTransaction() {
             synchronized (lock) {
                 // this may have the side-effect of ignoring the transaction
                 freezeTransactionName();
-
+                assignPriorityRootIfNotSet();
                 if (ignore) {
                     Agent.LOG.log(Level.FINER,
                             "Transaction {0} was cancelled: ignored. This is not an error condition.", this);