adding implementation for artifactory option

Author: Ziqun Ye

ads/opctl/backend/ads_model_deployment.py

@@ -175,7 +175,7 @@ def watch(self) -> None:
 
     def predict(self) -> None:
         ocid = self.config["execution"].get("ocid")
-        data = self.config["execution"].get("data")
+        data = self.config["execution"].get("payload")
         with AuthContext(auth=self.auth_type, profile=self.profile):
             model_deployment = ModelDeployment.from_id(ocid)
             data = json.loads(data)

ads/opctl/backend/local.py

@@ -7,6 +7,7 @@
 import copy
 import json
 import os
+import shutil
 import tempfile
 from concurrent.futures import Future, ThreadPoolExecutor
 from time import sleep
@@ -18,13 +19,16 @@
 from ads.common.decorator.runtime_dependency import (OptionalDependency,
                                                      runtime_dependency)
 from ads.common.oci_client import OCIClientFactory
-from ads.model.model_metadata import ModelCustomMetadata, ModelTaxonomyMetadata
+from ads.model.datascience_model import DataScienceModel
+from ads.model.model_metadata import ModelCustomMetadata
 from ads.opctl import logger
 from ads.opctl.backend.base import Backend
+from ads.opctl.conda.cmds import _install
 from ads.opctl.config.resolver import ConfigResolver
 from ads.opctl.constants import (DEFAULT_IMAGE_CONDA_DIR,
                                  DEFAULT_IMAGE_HOME_DIR,
                                  DEFAULT_IMAGE_SCRIPT_DIR,
+                                 DEFAULT_MODEL_FOLDER,
                                  DEFAULT_NOTEBOOK_SESSION_CONDA_DIR,
                                  DEFAULT_NOTEBOOK_SESSION_SPARK_CONF_DIR,
                                  ML_JOB_GPU_IMAGE, ML_JOB_IMAGE)
@@ -35,9 +39,7 @@
                              is_in_notebook_session, run_command,
                              run_container)
 from ads.pipeline.ads_pipeline import Pipeline, PipelineStep
-from ads.model.datascience_model import DataScienceModel
-
-DEFAULT_MODEL_FOLDER = "~/.ads_ops/models"
+from ads.model.runtime.runtime_info import RuntimeInfo
 
 
 class CondaPackNotFound(Exception):
@@ -55,6 +57,17 @@ def __init__(self, config: Dict) -> None:
             dictionary of configurations
         """
         self.config = config
+        self.auth_type = config["execution"].get("auth")
+        self.profile = config["execution"].get("oci_profile", None)
+        self.oci_config = config["execution"].get("oci_config", None)
+        
+        self.oci_auth = create_signer(
+            self.auth_type,
+            self.oci_config,
+            self.profile ,
+        )
+        
+        self.client = OCIClientFactory(**self.oci_auth).data_science
 
     def run(self):
         if self.config.get("version") == "v1.0":
@@ -179,15 +192,15 @@ def init_vscode_container(self) -> None:
                 f.write(json.dumps(dev_container, indent=2))
             print(f"File {os.path.abspath('.devcontainer.json')} created.")
 
-    def _run_with_conda_pack(self, bind_volumes: Dict, extra_cmd: str="") -> int:
+    def _run_with_conda_pack(self, bind_volumes: Dict, extra_cmd: str="", install: bool=False, conda_uri: str="") -> int:
         env_vars = self.config["execution"].get("env_vars", {})
         slug = self.config["execution"]["conda_slug"]
         image = self.config["execution"].get("image", None)
 
         # bind_volumes is modified in-place and does not need to be returned
         # it is returned just to be explicit that it is changed during this function call
         bind_volumes, env_vars = self._check_conda_pack_and_install_if_applicable(
-            slug, bind_volumes, env_vars
+            slug, bind_volumes, env_vars, install=install, conda_uri=conda_uri
         )
         bind_volumes = self._mount_source_folder_if_exists(bind_volumes)
         command = self._build_command_for_conda_run(extra_cmd)
@@ -274,6 +287,7 @@ def _run_with_image(self, bind_volumes: Dict) -> int:
         if self.config["execution"].get("source_folder", None):
             bind_volumes.update(self._mount_source_folder_if_exists(bind_volumes))
         bind_volumes.update(self.config["execution"]["volumes"])
+        
         return run_container(image, bind_volumes, env_vars, command, entrypoint)
 
     def _run_with_image_v1(self, bind_volumes: Dict) -> int:
@@ -296,15 +310,26 @@ def _run_with_image_v1(self, bind_volumes: Dict) -> int:
         )
 
     def _check_conda_pack_and_install_if_applicable(
-        self, slug: str, bind_volumes: Dict, env_vars: Dict
+        self, slug: str, bind_volumes: Dict, env_vars: Dict, install: bool=False, conda_uri: str = None
     ) -> Dict:
+        conda_pack_folder = os.path.abspath(os.path.expanduser(self.config['execution']["conda_pack_folder"]))
         conda_pack_path = os.path.join(
-            os.path.expanduser(self.config["execution"]["conda_pack_folder"]), slug
+            conda_pack_folder, slug
         )
         if not os.path.exists(conda_pack_path):
-            raise CondaPackNotFound(
-                f"Conda pack {conda_pack_path} not found. Please run `ads opctl conda create` or `ads opctl conda install`."
-            )
+            if install:
+                logger.info(f"Downloading the conda pack {slug} to this conda pack {conda_pack_folder}. If this conda pack is already installed locally in a different location, pass in `conda_pack_folder` to avoid downloading it again.")
+                _install(
+                    conda_uri=conda_uri,
+                    conda_pack_folder=conda_pack_folder,
+                    oci_config=self.oci_config,
+                    oci_profile=self.profile,
+                    auth_type=self.auth_type,
+                )
+            else:
+                raise CondaPackNotFound(
+                    f"Conda pack {conda_pack_path} not found. Please run `ads opctl conda create` or `ads opctl conda install`."
+                )
         if os.path.exists(os.path.join(conda_pack_path, "spark-defaults.conf")):
             if not is_in_notebook_session():
                 env_vars["SPARK_CONF_DIR"] = os.path.join(DEFAULT_IMAGE_CONDA_DIR, slug)
@@ -617,26 +642,35 @@ def _log_orchestration_message(self, str: str) -> None:
 class LocalModelDeploymentBackend(LocalBackend):
     def __init__(self, config: Dict) -> None:
         super().__init__(config)
-        self.oci_auth = create_signer(
-            config["execution"].get("auth"),
-            config["execution"].get("oci_config", None),
-            config["execution"].get("oci_profile", None),
-        )
-        self.auth_type = config["execution"].get("auth")
-        self.profile = config["execution"].get("oci_profile", None)
-        self.client = OCIClientFactory(**self.oci_auth).data_science
 
     def predict(self) -> None:
+        artifact_directory = self.config["execution"].get("artifact_directory")
         ocid = self.config["execution"].get("ocid")
-        data = self.config["execution"].get("data")
-        model_folder = self.config["execution"].get("model_folder", DEFAULT_MODEL_FOLDER)
-        conda_slug, conda_path = self._get_conda_info(ocid)
+        data = self.config["execution"].get("payload")
+        model_folder = os.path.expanduser(self.config["execution"].get("model_save_folder", DEFAULT_MODEL_FOLDER))
+        artifact_directory = artifact_directory or os.path.join(model_folder, str(ocid))
+        if ocid and (not os.path.exists(artifact_directory) or len(os.listdir(artifact_directory)) == 0):
+            
+            region = self.config["execution"].get("region", None)
+            bucket_uri = self.config["execution"].get("bucket_uri", None)
+            timeout = self.config["execution"].get("timeout", None)
+            logger.info(f"No cached model found. Downloading the model {ocid} to {artifact_directory}. If you already have a copy of the model, specify `artifact_directory` instead of `ocid`. You can specify `model_save_folder` to decide where to store the model artifacts.")
+            self._download_model(ocid=ocid, artifact_directory=artifact_directory, region=region, bucket_uri=bucket_uri, timeout=timeout)
+
+        if ocid:
+            conda_slug, conda_path = self._get_conda_info_from_catalog(ocid)
+        elif artifact_directory:
+            if not os.path.exists(artifact_directory) or len(os.listdir(artifact_directory)) == 0:
+                raise ValueError(f"`artifact_directory` {artifact_directory} does not exist or is empty.")
+            conda_slug, conda_path = self._get_conda_info_from_runtime(artifact_dir=artifact_directory)
+        else:
+            raise ValueError("Conda information cannot be detected.")
         compartment_id = self.config["execution"].get("compartment_id", self.config["infrastructure"].get("compartment_id"))
         project_id = self.config["execution"].get("project_id", self.config["infrastructure"].get("project_id"))
         if not compartment_id or not project_id:
             raise ValueError("`compartment_id` and `project_id` must be provided.")
 
-        extra_cmd = ocid + " " + data + " " + compartment_id + " " + project_id
+        extra_cmd = "/opt/ds/model/deployed_model/ " + data + " " + compartment_id + " " + project_id
         bind_volumes = {}
         if not is_in_notebook_session():
             bind_volumes = {
@@ -647,17 +681,18 @@ def predict(self) -> None:
             dir_path = os.path.dirname(os.path.realpath(__file__))
             script = "script.py"
             self.config["execution"]["source_folder"] = os.path.abspath(os.path.join(dir_path, ".."))
-
             self.config["execution"]["entrypoint"] = script
-            bind_volumes[os.path.join(model_folder)] = {"bind": script}
+            bind_volumes[artifact_directory] = {"bind": "/opt/ds/model/deployed_model/"}
+            
         if self.config["execution"].get("image"):
             exit_code = self._run_with_image(bind_volumes)
         elif self.config["execution"].get("conda_slug", conda_slug):
             self.config["execution"]["image"] = ML_JOB_IMAGE
             if not self.config["execution"].get("conda_slug"):
                 self.config["execution"]["conda_slug"] = conda_slug
+                self.config["execution"]["slug"] = conda_slug
             self.config["execution"]["conda_path"] = conda_path
-            exit_code = self._run_with_conda_pack(bind_volumes, extra_cmd)
+            exit_code = self._run_with_conda_pack(bind_volumes, extra_cmd, install=True, conda_uri=conda_path)
         else:
             raise ValueError("Either conda pack info or image should be specified.")
 
@@ -667,25 +702,39 @@ def predict(self) -> None:
                 f"Run with the --debug argument to view container logs."
             )
 
-    def _download_model(self, ocid, region, bucket_uri, timeout):
-        dsc_model = DataScienceModel.from_id(ocid)
-        dsc_model.download_artifact(
-        target_dir=os.path.join(self.config["execution"].get("source_folder", DEFAULT_MODEL_FOLDER), ocid),
-        force_overwrite=True,
-        overwrite_existing_artifact=True,
-        remove_existing_artifact=True,
-        auth=self.oci_auth,
-        region=region,
-        timeout=timeout,
-        bucket_uri=bucket_uri,
-        )
-    
-    def _get_conda_info(self, ocid):
+    def _download_model(self, ocid, artifact_directory, region, bucket_uri, timeout):
+        os.makedirs(artifact_directory, exist_ok=True)
+        os.chmod(artifact_directory, 777)
+        
+        try:
+            dsc_model = DataScienceModel.from_id(ocid)
+            dsc_model.download_artifact(
+            target_dir=artifact_directory,
+            force_overwrite=True,
+            overwrite_existing_artifact=True,
+            remove_existing_artifact=True,
+            auth=self.oci_auth,
+            region=region,
+            timeout=timeout,
+            bucket_uri=bucket_uri,
+            )
+        except:
+            shutil.rmtree(artifact_directory, ignore_errors=True)
+        
+    def _get_conda_info_from_catalog(self, ocid):
         response = self.client.get_model(ocid)
         custom_metadata = ModelCustomMetadata._from_oci_metadata(response.data.custom_metadata_list)
-        conda_env_path = custom_metadata['CondaEnvironmentPath'].value
+        conda_path = custom_metadata['CondaEnvironmentPath'].value
         conda_slug = custom_metadata['SlugName'].value
-        return conda_slug, conda_env_path
+        return conda_slug, conda_path
+    
+    def _get_conda_info_from_runtime(self, artifact_dir):
+        runtime_yaml_file = os.path.join(artifact_dir, "runtime.yaml")
+        runtime_info = RuntimeInfo.from_yaml(uri=runtime_yaml_file)
+        conda_slug = runtime_info.model_deployment.inference_conda_env.inference_env_slug
+        conda_path = runtime_info.model_deployment.inference_conda_env.inference_env_path
+        return conda_slug, conda_path
+        
 
     def _run_with_image(self, bind_volumes):
         ocid = self.config["execution"].get("ocid")

ads/opctl/cli.py

@@ -28,6 +28,7 @@
 from ads.opctl.utils import suppress_traceback
 from ads.opctl.config.merger import ConfigMerger
 from ads.opctl.constants import BACKEND_NAME
+from ads.opctl.backend.local import DEFAULT_MODEL_FOLDER
 
 import ads.opctl.conda.cli
 import ads.opctl.spark.cli
@@ -498,19 +499,25 @@ def deactivate(**kwargs):
     Deactivates a data science service.
     """
     suppress_traceback(kwargs["debug"])(deactivate_cmd)(**kwargs)
