ci: Remember failed workflows/td files and run them first in CI

misc/python/materialize/ci_util/__init__.py

@@ -48,7 +48,6 @@ def upload_junit_report(suite: str, junit_report: Path) -> None:
     """
     if not buildkite.is_in_buildkite():
         return
-    ui.section(f"Uploading report for suite {suite!r} to Buildkite Test Analytics")
     suite = suite.upper().replace("-", "_")
     token = os.getenv(f"BUILDKITE_TEST_ANALYTICS_API_KEY_{suite}")
     if not token:

misc/python/materialize/mzcompose/composition.py

@@ -36,14 +36,14 @@
 from contextlib import contextmanager
 from inspect import Traceback, getframeinfo, getmembers, isfunction, stack
 from tempfile import TemporaryFile
-from typing import Any, TextIO, cast
+from typing import Any, TextIO, TypeVar, cast
 
 import psycopg
 import sqlparse
 import yaml
 from psycopg import Connection, Cursor
 
-from materialize import MZ_ROOT, mzbuild, spawn, ui
+from materialize import MZ_ROOT, buildkite, mzbuild, spawn, ui
 from materialize.mzcompose import cluster_replica_size_map, loader
 from materialize.mzcompose.service import Service
 from materialize.mzcompose.services.materialized import (
@@ -1564,3 +1564,50 @@ def cloud_hostname(self, quiet: bool = False) -> str:
         # It is necessary to append the 'https://' protocol; otherwise, urllib can't parse it correctly.
         cloud_hostname = urllib.parse.urlparse("https://" + cloud_url).hostname
         return str(cloud_hostname)
+
+    T = TypeVar("T")
+
+    def test_parts(self, parts: list[T], process_func: Callable[[T], Any]) -> None:
+        from materialize.test_analytics.config.test_analytics_db_config import (
+            create_test_analytics_config,
+        )
+        from materialize.test_analytics.test_analytics_db import TestAnalyticsDb
+
+        priority: dict[str, int] = {}
+        test_analytics: TestAnalyticsDb | None = None
+
+        if buildkite.is_in_buildkite():
+            print("~~~ Fetching part priorities")
+            test_analytics_config = create_test_analytics_config(self)
+            test_analytics = TestAnalyticsDb(test_analytics_config)
+            try:
+                priority = test_analytics.builds.get_part_priorities(timeout=15)
+                print(f"Priorities: {priority}")
+            except Exception as e:
+                print(f"Failed to fetch part priorities, using default order: {e}")
+
+        sorted_parts = sorted(
+            parts, key=lambda part: priority.get(str(part), 0), reverse=True
+        )
+        exceptions: list[Exception] = []
+
+        try:
+            for part in sorted_parts:
+                try:
+                    process_func(part)
+                except Exception as e:
+                    if buildkite.is_in_buildkite():
+                        assert test_analytics
+                        test_analytics.builds.add_build_job_failure(str(part))
+                    # raise
+                    # We could also keep running, but then runtime is still
+                    # slow when a test fails, and the annotation only shows up
+                    # after the test finished:
+                    exceptions.append(e)
+        finally:
+            if buildkite.is_in_buildkite():
+                assert test_analytics
+                test_analytics.database_connector.submit_update_statements()
+        if exceptions:
+            print(f"Further exceptions were raised:\n{exceptions[1:]}")
+            raise exceptions[0]

misc/python/materialize/test_analytics/data/build/build_data_storage.py

@@ -173,3 +173,56 @@ def update_build_job_success(
         )
 
         self.database_connector.add_update_statements(sql_statements)
+
+    def add_build_job_failure(
+        self,
+        part: str,
+    ) -> None:
+        job_id = buildkite.get_var(BuildkiteEnvVar.BUILDKITE_JOB_ID)
+
+        sql_statements = []
+        sql_statements.append(
+            f"""
+            INSERT INTO build_job_failure
+            (
+                build_job_id,
+                part
+            )
+            VALUES
+            (
+                {as_sanitized_literal(job_id)},
+                {as_sanitized_literal(part)}
+            )
+            """
+        )
+
+        self.database_connector.add_update_statements(sql_statements)
+
+    def get_part_priorities(self, timeout: int) -> dict[str, int]:
+        branch = buildkite.get_var(BuildkiteEnvVar.BUILDKITE_BRANCH)
+        build_step_key = buildkite.get_var(BuildkiteEnvVar.BUILDKITE_STEP_KEY)
+        with self.database_connector.create_cursor() as cur:
+            cur.execute('SET cluster = "test_analytics"')
+            cur.execute(f"SET statement_timeout = '{timeout}s'".encode("utf-8"))
+            # 2 for failures in this PR
+            # 1 for failed recently in CI
+            cur.execute(
+                f"""
+            SELECT part, MAX(prio)
+            FROM (
+                SELECT part, 2 AS prio
+                FROM mv_build_job_failed_on_branch
+                WHERE branch = {as_sanitized_literal(branch)}
+                  AND build_step_key = {as_sanitized_literal(build_step_key)}
+              UNION
+                SELECT part, 1 AS prio
+                FROM mv_build_job_failed
+                WHERE build_step_key = {as_sanitized_literal(build_step_key)}
+            )
+            GROUP BY part;
+            """.encode(
+                    "utf-8"
+                )
+            )
+            results = cur.fetchall()
+            return {part: prio for part, prio in results}

misc/python/materialize/test_analytics/setup/tables/03-build-job-failure.sql

@@ -0,0 +1,17 @@
+-- Copyright Materialize, Inc. and contributors. All rights reserved.
+--
+-- Use of this software is governed by the Business Source License
+-- included in the LICENSE file at the root of this repository.
+--
+-- As of the Change Date specified in that file, in accordance with
+-- the Business Source License, use of this software will be governed
+-- by the Apache License, Version 2.0.
+
+-- part of a build job that failed, can be a testdrive file or a workflow
+CREATE TABLE build_job_failure (
+    build_job_id TEXT NOT NULL,
+    part TEXT NOT NULL
+);
+
+ALTER TABLE build_job_failure OWNER TO qa;
+GRANT SELECT, INSERT, UPDATE ON TABLE build_job_failure TO "hetzner-ci";

misc/python/materialize/test_analytics/setup/views/03-build-failure.sql

@@ -0,0 +1,27 @@
+-- Copyright Materialize, Inc. and contributors. All rights reserved.
+--
+-- Use of this software is governed by the Business Source License
+-- included in the LICENSE file at the root of this repository.
+--
+-- As of the Change Date specified in that file, in accordance with
+-- the Business Source License, use of this software will be governed
+-- by the Apache License, Version 2.0.
+
+CREATE OR REPLACE MATERIALIZED VIEW mv_build_job_failed_on_branch AS
+SELECT build_step_key, branch, part
+FROM build
+JOIN build_job ON build.build_id = build_job.build_id
+JOIN build_job_failure ON build_job.build_job_id = build_job_failure.build_job_id;
+
+CREATE OR REPLACE MATERIALIZED VIEW mv_build_job_failed AS
+SELECT build_step_key, part
+FROM build_job
+JOIN build_job_failure ON build_job.build_job_id = build_job_failure.build_job_id;
+
+CREATE DEFAULT INDEX ON mv_build_job_failed_on_branch;
+CREATE DEFAULT INDEX ON mv_build_job_failed;
+
+ALTER MATERIALIZED VIEW mv_build_job_failed_on_branch OWNER TO qa;
+GRANT SELECT ON TABLE mv_build_job_failed_on_branch TO "hetzner-ci";
+ALTER MATERIALIZED VIEW mv_build_job_failed OWNER TO qa;
+GRANT SELECT ON TABLE mv_build_job_failed TO "hetzner-ci";

test/0dt/mzcompose.py

@@ -29,6 +29,7 @@
     Materialized,
 )
 from materialize.mzcompose.services.mysql import MySql
+from materialize.mzcompose.services.mz import Mz
 from materialize.mzcompose.services.postgres import (
     CockroachOrPostgresMetadata,
     Postgres,
@@ -49,6 +50,7 @@
     Kafka(),
     SchemaRegistry(),
     CockroachOrPostgresMetadata(),
+    Mz(app_password=""),
     Materialized(
         name="mz_old",
         sanity_restart=False,
@@ -77,14 +79,17 @@
 
 
 def workflow_default(c: Composition) -> None:
-    for name in buildkite.shard_list(
-        list(c.workflows.keys()), lambda workflow: workflow
-    ):
+    def process(name: str) -> None:
         if name == "default":
-            continue
+            return
         with c.test_case(name):
             c.workflow(name)
 
+    workflows = buildkite.shard_list(
+        list(c.workflows.keys()), lambda workflow: workflow
+    )
+    c.test_parts(workflows, process)
+
 
 def workflow_read_only(c: Composition) -> None:
     """Verify read-only mode."""

test/aws-localstack/mzcompose.py

@@ -25,6 +25,7 @@
 )
 from materialize.mzcompose.services.localstack import Localstack
 from materialize.mzcompose.services.materialized import Materialized
+from materialize.mzcompose.services.mz import Mz
 from materialize.mzcompose.services.testdrive import Testdrive
 
 ENVIRONMENT_NAME = f"environment-{DEFAULT_ORG_ID}-{DEFAULT_ORDINAL}"
@@ -43,6 +44,7 @@
 
 SERVICES = [
     Localstack(),
+    Mz(app_password=""),
     Materialized(
         depends_on=["localstack"],
         environment_extra=[
@@ -66,10 +68,13 @@
 
 
 def workflow_default(c: Composition) -> None:
-    for name in ["secrets-manager", "aws-connection", "copy-to-s3"]:
+    def process(name: str) -> None:
         with c.test_case(name):
             c.workflow(name)
 
+    workflows = ["secrets-manager", "aws-connection", "copy-to-s3"]
+    c.test_parts(workflows, process)
+
 
 def workflow_secrets_manager(c: Composition) -> None:
     c.up("localstack")

test/balancerd/mzcompose.py

@@ -33,6 +33,7 @@
 from materialize.mzcompose.services.balancerd import Balancerd
 from materialize.mzcompose.services.frontegg import FronteggMock
 from materialize.mzcompose.services.materialized import Materialized
+from materialize.mzcompose.services.mz import Mz
 from materialize.mzcompose.services.test_certs import TestCerts
 from materialize.mzcompose.services.testdrive import Testdrive
 
@@ -118,6 +119,7 @@ def app_password(email: str) -> str:
             "secrets:/secrets",
         ],
     ),
+    Mz(app_password=""),
     Materialized(
         options=[
             # Enable TLS on the public port to verify that balancerd is connecting to the balancerd
@@ -185,11 +187,14 @@ def pg8000_sql_cursor(
 def workflow_default(c: Composition) -> None:
     c.down(destroy_volumes=True)
 
-    for name in c.workflows:
+    def process(name: str) -> None:
         if name in ["default", "plaintext"]:
-            continue
+            return
         with c.test_case(name):
             c.workflow(name)
+
+    c.test_parts(list(c.workflows.keys()), process)
+
     with c.test_case("plaintext"):
         c.workflow("plaintext")
 

test/bounded-memory/mzcompose.py

@@ -1091,16 +1091,14 @@ class KafkaScenario(Scenario):
 
 
 def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
-    for name in c.workflows:
-        if name == "default":
-            continue
-
-        if name == "minimization-search":
-            continue
-
+    def process(name: str) -> None:
+        if name in ["default", "minimization-search"]:
+            return
         with c.test_case(name):
             c.workflow(name)
 
+    c.test_parts(list(c.workflows.keys()), process)
+
 
 def workflow_main(c: Composition, parser: WorkflowArgumentParser) -> None:
     """Process various datasets in a memory-constrained environment in order

