feat(oracle): allow for specifying network configuration

This commit adds the ability to specify network configuration for
launching instances and for attaching vnics to instances. This
allows for more control over the network configuration of instances
and vnics.

Author: a-dubs

pycloudlib/oci/cloud.py

@@ -20,11 +20,13 @@
 )
 from pycloudlib.oci.instance import OciInstance
 from pycloudlib.oci.utils import (
+    generate_create_vnic_details,
     get_subnet_id,
     get_subnet_id_by_name,
     parse_oci_config_from_env_vars,
     wait_till_ready,
 )
+from pycloudlib.types import NetworkingConfig
 from pycloudlib.util import UBUNTU_RELEASE_VERSION_MAP, subp
 
 
@@ -251,6 +253,7 @@ def get_instance(self, instance_id, *, username: Optional[str] = None, **kwargs)
             availability_domain=self.availability_domain,
             oci_config=self.oci_config,
             username=username,
+            vcn_name=self.vcn_name,
         )
 
     def launch(
@@ -265,6 +268,7 @@ def launch(
         subnet_id: Optional[str] = None,
         subnet_name: Optional[str] = None,
         metadata: Dict = {},
+        primary_network_config: Optional[NetworkingConfig] = None,
         **kwargs,
     ) -> OciInstance:
         """Launch an instance.
@@ -286,6 +290,9 @@ def launch(
             username: username to use when connecting via SSH
             vcn_name: Name of the VCN to use. If not provided, the first VCN
                 found will be used
+            subnet_name: string, name of subnet to use for instance.
+            primary_network_config: NetworkingConfig object to use for configuring the primary
+                network interface
             **kwargs: dictionary of other arguments to pass as
                 LaunchInstanceDetails
 
@@ -296,18 +303,16 @@ def launch(
         if not image_id:
             raise ValueError(f"{self._type} launch requires image_id param. Found: {image_id}")
 
-        # provided subnet_id takes the highest precendence
         if not subnet_id:
             if subnet_name:
-                subnet_id = get_subnet_id_by_name(
-                    self.network_client, self.compartment_id, subnet_name
-                )
+                subnet_id = get_subnet_id_by_name(self.network_client, self.compartment_id, subnet_name)
             else:
                 subnet_id = get_subnet_id(
                     self.network_client,
                     self.compartment_id,
                     self.availability_domain,
                     vcn_name=self.vcn_name,
+                    networking_config=primary_network_config,
                 )
         default_metadata = {
             "ssh_authorized_keys": self.key_pair.public_key_content,
@@ -327,6 +332,10 @@ def launch(
             image_id=image_id,
             metadata={**default_metadata, **metadata},
             compute_cluster_id=cluster_id,
+            create_vnic_details=generate_create_vnic_details(
+                subnet_id=subnet_id,
+                networking_config=primary_network_config,
+            ),
             **kwargs,
         )
 
@@ -341,6 +350,32 @@ def launch(
         self.created_instances.append(instance)
         return instance
 
+    def find_compatible_subnet(self, networking_config: NetworkingConfig) -> str:
+        """
+        Automatically select a subnet that is compatible with the given networking_config.
+
+        In this case, compatible means that the subnet can support the necessary networking type
+        (ipv4 only, ipv6 only, or dual stack) and the private or public requirement.
+        This method will select the first subnet that matches the criteria.
+
+        Args:
+            networking_config: NetworkingConfig object to use for finding a subnet
+
+        Returns:
+            id of the subnet selected
+
+        Raises:
+            `PycloudlibError` if unable to determine `subnet_id` for the given `networking_config`
+        """
+        subnet_id = get_subnet_id(
+            network_client=self.network_client,
+            compartment_id=self.compartment_id,
+            availability_domain=self.availability_domain,
+            vcn_name=self.vcn_name,
+            networking_config=networking_config,
+        )
+        return subnet_id
+
     def snapshot(self, instance, clean=True, name=None):
         """Snapshot an instance and generate an image from it.
 

pycloudlib/oci/instance.py

@@ -10,7 +10,13 @@
 
 from pycloudlib.errors import PycloudlibError
 from pycloudlib.instance import BaseInstance
-from pycloudlib.oci.utils import get_subnet_id, get_subnet_id_by_name, wait_till_ready
+from pycloudlib.oci.utils import (
+    generate_create_vnic_details,
+    get_subnet_id,
+    get_subnet_id_by_name,
+    wait_till_ready,
+)
+from pycloudlib.types import NetworkingConfig
 
 
 class OciInstance(BaseInstance):
@@ -27,6 +33,7 @@ def __init__(
         oci_config=None,
         *,
         username: Optional[str] = None,
+        vcn_name: Optional[str] = None,
     ):
         """Set up the instance.
 
@@ -46,6 +53,7 @@ def __init__(
         self.availability_domain = availability_domain
         self._fault_domain = None
         self._ip = None
+        self._vcn_name: Optional[str] = vcn_name
 
         if oci_config is None:
             oci_config = oci.config.from_file("~/.oci/config")  # noqa: E501
@@ -145,7 +153,8 @@ def secondary_vnic_private_ip(self) -> Optional[str]:
             for vnic_attachment in vnic_attachments
         ]
         secondary_vnic_attachment = [vnic for vnic in vnics if not vnic.is_primary][0]
-        return secondary_vnic_attachment.private_ip
+        self._log.debug("secondary vnic attachment data:\n%s", secondary_vnic_attachment)
+        return secondary_vnic_attachment.private_ip or secondary_vnic_attachment.ipv6_addresses[0]
 
     @property
     def instance_data(self):
@@ -258,7 +267,7 @@ def get_secondary_vnic_ip(self) -> str:
     def add_network_interface(
         self,
         nic_index: int = 0,
-        use_private_subnet: bool = False,
+        networking_config: Optional[NetworkingConfig] = None,
         subnet_name: Optional[str] = None,
         **kwargs: Any,
     ) -> str:
@@ -270,13 +279,19 @@ def add_network_interface(
 
         Args:
             nic_index: The index of the NIC to add
-            subnet_name: Name of the subnet to add the NIC to. If not provided,
-                will use `use_private_subnet` to select first available subnet.
-            use_private_subnet: If True, will select the first available private
-                subnet. If False, will select the first available public subnet.
-                This is only used if `subnet_name` is not provided.
+            networking_config: Networking configuration to use when selecting subnet. This specifies
+                the networking type (ipv4, ipv6, or dualstack) and whether to use a public or
+                private subnet. If not provided, will default to selecting the first public subnet
+                found.
+            subnet_name: Name of the subnet to add the NIC to. If provided, this subnet will
+                blindly be selected and networking_config will be ignored.
+
+        Returns:
+            str: The private IP address of the added network interface.
         """
         if subnet_name:
+            if networking_config:
+                self._log.debug("Ignoring networking_config when subnet_name is provided.")
             subnet_id = get_subnet_id_by_name(
                 self.network_client,
                 self.compartment_id,
@@ -287,10 +302,11 @@ def add_network_interface(
                 self.network_client,
                 self.compartment_id,
                 self.availability_domain,
-                private=use_private_subnet,
+                networking_config=networking_config,
+                vcn_name=self._vcn_name,
             )
-        create_vnic_details = oci.core.models.CreateVnicDetails(  # noqa: E501
-            subnet_id=subnet_id,
+        create_vnic_details = generate_create_vnic_details(
+            subnet_id=subnet_id, networking_config=networking_config
         )
         attach_vnic_details = oci.core.models.AttachVnicDetails(  # noqa: E501
             create_vnic_details=create_vnic_details,
@@ -304,13 +320,29 @@ def add_network_interface(
             desired_state=vnic_attachment_data.LIFECYCLE_STATE_ATTACHED,
         )
         vnic_data = self.network_client.get_vnic(vnic_attachment_data.vnic_id).data
+        self._log.debug(
+            "Newly attached vnic data:\n%s",
+            vnic_data,
+        )
+        try:
+            new_ip = vnic_data.private_ip or vnic_data.ipv6_addresses[0]
+        except IndexError:
+            err_msg = (
+                "Unexpected error occurred when trying to retrieve local IP address of the "
+                "newly attached NIC. No private IP or IPv6 address found."
+            )
+            self._log.error(
+                err_msg + "Full vnic data for debugging purposes:\n%s",
+                vnic_data,
+            )
+            raise PycloudlibError(err_msg)
         self._log.info(
-            "Added network interface with private IP %s to instance %s on nic #%s",
-            vnic_data.private_ip,
+            "Added network interface with IP %s to instance %s on nic #%s",
+            new_ip,
             self.instance_id,
             nic_index,
         )
-        return vnic_data.private_ip
+        return new_ip
 
     def remove_network_interface(self, ip_address: str):
         """Remove network interface based on IP address.
@@ -355,14 +387,20 @@ def configure_secondary_vnic(self) -> str:
                         or if the IP address was not successfully assigned to the interface.
             PycloudlibError: If failed to fetch secondary VNIC data from the Oracle Cloud metadata service.
         """
-        if not self.secondary_vnic_private_ip:
+        secondary_ip = self.secondary_vnic_private_ip
+        if not secondary_ip:
             raise ValueError("Cannot configure secondary VNIC without a secondary VNIC attached")
+        if ":" in secondary_ip:
+            imds_url = "http://[fd00:c1::a9fe:a9fe]/opc/v1/vnics"
+        else:
+            imds_url = "http://169.254.169.254/opc/v1/vnics"
+
         secondary_vnic_imds_data: Optional[Dict[str, str]] = None
         # it can take a bit for the secondary VNIC to show up in the IMDS
         # so we need to retry fetching the data for roughly a minute
         for _ in range(60):
             # Fetch JSON data from the Oracle Cloud metadata service
-            imds_req = self.execute("curl -s http://169.254.169.254/opc/v1/vnics").stdout
+            imds_req = self.execute(f"curl -s {imds_url}").stdout
             vnics_data = json.loads(imds_req)
             if len(vnics_data) > 1:
                 self._log.debug("Successfully fetched secondary VNIC data from IMDS")