diff --git a/requirements.txt b/requirements.txt index 6feb3ea..1affea9 100644 --- a/requirements.txt +++ b/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.6 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.5 -pydruid[pandas] \ No newline at end of file +pydruid[pandas] +prometheus-client \ No newline at end of file diff --git a/scouter/activities/gates.py b/scouter/activities/gates.py index ea9d1bf..3c68aaf 100644 --- a/scouter/activities/gates.py +++ b/scouter/activities/gates.py @@ -3,6 +3,8 @@ from temporalio import workflow, activity 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 +18,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: """ diff --git a/scouter/activities/mongodb.py b/scouter/activities/mongodb.py index c4c6fe4..3a80436 100644 --- a/scouter/activities/mongodb.py +++ b/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): """ diff --git a/scouter/activities/redis.py b/scouter/activities/redis.py index 5174d87..33d3170 100644 --- a/scouter/activities/redis.py +++ b/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=metadata['pod_id'], + model_name=input_data['model_name'], + pipeline_name=input_data['schedule_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') diff --git a/scouter/metrics.py b/scouter/metrics.py new file mode 100644 index 0000000..3800279 --- /dev/null +++ b/scouter/metrics.py @@ -0,0 +1,15 @@ +from prometheus_client import Gauge, Counter + +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"], +) diff --git a/scouter/workflow/scouter.py b/scouter/workflow/scouter.py index ffcb625..c2afbff 100644 --- a/scouter/workflow/scouter.py +++ b/scouter/workflow/scouter.py @@ -5,6 +5,7 @@ with workflow.unsafe.imports_passed_through(): from typing import Any from datetime import timedelta from sientia_do.temporal.utils.policies import retry_policy + from os import getenv @workflow.defn(name="scouter") @@ -37,7 +38,8 @@ class Scouter: 'model_id': input_data['model_id'], 'model_name': input_data['model_name'], 'schedule_name': input_data['schedule_name'], - 'workflow_name': input_data['workflow_name'] + 'workflow_name': input_data['workflow_name'], + 'pod_id': getenv("HOSTNAME", "localhost") } } diff --git a/scouter/workflow/sub_workflows/core_scouter.py b/scouter/workflow/sub_workflows/core_scouter.py index 21a6120..c71f10b 100644 --- a/scouter/workflow/sub_workflows/core_scouter.py +++ b/scouter/workflow/sub_workflows/core_scouter.py @@ -1,5 +1,7 @@ from temporalio import workflow +from scouter import metrics + with workflow.unsafe.imports_passed_through(): from scouter.activities.activities import Activities from typing import Any @@ -86,6 +88,12 @@ class CoreScouter: start_to_close_timeout=timedelta(seconds=60) ) + metrics.LABORIOUS_DATA_WRITTEN_COUNT.labels( + pod_id=metadata['pod_id'], + model_name=input_data['model_name'], + pipeline_name=input_data['workflow_name'] + ).inc() + if input_data.get('debug_data_package', False): await workflow.execute_activity_method( Activities.store_data_package, diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..875113c --- /dev/null +++ b/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"} diff --git a/values.yaml b/values.yaml index 3e73d24..d7fb472 100644 --- a/values.yaml +++ b/values.yaml @@ -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.0" # 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,20 @@ tolerations: [] affinity: {} -service: {} +services: + metrics: + enabled: true + type: ClusterIP + port: 9090 + targetPort: 9090 + name: 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 +166,8 @@ env: - name: LOG_LEVEL value: "DEBUG" + - name: HTTP_METRICS_PORT + value: "9090" - name: PROJECT_NAME value: "sientia-scouter"