Drop rare sampling support on CSS

Author: amarziali

dd-trace-core/src/main/java/datadog/trace/common/metrics/ConflatingMetricsAggregator.java

@@ -319,11 +319,9 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel) {
             SPAN_KINDS.computeIfAbsent(
                 spanKind, UTF8BytesString::create), // save repeated utf8 conversions
             getPeerTags(span, spanKind.toString()));
-    boolean isNewKey = false;
     MetricKey key = keys.putIfAbsent(newKey, newKey);
     if (null == key) {
       key = newKey;
-      isNewKey = true;
     }
     long tag = (span.getError() > 0 ? ERROR_TAG : 0L) | (isTopLevel ? TOP_LEVEL_TAG : 0L);
     long durationNanos = span.getDurationNano();
@@ -340,7 +338,6 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel) {
       }
       // recycle the older key
       key = batch.getKey();
-      isNewKey = false;
     }
     batch = newBatch(key);
     batch.add(tag, durationNanos);
@@ -349,8 +346,8 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel) {
     pending.put(key, batch);
     // must offer to the queue after adding to pending
     inbox.offer(batch);
-    // force keep keys we haven't seen before or errors
-    return isNewKey || span.getError() > 0;
+    // force keep keys there are errors
+    return span.getError() > 0;
   }
 
   private List<UTF8BytesString> getPeerTags(CoreSpan<?> span, String spanKind) {

dd-trace-core/src/test/groovy/datadog/trace/common/metrics/ConflatingMetricAggregatorTest.groovy

@@ -661,45 +661,6 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
     aggregator.close()
   }
 
-  def "aggregator should force keep the first of each key it sees"() {
-    setup:
-    int maxAggregates = 10
-    MetricWriter writer = Mock(MetricWriter)
-    Sink sink = Stub(Sink)
-    DDAgentFeaturesDiscovery features = Mock(DDAgentFeaturesDiscovery)
-    features.supportsMetrics() >> true
-    features.peerTags() >> []
-    ConflatingMetricsAggregator aggregator = new ConflatingMetricsAggregator(empty,
-      features, HealthMetrics.NO_OP, sink, writer, maxAggregates, queueSize, 1, SECONDS)
-    long duration = 100
-    aggregator.start()
-
-    when:
-    def overrides = new boolean[10]
-    for (int i = 0; i < 5; ++i) {
-      overrides[i] = aggregator.publish([
-        new SimpleSpan("service" + i, "operation", "resource", "type", false, true, false, 0, duration, HTTP_OK)
-      ])
-    }
-    for (int i = 0; i < 5; ++i) {
-      overrides[i + 5] = aggregator.publish([
-        new SimpleSpan("service" + i, "operation", "resource", "type", false, true, false, 0, duration, HTTP_OK)
-      ])
-    }
-
-    then: "override only the first of each point in the interval"
-    for (int i = 0; i < 5; ++i) {
-      assert overrides[i]
-    }
-    // these were all repeats, so should be ignored
-    for (int i = 5; i < 10; ++i) {
-      assert !overrides[i]
-    }
-
-    cleanup:
-    aggregator.close()
-  }
-
   def "should be resilient to serialization errors"() {
     setup:
     int maxAggregates = 10