Adding ipv6 support

cloudshell/cp/vcenter/actions/vm_details.py

@@ -11,6 +11,7 @@
 )
 
 from cloudshell.cp.vcenter.actions.vm_network import VMNetworkActions
+from cloudshell.cp.vcenter.constants import IPProtocol
 from cloudshell.cp.vcenter.exceptions import VMIPNotFoundException
 from cloudshell.cp.vcenter.handlers.si_handler import SiHandler
 from cloudshell.cp.vcenter.handlers.vm_handler import PowerState, VmHandler
@@ -86,11 +87,16 @@ def _prepare_vm_network_data(
             vlan_id = vm.get_network_vlan_id(network)
 
             if vlan_id and (self.is_quali_network(network.name) or is_predefined):
-                vnic_ip = vnic.ipv4
+                if app_model.ip_protocol_version == IPProtocol.IPv6:
+                    vnic_ip = vnic.ipv6
+                else:
+                    vnic_ip = vnic.ipv4
+
                 is_primary = primary_ip == vnic_ip if primary_ip else False
 
                 network_data = [
-                    VmDetailsProperty(key="IP", value=vnic_ip),
+                    VmDetailsProperty(key="IP", value=vnic.ipv4),
+                    VmDetailsProperty(key="IPv6", value=vnic.ipv6),
                     VmDetailsProperty(key="MAC Address", value=vnic.mac_address),
                     VmDetailsProperty(key="Network Adapter", value=vnic.name),
                     VmDetailsProperty(key="Port Group Name", value=network.name),
@@ -113,10 +119,17 @@ def _get_primary_ip(self, app_model, vm) -> str | None:
         with suppress(VMIPNotFoundException):
             if isinstance(app_model, StaticVCenterDeployedApp):
                 # try to get VM IP without waiting
-                primary_ip = self.get_vm_ip(vm, timeout=0)
+                primary_ip = self.get_vm_ip(
+                    vm=vm, timeout=0, ip_protocol_version=app_model.ip_protocol_version
+                )
             elif vm.power_state is PowerState.ON:
                 # try to get VM IP without waiting, use IP regex if present
-                primary_ip = self.get_vm_ip(vm, ip_regex=app_model.ip_regex, timeout=0)
+                primary_ip = self.get_vm_ip(
+                    vm=vm,
+                    ip_regex=app_model.ip_regex,
+                    timeout=0,
+                    ip_protocol_version=app_model.ip_protocol_version,
+                )
         return primary_ip
 
     @staticmethod

cloudshell/cp/vcenter/actions/vm_network.py

@@ -7,6 +7,7 @@
 from datetime import datetime, timedelta
 from typing import TYPE_CHECKING
 
+from cloudshell.cp.vcenter.constants import IPProtocol
 from cloudshell.cp.vcenter.exceptions import VMIPNotFoundException
 from cloudshell.cp.vcenter.handlers.network_handler import NetworkHandler
 from cloudshell.cp.vcenter.handlers.vm_handler import VmHandler
@@ -41,20 +42,37 @@ def _find_vm_ip(
         vm: VmHandler,
         skip_networks: list[NetworkHandler],
         is_ip_pass_regex: callable[[str | None], bool],
+        ip_protocol_version: str = IPProtocol.IPv4,
     ) -> str | None:
-        logger.debug(f"Searching for the IPv4 address of the {vm}")
+        logger.debug(f"Searching for the IP address of the {vm}")
         ip = None
-        primary_ip = vm.primary_ipv4
-        if is_ip_pass_regex(primary_ip):
-            ip = primary_ip
-            logger.debug(f"Use primary IPv4 address of the {vm}")
+
+        if ip_protocol_version == IPProtocol.IPv4:
+            if is_ip_pass_regex(vm.primary_ipv4):
+                ip = vm.primary_ipv4
+                logger.debug(f"Use primary IPv4 address of the {vm}")
+            else:
+                for vnic in vm.vnics:
+                    logger.debug(f"Checking {vnic} with ip {vnic.ipv4}")
+                    if vnic.network not in skip_networks and is_ip_pass_regex(
+                        vnic.ipv4
+                    ):
+                        logger.debug(f"Found IP {vnic.ipv4} on {vnic}")
+                        ip = vnic.ipv4
+                        break
         else:
-            for vnic in vm.vnics:
-                logger.debug(f"Checking {vnic} with ip {vnic.ipv4}")
-                if vnic.network not in skip_networks and is_ip_pass_regex(vnic.ipv4):
-                    logger.debug(f"Found IP {vnic.ipv4} on {vnic}")
-                    ip = vnic.ipv4
-                    break
+            if is_ip_pass_regex(vm.primary_ipv6):
+                ip = vm.primary_ipv6
+                logger.debug(f"Use primary IPv6 address of the {vm}")
+            else:
+                for vnic in vm.vnics:
+                    logger.debug(f"Checking {vnic} with ip {vnic.ipv6}")
+                    if vnic.network not in skip_networks and is_ip_pass_regex(
+                        vnic.ipv6
+                    ):
+                        logger.debug(f"Found IP {vnic.ipv6} on {vnic}")
+                        ip = vnic.ipv6
+                        break
         return ip
 
     def get_vm_ip(
@@ -63,6 +81,7 @@ def get_vm_ip(
         ip_regex: str | None = None,
         timeout: int = 0,
         skip_networks: list[NetworkHandler] | None = None,
+        ip_protocol_version: str = IPProtocol.IPv4,
     ) -> str:
         logger.info(f"Getting IP address for the VM {vm.name} from the vCenter")
         timeout_time = datetime.now() + timedelta(seconds=timeout)
@@ -71,7 +90,12 @@ def get_vm_ip(
 
         while True:
             with self._cancellation_manager:
-                ip = self._find_vm_ip(vm, skip_networks, is_ip_pass_regex)
+                ip = self._find_vm_ip(
+                    vm=vm,
+                    skip_networks=skip_networks,
+                    is_ip_pass_regex=is_ip_pass_regex,
+                    ip_protocol_version=ip_protocol_version,
+                )
             if ip:
                 break
             if datetime.now() > timeout_time:

cloudshell/cp/vcenter/constants.py

@@ -1,3 +1,5 @@
+from enum import Enum
+
 SHELL_NAME = "VMware vCenter Cloud Provider 2G"
 STATIC_SHELL_NAME = "Generic Static vCenter VM 2G"
 
@@ -7,3 +9,8 @@
 VM_FROM_IMAGE_DEPLOYMENT_PATH = f"{SHELL_NAME}.vCenter VM From Image 2G"
 
 DEPLOYED_APPS_FOLDER = "Deployed Apps"
+
+
+class IPProtocol(Enum):
+    IPv4 = "ipv4"
+    IPv6 = "ipv6"

cloudshell/cp/vcenter/flows/refresh_ip.py

@@ -26,14 +26,17 @@ def refresh_ip(
 
     actions = VMNetworkActions(resource_conf, cancellation_manager)
     if isinstance(deployed_app, StaticVCenterDeployedApp):
-        ip = actions.get_vm_ip(vm)
+        ip = actions.get_vm_ip(
+            vm=vm, ip_protocol_version=deployed_app.ip_protocol_version
+        )
     else:
         default_net = dc.get_network(resource_conf.holding_network)
         ip = actions.get_vm_ip(
             vm,
             ip_regex=deployed_app.ip_regex,
             timeout=deployed_app.refresh_ip_timeout,
             skip_networks=[default_net],
+            ip_protocol_version=deployed_app.ip_protocol_version,
         )
     if ip != deployed_app.private_ip:
         deployed_app.update_private_ip(deployed_app.name, ip)

cloudshell/cp/vcenter/handlers/custom_spec_handler.py

@@ -90,6 +90,12 @@ def _set_network_params(self, networks: NetworksList, num_vm_nics: int):
                     network_adapter.ip = vim.vm.customization.FixedIp(
                         ipAddress=network.ipv4_address
                     )
+
+                if network.ipv6_address is not Empty:
+                    network_adapter.ip = vim.vm.customization.FixedIp(
+                        ipAddress=network.ipv6_address
+                    )
+
                 if network.subnet_mask is not Empty:
                     network_adapter.subnetMask = network.subnet_mask
 

cloudshell/cp/vcenter/handlers/vm_handler.py

@@ -49,7 +49,7 @@
     VnicWithMacNotFound,
 )
 from cloudshell.cp.vcenter.utils.connectivity_helpers import is_correct_vnic
-from cloudshell.cp.vcenter.utils.network_helpers import is_ipv4
+from cloudshell.cp.vcenter.utils.network_helpers import is_ipv4, is_ipv6
 from cloudshell.cp.vcenter.utils.units_converter import BASE_10
 
 logger = logging.getLogger(__name__)
@@ -151,6 +151,11 @@ def primary_ipv4(self) -> str | None:
         ip = self._vc_obj.guest.ipAddress
         return ip if is_ipv4(ip) else None
 
+    @property
+    def primary_ipv6(self) -> str | None:
+        ip = self._vc_obj.guest.ipAddress
+        return ip if is_ipv6(ip) else None
+
     @property
     def networks(self) -> list[NetworkHandler | DVPortGroupHandler]:
         return [get_network_handler(net, self.si) for net in self._vc_obj.network]

cloudshell/cp/vcenter/handlers/vnic_handler.py

@@ -15,7 +15,7 @@
     NetworkHandler,
 )
 from cloudshell.cp.vcenter.handlers.virtual_device_handler import VirtualDevice
-from cloudshell.cp.vcenter.utils.network_helpers import is_ipv4
+from cloudshell.cp.vcenter.utils.network_helpers import is_ipv4, is_ipv6
 
 if TYPE_CHECKING:
     from cloudshell.cp.vcenter.handlers.vm_handler import VmHandler
@@ -97,6 +97,12 @@ def ipv4(self) -> str | None:
         ipv4 = next(filter(is_ipv4, ips), None)
         return ipv4
 
+    @property
+    def ipv6(self) -> str | None:
+        ips = self.vm.get_ip_addresses_by_vnic(self)
+        ipv6 = next(filter(is_ipv6, ips), None)
+        return ipv6
+
     def connect(self, network: NetworkHandler | DVPortGroupHandler) -> None:
         if isinstance(network, NetworkHandler):
             nic_spec = self._create_spec_for_connecting_network(network)

cloudshell/cp/vcenter/models/deployed_app.py

@@ -45,6 +45,18 @@ class BaseVCenterDeployedApp(DeployedApp):
     autogenerated_name = ResourceBoolAttrRODeploymentPath(ATTR_NAMES.autogenerated_name)
     copy_source_uuid = ResourceBoolAttrRODeploymentPath(ATTR_NAMES.copy_source_uuid)
 
+    @property
+    def ip_protocol_version(self):
+        ip_protocol = self.attributes.get(f"{self._namespace}IP Protocol Version")
+        if ip_protocol and "4" in ip_protocol:
+            ip_protocol = constants.IPProtocol.IPv4
+        elif ip_protocol and "6" in ip_protocol:
+            ip_protocol = constants.IPProtocol.IPv6
+        else:
+            ip_protocol = constants.IPProtocol.IPv4
+
+        return ip_protocol
+
 
 class VMFromTemplateDeployedApp(BaseVCenterDeployedApp):
     ATTR_NAMES = VCenterVMFromTemplateDeploymentAppAttributeNames
@@ -86,6 +98,16 @@ class StaticVCenterDeployedApp(DeployedApp):
         ATTR_NAMES.vcenter_resource_name
     )
 
+    @property
+    def ip_protocol_version(self):
+        ip_protocol = self.attributes.get(f"{self._namespace}IP Protocol Version")
+        if ip_protocol and "6" in ip_protocol:
+            ip_protocol = constants.IPProtocol.IPv6
+        else:
+            ip_protocol = constants.IPProtocol.IPv4
+
+        return ip_protocol
+
 
 class VCenterGetVMDetailsRequestActions(GetVMDetailsRequestActions):
     deployed_app: BaseVCenterDeployedApp

cloudshell/cp/vcenter/utils/network_helpers.py

@@ -11,3 +11,13 @@ def is_ipv4(ip: str | None) -> bool:
     else:
         result = True
     return result
+
+
+def is_ipv6(ip: str | None) -> bool:
+    try:
+        ipaddress.IPv6Address(ip)
+    except ipaddress.AddressValueError:
+        result = False
+    else:
+        result = True
+    return result

version.txt

@@ -1 +1 @@
-7.0.1
+7.0.2