Initial implementation of remote config handling (signalfx#2878)

Author: robsunday

opamp/src/main/java/com/splunk/opentelemetry/opamp/OpampActivator.java

@@ -45,9 +45,9 @@ public class OpampActivator implements AgentListener {
 
   @Override
   public void afterAgent(AutoConfiguredOpenTelemetrySdk autoConfiguredOpenTelemetrySdk) {
-    OpampClientConfiguration config =
+    OpampClientConfiguration opampClientConfiguration =
         OpampClientConfigurationFactory.createConfiguration(autoConfiguredOpenTelemetrySdk);
-    if (!config.isEnabled()) {
+    if (!opampClientConfiguration.isEnabled()) {
       return;
     }
 
@@ -56,12 +56,13 @@ public void afterAgent(AutoConfiguredOpenTelemetrySdk autoConfiguredOpenTelemetr
         createEffectiveConfigFactory(autoConfiguredOpenTelemetrySdk);
     State.EffectiveConfig effectiveConfig = buildEffectiveConfig(effectiveConfigFactory);
 
+    ServerToAgentMessageHandler serverToAgentMessageHandler = new ServerToAgentMessageHandler();
+
     OpampClient client =
         startOpampClient(
-            effectiveConfig,
-            config.getEndpoint(),
+            opampClientConfiguration,
             resource,
-            config.getPollingInterval(),
+            effectiveConfig,
             new OpampClient.Callbacks() {
               @Override
               public void onConnect(OpampClient opampClient) {
@@ -81,7 +82,8 @@ public void onErrorResponse(
 
               @Override
               public void onMessage(OpampClient opampClient, MessageData messageData) {
-                logger.fine(messageData::toString);
+                logger.fine(() -> "Received message: " + messageData);
+                serverToAgentMessageHandler.handleMessage(messageData, opampClient);
               }
             });
 
@@ -105,14 +107,17 @@ private EffectiveConfigFactory createEffectiveConfigFactory(AutoConfiguredOpenTe
   }
 
   static OpampClient startOpampClient(
-      State.EffectiveConfig effectiveConfig,
-      String endpoint,
+      OpampClientConfiguration opampClientConfiguration,
       Resource resource,
-      long pollingDurationMillis,
+      State.EffectiveConfig effectiveConfig,
       OpampClient.Callbacks callbacks) {
 
     OpampClientBuilder builder = OpampClient.builder();
     builder.enableEffectiveConfigReporting();
+    builder.enableRemoteConfig();
+
+    String endpoint = opampClientConfiguration.getEndpoint();
+    long pollingDurationMillis = opampClientConfiguration.getPollingInterval();
     if (endpoint != null) {
       PeriodicDelay pollingDelay =
           PeriodicDelay.ofFixedDuration(Duration.ofMillis(pollingDurationMillis));
@@ -121,6 +126,7 @@ static OpampClient startOpampClient(
           HttpRequestService.create(okhttp, pollingDelay, DEFAULT_DELAY_BETWEEN_RETRIES);
       builder.setRequestService(httpSender);
     }
+
     OpampAgentAttributes agentAttributes = new OpampAgentAttributes(resource);
     agentAttributes.addIdentifyingAttributes(builder);
     agentAttributes.addNonIdentifyingAttributes(builder);

opamp/src/main/java/com/splunk/opentelemetry/opamp/RemoteConfigProcessor.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright Splunk Inc.
+ *
+ * Licensed 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 com.splunk.opentelemetry.opamp;
+
+import io.opentelemetry.opamp.client.OpampClient;
+import java.util.Map;
+import java.util.Objects;
+import java.util.logging.Logger;
+import opamp.proto.AgentConfigFile;
+import opamp.proto.AgentRemoteConfig;
+import opamp.proto.RemoteConfigStatus;
+import opamp.proto.RemoteConfigStatuses;
+
+public class RemoteConfigProcessor {
+  private static final Logger logger = Logger.getLogger(RemoteConfigProcessor.class.getName());
+
+  private static final String REMOTE_CONFIG_FILE_NAME = "splunk.remote.config";
+
+  public void applyConfig(AgentRemoteConfig remoteConfig, OpampClient opampClient) {
+    Objects.requireNonNull(opampClient);
+
+    Map<String, AgentConfigFile> configMap = remoteConfig.config.config_map;
+    AgentConfigFile configFile = configMap.get(REMOTE_CONFIG_FILE_NAME);
+    if (configFile == null) {
+      logger.warning(
+          "Received server message with remote config, but config is missing '"
+              + REMOTE_CONFIG_FILE_NAME
+              + "' file. Files provided: "
+              + configMap.keySet());
+      return;
+    }
+
+    // TODO: provide implementation
+
+    // Confirm to the OpAMP Server that remote config has been applied.
+    opampClient.setRemoteConfigStatus(
+        new RemoteConfigStatus.Builder()
+            .last_remote_config_hash(remoteConfig.config_hash)
+            .status(RemoteConfigStatuses.RemoteConfigStatuses_APPLIED)
+            .build());
+  }
+}

opamp/src/main/java/com/splunk/opentelemetry/opamp/ServerToAgentMessageHandler.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright Splunk Inc.
+ *
+ * Licensed 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 com.splunk.opentelemetry.opamp;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.opentelemetry.opamp.client.OpampClient;
+import io.opentelemetry.opamp.client.internal.response.MessageData;
+
+public class ServerToAgentMessageHandler {
+  private final RemoteConfigProcessor remoteConfigProcessor;
+
+  public ServerToAgentMessageHandler() {
+    this(new RemoteConfigProcessor());
+  }
+
+  @VisibleForTesting
+  ServerToAgentMessageHandler(RemoteConfigProcessor remoteConfigProcessor) {
+    this.remoteConfigProcessor = remoteConfigProcessor;
+  }
+
+  public void handleMessage(MessageData message, OpampClient opampClient) {
+    if (message.getRemoteConfig() != null) {
+      remoteConfigProcessor.applyConfig(message.getRemoteConfig(), opampClient);
+    }
+  }
+}

opamp/src/test/java/com/splunk/opentelemetry/opamp/OpampActivatorTest.java

@@ -138,12 +138,17 @@ void testOpamp() throws Exception {
         buildEffectiveConfig(new EnvVarsEffectiveConfigFileFactory(config));
 
     CompletableFuture<MessageData> result = new CompletableFuture<>();
+    OpampClientConfiguration opampClientConfiguration =
+        OpampClientConfiguration.builder()
+            .withEnabled(true)
+            .withEndpoint(server.httpUri().toString())
+            .withPollingInterval(500)
+            .build();
     OpampClient client =
         OpampActivator.startOpampClient(
-            effectiveConfig,
-            server.httpUri().toString(),
+            opampClientConfiguration,
             resource,
-            500,
+            effectiveConfig,
             new OpampClient.Callbacks() {
               @Override
               public void onConnect(OpampClient opampClient) {}

opamp/src/test/java/com/splunk/opentelemetry/opamp/RemoteConfigProcessorTest.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright Splunk Inc.
+ *
+ * Licensed 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 com.splunk.opentelemetry.opamp;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+import io.opentelemetry.opamp.client.OpampClient;
+import java.util.Map;
+import okio.ByteString;
+import opamp.proto.AgentConfigFile;
+import opamp.proto.AgentConfigMap;
+import opamp.proto.AgentRemoteConfig;
+import opamp.proto.RemoteConfigStatus;
+import opamp.proto.RemoteConfigStatuses;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+class RemoteConfigProcessorTest {
+  private final RemoteConfigProcessor handler = new RemoteConfigProcessor();
+  private final OpampClient opampClient = org.mockito.Mockito.mock(OpampClient.class);
+
+  @Test
+  void shouldMarkRemoteConfigAsApplied() {
+    // given
+    ByteString configHash = ByteString.encodeUtf8("test-config-hash");
+    AgentRemoteConfig remoteConfig =
+        createRemoteConfig(
+            configHash,
+            Map.of(
+                "splunk.remote.config",
+                new AgentConfigFile.Builder().body(ByteString.encodeUtf8("test-config")).build()));
+
+    // when
+    handler.applyConfig(remoteConfig, opampClient);
+
+    // then
+    ArgumentCaptor<RemoteConfigStatus> statusCaptor =
+        ArgumentCaptor.forClass(RemoteConfigStatus.class);
+    verify(opampClient).setRemoteConfigStatus(statusCaptor.capture());
+    RemoteConfigStatus status = statusCaptor.getValue();
+    assertThat(status.last_remote_config_hash).isEqualTo(configHash);
+    assertThat(status.status).isEqualTo(RemoteConfigStatuses.RemoteConfigStatuses_APPLIED);
+  }
+
+  @Test
+  void shouldIgnoreRemoteConfigWithoutExpectedConfigFile() {
+    // given
+    AgentRemoteConfig remoteConfig =
+        createRemoteConfig(
+            ByteString.encodeUtf8("test-config-hash"),
+            Map.of(
+                "other.config",
+                new AgentConfigFile.Builder().body(ByteString.encodeUtf8("test-config")).build()));
+
+    // when
+    handler.applyConfig(remoteConfig, opampClient);
+
+    // then
+    verify(opampClient, never()).setRemoteConfigStatus(org.mockito.ArgumentMatchers.any());
+  }
+
+  private static AgentRemoteConfig createRemoteConfig(
+      ByteString configHash, Map<String, AgentConfigFile> configMap) {
+    return new AgentRemoteConfig.Builder()
+        .config_hash(configHash)
+        .config(new AgentConfigMap.Builder().config_map(configMap).build())
+        .build();
+  }
+}