diff --git a/encrypt.py b/encrypt.py new file mode 100644 index 0000000..ad990a6 --- /dev/null +++ b/encrypt.py @@ -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() diff --git a/scouter/activities/gates.py b/scouter/activities/gates.py index b5da7dd..ca9e2ce 100644 --- a/scouter/activities/gates.py +++ b/scouter/activities/gates.py @@ -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) diff --git a/scouter/activities/redis.py b/scouter/activities/redis.py index 7619857..9c76fe4 100644 --- a/scouter/activities/redis.py +++ b/scouter/activities/redis.py @@ -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' diff --git a/scouter/workflow/scouter.py b/scouter/workflow/scouter.py index e028bc6..5888627 100644 --- a/scouter/workflow/scouter.py +++ b/scouter/workflow/scouter.py @@ -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) diff --git a/scouter/workflow/sub_workflows/core_scouter.py b/scouter/workflow/sub_workflows/core_scouter.py index 85634cf..1bbc4cb 100644 --- a/scouter/workflow/sub_workflows/core_scouter.py +++ b/scouter/workflow/sub_workflows/core_scouter.py @@ -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), diff --git a/tests/activities/test_gates.py b/tests/activities/test_gates.py index 7f0f026..c916ace 100644 --- a/tests/activities/test_gates.py +++ b/tests/activities/test_gates.py @@ -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, + ) diff --git a/tests/workflow/test_scouter.py b/tests/workflow/test_scouter.py index 3e16666..ba47683 100644 --- a/tests/workflow/test_scouter.py +++ b/tests/workflow/test_scouter.py @@ -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', diff --git a/values.yaml b/values.yaml index a44982e..7371af9 100644 --- a/values.yaml +++ b/values.yaml @@ -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"