Drop non-string labels if connected to older APM Server (#746)

Author: felixbarny

CHANGELOG.md

@@ -8,6 +8,7 @@
 ## Bug Fixes
  * Some JMS Consumers and Producers are filtered due to class name filtering in instrumentation matching
  * Jetty: When no display name is set and context path is "/" transaction service names will now correctly fall back to configured values
+ * Drops non-String labels when connected to APM Server < 6.7 to avoid validation errors (#687)
 
 # 1.7.0
 

apm-agent-core/src/main/java/co/elastic/apm/agent/bci/bytebuddy/CustomElementMatchers.java

@@ -25,6 +25,7 @@
 package co.elastic.apm.agent.bci.bytebuddy;
 
 import co.elastic.apm.agent.matcher.WildcardMatcher;
+import co.elastic.apm.agent.util.Version;
 import com.blogspot.mydailyjava.weaklockfree.WeakConcurrentMap;
 import net.bytebuddy.description.NamedElement;
 import net.bytebuddy.description.method.MethodDescription;
@@ -199,35 +200,6 @@ private static Version readImplementationVersionFromManifest(@Nullable Protectio
         return version;
     }
 
-    /**
-     * Based on <a href="https://gist.github.com/brianguertin/ada4b65c6d1c4f6d3eee3c12b6ce021b">https://gist.github.com/brianguertin</a>.
-     * This code was released into the public domain by Brian Guertin on July 8, 2016 citing, verbatim the unlicense.
-     */
-    private static class Version implements Comparable<Version> {
-        private final int[] numbers;
-
-        Version(String version) {
-            final String[] parts = version.split("\\-")[0].split("\\.");
-            numbers = new int[parts.length];
-            for (int i = 0; i < parts.length; i++) {
-                numbers[i] = Integer.valueOf(parts[i]);
-            }
-        }
-
-        @Override
-        public int compareTo(Version another) {
-            final int maxLength = Math.max(numbers.length, another.numbers.length);
-            for (int i = 0; i < maxLength; i++) {
-                final int left = i < numbers.length ? numbers[i] : 0;
-                final int right = i < another.numbers.length ? another.numbers[i] : 0;
-                if (left != right) {
-                    return left < right ? -1 : 1;
-                }
-            }
-            return 0;
-        }
-    }
-
     /**
      * Matches overridden methods of a super class or implemented methods of an interface.
      * Recursively traverses the superclasses and interfaces.

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

@@ -101,21 +101,27 @@ public ElasticApmTracerBuilder withConfig(String key, String value) {
     }
 
     public ElasticApmTracer build() {
+        boolean addApmServerConfigSource = false;
         if (configurationRegistry == null) {
+            addApmServerConfigSource = true;
             final List<ConfigurationSource> configSources = getConfigSources(agentArguments);
             configurationRegistry = getDefaultConfigurationRegistry(configSources);
         }
         final ApmServerClient apmServerClient = new ApmServerClient(configurationRegistry.getConfig(ReporterConfiguration.class));
-        final DslJsonSerializer payloadSerializer = new DslJsonSerializer(configurationRegistry.getConfig(StacktraceConfiguration.class));
+        final DslJsonSerializer payloadSerializer = new DslJsonSerializer(configurationRegistry.getConfig(StacktraceConfiguration.class), apmServerClient);
         final MetaData metaData = MetaData.create(configurationRegistry, null, null);
-        ApmServerConfigurationSource configurationSource = new ApmServerConfigurationSource(payloadSerializer, metaData, apmServerClient);
-        configurationRegistry.addConfigurationSource(configurationSource);
+        ApmServerConfigurationSource configurationSource = null;
+        if (addApmServerConfigSource) {
+            configurationSource = new ApmServerConfigurationSource(payloadSerializer, metaData, apmServerClient);
+            configurationRegistry.addConfigurationSource(configurationSource);
+        }
         if (reporter == null) {
             reporter = new ReporterFactory().createReporter(configurationRegistry, apmServerClient, metaData);
         }
         if (lifecycleListeners.isEmpty()) {
-            lifecycleListeners.add(new ApmServerHealthChecker(apmServerClient));
-            lifecycleListeners.add(configurationSource);
+            if (configurationSource != null) {
+                lifecycleListeners.add(configurationSource);
+            }
             lifecycleListeners.addAll(DependencyInjectingServiceLoader.load(LifecycleListener.class));
         }
         return new ElasticApmTracer(configurationRegistry, reporter, lifecycleListeners);

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

@@ -24,6 +24,7 @@
  */
 package co.elastic.apm.agent.report;
 
+import co.elastic.apm.agent.util.Version;
 import co.elastic.apm.agent.util.VersionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -42,6 +43,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -64,34 +66,44 @@ public class ApmServerClient {
 
     private static final Logger logger = LoggerFactory.getLogger(ApmServerClient.class);
     private static final String USER_AGENT = "elasticapm-java/" + VersionUtils.getAgentVersion();
+    private static final Version APM_SERVER_NON_STRING_LABEL_SUPPORT = new Version("6.7.0");
     private final ReporterConfiguration reporterConfiguration;
     private volatile List<URL> serverUrls;
+    private volatile Future<Version> apmServerVersion;
     private final AtomicInteger errorCount = new AtomicInteger();
+    private final ApmServerHealthChecker healthChecker;
 
     public ApmServerClient(ReporterConfiguration reporterConfiguration) {
         this(reporterConfiguration, shuffleUrls(reporterConfiguration.getServerUrls()));
     }
 
     public ApmServerClient(ReporterConfiguration reporterConfiguration, List<URL> shuffledUrls) {
         this.reporterConfiguration = reporterConfiguration;
+        this.healthChecker = new ApmServerHealthChecker(this);
         this.reporterConfiguration.getServerUrlsOption().addChangeListener(new ConfigurationOption.ChangeListener<List<URL>>() {
             @Override
             public void onChange(ConfigurationOption<?> configurationOption, List<URL> oldValue, List<URL> newValue) {
                 logger.debug("server_urls override with value = ({}).", newValue);
                 if (newValue != null && !newValue.isEmpty()) {
-                    serverUrls = shuffleUrls(newValue);
-                    errorCount.set(0);
+                    setServerUrls(shuffleUrls(newValue));
                 }
             }
         });
-        this.serverUrls = Collections.unmodifiableList(shuffledUrls);
+        setServerUrls(Collections.unmodifiableList(shuffledUrls));
+    }
+
+    private void setServerUrls(List<URL> serverUrls) {
+        this.serverUrls = serverUrls;
+        this.apmServerVersion = healthChecker.checkHealthAndGetMinVersion();
+        this.errorCount.set(0);
     }
 
     private static List<URL> shuffleUrls(List<URL> serverUrls) {
         // shuffling the URL list helps to distribute the load across the apm servers
         // when there are multiple agents, they should not all start connecting to the same apm server
-        Collections.shuffle(serverUrls);
-        return serverUrls;
+        List<URL> copy = new ArrayList<>(serverUrls);
+        Collections.shuffle(copy);
+        return copy;
     }
 
     private static void trustAll(HttpsURLConnection connection) {
@@ -212,12 +224,13 @@ public <V> V execute(String path, ConnectionHandler<V> connectionHandler) throws
         throw previousException;
     }
 
-    public void executeForAllUrls(String path, ConnectionHandler<Void> connectionHandler) {
+    public <T> List<T> executeForAllUrls(String path, ConnectionHandler<T> connectionHandler) {
+        List<T> results = new ArrayList<>(serverUrls.size());
         for (URL serverUrl : serverUrls) {
             HttpURLConnection connection = null;
             try {
                 connection = startRequestToUrl(appendPath(serverUrl, path));
-                connectionHandler.withConnection(connection);
+                results.add(connectionHandler.withConnection(connection));
             } catch (Exception e) {
                 logger.debug("Exception while interacting with APM Server", e);
             } finally {
@@ -226,6 +239,7 @@ public void executeForAllUrls(String path, ConnectionHandler<Void> connectionHan
                 }
             }
         }
+        return results;
     }
 
     URL getCurrentUrl() {
@@ -255,6 +269,23 @@ List<URL> getServerUrls() {
         return this.serverUrls;
     }
 
+    public boolean supportsNonStringLabels() {
+        return isAtLeast(APM_SERVER_NON_STRING_LABEL_SUPPORT);
+    }
+
+    public boolean isAtLeast(Version apmServerVersion) {
+        try {
+            Version localApmServerVersion = this.apmServerVersion.get();
+            if (localApmServerVersion == null) {
+                return false;
+            }
+            return localApmServerVersion.compareTo(apmServerVersion) >= 0;
+        } catch (Exception e) {
+            logger.debug(e.getMessage(), e);
+            return false;
+        }
+    }
+
     public interface ConnectionHandler<T> {
         @Nullable
         T withConnection(HttpURLConnection connection) throws IOException;

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

@@ -24,36 +24,49 @@
  */
 package co.elastic.apm.agent.report;
 
-import co.elastic.apm.agent.context.LifecycleListener;
-import co.elastic.apm.agent.impl.ElasticApmTracer;
 import co.elastic.apm.agent.util.ExecutorUtils;
+import co.elastic.apm.agent.util.Version;
+import com.dslplatform.json.DslJson;
+import com.dslplatform.json.JsonReader;
+import com.dslplatform.json.MapConverter;
+import com.dslplatform.json.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.HttpURLConnection;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
 import java.util.concurrent.ThreadPoolExecutor;
 
-public class ApmServerHealthChecker implements Runnable, LifecycleListener {
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+public class ApmServerHealthChecker implements Callable<Version> {
     private static final Logger logger = LoggerFactory.getLogger(ApmServerHealthChecker.class);
 
     private final ApmServerClient apmServerClient;
+    private final DslJson<Object> dslJson = new DslJson<>();
 
     public ApmServerHealthChecker(ApmServerClient apmServerClient) {
         this.apmServerClient = apmServerClient;
     }
 
-    @Override
-    public void start(ElasticApmTracer tracer) {
+    public Future<Version> checkHealthAndGetMinVersion() {
         ThreadPoolExecutor pool = ExecutorUtils.createSingleThreadDeamonPool("apm-server-healthcheck", 1);
-        pool.execute(this);
-        pool.shutdown();
+        try {
+            return pool.submit(this);
+        } finally {
+            pool.shutdown();
+        }
     }
 
+    @Nullable
     @Override
-    public void run() {
-        apmServerClient.executeForAllUrls("/", new ApmServerClient.ConnectionHandler<Void>() {
+    public Version call() {
+        List<Version> versions = apmServerClient.executeForAllUrls("/", new ApmServerClient.ConnectionHandler<Version>() {
             @Override
-            public Void withConnection(HttpURLConnection connection) {
+            public Version withConnection(HttpURLConnection connection) {
                 try {
                     if (logger.isDebugEnabled()) {
                         logger.debug("Starting healthcheck to {}", connection.getURL());
@@ -69,17 +82,23 @@ public Void withConnection(HttpURLConnection connection) {
                         }
                     } else {
                         // prints out the version info of the APM Server
-                        logger.info("Elastic APM server is available: {}", HttpUtils.getBody(connection));
+                        String body = HttpUtils.getBody(connection);
+                        logger.info("Elastic APM server is available: {}", body);
+                        final JsonReader<Object> reader = dslJson.newReader(body.getBytes(UTF_8));
+                        reader.startObject();
+                        String versionString = MapConverter.deserialize(reader).get("version");
+                        return new Version(versionString);
                     }
                 } catch (Exception e) {
                     logger.warn("Elastic APM server {} is not available ({})", connection.getURL(), e.getMessage());
                 }
                 return null;
             }
         });
-    }
-
-    @Override
-    public void stop() {
+        versions.remove(null);
+        if (!versions.isEmpty()) {
+            return Collections.min(versions);
+        }
+        return null;
     }
 }

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

@@ -47,7 +47,7 @@ public Reporter createReporter(ConfigurationRegistry configurationRegistry, ApmS
     private ReportingEventHandler getReportingEventHandler(ConfigurationRegistry configurationRegistry,
                                                            ReporterConfiguration reporterConfiguration, MetaData metaData, ApmServerClient apmServerClient) {
 
-        final DslJsonSerializer payloadSerializer = new DslJsonSerializer(configurationRegistry.getConfig(StacktraceConfiguration.class));
+        final DslJsonSerializer payloadSerializer = new DslJsonSerializer(configurationRegistry.getConfig(StacktraceConfiguration.class), apmServerClient);
         final ProcessorEventHandler processorEventHandler = ProcessorEventHandler.loadProcessors(configurationRegistry);
         return new IntakeV2ReportingEventHandler(reporterConfiguration, processorEventHandler, payloadSerializer, metaData, apmServerClient);
     }

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

@@ -55,6 +55,7 @@
 import co.elastic.apm.agent.metrics.Labels;
 import co.elastic.apm.agent.metrics.MetricRegistry;
 import co.elastic.apm.agent.metrics.MetricSet;
+import co.elastic.apm.agent.report.ApmServerClient;
 import co.elastic.apm.agent.util.PotentiallyMultiValuedMap;
 import com.dslplatform.json.BoolConverter;
 import com.dslplatform.json.DslJson;
@@ -101,11 +102,13 @@ public class DslJsonSerializer implements PayloadSerializer, MetricRegistry.Metr
     private final Collection<String> excludedStackFrames = Arrays.asList("java.lang.reflect", "com.sun", "sun.", "jdk.internal.");
     private final StringBuilder replaceBuilder = new StringBuilder(MAX_LONG_STRING_VALUE_LENGTH + 1);
     private final StacktraceConfiguration stacktraceConfiguration;
+    private final ApmServerClient apmServerClient;
     @Nullable
     private OutputStream os;
 
-    public DslJsonSerializer(StacktraceConfiguration stacktraceConfiguration) {
+    public DslJsonSerializer(StacktraceConfiguration stacktraceConfiguration, ApmServerClient apmServerClient) {
         this.stacktraceConfiguration = stacktraceConfiguration;
+        this.apmServerClient = apmServerClient;
         jw = new DslJson<>(new DslJson.Settings<>()).newWriter(BUFFER_SIZE);
     }
 
@@ -774,7 +777,7 @@ private void serializeContext(final TransactionContext context, TraceContext tra
         serializeResponse(context.getResponse());
         if (context.hasCustom()) {
             writeFieldName("custom");
-            serializeStringKeyScalarValueMap(context.getCustomIterator(), replaceBuilder, jw, true);
+            serializeStringKeyScalarValueMap(context.getCustomIterator(), replaceBuilder, jw, true, true);
             jw.writeByte(COMMA);
         }
         writeFieldName("tags");
@@ -786,31 +789,28 @@ private void serializeContext(final TransactionContext context, TraceContext tra
     // visible for testing
     void serializeLabels(AbstractContext context) {
         if (context.hasLabels()) {
-            serializeStringKeyScalarValueMap(context.getLabelIterator(), replaceBuilder, jw, false);
+            serializeStringKeyScalarValueMap(context.getLabelIterator(), replaceBuilder, jw, false, apmServerClient.supportsNonStringLabels());
         } else {
             jw.writeByte(OBJECT_START);
             jw.writeByte(OBJECT_END);
         }
     }
 
-    static void serializeStringLabels(Iterator<? extends Map.Entry<String, String>> iterator, StringBuilder replaceBuilder, JsonWriter jw) {
-        serializeStringKeyScalarValueMap(iterator, replaceBuilder, jw, false);
-    }
-
     private static void serializeStringKeyScalarValueMap(Iterator<? extends Map.Entry<String, ? /* String|Number|Boolean */>> it,
-                                                         StringBuilder replaceBuilder, JsonWriter jw, boolean extendedStringLimit) {
+                                                         StringBuilder replaceBuilder, JsonWriter jw, boolean extendedStringLimit,
+                                                         boolean supportsNonStringValues) {
         jw.writeByte(OBJECT_START);
         if (it.hasNext()) {
             Map.Entry<String, ?> kv = it.next();
             writeStringValue(sanitizeLabelKey(kv.getKey(), replaceBuilder), replaceBuilder, jw);
             jw.writeByte(JsonWriter.SEMI);
-            serializeScalarValue(replaceBuilder, jw, kv.getValue(), extendedStringLimit);
+            serializeScalarValue(replaceBuilder, jw, kv.getValue(), extendedStringLimit, supportsNonStringValues);
             while (it.hasNext()) {
                 jw.writeByte(COMMA);
                 kv = it.next();
                 writeStringValue(sanitizeLabelKey(kv.getKey(), replaceBuilder), replaceBuilder, jw);
                 jw.writeByte(JsonWriter.SEMI);
-                serializeScalarValue(replaceBuilder, jw, kv.getValue(), extendedStringLimit);
+                serializeScalarValue(replaceBuilder, jw, kv.getValue(), extendedStringLimit, supportsNonStringValues);
             }
         }
         jw.writeByte(OBJECT_END);
@@ -851,21 +851,29 @@ private static void serialize(Labels labels, StringBuilder replaceBuilder, JsonW
             }
             writeStringValue(sanitizeLabelKey(labels.getKey(i), replaceBuilder), replaceBuilder, jw);
             jw.writeByte(JsonWriter.SEMI);
-            serializeScalarValue(replaceBuilder, jw, labels.getValue(i), false);
+            serializeScalarValue(replaceBuilder, jw, labels.getValue(i), false, false);
         }
     }
 
-    private static void serializeScalarValue(StringBuilder replaceBuilder, JsonWriter jw, Object value, boolean extendedStringLimit) {
+    private static void serializeScalarValue(StringBuilder replaceBuilder, JsonWriter jw, Object value, boolean extendedStringLimit, boolean supportsNonStringValues) {
         if (value instanceof String) {
             if (extendedStringLimit) {
                 writeLongStringValue((String) value, replaceBuilder, jw);
             } else {
                 writeStringValue((String) value, replaceBuilder, jw);
             }
         } else if (value instanceof Number) {
-            NumberConverter.serialize(((Number) value).doubleValue(), jw);
+            if (supportsNonStringValues) {
+                NumberConverter.serialize(((Number) value).doubleValue(), jw);
+            } else {
+                jw.writeNull();
+            }
         } else if (value instanceof Boolean) {
-            BoolConverter.serialize((Boolean) value, jw);
+            if (supportsNonStringValues) {
+                BoolConverter.serialize((Boolean) value, jw);
+            } else {
+                jw.writeNull();
+            }
         } else {
             // can't happen, as AbstractContext enforces the values to be either String, Number or boolean
             jw.writeString("invalid value");