Drop unsampled transactions when connected to APM Server 8.0+ (elastic#2329)

Author: tobiasstadler

CHANGELOG.asciidoc

@@ -27,6 +27,7 @@ endif::[]
 ===== Features
 * Exceptions that are logged using the fatal log level are now captured (log4j2 only) - {pull}2377[#2377]
 * Replaced `authorization` in the default value of `sanitize_field_names` with `*auth*` - {pull}2326[#2326]
+* Unsampled transactions are dropped and not sent to the APM-Server if the APM-Server version is 8.0+ - {pull}2329[#2329]
 
 [float]
 ===== Bug fixes

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

@@ -368,7 +368,8 @@ public void endTransaction(Transaction transaction) {
                     new RuntimeException("this exception is just used to record where the transaction has been ended from"));
             }
         }
-        if (!transaction.isNoop()) {
+        if (!transaction.isNoop() &&
+            (transaction.isSampled() || apmServerClient.supportsKeepingUnsampledTransaction())) {
             // we do report non-sampled transactions (without the context)
             reporter.report(transaction);
         } else {

apm-agent-core/src/main/java/co/elastic/apm/agent/report/ApmServerClient.java

@@ -71,6 +71,7 @@ public class ApmServerClient {
     private static final Version VERSION_7_0 = Version.of("7.0.0");
     private static final Version VERSION_7_4 = Version.of("7.4.0");
     private static final Version VERSION_7_9 = Version.of("7.9.0");
+    private static final Version VERSION_8_0 = Version.of("8.0.0");
 
     private final ReporterConfiguration reporterConfiguration;
     @Nullable
@@ -338,6 +339,15 @@ public boolean supportsLogsEndpoint() {
         return isAtLeast(VERSION_7_9);
     }
 
+    public boolean supportsKeepingUnsampledTransaction() {
+        // supportsKeepingUnsampledTransaction is called from application threads
+        // return true instead of blocking the thread
+        if (apmServerVersion != null && !apmServerVersion.isDone()) {
+            return true;
+        }
+        return isLowerThan(VERSION_8_0);
+    }
+
     @Nullable
     Version getApmServerVersion(long timeout, TimeUnit timeUnit) throws InterruptedException, ExecutionException, TimeoutException {
         if (apmServerVersion != null) {
@@ -346,6 +356,10 @@ Version getApmServerVersion(long timeout, TimeUnit timeUnit) throws InterruptedE
         return null;
     }
 
+    public boolean isLowerThan(Version apmServerVersion) {
+        return !isAtLeast(apmServerVersion);
+    }
+
     public boolean isAtLeast(Version apmServerVersion) {
         if (this.apmServerVersion == null) {
             throw new IllegalStateException("Called before init event");

apm-agent-core/src/test/java/co/elastic/apm/agent/impl/DropUnsampledTransactionsTest.java

@@ -0,0 +1,105 @@
+/*
+ * 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;
+
+import co.elastic.apm.agent.AbstractInstrumentationTest;
+import co.elastic.apm.agent.MockReporter;
+import co.elastic.apm.agent.configuration.CoreConfiguration;
+import co.elastic.apm.agent.configuration.SpyConfiguration;
+import co.elastic.apm.agent.impl.metadata.MetaData;
+import co.elastic.apm.agent.objectpool.TestObjectPoolFactory;
+import co.elastic.apm.agent.report.ApmServerClient;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.stagemonitor.configuration.ConfigurationRegistry;
+
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class DropUnsampledTransactionsTest extends AbstractInstrumentationTest {
+
+    private static ApmServerClient apmServerClient = mock(ApmServerClient.class);
+
+    private static MockReporter reporter = new MockReporter();
+
+    private static ElasticApmTracer tracer;
+
+    @BeforeAll
+    static void startTracer() {
+        ConfigurationRegistry configurationRegistry = SpyConfiguration.createSpyConfig();
+        tracer = new ElasticApmTracer(configurationRegistry, reporter, new TestObjectPoolFactory(), apmServerClient, "ephemeralId", MetaData.create(configurationRegistry, "ephemeralId"));
+        tracer.start(false);
+    }
+
+    @AfterAll
+    static void stopTracer() {
+        tracer.stop();
+    }
+
+    @AfterEach
+    void resetReporter() {
+        reporter.reset();
+    }
+
+    @Test
+    void whenTheAPMServerSupportsKeepingUnsampledTransactionsUnsampledTransactionsShouldBeReported() throws IOException {
+        when(apmServerClient.supportsKeepingUnsampledTransaction()).thenReturn(true);
+        tracer.getConfig(CoreConfiguration.class).getSampleRate().update(0.0, SpyConfiguration.CONFIG_SOURCE_NAME);
+
+        tracer.startRootTransaction(null).end();
+
+        assertThat(reporter.getTransactions().size()).isEqualTo(1);
+    }
+
+    @Test
+    void whenTheAPMServerSupportsKeepingUnsampledTransactionsSampledTransactionsShouldBeReported() throws IOException {
+        when(apmServerClient.supportsKeepingUnsampledTransaction()).thenReturn(true);
+        tracer.getConfig(CoreConfiguration.class).getSampleRate().update(1.0, SpyConfiguration.CONFIG_SOURCE_NAME);
+
+        tracer.startRootTransaction(null).end();
+
+        assertThat(reporter.getTransactions().size()).isEqualTo(1);
+    }
+
+    @Test
+    void whenTheAPMServerDoesNotSupportsKeepingUnsampledTransactionsUnsampledTransactionsShouldNotBeReported() throws IOException {
+        when(apmServerClient.supportsKeepingUnsampledTransaction()).thenReturn(false);
+        tracer.getConfig(CoreConfiguration.class).getSampleRate().update(0.0, SpyConfiguration.CONFIG_SOURCE_NAME);
+
+        tracer.startRootTransaction(null).end();
+
+        assertThat(reporter.getTransactions().size()).isEqualTo(0);
+    }
+
+    @Test
+    void whenTheAPMServerDoesNotSupportsKeepingUnsampledTransactionsSampledTransactionsShouldBeReported() throws IOException {
+        when(apmServerClient.supportsKeepingUnsampledTransaction()).thenReturn(false);
+        tracer.getConfig(CoreConfiguration.class).getSampleRate().update(1.0, SpyConfiguration.CONFIG_SOURCE_NAME);
+
+        tracer.startRootTransaction(null).end();
+
+        assertThat(reporter.getTransactions().size()).isEqualTo(1);
+    }
+}