[CF] Convert lf_rollover_health query to clickhouse (#5819)

ZainRizvi · web-flow · commit 5b770fbabe10 · 2024-10-25T14:48:56.000-05:00
Converts the lf_rollover_health query to use clickhouse.
It also:
- Corrects a bug in the query about how it calculated the average
duration by fixing the denominator.
- Removes the temporary amz2023 label prefix handling, since those
workflows are obsolete now anyways

Validation:
- Ensured that the ClickHouse query showed the same results as the
Rockset query (except for the duration comparison chart, which also
shows matching results if I reintroduce the average duration bug)
diff --git a/torchci/clickhouse_queries/lf_rollover_health/query.sql b/torchci/clickhouse_queries/lf_rollover_health/query.sql
@@ -1,144 +1,106 @@
--- !!! Query is not converted to CH syntax yet.  Delete this line when it gets converted
-
-WITH normalized_jobs AS (
+-- This query is used to generate a chart on HUD to compare the success, cancellation, and duration rates
+-- of jobs in the Meta fleet vs jobs in the LF fleet over time.
+WITH
+normalized_jobs AS (
   SELECT
     j.started_at,
-    ROUND(DATE_DIFF('MINUTE', PARSE_TIMESTAMP_ISO8601(j.started_at), PARSE_TIMESTAMP_ISO8601(j.completed_at)), 1) as duration_min,
-    if(
-        strpos(l.label, 'amz2023.') = 0,
-        l.label,
-        CONCAT(
-            substr(l.label, 1, strpos(l.label, 'amz2023.') - 1),
-            substr(l.label, length('amz2023.') + strpos(l.label, 'amz2023.'))
-        )
-    ) as label,
-    REGEXP_EXTRACT(j.name, '([^,]*),?', 1) as job_name, -- remove shard number and label from job names
+    ROUND(dateDiff('minute', j.started_at, j.completed_at), 1) as duration_min,
+    l as label,
+    extract(j.name, '[^,]*') as job_name, -- remove shard number and label from job names
     j.workflow_name,
     j.conclusion,
-      DATE_TRUNC(:granularity, PARSE_TIMESTAMP_ISO8601(j.started_at))
-    AS bucket,
+    toStartOfInterval(j.started_at, INTERVAL 1 DAY) AS bucket
   FROM
-      commons.workflow_job j
-      CROSS JOIN UNNEST(j.labels as label) as l
-  WHERE 1=1
-    AND j.labels is not NULL
-    AND j._event_time > CURRENT_DATETIME() - DAYS(:days_ago)
+    -- Deliberatly not adding FINAL to this workflow_job.
+    -- Risks of not using it:
+    --   - You may get duplicate records for rows that were updated corresponding to their
+    --     before/after states, but as long as there’s some mechanism in the query to account
+    --     for that it’s okay (we check for j.status = 'completed`).
+    --   - In the worst case scenario, you may only see the ‘old’ version of the records for some rows
+    -- Costs of using it:
+    --   - Query procesing time increases from ~5 -> 16 seconds
+    --   - Memory usage grows from 7.5 GB -> 32 GB
+    -- So the tradeoff is worth it for this query.
+    workflow_job as j
+  ARRAY JOIN
+    j.labels as l
+  WHERE
+    j.labels IS NOT NULL -- prob unnecessary now
+    AND j.created_at > now() - interval {days_ago: Int64} day
     AND j.status = 'completed'
-    AND l.label != 'self-hosted'
-    AND l.label not like 'lf.c.%'
-    AND l.label not like '%canary%'
-
-), migrated_jobs AS (
+    AND l != 'self-hosted'
+    AND l NOT LIKE 'lf.c.%'
+    AND l NOT LIKE '%canary%'
+),
+lf_jobs AS (
   SELECT DISTINCT
     j.job_name
   FROM
-      normalized_jobs j
-  WHERE 1=1
-    AND j.label like 'lf%'
-), comparable_jobs AS (
-      SELECT
-        -- count(*)
-        j.bucket,
-        j.started_at,
-        j.duration_min,-- -- j.completed_at,
-        j.label,
-        j.job_name, -- remove shard number and label from job names
-        j.workflow_name,
-        j.conclusion,
-      FROM
-        normalized_jobs j
-        CROSS JOIN migrated_jobs mj
-      WHERE 1 = 1
-        AND j.job_name = mj.job_name
-        -- AND STRPOS(j.name, mj.job_clean) > 0
-
-), success_stats AS (
+    normalized_jobs as j
+  WHERE
+    j.label LIKE 'lf%'
+),
+comparable_jobs AS (
+  SELECT
+    j.bucket,
+    j.started_at,
+    j.duration_min,
+    j.label,
+    j.job_name,
+    j.workflow_name,
+    j.conclusion
+  FROM
+    normalized_jobs as j
+  CROSS JOIN lf_jobs as lfj
+  WHERE
+    j.job_name = lfj.job_name
+),
+success_stats AS (
   SELECT
     bucket,
     count(*) as group_size,
     job_name,
     workflow_name,
     label,
-    IF(SUBSTR(label, 1, 3) = 'lf.', True, False ) as lf_fleet,
-    SUM(
-        CASE
-            WHEN conclusion = 'success' THEN 1
-            ELSE 0
-        END
-    ) * 100 / (COUNT_IF(conclusion != 'cancelled') + 1) as success_rate, -- plus one is to handle divide by zero errors
-    SUM(
-        CASE
-            WHEN conclusion = 'failure' THEN 1
-            ELSE 0
-        END
-    ) * 100 / (COUNT_IF(conclusion != 'cancelled') + 1) as failure_rate,
-    SUM(
-        CASE
-            WHEN conclusion = 'cancelled' THEN 1
-            ELSE 0
-        END
-    ) * 100 / COUNT(*) as cancelled_rate, -- cancelled rate is calculated over all jobs
-    SUM(
-        CASE
-            WHEN conclusion = 'success' THEN 1
-            ELSE 0
-        END
-    ) as success_count,
-    SUM(
-        CASE
-            WHEN conclusion = 'failure' THEN 1
-            ELSE 0
-        END
-    ) as failure_count,
-    SUM(
-        CASE
-            WHEN conclusion = 'cancelled' THEN 1
-            ELSE 0
-        END
-    ) as cancelled_count,
-    COUNT(*) as total_count,
-    SUM(
-        CASE
-            WHEN conclusion = 'success' THEN duration_min
-            ELSE 0
-        END
-    ) / COUNT(*) as success_avg_duration,
-    SUM(
-        CASE
-            WHEN conclusion = 'failure' THEN duration_min
-            ELSE 0
-        END
-    ) / COUNT(*) as failure_avg_duration,
-    SUM(
-        CASE
-            WHEN conclusion = 'cancelled' THEN duration_min
-            ELSE 0
-        END
-    ) / COUNT(*) as cancelled_avg_duration,
-
+    if(startsWith(label, 'lf.'), 1, 0 ) as lf_fleet,
+    countIf(conclusion = 'success') * 100 / (countIf(conclusion != 'cancelled') + 1) as success_rate, -- plus one is to handle divide by zero errors
+    countIf(conclusion = 'failure') * 100 / (countIf(conclusion != 'cancelled') + 1) as failure_rate,
+    countIf(conclusion = 'cancelled') * 100 / COUNT() as cancelled_rate,
+    avgIf(duration_min, conclusion = 'success') as success_avg_duration,
+    avgIf(duration_min, conclusion = 'failure') as failure_avg_duration,
+    avgIf(duration_min, conclusion = 'cancelled') as cancelled_avg_duration
   FROM comparable_jobs
-  GROUP BY
-    bucket, job_name, workflow_name, label
-), comparison_stats AS (
-    SELECT
-        lf.bucket,
-        lf.workflow_name,
-        lf.job_name,
-        lf.group_size as sample_size_lf,
-        m.group_size as sample_size_meta,
-        lf.success_rate - m.success_rate as success_rate_delta,
-        lf.failure_rate - m.failure_rate as failure_rate_delta,
-        lf.cancelled_rate - m.cancelled_rate as cancelled_rate_delta,
-        IF(m.success_avg_duration = 0, 1, ROUND(lf.success_avg_duration * 1.0 / m.success_avg_duration, 2)) as success_duration_increase_ratio,
-    FROM success_stats lf
-    INNER JOIN success_stats m on lf.bucket = m.bucket
-    WHERE 1 = 1
-        AND lf.job_name = m.job_name
-        AND lf.workflow_name = m.workflow_name
-        AND lf.lf_fleet = True
-        AND m.lf_fleet = False
-        AND lf.group_size > 3
-        AND m.group_size > 3
+  GROUP BY bucket, job_name, workflow_name, label
+),
+comparison_stats AS (
+  SELECT
+    lf.bucket,
+    lf.workflow_name,
+    lf.job_name,
+    lf.group_size as sample_size_lf,
+    m.group_size as sample_size_meta,
+    lf.success_rate - m.success_rate as success_rate_delta,
+    lf.failure_rate - m.failure_rate as failure_rate_delta,
+    lf.cancelled_rate - m.cancelled_rate as cancelled_rate_delta,
+    if(m.success_avg_duration = 0, 1, round(lf.success_avg_duration / m.success_avg_duration, 2)) as success_duration_increase_ratio
+  FROM
+    success_stats as lf
+  JOIN
+    success_stats as m
+  ON
+    lf.bucket = m.bucket
+    AND lf.job_name = m.job_name
+    AND lf.workflow_name = m.workflow_name
+  WHERE
+    lf.lf_fleet = 1
+    AND m.lf_fleet = 0
+    -- the group size limit reduces noise from low sample sizes
+    AND lf.group_size > 3
+    AND m.group_size > 3
 )
-SELECT * from comparison_stats
-ORDER by bucket desc, job_name desc, success_rate_delta, workflow_name
+SELECT
+  *
+FROM
+  comparison_stats
+ORDER BY
+  bucket DESC, job_name DESC, success_rate_delta, workflow_name
diff --git a/torchci/pages/metrics.tsx b/torchci/pages/metrics.tsx
@@ -1144,7 +1144,7 @@ export default function Page() {
           </Typography>
           <p>
             These panels show the <b>delta</b> between states of the same job
-            run on the Linux Foundation and Meta fleets.
+            run on the Linux Foundation vs the Meta fleets.
           </p>
         </Grid>
 
@@ -1153,19 +1153,24 @@ export default function Page() {
             title={"LF vs Meta: Success rate delta"}
             queryName={"lf_rollover_health"}
             queryCollection={"metrics"}
-            queryParams={[
-              {
-                name: "days_ago",
-                type: "int",
-                value: timeRange,
-              },
-            ]}
+            queryParams={
+              useClickHouse
+                ? { ...timeParamsClickHouse, days_ago: timeRange }
+                : [
+                    {
+                      name: "days_ago",
+                      type: "int",
+                      value: timeRange,
+                    },
+                  ]
+            }
             granularity={"day"}
             timeFieldName={"bucket"}
             yAxisLabel={"rate delta"}
             yAxisFieldName={"success_rate_delta"}
             yAxisRenderer={(value) => value}
             groupByFieldName={"job_name"}
+            useClickHouse={useClickHouse}
           />
         </Grid>
 
@@ -1174,19 +1179,24 @@ export default function Page() {
             title={"LF vs Meta: Cancelled rate delta"}
             queryName={"lf_rollover_health"}
             queryCollection={"metrics"}
-            queryParams={[
-              {
-                name: "days_ago",
-                type: "int",
-                value: timeRange,
-              },
-            ]}
+            queryParams={
+              useClickHouse
+                ? { ...timeParamsClickHouse, days_ago: timeRange }
+                : [
+                    {
+                      name: "days_ago",
+                      type: "int",
+                      value: timeRange,
+                    },
+                  ]
+            }
             granularity={"day"}
             timeFieldName={"bucket"}
             yAxisLabel={"rate delta"}
             yAxisFieldName={"cancelled_rate_delta"}
             yAxisRenderer={(value) => value}
             groupByFieldName={"job_name"}
+            useClickHouse={useClickHouse}
           />
         </Grid>
 
@@ -1195,19 +1205,24 @@ export default function Page() {
             title={"LF vs Meta: Duration increase ratio"}
             queryName={"lf_rollover_health"}
             queryCollection={"metrics"}
-            queryParams={[
-              {
-                name: "days_ago",
-                type: "int",
-                value: timeRange,
-              },
-            ]}
+            queryParams={
+              useClickHouse
+                ? { ...timeParamsClickHouse, days_ago: timeRange }
+                : [
+                    {
+                      name: "days_ago",
+                      type: "int",
+                      value: timeRange,
+                    },
+                  ]
+            }
             granularity={"day"}
             timeFieldName={"bucket"}
             yAxisLabel="increase ratio"
             yAxisFieldName={"success_duration_increase_ratio"}
             yAxisRenderer={(value) => value}
             groupByFieldName={"job_name"}
+            useClickHouse={useClickHouse}
           />
         </Grid>
         <Grid item xs={12} height={ROW_HEIGHT}>
