Queue used to process server messages

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

@@ -125,7 +125,7 @@ private static ServerToAgentMessageHandler buildServerToAgentMessageHandler(
         buildRemoteConfigProcessor(effectiveConfigReporter, opampClientConfiguration);
     CommandDispatcher commandDispatcher =
         buildCommandDispatcher(autoConfiguredOpenTelemetrySdk, opampClientConfiguration);
-    return new ServerToAgentMessageHandler(remoteConfigProcessor, commandDispatcher);
+    return ServerToAgentMessageHandler.createAndStart(remoteConfigProcessor, commandDispatcher);
   }
 
   @NotNull

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

@@ -17,26 +17,82 @@
 package com.splunk.opentelemetry.opamp;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.logging.Level.WARNING;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.splunk.opamp.remotecontrol.CommandDispatcher;
+import com.splunk.opentelemetry.profiler.util.HelpfulExecutors;
 import io.opentelemetry.opamp.client.OpampClient;
 import io.opentelemetry.opamp.client.internal.response.MessageData;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.logging.Logger;
 import opamp.proto.AgentConfigFile;
 import opamp.proto.AgentRemoteConfig;
 
 public class ServerToAgentMessageHandler {
   public static final String MAGIC_CMD_STRING = "COMMAND_HACKS";
+
+  private static final Logger logger =
+      Logger.getLogger(ServerToAgentMessageHandler.class.getName());
+  private static final int MAX_MESSAGE_QUEUE_SIZE = 5;
+  private final BlockingQueue<ServerMessage> serverMessageQueue;
   private final RemoteConfigProcessor remoteConfigProcessor;
   private final CommandDispatcher commandDispatcher;
 
-  ServerToAgentMessageHandler(
+  public static ServerToAgentMessageHandler createAndStart(
       RemoteConfigProcessor remoteConfigProcessor, CommandDispatcher commandDispatcher) {
+    ExecutorService executor = HelpfulExecutors.newSingleThreadExecutor("Server Message Handler");
+    ServerToAgentMessageHandler messageHandler =
+        new ServerToAgentMessageHandler(
+            new LinkedBlockingDeque<>(MAX_MESSAGE_QUEUE_SIZE),
+            remoteConfigProcessor,
+            commandDispatcher);
+
+    messageHandler.start(executor);
+
+    return messageHandler;
+  }
+
+  @VisibleForTesting
+  ServerToAgentMessageHandler(
+      BlockingQueue<ServerMessage> serverMessageQueue,
+      RemoteConfigProcessor remoteConfigProcessor,
+      CommandDispatcher commandDispatcher) {
+    this.serverMessageQueue = serverMessageQueue;
     this.remoteConfigProcessor = remoteConfigProcessor;
     this.commandDispatcher = commandDispatcher;
   }
 
+  @VisibleForTesting
+  void start(ExecutorService executor) {
+    executor.submit(this::messageProcessingLoop);
+  }
+
+  private void messageProcessingLoop() {
+    while (true) {
+      try {
+        ServerMessage serverMessage = serverMessageQueue.take();
+        processMessage(serverMessage);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        logger.fine("ServerToAgentMessageHandler is shutting down");
+        return;
+      } catch (Exception e) {
+        logger.log(WARNING, "ServerToAgentMessageHandler encountered an unexpected exception", e);
+      }
+    }
+  }
+
   public void handleMessage(MessageData message, OpampClient opampClient) {
-    AgentRemoteConfig remoteConfig = message.getRemoteConfig();
+    if (!serverMessageQueue.offer(new ServerMessage(message, opampClient))) {
+      logger.severe("Message queue is full. Could not enqueue message " + message);
+    }
+  }
+
+  private void processMessage(ServerMessage serverMessage) {
+    AgentRemoteConfig remoteConfig = serverMessage.messageData.getRemoteConfig();
     if (remoteConfig != null) {
 
       if (remoteConfig.config.config_map.containsKey(MAGIC_CMD_STRING)) {
@@ -49,7 +105,17 @@ public void handleMessage(MessageData message, OpampClient opampClient) {
         }
       }
 
-      remoteConfigProcessor.applyConfig(remoteConfig, opampClient);
+      remoteConfigProcessor.applyConfig(remoteConfig, serverMessage.opampClient);
+    }
+  }
+
+  private static class ServerMessage {
+    private final MessageData messageData;
+    private final OpampClient opampClient;
+
+    ServerMessage(MessageData messageData, OpampClient opampClient) {
+      this.messageData = messageData;
+      this.opampClient = opampClient;
     }
   }
 }

profiler/src/main/java/com/splunk/opentelemetry/profiler/ProfilingSupervisor.java

@@ -17,20 +17,15 @@
 package com.splunk.opentelemetry.profiler;
 
 import static io.opentelemetry.sdk.autoconfigure.AutoConfigureUtil.getResource;
-import static java.util.logging.Level.WARNING;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.splunk.opentelemetry.instrumentation.jvmmetrics.otel.OtelAllocatedMemoryMetrics;
 import com.splunk.opentelemetry.instrumentation.jvmmetrics.otel.OtelGcMemoryMetrics;
-import com.splunk.opentelemetry.profiler.util.HelpfulExecutors;
 import com.splunk.opentelemetry.profiler.util.OptionalConfigurableSupplier;
 import io.opentelemetry.context.ContextStorage;
 import io.opentelemetry.sdk.autoconfigure.AutoConfigureUtil;
 import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk;
 import io.opentelemetry.sdk.resources.Resource;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -49,7 +44,6 @@ public class ProfilingSupervisor {
   private final OptionalConfigurableSupplier<ProfilerConfiguration> configSupplier;
   private final JFR jfr;
   private final AutoConfiguredOpenTelemetrySdk sdk;
-  private final BlockingQueue<ProfilingCommand> commandQueue;
   private final PeriodicRecordingFlusherFactory recordingFlusherFactory;
   private final OtelAllocatedMemoryMetrics allocatedMemoryMetrics;
   private final OtelGcMemoryMetrics gcMemoryMetrics;
@@ -64,14 +58,12 @@ public class ProfilingSupervisor {
       OptionalConfigurableSupplier<ProfilerConfiguration> configSupplier,
       JFR jfr,
       AutoConfiguredOpenTelemetrySdk sdk,
-      BlockingQueue<ProfilingCommand> commandQueue,
       PeriodicRecordingFlusherFactory recordingFlusherFactory,
       OtelAllocatedMemoryMetrics allocatedMemoryMetrics,
       OtelGcMemoryMetrics gcMemoryMetrics) {
     this.configSupplier = configSupplier;
     this.jfr = jfr;
     this.sdk = sdk;
-    this.commandQueue = commandQueue;
     this.recordingFlusherFactory = recordingFlusherFactory;
     this.allocatedMemoryMetrics = allocatedMemoryMetrics;
     this.gcMemoryMetrics = gcMemoryMetrics;
@@ -81,82 +73,21 @@ static ProfilingSupervisor createAndStart(AutoConfiguredOpenTelemetrySdk sdk) {
     if (SUPPLIER.isConfigured()) {
       throw new IllegalStateException("Already started");
     }
-    ExecutorService executor = HelpfulExecutors.newSingleThreadExecutor("JFR Profiler");
-    BlockingQueue<ProfilingCommand> queue = new LinkedBlockingQueue<>();
     ProfilingSupervisor supervisor =
         new ProfilingSupervisor(
             ProfilerConfiguration.SUPPLIER,
             JFR.getInstance(),
             sdk,
-            queue,
             new PeriodicRecordingFlusherFactory(),
             new OtelAllocatedMemoryMetrics(),
             new OtelGcMemoryMetrics());
     SUPPLIER.configure(supervisor);
-    supervisor.start(executor);
     supervisor.updateJvmMemoryMetrics();
 
     return supervisor;
   }
 
-  @VisibleForTesting
-  void start(ExecutorService executor) {
-    executor.submit(this::commandLoop);
-  }
-
-  private void commandLoop() {
-    while (true) {
-      try {
-        ProfilingCommand command = commandQueue.take();
-        handleCommand(command);
-      } catch (InterruptedException e) {
-        Thread.currentThread().interrupt();
-        logger.fine("ProfilingSupervisor is shutting down");
-        return;
-      } catch (Exception e) {
-        logger.log(WARNING, "ProfilingSupervisor encountered an unexpected exception", e);
-      }
-    }
-  }
-
-  public void requestStartProfiling() {
-    commandQueue.add(ProfilingCommand.START);
-  }
-
-  public void requestStopProfiling() {
-    commandQueue.add(ProfilingCommand.STOP);
-  }
-
-  public void requestReinitializeProfiling() {
-    commandQueue.add(ProfilingCommand.REINITIALIZE);
-  }
-
-  private void handleCommand(ProfilingCommand command) {
-    switch (command) {
-      case START:
-        tryStart();
-        break;
-      case STOP:
-        tryStop();
-        break;
-      case REINITIALIZE:
-        tryReinitialize();
-        break;
-    }
-  }
-
-  private void setJfrContextStorageEnabled(boolean enabled) {
-    JfrContextStorage contextStorage = jfrContextStorage.get();
-    if (contextStorage != null) {
-      contextStorage.setEnabled(enabled);
-    }
-  }
-
-  /**
-   * Try and start the profiler. This does not check configuration, just responds to a command
-   * request.
-   */
-  private void tryStart() {
+  public synchronized void requestStartProfiling() {
     if (isJfrRecordingActive()) {
       logger.fine("JFR is already running, not starting again.");
       return;
@@ -175,7 +106,7 @@ private void tryStart() {
     logger.info("Profiler is active.");
   }
 
-  private void tryStop() {
+  public synchronized void requestStopProfiling() {
     if (!isJfrRecordingActive()) {
       logger.fine("JFR is not running already, not stopping again.");
       return;
@@ -185,12 +116,19 @@ private void tryStop() {
     logger.info("Profiler is deactivated.");
   }
 
-  private void tryReinitialize() {
+  public synchronized void requestReinitializeProfiling() {
     updateJvmMemoryMetrics();
-    tryStop();
+    requestStopProfiling();
     // Start the profiler with current settings if it is enabled. New settings will be applied.
     if (configSupplier.get().isEnabled()) {
-      tryStart();
+      requestStartProfiling();
+    }
+  }
+
+  private void setJfrContextStorageEnabled(boolean enabled) {
+    JfrContextStorage contextStorage = jfrContextStorage.get();
+    if (contextStorage != null) {
+      contextStorage.setEnabled(enabled);
     }
   }
 
@@ -241,10 +179,4 @@ static void setupJfrContextStorage() {
           return storage;
         });
   }
-
-  enum ProfilingCommand {
-    START,
-    STOP,
-    REINITIALIZE
-  }
 }

profiler/src/test/java/com/splunk/opentelemetry/profiler/ProfilingSupervisorTest.java

@@ -35,10 +35,6 @@
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
-import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -59,7 +55,6 @@ class ProfilingSupervisorTest {
 
   ProfilerConfiguration config;
   OptionalConfigurableSupplier<ProfilerConfiguration> configSupplier;
-  ExecutorService executor;
   ProfilingSupervisor supervisor;
 
   @BeforeEach
@@ -80,13 +75,6 @@ void setUp(@TempDir Path tempDir) {
         .thenReturn(recordingFlusher);
   }
 
-  @AfterEach
-  void tearDown() {
-    if (executor != null) {
-      executor.shutdownNow();
-    }
-  }
-
   @Test
   void requestStartProfiling_doesNotStartProfilerWhenJfrIsUnavailable() {
     // given
@@ -330,17 +318,14 @@ private AutoConfiguredOpenTelemetrySdk createSdk(Map<String, String> properties)
   }
 
   private void startSupervisor(AutoConfiguredOpenTelemetrySdk sdk) {
-    executor = Executors.newSingleThreadExecutor();
     supervisor =
         new ProfilingSupervisor(
             configSupplier,
             jfr,
             sdk,
-            new LinkedBlockingQueue<>(),
             recordingFlusherFactory,
             allocatedMemoryMetrics,
             gcMemoryMetrics);
-    supervisor.start(executor);
   }
 
   private void startSupervisor() {