Merge pull request #14 from Aignosi/SIENTIAPDE-1174-mapear-e-implementar-metricas-a-serem-criadas

Sientiapde 1174 mapear e implementar metricas a serem criadas

requirements.txt

@@ -5,6 +5,7 @@ asyncua
 redis
 aiokafka
 pymongo
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.5
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.7
 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.5
-pydruid[pandas]
+pydruid[pandas]
+prometheus-client

scouter/activities/activities.py

@@ -8,6 +8,7 @@ with workflow.unsafe.imports_passed_through():
     from scouter.activities.gates import Gates
     from scouter.activities.mongodb import MongoDB
     from typing import Any
+    from os import getenv
 
 
 class Activities(Postgres, Redis, Gates, MongoDB,):
@@ -61,6 +62,8 @@ class Activities(Postgres, Redis, Gates, MongoDB,):
             notification_handler=notification_handler
         )
 
+        self.pod_id = getenv("HOSTNAME", "localhost")
+
     def shutdown(self):
         Postgres.close(self)
         MongoDB.shutdown(self)

scouter/activities/gates.py

@@ -1,8 +1,12 @@
 from temporalio import workflow, activity
 
+from scouter import metrics
+
 with workflow.unsafe.imports_passed_through():
     from sientia_do.notifications.models import NotificationLevel
     from sientia_do.temporal.activities.base import BaseActivity
+    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.temporal.utils.logger import Logger
     from scouter.utils.quality.filters import null_values_filter, out_of_bounds_filter
     from typing import Any
     import traceback
