Move heartbeat to standalone exporter (#2317)

* Move heartbeat to standalone exporter module

* Fix compilation

* Fix spotless

* Refactor

* Rename

* Use a simple mock

* Populate default

* Pass in a function

* Fix micrometer test IndexOutOfBoundsException

* Disable cache in test

* Use Consumer<List<TelemetryItem>> to heartbeat exporter

* Fix tests

* Fix micrometer test

Author: heyams

.github/workflows/pr.yaml

@@ -29,6 +29,7 @@ jobs:
       - name: Test
         uses: gradle/gradle-build-action@v2
         with:
+          cache-disabled: true
           arguments: check assemble
       - name: Upload snapshot
         # only upload from windows since only that build includes etw

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/init/SecondEntryPoint.java

@@ -28,6 +28,8 @@
 import com.azure.monitor.opentelemetry.exporter.implementation.MetricDataMapper;
 import com.azure.monitor.opentelemetry.exporter.implementation.SpanDataMapper;
 import com.azure.monitor.opentelemetry.exporter.implementation.configuration.ConnectionString;
+import com.azure.monitor.opentelemetry.exporter.implementation.heartbeat.HeartbeatExporter;
+import com.azure.monitor.opentelemetry.exporter.implementation.models.TelemetryItem;
 import com.azure.monitor.opentelemetry.exporter.implementation.quickpulse.QuickPulse;
 import com.azure.monitor.opentelemetry.exporter.implementation.utils.Strings;
 import com.azure.monitor.opentelemetry.exporter.implementation.utils.TempDirs;
@@ -41,7 +43,6 @@
 import com.microsoft.applicationinsights.agent.internal.exporter.AgentLogExporter;
 import com.microsoft.applicationinsights.agent.internal.exporter.AgentMetricExporter;
 import com.microsoft.applicationinsights.agent.internal.exporter.AgentSpanExporter;
-import com.microsoft.applicationinsights.agent.internal.heartbeat.HeartBeatProvider;
 import com.microsoft.applicationinsights.agent.internal.httpclient.LazyHttpClient;
 import com.microsoft.applicationinsights.agent.internal.legacyheaders.AiLegacyHeaderSpanProcessor;
 import com.microsoft.applicationinsights.agent.internal.legacyheaders.DelegatingPropagator;
@@ -57,6 +58,7 @@
 import com.microsoft.applicationinsights.agent.internal.telemetry.BatchItemProcessor;
 import com.microsoft.applicationinsights.agent.internal.telemetry.MetricFilter;
 import com.microsoft.applicationinsights.agent.internal.telemetry.TelemetryClient;
+import com.microsoft.applicationinsights.agent.internal.telemetry.TelemetryObservers;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.common.AttributesBuilder;
 import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdkBuilder;
@@ -82,6 +84,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 import org.slf4j.Logger;
@@ -153,7 +156,17 @@ public void customize(AutoConfigurationCustomizer autoConfiguration) {
 
     // interval longer than 15 minutes is not allowed since we use this data for usage telemetry
     long intervalSeconds = Math.min(config.heartbeat.intervalSeconds, MINUTES.toSeconds(15));
-    HeartBeatProvider.start(intervalSeconds, telemetryClient);
+    Consumer<List<TelemetryItem>> telemetryItemsConsumer =
+        telemetryItems -> {
+          for (TelemetryItem telemetryItem : telemetryItems) {
+            TelemetryObservers.INSTANCE
+                .getObservers()
+                .forEach(consumer -> consumer.accept(telemetryItem));
+            telemetryClient.getMetricsBatchItemProcessor().trackAsync(telemetryItem);
+          }
+        };
+    HeartbeatExporter.start(
+        intervalSeconds, telemetryClient::populateDefaults, telemetryItemsConsumer);
 
     TelemetryClient.setActive(telemetryClient);
 

agent/azure-monitor-exporter/src/main/java/com/azure/monitor/opentelemetry/exporter/AzureMonitorExporterBuilder.java

@@ -21,6 +21,8 @@
 
 package com.azure.monitor.opentelemetry.exporter;
 
+import static java.util.concurrent.TimeUnit.MINUTES;
+
 import com.azure.core.credential.TokenCredential;
 import com.azure.core.http.HttpClient;
 import com.azure.core.http.HttpPipeline;
@@ -40,6 +42,7 @@
 import com.azure.monitor.opentelemetry.exporter.implementation.MetricDataMapper;
 import com.azure.monitor.opentelemetry.exporter.implementation.SpanDataMapper;
 import com.azure.monitor.opentelemetry.exporter.implementation.configuration.ConnectionString;
+import com.azure.monitor.opentelemetry.exporter.implementation.heartbeat.HeartbeatExporter;
 import com.azure.monitor.opentelemetry.exporter.implementation.localstorage.LocalStorageStats;
 import com.azure.monitor.opentelemetry.exporter.implementation.localstorage.LocalStorageTelemetryPipelineListener;
 import com.azure.monitor.opentelemetry.exporter.implementation.models.ContextTagKeys;
@@ -267,13 +270,18 @@ public AzureMonitorTraceExporter buildTraceExporter() {
    * Creates an {@link AzureMonitorMetricExporter} based on the options set in the builder. This
    * exporter is an implementation of OpenTelemetry {@link MetricExporter}.
    *
+   * <p>When a new {@link MetricExporter} is created, it will automatically start {@link
+   * HeartbeatExporter}.
+   *
    * @return An instance of {@link AzureMonitorMetricExporter}.
    * @throws NullPointerException if the connection string is not set on this builder or if the
    *     environment variable "APPLICATIONINSIGHTS_CONNECTION_STRING" is not set.
    */
   public AzureMonitorMetricExporter buildMetricExporter() {
+    TelemetryItemExporter telemetryItemExporter = initExporterBuilder();
+    HeartbeatExporter.start(MINUTES.toSeconds(15), t -> {}, telemetryItemExporter::send);
     return new AzureMonitorMetricExporter(
-        new MetricDataMapper(instrumentationKey, t -> {}), initExporterBuilder());
+        new MetricDataMapper(instrumentationKey, t -> {}), telemetryItemExporter);
   }
 
   /**

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/heartbeat/DefaultHeartBeatPropertyProvider.java renamed to agent/azure-monitor-exporter/src/main/java/com/azure/monitor/opentelemetry/exporter/implementation/heartbeat/DefaultHeartBeatPropertyProvider.java

@@ -19,9 +19,9 @@
  * DEALINGS IN THE SOFTWARE.
  */
 
-package com.microsoft.applicationinsights.agent.internal.heartbeat;
+package com.azure.monitor.opentelemetry.exporter.implementation.heartbeat;
 
-import com.microsoft.applicationinsights.agent.internal.common.PropertyHelper;
+import com.azure.monitor.opentelemetry.exporter.implementation.utils.VersionGenerator;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.UUID;
@@ -68,7 +68,7 @@ public DefaultHeartBeatPropertyProvider() {
   }
 
   @Override
-  public Callable<Boolean> setDefaultPayload(HeartBeatProvider provider) {
+  public Callable<Boolean> setDefaultPayload(HeartbeatExporter provider) {
     return new Callable<Boolean>() {
 
       final Set<String> enabledProperties = defaultFields;
@@ -132,7 +132,7 @@ private static String getJreVersion() {
 
   /** Returns the Application Insights SDK version user is using to instrument his application. */
   private static String getSdkVersion() {
-    return PropertyHelper.getQualifiedSdkVersionString();
+    return VersionGenerator.getSdkVersion();
   }
 
   /** Returns the OS version on which application is running. */

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/heartbeat/HeartBeatPayloadProviderInterface.java renamed to agent/azure-monitor-exporter/src/main/java/com/azure/monitor/opentelemetry/exporter/implementation/heartbeat/HeartBeatPayloadProviderInterface.java

@@ -19,7 +19,7 @@
  * DEALINGS IN THE SOFTWARE.
  */
 
-package com.microsoft.applicationinsights.agent.internal.heartbeat;
+package com.azure.monitor.opentelemetry.exporter.implementation.heartbeat;
 
 import java.util.concurrent.Callable;
 
@@ -42,5 +42,5 @@ public interface HeartBeatPayloadProviderInterface {
    * @param provider The current heartbeat provider
    * @return Callable which can be executed to add the payload
    */
-  Callable<Boolean> setDefaultPayload(HeartBeatProvider provider);
+  Callable<Boolean> setDefaultPayload(HeartbeatExporter provider);
 }

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/heartbeat/HeartBeatPropertyPayload.java renamed to agent/azure-monitor-exporter/src/main/java/com/azure/monitor/opentelemetry/exporter/implementation/heartbeat/HeartBeatPropertyPayload.java

@@ -19,7 +19,7 @@
  * DEALINGS IN THE SOFTWARE.
  */
 
-package com.microsoft.applicationinsights.agent.internal.heartbeat;
+package com.azure.monitor.opentelemetry.exporter.implementation.heartbeat;
 
 import javax.annotation.Nullable;
 

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/heartbeat/HeartbeatDefaultPayload.java renamed to agent/azure-monitor-exporter/src/main/java/com/azure/monitor/opentelemetry/exporter/implementation/heartbeat/HeartbeatDefaultPayload.java

@@ -19,7 +19,7 @@
  * DEALINGS IN THE SOFTWARE.
  */
 
-package com.microsoft.applicationinsights.agent.internal.heartbeat;
+package com.azure.monitor.opentelemetry.exporter.implementation.heartbeat;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -43,7 +43,7 @@ public class HeartbeatDefaultPayload {
    * @param provider The HeartBeat provider
    * @return Callable to perform execution
    */
-  public static Callable<Boolean> populateDefaultPayload(HeartBeatProvider provider) {
+  public static Callable<Boolean> populateDefaultPayload(HeartbeatExporter provider) {
     return new Callable<Boolean>() {
       @Override
       public Boolean call() throws Exception {

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/heartbeat/HeartBeatProvider.java renamed to agent/azure-monitor-exporter/src/main/java/com/azure/monitor/opentelemetry/exporter/implementation/heartbeat/HeartbeatExporter.java

@@ -19,28 +19,31 @@
  * DEALINGS IN THE SOFTWARE.
  */
 
-package com.microsoft.applicationinsights.agent.internal.heartbeat;
+package com.azure.monitor.opentelemetry.exporter.implementation.heartbeat;
 
+import com.azure.monitor.opentelemetry.exporter.implementation.builders.AbstractTelemetryBuilder;
 import com.azure.monitor.opentelemetry.exporter.implementation.builders.MetricTelemetryBuilder;
 import com.azure.monitor.opentelemetry.exporter.implementation.models.ContextTagKeys;
 import com.azure.monitor.opentelemetry.exporter.implementation.models.TelemetryItem;
 import com.azure.monitor.opentelemetry.exporter.implementation.utils.ThreadPoolUtils;
-import com.microsoft.applicationinsights.agent.internal.telemetry.TelemetryClient;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /** Concrete implementation of Heartbeat functionality. */
-public class HeartBeatProvider {
+public class HeartbeatExporter {
 
-  private static final Logger logger = LoggerFactory.getLogger(HeartBeatProvider.class);
+  private static final Logger logger = LoggerFactory.getLogger(HeartbeatExporter.class);
 
   /** The name of the heartbeat metric. */
   private static final String HEARTBEAT_SYNTHETIC_METRIC_NAME = "HeartbeatState";
@@ -51,35 +54,42 @@ public class HeartBeatProvider {
   /** Map to hold heartbeat properties. */
   private final ConcurrentMap<String, HeartBeatPropertyPayload> heartbeatProperties;
 
-  /** Telemetry client instance used to send heartbeat. */
-  private TelemetryClient telemetryClient;
+  /** Telemetry item exporter used to send heartbeat. */
+  private final Consumer<List<TelemetryItem>> telemetryItemsConsumer;
+
+  /** Telemetry builder consumer used to populate defaults properties. */
+  private final Consumer<AbstractTelemetryBuilder> telemetryInitializer;
 
   /** ThreadPool used for adding properties to concurrent dictionary. */
   private final ExecutorService propertyUpdateService;
 
   /** Threadpool used to send data heartbeat telemetry. */
   private final ScheduledExecutorService heartBeatSenderService;
 
-  public static void start(long intervalSeconds, TelemetryClient telemetryClient) {
-    new HeartBeatProvider(intervalSeconds, telemetryClient);
+  public static void start(
+      long intervalSeconds,
+      Consumer<AbstractTelemetryBuilder> telemetryInitializer,
+      Consumer<List<TelemetryItem>> telemetryItemsConsumer) {
+    new HeartbeatExporter(intervalSeconds, telemetryInitializer, telemetryItemsConsumer);
   }
 
-  // visible for tests
-  HeartBeatProvider(long intervalSeconds, TelemetryClient telemetryClient) {
+  public HeartbeatExporter(
+      long intervalSeconds,
+      Consumer<AbstractTelemetryBuilder> telemetryInitializer,
+      Consumer<List<TelemetryItem>> telemetryItemsConsumer) {
     this.heartbeatProperties = new ConcurrentHashMap<>();
     this.heartbeatsSent = 0;
     this.propertyUpdateService =
         Executors.newCachedThreadPool(
             ThreadPoolUtils.createDaemonThreadFactory(
-                HeartBeatProvider.class, "propertyUpdateService"));
+                HeartbeatExporter.class, "propertyUpdateService"));
     this.heartBeatSenderService =
         Executors.newSingleThreadScheduledExecutor(
             ThreadPoolUtils.createDaemonThreadFactory(
-                HeartBeatProvider.class, "heartBeatSenderService"));
+                HeartbeatExporter.class, "heartBeatSenderService"));
 
-    if (this.telemetryClient == null) {
-      this.telemetryClient = telemetryClient;
-    }
+    this.telemetryItemsConsumer = telemetryItemsConsumer;
+    this.telemetryInitializer = telemetryInitializer;
 
     // Submit task to set properties to dictionary using separate thread. we do not wait for the
     // results to come out as some I/O bound properties may take time.
@@ -110,7 +120,7 @@ public boolean addHeartBeatProperty(
   /** Send the heartbeat item synchronously to application insights backend. */
   private void send() {
     try {
-      telemetryClient.trackAsync(gatherData());
+      telemetryItemsConsumer.accept(Collections.singletonList(gatherData()));
       logger.trace("No of heartbeats sent, {}", ++heartbeatsSent);
     } catch (RuntimeException e) {
       logger.warn("Error occured while sending heartbeat");
@@ -132,8 +142,8 @@ TelemetryItem gatherData() {
       numHealthy += payload.isHealthy() ? 0 : 1;
     }
     MetricTelemetryBuilder telemetryBuilder =
-        telemetryClient.newMetricTelemetryBuilder(HEARTBEAT_SYNTHETIC_METRIC_NAME, numHealthy);
-
+        MetricTelemetryBuilder.create(HEARTBEAT_SYNTHETIC_METRIC_NAME, numHealthy);
+    telemetryInitializer.accept(telemetryBuilder);
     telemetryBuilder.addTag(
         ContextTagKeys.AI_OPERATION_SYNTHETIC_SOURCE.toString(), HEARTBEAT_SYNTHETIC_METRIC_NAME);
 

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/heartbeat/WebAppsHeartbeatProvider.java renamed to agent/azure-monitor-exporter/src/main/java/com/azure/monitor/opentelemetry/exporter/implementation/heartbeat/WebAppsHeartbeatProvider.java

@@ -19,7 +19,7 @@
  * DEALINGS IN THE SOFTWARE.
  */
 
-package com.microsoft.applicationinsights.agent.internal.heartbeat;
+package com.azure.monitor.opentelemetry.exporter.implementation.heartbeat;
 
 import com.azure.monitor.opentelemetry.exporter.implementation.utils.Strings;
 import java.util.HashSet;
@@ -67,7 +67,7 @@ public WebAppsHeartbeatProvider() {
   }
 
   @Override
-  public Callable<Boolean> setDefaultPayload(HeartBeatProvider provider) {
+  public Callable<Boolean> setDefaultPayload(HeartbeatExporter provider) {
     return new Callable<Boolean>() {
 
       final Set<String> enabledProperties = defaultFields;

agent/agent-tooling/src/test/java/com/microsoft/applicationinsights/agent/internal/heartbeat/HeartbeatTests.java renamed to agent/azure-monitor-exporter/src/test/java/com/azure/monitor/opentelemetry/exporter/implementation/heartbeat/HeartbeatTests.java

@@ -19,26 +19,31 @@
  * DEALINGS IN THE SOFTWARE.
  */
 
-package com.microsoft.applicationinsights.agent.internal.heartbeat;
+package com.azure.monitor.opentelemetry.exporter.implementation.heartbeat;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
 
 import com.azure.monitor.opentelemetry.exporter.implementation.models.MetricsData;
-import com.microsoft.applicationinsights.agent.internal.telemetry.TelemetryClient;
+import com.azure.monitor.opentelemetry.exporter.implementation.models.TelemetryItem;
 import java.lang.reflect.Field;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.function.Consumer;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 import org.mockito.stubbing.Answer;
 
 class HeartbeatTests {
 
+  private final Consumer<List<TelemetryItem>> telemetryItemsConsumer = mock(Consumer.class);
+
   @Test
   void heartBeatPayloadContainsDataByDefault() throws InterruptedException {
     // given
-    HeartBeatProvider provider = new HeartBeatProvider(60, TelemetryClient.createForTest());
+    HeartbeatExporter provider = new HeartbeatExporter(60, t -> {}, telemetryItemsConsumer);
 
     // some of the initialization above happens in a separate thread
     Thread.sleep(500);
@@ -52,7 +57,7 @@ void heartBeatPayloadContainsDataByDefault() throws InterruptedException {
   @Test
   void heartBeatPayloadContainsSpecificProperties() {
     // given
-    HeartBeatProvider provider = new HeartBeatProvider(60, TelemetryClient.createForTest());
+    HeartbeatExporter provider = new HeartbeatExporter(60, t -> {}, telemetryItemsConsumer);
 
     // then
     assertThat(provider.addHeartBeatProperty("test", "testVal", true)).isTrue();
@@ -64,7 +69,7 @@ void heartBeatPayloadContainsSpecificProperties() {
   @Test
   void heartbeatMetricIsNonZeroWhenFailureConditionPresent() {
     // given
-    HeartBeatProvider provider = new HeartBeatProvider(60, TelemetryClient.createForTest());
+    HeartbeatExporter provider = new HeartbeatExporter(60, t -> {}, telemetryItemsConsumer);
 
     // then
     assertThat(provider.addHeartBeatProperty("test", "testVal", false)).isTrue();
@@ -76,7 +81,7 @@ void heartbeatMetricIsNonZeroWhenFailureConditionPresent() {
   @Test
   void heartbeatMetricCountsForAllFailures() {
     // given
-    HeartBeatProvider provider = new HeartBeatProvider(60, TelemetryClient.createForTest());
+    HeartbeatExporter provider = new HeartbeatExporter(60, t -> {}, telemetryItemsConsumer);
 
     // then
     assertThat(provider.addHeartBeatProperty("test", "testVal", false)).isTrue();
@@ -88,7 +93,7 @@ void heartbeatMetricCountsForAllFailures() {
 
   @Test
   void sentHeartbeatContainsExpectedDefaultFields() throws Exception {
-    HeartBeatProvider mockProvider = Mockito.mock(HeartBeatProvider.class);
+    HeartbeatExporter mockProvider = Mockito.mock(HeartbeatExporter.class);
     ConcurrentMap<String, String> props = new ConcurrentHashMap<>();
     Mockito.when(
             mockProvider.addHeartBeatProperty(
@@ -116,7 +121,7 @@ void sentHeartbeatContainsExpectedDefaultFields() throws Exception {
   @Test
   void heartBeatProviderDoesNotAllowDuplicateProperties() {
     // given
-    HeartBeatProvider provider = new HeartBeatProvider(60, TelemetryClient.createForTest());
+    HeartbeatExporter provider = new HeartbeatExporter(60, t -> {}, telemetryItemsConsumer);
 
     // then
     provider.addHeartBeatProperty("test01", "test val", true);
@@ -133,7 +138,7 @@ void cannotAddUnknownDefaultProperty() throws Exception {
     Set<String> defaultFields = (Set<String>) field.get(base);
     defaultFields.add(testKey);
 
-    HeartBeatProvider provider = new HeartBeatProvider(60, TelemetryClient.createForTest());
+    HeartbeatExporter provider = new HeartbeatExporter(60, t -> {}, telemetryItemsConsumer);
 
     base.setDefaultPayload(provider).call();
     MetricsData data = (MetricsData) provider.gatherData().getData().getBaseData();