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;
@@ -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();
     }
@@ -458,6 +461,12 @@ public final void end(long epochMicros) {
             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,72 @@
+/*
+ * 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 void init(long sum, String compressionStrategy) {
+        this.count.set(1);
+        this.sum.set(sum);
+        this.compressionStrategy = compressionStrategy;
+    }
+
+    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;
+    }
+}

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

@@ -19,6 +19,7 @@
 package co.elastic.apm.agent.impl.transaction;
 
 import co.elastic.apm.agent.configuration.CoreConfiguration;
+import co.elastic.apm.agent.configuration.SpanConfiguration;
 import co.elastic.apm.agent.impl.ElasticApmTracer;
 import co.elastic.apm.agent.impl.context.Db;
 import co.elastic.apm.agent.impl.context.Destination;
@@ -29,9 +30,11 @@
 import co.elastic.apm.agent.objectpool.Recyclable;
 import co.elastic.apm.agent.sdk.logging.Logger;
 import co.elastic.apm.agent.sdk.logging.LoggerFactory;
+import co.elastic.apm.agent.util.StringBuilderUtils;
 
 import javax.annotation.Nullable;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 
 public class Span extends AbstractSpan<Span> implements Recyclable {
@@ -65,6 +68,7 @@ public class Span extends AbstractSpan<Span> implements Recyclable {
      * Any other arbitrary data captured by the agent, optionally provided by the user
      */
     private final SpanContext context = new SpanContext();
+    private final Composite composite = new Composite();
     @Nullable
     private Throwable stacktrace;
     @Nullable
@@ -151,6 +155,14 @@ public SpanContext getContext() {
         return context;
     }
 
+    public boolean isComposite() {
+        return composite.getCount() > 0;
+    }
+
+    public Composite getComposite() {
+        return composite;
+    }
+
     /**
      * Keywords of specific relevance in the span's domain (eg: 'db', 'template', 'ext', etc)
      */
@@ -292,13 +304,123 @@ public void beforeEnd(long epochMicros) {
 
     @Override
     protected void afterEnd() {
-        this.tracer.endSpan(this);
+        if (tracer.getConfig(SpanConfiguration.class).isSpanCompressionEnabled()) {
+            Span buffered = parent.bufferedSpan.get();
+            if (!isCompressionEligible()) {
+                if (buffered != null) {
+                    if (parent.bufferedSpan.compareAndSet(buffered, null)) {
+                        this.tracer.endSpan(buffered);
+                    }
+                }
+                this.tracer.endSpan(this);
+                return;
+            }
+            if (buffered == null) {
+                if (!parent.bufferedSpan.compareAndSet(null, this)) {
+                    this.tracer.endSpan(this);
+                }
+                return;
+            }
+            if (!buffered.tryToCompress(this)) {
+                if (!parent.bufferedSpan.compareAndSet(buffered, this)) {
+                    this.tracer.endSpan(buffered);
+                    this.tracer.endSpan(this);
+                } else {
+                    this.tracer.endSpan(buffered);
+                }
+            } else if (isSampled()) {
+                Transaction transaction = getTransaction();
+                if (transaction != null) {
+                    transaction.getSpanCount().getDropped().incrementAndGet();
+                }
+                decrementReferences();
+            }
+        } else {
+            this.tracer.endSpan(this);
+        }
+    }
+
+    private boolean isCompressionEligible() {
+        return isExit() && isDiscardable() && (getOutcome() == null || getOutcome() == Outcome.SUCCESS);
+    }
+
+    private boolean tryToCompress(Span sibling) {
+        boolean canBeCompressed = isComposite() ? tryToCompressComposite(sibling) : tryToCompressRegular(sibling);
+        if (!canBeCompressed) {
+            return false;
+        }
+
+        long newDuration = sibling.getTimestamp() + sibling.duration - getTimestamp();
+        synchronized (composite) {
+            if (newDuration > duration) {
+                duration = newDuration;
+            }
+        }
+
+        composite.increaseCount();
+        composite.increaseSum(sibling.duration);
+
+        return true;
+    }
+
+    private boolean tryToCompressRegular(Span sibling) {
+        if (!isSameKind(sibling)) {
+            return false;
+        }
+
+        long maxExactMatchDuration = tracer.getConfig(SpanConfiguration.class).getSpanCompressionExactMatchMaxDuration().getMicros();
+        long maxSameKindDuration = tracer.getConfig(SpanConfiguration.class).getSpanCompressionSameKindMaxDuration().getMicros();
+
+        boolean isAlreadyComposite;
+        synchronized (composite) {
+            isAlreadyComposite = isComposite();
+            if (!isAlreadyComposite) {
+                if (StringBuilderUtils.equals(name, sibling.name)) {
+                    if (duration <= maxExactMatchDuration && sibling.duration <= maxExactMatchDuration) {
+                        composite.init(duration, "exact_match");
+                        return true;
+                    }
+                    return false;
+                }
+
+                if (duration <= maxSameKindDuration && sibling.duration <= maxSameKindDuration) {
+                    composite.init(duration, "same_kind");
+                    name.setLength(0);
+                    name.append("Calls to ").append(context.getDestination().getService().getResource());
+                    return true;
+                }
+            }
+        }
+
+        return isAlreadyComposite && tryToCompressComposite(sibling);
+    }
+
+    private boolean tryToCompressComposite(Span sibling) {
+        switch (composite.getCompressionStrategy()) {
+            case "exact_match":
+                long maxExactMatchDuration = tracer.getConfig(SpanConfiguration.class).getSpanCompressionExactMatchMaxDuration().getMicros();
+                return isSameKind(sibling) && StringBuilderUtils.equals(name, sibling.name) && sibling.duration <= maxExactMatchDuration;
+
+            case "same_kind":
+                long maxSameKindDuration = tracer.getConfig(SpanConfiguration.class).getSpanCompressionSameKindMaxDuration().getMicros();
+                return isSameKind(sibling) && sibling.duration <= maxSameKindDuration;
+            default:
+        }
+
+        return false;
+    }
+
+    private boolean isSameKind(Span other) {
+        return Objects.equals(type, other.type)
+            && Objects.equals(subtype, other.subtype)
+            && StringBuilderUtils.equals(context.getDestination().getService().getResource(), other.context.getDestination().getService().getResource());
     }
 
     @Override
     public void resetState() {
         super.resetState();
         context.resetState();
+        composite.resetState();
         stacktrace = null;
         type = null;
         subtype = null;