@@ -16,6 +20,10 @@ quality_gate_filters = {
 
 class Gates(BaseActivity):
 
+    def __init__(self, logger: Logger, notification_handler: NotificationHandler):
+        BaseActivity.__init__(
+            self, logger, notification_handler, set_error_counter=True)
+
     def apply_aggregation(self, group: DataFrame, aggr_function: str,
                           metadata: dict[str, Any]) -> float | None | str:
         """
@@ -68,7 +76,7 @@ class Gates(BaseActivity):
         Args:
             input_data (dict[str, Any]): The data to aggregate. Contains:
                 data (dict[str, Any]): The time series data.
-                model_tags (dict[str, Any]): The tags configuration 
+                model_tags (dict[str, Any]): The tags configuration
                 containing aggregation functions.
 
         Returns:
@@ -247,3 +255,18 @@ class Gates(BaseActivity):
         )
 
         return data.to_dict()
+
+    @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]
+        """
+        metadata = input_data['metadata']
+
+        metrics.LABORIOUS_DATA_WRITTEN_COUNT.labels(
+            pod_id=self.pod_id,
+            model_name=metadata['model_name'],
+            pipeline_name=metadata['workflow_name']
+        ).inc()

scouter/activities/mongodb.py

@@ -57,7 +57,8 @@ class MongoDB(BaseActivity):
 
         BaseActivity.__init__(self,
                               logger=logger,
-                              notification_handler=notification_handler)
+                              notification_handler=notification_handler,
+                              set_error_counter=True)
 
     def shutdown(self):
         """

scouter/activities/redis.py

@@ -1,8 +1,8 @@
-import traceback
 from temporalio import workflow, activity
 
 with workflow.unsafe.imports_passed_through():
     from logging import Logger
+    import traceback
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.notifications.models import NotificationLevel
     from sientia_do.temporal.activities.redis_base import Redis as RedisBase
@@ -10,6 +10,7 @@ with workflow.unsafe.imports_passed_through():
     from typing import Any
     from pandas import DataFrame
     from datetime import datetime
+    from scouter import metrics
 
 
 class Redis(RedisBase):
@@ -136,16 +137,29 @@ class Redis(RedisBase):
                 return data_hold
 
         try:
+
+            to_register_metrics = []
             for _, row in data.iterrows():
                 value = row['value']
 
                 data_hold[row['name']] = value
+                to_register_metrics.append(
+                    (row['name'], value))
 
             data_hold['timestamp'] = data['timestamp'].max() if not data.empty else \
                 datetime.now().strftime("%Y-%m-%d %H:%M:%S")
 
             self.set(key, data_hold, ttl=retention_time)
 
+            # Register metrics
+            for metric in to_register_metrics:
+                metrics.TAG_CHANGES_MONITOR.labels(
+                    pod_id=self.pod_id,
+                    model_name=metadata['model_name'],
+                    pipeline_name=metadata['workflow_name'],
+                    tag_name=metric[0]
+                ).set(metric[1])
+
             data_hold_df = DataFrame(data_hold, index=[0])
             data_hold_melted = data_hold_df.melt(
                 id_vars='timestamp', var_name='variable', value_name='value')

scouter/metrics.py

@@ -0,0 +1,21 @@
+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"]
+
+LABORIOUS_DATA_WRITTEN_COUNT = Counter(
+    "scouter_laborious_data_written_count",
+    "Number of writings to the database table laborious_data",
+    CORE_LABELS,
+)
+
+TAG_CHANGES_MONITOR = Gauge(
+    "scouter_tag_changes_monitor",
+    "Current value change of each tag",
+    [*CORE_LABELS, "tag_name"],
+)

scouter/worker/worker.py

@@ -12,19 +12,26 @@ with workflow.unsafe.imports_passed_through():
     from scouter.workflow.fake_data import FakeData
     from scouter.activities.faker import Faker
     import asyncio
+    from prometheus_client import start_http_server
+    from scouter import metrics
     from scouter.utils.connectors_config import (
         build_postgres_config,
         build_redis_config,
         build_mongodb_config
     )
 
+POD_ID = os.getenv("HOSTNAME", "localhost")
+
 
 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...')
 
@@ -77,7 +84,8 @@ async def main():
                 activities.aggregate_data,
                 activities.group_and_hold_data,
                 activities.export_data_to_postgres,
-                activities.store_data_package
+                activities.write_metrics,
+                activities.store_data_package,
             ],
             max_concurrent_workflow_tasks=100,
             max_concurrent_activities=100,
@@ -112,7 +120,20 @@ async def main():
         if activities:
             activities.shutdown()
         # Exit with a non-zero status code to indicate failure to Kubernetes
+        metrics.APP_UP.labels(pod_id=POD_ID).set(0)  # Mark app as DOWN
         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())

scouter/workflow/sub_workflows/core_scouter.py

@@ -86,6 +86,15 @@ class CoreScouter:
             start_to_close_timeout=timedelta(seconds=60)
         )
 
+        await workflow.execute_activity_method(
+            Activities.write_metrics,
+            {
+                **metadata,
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
         if input_data.get('debug_data_package', False):
             await workflow.execute_activity_method(
                 Activities.store_data_package,

tests/activities/test_gates.py

@@ -402,3 +402,18 @@ async def test_aggregate_data_raise_exception(gates_fixture):
         )
     else:
         assert False
+
+
+@pytest.mark.asyncio
+@patch('scouter.activities.gates.metrics')
+async def test_write_metrics(mock_metrics, gates_fixture):
+    """Test write_metrics method."""
+    input_data = {
+        'metadata': metadata['metadata']
+    }
+    await gates_fixture.write_metrics(input_data)
+    mock_metrics.LABORIOUS_DATA_WRITTEN_COUNT.labels.assert_called_once_with(
+        pod_id=gates_fixture.pod_id,
+        model_name=metadata['metadata']['model_name'],
+        pipeline_name=metadata['metadata']['workflow_name']
+    )

tests/activities/test_redis.py

@@ -20,6 +20,7 @@ def redis_activity(_mock_redis_init):
     activity.redis_client = MagicMock()
     activity.logger = logger
     activity.notification_handler = notification_handler
+    activity.pod_id = 'test_pod_id'
     return activity
 
 

tests/test_metrics.py

@@ -0,0 +1,25 @@
+# tests/unit/test_metrics.py
+
+import pytest
+from prometheus_client import Counter, Gauge, Histogram
+import scouter.metrics as metrics
+
+# --- Test Functions for Each Metric (Corrected for v0.22.0 _name behavior) ---
+
+
+def test_scouter_laborious_data_written_count():
+    """Verify the definition of LABORIOUS_DATA_WRITTEN_COUNT."""
+    assert metrics.LABORIOUS_DATA_WRITTEN_COUNT is not None
+    assert isinstance(metrics.LABORIOUS_DATA_WRITTEN_COUNT, Counter)
+    assert metrics.LABORIOUS_DATA_WRITTEN_COUNT._name == "scouter_laborious_data_written_count"
+    assert set(metrics.LABORIOUS_DATA_WRITTEN_COUNT._labelnames) == {
+        "pod_id", "model_name", "pipeline_name"}
+
+
+def test_scouter_tag_changes_monitor():
+    """Verify the definition of TAG_CHANGES_MONITOR."""
+    assert metrics.TAG_CHANGES_MONITOR is not None
+    assert isinstance(metrics.TAG_CHANGES_MONITOR, Gauge)
+    assert metrics.TAG_CHANGES_MONITOR._name == "scouter_tag_changes_monitor"
+    assert set(metrics.TAG_CHANGES_MONITOR._labelnames) == {
+        "pod_id", "model_name", "pipeline_name", "tag_name"}

values.yaml

@@ -3,7 +3,7 @@
 # Declare variables to be passed into your templates.
 
 # This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/
-replicaCount: 5
+replicaCount: 3
 
 # This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
 image:
@@ -11,7 +11,7 @@ image:
   # This sets the pull policy for images.
   pullPolicy: Always
   # Overrides the image tag whose default is the chart appVersion.
-  tag: "0.2.7"
+  tag: "0.3.2"
 
 # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
 imagePullSecrets:
@@ -111,14 +111,39 @@ tolerations: []
 
 affinity: {}
 
-service: {}
+services:
+  metrics:
+    enabled: true
+    type: ClusterIP
+    port: 9090
+    targetPort: 9090
+    name: metrics
+
+# Configuração do ServiceMonitor para o Prometheus Operator
+# ref: https://github.com/prometheus-operator/prometheus-operator
+serviceMonitor:
+  # Se true, um recurso ServiceMonitor será criado.
+  enabled: true
+  # O intervalo no qual as métricas devem ser coletadas (ex: 30s, 1m).
+  interval: 30s
+  # O path do endpoint de métricas na sua aplicação.
+  path: /metrics
+  # Labels adicionais para o recurso ServiceMonitor.
+  # Essencial para que o Prometheus Operator o descubra. Se você usa o helm chart kube-prometheus-stack,
+  # ele procura por ServiceMonitors com o label "release: kube-prometheus-stack".
+  additionalLabels:
+    release: kube-prometheus-stack
+  # Configurações de relabeling adicionais, se necessário.
+  # ref: https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config
+  relabelings: []
+  port: metrics
 
 env:
   # Entrypoint variables 
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1172-criar-pipeline-de-alertas-orquestrador"
+    value: "SIENTIAPDE-1174-mapear-e-implementar-metricas-a-serem-criadas"
   - name: PYTHON_APP
     value: "scouter.worker.worker"
 
@@ -160,6 +185,8 @@ env:
 
   - name: LOG_LEVEL
     value: "DEBUG"
+  - name: HTTP_METRICS_PORT
+    value: "9090"
   - name: PROJECT_NAME
     value: "sientia-scouter"
 
@@ -190,7 +217,7 @@ ssh:
 
 # kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp
 
-# helm upgrade --install sientia-scouter-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.4.0-uat
+# helm upgrade --install sientia-scouter-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.4.0
 
 # kubectl create secret generic git-ssh-key-sientia-scouter-worker \
 #   --namespace sientia \