Azure · FabianMeiswinkel · Sep 26, 2022 · Jul 19, 2022 · Jul 19, 2022 · Jul 19, 2022
@@ -2174,6 +2174,8 @@
       <Class name="com.azure.cosmos.implementation.ImplementationBridgeHelpers$DirectConnectionConfigHelper"/>
       <Class name="com.azure.cosmos.implementation.ImplementationBridgeHelpers$FeedResponseHelper"/>
       <Class name="com.azure.cosmos.implementation.ImplementationBridgeHelpers$PartitionKeyHelper"/>
+      <Class name="com.azure.cosmos.implementation.ImplementationBridgeHelpers$CosmosAsyncClientHelper"/>
+      <Class name="com.azure.cosmos.implementation.ImplementationBridgeHelpers$CosmosClientTelemetryConfigHelper"/>
     </Or>
     <Bug pattern="DM_EXIT"/>
   </Match>

@@ -354,6 +354,7 @@ cosmos_org.scala-lang:scala-library;2.12.10
 cosmos_org.scala-lang.modules:scala-java8-compat_2.12;0.8.0
 cosmos_io.projectreactor:reactor-scala-extensions_2.12;0.8.0
 cosmos_commons-io:commons-io;2.4
+cosmos_com.microsoft.azure:applicationinsights-core;2.6.4
 
 # Cosmos Spark connector tests only
 cosmos_org.scalatest:scalatest_2.12;3.2.2

