diff --git a/opamp/src/main/java/com/splunk/opentelemetry/opamp/OpampActivator.java b/opamp/src/main/java/com/splunk/opentelemetry/opamp/OpampActivator.java index 6299c695a..11d96976d 100644 --- a/opamp/src/main/java/com/splunk/opentelemetry/opamp/OpampActivator.java +++ b/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 diff --git a/opamp/src/main/java/com/splunk/opentelemetry/opamp/ServerToAgentMessageHandler.java b/opamp/src/main/java/com/splunk/opentelemetry/opamp/ServerToAgentMessageHandler.java index 5cbe7a2eb..22efa38f4 100644 --- a/opamp/src/main/java/com/splunk/opentelemetry/opamp/ServerToAgentMessageHandler.java +++ b/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 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 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; } } } diff --git a/profiler/src/main/java/com/splunk/opentelemetry/profiler/ProfilingSupervisor.java b/profiler/src/main/java/com/splunk/opentelemetry/profiler/ProfilingSupervisor.java index 05a0547ac..2b9be0154 100644 --- a/profiler/src/main/java/com/splunk/opentelemetry/profiler/ProfilingSupervisor.java +++ b/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 configSupplier; private final JFR jfr; private final AutoConfiguredOpenTelemetrySdk sdk; - private final BlockingQueue commandQueue; private final PeriodicRecordingFlusherFactory recordingFlusherFactory; private final OtelAllocatedMemoryMetrics allocatedMemoryMetrics; private final OtelGcMemoryMetrics gcMemoryMetrics; @@ -64,14 +58,12 @@ public class ProfilingSupervisor { OptionalConfigurableSupplier configSupplier, JFR jfr, AutoConfiguredOpenTelemetrySdk sdk, - BlockingQueue 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 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 - } } diff --git a/profiler/src/test/java/com/splunk/opentelemetry/profiler/ProfilingSupervisorTest.java b/profiler/src/test/java/com/splunk/opentelemetry/profiler/ProfilingSupervisorTest.java index 657985fad..b02f6e582 100644 --- a/profiler/src/test/java/com/splunk/opentelemetry/profiler/ProfilingSupervisorTest.java +++ b/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 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 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() {