Added (initial) support for compressing spans

CHANGELOG.asciidoc

@@ -27,6 +27,7 @@ endif::[]
 ===== Features
 * Added support for setting service name and version for a transaction via the public api - {pull}2451[#2451]
 * Added support for en-/disabling each public annotation on each own - {pull}2472[#2472]
+* Added support for compressing spans - {pull}2477[#2477]
 
 [float]
 ===== Performance improvements

apm-agent-core/src/main/java/co/elastic/apm/agent/configuration/SpanConfiguration.java

@@ -0,0 +1,71 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package co.elastic.apm.agent.configuration;
+
+import co.elastic.apm.agent.configuration.converter.TimeDuration;
+import co.elastic.apm.agent.configuration.converter.TimeDurationValueConverter;
+import org.stagemonitor.configuration.ConfigurationOption;
+import org.stagemonitor.configuration.ConfigurationOptionProvider;
+
+public class SpanConfiguration extends ConfigurationOptionProvider {
+
+    public static final String HUGE_TRACES_CATEGORY = "Huge Traces";
+
+    private final ConfigurationOption<Boolean> spanCompressionEnabled = ConfigurationOption.booleanOption()
+        .key("span_compression_enabled")
+        .configurationCategory(HUGE_TRACES_CATEGORY)
+        .tags("added[1.30.0]", "internal")
+        .description("Setting this option to true will enable span compression feature.\n" +
+            "Span compression reduces the collection, processing, and storage overhead, and removes clutter from the UI. " +
+            "The tradeoff is that some information such as DB statements of all the compressed spans will not be collected.")
+        .dynamic(true)
+        .buildWithDefault(false);
+
+    private final ConfigurationOption<TimeDuration> spanCompressionExactMatchMaxDuration = TimeDurationValueConverter.durationOption("ms")
+        .key("span_compression_exact_match_max_duration")
+        .configurationCategory(HUGE_TRACES_CATEGORY)
+        .tags("added[1.30.0]", "internal")
+        .description("Consecutive spans that are exact match and that are under this threshold will be compressed into a single composite span. " +
+            "This option does not apply to composite spans. This reduces the collection, processing, and storage overhead, and removes clutter from the UI. " +
+            "The tradeoff is that the DB statements of all the compressed spans will not be collected.")
+        .dynamic(true)
+        .buildWithDefault(TimeDuration.of("50ms"));
+
+    private final ConfigurationOption<TimeDuration> spanCompressionSameKindMaxDuration = TimeDurationValueConverter.durationOption("ms")
+        .key("span_compression_same_kind_max_duration")
+        .configurationCategory(HUGE_TRACES_CATEGORY)
+        .tags("added[1.30.0]", "internal")
+        .description("Consecutive spans to the same destination that are under this threshold will be compressed into a single composite span. " +
+            "This option does not apply to composite spans. This reduces the collection, processing, and storage overhead, and removes clutter from the UI. " +
+            "The tradeoff is that the DB statements of all the compressed spans will not be collected.")
+        .dynamic(true)
+        .buildWithDefault(TimeDuration.of("5ms"));
+
+    public boolean isSpanCompressionEnabled() {
+        return spanCompressionEnabled.get();
+    }
+
+    public TimeDuration getSpanCompressionExactMatchMaxDuration() {
+        return spanCompressionExactMatchMaxDuration.get();
+    }
+
+    public TimeDuration getSpanCompressionSameKindMaxDuration() {
+        return spanCompressionSameKindMaxDuration.get();
+    }
+}

apm-agent-core/src/main/java/co/elastic/apm/agent/configuration/converter/TimeDuration.java

@@ -63,6 +63,10 @@ public long getMillis() {
         return durationMs;
     }
 
+    public long getMicros() {
+        return 1000 * durationMs;
+    }
+
     @Override
     public String toString() {
         return durationString;

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/ElasticApmTracer.java

@@ -383,18 +383,20 @@ public void endSpan(Span span) {
             span.decrementReferences();
             return;
         }
-        if (span.getDuration() < coreConfiguration.getSpanMinDuration().getMillis() * 1000) {
-            logger.debug("Span faster than span_min_duration. Request discarding {}", span);
-            span.requestDiscarding();
-        }
-        if (span.isDiscarded()) {
-            logger.debug("Discarding span {}", span);
-            Transaction transaction = span.getTransaction();
-            if (transaction != null) {
-                transaction.getSpanCount().getDropped().incrementAndGet();
+        if (!span.isComposite()) {
+            if (span.getDuration() < coreConfiguration.getSpanMinDuration().getMicros()) {
+                logger.debug("Span faster than span_min_duration. Request discarding {}", span);
+                span.requestDiscarding();
+            }
+            if (span.isDiscarded()) {
+                logger.debug("Discarding span {}", span);
+                Transaction transaction = span.getTransaction();
+                if (transaction != null) {
+                    transaction.getSpanCount().getDropped().incrementAndGet();
+                }
+                span.decrementReferences();
+                return;
             }
-            span.decrementReferences();
-            return;
         }
         reportSpan(span);
     }

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/AbstractSpan.java

@@ -33,6 +33,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 
 public abstract class AbstractSpan<T extends AbstractSpan<T>> implements Recyclable {
     public static final int PRIO_USER_SUPPLIED = 1000;
@@ -53,7 +54,7 @@ public abstract class AbstractSpan<T extends AbstractSpan<T>> implements Recycla
     private long timestamp;
 
     // in microseconds
-    protected long duration;
+    protected final AtomicLong duration = new AtomicLong();
     private ChildDurationTimer childDurations = new ChildDurationTimer();
     protected AtomicInteger references = new AtomicInteger();
     protected volatile boolean finished = true;
@@ -105,6 +106,8 @@ public abstract class AbstractSpan<T extends AbstractSpan<T>> implements Recycla
 
     private boolean hasCapturedExceptions;
 
+    protected final AtomicReference<Span> bufferedSpan = new AtomicReference<>();
+
     public int getReferenceCount() {
         return references.get();
     }
@@ -211,15 +214,15 @@ public boolean isFinished() {
      * How long the transaction took to complete, in µs
      */
     public long getDuration() {
-        return duration;
+        return duration.get();
     }
 
     public long getSelfDuration() {
-        return duration - childDurations.getDuration();
+        return duration.get() - childDurations.getDuration();
     }
 
     public double getDurationMs() {
-        return duration / AbstractSpan.MS_IN_MICROS;
+        return duration.get() / AbstractSpan.MS_IN_MICROS;
     }
 
     /**
@@ -345,7 +348,7 @@ public void resetState() {
         finished = true;
         name.setLength(0);
         timestamp = 0;
-        duration = 0;
+        duration.set(0L);
         traceContext.resetState();
         childDurations.resetState();
         references.set(0);
@@ -356,6 +359,7 @@ public void resetState() {
         outcome = null;
         userOutcome = null;
         hasCapturedExceptions = false;
+        bufferedSpan.set(null);
     }
 
     public Span createSpan() {
@@ -451,13 +455,19 @@ public void end() {
 
     public final void end(long epochMicros) {
         if (!finished) {
-            this.duration = (epochMicros - timestamp);
+            this.duration.set(epochMicros - timestamp);
             if (name.length() == 0) {
                 name.append("unnamed");
             }
             childDurations.onSpanEnd(epochMicros);
             beforeEnd(epochMicros);
             this.finished = true;
+            Span buffered = bufferedSpan.get();
+            if (buffered != null) {
+                if (bufferedSpan.compareAndSet(buffered, null)) {
+                    this.tracer.endSpan(buffered);
+                }
+            }
             afterEnd();
         } else {
             logger.warn("End has already been called: {}", this);

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/Composite.java

@@ -0,0 +1,75 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package co.elastic.apm.agent.impl.transaction;
+
+import co.elastic.apm.agent.objectpool.Recyclable;
+
+import javax.annotation.Nullable;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class Composite implements Recyclable {
+
+    private final AtomicInteger count = new AtomicInteger(0);
+
+    private final AtomicLong sum = new AtomicLong(0L);
+
+    @Nullable
+    private String compressionStrategy;
+
+    public boolean init(long sum, String compressionStrategy) {
+        if (!this.count.compareAndSet(0, 1)) {
+            return false;
+        }
+        this.sum.set(sum);
+        this.compressionStrategy = compressionStrategy;
+        return true;
+    }
+
+    public int getCount() {
+        return count.get();
+    }
+
+    public void increaseCount() {
+        count.incrementAndGet();
+    }
+
+    public long getSum() {
+        return sum.get();
+    }
+
+    public double getSumMs() {
+        return sum.get() / 1000.0;
+    }
+
+    public void increaseSum(long delta) {
+        this.sum.addAndGet(delta);
+    }
+
+    public String getCompressionStrategy() {
+        return compressionStrategy;
+    }
+
+    @Override
+    public void resetState() {
+        this.count.set(0);
+        this.sum.set(0L);
+        this.compressionStrategy = null;
+    }
+}