@@ -76,6 +76,11 @@ abstract class AsyncBenchmark<T> {
     private AtomicBoolean warmupMode = new AtomicBoolean(false);
 
     AsyncBenchmark(Configuration cfg) {
+
+        logger = LoggerFactory.getLogger(this.getClass());
+        configuration = cfg;
+        MeterRegistry registry = configuration.getAzureMonitorMeterRegistry();
+
         CosmosClientBuilder cosmosClientBuilder = new CosmosClientBuilder()
             .endpoint(cfg.getServiceEndpoint())
             .key(cfg.getMasterKey())
@@ -84,6 +89,16 @@ abstract class AsyncBenchmark<T> {
             .contentResponseOnWriteEnabled(cfg.isContentResponseOnWriteEnabled())
             .clientTelemetryEnabled(cfg.isClientTelemetryEnabled());
 
+        if (registry != null) {
+            cosmosClientBuilder.clientTelemetryConfig().clientMetrics(registry);
+        }
+
+        registry = configuration.getGraphiteMeterRegistry();
+
+        if (registry != null) {
+            cosmosClientBuilder.clientTelemetryConfig().clientMetrics(registry);
+        }
+
         if (cfg.getConnectionMode().equals(ConnectionMode.DIRECT)) {
             cosmosClientBuilder = cosmosClientBuilder.directMode(DirectConnectionConfig.getDefaultConfig());
         } else {
@@ -92,8 +107,6 @@ abstract class AsyncBenchmark<T> {
             cosmosClientBuilder = cosmosClientBuilder.gatewayMode(gatewayConnectionConfig);
         }
         cosmosClient = cosmosClientBuilder.buildAsyncClient();
-        configuration = cfg;
-        logger = LoggerFactory.getLogger(this.getClass());
 
         try {
             cosmosAsyncDatabase = cosmosClient.getDatabase(this.configuration.getDatabaseId());
@@ -221,18 +234,6 @@ uuid, new PartitionKey(partitionKey), PojoizedJson.class)
                 .convertRatesTo(TimeUnit.SECONDS)
                 .build();
         }
-
-        MeterRegistry registry = configuration.getAzureMonitorMeterRegistry();
-
-        if (registry != null) {
-            BridgeInternal.monitorTelemetry(registry);
-        }
-
-        registry = configuration.getGraphiteMeterRegistry();
-
-        if (registry != null) {
-            BridgeInternal.monitorTelemetry(registry);
-        }
     }
 
     protected void init() {

@@ -15,6 +15,7 @@
 import io.micrometer.azuremonitor.AzureMonitorMeterRegistry;
 import io.micrometer.core.instrument.Clock;
 import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.config.NamingConvention;
 import io.micrometer.core.lang.Nullable;
 import io.micrometer.graphite.GraphiteConfig;
@@ -574,6 +575,7 @@ private synchronized MeterRegistry azureMonitorMeterRegistry(String instrumentat
         if (this.azureMonitorMeterRegistry == null) {
 
             Duration step = Duration.ofSeconds(Integer.getInteger("azure.cosmos.monitoring.azureMonitor.step", this.printingInterval));
+            String testCategoryTag = System.getProperty("azure.cosmos.monitoring.azureMonitor.testCategory");
             boolean enabled = !Boolean.getBoolean("azure.cosmos.monitoring.azureMonitor.disabled");
 
             final AzureMonitorConfig config = new AzureMonitorConfig() {
@@ -602,6 +604,11 @@ public boolean enabled() {
             };
 
             this.azureMonitorMeterRegistry = new AzureMonitorMeterRegistry(config, Clock.SYSTEM);
+            if (!Strings.isNullOrEmpty(testCategoryTag)) {
+                List<Tag> globalTags = new ArrayList<>();
+                globalTags.add(Tag.of("TestCategory", testCategoryTag));
+                this.azureMonitorMeterRegistry.config().commonTags(globalTags);
+            }
         }
 
         return this.azureMonitorMeterRegistry;

@@ -6,11 +6,11 @@
 import com.azure.cosmos.DirectConnectionConfig;
 import com.azure.cosmos.ThrottlingRetryOptions;
 import com.azure.cosmos.implementation.AsyncDocumentClient;
-import com.azure.cosmos.implementation.ClientTelemetryConfig;
 import com.azure.cosmos.implementation.ConnectionPolicy;
 import com.azure.cosmos.implementation.Database;
 import com.azure.cosmos.implementation.DatabaseForTest;
 import com.azure.cosmos.implementation.DocumentCollection;
+import com.azure.cosmos.implementation.ImplementationBridgeHelpers;
 import com.azure.cosmos.implementation.ResourceResponse;
 import com.azure.cosmos.implementation.TestConfigurations;
 import com.azure.cosmos.models.FeedResponse;
@@ -32,7 +32,11 @@ public static AsyncDocumentClient housekeepingClient() {
                         .withMasterKeyOrResourceToken(TestConfigurations.MASTER_KEY)
                         .withConnectionPolicy(connectionPolicy)
                         .withContentResponseOnWriteEnabled(true)
-                        .withClientTelemetryConfig(ClientTelemetryConfig.getDefaultConfig())
+                        .withClientTelemetryConfig(
+                            ImplementationBridgeHelpers
+                                .CosmosClientTelemetryConfigHelper
+                                .getCosmosClientTelemetryConfigAccessor()
+                                .getDefaultConfig())
                         .build();
     }
 

@@ -3,6 +3,7 @@
 ### 4.14.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added option to emit client-side metrics via micrometer.io MeterRegistry. - See [PR 30065](https://github.com/Azure/azure-sdk-for-java/pull/30065)
 
 #### Breaking Changes
 

@@ -3,6 +3,7 @@
 ### 4.14.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added option to emit client-side metrics via micrometer.io MeterRegistry. - See [PR 30065](https://github.com/Azure/azure-sdk-for-java/pull/30065) 
 
 #### Breaking Changes
 

@@ -0,0 +1,60 @@
+# Cosmos DB client-side metrics
+
+The Cosmos DB SDK for Java allows enabling micrometer.io metrics to track latency, request charge (RU/s), request rates etc. for logical operations (API calls into the SDK from your application/service) as well as actual requests to the Cosmos DB service (for example when due to Consistency level Strong or Bounded Staleness requests need to be sent to multiple replica). These metrics are tagged with different dimensions - so it is possible to look at metrics like latency, request charge etc. for a certain dimension like operation type, response status code etc.
+
+**<u>Please note:</u> There is some performance overhead when enabling metrics due to increased CPU usage. The overhead is small (<10%) but should be tested before enabling it in production especially for large containers/workloads (> 5 TB of data or > 1,000,000 RU/s)** 
+
+More information about the metrics being emitted by the Azure Cosmos DB SDK are available here: https://aka.ms/azure-cosmos-metrics
+
+
+
+## How to enable metrics?
+
+When using the Azure Cosmos DB Spark connector for Spark 3.* you can enable these client-side metrics and emit them to the Spark metric system (ganglia plus optionally in Log4j log files) as well as to Azure Monitor / ApplicationInsights simply via configuration.
+
+### How to enable in Azure Databricks?
+
+The Spark API that would allow to provide custom metrics is the Spark plugin API - see ` https://issues.apache.org/jira/browse/SPARK-24918`,  `https://github.com/apache/spark/pull/26170` and `http://blog.madhukaraphatak.com/spark-plugin-part-4/`
+
+The Azure Cosmos DB Spark connector contains two such plugins to emit metrics:
+
+- `com.azure.cosmos.spark.plugins.CosmosMetricsSparkPlugin`: This plugin emits the metrics to the Spark metric system (Spark config can be used to control where Spark emits metrics - Azure Databricks would usually push them to the Ganglia Metrics UI included in the Spark UI). Optionally the metrics can also be pushed to a Log4J log file (by setting the `spark.cosmos.metrics.slf4j.enabled` property in the spark config to `true`). The `spark.cosmos.metrics.intervalInSeconds` property can be used to determine how frequently metrics should be collected/aggregated (default is once per minute)
+- `com.azure.cosmos.spark.plugins.CosmosMetricsApplicationInsightsPlugin`: This plugin can be used to emit the metrics to Azure monitor. You can provide the Azure Monitor connection string in the `spark.cosmos.metrics.azureMonitor.connectionString` property. The `spark.cosmos.metrics.intervalInSeconds` property can be used to determine how frequently metrics should be collected/aggregated (default is once per minute)
+
+Last-but-not-least - to be able to use these Spark plugins the jar containing the plugins needs to be available at Cluster creation - so it is not sufficient to just install the Azure Cosmos DB Spark connector in the `Libraries` section of the cluster, but the jar needs to be copied into the `/databricks/jars` folder at cluster initialization via a start-up script.
+
+Follow these steps for the installation:
+
+- Download the latest Azure Cosmos DB Spark connector jar from Maven
+  - Spark 3.1: `https://repo1.maven.org/maven2/com/azure/cosmos/spark/azure-cosmos-spark_3-1_2-12/ReplaceWithTheLatestVersion/azure-cosmos-spark_3-1_2-12-ReplaceWithTheLatestVersion.jar`
+  - Spark 3.2: `https://repo1.maven.org/maven2/com/azure/cosmos/spark/azure-cosmos-spark_3-1_2-12/ReplaceWithTheLatestVersion/azure-cosmos-spark_3-2_2-12-ReplaceWithTheLatestVersion.jar` 
+
+- Upload this jar to your Databricks file system (in `/dbfs/FileStore/plugins` folder)
+- Create a text file (NOTE: Use Unix Line feeds (LF - not Windows line feeds) with the content below and upload this start-up script as well.
+- Configure the new start-up script in the cluster configuration
+
+- Content of the start-up script
+  ```sh
+  #!/bin/bash
+
+  STAGE_DIR="/dbfs/FileStore/plugins"
+
+  echo "BEGIN: Upload Spark Plugins"
+  cp -f $STAGE_DIR/*.jar /databricks/jars || { echo "Error copying Spark Plugin library file"; exit 1;}
+  echo "END: Upload Spark Plugin JARs"
+
+  ```
+
+- Now change the clusters's spark configuration to add these two options
+
+  - `spark.plugins ` com.azure.cosmos.spark.plugins.CosmosMetricsSparkPlugin,com.azure.cosmos.spark.plugins.CosmosMetricsApplicationInsightsPlugin
+  - `spark.cosmos.metrics.azureMonitor.connectionString  ` **REPLACE_WITH_YOUR_AZURE_MONITOR_CONNECTION_STRING**
+
+- Restart the cluster - and test...
+
+
+
+
+### How to enable in Azure Synapse?
+
+TBD
@@ -95,12 +95,6 @@
       <version>3.4.22</version> <!-- {x-version-update;io.projectreactor:reactor-test;external_dependency} -->
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-simple</artifactId>
-      <version>1.7.36</version> <!-- {x-version-update;org.slf4j:slf4j-simple;external_dependency} -->
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>org.assertj</groupId>
       <artifactId>assertj-core</artifactId>
@@ -148,6 +142,19 @@
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
       <version>1.7.36</version> <!-- {x-version-update;org.slf4j:slf4j-api;external_dependency} -->
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.micrometer</groupId>
+      <artifactId>micrometer-registry-azure-monitor</artifactId>
+      <version>1.9.3</version> <!-- {x-version-update;io.micrometer:micrometer-registry-azure-monitor;external_dependency} -->
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.microsoft.azure</groupId>
+      <artifactId>applicationinsights-core</artifactId>
+      <version>2.6.4</version> <!-- {x-version-update;cosmos_com.microsoft.azure:applicationinsights-core;external_dependency} -->
+      <scope>compile</scope>
     </dependency>
   </dependencies>
 
@@ -187,6 +194,8 @@
                 <include>org.scalastyle:scalastyle-maven-plugin:[1.0.0]</include> <!-- {x-include-update;cosmos_org.scalastyle:scalastyle-maven-plugin;external_dependency} -->
                 <include>com.fasterxml.jackson.core:jackson-databind:[2.13.3]</include>  <!-- {x-include-update;com.fasterxml.jackson.core:jackson-databind;external_dependency} -->
                 <include>com.fasterxml.jackson.module:jackson-module-scala_2.12:[2.13.3]</include> <!-- {x-include-update;com.fasterxml.jackson.module:jackson-module-scala_2.12;external_dependency} -->
+                <include>io.micrometer:micrometer-registry-azure-monitor:[1.9.3]</include> <!-- {x-include-update;io.micrometer:micrometer-registry-azure-monitor;external_dependency} -->
+                <include>com.microsoft.azure:applicationinsights-core:[2.6.4]</include> <!-- {x-include-update;cosmos_com.microsoft.azure:applicationinsights-core;external_dependency} -->
               </includes>
             </bannedDependencies>
           </rules>
@@ -378,10 +387,6 @@
                       <pattern>io.netty</pattern>
                       <shadedPattern>${shadingPrefix}.io.netty</shadedPattern>
                     </relocation>
-                    <relocation>
-                      <pattern>com.codahale</pattern>
-                      <shadedPattern>${shadingPrefix}.com.codahale</shadedPattern>
-                    </relocation>
                     <relocation>
                       <pattern>org.codehaus</pattern>
                       <shadedPattern>${shadingPrefix}.org.codehaus</shadedPattern>
@@ -418,8 +423,26 @@
                       <pattern>javax.xml</pattern>
                       <shadedPattern>${shadingPrefix}.javax.xml</shadedPattern>
                     </relocation>
+                    <relocation>
+                      <pattern>com.microsoft.azure</pattern>
+                      <shadedPattern>${shadingPrefix}.com.microsoft.azure</shadedPattern>
+                    </relocation>
+                    <relocation>
+                      <pattern>com.microsoft.applicationinsights</pattern>
+                      <shadedPattern>${shadingPrefix}.com.microsoft.applicationinsights</shadedPattern>
+                    </relocation>
+                    <relocation>
+                      <pattern>com.google</pattern>
+                      <shadedPattern>${shadingPrefix}.com.google</shadedPattern>
+                    </relocation>
                   </relocations>
                   <filters>
+                    <filter>
+                      <artifact>com.microsoft.azure:applicationinsights-core</artifact>
+                      <includes>
+                        <include>**</include>
+                      </includes>
+                    </filter>
                     <filter>
                       <artifact>*:*</artifact>
                       <excludes>

@@ -2,7 +2,8 @@
 // Licensed under the MIT License.
 package com.azure.cosmos.spark
 
-import com.azure.cosmos.implementation.{CosmosClientMetadataCachesSnapshot, CosmosDaemonThreadFactory, SparkBridgeImplementationInternal}
+import com.azure.cosmos.implementation.clienttelemetry.TagName
+import com.azure.cosmos.implementation.{CosmosClientMetadataCachesSnapshot, CosmosDaemonThreadFactory, SparkBridgeImplementationInternal, Strings}
 import com.azure.cosmos.spark.CosmosPredicates.isOnSparkDriver
 import com.azure.cosmos.spark.diagnostics.BasicLoggingTrait
 import com.azure.cosmos.{ConsistencyLevel, CosmosAsyncClient, CosmosClientBuilder, DirectConnectionConfig, ThrottlingRetryOptions}
@@ -121,6 +122,7 @@ private[spark] object CosmosClientCache extends BasicLoggingTrait {
                                cosmosClientStateHandle: Option[CosmosClientMetadataCachesSnapshot],
                                ownerInfo: OwnerInfo)
   : CosmosClientCacheItem = synchronized {
+
     val clientConfigWrapper = ClientConfigurationWrapper(cosmosClientConfiguration)
     cache.get(clientConfigWrapper) match {
       case Some(clientCacheMetadata) => clientCacheMetadata.createCacheItemForReuse(ownerInfo)
@@ -134,6 +136,31 @@ private[spark] object CosmosClientCache extends BasicLoggingTrait {
               .setMaxRetryAttemptsOnThrottledRequests(Int.MaxValue)
               .setMaxRetryWaitTime(Duration.ofSeconds((Integer.MAX_VALUE/1000) - 1)))
 
+        if (CosmosClientMetrics.meterRegistry.isDefined) {
+          val customApplicationNameSuffix = cosmosClientConfiguration.customApplicationNameSuffix
+            .getOrElse("")
+
+          val clientCorrelationId = SparkSession.getActiveSession match {
+            case Some(session) =>
+              val ctx = session.sparkContext
+
+              if (Strings.isNullOrWhiteSpace(customApplicationNameSuffix)) {
+                s"${CosmosClientMetrics.executorId}-${ctx.appName}"
+              } else {
+                s"$customApplicationNameSuffix-${CosmosClientMetrics.executorId}-${ctx.appName}"
+              }
+            case None => customApplicationNameSuffix
+          }
+
+          builder.clientTelemetryConfig().clientMetrics(CosmosClientMetrics.meterRegistry.get)
+          builder.clientTelemetryConfig().clientCorrelationId(clientCorrelationId)
+          builder.clientTelemetryConfig().metricTagNames(
+            s"${TagName.Container}, ${TagName.ClientCorrelationId}, ${TagName.Operation}, " +
+              s"${TagName.OperationStatusCode}, ${TagName.PartitionKeyRangeId}, ${TagName.ServiceEndpoint}, " +
+              s"${TagName.ServiceAddress}"
+          )
+        }
+
         if (cosmosClientConfiguration.disableTcpConnectionEndpointRediscovery) {
           builder.endpointDiscoveryEnabled(false)
         }
@@ -350,6 +377,12 @@ private[spark] object CosmosClientCache extends BasicLoggingTrait {
     }
   }
 
+  def clearCache(): Unit = {
+    cache.readOnlySnapshot().keys.foreach(clientCfgWrapper => purgeImpl(clientCfgWrapper, forceClosure = true))
+    cache.clear()
+    cleanUpToBeClosedWhenNotActiveAnymore(forceClosure = true)
+  }
+
   private[this] class CacheItemImpl
   (
     val cosmosClient: CosmosAsyncClient,
@@ -384,9 +417,7 @@ private[spark] object CosmosClientCache extends BasicLoggingTrait {
           case Some(_) =>
             logInfo(
               s"ApplicationEndListener:onApplicationEnd(${ctx.hashCode}) closed - purging all cosmos clients")
-            cache.readOnlySnapshot().keys.foreach(clientCfgWrapper => purgeImpl(clientCfgWrapper, forceClosure = true))
-            cache.clear()
-            cleanUpToBeClosedWhenNotActiveAnymore(forceClosure = true)
+            clearCache()
           case None =>
             logWarning(s"ApplicationEndListener:onApplicationEnd (${ctx.hashCode}) - not monitored anymore")
         }