add support for elastic.otel.opamp.headers config

CHANGELOG.next-release.md

@@ -13,8 +13,9 @@ This file contains all changes which are not released yet.
 <!--FIXES-END-->
 # Features and enhancements
 <!--ENHANCEMENTS-START-->
-Inferred spans can now be disabled and re-enabled via central config - [#838](https://github.com/elastic/elastic-otel-java/pull/838)
-The agent config is now logged on startup, use option elastic.otel.java.experimental.configuration.logging.enabled (default true) to disable if needed - [835](https://github.com/elastic/elastic-otel-java/pull/835)
+* Inferred spans can now be disabled and re-enabled via central config - [#838](https://github.com/elastic/elastic-otel-java/pull/838)
+* The agent config is now logged on startup, use option elastic.otel.java.experimental.configuration.logging.enabled (default true) to disable if needed - [835](https://github.com/elastic/elastic-otel-java/pull/835)
+* add header support for OpAMP integration [#848](https://github.com/elastic/elastic-otel-java/pull/848)
 <!--ENHANCEMENTS-END-->
 # Deprecations
 <!--DEPRECATIONS-START-->

custom/build.gradle.kts

@@ -13,6 +13,7 @@ dependencies {
     exclude(group = "io.opentelemetry", module = "opentelemetry-api")
   }
   implementation(libs.dslJson)
+  implementation(libs.okhttp)
   implementation(project(":inferred-spans"))
   implementation(project(":universal-profiling-integration"))
   implementation(project(":resources"))

custom/src/main/java/co/elastic/otel/dynamicconfig/CentralConfig.java

@@ -35,38 +35,41 @@
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import javax.annotation.Nullable;
 
 public class CentralConfig {
   private static final Logger logger = Logger.getLogger(CentralConfig.class.getName());
 
+  private static final String OPAMP_HEADERS = "elastic.otel.opamp.headers";
+
   static {
     DynamicConfigurationPropertyChecker.startCheckerThread();
   }
 
   public static void init(SdkTracerProviderBuilder providerBuilder, ConfigProperties properties) {
-    String endpoint = properties.getString("elastic.otel.opamp.endpoint");
+    String endpoint = getEndpoint(properties);
     if (endpoint == null || endpoint.isEmpty()) {
       logger.fine("OpAMP is disabled");
       return;
     }
-    logger.info("Enabling OpAMP as endpoint is defined: " + endpoint);
-    if (!endpoint.endsWith("v1/opamp")) {
-      if (endpoint.endsWith("/")) {
-        endpoint += "v1/opamp";
-      } else {
-        endpoint += "/v1/opamp";
-      }
-    }
+
     String serviceName = getServiceName(properties);
     String environment = getServiceEnvironment(properties);
-    logger.info("Starting OpAmp client for: " + serviceName + " on endpoint " + endpoint);
+    Map<String, String> headers = properties.getMap(OPAMP_HEADERS);
+    if (logger.isLoggable(Level.FINE)) {
+      // only log header names, not the values to prevent potential leaks
+      headers.forEach((k, v) -> logger.fine("OpAMP header: " + k));
+    }
+
+    logger.info("Starting OpAMP client for: " + serviceName + " on endpoint " + endpoint);
     DynamicInstrumentation.setTracerConfigurator(
         providerBuilder, DynamicConfiguration.UpdatableConfigurator.INSTANCE);
     OpampManager opampManager =
         OpampManager.builder()
             .setServiceName(serviceName)
             .setPollingInterval(Duration.ofSeconds(30))
-            .setConfigurationEndpoint(endpoint)
+            .setEndpointUrl(endpoint)
+            .setEndpointHeaders(headers)
             .setServiceEnvironment(environment)
             .build();
 
@@ -81,7 +84,7 @@ public static void init(SdkTracerProviderBuilder providerBuilder, ConfigProperti
         .addShutdownHook(
             new Thread(
                 () -> {
-                  logger.info("=========== Shutting down OpAMP client for: " + serviceName);
+                  logger.info("Shutting down OpAMP client for: " + serviceName);
                   try {
                     opampManager.close();
                   } catch (IOException e) {
@@ -90,31 +93,46 @@ public static void init(SdkTracerProviderBuilder providerBuilder, ConfigProperti
                 }));
   }
 
-  private static String getServiceName(ConfigProperties properties) {
+  // package private for testing
+  @Nullable
+  static String getEndpoint(ConfigProperties properties) {
+    String endpoint = properties.getString("elastic.otel.opamp.endpoint");
+    if (endpoint == null || endpoint.isEmpty()) {
+      return null;
+    }
+    if (!endpoint.endsWith("v1/opamp")) {
+      if (endpoint.endsWith("/")) {
+        endpoint += "v1/opamp";
+      } else {
+        endpoint += "/v1/opamp";
+      }
+    }
+    return endpoint;
+  }
+
+  // package private for testing
+  static String getServiceName(ConfigProperties properties) {
     String serviceName = properties.getString("otel.service.name");
     if (serviceName != null) {
       return serviceName;
     }
     Map<String, String> resourceMap = properties.getMap("otel.resource.attributes");
-    if (resourceMap != null) {
-      serviceName = resourceMap.get("service.name");
-      if (serviceName != null) {
-        return serviceName;
-      }
+    serviceName = resourceMap.get("service.name");
+    if (serviceName != null) {
+      return serviceName;
     }
     return "unknown_service:java"; // Specified default
   }
 
-  private static String getServiceEnvironment(ConfigProperties properties) {
+  // package private for testing
+  @Nullable
+  static String getServiceEnvironment(ConfigProperties properties) {
     Map<String, String> resourceMap = properties.getMap("otel.resource.attributes");
-    if (resourceMap != null) {
-      String environment = resourceMap.get("deployment.environment.name"); // semconv
-      if (environment != null) {
-        return environment;
-      }
-      return resourceMap.get("deployment.environment"); // backward compatible, can be null
+    String environment = resourceMap.get("deployment.environment.name"); // semconv
+    if (environment != null) {
+      return environment;
     }
-    return null;
+    return resourceMap.get("deployment.environment"); // backward compatible, can be null
   }
 
   public static class Configs {

custom/src/main/java/co/elastic/otel/dynamicconfig/internal/OpampManager.java

@@ -38,6 +38,8 @@
 import java.util.logging.Logger;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
 import okio.ByteString;
 import opamp.proto.AgentConfigFile;
 import opamp.proto.AgentRemoteConfig;
@@ -63,7 +65,20 @@ public void start(CentralConfigurationProcessor processor) {
 
     OpampClientBuilder builder = OpampClient.builder();
     builder.enableRemoteConfig();
-    OkHttpSender httpSender = OkHttpSender.create(configuration.configurationEndpoint);
+
+    OkHttpClient.Builder okHttpClient = new OkHttpClient().newBuilder();
+
+    // TODO: revisit this later once the upstream opamp client provides a simpler way to add headers
+    okHttpClient
+        .interceptors()
+        .add(
+            chain -> {
+              Request.Builder modifiedRequest = chain.request().newBuilder();
+              configuration.headers.forEach(modifiedRequest::addHeader);
+              return chain.proceed(modifiedRequest.build());
+            });
+
+    OkHttpSender httpSender = OkHttpSender.create(configuration.endpointUrl, okHttpClient.build());
     if (configuration.serviceName != null) {
       builder.putIdentifyingAttribute("service.name", configuration.serviceName);
     }
@@ -164,9 +179,10 @@ public static Builder builder() {
 
   public static class Builder {
     private String serviceName;
-    private String environment;
-    private String configurationEndpoint = "http://localhost:4320/v1/opamp";
+    @Nullable private String environment;
+    private String endpointUrl = "http://localhost:4320/v1/opamp";
     private Duration pollingInterval = Duration.ofSeconds(30);
+    private Map<String, String> headers = Collections.emptyMap();
 
     private Builder() {}
 
@@ -175,8 +191,13 @@ public Builder setServiceName(String serviceName) {
       return this;
     }
 
-    public Builder setConfigurationEndpoint(String configurationEndpoint) {
-      this.configurationEndpoint = configurationEndpoint;
+    public Builder setEndpointHeaders(Map<String, String> headers) {
+      this.headers = headers;
+      return this;
+    }
+
+    public Builder setEndpointUrl(String endpointUrl) {
+      this.endpointUrl = endpointUrl;
       return this;
     }
 
@@ -185,14 +206,14 @@ public Builder setPollingInterval(Duration pollingInterval) {
       return this;
     }
 
-    public Builder setServiceEnvironment(String environment) {
+    public Builder setServiceEnvironment(@Nullable String environment) {
       this.environment = environment;
       return this;
     }
 
     public OpampManager build() {
       return new OpampManager(
-          new Configuration(serviceName, environment, configurationEndpoint, pollingInterval));
+          new Configuration(serviceName, environment, endpointUrl, pollingInterval, headers));
     }
   }
 
@@ -208,19 +229,22 @@ enum Result {
 
   private static class Configuration {
     private final String serviceName;
-    private final String environment;
-    private final String configurationEndpoint;
+    @Nullable private final String environment;
+    private final String endpointUrl;
     private final Duration pollingInterval;
+    private final Map<String, String> headers;
 
     private Configuration(
         String serviceName,
-        String environment,
-        String configurationEndpoint,
-        Duration pollingInterval) {
+        @Nullable String environment,
+        String endpointUrl,
+        Duration pollingInterval,
+        Map<String, String> headers) {
       this.serviceName = serviceName;
       this.environment = environment;
-      this.configurationEndpoint = configurationEndpoint;
+      this.endpointUrl = endpointUrl;
       this.pollingInterval = pollingInterval;
+      this.headers = headers;
     }
   }
 

custom/src/test/java/co/elastic/otel/dynamicconfig/CentralConfigTest.java

@@ -0,0 +1,120 @@
+/*
+ * 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.otel.dynamicconfig;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
+import io.opentelemetry.sdk.autoconfigure.spi.internal.DefaultConfigProperties;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+class CentralConfigTest {
+
+  @Test
+  void getEndpoint() {
+    testEndpoint(null, null, "missing config should return null");
+    testEndpoint("", null, "empty config should return null");
+    testEndpoint(
+        "http://localhost:8080/v1/opamp",
+        "http://localhost:8080/v1/opamp",
+        "opamp suffix should be automatically added");
+    testEndpoint(
+        "http://localhost:8080/",
+        "http://localhost:8080/v1/opamp",
+        "opamp suffix should be automatically added");
+    testEndpoint(
+        "http://localhost:8080",
+        "http://localhost:8080/v1/opamp",
+        "opamp suffix should be automatically added");
+  }
+
+  private static void testEndpoint(
+      String configValue, String expectedEndpoint, String description) {
+    Map<String, String> map = Collections.emptyMap();
+    if (configValue != null) {
+      map = Collections.singletonMap("elastic.otel.opamp.endpoint", configValue);
+    }
+    assertThat(CentralConfig.getEndpoint(DefaultConfigProperties.createFromMap(map)))
+        .describedAs(description)
+        .isEqualTo(expectedEndpoint);
+  }
+
+  @Test
+  void getServiceName() {
+    Map<String, String> map = Collections.emptyMap();
+    testServiceName(map, "unknown_service:java", "default service name should be provided");
+
+    map = Collections.singletonMap("otel.service.name", "my-service-1");
+    testServiceName(map, "my-service-1", "set through service name config");
+
+    map = Collections.singletonMap("otel.resource.attributes", "service.name=my-service-2");
+    testServiceName(map, "my-service-2", "set through resource attributes config");
+
+    map = new HashMap<>();
+    map.put("otel.service.name", "my-service-3");
+    map.put("otel.resource.attributes", "service.name=my-service-4");
+    testServiceName(map, "my-service-3", "service name takes precedence over resource attributes");
+
+    map.clear();
+    map.put("otel.resource.attributes", "");
+    testServiceName(map, "unknown_service:java", "default service name should be provided");
+
+    map.clear();
+    map.put("otel.resource.attributes", "service.name=");
+    testServiceName(map, "unknown_service:java", "default service name should be provided");
+  }
+
+  private static void testServiceName(
+      Map<String, String> map, String expectedServiceName, String description) {
+    ConfigProperties configProperties = DefaultConfigProperties.createFromMap(map);
+    assertThat(CentralConfig.getServiceName(configProperties))
+        .isNotNull()
+        .describedAs(description)
+        .isEqualTo(expectedServiceName);
+  }
+
+  @Test
+  void getServiceEnvironment() {
+    Map<String, String> map = Collections.emptyMap();
+    testServiceEnvironment(map, null, "no environment by default");
+
+    map = Collections.singletonMap("otel.resource.attributes", "deployment.environment.name=test1");
+    testServiceEnvironment(map, "test1", "environment set through resource attribute");
+
+    map = Collections.singletonMap("otel.resource.attributes", "deployment.environment=test2");
+    testServiceEnvironment(map, "test2", "environment set through legacy resource attribute");
+
+    map =
+        Collections.singletonMap(
+            "otel.resource.attributes",
+            "deployment.environment=test3,deployment.environment.name=test4");
+    testServiceEnvironment(map, "test4", "when both set semconv attribute takes precedence");
+  }
+
+  private static void testServiceEnvironment(
+      Map<String, String> map, String expectedEnvironment, String description) {
+    ConfigProperties configProperties = DefaultConfigProperties.createFromMap(map);
+    assertThat(CentralConfig.getServiceEnvironment(configProperties))
+        .describedAs(description)
+        .isEqualTo(expectedEnvironment);
+  }
+}