Implement ads init and ads run for marketplace operator

Author: amitkrprajapati

ads/opctl/backend/marketplace/local_marketplace.py

@@ -1,5 +1,15 @@
 import json
-from typing import Optional, Dict, Union
+import os
+import subprocess
+import tempfile
+import time
+from subprocess import CompletedProcess
+from typing import Optional, Dict, Union, Any
+
+import fsspec
+import yaml
+from kubernetes import config, client
+from kubernetes.watch import watch
 
 from ads.opctl.backend.marketplace.marketplace_type import (
  HelmMarketplaceListingDetails,
@@ -40,7 +50,7 @@ class LocalMarketplaceOperatorBackend(Backend):
  """
 
  def __init__(
- self, config: Optional[Dict], operator_info: OperatorInfo = None
+  self, config: Optional[Dict], operator_info: OperatorInfo = None
  ) -> None:
  """
  Instantiates the operator backend.
@@ -70,8 +80,62 @@ def __init__(
 
  self.operator_info = operator_info
 
+ def _set_kubernete_env(self) -> None:
+ os.environ["OCI_CLI_AUTH"] = 'security_token'
+ os.environ["OCI_CLI_PROFILE"] = 'SESSION_PROFILE'
+
+ def _run_helm_install(self, name, chart, **kwargs) -> CompletedProcess:
+ self._set_kubernete_env()
+ flags = []
+ for key, value in kwargs.items():
+ flags.extend([f"--{key}", f"{value}"])
+
+ helm_cmd = ["helm", "install", name, chart, *flags]
+ print("Running Command:", " ".join(helm_cmd))
+ return subprocess.run(helm_cmd)
+
+ def _save_helm_value_to_yaml(self, helm_values: Dict[str, Any]) -> str:
+ override_value_path = os.path.join(tempfile.TemporaryDirectory().name, f"values.yaml")
+ with fsspec.open(override_value_path, mode="w") as f:
+ f.write(yaml.dump(helm_values))
+ return override_value_path
+
+ def _delete_temp_file(self, temp_file) -> bool:
+ if os.path.exists(temp_file):
+ os.remove(temp_file)
+ return True
+
+ return False
+
+ def _wait_for_pod_ready(self, namespace, pod_name):
+ # Configs can be set in Configuration class directly or using helper utility
+ self._set_kubernete_env()
+ config.load_kube_config()
+ v1 = client.CoreV1Api()
+
+ def is_pod_ready(pod):
+ for condition in pod.status.conditions:
+ if condition.type == "Ready":
+ return condition.status == 'True'
+ return False
+
+ start_time = time.time()
+ timeout_seconds = 10 * 60
+ sleep_time = 20
+ while True:
+ pod = v1.list_namespaced_pod(namespace=namespace, label_selector=f"app.kubernetes.io/instance={pod_name}").items[0]
+ if is_pod_ready(pod):
+ return 0
+ if time.time() - start_time >= timeout_seconds:
+ print("Timed out waiting for pad to get ready.")
+ break
+ print(f"Waiting for pod {pod_name} to be ready...")
+ time.sleep(sleep_time)
+ return -1
+
  def _run_with_python(self, **kwargs: Dict) -> int:
- """Runs the operator within a local python environment.
+ """
+ Runs the operator within a local python environment.
 
  Returns
  -------
@@ -92,10 +156,28 @@ def _run_with_python(self, **kwargs: Dict) -> int:
  listing_details: MarketplaceListingDetails = operator.get_listing_details(
  operator_spec
  )
-
+ # operator_spec = self.operator_config['spec']
+ # helm_values = operator_spec['helmValues']
  ##Perform backend logic##
+ # name = 'fs-dp-api-test'
+ # chart = 'oci://iad.ocir.io/idogsu2ylimg/feature-store-dataplane-api/helm-chart/feature-store-dp-api'
+ if isinstance(listing_details, HelmMarketplaceListingDetails):
+ override_value_path = self._save_helm_value_to_yaml(listing_details.helm_values)
+ helm_install_status = self._run_helm_install(
+ name=listing_details.name,
+ chart=listing_details.chart,
+ **{
+ "version": listing_details.version,
+ "namespace": listing_details.namespace,
+ "values": override_value_path
+ }
+ )
 
- return 0
+ self._delete_temp_file(override_value_path)
+ if helm_install_status.returncode == 0:
+ return self._wait_for_pod_ready(listing_details.namespace, listing_details.name)
+ else:
+ return -1
 
  def run(self, **kwargs: Dict) -> None:
  """Runs the operator."""
@@ -130,11 +212,11 @@ def run(self, **kwargs: Dict) -> None:
  )
 
  def init(
- self,
- uri: Union[str, None] = None,
- overwrite: bool = False,
- runtime_type: Union[str, None] = None,
- **kwargs: Dict,
+  self,
+  uri: Union[str, None] = None,
+  overwrite: bool = False,
+  runtime_type: Union[str, None] = None,
+  **kwargs: Dict,
  ) -> Union[str, None]:
  """Generates a starter YAML specification for the operator local runtime.
 
@@ -177,8 +259,8 @@ def init(
 
  return (
  operator_runtime_const.MARKETPLACE_RUNTIME_MAP[runtime_type]
- .init(**RUNTIME_KWARGS_MAP[runtime_type])
- .to_yaml(
+  .init(**RUNTIME_KWARGS_MAP[runtime_type])
+  .to_yaml(
  uri=uri,
  overwrite=overwrite,
  note=note,

ads/opctl/backend/marketplace/marketplace_type.py

@@ -11,15 +11,21 @@ def __init__(self, listing_id: str):
 
 class HelmMarketplaceListingDetails(MarketplaceListingDetails):
  def __init__(
- self,
- listing_id: str,
- cluster_id: str,
- namespace: str,
- ocir_repo: str,
- helm_values,
+ self,
+ listing_id: str,
+ name: str,
+ chart: str,
+ version: str,
+ cluster_id: str,
+ namespace: str,
+ ocir_repo: str,
+ helm_values,
  ):
  super().__init__(listing_id)
  self.cluster_id = cluster_id
+ self.name = name
+ self.chart = chart
+ self.version = version
  self.namespace = namespace
  self.ocir_repo = ocir_repo
  self.helm_values = helm_values

ads/opctl/operator/lowcode/feature_store_marketplace/__main__.py

@@ -1,3 +1,4 @@
+import json
 import sys
 
 from ads.opctl.backend.marketplace.marketplace_operator_runner import (
@@ -12,12 +13,18 @@
 
 class FeatureStoreOperatorRunner(MarketplaceOperatorRunner):
  def get_listing_details(self, operator_config: str) -> MarketplaceListingDetails:
+ operator_config_json = json.loads(operator_config)
+ operator_config_spec = operator_config_json['spec']
  return HelmMarketplaceListingDetails(
- listing_id="sada",
- helm_values="{}",
- cluster_id="sds",
- namespace="asdas",
- ocir_repo="sad",
+ listing_id="<listing_id>",
+ name='fs-dp-api-test',
+ chart='oci://iad.ocir.io/idogsu2ylimg/feature-store-dataplane-api/helm-chart/feature-store-dp-api',
+ version="1.0",
+ helm_values=operator_config_spec['helmValues'],
+ cluster_id="<cluster id>",
+ namespace=operator_config_spec['clusterDetails']['namespace'],
+ ocir_repo="<ocir_repo>",
+
  )
 
 

ads/opctl/operator/lowcode/feature_store_marketplace/cmd.py

@@ -4,7 +4,7 @@
 # Copyright (c) 2023 Oracle and/or its affiliates.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
 
-from typing import Dict
+from typing import Dict, Any
 
 import click
 
@@ -14,6 +14,31 @@
 
 from .const import SupportedDatabases
 
+VAULT_OCID_PLACEHOLDER = '<VAULT_OCID>'
+SECRET_NAME_PLACEHOLDER = '<SECRET_NAME>'
+JDBC_CONNECTION_URL_PLACEHOLDER = '<JDBC_CONNECTION_URL>'
+DB_URL_PLACEHOLDER = '<DB_URL>'
+
+
+def __get_db_config(db_type: str) -> Dict[str, Any]:
+ if db_type == SupportedDatabases.MySQL:
+ return {
+ "mysqlDBConfig": {
+ "vaultOCID": VAULT_OCID_PLACEHOLDER,
+ "secretName": SECRET_NAME_PLACEHOLDER,
+ "jdbcConnectionUrl": JDBC_CONNECTION_URL_PLACEHOLDER,
+ }
+ }
+ elif db_type == SupportedDatabases.ATP:
+ return {
+ "atpDBConfig": {
+ "vaultOCID": JDBC_CONNECTION_URL_PLACEHOLDER,
+ "dbURL": DB_URL_PLACEHOLDER
+ }
+ }
+ else:
+ return {}
+
 
 def init(**kwargs: Dict) -> dict:
  """
@@ -39,7 +64,7 @@ def init(**kwargs: Dict) -> dict:
  type=click.Choice(SupportedDatabases.values()),
  default=SupportedDatabases.MySQL,
  )
-
+ selected_db_config = __get_db_config(db_type)
  return YamlGenerator(
  schema=_load_yaml_from_uri(__file__.replace("cmd.py", "schema.yaml"))
- ).generate_example_dict(values={"configuredDB": db_type})
+ ).generate_example_dict(values={"configuredDB": db_type, **selected_db_config})