Mdr upgrade scenario

conf/README.md

@@ -212,6 +212,7 @@ higher priority).
 * `skip_ocp_deployment` - Skip the OCP deployment step or not (Default: false)
 * `skip_ocs_deployment` - Skip the OCS deployment step or not (Default: false)
 * `ocs_version` - Version of OCS that is being deployed
+* `acm_version` - Version of acm to be used for this run (applicable mostly to DR scenarios)
 * `vm_template` - VMWare template to use for RHCOS images
 * `fio_storageutilization_min_mbps` - Minimal write speed of FIO used in workload_fio_storageutilization
 * `TF_LOG_LEVEL` - Terraform log level
@@ -362,6 +363,9 @@ Upgrade related configuration data.
 * `ocp_arch` - Architecture type of the OCP image
 * `upgrade_logging_channel` - OCP logging channel to upgrade with
 * `upgrade_ui` - Perform upgrade via UI (Not all the versions are supported, please look at the code)
+* `upgrade_acm_version` - ACM version to which we have to upgrade
+* `upgrade_acm_registry_image` - ACM Image tag from brew which should be used to upgrade 
+example: <brew_registry_url>/rh-osbs/iib:565330
 
 #### AUTH
 

ocs_ci/framework/__init__.py

@@ -37,6 +37,10 @@ class Config:
     COMPONENTS: dict = field(default_factory=dict)
     # Used for multicluster only
     MULTICLUSTER: dict = field(default_factory=dict)
+    # Use this variable to store any arbitrary key/values related
+    # to the upgrade context. Applicable only in the multicluster upgrade
+    # scenario
+    PREUPGRADE_CONFIG: dict = field(default_factory=dict)
 
     def __post_init__(self):
         self.reset()

ocs_ci/framework/conf/default_config.yaml

@@ -362,3 +362,7 @@ MULTICLUSTER:
   acm_cluster: False
   primary_cluster: False
   active_acm_cluster: False
+
+PREUPGRADE_CONFIG:
+  AUTH: null
+  MULTICLUSTER: null

ocs_ci/framework/pytest_customization/marks.py

@@ -14,6 +14,9 @@
     ORDER_BEFORE_OCP_UPGRADE,
     ORDER_BEFORE_UPGRADE,
     ORDER_OCP_UPGRADE,
+    ORDER_MCO_UPGRADE,
+    ORDER_DR_HUB_UPGRADE,
+    ORDER_ACM_UPGRADE,
     ORDER_OCS_UPGRADE,
     ORDER_AFTER_OCP_UPGRADE,
     ORDER_AFTER_OCS_UPGRADE,
@@ -117,12 +120,28 @@
 order_pre_ocp_upgrade = pytest.mark.order(ORDER_BEFORE_OCP_UPGRADE)
 order_pre_ocs_upgrade = pytest.mark.order(ORDER_BEFORE_OCS_UPGRADE)
 order_ocp_upgrade = pytest.mark.order(ORDER_OCP_UPGRADE)
+order_mco_upgrade = pytest.mark.order(ORDER_MCO_UPGRADE)
+order_dr_hub_upgrade = pytest.mark.order(ORDER_DR_HUB_UPGRADE)
+# dr cluster operator order is same as hub operator order except that
+# it's applicable only on the managed clusters
+order_dr_cluster_operator_upgrade = pytest.mark.order(ORDER_DR_HUB_UPGRADE)
+order_acm_upgrade = pytest.mark.order(ORDER_ACM_UPGRADE)
 order_ocs_upgrade = pytest.mark.order(ORDER_OCS_UPGRADE)
 order_post_upgrade = pytest.mark.order(ORDER_AFTER_UPGRADE)
 order_post_ocp_upgrade = pytest.mark.order(ORDER_AFTER_OCP_UPGRADE)
 order_post_ocs_upgrade = pytest.mark.order(ORDER_AFTER_OCS_UPGRADE)
 ocp_upgrade = compose(order_ocp_upgrade, pytest.mark.ocp_upgrade)
