feat: introduce p50, p90 and p99 latency metrics for graphs and subgraphs

connect/src/wg/cosmo/platform/v1/platform_pb.ts

@@ -10248,6 +10248,21 @@ export class MetricsSeriesItem extends Message<MetricsSeriesItem> {
    */
   previousValue?: string;
 
+  /**
+   * @generated from field: optional string p50 = 4;
+   */
+  p50?: string;
+
+  /**
+   * @generated from field: optional string p90 = 5;
+   */
+  p90?: string;
+
+  /**
+   * @generated from field: optional string p99 = 6;
+   */
+  p99?: string;
+
   constructor(data?: PartialMessage<MetricsSeriesItem>) {
     super();
     proto3.util.initPartial(data, this);
@@ -10259,6 +10274,9 @@ export class MetricsSeriesItem extends Message<MetricsSeriesItem> {
     { no: 1, name: "timestamp", kind: "scalar", T: 9 /* ScalarType.STRING */ },
     { no: 2, name: "value", kind: "scalar", T: 9 /* ScalarType.STRING */ },
     { no: 3, name: "previousValue", kind: "scalar", T: 9 /* ScalarType.STRING */, opt: true },
+    { no: 4, name: "p50", kind: "scalar", T: 9 /* ScalarType.STRING */, opt: true },
+    { no: 5, name: "p90", kind: "scalar", T: 9 /* ScalarType.STRING */, opt: true },
+    { no: 6, name: "p99", kind: "scalar", T: 9 /* ScalarType.STRING */, opt: true },
   ]);
 
   static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): MetricsSeriesItem {

controlplane/src/core/repositories/analytics/MetricsRepository.ts

@@ -44,6 +44,14 @@ interface GetMetricsProps {
   queryParams?: CoercedFilterValues;
 }
 
+interface LatencySeries {
+  timestamp: string;
+  value: string;
+  p50?: number;
+  p90?: number;
+  p99?: number;
+}
+
 export class MetricsRepository {
   constructor(private client: ClickHouseClient) {}
 
@@ -113,7 +121,7 @@ export class MetricsRepository {
 
     // get time series of last [range] hours
     const querySeries = (start: number, end: number) => {
-      return this.client.queryPromise<{ value: number | null }[]>(
+      return this.client.queryPromise<LatencySeries>(
         `
       WITH
         toStartOfInterval(toDateTime('${start}'), INTERVAL ${granule} MINUTE) AS startDate,
@@ -244,13 +252,14 @@ export class MetricsRepository {
 
     // get time series of last [range] hours
     const querySeries = (quantile: string, start: number, end: number) => {
-      return this.client.queryPromise<{ value: number | null }[]>(
+      return this.client.queryPromise<LatencySeries>(
         `
         WITH
           toStartOfInterval(toDateTime('${start}'), INTERVAL ${granule} MINUTE) AS startDate,
           toDateTime('${end}') AS endDate
         SELECT
             toStartOfInterval(Timestamp, INTERVAL ${granule} MINUTE) AS timestamp,
+            -- Default
             func_rank(${quantile}, BucketCounts) as rank,
             func_rank_bucket_lower_index(rank, BucketCounts) as b,
             func_histogram_v2(
@@ -259,6 +268,33 @@ export class MetricsRepository {
                 BucketCounts,
                 anyLast(ExplicitBounds)
             ) as value,
+            -- P50
+            func_rank(0.50, BucketCounts) as rank50,
+            func_rank_bucket_lower_index(rank50, BucketCounts) as b50,
+            func_histogram_v2(
+                rank50,
+                b50,
+                BucketCounts,
+                anyLast(ExplicitBounds)
+            ) as p50,
+            -- P90
+            func_rank(0.90, BucketCounts) as rank90,
+            func_rank_bucket_lower_index(rank90, BucketCounts) as b90,
+            func_histogram_v2(
+                rank90,
+                b90,
+                BucketCounts,
+                anyLast(ExplicitBounds)
+            ) as p90,
+            -- P90
+            func_rank(0.99, BucketCounts) as rank99,
+            func_rank_bucket_lower_index(rank99, BucketCounts) as b99,
+            func_histogram_v2(
+                rank99,
+                b99,
+                BucketCounts,
+                anyLast(ExplicitBounds)
+            ) as p99,
 
             -- Histogram aggregations
             sumForEachMerge(BucketCounts) as BucketCounts
@@ -380,7 +416,7 @@ export class MetricsRepository {
 
     // get time series of last [range] hours
     const getSeries = (start: number, end: number) => {
-      return this.client.queryPromise<{ value: number | null }[]>(
+      return this.client.queryPromise<LatencySeries>(
         `
       WITH
         toStartOfInterval(toDateTime('${start}'), INTERVAL ${granule} MINUTE) AS startDate,
@@ -652,17 +688,19 @@ export class MetricsRepository {
    * @param previousSeries
    * @returns
    */
-  protected mapSeries(diff: number, series: any[] = [], previousSeries?: any[]) {
+  protected mapSeries(diff: number, series: LatencySeries[] = [], previousSeries?: LatencySeries[]) {
     return series.map((s) => {
       const timestamp = new Date(s.timestamp + 'Z').getTime();
       const prevTimestamp = toISO9075(new Date(timestamp - diff * 60 * 60 * 1000));
+      const prevValue = previousSeries?.find((s) => s.timestamp === prevTimestamp)?.value ?? '0';
 
       return {
         timestamp: String(timestamp),
         value: String(s.value),
-        previousValue: String(
-          Number.parseFloat(previousSeries?.find((s) => s.timestamp === prevTimestamp)?.value || '0'),
-        ),
+        previousValue: String(Number.parseFloat(prevValue)),
+        p50: s.p50 === undefined ? undefined : String(s.p50),
+        p90: s.p90 === undefined ? undefined : String(s.p90),
+        p99: s.p99 === undefined ? undefined : String(s.p99),
       };
     });
   }

controlplane/src/core/repositories/analytics/SubgraphMetricsRepository.ts

@@ -52,6 +52,14 @@ interface GetSubgraphMetricsProps {
   queryParams?: CoercedFilterValues;
 }
 
+interface LatencySeries {
+  timestamp: string;
+  value: string;
+  p50?: number;
+  p90?: number;
+  p99?: number;
+}
+
 export class SubgraphMetricsRepository {
   constructor(
     private logger: FastifyBaseLogger,
@@ -125,7 +133,7 @@ export class SubgraphMetricsRepository {
 
     // get time series of last [range] hours
     const querySeries = (start: number, end: number) => {
-      return this.client.queryPromise<{ value: number | null }[]>(
+      return this.client.queryPromise<LatencySeries>(
         `
       WITH
         toStartOfInterval(toDateTime('${start}'), INTERVAL ${granule} MINUTE) AS startDate,
@@ -256,13 +264,14 @@ export class SubgraphMetricsRepository {
 
     // get time series of last [range] hours
     const querySeries = (quantile: string, start: number, end: number) => {
-      return this.client.queryPromise<{ value: number | null }[]>(
+      return this.client.queryPromise<LatencySeries>(
         `
         WITH
           toStartOfInterval(toDateTime('${start}'), INTERVAL ${granule} MINUTE) AS startDate,
           toDateTime('${end}') AS endDate
         SELECT
             toStartOfInterval(Timestamp, INTERVAL ${granule} MINUTE) AS timestamp,
+            -- Default
             func_rank(${quantile}, BucketCounts) as rank,
             func_rank_bucket_lower_index(rank, BucketCounts) as b,
             func_histogram_v2(
@@ -271,6 +280,33 @@ export class SubgraphMetricsRepository {
                 BucketCounts,
                 anyLast(ExplicitBounds)
             ) as value,
+            -- P50
+            func_rank(0.50, BucketCounts) as rank50,
+            func_rank_bucket_lower_index(rank50, BucketCounts) as b50,
+            func_histogram_v2(
+                rank50,
+                b50,
+                BucketCounts,
+                anyLast(ExplicitBounds)
+            ) as p50,
+            -- P90
+            func_rank(0.90, BucketCounts) as rank90,
+            func_rank_bucket_lower_index(rank50, BucketCounts) as b90,
+            func_histogram_v2(
+                rank90,
+                b90,
+                BucketCounts,
+                anyLast(ExplicitBounds)
+            ) as p90,
+            -- P99
+            func_rank(0.99, BucketCounts) as rank99,
+            func_rank_bucket_lower_index(rank50, BucketCounts) as b99,
+            func_histogram_v2(
+                rank99,
+                b99,
+                BucketCounts,
+                anyLast(ExplicitBounds)
+            ) as p99,
 
             -- Histogram aggregations
             sumForEachMerge(BucketCounts) as BucketCounts
@@ -392,7 +428,7 @@ export class SubgraphMetricsRepository {
 
     // get time series of last [range] hours
     const getSeries = (start: number, end: number) => {
-      return this.client.queryPromise<{ value: number | null }[]>(
+      return this.client.queryPromise<LatencySeries>(
         `
       WITH
         toStartOfInterval(toDateTime('${start}'), INTERVAL ${granule} MINUTE) AS startDate,
@@ -702,17 +738,19 @@ export class SubgraphMetricsRepository {
    * @param previousSeries
    * @returns
    */
-  protected mapSeries(diff: number, series: any[] = [], previousSeries?: any[]) {
+  protected mapSeries(diff: number, series: LatencySeries[] = [], previousSeries?: LatencySeries[]) {
     return series.map((s) => {
       const timestamp = new Date(s.timestamp + 'Z').getTime();
       const prevTimestamp = toISO9075(new Date(timestamp - diff * 60 * 60 * 1000));
+      const prevValue = previousSeries?.find((s) => s.timestamp === prevTimestamp)?.value ?? '0';
 
       return {
         timestamp: String(timestamp),
         value: String(s.value),
-        previousValue: String(
-          Number.parseFloat(previousSeries?.find((s) => s.timestamp === prevTimestamp)?.value || '0'),
-        ),
+        previousValue: String(Number.parseFloat(prevValue)),
+        p50: s.p50 === undefined ? undefined : String(s.p50),
+        p90: s.p90 === undefined ? undefined : String(s.p90),
+        p99: s.p99 === undefined ? undefined : String(s.p99),
       };
     });
   }

proto/wg/cosmo/platform/v1/platform.proto

@@ -1342,6 +1342,9 @@ message MetricsSeriesItem {
   string timestamp = 1;
   string value = 2;
   optional string previousValue = 3;
+  optional string p50 = 4;
+  optional string p90 = 5;
+  optional string p99 = 6;
 }
 
 message MetricsDashboard {