Tests - Cluster Logging Capture

[GeekSheikh]

Let's ensure that all clusters with logging enabled are captured. The cluster_id must be found for all scenarios.

Code to validate is here.

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

Lines 640 to 658 in 589c353

	val incrementalClusterBase = df
	.selectExpr("*", "requestParams.*").drop("requestParams")
	.filter('serviceName === "clusters" && 'actionName.isin("create", "edit"))
	.withColumn("cluster_id", when('actionName === "create", get_json_object($"response.result", "$.cluster_id"))
	.when('actionName =!= "create" && 'cluster_id.isNull, 'clusterId)
	.otherwise('cluster_id).alias("cluster_id")
	)
	.select('organization_id, 'timestamp, 'cluster_id, 'cluster_name, 'cluster_log_conf)
	// Get incremental snapshot of clusters during current run
	// This captures clusters that have not been edited/restarted (still not terminated) since the last run with
	// log confs as they will not be in the audit logs
	val latestSnapW = Window.partitionBy('organization_id).orderBy('Pipeline_SnapTS.desc)
	val currentlyDefinedClustersWithLogging = clusterSnapshot.asDF
	.withColumn("snapRnk", rank.over(latestSnapW))
	.filter('snapRnk === 1)
	.withColumn("cluster_log_conf", to_json('cluster_log_conf))
	.filter('cluster_id.isNotNull && 'cluster_log_conf.isNotNull)
	.select('cluster_id, 'cluster_log_conf)