[CARMEL-484] Collect and expose metrics in HiveExternalCatalog to mon…

…itor the performance of HiveMetastore (#43)
apache · Aug 21, 2020 · d4d52c3 · d4d52c3
1 parent 886bade
commit d4d52c3
Show file tree

Hide file tree

Showing 3 changed files with 190 additions and 121 deletions.
diff --git a/core/src/main/scala/org/apache/spark/metrics/source/StaticSources.scala b/core/src/main/scala/org/apache/spark/metrics/source/StaticSources.scala
@@ -90,6 +90,57 @@ object HiveCatalogMetrics extends Source {
   val METRIC_PARALLEL_LISTING_JOB_COUNT = metricRegistry.counter(
     MetricRegistry.name("parallelListingJobCount"))
 
+  /**
+   * Tracks the overall catalog call duration.
+   */
+  val METRIC_CALL_DURATION = metricRegistry.histogram(
+    MetricRegistry.name("callDuration"))
+
+  private val hiveCatalogCalls =
+    "createDatabase" ::
+      "dropDatabase" ::
+      "alterDatabase" ::
+      "getDatabase" ::
+      "databaseExists" ::
+      "listDatabases" ::
+      "tableExists" ::
+      "getTableOption" ::
+      "createTable" ::
+      "dropTable" ::
+      "alterTable" ::
+      "alterTableDataSchema" ::
+      "createPartitions" ::
+      "dropPartitions" ::
+      "renamePartitions" ::
+      "alterPartitions" ::
+      "getPartitionNames" ::
+      "getPartitionOption" ::
+      "getPartitions" ::
+      "getPartitionsByFilter" ::
+      "listTables" ::
+      "listTables" ::
+      "loadPartition" ::
+      "loadTable" ::
+      "loadDynamicPartitions" ::
+      "createFunction" ::
+      "dropFunction" ::
+      "renameFunction" ::
+      "alterFunction" ::
+      "getFunctionOption" ::
+      "listFunctions" :: Nil
+
+  private lazy val hiveCallCountMetrics =
+    hiveCatalogCalls.map { call =>
+      val name = call + "Count"
+      name -> metricRegistry.counter(name)
+    }.toMap
+
+  private lazy val hiveCallDurationMetrics =
+    hiveCatalogCalls.map { call =>
+      val name = call + "Duration"
+      name -> metricRegistry.histogram(name)
+    }.toMap
+
   /**
    * Resets the values of all metrics to zero. This is useful in tests.
    */
@@ -107,4 +158,10 @@ object HiveCatalogMetrics extends Source {
   def incrementFileCacheHits(n: Int): Unit = METRIC_FILE_CACHE_HITS.inc(n)
   def incrementHiveClientCalls(n: Int): Unit = METRIC_HIVE_CLIENT_CALLS.inc(n)
   def incrementParallelListingJobCount(n: Int): Unit = METRIC_PARALLEL_LISTING_JOB_COUNT.inc(n)
+  def updateCallDuration(call: String, duration: Long): Unit = {
+    hiveCallDurationMetrics.get(call + "Duration").foreach(_.update(duration))
+    METRIC_CALL_DURATION.update(duration)
+  }
+  def incrementCallCount(call: String, count: Int): Unit =
+    hiveCallCountMetrics.get(call + "Count").foreach(_.inc(count))
 }
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClient.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClient.scala
@@ -296,7 +296,7 @@ private[hive] trait HiveClient {
   def newSession(): HiveClient
 
   /** Run a function within Hive state (SessionState, HiveConf, Hive client and class loader) */
-  def withHiveState[A](f: => A): A
+  def withHiveState[A](f: => A, operation: String = ""): A
 
   /** Used for testing only.  Removes all metadata from this instance of Hive. */
   def reset(): Unit