From 7bb24b42f615d09482fe9c94be8e2b988a98f2c1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 1 Aug 2025 12:01:03 -0300 Subject: [PATCH 1/6] SIENTIAPDE-1174 feat: update dependencies and enhance metrics integration - Updated sientia-dataops-library dependency version from 1.3.5 to 1.3.7 in requirements.txt. - Incremented image tag in values.yaml from 0.2.7 to 0.3.1. - Added Prometheus metrics service configuration in values.yaml, enabling metrics collection. - Implemented Prometheus client in worker.py to start a metrics server and track application health. --- orchestrator/metrics.py | 16 ++++++++++++++ orchestrator/worker/worker.py | 22 ++++++++++++++++++- requirements.txt | 3 ++- values.yaml | 40 +++++++++++++++++++++++------------ 4 files changed, 66 insertions(+), 15 deletions(-) create mode 100644 orchestrator/metrics.py diff --git a/orchestrator/metrics.py b/orchestrator/metrics.py new file mode 100644 index 0000000..c7bc20e --- /dev/null +++ b/orchestrator/metrics.py @@ -0,0 +1,16 @@ +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"] + + +EMAIL_SENT_COUNT = Counter( + "email_sent_count", + "Number of emails sent", + [*CORE_LABELS, "email_group"], +) diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index fcd815d..89959e3 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -22,6 +22,10 @@ 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 prometheus_client import start_http_server + from orchestrator import metrics + +POD_ID = os.getenv("POD_ID") async def main(): @@ -29,7 +33,10 @@ async def main(): namespace = os.getenv('TEMPORAL_NAMESPACE', 'default') 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...') @@ -164,7 +171,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) + except Exception as e: + print(f"Failed to start Prometheus server: {e}") + os._exit(1) + + if __name__ == '__main__': asyncio.run(main()) diff --git a/requirements.txt b/requirements.txt index 166f96d..f7a80cf 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,4 +5,5 @@ redis couchbase pymongo jinja2 -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 +prometheus-client diff --git a/values.yaml b/values.yaml index 570ae29..802fba4 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.1" # 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: @@ -112,19 +112,31 @@ tolerations: [] affinity: {} services: - api: - enabled: false + metrics: + enabled: true type: ClusterIP - port: 4841 - targetPort: 4841 - name: api + port: 9090 + targetPort: 9090 + name: metrics - opc: - enabled: false - type: ClusterIP - port: 4840 - targetPort: 4840 - name: server +# 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: @@ -132,7 +144,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_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: "orchestrator.worker.worker" @@ -203,6 +215,8 @@ env: - name: LOG_LEVEL value: "DEBUG" + - name: HTTP_METRICS_PORT + value: "9090" - name: PROJECT_NAME value: "sientia-orchestrator" From ae0fa2fc8afc358c5ad86b51ec88475ced12eab9 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 1 Aug 2025 12:27:59 -0300 Subject: [PATCH 2/6] SIENTIAPDE-1174 feat: integrate metrics tracking for sent emails in Email activity - Added metrics tracking for the number of emails sent, incorporating pod ID, model name, pipeline name, and email group for better monitoring. - Enhanced logging to provide detailed information on email sending status for each receiver group. --- orchestrator/activities/email.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/orchestrator/activities/email.py b/orchestrator/activities/email.py index b59ff91..e47ec1e 100644 --- a/orchestrator/activities/email.py +++ b/orchestrator/activities/email.py @@ -2,6 +2,8 @@ from smtplib import SMTPServerDisconnected from temporalio import workflow, activity +from orchestrator import metrics + with workflow.unsafe.imports_passed_through(): import traceback import smtplib @@ -147,7 +149,6 @@ class Email(BaseActivity): for group_name, group_config in receiver_groups.items(): try: - receivers = ", ".join(group_config['members']) self.info(f"Sending email to {group_name}: {receivers}", @@ -177,6 +178,12 @@ class Email(BaseActivity): group_config['status'] = 'failed' else: group_config['status'] = 'sent' + metrics.EMAIL_SENT_COUNT.labels( + pod_id=self.pod_id, + model_name=metadata['model_name'], + pipeline_name=metadata['workflow_name'], + email_group=group_name + ).inc() self.info(f"Email sent to {group_name}: {receivers}", metadata=metadata) From 6c1b7c4bf7b6a95e0b466165279155ae52c27e19 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 1 Aug 2025 12:41:40 -0300 Subject: [PATCH 3/6] SIENTIAPDE-1174 feat: enhance notification filtering in Formatters and SlotManager classes - Added logic to remove receiver groups with no notifications, improving the efficiency of notification handling. - Implemented early return in Alerts and Reports workflows when there are no notifications or receiver groups, streamlining the processing flow. --- orchestrator/activities/formatters.py | 7 +++++++ orchestrator/activities/slot_manager.py | 7 +++++++ orchestrator/workflows/alerts.py | 3 +++ orchestrator/workflows/reports.py | 3 +++ 4 files changed, 20 insertions(+) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 1ebcc6b..04a983d 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -574,4 +574,11 @@ class Formatters(BaseActivity): notification) already_added_keys.append(key) + # Remove groups with no notifications + receiver_groups = { + group_name: group + for group_name, group in receiver_groups.items() + if group['notifications'] + } + return receiver_groups diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 076a1a8..18e96d0 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -318,6 +318,13 @@ class SlotManager(Redis): notification) already_added_keys.append(key) + # Remove groups with no notifications + receiver_groups = { + group_name: group + for group_name, group in receiver_groups.items() + if group['notifications'] + } + return receiver_groups @activity.defn(name="store_notification_cache") diff --git a/orchestrator/workflows/alerts.py b/orchestrator/workflows/alerts.py index 6fbe231..1a8509a 100644 --- a/orchestrator/workflows/alerts.py +++ b/orchestrator/workflows/alerts.py @@ -82,6 +82,9 @@ class Alerts: } ) + if not log_report: + return + # Store the notification_id sendings to avoid sending them again await workflow.execute_activity_method( Activities.store_notification_cache, diff --git a/orchestrator/workflows/reports.py b/orchestrator/workflows/reports.py index 75e2851..eb00231 100644 --- a/orchestrator/workflows/reports.py +++ b/orchestrator/workflows/reports.py @@ -65,6 +65,9 @@ class Reports: retry_policy=retry_policy ) + if not receiver_groups: + return + # Call subworkflow "process_notifications" passing the notification package await workflow.execute_child_workflow( 'process_notifications', From 4f4bc9e7ec5a208486f9f0cb44107c35c1da633d Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 4 Aug 2025 10:01:14 -0300 Subject: [PATCH 4/6] SIENTIAPDE-1174 chore: update helm upgrade command in values.yaml to reflect version change - Modified the helm upgrade command in values.yaml to use version 0.4.0 instead of 0.4.0-uat for consistency in deployment. --- values.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/values.yaml b/values.yaml index 802fba4..b1175cf 100644 --- a/values.yaml +++ b/values.yaml @@ -237,7 +237,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-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.4.0-uat +# helm upgrade --install sientia-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.4.0 # kubectl create secret generic git-ssh-key-sientia-orchestrator-worker \ # --namespace sientia \ From 68d4998ce762b230ce0956739b82099d2c000627 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 4 Aug 2025 15:27:40 -0300 Subject: [PATCH 5/6] SIENTIAPDE-1174 chore: increment image tag in values.yaml from 0.3.1 to 0.3.2 for version update --- values.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/values.yaml b/values.yaml index b1175cf..1ae2b9b 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.3.1" + 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: From afaf0173400c90f589746d3624f7d3d3b5617c58 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 5 Aug 2025 13:51:34 -0300 Subject: [PATCH 6/6] SIENTIAPDE-1174 SIENTIAPDE-1174 chore: update sientia-dataops-library dependency version from 1.3.7 to 1.3.8 in requirements.txt - Removed unused test group from test_formatters.py. - Added new test case for handling scenarios with no log reports in test_alerts.py. - Added new test case for handling scenarios with no groups in test_reports.py. --- requirements.txt | 2 +- .../activities/test_formatters.py | 5 ----- tests/orchestrator/workflows/test_alerts.py | 21 ++++++++++++++++++- tests/orchestrator/workflows/test_reports.py | 16 ++++++++++++++ 4 files changed, 37 insertions(+), 7 deletions(-) diff --git a/requirements.txt b/requirements.txt index f7a80cf..53ef9e2 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,5 +5,5 @@ redis couchbase pymongo jinja2 -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.7 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.8 prometheus-client diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index ef89e9b..f3c1397 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -816,10 +816,5 @@ async def test_filter_notification_reports(formatters): 'notification_id': 'test_notification_id_3' } ] - }, - 'test_group_2': { - 'group_name': 'test_group_2', - 'contents': ['core_alerts'], - 'notifications': [] } } diff --git a/tests/orchestrator/workflows/test_alerts.py b/tests/orchestrator/workflows/test_alerts.py index 0dfffd3..56c1406 100644 --- a/tests/orchestrator/workflows/test_alerts.py +++ b/tests/orchestrator/workflows/test_alerts.py @@ -1,4 +1,4 @@ -from unittest.mock import AsyncMock, patch, ANY, call +from unittest.mock import AsyncMock, MagicMock, patch, ANY, call from pytest import fixture, mark from orchestrator.workflows.alerts import Alerts from orchestrator.activities.activities import Activities @@ -115,3 +115,22 @@ async def test_run_no_data(workflow_mock, alerts): ]) workflow_mock.execute_local_activity_method.assert_not_called() + + +@mark.asyncio +@patch("orchestrator.workflows.alerts.workflow", new_callable=AsyncMock) +async def test_run_no_log_report(workflow_mock, alerts): + workflow_mock.execute_child_workflow.side_effect = [ + MagicMock(), + [] + ] + + input_data = { + 'schedule_name': 'test-schedule-name', + 'notification_ttl': 300, + 'sent_ttl': 600 + } + + await alerts.run(input_data) + + workflow_mock.execute_activity_method.assert_not_called() diff --git a/tests/orchestrator/workflows/test_reports.py b/tests/orchestrator/workflows/test_reports.py index 1b1983d..d0134d8 100644 --- a/tests/orchestrator/workflows/test_reports.py +++ b/tests/orchestrator/workflows/test_reports.py @@ -97,3 +97,19 @@ async def test_run_no_data(workflow_mock, reports): ]) workflow_mock.execute_local_activity_method.assert_not_called() + + +@mark.asyncio +@patch("orchestrator.workflows.reports.workflow", new_callable=AsyncMock) +async def test_run_no_groups(workflow_mock, reports): + workflow_mock.execute_local_activity_method.return_value = [] + + input_data = { + 'schedule_name': 'test-schedule-name', + 'notification_ttl': 300, + 'sent_ttl': 600 + } + + await reports.run(input_data) + + workflow_mock.execute_child_workflow.assert_called_once()