add dso-visible tps stats to bigquery (#2100)

Signed-off-by: Itai Segall <itai.segall@digitalasset.com>

Author: isegall-da

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ScanTotalSupplyBigQueryIntegrationTest.scala

@@ -23,6 +23,7 @@ import scala.concurrent.Future
 import scala.jdk.CollectionConverters.*
 import scala.jdk.DurationConverters.*
 import scala.sys.process.Process
+import java.time.temporal.ChronoUnit
 
 class ScanTotalSupplyBigQueryIntegrationTest
     extends SpliceTests.IntegrationTest
@@ -69,6 +70,12 @@ class ScanTotalSupplyBigQueryIntegrationTest
   private val unmintedAmount = BigDecimal("570776.255709163")
   private val amuletHolders = 5
   private val validators = 4 // one SV + 3 validators
+  // The test currently produces 80 transactions, which is 0.000926 tps over 24 hours,
+  // so we assert for a range of 70-85 transactions, or 0.0008-0.00099 tps.
+  private val avgTps = (0.0008, 0.00099)
+  // The peak is 17 transactions in a (simulated) minute, or 0.28333 tps over a minute,
+  // so we assert 15-20 transactions, or 0.25-0.34 tps
+  private val peakTps = (0.25, 0.34)
 
   override def beforeAll() = {
     super.beforeAll()
@@ -501,7 +508,8 @@ class ScanTotalSupplyBigQueryIntegrationTest
     */
   private def runTotalSupplyQueries()(implicit env: FixtureParam): ExpectedMetrics = {
     val project = bigquery.getOptions.getProjectId
-    val timestamp = getLedgerTime.toInstant.plusSeconds(10).toString
+    // The TPS query assumes staleness of up to 4 hours, so we query for stats 5 hours after the current ledger time.
+    val timestamp = getLedgerTime.toInstant.plus(5, ChronoUnit.HOURS).toString
     val sql =
       s"SELECT * FROM `$project.$functionsDatasetName.all_stats`('$timestamp', 0);"
 
@@ -536,29 +544,32 @@ class ScanTotalSupplyBigQueryIntegrationTest
       burned: BigDecimal,
       numAmuletHolders: Long,
       numActiveValidators: Long,
+      avgTps: Double,
+      peakTps: Double,
   )
 
   private def parseQueryResults(result: bq.TableResult) = {
     // We expect the final query to return a single row with all metrics
     val row = result.iterateAll().iterator().next()
     logger.debug(s"Query row: $row; schema ${result.getSchema}")
 
-    def bd(column: String) = {
+    def required(column: String) = {
       val field = row get column
       if (field.isNull)
         fail(s"Column '$column' in all-stats results is null")
-      else
-        BigDecimal(field.getStringValue)
+      field
+    }
+
+    def bd(column: String) = {
+      BigDecimal(required(column).getStringValue)
     }
 
     def int(column: String) = {
-      val field = row get column
-      if (field.isNull)
-        fail(s"Column '$column' in all-stats results is null")
-      else {
-        field.getLongValue
-      }
+      required(column).getLongValue
+    }
 
+    def float(column: String) = {
+      required(column).getDoubleValue
     }
 
     ExpectedMetrics(
@@ -571,6 +582,8 @@ class ScanTotalSupplyBigQueryIntegrationTest
       burned = bd("burned"),
       numAmuletHolders = int("num_amulet_holders"),
       numActiveValidators = int("num_active_validators"),
+      avgTps = float("average_tps"),
+      peakTps = float("peak_tps"),
     )
   }
 
@@ -594,6 +607,16 @@ class ScanTotalSupplyBigQueryIntegrationTest
       actual shouldBe expected withClue clue
     }
 
+    forEvery(
+      Seq(
+        ("average_tps", results.avgTps, avgTps),
+        ("peak_tps", results.peakTps, peakTps),
+      )
+    ) { case (clue, actual, expected) =>
+      actual shouldBe >=(expected._1) withClue clue
+      actual shouldBe <=(expected._2) withClue clue
+    }
+
     // other derived metrics
     (mintedAmount - burnedAmount) shouldBe (
       lockedAmount + unlockedAmount

cluster/pulumi/canton-network/src/bigQuery_functions.ts

@@ -8,6 +8,7 @@ import {
   BQFunctionArgument,
   BQScalarFunction,
   BQTableFunction,
+  FLOAT64,
   INT64,
   json,
   STRING,
@@ -448,6 +449,56 @@ const num_active_validators = new BQScalarFunction(
   `
 );
 
+const one_day_updates = new BQTableFunction(
+  'one_day_updates',
+  as_of_args,
+  [new BQColumn('update_id', STRING), new BQColumn('record_time', INT64)],
+  `
+    -- All update IDs over a period of 24 hours. Since the data might be up to 4 hours stale, we take a window of 4-28 hours ago.
+    SELECT DISTINCT(update_id), record_time FROM (
+        SELECT
+          update_id,
+          record_time
+        FROM \`$$SCAN_DATASET$$.scan_sv_1_update_history_exercises\` e
+        WHERE \`$$FUNCTIONS_DATASET$$.in_time_window\`(TIMESTAMP_SUB(as_of_record_time, INTERVAL 28 HOUR), 0, TIMESTAMP_SUB(as_of_record_time, INTERVAL 4 HOUR), migration_id, e.record_time, e.migration_id)
+      UNION ALL
+        SELECT
+          update_id,
+          record_time
+        FROM \`$$SCAN_DATASET$$.scan_sv_1_update_history_creates\` c
+        WHERE \`$$FUNCTIONS_DATASET$$.in_time_window\`(TIMESTAMP_SUB(as_of_record_time, INTERVAL 28 HOUR), 0, TIMESTAMP_SUB(as_of_record_time, INTERVAL 4 HOUR), migration_id, c.record_time, c.migration_id)
+    )
+  `
+);
+
+const average_tps = new BQScalarFunction(
+  'average_tps',
+  as_of_args,
+  FLOAT64,
+  `
+    (SELECT COUNT(*) / 86400.0
+      FROM \`$$FUNCTIONS_DATASET$$.one_day_updates\`(as_of_record_time, migration_id)
+    )
+  `
+);
+
+const peak_tps = new BQScalarFunction(
+  'peak_tps',
+  as_of_args,
+  FLOAT64,
+  `
+    (SELECT
+      MAX(events_per_minute) / 60.0
+    FROM (
+      SELECT
+        COUNT(*) as events_per_minute
+      FROM \`$$FUNCTIONS_DATASET$$.one_day_updates\`(as_of_record_time, migration_id)
+      GROUP BY TIMESTAMP_TRUNC(TIMESTAMP_MICROS(record_time), MINUTE)
+    )
+  )
+  `
+);
+
 const all_stats = new BQTableFunction(
   'all_stats',
   as_of_args,
@@ -463,6 +514,8 @@ const all_stats = new BQTableFunction(
     new BQColumn('burned', BIGNUMERIC),
     new BQColumn('num_amulet_holders', INT64),
     new BQColumn('num_active_validators', INT64),
+    new BQColumn('average_tps', FLOAT64),
+    new BQColumn('peak_tps', FLOAT64),
   ],
   `
     SELECT
@@ -476,7 +529,9 @@ const all_stats = new BQTableFunction(
       \`$$FUNCTIONS_DATASET$$.minted\`(as_of_record_time, migration_id) + \`$$FUNCTIONS_DATASET$$.unminted\`(as_of_record_time, migration_id) as allowed_mint,
       \`$$FUNCTIONS_DATASET$$.burned\`(as_of_record_time, migration_id) as burned,
       \`$$FUNCTIONS_DATASET$$.num_amulet_holders\`(as_of_record_time, migration_id) as num_amulet_holders,
-      \`$$FUNCTIONS_DATASET$$.num_active_validators\`(as_of_record_time, migration_id) as num_active_validators
+      \`$$FUNCTIONS_DATASET$$.num_active_validators\`(as_of_record_time, migration_id) as num_active_validators,
+      \`$$FUNCTIONS_DATASET$$.average_tps\`(as_of_record_time, migration_id) as average_tps,
+      \`$$FUNCTIONS_DATASET$$.peak_tps\`(as_of_record_time, migration_id) as peak_tps
   `
 );
 
@@ -501,5 +556,8 @@ export const allFunctions = [
   num_amulet_holders,
   all_validators,
   num_active_validators,
+  one_day_updates,
+  average_tps,
+  peak_tps,
   all_stats,
 ];

cluster/pulumi/canton-network/src/bigQuery_functions_types.ts

@@ -31,6 +31,7 @@ export const TIMESTAMP = new BQBasicType('TIMESTAMP');
 export const BIGNUMERIC = new BQBasicType('BIGNUMERIC');
 export const json = new BQBasicType('JSON'); // JSON is a reserved word in TypeScript
 export const BOOL = new BQBasicType('BOOL');
+export const FLOAT64 = new BQBasicType('FLOAT64');
 
 export class BQArray extends BQType {
   private readonly elementType: BQType;