+# multicluster orchestrator
+mco_upgrade = compose(order_mco_upgrade, pytest.mark.mco_upgrade)
+# dr hub operator
+dr_hub_upgrade = compose(order_dr_hub_upgrade, pytest.mark.dr_hub_upgrade)
+dr_cluster_operator_upgrade = compose(
+    order_dr_cluster_operator_upgrade, pytest.mark.dr_cluster_operator_upgrade
+)
+# acm operator
+acm_upgrade = compose(order_acm_upgrade, pytest.mark.acm_upgrade)
 ocs_upgrade = compose(order_ocs_upgrade, pytest.mark.ocs_upgrade)
+# pre_*_upgrade markers
 pre_upgrade = compose(order_pre_upgrade, pytest.mark.pre_upgrade)
 pre_ocp_upgrade = compose(
     order_pre_ocp_upgrade,
@@ -132,12 +151,16 @@
     order_pre_ocs_upgrade,
     pytest.mark.pre_ocs_upgrade,
 )
+# post_*_upgrade markers
 post_upgrade = compose(order_post_upgrade, pytest.mark.post_upgrade)
 post_ocp_upgrade = compose(order_post_ocp_upgrade, pytest.mark.post_ocp_upgrade)
 post_ocs_upgrade = compose(order_post_ocs_upgrade, pytest.mark.post_ocs_upgrade)
 
 upgrade_marks = [
     ocp_upgrade,
+    mco_upgrade,
+    dr_hub_upgrade,
+    acm_upgrade,
     ocs_upgrade,
     pre_upgrade,
     pre_ocp_upgrade,
@@ -685,3 +708,12 @@ def get_current_test_marks():
     config.DEPLOYMENT.get("kms_deployment") is True,
     reason="This test is not supported for KMS deployment.",
 )
+
+# Mark the test with marker below to allow re-tries in ceph health fixture
+# for known issues when waiting in re-balance and flip flop from health OK
+# to 1-2 PGs waiting to be Clean
+ceph_health_retry = pytest.mark.ceph_health_retry
+
+# Mark for Multicluster upgrade scenarios
+config_index = pytest.mark.config_index
+multicluster_roles = pytest.mark.multicluster_roles

ocs_ci/framework/pytest_customization/ocscilib.py

@@ -263,6 +263,16 @@ def pytest_addoption(parser):
             "(e.g. quay.io/rhceph-dev/ocs-olm-operator:latest-4.3)"
         ),
     )
+    parser.addoption(
+        "--acm-version",
+        dest="acm_version",
+        help="acm version(e.g. 2.8) to be used for the current run",
+    )
+    parser.addoption(
+        "--upgrade-acm-version",
+        dest="upgrade_acm_version",
+        help="acm version to upgrade(e.g. 2.8), use only with DR upgrade scenario",
+    )
     parser.addoption(
         "--flexy-env-file", dest="flexy_env_file", help="Path to flexy environment file"
     )
@@ -567,6 +577,11 @@ def process_cluster_cli_params(config):
     upgrade_ocs_version = get_cli_param(config, "upgrade_ocs_version")
     if upgrade_ocs_version:
         ocsci_config.UPGRADE["upgrade_ocs_version"] = upgrade_ocs_version
+        # Storing previous version explicitly
+        # Useful in DR upgrade scenarios
+        ocsci_config.UPGRADE["pre_upgrade_ocs_version"] = ocsci_config.ENV_DATA[
+            "ocs_version"
+        ]
     ocs_registry_image = get_cli_param(config, f"ocs_registry_image{suffix}")
     if ocs_registry_image:
         ocsci_config.DEPLOYMENT["ocs_registry_image"] = ocs_registry_image
@@ -661,6 +676,12 @@ def process_cluster_cli_params(config):
     if custom_kubeconfig_location:
         os.environ["KUBECONFIG"] = custom_kubeconfig_location
         ocsci_config.RUN["kubeconfig"] = custom_kubeconfig_location
+    acm_version = get_cli_param(config, "--acm-version")
+    if acm_version:
+        ocsci_config.ENV_DATA["acm_version"] = acm_version
+    upgrade_acm_version = get_cli_param(config, "--upgrade-acm-version")
+    if upgrade_acm_version:
+        ocsci_config.UPGRADE["upgrade_acm_version"] = upgrade_acm_version
 
 
 def pytest_collection_modifyitems(session, config, items):

ocs_ci/ocs/acm_upgrade.py

