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 046700f6d..6299c695a 100644 --- a/opamp/src/main/java/com/splunk/opentelemetry/opamp/OpampActivator.java +++ b/opamp/src/main/java/com/splunk/opentelemetry/opamp/OpampActivator.java @@ -33,6 +33,7 @@ import com.splunk.opentelemetry.profiler.ProfilingDataType; import com.splunk.opentelemetry.profiler.ProfilingSupervisor; import com.splunk.opentelemetry.profiler.exporter.PprofLogDataExporter; +import com.splunk.opentelemetry.profiler.snapshot.SnapshotProfilingSupervisor; import io.opentelemetry.javaagent.extension.AgentListener; import io.opentelemetry.opamp.client.OpampClient; import io.opentelemetry.opamp.client.OpampClientBuilder; @@ -152,7 +153,9 @@ private static RemoteConfigProcessor buildRemoteConfigProcessor( OpampClientConfiguration opampClientConfiguration) { if (opampClientConfiguration.isRemoteConfigurationEnabled()) { return new RemoteConfigProcessorImpl( - ProfilingSupervisor.SUPPLIER.get(), effectiveConfigReporter); + ProfilingSupervisor.SUPPLIER.get(), + SnapshotProfilingSupervisor.SUPPLIER.get(), + effectiveConfigReporter); } return RemoteConfigProcessor.NOOP; } diff --git a/opamp/src/main/java/com/splunk/opentelemetry/opamp/RemoteConfigProcessorImpl.java b/opamp/src/main/java/com/splunk/opentelemetry/opamp/RemoteConfigProcessorImpl.java index 340a7ad57..d4996a815 100644 --- a/opamp/src/main/java/com/splunk/opentelemetry/opamp/RemoteConfigProcessorImpl.java +++ b/opamp/src/main/java/com/splunk/opentelemetry/opamp/RemoteConfigProcessorImpl.java @@ -23,6 +23,9 @@ import com.splunk.opentelemetry.profiler.ProfilerConfiguration; import com.splunk.opentelemetry.profiler.ProfilerDeclarativeConfigurationFactory; import com.splunk.opentelemetry.profiler.ProfilingSupervisor; +import com.splunk.opentelemetry.profiler.snapshot.SnapshotProfilingConfiguration; +import com.splunk.opentelemetry.profiler.snapshot.SnapshotProfilingDeclarativeConfigurationFactory; +import com.splunk.opentelemetry.profiler.snapshot.SnapshotProfilingSupervisor; import io.opentelemetry.api.incubator.config.DeclarativeConfigProperties; import io.opentelemetry.opamp.client.OpampClient; import io.opentelemetry.sdk.autoconfigure.declarativeconfig.DeclarativeConfiguration; @@ -44,15 +47,18 @@ public class RemoteConfigProcessorImpl implements RemoteConfigProcessor { private static final String PROFILING_NODE_NAME = "profiling"; private final ProfilingSupervisor profilingSupervisor; + private final SnapshotProfilingSupervisor snapshotProfilingSupervisor; private final EffectiveConfigReporter effectiveConfigReporter; public RemoteConfigProcessorImpl( - ProfilingSupervisor profilingSupervisor, EffectiveConfigReporter effectiveConfigReporter) { + ProfilingSupervisor profilingSupervisor, + SnapshotProfilingSupervisor snapshotProfilingSupervisor, + EffectiveConfigReporter effectiveConfigReporter) { this.profilingSupervisor = Objects.requireNonNull(profilingSupervisor); + this.snapshotProfilingSupervisor = Objects.requireNonNull(snapshotProfilingSupervisor); this.effectiveConfigReporter = Objects.requireNonNull(effectiveConfigReporter); } - @Override public void applyConfig(AgentRemoteConfig remoteConfig, OpampClient opampClient) { Objects.requireNonNull(opampClient); @@ -77,22 +83,8 @@ public void applyConfig(AgentRemoteConfig remoteConfig, OpampClient opampClient) // Update profiler configuration only when profiling node exists if (distributionRemoteConfigProperties.getPropertyKeys().contains(PROFILING_NODE_NAME)) { - ProfilerConfiguration receivedProfilerConfig = - ProfilerDeclarativeConfigurationFactory.create( - distributionRemoteConfigProperties.getStructured(PROFILING_NODE_NAME, empty())); - - ProfilerConfiguration currentProfilerConfiguration = ProfilerConfiguration.SUPPLIER.get(); - ProfilerConfiguration updatedProfilerConfig = - currentProfilerConfiguration.toBuilder() - .setEnabled(receivedProfilerConfig.isEnabled()) - .setCallStackInterval(receivedProfilerConfig.getCallStackInterval()) - .setMemoryEnabled(receivedProfilerConfig.getMemoryEnabled()) - .build(); - - if (!currentProfilerConfiguration.equals(updatedProfilerConfig)) { - ProfilerConfiguration.SUPPLIER.configure(updatedProfilerConfig); - profilingSupervisor.requestReinitializeProfiling(); - } + applyAlwaysOnProfilingConfiguration(distributionRemoteConfigProperties); + applySnapshotProfilingConfiguration(distributionRemoteConfigProperties); } // Confirm to the OpAMP Server that remote config has been applied. @@ -115,6 +107,43 @@ public void applyConfig(AgentRemoteConfig remoteConfig, OpampClient opampClient) effectiveConfigReporter.reportEffectiveConfigIfChanged(); } + private void applyAlwaysOnProfilingConfiguration( + DeclarativeConfigProperties distributionRemoteConfigProperties) { + ProfilerConfiguration receivedConfiguration = + ProfilerDeclarativeConfigurationFactory.create( + distributionRemoteConfigProperties.getStructured(PROFILING_NODE_NAME, empty())); + + ProfilerConfiguration currentConfiguration = ProfilerConfiguration.SUPPLIER.get(); + ProfilerConfiguration updatedConfiguration = + currentConfiguration.toBuilder() + .setEnabled(receivedConfiguration.isEnabled()) + .setCallStackInterval(receivedConfiguration.getCallStackInterval()) + .setMemoryEnabled(receivedConfiguration.getMemoryEnabled()) + .build(); + + if (!currentConfiguration.equals(updatedConfiguration)) { + ProfilerConfiguration.SUPPLIER.configure(updatedConfiguration); + profilingSupervisor.requestReinitializeProfiling(); + } + } + + private void applySnapshotProfilingConfiguration( + DeclarativeConfigProperties distributionRemoteConfigProperties) { + SnapshotProfilingConfiguration receivedConfiguration = + SnapshotProfilingDeclarativeConfigurationFactory.create( + distributionRemoteConfigProperties.getStructured(PROFILING_NODE_NAME, empty())); + + SnapshotProfilingConfiguration currentConfiguration = + SnapshotProfilingConfiguration.SUPPLIER.get(); + SnapshotProfilingConfiguration updatedConfiguration = + currentConfiguration.toBuilder().setEnabled(receivedConfiguration.isEnabled()).build(); + + if (!currentConfiguration.equals(updatedConfiguration)) { + SnapshotProfilingConfiguration.SUPPLIER.configure(updatedConfiguration); + snapshotProfilingSupervisor.requestReinitializeProfiling(); + } + } + @VisibleForTesting static DeclarativeConfigProperties toDeclarativeConfigProperties(AgentConfigFile configFile) { return DeclarativeConfiguration.toConfigProperties( diff --git a/opamp/src/test/java/com/splunk/opentelemetry/opamp/RemoteConfigProcessorImplTest.java b/opamp/src/test/java/com/splunk/opentelemetry/opamp/RemoteConfigProcessorImplTest.java index 26572be00..69551afc6 100644 --- a/opamp/src/test/java/com/splunk/opentelemetry/opamp/RemoteConfigProcessorImplTest.java +++ b/opamp/src/test/java/com/splunk/opentelemetry/opamp/RemoteConfigProcessorImplTest.java @@ -25,6 +25,8 @@ import com.splunk.opentelemetry.opamp.effectiveconfig.EffectiveConfigReporter; import com.splunk.opentelemetry.profiler.ProfilerConfiguration; import com.splunk.opentelemetry.profiler.ProfilingSupervisor; +import com.splunk.opentelemetry.profiler.snapshot.SnapshotProfilingConfiguration; +import com.splunk.opentelemetry.profiler.snapshot.SnapshotProfilingSupervisor; import io.opentelemetry.opamp.client.OpampClient; import java.util.Map; import okio.ByteString; @@ -35,6 +37,7 @@ import opamp.proto.RemoteConfigStatuses; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.ArgumentCaptor; @@ -44,6 +47,7 @@ @ExtendWith(MockitoExtension.class) class RemoteConfigProcessorImplTest { @Mock ProfilingSupervisor profilingSupervisor; + @Mock SnapshotProfilingSupervisor snapshotProfilingSupervisor; @Mock EffectiveConfigReporter effectiveConfigReporter; @Mock OpampClient opampClient; private RemoteConfigProcessorImpl handler; @@ -51,12 +55,17 @@ class RemoteConfigProcessorImplTest { @BeforeEach void setUp() { ProfilerConfiguration.SUPPLIER.configure(ProfilerConfiguration.builder().build()); - handler = new RemoteConfigProcessorImpl(profilingSupervisor, effectiveConfigReporter); + SnapshotProfilingConfiguration.SUPPLIER.configure( + SnapshotProfilingConfiguration.builder().build()); + handler = + new RemoteConfigProcessorImpl( + profilingSupervisor, snapshotProfilingSupervisor, effectiveConfigReporter); } @AfterEach void tearDown() { ProfilerConfiguration.SUPPLIER.reset(); + SnapshotProfilingConfiguration.SUPPLIER.reset(); } @Test @@ -83,35 +92,7 @@ void shouldMarkRemoteConfigAsAppliedWhenProfilingConfigIsNotProvided() { assertThat(status.error_message).isEmpty(); assertThat(ProfilerConfiguration.SUPPLIER.get().isEnabled()).isFalse(); verify(effectiveConfigReporter).reportEffectiveConfigIfChanged(); - verifyNoInteractions(profilingSupervisor); - } - - @Test - void shouldStartProfilingWhenRemoteConfigEnablesProfiler() { - // given - String remoteConfigYaml = - """ - distribution: - splunk: - profiling: - always_on: - cpu_profiler: - """; - ByteString configHash = ByteString.encodeUtf8("test-config-hash"); - AgentRemoteConfig remoteConfig = createRemoteConfig(configHash, remoteConfigYaml); - - // when - handler.applyConfig(remoteConfig, opampClient); - - // then - RemoteConfigStatus status = getReportedRemoteConfigStatus(); - assertThat(status.last_remote_config_hash).isEqualTo(configHash); - assertThat(status.status).isEqualTo(RemoteConfigStatuses.RemoteConfigStatuses_APPLIED); - assertThat(status.error_message).isEmpty(); - assertThat(ProfilerConfiguration.SUPPLIER.get().isEnabled()).isTrue(); - verify(profilingSupervisor).requestReinitializeProfiling(); - verifyNoMoreInteractions(profilingSupervisor); - verify(effectiveConfigReporter).reportEffectiveConfigIfChanged(); + verifyNoInteractions(profilingSupervisor, snapshotProfilingSupervisor); } @Test @@ -138,67 +119,7 @@ void shouldReportErrorWhenRemoteConfigProcessingFailed() { assertThat(ProfilerConfiguration.SUPPLIER.get().isEnabled()).isFalse(); assertThat(ProfilerConfiguration.SUPPLIER.get().getCallStackInterval().toMillis()) .isEqualTo(10000); - verifyNoInteractions(profilingSupervisor); - verify(effectiveConfigReporter).reportEffectiveConfigIfChanged(); - } - - @Test - void shouldUpdateProfilerConfigWhileRunning() { - // given - ProfilerConfiguration.SUPPLIER.configure( - ProfilerConfiguration.builder().setEnabled(true).build()); - - String remoteConfigYaml = - """ - distribution: - splunk: - profiling: - always_on: - cpu_profiler: - sampling_interval: 123 - memory_profiler: - """; - ByteString configHash = ByteString.encodeUtf8("test-config-hash"); - AgentRemoteConfig remoteConfig = createRemoteConfig(configHash, remoteConfigYaml); - - // when - handler.applyConfig(remoteConfig, opampClient); - - // then - assertThat(ProfilerConfiguration.SUPPLIER.get().isEnabled()).isTrue(); - assertThat(ProfilerConfiguration.SUPPLIER.get().getCallStackInterval().toMillis()) - .isEqualTo(123); - assertThat(ProfilerConfiguration.SUPPLIER.get().getMemoryEnabled()).isTrue(); - verify(profilingSupervisor).requestReinitializeProfiling(); - verifyNoMoreInteractions(profilingSupervisor); - verify(effectiveConfigReporter).reportEffectiveConfigIfChanged(); - } - - @Test - void shouldStopProfilingWhenRemoteConfigDisablesProfiler() { - // given - ProfilerConfiguration.SUPPLIER.configure( - ProfilerConfiguration.builder().setEnabled(true).build()); - String remoteConfigYaml = - """ - distribution: - splunk: - profiling: - """; - ByteString configHash = ByteString.encodeUtf8("test-config-hash"); - AgentRemoteConfig remoteConfig = createRemoteConfig(configHash, remoteConfigYaml); - - // when - handler.applyConfig(remoteConfig, opampClient); - - // then - RemoteConfigStatus status = getReportedRemoteConfigStatus(); - assertThat(status.last_remote_config_hash).isEqualTo(configHash); - assertThat(status.status).isEqualTo(RemoteConfigStatuses.RemoteConfigStatuses_APPLIED); - assertThat(status.error_message).isEmpty(); - assertThat(ProfilerConfiguration.SUPPLIER.get().isEnabled()).isFalse(); - verify(profilingSupervisor).requestReinitializeProfiling(); - verify(profilingSupervisor, never()).requestStartProfiling(); + verifyNoInteractions(profilingSupervisor, snapshotProfilingSupervisor); verify(effectiveConfigReporter).reportEffectiveConfigIfChanged(); } @@ -217,10 +138,188 @@ void shouldIgnoreRemoteConfigWithoutExpectedConfigFile() { // then assertThat(ProfilerConfiguration.SUPPLIER.get().isEnabled()).isFalse(); - verifyNoInteractions(opampClient, profilingSupervisor); + verifyNoInteractions(opampClient, profilingSupervisor, snapshotProfilingSupervisor); verify(effectiveConfigReporter, never()).reportEffectiveConfigIfChanged(); } + @Nested + class AlwaysOnProfilerTest { + @Test + void shouldStartProfilingWhenRemoteConfigEnablesProfiler() { + // given + String remoteConfigYaml = + """ + distribution: + splunk: + profiling: + always_on: + cpu_profiler: + """; + ByteString configHash = ByteString.encodeUtf8("test-config-hash"); + AgentRemoteConfig remoteConfig = createRemoteConfig(configHash, remoteConfigYaml); + + // when + handler.applyConfig(remoteConfig, opampClient); + + // then + RemoteConfigStatus status = getReportedRemoteConfigStatus(); + assertThat(status.last_remote_config_hash).isEqualTo(configHash); + assertThat(status.status).isEqualTo(RemoteConfigStatuses.RemoteConfigStatuses_APPLIED); + assertThat(status.error_message).isEmpty(); + assertThat(ProfilerConfiguration.SUPPLIER.get().isEnabled()).isTrue(); + verify(profilingSupervisor).requestReinitializeProfiling(); + verifyNoMoreInteractions(profilingSupervisor); + verifyNoInteractions(snapshotProfilingSupervisor); + verify(effectiveConfigReporter).reportEffectiveConfigIfChanged(); + } + + @Test + void shouldUpdateProfilerConfigWhileRunning() { + // given + ProfilerConfiguration.SUPPLIER.configure( + ProfilerConfiguration.builder().setEnabled(true).build()); + + String remoteConfigYaml = + """ + distribution: + splunk: + profiling: + always_on: + cpu_profiler: + sampling_interval: 123 + memory_profiler: + """; + ByteString configHash = ByteString.encodeUtf8("test-config-hash"); + AgentRemoteConfig remoteConfig = createRemoteConfig(configHash, remoteConfigYaml); + + // when + handler.applyConfig(remoteConfig, opampClient); + + // then + assertThat(ProfilerConfiguration.SUPPLIER.get().isEnabled()).isTrue(); + assertThat(ProfilerConfiguration.SUPPLIER.get().getCallStackInterval().toMillis()) + .isEqualTo(123); + assertThat(ProfilerConfiguration.SUPPLIER.get().getMemoryEnabled()).isTrue(); + verify(profilingSupervisor).requestReinitializeProfiling(); + verifyNoMoreInteractions(profilingSupervisor); + verifyNoInteractions(snapshotProfilingSupervisor); + verify(effectiveConfigReporter).reportEffectiveConfigIfChanged(); + } + + @Test + void shouldStopProfilingWhenRemoteConfigDisablesProfiler() { + // given + ProfilerConfiguration.SUPPLIER.configure( + ProfilerConfiguration.builder().setEnabled(true).build()); + String remoteConfigYaml = + """ + distribution: + splunk: + profiling: + """; + ByteString configHash = ByteString.encodeUtf8("test-config-hash"); + AgentRemoteConfig remoteConfig = createRemoteConfig(configHash, remoteConfigYaml); + + // when + handler.applyConfig(remoteConfig, opampClient); + + // then + RemoteConfigStatus status = getReportedRemoteConfigStatus(); + assertThat(status.last_remote_config_hash).isEqualTo(configHash); + assertThat(status.status).isEqualTo(RemoteConfigStatuses.RemoteConfigStatuses_APPLIED); + assertThat(status.error_message).isEmpty(); + assertThat(ProfilerConfiguration.SUPPLIER.get().isEnabled()).isFalse(); + verify(profilingSupervisor).requestReinitializeProfiling(); + verify(profilingSupervisor, never()).requestStartProfiling(); + verifyNoInteractions(snapshotProfilingSupervisor); + verify(effectiveConfigReporter).reportEffectiveConfigIfChanged(); + } + } + + @Nested + class SnapshotProfilingSupervisorTest { + @Test + void shouldStartProfilingWhenRemoteConfigEnablesProfiler() { + // given + String remoteConfigYaml = + """ + distribution: + splunk: + profiling: + callgraphs: + """; + ByteString configHash = ByteString.encodeUtf8("test-config-hash"); + AgentRemoteConfig remoteConfig = createRemoteConfig(configHash, remoteConfigYaml); + + // when + handler.applyConfig(remoteConfig, opampClient); + + // then + RemoteConfigStatus status = getReportedRemoteConfigStatus(); + assertThat(status.last_remote_config_hash).isEqualTo(configHash); + assertThat(status.status).isEqualTo(RemoteConfigStatuses.RemoteConfigStatuses_APPLIED); + assertThat(status.error_message).isEmpty(); + assertThat(SnapshotProfilingConfiguration.SUPPLIER.get().isEnabled()).isTrue(); + verify(snapshotProfilingSupervisor).requestReinitializeProfiling(); + verifyNoMoreInteractions(snapshotProfilingSupervisor); + verifyNoInteractions(profilingSupervisor); + verify(effectiveConfigReporter).reportEffectiveConfigIfChanged(); + } + + @Test + void shouldNotReinitializeProfilingWhenConfigurationDoesNotChange() { + // given + SnapshotProfilingConfiguration.SUPPLIER.configure( + SnapshotProfilingConfiguration.builder().setEnabled(true).build()); + String remoteConfigYaml = + """ + distribution: + splunk: + profiling: + callgraphs: + """; + ByteString configHash = ByteString.encodeUtf8("test-config-hash"); + AgentRemoteConfig remoteConfig = createRemoteConfig(configHash, remoteConfigYaml); + + // when + handler.applyConfig(remoteConfig, opampClient); + + // then + assertThat(SnapshotProfilingConfiguration.SUPPLIER.get().isEnabled()).isTrue(); + verifyNoInteractions(profilingSupervisor, snapshotProfilingSupervisor); + verify(effectiveConfigReporter).reportEffectiveConfigIfChanged(); + } + + @Test + void shouldStopProfilingWhenRemoteConfigDisablesProfiler() { + // given + SnapshotProfilingConfiguration.SUPPLIER.configure( + SnapshotProfilingConfiguration.builder().setEnabled(true).build()); + String remoteConfigYaml = + """ + distribution: + splunk: + profiling: + """; + ByteString configHash = ByteString.encodeUtf8("test-config-hash"); + AgentRemoteConfig remoteConfig = createRemoteConfig(configHash, remoteConfigYaml); + + // when + handler.applyConfig(remoteConfig, opampClient); + + // then + RemoteConfigStatus status = getReportedRemoteConfigStatus(); + assertThat(status.last_remote_config_hash).isEqualTo(configHash); + assertThat(status.status).isEqualTo(RemoteConfigStatuses.RemoteConfigStatuses_APPLIED); + assertThat(status.error_message).isEmpty(); + assertThat(SnapshotProfilingConfiguration.SUPPLIER.get().isEnabled()).isFalse(); + verify(snapshotProfilingSupervisor).requestReinitializeProfiling(); + verifyNoMoreInteractions(snapshotProfilingSupervisor); + verifyNoInteractions(profilingSupervisor); + verify(effectiveConfigReporter).reportEffectiveConfigIfChanged(); + } + } + private RemoteConfigStatus getReportedRemoteConfigStatus() { ArgumentCaptor statusCaptor = ArgumentCaptor.forClass(RemoteConfigStatus.class); 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..98646ab5b 100644 --- a/profiler/src/main/java/com/splunk/opentelemetry/profiler/ProfilingSupervisor.java +++ b/profiler/src/main/java/com/splunk/opentelemetry/profiler/ProfilingSupervisor.java @@ -49,7 +49,7 @@ public class ProfilingSupervisor { private final OptionalConfigurableSupplier configSupplier; private final JFR jfr; private final AutoConfiguredOpenTelemetrySdk sdk; - private final BlockingQueue commandQueue; + private final BlockingQueue commandQueue = new LinkedBlockingQueue<>(); private final PeriodicRecordingFlusherFactory recordingFlusherFactory; private final OtelAllocatedMemoryMetrics allocatedMemoryMetrics; private final OtelGcMemoryMetrics gcMemoryMetrics; @@ -64,14 +64,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; @@ -82,13 +80,11 @@ static ProfilingSupervisor createAndStart(AutoConfiguredOpenTelemetrySdk sdk) { 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()); diff --git a/profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingAgentListener.java b/profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingAgentListener.java index f289ca945..5264f364c 100644 --- a/profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingAgentListener.java +++ b/profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingAgentListener.java @@ -45,7 +45,7 @@ public void afterAgent(AutoConfiguredOpenTelemetrySdk sdk) { SnapshotProfilingConfiguration configuration = SnapshotProfilingConfiguration.SUPPLIER.get(); if (configuration.isEnabled()) { - supervisor.startProfiling(); + supervisor.requestStartProfiling(); } } } diff --git a/profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingSupervisor.java b/profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingSupervisor.java index b78992333..a38f33682 100644 --- a/profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingSupervisor.java +++ b/profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingSupervisor.java @@ -16,11 +16,22 @@ package com.splunk.opentelemetry.profiler.snapshot; +import static java.util.logging.Level.WARNING; + import com.google.common.annotations.VisibleForTesting; import com.splunk.opentelemetry.profiler.OtelLoggerFactory; +import com.splunk.opentelemetry.profiler.util.DeclarativeConfigPropertiesUtil; +import com.splunk.opentelemetry.profiler.util.HelpfulExecutors; import com.splunk.opentelemetry.profiler.util.OptionalConfigurableSupplier; +import io.opentelemetry.api.incubator.config.DeclarativeConfigProperties; import io.opentelemetry.sdk.autoconfigure.AutoConfigureUtil; import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; +import io.opentelemetry.sdk.resources.Resource; +import java.time.Duration; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; import java.util.logging.Logger; public class SnapshotProfilingSupervisor { @@ -30,6 +41,10 @@ public class SnapshotProfilingSupervisor { Logger.getLogger(SnapshotProfilingSupervisor.class.getName()); private final OptionalConfigurableSupplier configurationSupplier; + private final BlockingQueue commandQueue = new LinkedBlockingQueue<>(); + private final ConfigurableSupplier stagingAreaSupplier; + private final ConfigurableSupplier stackTraceSamplerSupplier; + private final ConfigurableSupplier stackTraceExporterSupplier; private final ConfigurableSupplier spanTrackerSupplier; private final OptionalConfigurableSupplier traceThreadChangeDetectorSupplier; @@ -37,17 +52,23 @@ public class SnapshotProfilingSupervisor { profilingSpanProcessorSupplier; private final AutoConfiguredOpenTelemetrySdk sdk; private final OtelLoggerFactory otelLoggerFactory; - private boolean running; + private volatile boolean running; @VisibleForTesting SnapshotProfilingSupervisor( OptionalConfigurableSupplier configurationSupplier, + ConfigurableSupplier stagingAreaSupplier, + ConfigurableSupplier stackTraceSamplerSupplier, + ConfigurableSupplier stackTraceExporterSupplier, ConfigurableSupplier spanTrackerSupplier, OptionalConfigurableSupplier traceThreadChangeDetectorSupplier, OptionalConfigurableSupplier profilingSpanProcessorSupplier, AutoConfiguredOpenTelemetrySdk sdk, OtelLoggerFactory otelLoggerFactory) { this.configurationSupplier = configurationSupplier; + this.stagingAreaSupplier = stagingAreaSupplier; + this.stackTraceSamplerSupplier = stackTraceSamplerSupplier; + this.stackTraceExporterSupplier = stackTraceExporterSupplier; this.spanTrackerSupplier = spanTrackerSupplier; this.traceThreadChangeDetectorSupplier = traceThreadChangeDetectorSupplier; this.profilingSpanProcessorSupplier = profilingSpanProcessorSupplier; @@ -60,59 +81,169 @@ public static SnapshotProfilingSupervisor initialize(AutoConfiguredOpenTelemetry throw new IllegalStateException("Snapshot profiling already initialized"); } + ExecutorService executor = + HelpfulExecutors.newSingleThreadExecutor("Snapshot Profiling Supervisor"); SnapshotProfilingSupervisor supervisor = new SnapshotProfilingSupervisor( SnapshotProfilingConfiguration.SUPPLIER, + StagingArea.SUPPLIER, + StackTraceSampler.SUPPLIER, + StackTraceExporter.SUPPLIER, SpanTracker.SUPPLIER, TraceThreadChangeDetector.SUPPLIER, SnapshotProfilingSpanProcessor.SUPPLIER, sdk, new OtelLoggerFactory()); SUPPLIER.configure(supervisor); + supervisor.start(executor); return supervisor; } - public synchronized void startProfiling() { + @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("SnapshotProfilingSupervisor is shutting down"); + return; + } catch (Exception e) { + logger.log(WARNING, "SnapshotProfilingSupervisor 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); + } + + @VisibleForTesting + boolean isRunning() { + return running; + } + + private void handleCommand(ProfilingCommand command) { + switch (command) { + case START: + tryStart(); + break; + case STOP: + tryStop(); + break; + case REINITIALIZE: + tryReinitialize(); + break; + } + } + + private void tryStart() { if (running) { return; } - configurationSupplier.get().log(); + SnapshotProfilingConfiguration configuration = configurationSupplier.get(); + configuration.log(); - StackTraceSamplerInitializer.setupStackTraceSampler(configurationSupplier.get()); - StackTraceSamplerInitializer.setupStackTraceExporter( - configurationSupplier.get(), AutoConfigureUtil.getResource(sdk), otelLoggerFactory); + // Create a new components + stagingAreaSupplier.configure(createStagingArea(configuration)); + stackTraceSamplerSupplier.configure(createStackTraceSampler(configuration)); + stackTraceExporterSupplier.configure(createStackTraceExporter(configuration)); + // Enable components created during SDK initialization spanTrackerSupplier.get().setEnabled(true); traceThreadChangeDetectorSupplier.get().setEnabled(true); - profilingSpanProcessorSupplier.get().setEnabled(true); running = true; logger.info("Snapshot profiling is active."); } - public synchronized void stopProfiling() { + private void tryStop() { if (!running) { return; } - StackTraceSampler.SUPPLIER.get().close(); - StackTraceSampler.SUPPLIER.reset(); + // Dispose components that can be recreated + stackTraceSamplerSupplier.get().close(); + stackTraceSamplerSupplier.reset(); - StagingArea.SUPPLIER.get().close(); - StagingArea.SUPPLIER.reset(); + stagingAreaSupplier.get().close(); + stagingAreaSupplier.reset(); - StackTraceExporter.SUPPLIER.get().close(); - StackTraceExporter.SUPPLIER.reset(); + stackTraceExporterSupplier.get().close(); + stackTraceExporterSupplier.reset(); + // Disable components created during SDK initialization spanTrackerSupplier.get().setEnabled(false); traceThreadChangeDetectorSupplier.get().setEnabled(false); - profilingSpanProcessorSupplier.get().setEnabled(false); running = false; logger.info("Snapshot profiling is deactivated."); } + + private void tryReinitialize() { + if (running) { + tryStop(); + } + + if (configurationSupplier.get().isEnabled()) { + tryStart(); + } + } + + StagingArea createStagingArea(SnapshotProfilingConfiguration configuration) { + Duration interval = configuration.getExportInterval(); + int capacity = configuration.getStagingCapacity(); + return new PeriodicallyExportingStagingArea(stackTraceExporterSupplier, interval, capacity); + } + + StackTraceSampler createStackTraceSampler(SnapshotProfilingConfiguration configuration) { + Duration samplingPeriod = configuration.getSamplingInterval(); + return new PeriodicStackTraceSampler(stagingAreaSupplier, spanTrackerSupplier, samplingPeriod); + } + + StackTraceExporter createStackTraceExporter(SnapshotProfilingConfiguration configuration) { + Resource resource = AutoConfigureUtil.getResource(sdk); + io.opentelemetry.api.logs.Logger otelLogger = + buildLogger(otelLoggerFactory, resource, configuration.getConfigProperties()); + + return new AsyncStackTraceExporter(otelLogger, configuration.getStackDepth()); + } + + private io.opentelemetry.api.logs.Logger buildLogger( + OtelLoggerFactory otelLoggerFactory, Resource resource, Object configProperties) { + if (configProperties instanceof DeclarativeConfigProperties) { + DeclarativeConfigProperties exporterConfig = + DeclarativeConfigPropertiesUtil.getStructuredOrEmpty( + (DeclarativeConfigProperties) configProperties, "exporter"); + return otelLoggerFactory.build(exporterConfig, resource); + } + if (configProperties instanceof ConfigProperties) { + return otelLoggerFactory.build((ConfigProperties) configProperties, resource); + } + throw new IllegalArgumentException( + "Unsupported config properties type: " + configProperties.getClass().getName()); + } + + enum ProfilingCommand { + START, + STOP, + REINITIALIZE + } } diff --git a/profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/StackTraceSamplerInitializer.java b/profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/StackTraceSamplerInitializer.java deleted file mode 100644 index d6ba287a1..000000000 --- a/profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/StackTraceSamplerInitializer.java +++ /dev/null @@ -1,71 +0,0 @@ -/* - * 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.profiler.snapshot; - -import com.splunk.opentelemetry.profiler.OtelLoggerFactory; -import com.splunk.opentelemetry.profiler.util.DeclarativeConfigPropertiesUtil; -import io.opentelemetry.api.incubator.config.DeclarativeConfigProperties; -import io.opentelemetry.api.logs.Logger; -import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; -import io.opentelemetry.sdk.resources.Resource; -import java.time.Duration; - -class StackTraceSamplerInitializer { - private StackTraceSamplerInitializer() {} - - static void setupStackTraceSampler(SnapshotProfilingConfiguration configuration) { - Duration samplingPeriod = configuration.getSamplingInterval(); - StagingArea.SUPPLIER.configure(createStagingArea(configuration)); - - StackTraceSampler sampler = - new PeriodicStackTraceSampler(StagingArea.SUPPLIER, SpanTracker.SUPPLIER, samplingPeriod); - - StackTraceSampler.SUPPLIER.configure(sampler); - } - - static void setupStackTraceExporter( - SnapshotProfilingConfiguration configuration, - Resource resource, - OtelLoggerFactory otelLoggerFactory) { - int maxDepth = configuration.getStackDepth(); - Logger otelLogger = - buildLogger(otelLoggerFactory, resource, configuration.getConfigProperties()); - AsyncStackTraceExporter exporter = new AsyncStackTraceExporter(otelLogger, maxDepth); - StackTraceExporter.SUPPLIER.configure(exporter); - } - - private static StagingArea createStagingArea(SnapshotProfilingConfiguration configuration) { - Duration interval = configuration.getExportInterval(); - int capacity = configuration.getStagingCapacity(); - return new PeriodicallyExportingStagingArea(StackTraceExporter.SUPPLIER, interval, capacity); - } - - private static Logger buildLogger( - OtelLoggerFactory otelLoggerFactory, Resource resource, Object configProperties) { - if (configProperties instanceof DeclarativeConfigProperties) { - DeclarativeConfigProperties exporterConfig = - DeclarativeConfigPropertiesUtil.getStructuredOrEmpty( - (DeclarativeConfigProperties) configProperties, "exporter"); - return otelLoggerFactory.build(exporterConfig, resource); - } - if (configProperties instanceof ConfigProperties) { - return otelLoggerFactory.build((ConfigProperties) configProperties, resource); - } - throw new IllegalArgumentException( - "Unsupported config properties type: " + configProperties.getClass().getName()); - } -} 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..2f40d9216 100644 --- a/profiler/src/test/java/com/splunk/opentelemetry/profiler/ProfilingSupervisorTest.java +++ b/profiler/src/test/java/com/splunk/opentelemetry/profiler/ProfilingSupervisorTest.java @@ -37,7 +37,6 @@ 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; @@ -336,7 +335,6 @@ private void startSupervisor(AutoConfiguredOpenTelemetrySdk sdk) { configSupplier, jfr, sdk, - new LinkedBlockingQueue<>(), recordingFlusherFactory, allocatedMemoryMetrics, gcMemoryMetrics); diff --git a/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingAgentListenerTest.java b/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingAgentListenerTest.java index 6b5398dd9..77c7ce614 100644 --- a/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingAgentListenerTest.java +++ b/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingAgentListenerTest.java @@ -16,8 +16,8 @@ package com.splunk.opentelemetry.profiler.snapshot; +import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertInstanceOf; -import static org.junit.jupiter.api.Assertions.assertNotSame; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; @@ -66,8 +66,8 @@ void configureStackTraceExporterProvider() { new OtelLoggerFactory(() -> logExporter, declarativeConfigProperties -> logExporter)) .afterAgent(sdk); + await().until(() -> StackTraceExporter.SUPPLIER.get() != StackTraceExporter.NOOP); var exporter = StackTraceExporter.SUPPLIER.get(); - assertNotSame(StackTraceExporter.NOOP, exporter); assertInstanceOf(AsyncStackTraceExporter.class, exporter); } @@ -85,8 +85,8 @@ void declarativeConfigWithoutExporterPreservesComponentLoader() { new SnapshotProfilingAgentListener().afterAgent(sdk); + await().until(() -> StackTraceExporter.SUPPLIER.get() != StackTraceExporter.NOOP); var exporter = StackTraceExporter.SUPPLIER.get(); - assertNotSame(StackTraceExporter.NOOP, exporter); assertInstanceOf(AsyncStackTraceExporter.class, exporter); assertTrue(componentLoader.loaded(HttpSenderProvider.class)); } diff --git a/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingSupervisorTest.java b/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingSupervisorTest.java new file mode 100644 index 000000000..8255ac98c --- /dev/null +++ b/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingSupervisorTest.java @@ -0,0 +1,279 @@ +/* + * 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.profiler.snapshot; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.awaitility.Awaitility.await; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.verifyNoMoreInteractions; + +import com.splunk.opentelemetry.profiler.OtelLoggerFactory; +import com.splunk.opentelemetry.profiler.util.OptionalConfigurableSupplier; +import io.opentelemetry.sdk.autoconfigure.AutoConfigureUtil; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; +import io.opentelemetry.sdk.resources.Resource; +import java.time.Duration; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +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; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class SnapshotProfilingSupervisorTest { + private static final Resource RESOURCE = Resource.empty(); + + @Mock AutoConfiguredOpenTelemetrySdk sdk; + @Mock OtelLoggerFactory otelLoggerFactory; + @Mock SpanTracker spanTracker; + @Mock TraceThreadChangeDetector traceThreadChangeDetector; + @Mock SnapshotProfilingSpanProcessor profilingSpanProcessor; + @Mock StackTraceSampler stackTraceSampler; + @Mock StagingArea stagingArea; + @Mock StackTraceExporter stackTraceExporter; + @Mock ConfigProperties configProperties; + + private OptionalConfigurableSupplier configurationSupplier; + private ConfigurableSupplier stackTraceSamplerSupplier; + private ConfigurableSupplier stagingAreaSupplier; + private ConfigurableSupplier stackTraceExporterSupplier; + private ExecutorService executor; + private SnapshotProfilingSupervisor supervisor; + private MockedStatic autoConfigureUtil; + + @BeforeEach + void setUp() { + configurationSupplier = new OptionalConfigurableSupplier<>(); + + ConfigurableSupplier spanTrackerSupplier = new ConfigurableSupplier<>(spanTracker); + OptionalConfigurableSupplier traceThreadChangeDetectorSupplier = + new OptionalConfigurableSupplier<>(); + traceThreadChangeDetectorSupplier.configure(traceThreadChangeDetector); + OptionalConfigurableSupplier profilingSpanProcessorSupplier = + new OptionalConfigurableSupplier<>(); + profilingSpanProcessorSupplier.configure(profilingSpanProcessor); + stackTraceSamplerSupplier = new ConfigurableSupplier<>(StackTraceSampler.NOOP); + stagingAreaSupplier = new ConfigurableSupplier<>(StagingArea.NOOP); + stackTraceExporterSupplier = new ConfigurableSupplier<>(StackTraceExporter.NOOP); + + supervisor = + new SnapshotProfilingSupervisor( + configurationSupplier, + stagingAreaSupplier, + stackTraceSamplerSupplier, + stackTraceExporterSupplier, + spanTrackerSupplier, + traceThreadChangeDetectorSupplier, + profilingSpanProcessorSupplier, + sdk, + otelLoggerFactory); + executor = Executors.newSingleThreadExecutor(); + supervisor.start(executor); + + autoConfigureUtil = mockStatic(AutoConfigureUtil.class); + autoConfigureUtil.when(() -> AutoConfigureUtil.getResource(sdk)).thenReturn(RESOURCE); + } + + @AfterEach + void tearDown() { + supervisor.requestStopProfiling(); + await().untilAsserted(this::assertRuntimeComponentsReset); + executor.shutdownNow(); + autoConfigureUtil.close(); + Snapshotting.resetProfiling(); + } + + @Test + void initializeRegistersSupervisor() { + SnapshotProfilingSupervisor initialized = SnapshotProfilingSupervisor.initialize(sdk); + + assertThat(SnapshotProfilingSupervisor.SUPPLIER.get()).isSameAs(initialized); + assertThatThrownBy(() -> SnapshotProfilingSupervisor.initialize(sdk)) + .isInstanceOf(IllegalStateException.class) + .hasMessage("Snapshot profiling already initialized"); + } + + @Test + void startProfilingOnlyOnce() { + SnapshotProfilingConfiguration configuration = configuration(true); + configurationSupplier.configure(configuration); + + requestStartProfiling(); + StackTraceSampler configuredSampler = stackTraceSamplerSupplier.get(); + StagingArea configuredStagingArea = stagingAreaSupplier.get(); + StackTraceExporter configuredExporter = stackTraceExporterSupplier.get(); + supervisor.requestStartProfiling(); + + await() + .during(Duration.ofMillis(200)) + .untilAsserted( + () -> { + verifyEnabled(true); + assertThat(stackTraceSamplerSupplier.get()).isSameAs(configuredSampler); + assertThat(stagingAreaSupplier.get()).isSameAs(configuredStagingArea); + assertThat(stackTraceExporterSupplier.get()).isSameAs(configuredExporter); + assertRuntimeComponentsConfigured(); + }); + } + + @Test + void stopProfilingOnlyOnce() { + SnapshotProfilingConfiguration configuration = configuration(true); + configurationSupplier.configure(configuration); + requestStartProfiling(); + configureRuntimeComponents(); + + supervisor.requestStopProfiling(); + supervisor.requestStopProfiling(); + + await().untilAsserted(this::verifyClosedRuntimeComponents); + verify(spanTracker).setEnabled(true); + verify(spanTracker).setEnabled(false); + verify(traceThreadChangeDetector).setEnabled(true); + verify(traceThreadChangeDetector).setEnabled(false); + verify(profilingSpanProcessor).setEnabled(true); + verify(profilingSpanProcessor).setEnabled(false); + verifyNoMoreInteractions(spanTracker, traceThreadChangeDetector, profilingSpanProcessor); + assertRuntimeComponentsReset(); + } + + @Test + void doNotStartProfilingWhenReinitializedWithDisabledConfiguration() { + configurationSupplier.configure(configuration(false)); + + supervisor.requestReinitializeProfiling(); + + await() + .during(Duration.ofMillis(200)) + .untilAsserted( + () -> { + verifyNoInteractions(spanTracker, traceThreadChangeDetector, profilingSpanProcessor); + assertRuntimeComponentsReset(); + }); + } + + @Test + void restartProfilingWhenReinitializedWithEnabledConfiguration() { + SnapshotProfilingConfiguration initialConfiguration = configuration(true); + configurationSupplier.configure(initialConfiguration); + requestStartProfiling(); + StackTraceSampler initialSampler = stackTraceSamplerSupplier.get(); + StagingArea initialStagingArea = stagingAreaSupplier.get(); + StackTraceExporter initialExporter = stackTraceExporterSupplier.get(); + configureRuntimeComponents(); + clearInvocations(spanTracker, traceThreadChangeDetector, profilingSpanProcessor); + + SnapshotProfilingConfiguration updatedConfiguration = + configuration(true).toBuilder().setStackDepth(512).build(); + configurationSupplier.configure(updatedConfiguration); + supervisor.requestReinitializeProfiling(); + + await() + .untilAsserted( + () -> { + verifyClosedRuntimeComponents(); + verify(spanTracker).setEnabled(false); + verify(spanTracker).setEnabled(true); + verify(traceThreadChangeDetector).setEnabled(false); + verify(traceThreadChangeDetector).setEnabled(true); + verify(profilingSpanProcessor).setEnabled(false); + verify(profilingSpanProcessor).setEnabled(true); + verifyNoMoreInteractions( + spanTracker, traceThreadChangeDetector, profilingSpanProcessor); + }); + assertThat(stackTraceSamplerSupplier.get()).isNotSameAs(initialSampler); + assertThat(stagingAreaSupplier.get()).isNotSameAs(initialStagingArea); + assertThat(stackTraceExporterSupplier.get()).isNotSameAs(initialExporter); + assertRuntimeComponentsConfigured(); + } + + @Test + void stopProfilingWhenReinitializedWithDisabledConfiguration() { + SnapshotProfilingConfiguration initialConfiguration = configuration(true); + configurationSupplier.configure(initialConfiguration); + requestStartProfiling(); + configureRuntimeComponents(); + clearInvocations(spanTracker, traceThreadChangeDetector, profilingSpanProcessor); + + SnapshotProfilingConfiguration disabledConfiguration = configuration(false); + configurationSupplier.configure(disabledConfiguration); + supervisor.requestReinitializeProfiling(); + + await() + .untilAsserted( + () -> { + verifyClosedRuntimeComponents(); + verifyEnabled(false); + assertRuntimeComponentsReset(); + }); + } + + private void requestStartProfiling() { + supervisor.requestStartProfiling(); + await().untilAsserted(this::assertRuntimeComponentsConfigured); + } + + private void configureRuntimeComponents() { + stackTraceSamplerSupplier.get().close(); + stagingAreaSupplier.get().close(); + stackTraceExporterSupplier.get().close(); + stackTraceSamplerSupplier.configure(stackTraceSampler); + stagingAreaSupplier.configure(stagingArea); + stackTraceExporterSupplier.configure(stackTraceExporter); + } + + private void verifyClosedRuntimeComponents() { + verify(stackTraceSampler).close(); + verify(stagingArea).close(); + verify(stackTraceExporter).close(); + } + + private void verifyEnabled(boolean enabled) { + verify(spanTracker).setEnabled(enabled); + verify(traceThreadChangeDetector).setEnabled(enabled); + verify(profilingSpanProcessor).setEnabled(enabled); + verifyNoMoreInteractions(spanTracker, traceThreadChangeDetector, profilingSpanProcessor); + } + + private void assertRuntimeComponentsConfigured() { + assertThat(stackTraceSamplerSupplier.get()).isInstanceOf(PeriodicStackTraceSampler.class); + assertThat(stagingAreaSupplier.get()).isInstanceOf(PeriodicallyExportingStagingArea.class); + assertThat(stackTraceExporterSupplier.get()).isInstanceOf(AsyncStackTraceExporter.class); + } + + private void assertRuntimeComponentsReset() { + assertThat(stackTraceSamplerSupplier.get()).isSameAs(StackTraceSampler.NOOP); + assertThat(stagingAreaSupplier.get()).isSameAs(StagingArea.NOOP); + assertThat(stackTraceExporterSupplier.get()).isSameAs(StackTraceExporter.NOOP); + } + + private SnapshotProfilingConfiguration configuration(boolean enabled) { + return SnapshotProfilingConfiguration.builder() + .setEnabled(enabled) + .setConfigProperties(configProperties) + .build(); + } +} diff --git a/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/Snapshotting.java b/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/Snapshotting.java index fa0c2fe2f..a42341ef7 100644 --- a/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/Snapshotting.java +++ b/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/Snapshotting.java @@ -16,16 +16,21 @@ package com.splunk.opentelemetry.profiler.snapshot; +import static org.awaitility.Awaitility.await; + import com.splunk.opentelemetry.profiler.OtelLoggerFactory; +import com.splunk.opentelemetry.profiler.util.HelpfulExecutors; import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.SpanContext; import io.opentelemetry.api.trace.TraceFlags; import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk; import io.opentelemetry.sdk.testing.exporter.InMemoryLogRecordExporter; import io.opentelemetry.sdk.trace.IdGenerator; import java.time.Duration; import java.time.Instant; import java.util.Random; +import java.util.concurrent.atomic.AtomicReference; class Snapshotting { private static final Random RANDOM = new Random(); @@ -35,15 +40,34 @@ static SnapshotProfilingSdkCustomizerBuilder customizer() { } static SnapshotProfilingAgentListener agentListener(OtelLoggerFactory otelLoggerFactory) { + AtomicReference supervisorReference = new AtomicReference<>(); return new SnapshotProfilingAgentListener( - sdk -> - new SnapshotProfilingSupervisor( - SnapshotProfilingConfiguration.SUPPLIER, - SpanTracker.SUPPLIER, - TraceThreadChangeDetector.SUPPLIER, - SnapshotProfilingSpanProcessor.SUPPLIER, - sdk, - otelLoggerFactory)); + sdk -> { + SnapshotProfilingSupervisor supervisor = + new SnapshotProfilingSupervisor( + SnapshotProfilingConfiguration.SUPPLIER, + StagingArea.SUPPLIER, + StackTraceSampler.SUPPLIER, + StackTraceExporter.SUPPLIER, + SpanTracker.SUPPLIER, + TraceThreadChangeDetector.SUPPLIER, + SnapshotProfilingSpanProcessor.SUPPLIER, + sdk, + otelLoggerFactory); + supervisorReference.set(supervisor); + supervisor.start( + HelpfulExecutors.newSingleThreadExecutor("Test Snapshot Profiling Supervisor")); + SnapshotProfilingSupervisor.SUPPLIER.configure(supervisor); + return supervisor; + }) { + @Override + public void afterAgent(AutoConfiguredOpenTelemetrySdk sdk) { + super.afterAgent(sdk); + if (SnapshotProfilingConfiguration.SUPPLIER.get().isEnabled()) { + await().until(() -> supervisorReference.get().isRunning()); + } + } + }; } static SnapshotProfilingAgentListener agentListener() { @@ -62,7 +86,8 @@ static void resetProfiling() { SnapshotProfilingConfiguration.SUPPLIER.reset(); if (SnapshotProfilingSupervisor.SUPPLIER.isConfigured()) { - SnapshotProfilingSupervisor.SUPPLIER.get().stopProfiling(); + SnapshotProfilingSupervisor.SUPPLIER.get().requestStopProfiling(); + await().until(() -> !SnapshotProfilingSupervisor.SUPPLIER.get().isRunning()); SnapshotProfilingSupervisor.SUPPLIER.reset(); } diff --git a/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/StackTraceSamplerInitializerTest.java b/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/StackTraceSamplerInitializerTest.java deleted file mode 100644 index 44d2b13cc..000000000 --- a/profiler/src/test/java/com/splunk/opentelemetry/profiler/snapshot/StackTraceSamplerInitializerTest.java +++ /dev/null @@ -1,50 +0,0 @@ -/* - * 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.profiler.snapshot; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Test; - -class StackTraceSamplerInitializerTest { - @AfterEach - void cleanupSuppliers() { - StackTraceSampler.SUPPLIER.get().close(); - StagingArea.SUPPLIER.get().close(); - StackTraceSampler.SUPPLIER.reset(); - StagingArea.SUPPLIER.reset(); - } - - @Test - void setupStackTraceSamplerConfiguresDefaults() { - // given - SnapshotProfilingConfiguration configuration = mock(SnapshotProfilingConfiguration.class); - when(configuration.getStagingCapacity()).thenReturn(15); - - // when - StackTraceSamplerInitializer.setupStackTraceSampler(configuration); - - // then - StackTraceSampler configuredSampler = StackTraceSampler.SUPPLIER.get(); - StagingArea configuredStagingArea = StagingArea.SUPPLIER.get(); - assertThat(configuredSampler).isInstanceOf(PeriodicStackTraceSampler.class); - assertThat(configuredStagingArea).isInstanceOf(PeriodicallyExportingStagingArea.class); - } -}