test/chbench/mzcompose.py

@@ -19,6 +19,7 @@
 from materialize.mzcompose.services.materialized import Materialized
 from materialize.mzcompose.services.metabase import Metabase
 from materialize.mzcompose.services.mysql import MySql
+from materialize.mzcompose.services.mz import Mz
 from materialize.mzcompose.services.schema_registry import SchemaRegistry
 from materialize.mzcompose.services.zookeeper import Zookeeper
 
@@ -28,6 +29,7 @@
     SchemaRegistry(),
     Debezium(),
     MySql(root_password="rootpw"),
+    Mz(app_password=""),
     Materialized(),
     Metabase(),
     Service(
@@ -42,13 +44,14 @@
 
 
 def workflow_default(c: Composition) -> None:
-    for name in c.workflows:
+    def process(name: str) -> None:
         if name == "default":
-            continue
-
+            return
         with c.test_case(name):
             c.workflow(name)
 
+    c.test_parts(list(c.workflows.keys()), process)
+
 
 def workflow_no_load(c: Composition, parser: WorkflowArgumentParser) -> None:
     """Run CH-benCHmark without any load on Materialize"""

test/cloud-canary/mzcompose.py

@@ -243,7 +243,7 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
 
     args = parser.parse_args()
 
-    globs = list(
+    files = list(
         itertools.chain.from_iterable(
             [
                 glob.glob(file_glob, root_dir=MZ_ROOT / "test" / "cloud-canary")
@@ -275,15 +275,18 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
         # Takes about 40 min to spin up
         redpanda = (
             Redpanda(c, cleanup=args.cleanup)
-            if any(["redpanda" in filename for filename in globs])
+            if any(["redpanda" in filename for filename in files])
             else None
         )
 
         try:
             print("Running .td files ...")
             td(c, text="> CREATE CLUSTER canary_sources SIZE '25cc'")
-            for filename in globs:
+
+            def process(filename: str) -> None:
                 td(c, filename, redpanda=redpanda)
+
+            c.test_parts(files, process)
             test_failed = False
         finally:
             if args.cleanup and redpanda is not None: