diff --git a/requirements.txt b/requirements.txt index 6feb3ea..1b3009f 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.7 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/activities.py b/scouter/activities/activities.py index 775f958..9ada6d3 100644 --- a/scouter/activities/activities.py +++ b/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) diff --git a/scouter/activities/gates.py b/scouter/activities/gates.py index ea9d1bf..c7840f4 100644 --- a/scouter/activities/gates.py +++ b/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() 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..bffc478 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=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') diff --git a/scouter/metrics.py b/scouter/metrics.py new file mode 100644 index 0000000..a3114be --- /dev/null +++ b/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"], +) diff --git a/scouter/worker/worker.py b/scouter/worker/worker.py index 239ecd8..aff8341 100644 --- a/scouter/worker/worker.py +++ b/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()) diff --git a/scouter/workflow/sub_workflows/core_scouter.py b/scouter/workflow/sub_workflows/core_scouter.py index 21a6120..b9f4a35 100644 --- a/scouter/workflow/sub_workflows/core_scouter.py +++ b/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, diff --git a/tests/activities/test_gates.py b/tests/activities/test_gates.py index dec6b9d..f0057fd 100644 --- a/tests/activities/test_gates.py +++ b/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'] + ) diff --git a/tests/activities/test_redis.py b/tests/activities/test_redis.py index ce3215e..bed4c10 100644 --- a/tests/activities/test_redis.py +++ b/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 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..a729358 100644 --- a/values.yaml +++ b/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 \