watching logs and pushing them to telemetry

Author: agrimk

ads/aqua/modeldeployment/deployment.py

@@ -4,6 +4,7 @@
 
 
 import json
+import re
 import shlex
 import threading
 from datetime import datetime, timedelta
@@ -732,28 +733,18 @@ def _create_deployment(
             f"Aqua model deployment {deployment_id} created for model {aqua_model_id}. Work request Id is {deployment.dsc_model_deployment.workflow_req_id}"
         )
         status_list = []
-        progress_thread_1 = threading.Thread(
-            target=deployment.watch,
-            args=(status_list),
-            daemon=True,
-        )
-        progress_thread_1.start()
 
-        progress_thread_2 = threading.Thread(
+        progress_thread = threading.Thread(
             target=self.get_deployment_status,
             args=(
                 deployment_id,
                 deployment.dsc_model_deployment.workflow_req_id,
                 model_type,
                 model_name,
-                status_list,
             ),
             daemon=True,
         )
-        progress_thread_2.start()
-
-        progress_thread_1.join()
-        progress_thread_2.join()
+        progress_thread.start()
 
         # we arbitrarily choose last 8 characters of OCID to identify MD in telemetry
         telemetry_kwargs = {"ocid": get_ocid_substring(deployment_id, key_len=8)}
@@ -1245,11 +1236,11 @@ def list_shapes(self, **kwargs) -> List[ComputeShapeSummary]:
 
     def get_deployment_status(
         self,
+        deployment,
         model_deployment_id: str,
         work_request_id: str,
         model_type: str,
         model_name: str,
-        status_list: List[str] = [],
     ) -> None:
         """Waits for the data science  model deployment to be completed and log its status in telemetry.
 
@@ -1262,17 +1253,14 @@ def get_deployment_status(
             The work request Id of the model deployment.
         model_type: str
             The type of aqua model to be deployed. Allowed values are: `custom`, `service` and `multi_model`.
-        status_list: List[str]
-            The list of status frmo streams the access and/or predict logs of model deployment.
 
         Returns
         -------
         AquaDeployment
             An Aqua deployment instance.
         """
-
         ocid = get_ocid_substring(model_deployment_id, key_len=8)
-        telemetry_kwargs = {"ocid": ocid, "model_name": model_name}
+        status_list: List[str] = []
 
         data_science_work_request: DataScienceWorkRequest = DataScienceWorkRequest(
             work_request_id
@@ -1284,25 +1272,104 @@ def get_deployment_status(
                 max_wait_time=DEFAULT_WAIT_TIME,
                 poll_interval=DEFAULT_POLL_INTERVAL,
             )
-        except Exception:
+            predict_logs = deployment.tail_logs("predict")
+            access_logs = deployment.tail_logs("access")
+
+            status = ""
+            if access_logs and len(access_logs) > 0:
+                print("access log list ############################")
+                print(access_logs)
+                status = access_logs[0]["message"]
+
+            if predict_logs and len(predict_logs) > 0:
+                print("predict_logs  ############################")
+                print(predict_logs)
+                status += predict_logs[0]["message"]
+
+            status = re.sub(r"[^a-zA-Z0-9]", "", status)
+            telemetry_kwargs = {
+                "ocid": ocid,
+                "model_name": model_name,
+                "status": status,
+            }
+            print(telemetry_kwargs)
+            print("############################")
+
+            self.telemetry.record_event(
+                category=f"aqua/{model_type}/deployment/status",
+                action="LAST_LOG",
+                # detail=error_str,
+                **telemetry_kwargs,
+            )
+
+        except Exception as e:
             if data_science_work_request._error_message:
                 error_str = ""
                 for error in data_science_work_request._error_message:
                     error_str = error_str + " " + error.message
 
                 status = ""
-                if len(status_list) > 0:
-                    status = status_list[-1]
-
-                telemetry_kwargs["status"] = status
+                predict_logs = deployment.tail_logs("predict")
+                access_logs = deployment.tail_logs("access")
+                if access_logs and len(access_logs) > 0:
+                    print(access_logs)
+                    status = access_logs[0]["message"]
+
+                if predict_logs and len(predict_logs) > 0:
+                    print("predict_logs  ############################")
+                    print(predict_logs)
+                    status += predict_logs[0]["message"]
+                status = re.sub(r"[^a-zA-Z0-9]", "", status)
+                error_str = re.sub(r"[^a-zA-Z0-9]", "", error_str)
+                telemetry_kwargs = {
+                    "ocid": ocid,
+                    "model_name": model_name,
+                    "status": error_str + " " + status,
+                }
+                print(telemetry_kwargs)
+                print("############################")
 
                 self.telemetry.record_event(
                     category=f"aqua/{model_type}/deployment/status",
                     action="FAILED",
                     detail=error_str,
                     **telemetry_kwargs,
                 )
+            else:
+                print(str(e))
+                status = str(e)
+                predict_logs = deployment.tail_logs("predict")
+                access_logs = deployment.tail_logs("access")
+                if access_logs and len(access_logs) > 0:
+                    print("access log list ############################")
+                    print(access_logs)
+                    status = access_logs[0]["message"]
+
+                if predict_logs and len(predict_logs) > 0:
+                    print("predict_logs  ############################")
+                    print(predict_logs)
+                    status += predict_logs[0]["message"]
+
+                status = re.sub(r"[^a-zA-Z0-9]", "", status)
+                error_str = re.sub(r"[^a-zA-Z0-9]", "", error_str)
+
+                telemetry_kwargs = {
+                    "ocid": ocid,
+                    "model_name": model_name,
+                    "status": error_str + " " + status,
+                }
+                print(telemetry_kwargs)
+                print("############################")
+
+                self.telemetry.record_event(
+                    category=f"aqua/{model_type}/deployment/status",
+                    action="FAILED",
+                    # detail=error_str,
+                    **telemetry_kwargs,
+                )
+
         else:
+            telemetry_kwargs = {"ocid": ocid, "model_name": model_name}
             self.telemetry.record_event_async(
                 category=f"aqua/{model_type}/deployment/status",
                 action="SUCCEEDED",

ads/common/oci_logging.py

@@ -1,22 +1,21 @@
 #!/usr/bin/env python
-# -*- coding: utf-8; -*-
 
 # Copyright (c) 2021, 2024 Oracle and/or its affiliates.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
 
 import datetime
 import logging
 import time
-from typing import Dict, Union, List
+from typing import Dict, List, Union
 
+import oci.exceptions
 import oci.logging
 import oci.loggingsearch
-import oci.exceptions
+
 from ads.common.decorator.utils import class_or_instance_method
 from ads.common.oci_mixin import OCIModelMixin, OCIWorkRequestMixin
 from ads.common.oci_resource import OCIResource, ResourceNotFoundError
 
-
 logger = logging.getLogger(__name__)
 
 # Maximum number of log records to be returned by default.
@@ -862,9 +861,48 @@ def tail(
             time_start=time_start,
             log_filter=log_filter,
         )
-        self._print(
-            sorted(tail_logs, key=lambda log: log["time"])
+        self._print(sorted(tail_logs, key=lambda log: log["time"]))
+
+    def get_tail_logs(
+        self,
+        source: str = None,
+        limit: int = LOG_RECORDS_LIMIT,
+        time_start: datetime.datetime = None,
+        log_filter: str = None,
+    ) -> List[Union[oci.loggingsearch.models.SearchResult, dict]]:
+        """Returns the most recent consolidated log records.
+
+        Parameters
+        ----------
+        source : str, optional
+            Expression or OCID to filter the "source" field of the OCI log record.
+            Defaults to None.
+        limit : int, optional.
+            Maximum number of records to be returned.
+            If limit is set to None, all logs from time_start to now will be returned.
+            Defaults to 100.
+        time_start : datetime.datetime, optional
+            Starting time for the log query.
+            Defaults to None.
+        log_filter : str, optional
+            Expression for filtering the logs. This will be the WHERE clause of the query.
+            Defaults to None.
+
+        Returns
+        -------
+        list
+            A list of oci.loggingsearch.models.SearchResult objects or log records sorted in descending order by time
+            Each log record is a dictionary with the following keys: `annotation`, `id`, `time`,
+            `message` and `datetime`.
+        """
+        tail_logs = self._search_and_format(
+            source=source,
+            limit=limit,
+            sort_order=SortOrder.DESC,
+            time_start=time_start,
+            log_filter=log_filter,
         )
+        return sorted(tail_logs, key=lambda log: log["time"])
 
     def head(
         self,

ads/model/deployment/model_deployment.py

@@ -729,13 +729,46 @@ def update(
 
         return self._update_from_oci_model(response)
 
+    def tail_logs(
+        self, log_type: str = None, time_start: datetime = None, log_filter: str = None
+    ) -> List[Union[oci.loggingsearch.models.SearchResult, dict]]:
+        """Returns the most recent consolidated log records for the model deployment
+
+        Parameters
+        ----------
+        source : str, optional
+            Expression or OCID to filter the "source" field of the OCI log record.
+            Defaults to None.
+        limit : int, optional.
+            Maximum number of records to be returned.
+            If limit is set to None, all logs from time_start to now will be returned.
+            Defaults to 100.
+        time_start : datetime.datetime, optional
+            Starting time for the log query.
+            Defaults to None.
+        log_filter : str, optional
+            Expression for filtering the logs. This will be the WHERE clause of the query.
+            Defaults to None.
+
+        Returns
+        -------
+        list
+            A list of oci.loggingsearch.models.SearchResult objects or log records sorted in descending order by time
+            Each log record is a dictionary with the following keys: `annotation`, `id`, `time`,
+            `message` and `datetime`.
+        """
+        return self.logs(log_type).get_tail_logs(
+            source=self.model_deployment_id,
+            time_start=time_start,
+            log_filter=log_filter,
+        )
+
     def watch(
         self,
         log_type: str = None,
         time_start: datetime = None,
         interval: int = LOG_INTERVAL,
         log_filter: str = None,
-        status_list: List[str] = None,
     ) -> "ModelDeployment":
         """Streams the access and/or predict logs of model deployment.
 
@@ -764,8 +797,6 @@ def watch(
         status = ""
         while not self._stop_condition():
             status = self._check_and_print_status(status)
-            if status not in status_list:
-                status_list.append(status)
             time.sleep(interval)
 
         time_start = time_start or self.time_created