gitea_admin/sientia-dataops-scouter_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #23 from Aignosi/fix/SIENTIAPDE-1314-ajustes-nas-camadas-de-monitoramento-do-sientia

Browse Source 
Implement tag value registration in Gates activity for improved metrics tracking
This commit is contained in:
Matheus Demoner
2025-10-27 14:28:51 -03:00
committed by GitHub
parent c9136ca537 5dc2971d8c
commit 7265819ae0
8 changed files with 163 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
encrypt.py Normal file
View File

@@ -0,0 +1,113 @@
import os
import argparse
from pathspec import PathSpec
import yaml # type: ignore
from typing import Any
'''
Usage:
python .\encrypt.py path_to_dir output_file --ignore ignore_file --chunk-size 100000
'''
def load_ignore_patterns(ignore_file, include_library):
# Ensure the .gitignore file exists
if not os.path.exists(ignore_file):
raise FileNotFoundError(f"Ignore file not found at {ignore_file}")
# Load and parse the .gitignore patterns
with open(ignore_file, 'r') as file:
patterns = file.readlines()
if not include_library:
patterns.append('**/deploy/library/')
spec = PathSpec.from_lines('gitwildmatch', patterns)
return spec
def is_ignored(file_path, spec):
"""Check if a file should be ignored based on the ignore patterns."""
return spec.match_file(file_path) if spec else False
def encode_file_tree_to_yaml(directory, ignore_file, include_library):
"""Encode the file tree into a single YAML file."""
ignore_patterns = load_ignore_patterns(
ignore_file, include_library) if ignore_file else None
file_tree: dict[str, Any] = {}
for root, dirs, files in os.walk(directory):
# Skip ignored directories
dirs[:] = [d for d in dirs if not is_ignored(
os.path.join(root, d), ignore_patterns)]
for file in files:
file_path = os.path.join(root, file)
# Skip ignored files
if is_ignored(file_path, ignore_patterns):
continue
# Read file content
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
except Exception as e:
print(f"Error reading file {file_path}: {e}")
raise
# Create nested dictionary structure
path_parts = os.path.relpath(file_path, directory).split(os.sep)
current_level = file_tree
# all except the last part (the file name)
for part in path_parts[:-1]:
current_level = current_level.setdefault(part, {})
# Add the file and its content
current_level[path_parts[-1]] = content
return yaml.dump(file_tree, default_flow_style=False)
def chunk_and_write_file_tree_to_yaml(yaml_content, output_file, chunk_size=None):
"""Chunk the YAML content and write it to the output file."""
chunks = [yaml_content] if chunk_size is None else [
yaml_content[i:i + chunk_size] for i in range(0, len(yaml_content), chunk_size)]
for i, chunk in enumerate(chunks):
chunk_file = f"{output_file}_{i}.yaml"
# Write the file tree to the output YAML file
with open(chunk_file, 'w', encoding='utf-8') as yaml_file:
yaml_file.write(chunk)
def main():
parser = argparse.ArgumentParser(
description="Encrypts file tree to yaml file")
parser.add_argument("input_directory", help="Directory to encode")
parser.add_argument("output_yaml_file", help="Output YAML file")
parser.add_argument("--ignore", default=None,
help="Path to the ignore file")
parser.add_argument("--chunk-size", type=int, default=None,
help="Chunk size for the output YAML file")
parser.add_argument("--library", type=bool, default=False,
help="Incude the library in the output YAML file")
# Parse arguments
args = parser.parse_args()
# Example usage
directory_to_encode = args.input_directory
ignore_file_path = args.ignore
output_yaml_file = args.output_yaml_file
include_library = args.library
content = encode_file_tree_to_yaml(
directory_to_encode, ignore_file_path, include_library)
chunk_and_write_file_tree_to_yaml(
content, output_yaml_file, args.chunk_size)
if __name__ == "__main__":
main()

10
scouter/activities/gates.py
View File

@@ -284,6 +284,7 @@ class Gates(BaseActivity):
metadata: dict[str, Any]
"""
metadata = input_data['metadata']
tag_values = DataFrame(input_data['tag_values'])
self.info(f'Writing metrics for {metadata["model_name"]}', metadata=metadata)
@@ -293,4 +294,13 @@ class Gates(BaseActivity):
pipeline_name=metadata['workflow_name'],
).inc()
# Register metrics
for _, row in tag_values.iterrows():
metrics.TAG_CHANGES_MONITOR.labels(
pod_id=self.pod_id,
model_name=metadata['model_name'],
pipeline_name=metadata['workflow_name'],
tag_name=row['variable'],
).set(row['value'])
self.info(f'Metrics written for {metadata["model_name"]}', metadata=metadata)

14
scouter/activities/redis.py
View File

@@ -13,8 +13,6 @@ with workflow.unsafe.imports_passed_through():
from sientia_do.temporal.activities.redis_base import Redis as RedisBase
from sientia_do.temporal.constants import DATETIME_FORMAT, now
from scouter import metrics
class Redis(RedisBase):
"""
@@ -220,12 +218,10 @@ class Redis(RedisBase):
data_hold = {tag: content for tag, content in data_hold.items() if tag in tags}
self.debug(f'Data hold after removing removed tags: {data_hold}', metadata=metadata)
to_register_metrics = []
for _, row in data.iterrows():
value = row['value']
data_hold[row['name']] = value
to_register_metrics.append((row['name'], value))
if fill_missing_tags:
self.debug('Filling missing tags in data package', metadata=metadata)
@@ -240,16 +236,6 @@ class Redis(RedisBase):
self.set(key, data_hold, ttl=retention_time)
# Register metrics
self.debug(f'Metrics to register: {to_register_metrics}', metadata=metadata)
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'

2
scouter/workflow/scouter.py
View File

@@ -113,4 +113,4 @@ class Scouter:
input_data['data'] = data
input_data['metadata'] = metadata
await workflow.execute_child_workflow('core_scouter', input_data)
await workflow.execute_child_workflow('subworkflow.core_scouter', input_data)

3
scouter/workflow/sub_workflows/core_scouter.py
View File

@@ -10,7 +10,7 @@ with workflow.unsafe.imports_passed_through():
from scouter.activities.activities import Activities
@workflow.defn(name='core_scouter')
@workflow.defn(name='subworkflow.core_scouter')
class CoreScouter:
"""
Core data processing workflow that handles data quality, aggregation, and export.
@@ -119,6 +119,7 @@ class CoreScouter:
Activities.write_metrics,
{
**metadata,
'tag_values': held_data,
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60),

37
tests/activities/test_gates.py
View File

@@ -1,5 +1,5 @@
from typing import Any
from unittest.mock import ANY, MagicMock, Mock, patch
from unittest.mock import ANY, MagicMock, Mock, call, patch
import numpy as np
import pandas as pd
@@ -359,10 +359,43 @@ async def test_aggregate_data_raise_exception(gates_fixture):
@patch('scouter.activities.gates.metrics')
async def test_write_metrics(mock_metrics, gates_fixture):
"""Test write_metrics method."""
input_data = {'metadata': metadata['metadata']}
input_data = {
'metadata': metadata['metadata'],
'tag_values': {
'variable': ['tag1', 'tag2'],
'value': [1.0, 2.0],
},
}
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'],
)
mock_metrics.LABORIOUS_DATA_WRITTEN_COUNT.labels.return_value.inc.assert_called_once()
mock_metrics.TAG_CHANGES_MONITOR.labels.return_value.set.assert_has_calls(
[
call(1.0),
call(2.0),
],
any_order=True,
)
mock_metrics.TAG_CHANGES_MONITOR.labels.assert_has_calls(
[
call(
pod_id=gates_fixture.pod_id,
model_name=metadata['metadata']['model_name'],
pipeline_name=metadata['metadata']['workflow_name'],
tag_name='tag1',
),
call(
pod_id=gates_fixture.pod_id,
model_name=metadata['metadata']['model_name'],
pipeline_name=metadata['metadata']['workflow_name'],
tag_name='tag2',
),
],
any_order=True,
)

2
tests/workflow/test_scouter.py
View File

@@ -75,7 +75,7 @@ async def test_scouter_workflow(mock_workflow, scouter):
)
mock_workflow.execute_child_workflow.assert_called_once_with(
'core_scouter',
'subworkflow.core_scouter',
{
'metadata': expected_metadata,
'topic': 'test_topic',

2
values.yaml
View File

@@ -150,7 +150,7 @@ env:
- name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git"
- name: GITHUB_BRANCH
value: "SIENTIAPDE-1312-melhorias-e-correcoes-nas-pipelines-de-dados"
value: "fix/SIENTIAPDE-1314-ajustes-nas-camadas-de-monitoramento-do-sientia"
- name: PYTHON_APP
value: "scouter.worker.worker"