[patch] Add standalone Db2uCluster to Db2uInstance migration pipeline and CLI command

image/cli/app-root/src/.bashrc

@@ -32,6 +32,7 @@ echo "  - ${TEXT_BOLD}${COLOR_GREEN}mas must-gather${TEXT_RESET} to perform must
 echo "  - ${TEXT_BOLD}${COLOR_GREEN}mas uninstall${TEXT_RESET} to uninstall a MAS instance"
 echo "  - ${TEXT_BOLD}${COLOR_GREEN}mas backup${TEXT_RESET} to backup a MAS instance"
 echo "  - ${TEXT_BOLD}${COLOR_GREEN}mas restore${TEXT_RESET} to restore a MAS instance"
+echo "  - ${TEXT_BOLD}${COLOR_GREEN}mas db2ucluster-migration${TEXT_RESET} to migrate Db2uCluster to Db2uInstance"
 
 # None of these functions are tested/supported on s390x /ppc64le yet
 if  [ $arch != "s390x" ] && [ $arch != "ppc64le" ]; then

image/cli/mascli/mas

@@ -372,6 +372,16 @@ case $1 in
     mas-cli restore "$@"
     ;;
 
+  db2ucluster-migration)
+    echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" >> $LOGFILE
+    echo "!! db2ucluster-migration                                                     !!" >> $LOGFILE
+    echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" >> $LOGFILE
+    echo "${TEXT_UNDERLINE}IBM Maximo Application Suite DB2 Migration Manager (v${VERSION})${TEXT_RESET}"
+    echo "Powered by ${COLOR_CYAN}${TEXT_UNDERLINE}https://github.com/ibm-mas/ansible-devops/${TEXT_RESET} and ${COLOR_CYAN}${TEXT_UNDERLINE}https://tekton.dev/${TEXT_RESET}"
+    shift
+    mas-cli db2ucluster-migration "$@"
+    ;;
+
   gitops-bootstrap)
     echo "${TEXT_UNDERLINE}IBM Maximo Application Suite GitOps Manager (v${VERSION})${TEXT_RESET}"
     echo "Powered by ${COLOR_CYAN}${TEXT_UNDERLINE}https://github.com/ibm-mas/gitops/${TEXT_RESET}"

python/src/mas/cli/__main__.py

@@ -50,6 +50,7 @@ def usage() -> None:
             + " - <ForestGreen>mas-cli must-gather</ForestGreen> Collect diagnostic information\n"
             + " - <ForestGreen>mas-cli setup-rbac</ForestGreen>  Set up RBAC resources for MAS installation\n"
             + " - <ForestGreen>mas-cli pre-install</ForestGreen>  Set up pre-install RBAC for MAS\n"
+            + " - <ForestGreen>mas-cli db2ucluster-migration</ForestGreen> Migrate Db2uCluster to Db2uInstance\n"
         )
     )
     print_formatted_text(HTML("For usage information run <ForestGreen>mas-cli [action] --help</ForestGreen>\n"))
@@ -131,6 +132,12 @@ def main() -> None:
             app = MustGatherApp()
             app.mustGather(argv[2:])
             return
+        if function == "db2ucluster-migration":
+            from mas.cli.db2_migration.app import Db2MigrationApp
+
+            app = Db2MigrationApp()
+            app.migrate(argv[2:])
+            return
         if function in ["-h", "--help"]:
             usage()
             raise SystemExit(0)

python/src/mas/cli/db2_migration/__init__.py

@@ -0,0 +1,9 @@
+# -----------------------------------------------------------
+# Licensed Materials - Property of IBM
+# 5737-M66
+# (C) Copyright IBM Corp. 2026 All Rights Reserved.
+# US Government Users Restricted Rights - Use, duplication, or disclosure
+# restricted by GSA ADP Schedule Contract with IBM Corp.
+# -----------------------------------------------------------
+
+from ..cli import BaseApp  # noqa: F401

python/src/mas/cli/db2_migration/app.py

