ci: compare benchmarks against common-ancestor (#22355)

ci/nightly/pipeline.template.yml

@@ -39,7 +39,7 @@ steps:
       queue: builder-linux-x86_64
 
   - id: feature-benchmark
-    label: "Feature benchmark against 'latest'"
+    label: "Feature benchmark against merge base or 'latest'"
     timeout_in_minutes: 360
     agents:
       queue: linux-x86_64-large
@@ -49,7 +49,8 @@ steps:
           composition: feature-benchmark
           args:
              - --other-tag
-             - latest
+             # common-ancestor will default to latest if not in a PR
+             - common-ancestor
 
   - id: kafka-matrix
     label: Kafka smoke test against previous Kafka versions

misc/buildkite/git.bash

@@ -16,8 +16,8 @@ set -euo pipefail
 export BUILDKITE_REPO_REF="${BUILDKITE_REPO_REF:-origin}"
 export BUILDKITE_PULL_REQUEST_BASE_BRANCH="${BUILDKITE_PULL_REQUEST_BASE_BRANCH:-main}"
 
-configure_git_user() {
-  if [[ "$BUILDKITE" == "true" ]]; then
+configure_git_user_if_in_buildkite() {
+  if [[ "${BUILDKITE:-}" == "true" ]]; then
     ci_collapsed_heading "Configure git"
     run git config --global user.email "[email protected]"
     run git config --global user.name "Buildkite"
@@ -30,7 +30,7 @@ fetch_pr_target_branch() {
 }
 
 merge_pr_target_branch() {
-  configure_git_user
+  configure_git_user_if_in_buildkite
 
   ci_collapsed_heading "Merge target branch"
   run git merge "$BUILDKITE_REPO_REF"/"$BUILDKITE_PULL_REQUEST_BASE_BRANCH" --message "Merge"

misc/python/materialize/benchmark_utils.py

@@ -0,0 +1,23 @@
+# 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.
+
+"""Benchmark utilities."""
+
+from materialize import buildkite
+
+
+def resolve_tag_of_common_ancestor(tag_when_on_default_branch: str = "latest") -> str:
+    if buildkite.is_on_default_branch():
+        print(f"On default branch, using {tag_when_on_default_branch} as tag")
+        return tag_when_on_default_branch
+    else:
+        commit_hash = buildkite.get_merge_base()
+        tag = f"unstable-{commit_hash}"
+        print(f"Resolved common-ancestor to {tag} (commit: {commit_hash})")
+        return tag

misc/python/materialize/buildkite.py

@@ -0,0 +1,48 @@
+# 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.
+
+"""Buildkite utilities."""
+
+import os
+
+from materialize import git
+
+
+def is_in_buildkite() -> bool:
+    return os.getenv("BUILDKITE", "false") == "true"
+
+
+def is_in_pull_request() -> bool:
+    """
+    Note that this does not work in (manually triggered) nightly builds because they don't carry this information!
+    Consider using #is_on_default_branch() instead.
+    """
+    return os.getenv("BUILDKITE_PULL_REQUEST", "false") != "false"
+
+
+def is_on_default_branch() -> bool:
+    current_branch = os.getenv("BUILDKITE_BRANCH", "unknown")
+    default_branch = os.getenv("BUILDKITE_PIPELINE_DEFAULT_BRANCH", "main")
+    return current_branch == default_branch
+
+
+def get_pull_request_base_branch(fallback: str = "main"):
+    return os.getenv("BUILDKITE_PULL_REQUEST_BASE_BRANCH", fallback)
+
+
+def get_pipeline_default_branch(fallback: str = "main"):
+    return os.getenv("BUILDKITE_PIPELINE_DEFAULT_BRANCH", fallback)
+
+
+def get_merge_base(remote="origin") -> str:
+    base_branch = get_pull_request_base_branch() or get_pipeline_default_branch()
+    merge_base = git.get_common_ancestor_commit(
+        remote, branch=base_branch, fetch_branch=True
+    )
+    return merge_base

misc/python/materialize/cli/ci_coverage_pr_report.py

@@ -16,7 +16,7 @@
 
 import junit_xml
 
-from materialize import MZ_ROOT, ci_util
+from materialize import MZ_ROOT, buildkite, ci_util
 
 # - None value indicates that this line is interesting, but we don't know yet
 #   if it can actually be covered.
@@ -55,18 +55,8 @@ def find_modified_lines() -> Coverage:
     """
     Find each line that has been added or modified in the current pull request.
     """
-    base_branch = os.getenv("BUILDKITE_PULL_REQUEST_BASE_BRANCH", "main") or os.getenv(
-        "BUILDKITE_PIPELINE_DEFAULT_BRANCH", "main"
-    )
-    # Make sure we have the latest state to correctly identify the merge base
-    subprocess.run(["git", "fetch", "origin", base_branch], check=True)
-    result = subprocess.run(
-        ["git", "merge-base", "HEAD", f"origin/{base_branch}"],
-        check=True,
-        capture_output=True,
-    )
-    merge_base = result.stdout.strip()
-    print(f"Merge base: {merge_base.decode('utf-8')}")
+    merge_base = buildkite.get_merge_base()
+    print(f"Merge base: {merge_base}")
     result = subprocess.run(
         ["git", "diff", "-U0", merge_base], check=True, capture_output=True
     )

misc/python/materialize/git.py

@@ -95,7 +95,7 @@ def get_version_tags(*, fetch: bool = True, prefix: str = "v") -> list[Version]:
             tag as a version.
     """
     if fetch:
-        _fetch()
+        _fetch(all_remotes=True, include_tags=True, force=True)
     tags = []
     for t in spawn.capture(["git", "tag"]).splitlines():
         if not t.startswith(prefix):
@@ -139,13 +139,52 @@ def describe() -> str:
     return spawn.capture(["git", "describe"]).strip()
 
 
-def fetch() -> str:
-    """Fetch from all configured default fetch remotes"""
-    return spawn.capture(["git", "fetch", "--all", "--tags", "--force"]).strip()
+def fetch(
+    remote: str | None = None,
+    all_remotes: bool = False,
+    include_tags: bool = False,
+    force: bool = False,
+    branch: str | None = None,
+) -> str:
+    """Fetch from remotes"""
+
+    if remote and all_remotes:
+        raise RuntimeError("all_remotes must be false when a remote is specified")
+
+    if branch and not remote:
+        raise RuntimeError("remote must be specified when a branch is specified")
+
+    command = ["git", "fetch"]
+
+    if remote:
+        command.append(remote)
+
+    if branch:
+        command.append(branch)
+
+    if all_remotes:
+        command.append("--all")
+
+    if include_tags:
+        command.append("--tags")
+
+    if force:
+        command.append("--force")
+
+    return spawn.capture(command).strip()
 
 
 _fetch = fetch  # renamed because an argument shadows the fetch name in get_tags
 
+
+def get_common_ancestor_commit(remote: str, branch: str, fetch_branch: bool) -> str:
+    if fetch_branch:
+        fetch(remote=remote, branch=branch)
+
+    command = ["git", "merge-base", "HEAD", f"{remote}/{branch}"]
+    return spawn.capture(command).strip()
+
+
 # Work tree mutation
 
 

misc/python/materialize/mzcompose/composition.py

@@ -625,18 +625,35 @@ def exec(
         )
 
     def pull_if_variable(self, services: list[str], max_tries: int = 2) -> None:
-        """Pull fresh service images in case the tag indicates thee underlying image may change over time.
+        """Pull fresh service images in case the tag indicates the underlying image may change over time.
 
         Args:
             services: List of service names
         """
 
         for service in services:
             if "image" in self.compose["services"][service] and any(
-                self.compose["services"][service]["image"].endswith(tag)
+                tag in self.compose["services"][service]["image"]
                 for tag in [":latest", ":unstable", ":rolling"]
             ):
-                self.invoke("pull", service, max_tries=max_tries)
+                self.pull_single_image_by_service_name(service, max_tries=max_tries)
+
+    def pull_single_image_by_service_name(
+        self, service_name: str, max_tries: int
+    ) -> None:
+        self.invoke("pull", service_name, max_tries=max_tries)
+
+    def try_pull_service_image(self, service: Service, max_tries: int = 2) -> bool:
+        """Tries to pull the specified image and returns if this was successful."""
+
+        with self.override(service):
+            try:
+                self.pull_single_image_by_service_name(
+                    service.name, max_tries=max_tries
+                )
+                return True
+            except UIError:
+                return False
 
     def up(
         self,

misc/python/materialize/scalability/endpoints.py

@@ -94,9 +94,17 @@ def up(self) -> None:
 
 
 class MaterializeContainer(MaterializeNonRemote):
-    def __init__(self, composition: Composition, image: str | None = None) -> None:
+    def __init__(
+        self,
+        composition: Composition,
+        image: str | None = None,
+        alternative_image: str | None = None,
+    ) -> None:
         self.composition = composition
         self.image = image
+        self.alternative_image = (
+            alternative_image if image != alternative_image else None
+        )
         self._port: int | None = None
         super().__init__()
 
@@ -109,10 +117,26 @@ def priv_port(self) -> int:
 
     def up(self) -> None:
         self.composition.down(destroy_volumes=True)
+
+        if (
+            self.image is not None
+            and self.alternative_image is not None
+            and not self.composition.try_pull_service_image(
+                Materialized(image=self.image)
+            )
+        ):
+            # explicitly specified image cannot be found and alternative exists
+            print(
+                f"Unable to find image {self.image}, proceeding with alternative image {self.alternative_image}!"
+            )
+            self.image = self.alternative_image
+
+        self.up_internal()
+        self.lift_limits()
+
+    def up_internal(self) -> None:
         with self.composition.override(
             Materialized(image=self.image, sanity_restart=False)
         ):
             self.composition.up("materialized")
             self._port = self.composition.default_port("materialized")
-
-        self.lift_limits()

misc/python/materialize/version_list.py

@@ -13,7 +13,7 @@
 
 import frontmatter
 
-from materialize.git import get_version_tags
+from materialize import git
 from materialize.util import MzVersion
 
 MZ_ROOT = Path(os.environ["MZ_ROOT"])
@@ -84,7 +84,7 @@ class VersionsFromGit(VersionList):
 
     def __init__(self) -> None:
         self.versions = list(
-            {MzVersion.from_semver(t) for t in get_version_tags(fetch=True)}
+            {MzVersion.from_semver(t) for t in git.get_version_tags(fetch=True)}
             - INVALID_VERSIONS
         )
         self.versions.sort()