-    
-    
+
+
 @commands.command()
-@click.argument("ocid", nargs=1)
-@click.argument("data", nargs=1)
-@click.argument("conda_slug", nargs=1, required=False)
-@add_options(_options)
+@click.option("--ocid", nargs=1, required=False, help="This can be either a model id or model deployment id. When model id is passed, it conducts a local predict/test. This is designed for local dev purpose in order to test whether deployment will be successful locally. When you pass in `model_save_folder`, the model artifact will be downloaded and saved to a subdirectory of `model_save_folder` where model id is the name of subdirectory. Or you can pass in a model deployment id and this will invoke the remote endpoint and conduct a prediction on the server.")
+@click.option("--model-save-folder", nargs=1, required=False, default=DEFAULT_MODEL_FOLDER, help="Which location to store model artifact folders. Defaults to ~/.ads_ops/models. This is only used when model id is passed to `ocid` and a local predict is conducted.")
+@click.option("--conda-pack-folder", nargs=1, required=False, help="Which location to store the conda pack locally. Defaults to ~/.ads_ops/conda. This is only used when model id is passed to `ocid` and a local predict is conducted.")
+@click.option("--bucket-uri", nargs=1, required=False, help="The OCI Object Storage URI where model artifacts will be copied to. The `bucket_uri` is only necessary for uploading large artifacts which size is greater than 2GB. Example: `oci://<bucket_name>@<namespace>/prefix/`. This is only used when the model id is passed.")
+@click.option("--region", nargs=1, required=False, help="The destination Object Storage bucket region. By default the value will be extracted from the `OCI_REGION_METADATA` environment variables. This is only used when the model id is passed.")
+@click.option("--timeout", nargs=1, required=False, help="The connection timeout in seconds for the client. This is only used when the model id is passed.")
+@click.option("--artifact-directory", nargs=1, required=False, default=None, help="The artifact directory where stores your models, score.py and etc. This is used when you have a model artifact locally and have not saved it to the model catalog yet. In this case, you dont need to pass in model ")
+@click.option("--payload", nargs=1, help="The payload sent to the model for prediction.")
+@click.option("--conda-slug", nargs=1, required=False, help="The conda env used to load the model and conduct the prediction. This is only used when model id is passed to `ocid` and a local predict is conducted. It should match the inference conda env specified in the runtime.yaml file which is the conda pack being used when conducting real model deployment.")
+@click.option("--debug", "-d", help="set debug mode", is_flag=True, default=False)
 def predict(**kwargs):
     """
     Deactivates a data science service.
     """
     suppress_traceback(kwargs["debug"])(predict_cmd)(**kwargs)
