Test full upgrade path of self managed releases (#33985)

<!--
Describe the contents of the PR briefly but completely.

If you write detailed commit messages, it is acceptable to copy/paste
them
here, or write "see commit messages for details." If there is only one
commit
in the PR, GitHub will have already added its commit message above.
-->

### Motivation

Partially does
MaterializeInc/database-issues#9861
- Specifically the "Paths" part

The plan is to have this cover all releases(including patches) from
25.0.0 to 26.0.0.

Bumped the parallelism but looking for guidance on if I should do this /
if we should keep this in the nightly
<!--
Which of the following best describes the motivation behind this PR?

  * This PR fixes a recognized bug.

    [Ensure issue is linked somewhere.]

  * This PR adds a known-desirable feature.

    [Ensure issue is linked somewhere.]

  * This PR fixes a previously unreported bug.

    [Describe the bug in detail, as if you were filing a bug report.]

  * This PR adds a feature that has not yet been specified.

[Write a brief specification for the feature, including justification
for its inclusion in Materialize, as if you were writing the original
     feature specification.]

   * This PR refactors existing code.

[Describe what was wrong with the existing code, if it is not obvious.]
-->

### Tips for reviewer

<!--
Leave some tips for your reviewer, like:

    * The diff is much smaller if viewed with whitespace hidden.
    * [Some function/module/file] deserves extra attention.
* [Some function/module/file] is pure code movement and only needs a
skim.

Delete this section if no tips.
-->

### Checklist

- [ ] This PR has adequate test coverage / QA involvement has been duly
considered. ([trigger-ci for additional test/nightly
runs](https://trigger-ci.dev.materialize.com/))
- [ ] This PR has an associated up-to-date [design
doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md),
is a design doc
([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)),
or is sufficiently small to not require a design.
  <!-- Reference the design in the description. -->
- [ ] If this PR evolves [an existing `$T ⇔ Proto$T`
mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md)
(possibly in a backwards-incompatible way), then it is tagged with a
`T-proto` label.
- [ ] If this PR will require changes to cloud orchestration or tests,
there is a companion cloud PR to account for those changes that is
tagged with the release-blocker label
([example](MaterializeInc/cloud#5021)).
<!-- Ask in #team-cloud on Slack if you need help preparing the cloud
PR. -->
- [ ] If this PR includes major [user-facing behavior
changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note),
I have pinged the relevant PM to schedule a changelog post.

Author: SangJunBak

ci/nightly/pipeline.template.yml

@@ -1110,29 +1110,29 @@ steps:
               args: [-m=long, test/cloudtest/test_upgrade.py, --no-test-parallelism]
         sanitizer: skip
         skip: "TODO(def-) Reenable in one version when labels are fixed in old version"
-      - id: checks-self-managed-25-2-upgrade-manipulate-before-upgrade
-        label: "Checks Self-Managed upgrade across all v25.2 patch releases, manipulate before upgrade"
+      - id: checks-self-managed-linear-upgrade-path-manipulate-before-upgrade
+        label: "Checks Self-Managed upgrade across all supported patch releases, manipulate before upgrade"
         depends_on: build-x86_64
         timeout_in_minutes: 180
-        parallelism: 3
+        parallelism: 6
         agents:
           queue: hetzner-x86-64-8cpu-16gb
         plugins:
           - ./ci/plugins/mzcompose:
               composition: platform-checks
-              args: [--scenario=SelfManagedv25Point2_Upgrade_ManipulateBeforeUpgrade, "--seed=$BUILDKITE_JOB_ID"]
+              args: [--scenario=SelfManagedLinearUpgradePathManipulateBeforeUpgrade, "--seed=$BUILDKITE_JOB_ID"]
 
-      - id: checks-self-managed-25-2-upgrade-manipulate-during-upgrade
-        label: "Checks Self-Managed upgrade across all v25.2 patch releases, manipulate during upgrade"
+      - id: checks-self-managed-linear-upgrade-path-manipulate-during-upgrade
+        label: "Checks Self-Managed upgrade across all supported patch releases, manipulate during upgrade"
         depends_on: build-x86_64
         timeout_in_minutes: 180
-        parallelism: 3
+        parallelism: 6
         agents:
           queue: hetzner-x86-64-8cpu-16gb
         plugins:
           - ./ci/plugins/mzcompose:
               composition: platform-checks
-              args: [--scenario=SelfManagedv25Point2_Upgrade_ManipulateDuringUpgrade, "--seed=$BUILDKITE_JOB_ID"]
+              args: [--scenario=SelfManagedLinearUpgradePathManipulateDuringUpgrade, "--seed=$BUILDKITE_JOB_ID"]
 
   - group: "K8s node recovery cloudtest"
     key: k8s-node-recovery

misc/python/materialize/checks/scenarios_upgrade.py

@@ -28,9 +28,9 @@
 from materialize.mzcompose import get_default_system_parameters
 from materialize.mzcompose.services.materialized import LEADER_STATUS_HEALTHCHECK
 from materialize.version_list import (
-    fetch_self_managed_versions,
     get_published_minor_mz_versions,
     get_self_managed_versions,
+    get_supported_self_managed_versions,
 )
 
 # late initialization
@@ -515,48 +515,62 @@ def actions(self) -> list[Action]:
         ]
 
 
-def upgrade_actions(
+@dataclass
+class MzServiceUpgradeInfo:
+    # Version of the MZ instance
+    version: MzVersion | None
+    # Name of the docker service
+    service_name: str
+    # Generation of the MZ instance
+    deploy_generation: int
+    system_parameter_defaults: dict[str, str]
+
+
+def create_mz_service_upgrade_info_list(
+    versions: list[MzVersion | None],
+) -> list[MzServiceUpgradeInfo]:
+    # We use the first version to get the system parameters since the defaults for
+    # newer versions include cutting edge features than can break backwards compatibility.
+    # TODO (multiversion1): Get minimal system parameters by default to avoid cutting edge features.
+    system_parameter_defaults = get_default_system_parameters(versions[0])
+    return [
+        MzServiceUpgradeInfo(
+            version=version,
+            service_name=f"mz_{(i % 2) + 1}",
+            deploy_generation=i,
+            system_parameter_defaults=system_parameter_defaults,
+        )
+        for i, version in enumerate(versions)
+    ]
+
+
+def upgrade_service_actions(
     scenario: Scenario,
-    version: MzVersion | None,
-    deploy_generation: int,
-    service_name: str,
-    previous_service_name: str,
-    system_parameter_defaults: dict[str, str],
-    should_validate: bool = True,
+    service_info: MzServiceUpgradeInfo,
+    previous_service_info: MzServiceUpgradeInfo,
 ) -> list[Action]:
     return [
         start_mz_read_only(
             scenario,
-            tag=version,
-            deploy_generation=deploy_generation,
-            mz_service=service_name,
-            system_parameter_defaults=system_parameter_defaults,
+            tag=service_info.version,
+            deploy_generation=service_info.deploy_generation,
+            mz_service=service_info.service_name,
+            system_parameter_defaults=service_info.system_parameter_defaults,
         ),
-        WaitReadyMz(service_name),
-        PromoteMz(service_name),
-        *([Validate(scenario, mz_service=service_name)] if should_validate else []),
-        KillMz(capture_logs=True, mz_service=previous_service_name),
+        WaitReadyMz(service_info.service_name),
+        PromoteMz(service_info.service_name),
+        # Cleanup the previous service
+        KillMz(capture_logs=True, mz_service=previous_service_info.service_name),
     ]
 
 
-@dataclass
-class MzServiceUpgradeInfo:
-    version: MzVersion | None
-    service_name: str
-
-
-def get_self_managed_v25_2_versions() -> list[MzVersion]:
-    self_managed_versions = fetch_self_managed_versions()
-    return sorted(
-        [
-            v.version
-            for v in self_managed_versions
-            if v.helm_version.major == 25 and v.helm_version.minor == 2
-        ]
+def print_upgrade_path(versions: list[MzVersion | None]):
+    print(
+        f"Upgrading through versions {[str(version) if version is not None else "current" for version in versions]}"
     )
 
 
-class SelfManagedv25Point2_Upgrade_ManipulateBeforeUpgrade(Scenario):
+class SelfManagedLinearUpgradePathManipulateBeforeUpgrade(Scenario):
     """
     Upgrade from the oldest v25.2 patch release to the latest v25.2 patch release to main.
     Run all manipulation phases before any upgrades.
@@ -569,63 +583,52 @@ def __init__(
         features: Features,
         seed: str | None = None,
     ):
-        self.v25_2_versions = get_self_managed_v25_2_versions()
+        self.self_managed_versions = get_supported_self_managed_versions()
         super().__init__(checks, executor, features, seed)
 
     def base_version(self) -> MzVersion:
-        return self.v25_2_versions[0]
+        return self.self_managed_versions[0]
 
     def actions(self) -> list[Action]:
-        print(f"Upgrading from tag {self.base_version()}")
-        system_parameter_defaults = get_default_system_parameters(self.base_version())
-        actions = []
-        versions = self.v25_2_versions + [None]
-
-        mz_services = [
-            MzServiceUpgradeInfo(
-                version=version,
-                # We alternate between mz_1 and mz_2
-                service_name=f"mz_{(i % 2) + 1}",
-            )
-            for i, version in enumerate(versions)
+        versions = self.self_managed_versions + [None]
+
+        print_upgrade_path(versions)
+
+        mz_services = create_mz_service_upgrade_info_list(versions)
+
+        actions = [
+            StartMz(
+                self,
+                tag=mz_services[0].version,
+                mz_service=mz_services[0].service_name,
+                system_parameter_defaults=mz_services[0].system_parameter_defaults,
+            ),
+            Initialize(self, mz_service=mz_services[0].service_name),
+            Manipulate(self, phase=1, mz_service=mz_services[0].service_name),
+            Manipulate(self, phase=2, mz_service=mz_services[0].service_name),
+            Validate(self, mz_service=mz_services[0].service_name),
         ]
 
-        for i, service in enumerate(mz_services):
-            if i == 0:
-                actions.extend(
-                    [
-                        StartMz(
-                            self,
-                            tag=service.version,
-                            mz_service=service.service_name,
-                            system_parameter_defaults=system_parameter_defaults,
-                        ),
-                        Initialize(self, mz_service=service.service_name),
-                        Manipulate(self, phase=1, mz_service=service.service_name),
-                        Manipulate(self, phase=2, mz_service=service.service_name),
-                        Validate(self, mz_service=service.service_name),
-                    ]
-                )
-            else:
-                # Because upgrades start on generation 1, we can set it to the index
-                deploy_generation = i
-                actions.extend(
-                    upgrade_actions(
-                        self,
-                        service.version,
-                        deploy_generation,
-                        service.service_name,
-                        mz_services[i - 1].service_name,
-                        system_parameter_defaults=system_parameter_defaults,
-                    )
+        for i, service_info in enumerate[MzServiceUpgradeInfo](
+            mz_services[1:], start=1
+        ):
+            actions.extend(
+                upgrade_service_actions(
+                    self,
+                    service_info=service_info,
+                    previous_service_info=mz_services[i - 1],
                 )
+                + [
+                    Validate(self, mz_service=service_info.service_name),
+                ]
+            )
 
         return actions
 
 
-class SelfManagedv25Point2_Upgrade_ManipulateDuringUpgrade(Scenario):
+class SelfManagedLinearUpgradePathManipulateDuringUpgrade(Scenario):
     """
-    Upgrade from the oldest v25.2 patch release to the latest v25.2 patch release to main.
+    Upgrade from the oldest Self-Managed version to the latest Self-Managed version to main.
     Run the first manipulation phase before all upgrades and the second during the upgrade.
     """
 
@@ -636,73 +639,54 @@ def __init__(
         features: Features,
         seed: str | None = None,
     ):
-        self.v25_2_versions = get_self_managed_v25_2_versions()
+        self.self_managed_versions = get_supported_self_managed_versions()
         super().__init__(checks, executor, features, seed)
 
     def base_version(self) -> MzVersion:
-        return self.v25_2_versions[0]
+        return self.self_managed_versions[0]
 
     def actions(self) -> list[Action]:
-        print(f"Upgrading from tag {self.base_version()}")
-        system_parameter_defaults = get_default_system_parameters(self.base_version())
+        versions = self.self_managed_versions + [None]
 
-        actions = []
-        versions = self.v25_2_versions + [None]
+        print_upgrade_path(versions)
 
-        mz_services = [
-            MzServiceUpgradeInfo(
-                version=version,
-                # We alternate between mz_1 and mz_2
-                service_name=f"mz_{(i % 2) + 1}",
-            )
-            for i, version in enumerate(versions)
+        mz_services = create_mz_service_upgrade_info_list(
+            versions,
+        )
+
+        actions = [
+            StartMz(
+                self,
+                tag=mz_services[0].version,
+                mz_service=mz_services[0].service_name,
+                system_parameter_defaults=mz_services[0].system_parameter_defaults,
+            ),
+            Initialize(self, mz_service=mz_services[0].service_name),
+            Manipulate(self, phase=1, mz_service=mz_services[0].service_name),
         ]
 
-        for i, service in enumerate(mz_services):
-            if i == 0:
+        for i, service_info in enumerate[MzServiceUpgradeInfo](
+            mz_services[1:], start=1
+        ):
+            actions.extend(
+                upgrade_service_actions(
+                    self,
+                    service_info=service_info,
+                    previous_service_info=mz_services[i - 1],
+                )
+            )
+
+            if i == 1:
+                # Manipulate the MZ instance after the first upgrade
                 actions.extend(
                     [
-                        StartMz(
-                            self,
-                            tag=service.version,
-                            mz_service=service.service_name,
-                            system_parameter_defaults=system_parameter_defaults,
-                        ),
-                        Initialize(self, mz_service=service.service_name),
-                        Manipulate(self, phase=1, mz_service=service.service_name),
+                        Manipulate(self, phase=2, mz_service=service_info.service_name),
+                        Validate(self, mz_service=service_info.service_name),
                     ]
                 )
             else:
-                # Because upgrades start on generation 1, we can set it to the index
-                deploy_generation = i
-
-                if i == 1:
-                    # Manipulate the MZ instance after the first upgrade.
-                    actions.extend(
-                        upgrade_actions(
-                            self,
-                            service.version,
-                            deploy_generation,
-                            service.service_name,
-                            mz_services[i - 1].service_name,
-                            system_parameter_defaults=system_parameter_defaults,
-                            # We can't validate on the first upgrade since all manipulations haven't completed yet.
-                            should_validate=False,
-                        )
-                        + [
-                            Manipulate(self, phase=2, mz_service=service.service_name),
-                        ]
-                    )
-                else:
-                    actions.extend(
-                        upgrade_actions(
-                            self,
-                            service.version,
-                            deploy_generation,
-                            service.service_name,
-                            mz_services[i - 1].service_name,
-                            system_parameter_defaults=system_parameter_defaults,
-                        )
-                    )
+                actions.append(
+                    Validate(self, mz_service=service_info.service_name),
+                )
 
         return actions