@@ -0,0 +1,152 @@
+"""
+ACM operator upgrade classes and utilities
+
+"""
+
+import logging
+import tempfile
+from pkg_resources import parse_version
+
+import requests
+
+from ocs_ci.ocs import constants
+from ocs_ci.framework import config
+from ocs_ci.ocs.ocp import OCP
+from ocs_ci.utility import templating
+from ocs_ci.utility.utils import get_ocp_version, get_running_acm_version, run_cmd
+
+
+logger = logging.getLogger(__name__)
+
+
+class ACMUpgrade(object):
+    def __init__(self):
+        self.namespace = constants.ACM_HUB_NAMESPACE
+        self.operator_name = constants.ACM_HUB_OPERATOR_NAME
+        # Since ACM upgrade happens followed by OCP upgrade in the sequence
+        # the config would have loaded upgrade parameters rather than pre-upgrade params
+        # Hence we can't rely on ENV_DATA['acm_version'] for the pre-upgrade version
+        # we need to dynamically find it
+        self.version_before_upgrade = self.get_acm_version_before_upgrade()
+        self.upgrade_version = config.UPGRADE["upgrade_acm_version"]
+        # In case if we are using registry image
+        self.acm_registry_image = config.UPGRADE.get("upgrade_acm_registry_image", "")
+        self.zstream_upgrade = False
+
+    def get_acm_version_before_upgrade(self):
+        running_acm_version = get_running_acm_version()
+        return running_acm_version
+
+    def get_parsed_versions(self):
+        parsed_version_before_upgrade = parse_version(self.version_before_upgrade)
+        parsed_upgrade_version = parse_version(self.upgrade_version)
+
+        return parsed_version_before_upgrade, parsed_upgrade_version
+
+    def run_upgrade(self):
+        self.version_change = (
+            self.get_parsed_versions()[1] > self.get_parsed_versions()[0]
+        )
+        if not self.version_change:
+            self.zstream_upgrade = True
+        # either this would be GA to Unreleased upgrade of same version OR
+        # GA to unreleased upgrade to higher version
+        if self.acm_registry_image and self.version_change:
+            self.upgrade_with_registry()
+            self.annotate_mch()
+            run_cmd(f"oc create -f {constants.ACM_BREW_ICSP_YAML}")
+            self.patch_channel()
+        else:
+            # GA to GA
+            self.upgrade_without_registry()
+        self.validate_upgrade()
+
+    def upgrade_without_registry(self):
+        self.patch_channel()
+
+    def patch_channel(self):
+        """
+        GA to GA acm upgrade
+
+        """
+        patch = f'\'{{"spec": {{"channel": "release-{self.upgrade_version}"}}}}\''
+        self.acm_patch_subscription(patch)
+
+    def upgrade_with_registry(self):
+        """
+        There are 2 scenarios with registry
+        1. GA to unreleased same version (ex: 2.8.1 GA to 2.8.2 Unreleased)
+        2. GA to unreleased higher version (ex: 2.8.9 GA to 2.9.1 Unreleased)
+
+        """
+        if self.acm_registry_image and (not self.version_change):
+            # This is GA to unreleased: same version
+            self.create_catalog_source()
+        else:
+            # This is GA to unreleased version: upgrade to next version
+            self.create_catalog_source()
+            patch = f'\'{{"spec":{{"source": "{constants.ACM_CATSRC_NAME}"}}}}\''
+            self.acm_patch_subscription(patch)
+
+    def annotate_mch(self):
+        annotation = f'\'{{"source": "{constants.ACM_CATSRC_NAME}"}}\''
+        annotate_cmd = (
+            f"oc -n {constants.ACM_HUB_NAMESPACE} annotate mch multiclusterhub "
+            f"installer.open-cluster-management.io/mce-subscription-spec={annotation}"
+        )
+        run_cmd(annotate_cmd)
+
+    def acm_patch_subscription(self, patch):
+        patch_cmd = (
+            f"oc -n {constants.ACM_HUB_NAMESPACE} patch sub advanced-cluster-management "
+            f"-p {patch} --type merge"
+        )
+        run_cmd(patch_cmd)
+
+    def create_catalog_source(self):
+        logger.info("Creating ACM catalog source")
+        acm_catsrc = templating.load_yaml(constants.ACM_CATSRC)
+        if self.acm_registry_image:
+            acm_catsrc["spec"]["image"] = self.acm_registry_image
+        else:
+            # Update catalog source
+            resp = requests.get(constants.ACM_BREW_BUILD_URL, verify=False)
+            raw_msg = resp.json()["raw_messages"]
+            # TODO: Find way to get ocp version before upgrade
+            version_tag = raw_msg[0]["msg"]["pipeline"]["index_image"][
+                f"v{get_ocp_version()}"
+            ].split(":")[1]
+            acm_catsrc["spec"]["image"] = ":".jon(
+                [constants.ACM_BREW_REPO, version_tag]
+            )
+        acm_catsrc["metadata"]["name"] = constants.ACM_CATSRC_NAME
+        acm_catsrc["spec"]["publisher"] = "grpc"
+        acm_data_yaml = tempfile.NamedTemporaryFile(
+            mode="w+", prefix="acm_catsrc", delete=False
+        )
+        templating.dump_data_to_temp_yaml(acm_catsrc, acm_data_yaml.name)
+        run_cmd(f"oc create -f {acm_data_yaml.name}", timeout=300)
+
+    def validate_upgrade(self):
+        acm_sub = OCP(
+            namespace=self.namespace,
+            resource_name=self.operator_name,
+            kind="Subscription.operators.coreos.com",
+        )
+        if not self.zstream_upgrade:
+            acm_prev_channel = f"release-{self.upgrade_version}"
+        else:
+            acm_prev_channel = config.ENV_DATA["acm_hub_channel"]
+        assert acm_sub.get().get("spec").get("channel") == acm_prev_channel
+        logger.info("Checking ACM status")
+        acm_mch = OCP(
+            kind=constants.ACM_MULTICLUSTER_HUB,
+            namespace=constants.ACM_HUB_NAMESPACE,
+        )
+        acm_mch.wait_for_resource(
+            condition=constants.STATUS_RUNNING,
+            resource_name=constants.ACM_MULTICLUSTER_RESOURCE,
+            column="STATUS",
+            timeout=720,
+            sleep=5,
+        )

