0713 Release Staging (#823)

* initial 0713 commit

* fixed (#815)

* fixed (#816)

* fixed (#818)

* jrcp worker potential fixed (#822)

Co-authored-by: geeksheikh <[email protected]>

* omitted sql warehouses from cluster spec silver (#831)

* omitted sql warehouses from cluster spec silver

* only publish successful change requsts

* added constant for response status

* backporting fix for 709 into 713 (#833)

---------

Co-authored-by: geeksheikh <[email protected]>

Author: GeekSheikh

build.sbt

@@ -2,7 +2,7 @@ name := "overwatch"
 
 organization := "com.databricks.labs"
 
-version := "0.7.1.2"
+version := "0.7.1.3"
 
 scalaVersion := "2.12.12"
 scalacOptions ++= Seq("-Xmax-classfile-name", "78")

src/main/scala/com/databricks/labs/overwatch/pipeline/GoldTransforms.scala

@@ -333,13 +333,22 @@ trait GoldTransforms extends SparkSessionWrapper {
       // using noise generators to fill with two passes to reduce skew
       .fillMeta(clusterPotMetaToFill, clusterPotKeys, clusterPotIncrementals, noiseBuckets = getTotalCores) // scan back then forward to fill all
 
-
-    val workerPotentialCoreS = when('databricks_billable, $"workerSpecs.vCPUs" * 'current_num_workers * 'uptime_in_state_S).otherwise(lit(0))
+    val isSingleNode = when(get_json_object('custom_tags, "$.ResourceClass") === "SingleNode", lit(true)).otherwise(lit(false))
+    val workerPotentialCoreS = when('databricks_billable && isSingleNode, $"driverSpecs.vCPUs" * 'uptime_in_state_S)
+      .when('databricks_billable, $"workerSpecs.vCPUs" * 'current_num_workers * 'uptime_in_state_S)
+      .otherwise(lit(0))
     val isPhotonEnabled = upper('runtime_engine).equalTo("PHOTON")
     val isNotAnSQlWarehouse = !upper('sku).equalTo("SQLCOMPUTE")
-    val photonDBUMultiplier = when(isPhotonEnabled && isNotAnSQlWarehouse, lit(2)).otherwise(lit(1))
+    val photonDBUMultiplier = when(isPhotonEnabled && isNotAnSQlWarehouse && isAutomated('cluster_name), lit(2.9))
+      .when(isPhotonEnabled && isNotAnSQlWarehouse && !isAutomated('cluster_name), lit(2.0))
+      .otherwise(lit(1.0))
     val driverDBUs = when('databricks_billable, $"driverSpecs.Hourly_DBUs" * 'uptime_in_state_H * photonDBUMultiplier).otherwise(lit(0)).alias("driver_dbus")
-    val workerDBUs = when('databricks_billable, $"workerSpecs.Hourly_DBUs" * 'current_num_workers * 'uptime_in_state_H * photonDBUMultiplier).otherwise(lit(0)).alias("worker_dbus")
+    val workerDBUs = when('databricks_billable && !isSingleNode, $"workerSpecs.Hourly_DBUs" * 'current_num_workers * 'uptime_in_state_H * photonDBUMultiplier).otherwise(lit(0)).alias("worker_dbus")
+    val driverCoreHours = round(TransformFunctions.getNodeInfo("driver", "vCPUs", true) / lit(3600), 2)
+    val workerCoreHours = round(TransformFunctions.getNodeInfo("worker", "vCPUs", true) / lit(3600), 2)
+    val coreHours = when('isRunning && isSingleNode, driverCoreHours)
+      .when('isRunning && !isSingleNode, workerCoreHours)
+      .otherwise(lit(0.0)).alias("core_hours")
     val driverComputeCost = Costs.compute('cloud_billable, $"driverSpecs.Compute_Contract_Price", lit(1), 'uptime_in_state_H).alias("driver_compute_cost")
     val workerComputeCost = Costs.compute('cloud_billable, $"workerSpecs.Compute_Contract_Price", 'target_num_workers, 'uptime_in_state_H).alias("worker_compute_cost")
     val driverDBUCost = Costs.dbu(driverDBUs, 'dbu_rate).alias("driver_dbu_cost")
@@ -375,10 +384,7 @@ trait GoldTransforms extends SparkSessionWrapper {
       driverDBUs,
       workerDBUs,
       (driverDBUs + workerDBUs).alias("total_dbus"),
-      when('isRunning,
-        round(TransformFunctions.getNodeInfo("driver", "vCPUs", true) / lit(3600), 2) +
-          round(TransformFunctions.getNodeInfo("worker", "vCPUs", true) / lit(3600), 2)
-      ).otherwise(lit(0.0)).alias("core_hours"),
+      coreHours,
       'driverSpecs,
       'workerSpecs,
       driverComputeCost,

src/main/scala/com/databricks/labs/overwatch/pipeline/SilverTransforms.scala

@@ -14,6 +14,7 @@ trait SilverTransforms extends SparkSessionWrapper {
   import spark.implicits._
 
   private val logger: Logger = Logger.getLogger(this.getClass)
+  private val responseSuccessFilter: Column = $"response.statusCode" === 200
 
   private def appendPowerProperties: Column = {
     struct(
@@ -419,6 +420,7 @@ trait SilverTransforms extends SparkSessionWrapper {
     $"userIdentity.email".alias("userEmail"), 'requestId, 'response)
 
   private def clusterBase(auditRawDF: DataFrame): DataFrame = {
+    val isWarehouse = get_json_object('custom_tags, "$.SqlEndpointId").isNotNull
     val cluster_id_gen_w = Window.partitionBy('organization_id, 'cluster_name).orderBy('timestamp).rowsBetween(Window.currentRow, 1000)
     val cluster_name_gen_w = Window.partitionBy('organization_id, 'cluster_id).orderBy('timestamp).rowsBetween(Window.currentRow, 1000)
     val cluster_id_gen = first('cluster_id, true).over(cluster_id_gen_w)
@@ -465,7 +467,9 @@ trait SilverTransforms extends SparkSessionWrapper {
     val clusterRaw = auditRawDF
       .filter('serviceName === "clusters" && !'actionName.isin("changeClusterAcl"))
       .selectExpr("*", "requestParams.*").drop("requestParams", "Overwatch_RunID")
+      .filter(responseSuccessFilter) // only publish successful edits into the spec table
       .select(clusterSummaryCols: _*)
+      .filter(!isWarehouse)
       .withColumn("cluster_id", cluster_id_gen)
       .withColumn("cluster_name", cluster_name_gen)
 
@@ -512,14 +516,14 @@ trait SilverTransforms extends SparkSessionWrapper {
         'poolSnapDetails
       )
 
-    val deleteCol = when('actionName === "delete" && $"response.statusCode" === 200,
+    val deleteCol = when('actionName === "delete" && responseSuccessFilter,
       struct(
         $"userIdentity.email".alias("deleted_by"),
         'timestamp.alias("deleted_at_epochMillis"),
         from_unixtime('timestamp / 1000).cast("timestamp").alias("deleted_at")
       )).otherwise(lit(null).cast(Schema.poolsDeleteSchema)).alias("delete_details")
 
-    val createCol = when('actionName === "create" && $"response.statusCode" === 200,
+    val createCol = when('actionName === "create" && responseSuccessFilter,
       struct(
         $"userIdentity.email".alias("created_by"),
         'timestamp.alias("created_at_epochMillis"),
@@ -873,7 +877,6 @@ trait SilverTransforms extends SparkSessionWrapper {
     )
 
     val clustersRemoved = clusterBaseDF
-      .filter($"response.statusCode" === 200) // only successful delete statements get applied
       .filter('actionName.isin("permanentDelete"))
       .select('organization_id, 'cluster_id, 'userEmail.alias("deleted_by"))
 

src/main/scala/com/databricks/labs/overwatch/pipeline/WorkflowsTransforms.scala

@@ -653,7 +653,7 @@ object WorkflowsTransforms extends SparkSessionWrapper {
         'jobId.cast("long").alias("submissionJobId"),
         'runId.alias("jobRunId"),
         'timestamp.alias("submissionTime"),
-        'jobTriggerType,
+        'jobTriggerType.alias("jobTriggerType_Triggered"),
         'requestId.alias("submitRequestID"),
         'response.alias("submitResponse"),
         'userIdentity.alias("submittedBy")
@@ -833,7 +833,7 @@ object WorkflowsTransforms extends SparkSessionWrapper {
         coalesce('parentRunId_Started, 'parentRunId_Completed).cast("long").alias("parentRunId"),
         coalesce('taskRunId, 'idInJob).cast("long").alias("idInJob"),
         TransformFunctions.subtractTime(
-          'submissionTime,
+          'startTime,
           array_max(array('completionTime, 'cancellationTime)) // endTS must remain null if still open
         ).alias("TaskRunTime"), // run launch time until terminal event
         TransformFunctions.subtractTime(
@@ -843,7 +843,7 @@ object WorkflowsTransforms extends SparkSessionWrapper {
         'run_name,
         coalesce('jobClusterType_Started, 'jobClusterType_Completed).alias("clusterType"),
         coalesce('jobTaskType_Started, 'jobTaskType_Completed).alias("taskType"),
-        coalesce('jobTriggerType_Started, 'jobTriggerType_Completed, 'jobTriggerType_runNow).alias("jobTriggerType"),
+        coalesce('jobTriggerType_Triggered,'jobTriggerType_Started, 'jobTriggerType_Completed, 'jobTriggerType_runNow).alias("jobTriggerType"),
         when('cancellationRequestId.isNotNull, "Cancelled")
           .otherwise('jobTerminalState)
           .alias("terminalState"),
@@ -1221,6 +1221,8 @@ object WorkflowsTransforms extends SparkSessionWrapper {
       .withColumn("runtime_in_cluster_state", // all runs have an initial cluster state
         when('state.isin("CREATING", "STARTING") || 'cluster_type === "new", 'uptime_in_state_H * 1000 * 3600) // get true cluster time when state is guaranteed fully initial
           .otherwise(runStateFirstToEnd - $"task_runtime.startEpochMS")) // otherwise use jobStart as beginning time and min of stateEnd or jobEnd for end time )
+      .withColumn("hourly_core_potential", 'cluster_state_worker_potential_core_H / 'uptime_in_state_H) // executor potential per hour
+      .withColumn("worker_potential_core_H", ('runtime_in_cluster_state / 1000.0 / 60.0 / 60.0) * 'hourly_core_potential)
       .withColumn("lifecycleState", lit("init"))
   }
 
@@ -1256,6 +1258,8 @@ object WorkflowsTransforms extends SparkSessionWrapper {
       )
       .join(jobRunInitialStates.select(stateLifecycleKeys map col: _*), stateLifecycleKeys, "leftanti") // filter out beginning states
       .withColumn("runtime_in_cluster_state", taskRunEndOrPipelineEnd - runStateLastToStart)
+      .withColumn("hourly_core_potential", 'cluster_state_worker_potential_core_H / 'uptime_in_state_H) // executor potential per hour
+      .withColumn("worker_potential_core_H", ('runtime_in_cluster_state / 1000.0 / 60.0 / 60.0) * 'hourly_core_potential)
       .withColumn("lifecycleState", lit("terminal"))
   }
 
@@ -1295,6 +1299,8 @@ object WorkflowsTransforms extends SparkSessionWrapper {
       .join(jobRunInitialStates.select(stateLifecycleKeys map col: _*), stateLifecycleKeys, "leftanti") // filter out beginning states
       .join(jobRunTerminalStates.select(stateLifecycleKeys map col: _*), stateLifecycleKeys, "leftanti") // filter out ending states
       .withColumn("runtime_in_cluster_state", 'unixTimeMS_state_end - 'unixTimeMS_state_start)
+      .withColumn("hourly_core_potential", 'cluster_state_worker_potential_core_H / 'uptime_in_state_H) // executor potential per hour
+      .withColumn("worker_potential_core_H", ('runtime_in_cluster_state / 1000.0 / 60.0 / 60.0) * 'hourly_core_potential)
       .withColumn("lifecycleState", lit("intermediate"))
 
   }
@@ -1327,7 +1333,7 @@ object WorkflowsTransforms extends SparkSessionWrapper {
       $"workerSpecs.vCPUs".alias("worker_cores"),
       'isAutomated,
       'dbu_rate,
-      'worker_potential_core_H,
+      'worker_potential_core_H.alias("cluster_state_worker_potential_core_H"),
       'driver_compute_cost,
       'worker_compute_cost,
       'driver_dbu_cost,