-    
+
 
 commands.add_command(ads.opctl.conda.cli.commands)
 commands.add_command(ads.opctl.spark.cli.commands)

ads/opctl/cmds.py

@@ -488,7 +488,8 @@ def predict(**kwargs) -> None:
     p = ConfigProcessor().step(ConfigMerger, **kwargs)
     if "datasciencemodeldeployment" in p.config["execution"].get("ocid", ""):
         return ModelDeploymentBackend(p.config).predict()
-    elif "datasciencemodel" in p.config["execution"].get("ocid", ""):
+    else:
+        # model ocid or artifact directory
         return LocalModelDeploymentBackend(p.config).predict()
 
 

ads/opctl/constants.py

@@ -9,6 +9,7 @@
 DEFAULT_OCI_CONFIG_FILE = "~/.oci/config"
 DEFAULT_PROFILE = "DEFAULT"
 DEFAULT_CONDA_PACK_FOLDER = "~/conda"
+DEFAULT_MODEL_FOLDER = "~/.ads_ops/models"
 CONDA_PACK_OS_PREFIX_FORMAT = "oci://<bucket>@<namespace>/<prefix>"
 DEFAULT_ADS_CONFIG_FOLDER = "~/.ads_ops"
 OPS_IMAGE_BASE = "ads-operators-base"

ads/opctl/script.py

@@ -7,7 +7,7 @@
 
 def verify(artifact_dir, data, compartment_id, project_id):
     with tempfile.TemporaryDirectory() as td:
-        model = GenericModel.from_model_artifact(artifact_dir=artifact_dir,
+        model = GenericModel.from_model_artifact(uri = artifact_dir, artifact_dir=artifact_dir,
                                                 force_overwrite=True,
                                                 compartment_id=compartment_id,
                                                 project_id=project_id)