ocs_ci/ocs/cluster.py

@@ -69,6 +69,19 @@
 logger = logging.getLogger(__name__)
 
 
+class CephClusterMultiCluster(object):
+    """
+    TODO: Implement this class later
+    This class will be used in case of multicluster scenario
+    and current cluster is ACM hence this cluster should point to
+    the ODF which is not in current context
+
+    """
+
+    def __init__(self, cluster_conf=None):
+        pass
+
+
 class CephCluster(object):
     """
     Handles all cluster related operations from ceph perspective
@@ -84,11 +97,17 @@ class CephCluster(object):
         namespace (str): openshift Namespace where this cluster lives
     """
 
-    def __init__(self):
+    def __init__(self, cluster_config=None):
         """
         Cluster object initializer, this object needs to be initialized
         after cluster deployment. However its harmless to do anywhere.
         """
+        if cluster_config:
+            logger.info(
+                "CephClusterMulticluster will be used to handle multicluster case"
+            )
+            return CephClusterMultiCluster()
+
         if config.ENV_DATA["mcg_only_deployment"] or (
             config.ENV_DATA.get("platform") == constants.FUSIONAAS_PLATFORM
             and config.ENV_DATA["cluster_type"].lower() == "consumer"
@@ -1035,6 +1054,18 @@ def delete_blockpool(self, pool_name):
         self.RBD.exec_oc_cmd(f"patch {patch}")
 
 
+class MulticlusterCephHealthMonitor(object):
+    # TODO: This will be a placeholder for now
+    def __init__(self, ceph_cluster=None):
+        pass
+
+    def __enter__(self):
+        pass
+
+    def __exit__(self, exception_type, value, traceback):
+        pass
+
+
 class CephHealthMonitor(threading.Thread):
     """
     Context manager class for monitoring ceph health status of CephCluster.
@@ -1052,6 +1083,8 @@ def __init__(self, ceph_cluster, sleep=5):
             sleep (int): Number of seconds to sleep between health checks.
 
         """
+        if isinstance(ceph_cluster, CephClusterMultiCluster):
+            return MulticlusterCephHealthMonitor()
         self.ceph_cluster = ceph_cluster
         self.sleep = sleep
         self.health_error_status = None