@@ -0,0 +1,220 @@
+# -----------------------------------------------------------
+# Licensed Materials - Property of IBM
+# 5737-M66
+# (C) Copyright IBM Corp. 2026 All Rights Reserved.
+# US Government Users Restricted Rights - Use, duplication, or disclosure
+# restricted by GSA ADP Schedule Contract with IBM Corp.
+# -----------------------------------------------------------
+
+import logging
+
+from typing import List, Dict, Any
+from halo import Halo
+from prompt_toolkit import print_formatted_text, HTML
+from openshift.dynamic.exceptions import NotFoundError
+
+from ..cli import BaseApp
+from .argParser import db2MigrationArgParser
+from mas.devops.ocp import createNamespace
+from mas.devops.tekton import preparePipelinesNamespace, installOpenShiftPipelines, updateTektonDefinitions, launchDb2MigrationPipeline
+
+logger = logging.getLogger(__name__)
+
+
+class Db2MigrationApp(BaseApp):
+    """Application class for DB2 cluster migration"""
+
+    def detectDb2uClusters(self, namespace: str) -> List[Dict[str, Any]]:
+        """Detect all Db2uCluster instances in the specified namespace.
+
+        Args:
+            namespace (str): Kubernetes namespace to search
+
+        Returns:
+            List[Dict[str, Any]]: List of Db2uCluster resources found
+        """
+        try:
+            db2ClusterAPI = self.dynamicClient.resources.get(api_version="db2u.databases.ibm.com/v1", kind="Db2uCluster")
+            clusters = db2ClusterAPI.get(namespace=namespace)
+            return clusters.items if clusters else []
+        except NotFoundError:
+            return []
+
+    def promptForCluster(self, clusters: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """Prompt user to select a cluster from the detected list.
+
+        Args:
+            clusters (List[Dict[str, Any]]): List of available clusters
+
+        Returns:
+            Dict[str, Any]: Selected cluster resource
+        """
+        if len(clusters) == 1:
+            cluster = clusters[0]
+            clusterName = cluster.metadata.name
+            self.printHighlight(f"Found 1 Db2uCluster: {clusterName}")
+            return cluster
+
+        # Multiple clusters - Prompt for selection
+        # self.printH2("Available Db2uClusters")
+        # options = []
+        # for i, cluster in enumerate(clusters):
+        #    name = cluster.metadata.name
+        #    version = cluster.spec.version if hasattr(cluster.spec, "version") else "unknown"
+        #    status = cluster.status.state if hasattr(cluster, "status") and hasattr(cluster.status, "state") else "unknown"
+        #    options.append(f"{name} (version: {version}, status: {status})")
+
+        # selectedIndex = self.promptForListSelect("Select cluster to migrate", options)
+        # return clusters[selectedIndex]
+
+        self.printH2("Available Db2uClusters")
+        for i, cluster in enumerate(clusters):
+            name = cluster.metadata.name
+            version = cluster.spec.version if hasattr(cluster.spec, "version") else "unknown"
+            status = cluster.status.state if hasattr(cluster, "status") and hasattr(cluster.status, "state") else "unknown"
+            print(f"  {i+1}. {name} (version: {version}, status: {status})")
+
+        selectedIndex = self.promptForInt("Select cluster to migrate", min=1, max=len(clusters))
+        return clusters[selectedIndex - 1]
+
+    def promptForBackup(self) -> bool:
+        """Prompt user whether to perform backup before migration.
+
+        Returns:
+            bool: True if backup should be performed, False otherwise
+        """
+        self.printH2("Backup Configuration")
+        print_formatted_text(
+            HTML(
+                "<Yellow>It is strongly recommended to backup before migration.</Yellow>\n"
+                "This will create a full database backup that can be used for rollback.\n"
+            )
+        )
+        return self.yesOrNo("Perform backup before migration")
+
+    def migrate(self, argv: List[str]) -> None:
+        """Main entry point for DB2 migration command.
+
+        Args:
+            argv (List[str]): Command line arguments
+        """
+        args = db2MigrationArgParser.parse_args(argv)
+        self.noConfirm = args.no_confirm
+
+        # Connect to cluster
+        self.connect()
+
+        # Determine mode: interactive vs non-interactive
+        isInteractive = args.namespace is None
+
+        if isInteractive:
+            # Interactive mode
+            self.printH1("DB2 Cluster Migration")
+
+            # List db2u namespaces
+            with Halo(text="Detecting db2u namespaces", spinner=self.spinner) as h:
+                try:
+                    namespaceAPI = self.dynamicClient.resources.get(api_version="v1", kind="Namespace")
+                    allNamespaces = namespaceAPI.get()
+                    db2uNamespaces = [ns.metadata.name for ns in allNamespaces.items if ns.metadata.name.startswith("db2u")]
+                    # v1 = client.CoreV1Api()
+                    # allNamespaces = v1.list_namespace()
+                    # db2uNamespaces = [ns.metadata.name for ns in allNamespaces.items if ns.metadata.name.startswith("db2u")]
+
+                    if db2uNamespaces:
+                        h.succeed(f"Found {len(db2uNamespaces)} db2u namespace(s)")
+                        print_formatted_text(HTML("<ansicyan>Available db2u namespaces:</ansicyan>"))
+                        for ns in sorted(db2uNamespaces):
+                            print(f"  - {ns}")
+                        print()
+                    else:
+                        h.info("No db2u namespaces found")
+                except Exception as e:
+                    h.fail(f"Failed to list namespaces: {e}")
+
+            # Prompt for namespace with default
+            namespace = self.promptForString("Enter namespace containing Db2uClusters", default="db2u")
+
+            # Detect clusters
+            with Halo(text=f"Detecting Db2uClusters in namespace {namespace}", spinner=self.spinner) as h:
+                clusters = self.detectDb2uClusters(namespace)
+                if not clusters:
+                    h.fail(f"No Db2uClusters found in namespace {namespace}")
+                    self.fatalError(f"No Db2uClusters found in namespace {namespace}")
+                h.succeed(f"Found {len(clusters)} Db2uCluster(s)")
+
+            # Select cluster
+            selectedCluster = self.promptForCluster(clusters)
+            clusterName = selectedCluster.metadata.name
+
+            # Prompt for backup
+            enableBackup = self.promptForBackup()
+
+        else:
+            # Non-interactive mode
+            namespace = args.namespace
+            clusterName = args.cluster_name
+            enableBackup = args.backup == "true" if args.backup else True
+
+            # Validate cluster exists if name provided
+            if clusterName:
+                clusters = self.detectDb2uClusters(namespace)
+                clusterNames = [c.metadata.name for c in clusters]
+                if clusterName not in clusterNames:
+                    self.fatalError(f"Cluster {clusterName} not found in namespace {namespace}")
+            else:
+                # Auto-select if only one cluster
+                clusters = self.detectDb2uClusters(namespace)
+                if len(clusters) == 0:
+                    self.fatalError(f"No Db2uClusters found in namespace {namespace}")
+                elif len(clusters) == 1:
+                    clusterName = clusters[0].metadata.name
+                else:
+                    self.fatalError("Multiple clusters found. Please specify --cluster-name")
+
+        # Confirmation
+        if not self.noConfirm:
+            self.printH2("Migration Summary")
+            print_formatted_text(
+                HTML(
+                    f"<Yellow>Namespace:</Yellow> {namespace}\n"
+                    f"<Yellow>Cluster:</Yellow> {clusterName}\n"
+                    f"<Yellow>Backup:</Yellow> {'Enabled' if enableBackup else 'Disabled'}\n"
+                )
+            )
+            if not self.yesOrNo("Proceed with migration"):
+                print_formatted_text(HTML("<Red>Migration cancelled</Red>"))
+                return
+
+        # Set parameters
+        self.setParam("db2_migration_namespace", namespace)
+        self.setParam("db2_migration_cluster_name", clusterName)
+        self.setParam("db2_migration_backup_enabled", str(enableBackup).lower())
+
+        # Prepare pipeline namespace
+        pipelinesNamespace = "mas-pipelines"
+
+        with Halo(text="Validating OpenShift Pipelines installation", spinner=self.spinner) as h:
+            if installOpenShiftPipelines(self.dynamicClient):
+                h.succeed("OpenShift Pipelines Operator is installed and ready")
+            else:
+                h.fail("OpenShift Pipelines Operator installation failed")
+                self.fatalError("Installation failed")
+
+        with Halo(text=f"Preparing namespace ({pipelinesNamespace})", spinner=self.spinner) as h:
+            createNamespace(self.dynamicClient, pipelinesNamespace)
+            preparePipelinesNamespace(dynClient=self.dynamicClient)
+            h.succeed(f"Namespace {pipelinesNamespace} is ready")
+
+        with Halo(text=f"Installing latest Tekton definitions (v{self.version})", spinner=self.spinner) as h:
+            updateTektonDefinitions(self.dynamicClient, pipelinesNamespace, self.tektonDefsPath)
+            h.succeed(f"Latest Tekton definitions are installed (v{self.version})")
+
+        # Launch pipeline
+        with Halo(text="Submitting PipelineRun for DB2 migration", spinner=self.spinner) as h:
+            pipelineURL = launchDb2MigrationPipeline(dynClient=self.dynamicClient, params=self.params)
+            if pipelineURL:
+                h.succeed("PipelineRun for DB2 migration submitted")
+                print_formatted_text(HTML(f"\nView progress:\n  <Cyan><u>{pipelineURL}</u></Cyan>\n"))
+            else:
+                h.fail("Failed to submit PipelineRun, see log file for details")

python/src/mas/cli/db2_migration/argParser.py

@@ -0,0 +1,66 @@
+# -----------------------------------------------------------
+# Licensed Materials - Property of IBM
+# 5737-M66
+# (C) Copyright IBM Corp. 2026 All Rights Reserved.
+# US Government Users Restricted Rights - Use, duplication, or disclosure
+# restricted by GSA ADP Schedule Contract with IBM Corp.
+# -----------------------------------------------------------
+
+import argparse
+import sys
+
+from ..cli import getHelpFormatter
+
+
+class Db2MigrationArgumentParser(argparse.ArgumentParser):
+    """Custom argument parser for DB2 migration command"""
+
+    def format_usage(self):
+        prog = self.prog
+        return (
+            f"Usage (non-interactive mode):\n"
+            f"  {prog} --namespace NAMESPACE [--cluster-name CLUSTER_NAME]\n"
+            f"                  [--backup {{true,false}}] [--no-confirm]\n"
+            f"\n"
+            f"Usage (interactive mode):\n"
+            f"  {prog}\n"
+            f"\n"
+            f"Usage (help):\n"
+            f"  {prog} -h\n"
+        )
+
+    def parse_args(self, args=None, namespace=None):
+        parsedArgs = super().parse_args(args, namespace)
+
+        providedArgs = []
+        if args is not None:
+            providedArgs = [arg for arg in args if arg.startswith("-")]
+        else:
+            providedArgs = [arg for arg in sys.argv[1:] if arg.startswith("-")]
+
+        hasAnyArgs = len(providedArgs) > 0
+        hasNamespace = parsedArgs.namespace is not None
+        helpOnly = "--help" in providedArgs or "-h" in providedArgs
+
+        if hasAnyArgs and not hasNamespace and not helpOnly:
+            self.error("non-interactive mode requires --namespace parameter")
+
+        return parsedArgs
+
+
+db2MigrationArgParser = Db2MigrationArgumentParser(
+    prog="mas db2ucluster-migration",
+    description="Migrate Db2uCluster to Db2uInstance by launching the DB2 Migration Tekton Pipeline.",
+    epilog="Refer to the online documentation for more information: https://ibm-mas.github.io/cli/",
+    formatter_class=getHelpFormatter(),
+    add_help=False,
+)
+
+migrationArgGroup = db2MigrationArgParser.add_argument_group("Migration Configuration", "Configure the DB2 migration parameters.")
+migrationArgGroup.add_argument("--namespace", required=False, help="Namespace containing Db2uCluster instances")
+migrationArgGroup.add_argument("--cluster-name", required=False, help="Specific Db2uCluster name to migrate")
+migrationArgGroup.add_argument("--backup", required=False, choices=["true", "false"], help="Enable or disable backup before migration")
+
+otherArgGroup = db2MigrationArgParser.add_argument_group("More", "Additional options.")
+otherArgGroup.add_argument("--no-confirm", required=False, action="store_true", default=False, help="Launch without prompting for confirmation")
+otherArgGroup.add_argument("-h", "--help", action="help", default=False, help="Show this help message and exit")

tekton/generate-tekton-pipelines.yml

@@ -22,6 +22,7 @@
       - mas-update
       - mas-upgrade
       - mas-uninstall
+      - mas-devops-db2-migration 
       # MAS FVT Pipelines
       - mas-fvt-assist
       - mas-fvt-core