Merge branch 'develop' into ODSC-40043/dataflow_archive

Author: z7ye

ads/model/deployment/model_deployment.py

@@ -1538,6 +1538,9 @@ def _build_model_deployment_configuration_details(self) -> Dict:
                 or MODEL_DEPLOYMENT_INSTANCE_MEMORY_IN_GBS,
             }
 
+        if infrastructure.subnet_id:
+            instance_configuration[infrastructure.CONST_SUBNET_ID] = infrastructure.subnet_id
+
         scaling_policy = {
             infrastructure.CONST_POLICY_TYPE: "FIXED_SIZE",
             infrastructure.CONST_INSTANCE_COUNT: infrastructure.replica

ads/model/deployment/model_deployment_infrastructure.py

@@ -39,6 +39,8 @@ class ModelDeploymentInfrastructure(Builder):
         The access and predict log id of model deployment
     web_concurrency: int
         The web concurrency of model deployment
+    subnet_id: str
+        The subnet id of model deployment
 
     Methods
     -------
@@ -64,6 +66,8 @@ class ModelDeploymentInfrastructure(Builder):
         Sets the access and predict log id of model deployment
     with_web_concurrency(web_concurrency)
         Sets the web concurrency of model deployment
+    with_subnet_id(subnet_id) 
+        Sets the subnet id of model deployment
 
     Example
     -------
@@ -79,6 +83,7 @@ class ModelDeploymentInfrastructure(Builder):
     ...        .with_replica(1)
     ...        .with_bandwidth_mbps(10)
     ...        .with_web_concurrency(10)
+    ...        .with_subnet_id(<subnet_id>)
     ...        .with_access_log(
     ...            log_group_id=<log_group_id>,
     ...            log_id=<log_id>
@@ -121,6 +126,7 @@ class ModelDeploymentInfrastructure(Builder):
     CONST_LOG_GROUP_ID = "logGroupId"
     CONST_WEB_CONCURRENCY = "webConcurrency"
     CONST_STREAM_CONFIG_DETAILS = "streamConfigurationDetails"
+    CONST_SUBNET_ID = "subnetId"
 
     attribute_map = {
         CONST_PROJECT_ID: "project_id",
@@ -136,6 +142,7 @@ class ModelDeploymentInfrastructure(Builder):
         CONST_LOG_ID: "log_id",
         CONST_LOG_GROUP_ID: "log_group_id",
         CONST_WEB_CONCURRENCY: "web_concurrency",
+        CONST_SUBNET_ID: "subnet_id"
     }
 
     shape_config_details_attribute_map = {
@@ -162,6 +169,7 @@ class ModelDeploymentInfrastructure(Builder):
         CONST_COMPARTMENT_ID: "compartment_id",
         CONST_SHAPE_NAME: f"{MODEL_CONFIG_DETAILS_PATH}.instance_configuration.instance_shape_name",
         CONST_SHAPE_CONFIG_DETAILS: f"{MODEL_CONFIG_DETAILS_PATH}.instance_configuration.model_deployment_instance_shape_config_details",
+        CONST_SUBNET_ID: f"{MODEL_CONFIG_DETAILS_PATH}.instance_configuration.subnet_id",
         CONST_REPLICA: f"{MODEL_CONFIG_DETAILS_PATH}.scaling_policy.instance_count",
         CONST_BANDWIDTH_MBPS: f"{MODEL_CONFIG_DETAILS_PATH}.bandwidth_mbps",
         CONST_ACCESS_LOG: "category_log_details.access",
@@ -518,3 +526,29 @@ def with_web_concurrency(
             The ModelDeploymentInfrastructure instance (self).
         """
         return self.set_spec(self.CONST_WEB_CONCURRENCY, web_concurrency)
+
+    def with_subnet_id(self, subnet_id: str) -> "ModelDeploymentInfrastructure":
+        """Sets the subnet id of model deployment.
+
+        Parameters
+        ----------
+        subnet_id : str
+            The subnet id of model deployment.
+
+        Returns
+        -------
+        ModelDeploymentInfrastructure
+            The ModelDeploymentInfrastructure instance (self).
+        """
+        return self.set_spec(self.CONST_SUBNET_ID, subnet_id)
+
+    @property
+    def subnet_id(self) -> str:
+        """The model deployment subnet id.
+        
+        Returns
+        -------
+        str
+            The model deployment subnet id.
+        """
+        return self.get_spec(self.CONST_SUBNET_ID, None)

ads/model/generic_model.py

@@ -1902,6 +1902,7 @@ def deploy(
         display_name: Optional[str] = None,
         description: Optional[str] = None,
         deployment_instance_shape: Optional[str] = None,
+        deployment_instance_subnet_id: Optional[str] = None,
         deployment_instance_count: Optional[int] = None,
         deployment_bandwidth_mbps: Optional[int] = None,
         deployment_log_group_id: Optional[str] = None,
@@ -1950,6 +1951,8 @@ def deploy(
             The description of the model.
         deployment_instance_shape: (str, optional). Default to `VM.Standard2.1`.
             The shape of the instance used for deployment.
+        deployment_instance_subnet_id: (str, optional). Default to None.
+            The subnet id of the instance used for deployment.
         deployment_instance_count: (int, optional). Defaults to 1.
             The number of instance used for deployment.
         deployment_bandwidth_mbps: (int, optional). Defaults to 10.
@@ -2084,6 +2087,10 @@ def deploy(
                 self.properties.deployment_instance_shape
                 or existing_infrastructure.shape_name
             )
+            .with_subnet_id(
+                self.properties.deployment_instance_subnet_id
+                or existing_infrastructure.subnet_id
+            )
             .with_replica(
                 self.properties.deployment_instance_count
                 or existing_infrastructure.replica
@@ -2253,6 +2260,7 @@ def prepare_save_deploy(
         deployment_display_name: Optional[str] = None,
         deployment_description: Optional[str] = None,
         deployment_instance_shape: Optional[str] = None,
+        deployment_instance_subnet_id: Optional[str] = None,
         deployment_instance_count: Optional[int] = None,
         deployment_bandwidth_mbps: Optional[int] = None,
         deployment_log_group_id: Optional[str] = None,
@@ -2340,6 +2348,8 @@ def prepare_save_deploy(
             The description of the model.
         deployment_instance_shape: (str, optional). Default to `VM.Standard2.1`.
             The shape of the instance used for deployment.
+        deployment_instance_subnet_id: (str, optional). Default to None.
+            The subnet id of the instance used for deployment.
         deployment_instance_count: (int, optional). Defaults to 1.
             The number of instance used for deployment.
         deployment_bandwidth_mbps: (int, optional). Defaults to 10.
@@ -2481,6 +2491,7 @@ def prepare_save_deploy(
             display_name=deployment_display_name,
             description=deployment_description,
             deployment_instance_shape=self.properties.deployment_instance_shape,
+            deployment_instance_subnet_id=self.properties.deployment_instance_subnet_id,
             deployment_instance_count=self.properties.deployment_instance_count,
             deployment_bandwidth_mbps=self.properties.deployment_bandwidth_mbps,
             deployment_log_group_id=self.properties.deployment_log_group_id,

ads/model/model_properties.py

@@ -28,6 +28,7 @@ class ModelProperties(BaseProperties):
     remove_existing_artifact: bool = None
     overwrite_existing_artifact: bool = None
     deployment_instance_shape: str = None
+    deployment_instance_subnet_id: str = None
     deployment_instance_count: int = None
     deployment_bandwidth_mbps: int = None
     deployment_log_group_id: str = None

ads/opctl/utils.py

@@ -10,12 +10,12 @@
 import os
 import subprocess
 import sys
+import shlex
 import tempfile
 import urllib.parse
 from distutils import dir_util
 from subprocess import Popen, PIPE, STDOUT
 from typing import Union, List, Tuple, Dict
-import shlex
 import yaml
 
 import ads
@@ -33,6 +33,9 @@
 )
 
 
+CONTAINER_NETWORK = "CONTAINER_NETWORK"
+
+
 def get_service_pack_prefix() -> Dict:
     curr_dir = os.path.dirname(os.path.abspath(__file__))
     service_config_file = os.path.join(curr_dir, "conda", "config.yaml")
@@ -175,11 +178,13 @@ def build_image(
             command += ["--build-arg", f"http_proxy={os.environ['http_proxy']}"]
         if os.environ.get("https_proxy"):
             command += ["--build-arg", f"https_proxy={os.environ['https_proxy']}"]
+        if os.environ.get(CONTAINER_NETWORK):
+            command += ["--network", os.environ[CONTAINER_NETWORK]]
         command += [os.path.abspath(curr_dir)]
-        logger.info(f"Build image with command {command}")
+        logger.info("Build image with command %s", command)
         proc = run_command(command)
     if proc.returncode != 0:
-        raise RuntimeError(f"Docker build failed.")
+        raise RuntimeError("Docker build failed.")
 
 
 def _get_image_name_dockerfile_target(type: str, gpu: bool) -> str:
@@ -321,27 +326,51 @@ def run_container(
     logger.info(f"command: {command}")
     logger.info(f"entrypoint: {entrypoint}")
 
+    # Print out the equivalent docker run command for debugging purpose
+    docker_run_cmd = [">>> docker run --rm"]
+    if entrypoint:
+        docker_run_cmd.append(f"--entrypoint {entrypoint}")
+    if env_vars:
+        docker_run_cmd.extend([f"-e {key}={val}" for key, val in env_vars.items()])
+    if bind_volumes:
+        docker_run_cmd.extend(
+            [f'-v {source}:{bind.get("bind")}' for source, bind in bind_volumes.items()]
+        )
+    docker_run_cmd.append(image)
+    if command:
+        docker_run_cmd.append(command)
+    logger.debug(" ".join(docker_run_cmd))
+
     client = get_docker_client()
     try:
         client.api.inspect_image(image)
     except docker.errors.ImageNotFound:
-        logger.warn(f"Image {image} not found. Try pulling it now....")
+        logger.warning(f"Image {image} not found. Try pulling it now....")
         run_command(["docker", "pull", f"{image}"], None)
-    container = client.containers.run(
-        image=image,
-        volumes=bind_volumes,
-        command=shlex.split(command) if command else None,
-        environment=env_vars,
-        detach=True,
-        entrypoint=entrypoint,
-        user=0,
-        # auto_remove=True,
-    )
-    logger.info(f"Container ID: {container.id}")
-
-    for line in container.logs(stream=True, follow=True):
-        logger.info(line.decode("utf-8").strip())
-
-    result = container.wait()
-    container.remove()
-    return result.get("StatusCode", -1)
+    try:
+        kwargs = {}
+        if CONTAINER_NETWORK in os.environ:
+            kwargs["network_mode"] = os.environ[CONTAINER_NETWORK]
+        container = client.containers.run(
+            image=image,
+            volumes=bind_volumes,
+            command=shlex.split(command) if command else None,
+            environment=env_vars,
+            detach=True,
+            entrypoint=entrypoint,
+            user=0,
+            **kwargs
+            # auto_remove=True,
+        )
+        logger.info("Container ID: %s", container.id)
+        for line in container.logs(stream=True, follow=True):
+            print(line.decode("utf-8"), end="")
+
+        result = container.wait()
+        return result.get("StatusCode", -1)
+    except docker.errors.APIError as ex:
+        logger.error(ex.explanation)
+        return -1
+    finally:
+        # Remove the container
+        container.remove()