update config name

Author: annie-mac

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/changeFeedMetrics/ChangeFeedLsnGapMetric.scala

@@ -0,0 +1,17 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos.changeFeedMetrics
+
+import com.azure.cosmos.spark.CosmosConstants.MetricNames
+import org.apache.spark.sql.connector.metric.CustomSumMetric
+
+/***
+ * This metric is used to track the lsn range for partition within a change feed micro-batch
+ */
+private[cosmos] class ChangeFeedLsnRangeMetric extends CustomSumMetric {
+
+ override def name(): String = MetricNames.ChangeFeedLsnRange
+
+ override def description(): String = MetricNames.ChangeFeedLsnRange
+}

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/changeFeedMetrics/ChangeFeedLsnRangeMetric.scala

@@ -19,7 +19,7 @@ private[cosmos] class ChangeFeedMetricsListener(
   try {
    val metrics = SparkInternalsBridge.getInternalCustomTaskMetricsAsSQLMetric(
     Set(
-     CosmosConstants.MetricNames.ChangeFeedLsnGap,
+     CosmosConstants.MetricNames.ChangeFeedLsnRange,
      CosmosConstants.MetricNames.ChangeFeedFetchedChangesCnt,
      CosmosConstants.MetricNames.ChangeFeedPartitionIndex
     ),
@@ -34,16 +34,16 @@ private[cosmos] class ChangeFeedMetricsListener(
      partitionMetricsMap.putIfAbsent(normalizedRange, new ChangeFeedMetricsTracker(index, normalizedRange))
      val metricsTracker = partitionMetricsMap.get(normalizedRange)
      val changesCnt = getChangesCnt(metrics)
-     val lsnGap = getLsnGap(metrics)
+     val lsnRange = getLsnRange(metrics)
 
-     if (changesCnt >= 0 && lsnGap >= 0) {
+     if (changesCnt >= 0 && lsnRange >= 0) {
       metricsTracker.track(
-       metrics(CosmosConstants.MetricNames.ChangeFeedLsnGap).value,
+       metrics(CosmosConstants.MetricNames.ChangeFeedLsnRange).value,
        metrics(CosmosConstants.MetricNames.ChangeFeedFetchedChangesCnt).value
       )
      }
 
-     logInfo(s"onTaskEnd for partition index $index, changesCnt $changesCnt, lsnGap $lsnGap")
+     logInfo(s"onTaskEnd for partition index $index, changesCnt $changesCnt, lsnRange $lsnRange")
     }
    }
   } catch {
@@ -62,9 +62,9 @@ private[cosmos] class ChangeFeedMetricsListener(
   }
  }
 
- def getLsnGap(metrics: Map[String, SQLMetric]): Long = {
-  if (metrics.contains(CosmosConstants.MetricNames.ChangeFeedLsnGap)) {
-   metrics(CosmosConstants.MetricNames.ChangeFeedLsnGap).value
+ def getLsnRange(metrics: Map[String, SQLMetric]): Long = {
+  if (metrics.contains(CosmosConstants.MetricNames.ChangeFeedLsnRange)) {
+   metrics(CosmosConstants.MetricNames.ChangeFeedLsnRange).value
   } else {
    -1
   }

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/changeFeedMetrics/ChangeFeedMetricsListener.scala

@@ -40,12 +40,12 @@ private[cosmos] class ChangeFeedMetricsTracker(
   /**
    * Track the normalized change feed changes per lsn
    *
-   * @param lsnGap the lsn gap has been observed.
+   * @param lsnRange the lsn range of the fetched changes.
    * @param changesFetchedCnt the total fetched changes.
    */
-  def track(lsnGap: Long, changesFetchedCnt: Long): Unit = {
+  def track(lsnRange: Long, changesFetchedCnt: Long): Unit = {
     val effectiveChangesFetchedCnt = Math.max(1, changesFetchedCnt)
-    val changesPerLsn = if (lsnGap == 0) effectiveChangesFetchedCnt.toDouble else effectiveChangesFetchedCnt.toDouble / lsnGap
+    val changesPerLsn = if (lsnRange == 0) effectiveChangesFetchedCnt.toDouble else effectiveChangesFetchedCnt.toDouble / lsnRange
     synchronized {
       changeFeedChangesPerLsnHistory.add(changesPerLsn)
       calculateWeightedChangesPerLsn()
@@ -57,7 +57,7 @@ private[cosmos] class ChangeFeedMetricsTracker(
    * Uses exponential weighting: weight = decayFactor^(n-i-1) where:
    * n = number of measurements
    * i = index of measurement (0 being oldest)
-   * @return Weighted average of LSN gaps
+   * @return Weighted changes per lsn
    */
   private def calculateWeightedChangesPerLsn(): Unit = {
     if (!changeFeedChangesPerLsnHistory.isEmpty) {
@@ -80,8 +80,8 @@ private[cosmos] class ChangeFeedMetricsTracker(
    * Gets current weighted changes per lsn
    * @return weighted changes per lsn
    */
-  def getWeightedAvgChangesPerLsn: Option[Double] = {
-    logDebug(s"getWeightedAvgChangesPerLsn for feedRangeIndex [$partitionIndex], feedRange [$feedRange] value [$currentChangesPerLsnOpt]")
+  def getWeightedChangesPerLsn: Option[Double] = {
+    logDebug(s"getWeightedChangesPerLsn for feedRangeIndex [$partitionIndex], feedRange [$feedRange] value [$currentChangesPerLsnOpt]")
     this.currentChangesPerLsnOpt
   }
 }

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/changeFeedMetrics/ChangeFeedMetricsTracker.scala

@@ -193,7 +193,7 @@ private class ChangeFeedBatch
             SparkBridgeImplementationInternal
               .extractChangeFeedStateForRange(initialOffsetJson, partition.feedRange),
             clearEndLsn = !hasBatchCheckpointLocation))
-       .map(_.asInstanceOf[InputPartition])
+        .map(_.asInstanceOf[InputPartition])
 
       log.logInfo(s"<-- planInputPartitions $batchId (creating ${inputPartitions.length} partitions)")
       inputPartitions

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/spark/ChangeFeedBatch.scala

@@ -51,9 +51,9 @@ private case class ChangeFeedPartitionReader
   private val startLsn = getPartitionStartLsn
   private var latestLsnReturned: Option[Long] = None
 
-  private val changeFeedLSNGapMetric = new CustomTaskMetric {
-    override def name(): String = MetricNames.ChangeFeedLsnGap
-    override def value(): Long = getChangeFeedLSNGap
+  private val changeFeedLSNRangeMetric = new CustomTaskMetric {
+    override def name(): String = MetricNames.ChangeFeedLsnRange
+    override def value(): Long = getChangeFeedLSNRange
   }
   private val changeFeedFetchedChangesCntMetric = new CustomTaskMetric {
     override def name(): String = MetricNames.ChangeFeedFetchedChangesCnt
@@ -109,7 +109,7 @@ private case class ChangeFeedPartitionReader
 
   override def currentMetricsValues(): Array[CustomTaskMetric] = {
     Array(
-      changeFeedLSNGapMetric,
+      changeFeedLSNRangeMetric,
       changeFeedFetchedChangesCntMetric,
       changeFeedPartitionIndexMetric
     )
@@ -290,7 +290,7 @@ private case class ChangeFeedPartitionReader
     this.iterator.getTotalChangesFetched
   }
 
-  private def getChangeFeedLSNGap: Long = {
+  private def getChangeFeedLSNRange: Long = {
     // calculate the changes per lsn
     val latestLsnOpt = this.iterator.getLatestContinuationToken match {
       case Some(continuationToken) =>

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/spark/ChangeFeedPartitionReader.scala

@@ -2,7 +2,7 @@
 // Licensed under the MIT License.
 package com.azure.cosmos.spark
 
-import com.azure.cosmos.changeFeedMetrics.{ChangeFeedFetchedChangesCntMetric, ChangeFeedLsnGapMetric, ChangeFeedPartitionIndexMetric}
+import com.azure.cosmos.changeFeedMetrics.{ChangeFeedFetchedChangesCntMetric, ChangeFeedLsnRangeMetric, ChangeFeedPartitionIndexMetric}
 import com.azure.cosmos.spark.diagnostics.LoggerHelper
 import org.apache.spark.broadcast.Broadcast
 import org.apache.spark.sql.SparkSession
@@ -69,7 +69,7 @@ private case class ChangeFeedScan
 
   override def supportedCustomMetrics(): Array[CustomMetric] = {
     Array(
-      new ChangeFeedLsnGapMetric,
+      new ChangeFeedLsnRangeMetric,
       new ChangeFeedFetchedChangesCntMetric,
       new ChangeFeedPartitionIndexMetric
     )

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/spark/ChangeFeedScan.scala

@@ -35,7 +35,7 @@ private[cosmos] object CosmosConstants {
     val RecordsWritten = "recordsWritten"
     val TotalRequestCharge = "cosmos.totalRequestCharge"
 
-    val ChangeFeedLsnGap = "cosmos.changeFeed.partition.lsnGap"
+    val ChangeFeedLsnRange = "cosmos.changeFeed.partition.lsnRange"
     val ChangeFeedFetchedChangesCnt = "cosmos.changeFeed.partition.fetchedChangesCnt"
     val ChangeFeedPartitionIndex = "cosmos.changeFeed.partition.index"
 
@@ -79,11 +79,11 @@ private[cosmos] object CosmosConstants {
   }
 
   object ChangeFeedMetricsListenerConfig {
-    val metricsListenerEnabledPropertyName = "cosmos.changeFeed.metricsListener.enabled"
-    val metricsListenerEnabledEnvName = "cosmos_changeFeed_metricsListener_enabled"
-    val enableByDefalult = "true"
+    val metricsListenerEnabledPropertyName = "cosmos.changeFeed.performance.monitoring.enabled"
+    val metricsListenerEnabledEnvName = "COSMOS_CHANGEFEED_PERFORMANCE_MONITORING_ENABLED"
+    val enableByDefault = "true"
     val metricsListenerEnabled = Option(System.getProperty(metricsListenerEnabledPropertyName))
      .orElse(sys.env.get(metricsListenerEnabledEnvName))
-     .getOrElse(enableByDefalult).toBoolean
+     .getOrElse(enableByDefault).toBoolean
   }
 }

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/spark/CosmosConstants.scala

@@ -555,7 +555,7 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait {
       partitionMetricsMap match {
         case Some(metricsMap) =>
           if (metricsMap.containsKey(metadata.feedRange)) {
-            changesPerLsnFromMetricsOpt = metricsMap.get(metadata.feedRange).getWeightedAvgChangesPerLsn
+            changesPerLsnFromMetricsOpt = metricsMap.get(metadata.feedRange).getWeightedChangesPerLsn
           }
         case None =>
       }
@@ -572,9 +572,9 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait {
     var changesPerLsnFromMetricsOpt: Option[Double] = None
     if (isChangeFeed) {
       partitionMetricsMap match {
-        case Some(matricsMap) =>
-          if (matricsMap.containsKey(metadata.feedRange)) {
-            changesPerLsnFromMetricsOpt = partitionMetricsMap.get.get(metadata.feedRange).getWeightedAvgChangesPerLsn
+        case Some(metricsMap) =>
+          if (metricsMap.containsKey(metadata.feedRange)) {
+            changesPerLsnFromMetricsOpt = partitionMetricsMap.get.get(metadata.feedRange).getWeightedChangesPerLsn
           }
         case None =>
       }