SIENTIAPDE-1174

Update dependencies, modify replica count, and implement metrics tracking

- Updated sientia-dataops-library version from 1.3.5 to 1.3.7 in requirements.txt.
- Changed replicaCount in values.yaml from 5 to 3 and incremented image tag from 0.2.7 to 0.3.1.
- Added Prometheus metrics tracking in gates.py and worker.py, including a new write_metrics method.
- Configured Prometheus service and ServiceMonitor in values.yaml for metrics collection.

laborious/activities/gates.py

@@ -15,6 +15,7 @@ with workflow.unsafe.imports_passed_through():
     )
     from pandas import DataFrame
     from datetime import datetime
+    from laborious import metrics
 
 input_filter_functions = {
     'SPECIFIC_VARIABLES_NULL_VALUES': filter_specific_variables_null_values,
@@ -309,3 +310,34 @@ class Gates(BaseActivity):
         if data.empty:
             return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
         return max(data['timestamp'].values.tolist())
+
+    @activity.defn(name="write_metrics")
+    async def write_metrics(self, input_data: dict[str, Any]):
+        """
+        Write metrics to the database.
+        input_data:
+            metadata: dict[str, Any]
+            prediction: dict[str, Any]
+        """
+        metadata = input_data['metadata']
+        prediction = DataFrame(input_data['prediction'])
+        prediction_confidence = prediction['prediction_confidence'].values[0]
+        response_time = prediction['response_time'].values[0]
+
+        metrics.PREDICTIONS_WRITTEN_COUNT.labels(
+            pod_id=self.pod_id,
+            model_name=metadata['model_name'],
+            pipeline_name=metadata['workflow_name']
+        ).inc()
+
+        metrics.PREDICTION_CONFIDENCE_MONITOR.labels(
+            pod_id=self.pod_id,
+            model_name=metadata['model_name'],
+            pipeline_name=metadata['workflow_name']
+        ).set(prediction_confidence)
+
+        metrics.PREDICTION_RESPONSE_TIME_MONITOR.labels(
+            pod_id=self.pod_id,
+            model_name=metadata['model_name'],
+            pipeline_name=metadata['workflow_name']
+        ).set(response_time)

laborious/metrics.py

@@ -0,0 +1,27 @@
+from prometheus_client import Gauge, Counter
+
+APP_UP = Gauge(
+    "app_up",
+    "Indicates if the application is running (1) or shutting down (0)",
+    ["pod_id"],
+)
+
+CORE_LABELS = ["pod_id", "model_name", "pipeline_name"]
+
+PREDICTIONS_WRITTEN_COUNT = Counter(
+    "laborious_predictions_written_count",
+    "Number of predictions written to the database table predictions",
+    CORE_LABELS,
+)
+
+PREDICTION_CONFIDENCE_MONITOR = Gauge(
+    "laborious_prediction_confidence_monitor",
+    "Current confidence of each prediction",
+    CORE_LABELS,
+)
+
+PREDICTION_RESPONSE_TIME_MONITOR = Gauge(
+    "laborious_prediction_response_time_monitor",
+    "Current response time of each prediction",
+    CORE_LABELS,
+)

laborious/utils/repository/model_repository.py

@@ -68,7 +68,7 @@ class MLFlowRepository():
         try:
             start_time = datetime.now()
             data = self.model_serving.get_cached_predict(
-                model_name, data, model_retention)[-1:]
+                model_name, data, model_retention)[0:1]
 
             end_time = datetime.now()
             data = pd.DataFrame(data, columns=['prediction'])

laborious/worker/worker.py

@@ -20,13 +20,20 @@ with workflow.unsafe.imports_passed_through():
     )
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.temporal.utils.logger import get_logger
+    from laborious import metrics
+    from prometheus_client import start_http_server
+
+POD_ID = os.getenv('POD_ID')
 
 
 async def main():
     host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
     logger = get_logger(__name__)
 
-    logger.info('Starting Worker...')
+    logger.info(f'Starting Worker with POD_ID: {POD_ID}')
+
+    logger.info("Starting prometheus client...")
+    start_prometheus_server()
 
     logger.info('Starting Notification Handler...')
 
@@ -125,5 +132,17 @@ async def main():
         # Exit with a non-zero status code to indicate failure to Kubernetes
         sys.exit(1)
 
+
+def start_prometheus_server():
+    try:
+        port = int(os.getenv("HTTP_METRICS_PORT", 9090))
+        start_http_server(port)
+        print(f"Prometheus server started on port {port}.")
+        metrics.APP_UP.labels(pod_id=POD_ID).set(1)  # Mark app as UP
+    except Exception as e:
+        print(f"Failed to start Prometheus server: {e}")
+        os._exit(1)
+
+
 if __name__ == '__main__':
     asyncio.run(main())

laborious/workflows/sub_workflows/format_and_export_prediction.py

@@ -95,3 +95,13 @@ class FormatAndExportPrediction():
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=60)
         )
+
+        await workflow.execute_activity_method(
+            Activities.write_metrics,
+            {
+                **metadata,
+                'prediction': prediction
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )