From f3833a0f98325c99fad5beab0e3261de922c13c9 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Dec 2025 09:48:28 -0300 Subject: [PATCH 01/11] SIENTIAPDE-1441 Remove encrypt.py and test.ipynb files; update values.yaml to set image tag to 1.0.0 and change GITHUB_BRANCH to release/SIENTIAPDE-1441 for version consistency. --- encrypt.py | 113 ------ git-requirements-mapping.txt | 1 + test.ipynb | 724 ----------------------------------- values.yaml | 10 +- 4 files changed, 5 insertions(+), 843 deletions(-) delete mode 100644 encrypt.py create mode 100644 git-requirements-mapping.txt delete mode 100644 test.ipynb diff --git a/encrypt.py b/encrypt.py deleted file mode 100644 index ad990a6..0000000 --- a/encrypt.py +++ /dev/null @@ -1,113 +0,0 @@ -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/git-requirements-mapping.txt b/git-requirements-mapping.txt new file mode 100644 index 0000000..1362493 --- /dev/null +++ b/git-requirements-mapping.txt @@ -0,0 +1 @@ +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git:sientia-do \ No newline at end of file diff --git a/test.ipynb b/test.ipynb deleted file mode 100644 index 7e5c2a0..0000000 --- a/test.ipynb +++ /dev/null @@ -1,724 +0,0 @@ -{ - "cells": [ - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "from sqlalchemy.engine import create_engine\n", - "\n", - "engine = create_engine('druid://localhost:8082/druid/v2/sql/')" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "from sqlalchemy import MetaData, Table\n", - "\n", - "metadata = MetaData()\n", - "places = Table('raw_scouter-opcua-orchestrated-pipeline', metadata, autoload_with=engine)" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "from sqlalchemy import select\n", - "\n", - "stmt = select(places)\n", - "with engine.connect() as conn:\n", - " result = conn.execute(stmt)\n", - " for row in result:\n", - " print(row)" - ] - }, - { - "cell_type": "code", - "execution_count": 5, - "metadata": {}, - "outputs": [ - { - "name": "stderr", - "output_type": "stream", - "text": [ - "/tmp/ipykernel_30057/1086103244.py:5: SADeprecationWarning: The dbapi() classmethod on dialect classes has been renamed to import_dbapi(). Implement an import_dbapi() classmethod directly on class to remove this warning; the old .dbapi() classmethod may be maintained for backwards compatibility.\n", - " engine = create_engine('druid://localhost:8082/druid/v2/sql/')\n", - "/home/grezewave/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/pydruid/db/sqlalchemy.py:188: SAWarning: Dialect druid:rest will not make use of SQL compilation caching as it does not set the 'supports_statement_cache' attribute to ``True``. This can have significant performance implications including some performance degradations in comparison to prior SQLAlchemy versions. Dialect maintainers should seek to set this attribute to True after appropriate development and testing for SQLAlchemy 1.4 caching support. Alternatively, this attribute may be set to False which will disable this warning. (Background on this warning at: https://sqlalche.me/e/20/cprf)\n", - " result = connection.execute(text(query))\n" - ] - }, - { - "data": { - "text/html": [ - "
\n", - "\n", - "\n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - " \n", - "
namekafka.topictagvaluetimestampinserted_at
0Counterraw_scouter-opcua-orchestrated-pipelinens=2;i=2-50.1322025-07-02 13:05:192025-07-02 13:05:19.729000
1Rolloutraw_scouter-opcua-orchestrated-pipelinens=2;i=370.7672025-07-02 13:05:192025-07-02 13:05:19.731000
2Squareraw_scouter-opcua-orchestrated-pipelinens=2;i=4-58.4482025-07-02 13:05:192025-07-02 13:05:19.732000
3Counterraw_scouter-opcua-orchestrated-pipelinens=2;i=2-50.1262025-07-02 13:05:242025-07-02 13:05:24.728000
4Rolloutraw_scouter-opcua-orchestrated-pipelinens=2;i=369.1992025-07-02 13:05:242025-07-02 13:05:24.730000
.....................
373Rolloutraw_scouter-opcua-orchestrated-pipelinens=2;i=385.9212025-07-02 13:15:402025-07-02 13:15:40.230000
374Squareraw_scouter-opcua-orchestrated-pipelinens=2;i=4-69.2962025-07-02 13:15:402025-07-02 13:15:40.232000
375Counterraw_scouter-opcua-orchestrated-pipelinens=2;i=2-71.2072025-07-02 13:15:452025-07-02 13:15:45.228000
376Rolloutraw_scouter-opcua-orchestrated-pipelinens=2;i=384.6652025-07-02 13:15:452025-07-02 13:15:45.231000
377Squareraw_scouter-opcua-orchestrated-pipelinens=2;i=4-67.5922025-07-02 13:15:452025-07-02 13:15:45.233000
\n", - "

378 rows × 6 columns

\n", - "
" - ], - "text/plain": [ - " name kafka.topic tag value \\\n", - "0 Counter raw_scouter-opcua-orchestrated-pipeline ns=2;i=2 -50.132 \n", - "1 Rollout raw_scouter-opcua-orchestrated-pipeline ns=2;i=3 70.767 \n", - "2 Square raw_scouter-opcua-orchestrated-pipeline ns=2;i=4 -58.448 \n", - "3 Counter raw_scouter-opcua-orchestrated-pipeline ns=2;i=2 -50.126 \n", - "4 Rollout raw_scouter-opcua-orchestrated-pipeline ns=2;i=3 69.199 \n", - ".. ... ... ... ... \n", - "373 Rollout raw_scouter-opcua-orchestrated-pipeline ns=2;i=3 85.921 \n", - "374 Square raw_scouter-opcua-orchestrated-pipeline ns=2;i=4 -69.296 \n", - "375 Counter raw_scouter-opcua-orchestrated-pipeline ns=2;i=2 -71.207 \n", - "376 Rollout raw_scouter-opcua-orchestrated-pipeline ns=2;i=3 84.665 \n", - "377 Square raw_scouter-opcua-orchestrated-pipeline ns=2;i=4 -67.592 \n", - "\n", - " timestamp inserted_at \n", - "0 2025-07-02 13:05:19 2025-07-02 13:05:19.729000 \n", - "1 2025-07-02 13:05:19 2025-07-02 13:05:19.731000 \n", - "2 2025-07-02 13:05:19 2025-07-02 13:05:19.732000 \n", - "3 2025-07-02 13:05:24 2025-07-02 13:05:24.728000 \n", - "4 2025-07-02 13:05:24 2025-07-02 13:05:24.730000 \n", - ".. ... ... \n", - "373 2025-07-02 13:15:40 2025-07-02 13:15:40.230000 \n", - "374 2025-07-02 13:15:40 2025-07-02 13:15:40.232000 \n", - "375 2025-07-02 13:15:45 2025-07-02 13:15:45.228000 \n", - "376 2025-07-02 13:15:45 2025-07-02 13:15:45.231000 \n", - "377 2025-07-02 13:15:45 2025-07-02 13:15:45.233000 \n", - "\n", - "[378 rows x 6 columns]" - ] - }, - "metadata": {}, - "output_type": "display_data" - } - ], - "source": [ - "from sqlalchemy import create_engine, MetaData, Table, select, func, text\n", - "import pandas as pd\n", - "from datetime import datetime\n", - "\n", - "engine = create_engine('druid://localhost:8082/druid/v2/sql/')\n", - "metadata = MetaData()\n", - "places = Table('raw_scouter-opcua-orchestrated-pipeline', metadata, autoload_with=engine)\n", - "date_str = '2025-01-01'\n", - "stmt = select(places).where(text(f'\"__time\" > TIMESTAMP \\'{date_str}\\''))\n", - "\n", - "result = pd.read_sql(stmt, engine)\n", - "\n", - "result[\"inserted_at\"] = pd.to_datetime(result[\"__time\"]).dt.strftime(\n", - " \"%Y-%m-%d %H:%M:%S.%f\")\n", - "\n", - "result.drop(columns=[\"__time\"], inplace=True)\n", - "\n", - "display(result)" - ] - }, - { - "cell_type": "code", - "execution_count": 1, - "metadata": {}, - "outputs": [], - "source": [ - "data = {\n", - " \"id\": \"1\",\n", - " \"schedule_name\": \"scouter-opcua-orchestrated-pipeline\",\n", - " \"model_id\": \"1\",\n", - " \"workflow_type\": \"scouter\",\n", - " \"frequency\": \"30s\",\n", - " \"max_retry_policy\": 1,\n", - " \"read_tags\": [\n", - " {\n", - " \"tag_name\": \"Counter\",\n", - " \"server_id\": \"1\",\n", - " \"aggr_func\": \"avg\",\n", - " \"tag_address\": \"ns=2;i=2\",\n", - " \"frequency\": \"15000\",\n", - " \"data_range\": [\n", - " -100,\n", - " 100\n", - " ]\n", - " },\n", - " {\n", - " \"tag_name\": \"Rollout\",\n", - " \"server_id\": \"1\",\n", - " \"aggr_func\": \"mdn\",\n", - " \"tag_address\": \"ns=2;i=3\",\n", - " \"frequency\": \"15000\",\n", - " \"data_range\": [\n", - " -100,\n", - " 100\n", - " ]\n", - " },\n", - " {\n", - " \"tag_name\": \"Square\",\n", - " \"server_id\": \"1\",\n", - " \"aggr_func\": \"lts\",\n", - " \"tag_address\": \"ns=2;i=4\",\n", - " \"frequency\": \"15000\",\n", - " \"data_range\": [\n", - " -100,\n", - " 100\n", - " ]\n", - " }\n", - " ],\n", - " \"filters\": [\n", - " {\n", - " \"filter_name\": \"OUT_OF_BOUNDS_FILTER\",\n", - " \"policy\": \"DISCARD\"\n", - " },\n", - " {\n", - " \"filter_name\": \"NULL_VALUES_FILTER\",\n", - " \"policy\": \"DISCARD\"\n", - " }\n", - " ],\n", - " \"tag_retention_minutes\": 60\n", - "}" - ] - }, - { - "cell_type": "code", - "execution_count": 2, - "metadata": {}, - "outputs": [], - "source": [ - "import json\n", - "\n", - "\n", - "# Generate 29 more, changing only id and schedule_name\n", - "json_list = []\n", - "for i in range(30):\n", - " obj = data.copy()\n", - " obj['id'] = i + 1 # or any other unique id logic\n", - " obj['schedule_name'] = f\"scouter-opcua-pipeline-{i+1}\"\n", - " json_list.append(obj)\n", - "\n", - "# Save to a new file\n", - "with open('input_samples_30.json', 'w') as f:\n", - " json.dump(json_list, f, indent=2)" - ] - }, - { - "cell_type": "code", - "execution_count": 9, - "metadata": {}, - "outputs": [], - "source": [ - "spec = {\n", - " \"type\": \"kafka\",\n", - " \"spec\": {\n", - " \"dataSchema\": {\n", - " \"dataSource\": \"raw_scouter-opcua-orchestrated-pipeline\",\n", - " \"timestampSpec\": {\n", - " \"column\": \"kafka.timestamp\",\n", - " \"format\": \"millis\",\n", - " \"missingValue\": None\n", - " },\n", - " \"dimensionsSpec\": {\n", - " \"dimensions\": [],\n", - " \"dimensionExclusions\": [\n", - " \"__time\",\n", - " \"kafka.timestamp\"\n", - " ],\n", - " \"includeAllDimensions\": False,\n", - " \"useSchemaDiscovery\": True\n", - " },\n", - " \"metricsSpec\": [],\n", - " \"granularitySpec\": {\n", - " \"type\": \"uniform\",\n", - " \"segmentGranularity\": \"DAY\",\n", - " \"queryGranularity\": {\n", - " \"type\": \"none\"\n", - " },\n", - " \"rollup\": False,\n", - " \"intervals\": []\n", - " },\n", - " \"transformSpec\": {\n", - " \"filter\": None,\n", - " \"transforms\": []\n", - " }\n", - " },\n", - " \"ioConfig\": {\n", - " \"topic\": \"raw_scouter-opcua-orchestrated-pipeline\",\n", - " \"topicPattern\": None,\n", - " \"inputFormat\": {\n", - " \"type\": \"kafka\",\n", - " \"headerFormat\": None,\n", - " \"keyFormat\": None,\n", - " \"valueFormat\": {\n", - " \"type\": \"json\",\n", - " \"keepNoneColumns\": False,\n", - " \"assumeNewlineDelimited\": False,\n", - " \"useJsonNodeReader\": False\n", - " },\n", - " \"headerColumnPrefix\": \"kafka.header.\",\n", - " \"keyColumnName\": \"kafka.key\",\n", - " \"timestampColumnName\": \"kafka.timestamp\",\n", - " \"topicColumnName\": \"kafka.topic\"\n", - " },\n", - " \"replicas\": 1,\n", - " \"taskCount\": 1,\n", - " \"taskDuration\": \"PT3600S\",\n", - " \"consumerProperties\": {\n", - " \"bootstrap.servers\": \"kafka.kafka.svc.cluster.local:9092\"\n", - " },\n", - " \"autoScalerConfig\": None,\n", - " \"pollTimeout\": 100,\n", - " \"startDelay\": \"PT5S\",\n", - " \"period\": \"PT30S\",\n", - " \"useEarliestOffset\": True,\n", - " \"completionTimeout\": \"PT1800S\",\n", - " \"lateMessageRejectionPeriod\": None,\n", - " \"earlyMessageRejectionPeriod\": None,\n", - " \"lateMessageRejectionStartDateTime\": None,\n", - " \"configOverrides\": None,\n", - " \"idleConfig\": None,\n", - " \"stopTaskCount\": None,\n", - " \"stream\": \"raw_scouter-opcua-orchestrated-pipeline\",\n", - " \"useEarliestSequenceNumber\": True\n", - " },\n", - " \"tuningConfig\": {\n", - " \"type\": \"kafka\",\n", - " \"appendableIndexSpec\": {\n", - " \"type\": \"onheap\",\n", - " \"preserveExistingMetrics\": False\n", - " },\n", - " \"maxRowsInMemory\": 150000,\n", - " \"maxBytesInMemory\": 0,\n", - " \"skipBytesInMemoryOverheadCheck\": False,\n", - " \"maxRowsPerSegment\": 5000000,\n", - " \"maxTotalRows\": None,\n", - " \"intermediatePersistPeriod\": \"PT10M\",\n", - " \"maxPendingPersists\": 0,\n", - " \"indexSpec\": {\n", - " \"bitmap\": {\n", - " \"type\": \"roaring\"\n", - " },\n", - " \"dimensionCompression\": \"lz4\",\n", - " \"stringDictionaryEncoding\": {\n", - " \"type\": \"utf8\"\n", - " },\n", - " \"metricCompression\": \"lz4\",\n", - " \"longEncoding\": \"longs\"\n", - " },\n", - " \"indexSpecForIntermediatePersists\": {\n", - " \"bitmap\": {\n", - " \"type\": \"roaring\"\n", - " },\n", - " \"dimensionCompression\": \"lz4\",\n", - " \"stringDictionaryEncoding\": {\n", - " \"type\": \"utf8\"\n", - " },\n", - " \"metricCompression\": \"lz4\",\n", - " \"longEncoding\": \"longs\"\n", - " },\n", - " \"reportParseExceptions\": False,\n", - " \"handoffConditionTimeout\": 900000,\n", - " \"resetOffsetAutomatically\": False,\n", - " \"segmentWriteOutMediumFactory\": None,\n", - " \"workerThreads\": None,\n", - " \"chatRetries\": 8,\n", - " \"httpTimeout\": \"PT10S\",\n", - " \"shutdownTimeout\": \"PT80S\",\n", - " \"offsetFetchPeriod\": \"PT30S\",\n", - " \"intermediateHandoffPeriod\": \"P2147483647D\",\n", - " \"logParseExceptions\": False,\n", - " \"maxParseExceptions\": 2147483647,\n", - " \"maxSavedParseExceptions\": 0,\n", - " \"numPersistThreads\": 1,\n", - " \"skipSequenceNumberAvailabilityCheck\": False,\n", - " \"repartitionTransitionDuration\": \"PT120S\"\n", - " }\n", - " },\n", - " \"context\": None,\n", - " \"suspended\": False\n", - "}" - ] - }, - { - "cell_type": "code", - "execution_count": 18, - "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "raw_scouter-opcua-pipeline-2\n", - "raw_scouter-opcua-pipeline-3\n", - "raw_scouter-opcua-pipeline-4\n", - "raw_scouter-opcua-pipeline-5\n", - "raw_scouter-opcua-pipeline-6\n", - "raw_scouter-opcua-pipeline-7\n", - "raw_scouter-opcua-pipeline-8\n", - "raw_scouter-opcua-pipeline-9\n", - "raw_scouter-opcua-pipeline-10\n", - "raw_scouter-opcua-pipeline-11\n", - "raw_scouter-opcua-pipeline-12\n", - "raw_scouter-opcua-pipeline-13\n", - "raw_scouter-opcua-pipeline-14\n", - "raw_scouter-opcua-pipeline-15\n", - "raw_scouter-opcua-pipeline-16\n", - "raw_scouter-opcua-pipeline-17\n", - "raw_scouter-opcua-pipeline-18\n", - "raw_scouter-opcua-pipeline-19\n", - "raw_scouter-opcua-pipeline-20\n", - "raw_scouter-opcua-pipeline-21\n", - "raw_scouter-opcua-pipeline-22\n", - "raw_scouter-opcua-pipeline-23\n", - "raw_scouter-opcua-pipeline-24\n", - "raw_scouter-opcua-pipeline-25\n", - "raw_scouter-opcua-pipeline-26\n", - "raw_scouter-opcua-pipeline-27\n", - "raw_scouter-opcua-pipeline-28\n", - "raw_scouter-opcua-pipeline-29\n", - "raw_scouter-opcua-pipeline-30\n" - ] - } - ], - "source": [ - "import json\n", - "from copy import deepcopy\n", - "\n", - "json_list = []\n", - "i = 0\n", - "for i in range(1, 30):\n", - " topic = f\"raw_scouter-opcua-pipeline-{i+1}\"\n", - " print(topic)\n", - " obj = deepcopy(spec)\n", - " obj['spec']['dataSchema']['dataSource'] = topic\n", - " obj['spec']['ioConfig']['topic'] = topic\n", - " obj['spec']['ioConfig']['stream'] = topic\n", - "\n", - " json_list.append(obj)\n", - "\n", - "# Save to a new file\n", - "with open('specs_30.json', 'w') as f:\n", - " json.dump(json_list, f, indent=2)" - ] - }, - { - "cell_type": "code", - "execution_count": 19, - "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "[OK] raw_scouter-opcua-pipeline-2 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-3 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-4 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-5 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-6 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-7 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-8 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-9 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-10 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-11 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-12 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-13 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-14 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-15 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-16 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-17 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-18 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-19 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-20 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-21 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-22 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-23 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-24 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-25 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-26 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-27 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-28 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-29 enviado.\n", - "[OK] raw_scouter-opcua-pipeline-30 enviado.\n" - ] - } - ], - "source": [ - "import os\n", - "import requests\n", - "\n", - "DRUID_OVERLORD = os.getenv(\"DRUID_OVERLORD\", \"http://localhost:8082\")\n", - "SUPERVISOR_ENDPOINT = f\"{DRUID_OVERLORD}/druid/indexer/v1/supervisor\"\n", - "\n", - "def enviar_supervisores(specs):\n", - " for spec in specs:\n", - " resp = requests.post(\n", - " SUPERVISOR_ENDPOINT,\n", - " headers={\"Content-Type\": \"application/json\"},\n", - " json=spec\n", - " )\n", - " if resp.status_code == 200:\n", - " print(f\"[OK] {spec['spec']['dataSchema']['dataSource']} enviado.\")\n", - " else:\n", - " print(f\"[ERRO] {spec['spec']['dataSchema']['dataSource']}: {resp.status_code} → {resp.text}\")\n", - "\n", - "with open(\"./specs_30.json\", \"r\") as f:\n", - " specs = json.load(f)\n", - "enviar_supervisores(specs)\n" - ] - }, - { - "cell_type": "code", - "execution_count": 20, - "metadata": {}, - "outputs": [ - { - "ename": "ConnectionError", - "evalue": "HTTPConnectionPool(host='localhost', port=8090): Max retries exceeded with url: /druid/indexer/v1/supervisor/raw_scouter-opcua-orchestrated-pipeline-2/terminate (Caused by NewConnectionError(': Failed to establish a new connection: [Errno 111] Connection refused'))", - "output_type": "error", - "traceback": [ - "\u001b[31m---------------------------------------------------------------------------\u001b[39m", - "\u001b[31mConnectionRefusedError\u001b[39m Traceback (most recent call last)", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connection.py:198\u001b[39m, in \u001b[36mHTTPConnection._new_conn\u001b[39m\u001b[34m(self)\u001b[39m\n\u001b[32m 197\u001b[39m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[32m--> \u001b[39m\u001b[32m198\u001b[39m sock = \u001b[43mconnection\u001b[49m\u001b[43m.\u001b[49m\u001b[43mcreate_connection\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m 199\u001b[39m \u001b[43m \u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_dns_host\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43mport\u001b[49m\u001b[43m)\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 200\u001b[39m \u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43mtimeout\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 201\u001b[39m \u001b[43m \u001b[49m\u001b[43msource_address\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43msource_address\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 202\u001b[39m \u001b[43m \u001b[49m\u001b[43msocket_options\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43msocket_options\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 203\u001b[39m \u001b[43m \u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 204\u001b[39m \u001b[38;5;28;01mexcept\u001b[39;00m socket.gaierror \u001b[38;5;28;01mas\u001b[39;00m e:\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/util/connection.py:85\u001b[39m, in \u001b[36mcreate_connection\u001b[39m\u001b[34m(address, timeout, source_address, socket_options)\u001b[39m\n\u001b[32m 84\u001b[39m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[32m---> \u001b[39m\u001b[32m85\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m err\n\u001b[32m 86\u001b[39m \u001b[38;5;28;01mfinally\u001b[39;00m:\n\u001b[32m 87\u001b[39m \u001b[38;5;66;03m# Break explicitly a reference cycle\u001b[39;00m\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/util/connection.py:73\u001b[39m, in \u001b[36mcreate_connection\u001b[39m\u001b[34m(address, timeout, source_address, socket_options)\u001b[39m\n\u001b[32m 72\u001b[39m sock.bind(source_address)\n\u001b[32m---> \u001b[39m\u001b[32m73\u001b[39m \u001b[43msock\u001b[49m\u001b[43m.\u001b[49m\u001b[43mconnect\u001b[49m\u001b[43m(\u001b[49m\u001b[43msa\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 74\u001b[39m \u001b[38;5;66;03m# Break explicitly a reference cycle\u001b[39;00m\n", - "\u001b[31mConnectionRefusedError\u001b[39m: [Errno 111] Connection refused", - "\nThe above exception was the direct cause of the following exception:\n", - "\u001b[31mNewConnectionError\u001b[39m Traceback (most recent call last)", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connectionpool.py:787\u001b[39m, in \u001b[36mHTTPConnectionPool.urlopen\u001b[39m\u001b[34m(self, method, url, body, headers, retries, redirect, assert_same_host, timeout, pool_timeout, release_conn, chunked, body_pos, preload_content, decode_content, **response_kw)\u001b[39m\n\u001b[32m 786\u001b[39m \u001b[38;5;66;03m# Make the request on the HTTPConnection object\u001b[39;00m\n\u001b[32m--> \u001b[39m\u001b[32m787\u001b[39m response = \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_make_request\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m 788\u001b[39m \u001b[43m \u001b[49m\u001b[43mconn\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 789\u001b[39m \u001b[43m \u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 790\u001b[39m \u001b[43m \u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 791\u001b[39m \u001b[43m \u001b[49m\u001b[43mtimeout\u001b[49m\u001b[43m=\u001b[49m\u001b[43mtimeout_obj\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 792\u001b[39m \u001b[43m \u001b[49m\u001b[43mbody\u001b[49m\u001b[43m=\u001b[49m\u001b[43mbody\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 793\u001b[39m \u001b[43m \u001b[49m\u001b[43mheaders\u001b[49m\u001b[43m=\u001b[49m\u001b[43mheaders\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 794\u001b[39m \u001b[43m \u001b[49m\u001b[43mchunked\u001b[49m\u001b[43m=\u001b[49m\u001b[43mchunked\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 795\u001b[39m \u001b[43m \u001b[49m\u001b[43mretries\u001b[49m\u001b[43m=\u001b[49m\u001b[43mretries\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 796\u001b[39m \u001b[43m \u001b[49m\u001b[43mresponse_conn\u001b[49m\u001b[43m=\u001b[49m\u001b[43mresponse_conn\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 797\u001b[39m \u001b[43m \u001b[49m\u001b[43mpreload_content\u001b[49m\u001b[43m=\u001b[49m\u001b[43mpreload_content\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 798\u001b[39m \u001b[43m \u001b[49m\u001b[43mdecode_content\u001b[49m\u001b[43m=\u001b[49m\u001b[43mdecode_content\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 799\u001b[39m \u001b[43m \u001b[49m\u001b[43m*\u001b[49m\u001b[43m*\u001b[49m\u001b[43mresponse_kw\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 800\u001b[39m \u001b[43m\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 802\u001b[39m \u001b[38;5;66;03m# Everything went great!\u001b[39;00m\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connectionpool.py:493\u001b[39m, in \u001b[36mHTTPConnectionPool._make_request\u001b[39m\u001b[34m(self, conn, method, url, body, headers, retries, timeout, chunked, response_conn, preload_content, decode_content, enforce_content_length)\u001b[39m\n\u001b[32m 492\u001b[39m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[32m--> \u001b[39m\u001b[32m493\u001b[39m \u001b[43mconn\u001b[49m\u001b[43m.\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m 494\u001b[39m \u001b[43m \u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 495\u001b[39m \u001b[43m \u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 496\u001b[39m \u001b[43m \u001b[49m\u001b[43mbody\u001b[49m\u001b[43m=\u001b[49m\u001b[43mbody\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 497\u001b[39m \u001b[43m \u001b[49m\u001b[43mheaders\u001b[49m\u001b[43m=\u001b[49m\u001b[43mheaders\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 498\u001b[39m \u001b[43m \u001b[49m\u001b[43mchunked\u001b[49m\u001b[43m=\u001b[49m\u001b[43mchunked\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 499\u001b[39m \u001b[43m \u001b[49m\u001b[43mpreload_content\u001b[49m\u001b[43m=\u001b[49m\u001b[43mpreload_content\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 500\u001b[39m \u001b[43m \u001b[49m\u001b[43mdecode_content\u001b[49m\u001b[43m=\u001b[49m\u001b[43mdecode_content\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 501\u001b[39m \u001b[43m \u001b[49m\u001b[43menforce_content_length\u001b[49m\u001b[43m=\u001b[49m\u001b[43menforce_content_length\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 502\u001b[39m \u001b[43m \u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 504\u001b[39m \u001b[38;5;66;03m# We are swallowing BrokenPipeError (errno.EPIPE) since the server is\u001b[39;00m\n\u001b[32m 505\u001b[39m \u001b[38;5;66;03m# legitimately able to close the connection after sending a valid response.\u001b[39;00m\n\u001b[32m 506\u001b[39m \u001b[38;5;66;03m# With this behaviour, the received response is still readable.\u001b[39;00m\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connection.py:494\u001b[39m, in \u001b[36mHTTPConnection.request\u001b[39m\u001b[34m(self, method, url, body, headers, chunked, preload_content, decode_content, enforce_content_length)\u001b[39m\n\u001b[32m 493\u001b[39m \u001b[38;5;28mself\u001b[39m.putheader(header, value)\n\u001b[32m--> \u001b[39m\u001b[32m494\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43mendheaders\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 496\u001b[39m \u001b[38;5;66;03m# If we're given a body we start sending that in chunks.\u001b[39;00m\n", - "\u001b[36mFile \u001b[39m\u001b[32m/usr/lib/python3.11/http/client.py:1298\u001b[39m, in \u001b[36mHTTPConnection.endheaders\u001b[39m\u001b[34m(self, message_body, encode_chunked)\u001b[39m\n\u001b[32m 1297\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m CannotSendHeader()\n\u001b[32m-> \u001b[39m\u001b[32m1298\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_send_output\u001b[49m\u001b[43m(\u001b[49m\u001b[43mmessage_body\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mencode_chunked\u001b[49m\u001b[43m=\u001b[49m\u001b[43mencode_chunked\u001b[49m\u001b[43m)\u001b[49m\n", - "\u001b[36mFile \u001b[39m\u001b[32m/usr/lib/python3.11/http/client.py:1058\u001b[39m, in \u001b[36mHTTPConnection._send_output\u001b[39m\u001b[34m(self, message_body, encode_chunked)\u001b[39m\n\u001b[32m 1057\u001b[39m \u001b[38;5;28;01mdel\u001b[39;00m \u001b[38;5;28mself\u001b[39m._buffer[:]\n\u001b[32m-> \u001b[39m\u001b[32m1058\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43msend\u001b[49m\u001b[43m(\u001b[49m\u001b[43mmsg\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 1060\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m message_body \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[32m 1061\u001b[39m \n\u001b[32m 1062\u001b[39m \u001b[38;5;66;03m# create a consistent interface to message_body\u001b[39;00m\n", - "\u001b[36mFile \u001b[39m\u001b[32m/usr/lib/python3.11/http/client.py:996\u001b[39m, in \u001b[36mHTTPConnection.send\u001b[39m\u001b[34m(self, data)\u001b[39m\n\u001b[32m 995\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m.auto_open:\n\u001b[32m--> \u001b[39m\u001b[32m996\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43mconnect\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 997\u001b[39m \u001b[38;5;28;01melse\u001b[39;00m:\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connection.py:325\u001b[39m, in \u001b[36mHTTPConnection.connect\u001b[39m\u001b[34m(self)\u001b[39m\n\u001b[32m 324\u001b[39m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34mconnect\u001b[39m(\u001b[38;5;28mself\u001b[39m) -> \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[32m--> \u001b[39m\u001b[32m325\u001b[39m \u001b[38;5;28mself\u001b[39m.sock = \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_new_conn\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 326\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m._tunnel_host:\n\u001b[32m 327\u001b[39m \u001b[38;5;66;03m# If we're tunneling it means we're connected to our proxy.\u001b[39;00m\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connection.py:213\u001b[39m, in \u001b[36mHTTPConnection._new_conn\u001b[39m\u001b[34m(self)\u001b[39m\n\u001b[32m 212\u001b[39m \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mOSError\u001b[39;00m \u001b[38;5;28;01mas\u001b[39;00m e:\n\u001b[32m--> \u001b[39m\u001b[32m213\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m NewConnectionError(\n\u001b[32m 214\u001b[39m \u001b[38;5;28mself\u001b[39m, \u001b[33mf\u001b[39m\u001b[33m\"\u001b[39m\u001b[33mFailed to establish a new connection: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00me\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m\"\u001b[39m\n\u001b[32m 215\u001b[39m ) \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34;01me\u001b[39;00m\n\u001b[32m 217\u001b[39m sys.audit(\u001b[33m\"\u001b[39m\u001b[33mhttp.client.connect\u001b[39m\u001b[33m\"\u001b[39m, \u001b[38;5;28mself\u001b[39m, \u001b[38;5;28mself\u001b[39m.host, \u001b[38;5;28mself\u001b[39m.port)\n", - "\u001b[31mNewConnectionError\u001b[39m: : Failed to establish a new connection: [Errno 111] Connection refused", - "\nThe above exception was the direct cause of the following exception:\n", - "\u001b[31mMaxRetryError\u001b[39m Traceback (most recent call last)", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/requests/adapters.py:667\u001b[39m, in \u001b[36mHTTPAdapter.send\u001b[39m\u001b[34m(self, request, stream, timeout, verify, cert, proxies)\u001b[39m\n\u001b[32m 666\u001b[39m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[32m--> \u001b[39m\u001b[32m667\u001b[39m resp = \u001b[43mconn\u001b[49m\u001b[43m.\u001b[49m\u001b[43murlopen\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m 668\u001b[39m \u001b[43m \u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m=\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m.\u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 669\u001b[39m \u001b[43m \u001b[49m\u001b[43murl\u001b[49m\u001b[43m=\u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 670\u001b[39m \u001b[43m \u001b[49m\u001b[43mbody\u001b[49m\u001b[43m=\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m.\u001b[49m\u001b[43mbody\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 671\u001b[39m \u001b[43m \u001b[49m\u001b[43mheaders\u001b[49m\u001b[43m=\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m.\u001b[49m\u001b[43mheaders\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 672\u001b[39m \u001b[43m \u001b[49m\u001b[43mredirect\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43;01mFalse\u001b[39;49;00m\u001b[43m,\u001b[49m\n\u001b[32m 673\u001b[39m \u001b[43m \u001b[49m\u001b[43massert_same_host\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43;01mFalse\u001b[39;49;00m\u001b[43m,\u001b[49m\n\u001b[32m 674\u001b[39m \u001b[43m \u001b[49m\u001b[43mpreload_content\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43;01mFalse\u001b[39;49;00m\u001b[43m,\u001b[49m\n\u001b[32m 675\u001b[39m \u001b[43m \u001b[49m\u001b[43mdecode_content\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43;01mFalse\u001b[39;49;00m\u001b[43m,\u001b[49m\n\u001b[32m 676\u001b[39m \u001b[43m \u001b[49m\u001b[43mretries\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43mmax_retries\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 677\u001b[39m \u001b[43m \u001b[49m\u001b[43mtimeout\u001b[49m\u001b[43m=\u001b[49m\u001b[43mtimeout\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 678\u001b[39m \u001b[43m \u001b[49m\u001b[43mchunked\u001b[49m\u001b[43m=\u001b[49m\u001b[43mchunked\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 679\u001b[39m \u001b[43m \u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 681\u001b[39m \u001b[38;5;28;01mexcept\u001b[39;00m (ProtocolError, \u001b[38;5;167;01mOSError\u001b[39;00m) \u001b[38;5;28;01mas\u001b[39;00m err:\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connectionpool.py:841\u001b[39m, in \u001b[36mHTTPConnectionPool.urlopen\u001b[39m\u001b[34m(self, method, url, body, headers, retries, redirect, assert_same_host, timeout, pool_timeout, release_conn, chunked, body_pos, preload_content, decode_content, **response_kw)\u001b[39m\n\u001b[32m 839\u001b[39m new_e = ProtocolError(\u001b[33m\"\u001b[39m\u001b[33mConnection aborted.\u001b[39m\u001b[33m\"\u001b[39m, new_e)\n\u001b[32m--> \u001b[39m\u001b[32m841\u001b[39m retries = \u001b[43mretries\u001b[49m\u001b[43m.\u001b[49m\u001b[43mincrement\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m 842\u001b[39m \u001b[43m \u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43merror\u001b[49m\u001b[43m=\u001b[49m\u001b[43mnew_e\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m_pool\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m_stacktrace\u001b[49m\u001b[43m=\u001b[49m\u001b[43msys\u001b[49m\u001b[43m.\u001b[49m\u001b[43mexc_info\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\u001b[43m[\u001b[49m\u001b[32;43m2\u001b[39;49m\u001b[43m]\u001b[49m\n\u001b[32m 843\u001b[39m \u001b[43m\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 844\u001b[39m retries.sleep()\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/util/retry.py:519\u001b[39m, in \u001b[36mRetry.increment\u001b[39m\u001b[34m(self, method, url, response, error, _pool, _stacktrace)\u001b[39m\n\u001b[32m 518\u001b[39m reason = error \u001b[38;5;129;01mor\u001b[39;00m ResponseError(cause)\n\u001b[32m--> \u001b[39m\u001b[32m519\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m MaxRetryError(_pool, url, reason) \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34;01mreason\u001b[39;00m \u001b[38;5;66;03m# type: ignore[arg-type]\u001b[39;00m\n\u001b[32m 521\u001b[39m log.debug(\u001b[33m\"\u001b[39m\u001b[33mIncremented Retry for (url=\u001b[39m\u001b[33m'\u001b[39m\u001b[38;5;132;01m%s\u001b[39;00m\u001b[33m'\u001b[39m\u001b[33m): \u001b[39m\u001b[38;5;132;01m%r\u001b[39;00m\u001b[33m\"\u001b[39m, url, new_retry)\n", - "\u001b[31mMaxRetryError\u001b[39m: HTTPConnectionPool(host='localhost', port=8090): Max retries exceeded with url: /druid/indexer/v1/supervisor/raw_scouter-opcua-orchestrated-pipeline-2/terminate (Caused by NewConnectionError(': Failed to establish a new connection: [Errno 111] Connection refused'))", - "\nDuring handling of the above exception, another exception occurred:\n", - "\u001b[31mConnectionError\u001b[39m Traceback (most recent call last)", - "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[20]\u001b[39m\u001b[32m, line 16\u001b[39m\n\u001b[32m 13\u001b[39m \u001b[38;5;28mprint\u001b[39m(\u001b[33mf\u001b[39m\u001b[33m\"\u001b[39m\u001b[33m❌ Erro \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mresp.status_code\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mresp.text\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m\"\u001b[39m)\n\u001b[32m 15\u001b[39m \u001b[38;5;28;01mfor\u001b[39;00m i \u001b[38;5;129;01min\u001b[39;00m \u001b[38;5;28mrange\u001b[39m(\u001b[32m1\u001b[39m, \u001b[32m30\u001b[39m):\n\u001b[32m---> \u001b[39m\u001b[32m16\u001b[39m \u001b[43mterminate_supervisor\u001b[49m\u001b[43m(\u001b[49m\u001b[33;43mf\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43mraw_scouter-opcua-orchestrated-pipeline-\u001b[39;49m\u001b[38;5;132;43;01m{\u001b[39;49;00m\u001b[43mi\u001b[49m\u001b[43m+\u001b[49m\u001b[32;43m1\u001b[39;49m\u001b[38;5;132;43;01m}\u001b[39;49;00m\u001b[33;43m\"\u001b[39;49m\u001b[43m)\u001b[49m\n", - "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[20]\u001b[39m\u001b[32m, line 7\u001b[39m, in \u001b[36mterminate_supervisor\u001b[39m\u001b[34m(supervisor_id)\u001b[39m\n\u001b[32m 5\u001b[39m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34mterminate_supervisor\u001b[39m(supervisor_id):\n\u001b[32m 6\u001b[39m url = \u001b[33mf\u001b[39m\u001b[33m\"\u001b[39m\u001b[38;5;132;01m{\u001b[39;00mDRUID\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m/druid/indexer/v1/supervisor/\u001b[39m\u001b[38;5;132;01m{\u001b[39;00msupervisor_id\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m/terminate\u001b[39m\u001b[33m\"\u001b[39m\n\u001b[32m----> \u001b[39m\u001b[32m7\u001b[39m resp = \u001b[43mrequests\u001b[49m\u001b[43m.\u001b[49m\u001b[43mpost\u001b[49m\u001b[43m(\u001b[49m\u001b[43murl\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 8\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m resp.status_code == \u001b[32m200\u001b[39m:\n\u001b[32m 9\u001b[39m \u001b[38;5;28mprint\u001b[39m(\u001b[33mf\u001b[39m\u001b[33m\"\u001b[39m\u001b[33m✅ Supervisor \u001b[39m\u001b[33m'\u001b[39m\u001b[38;5;132;01m{\u001b[39;00msupervisor_id\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m'\u001b[39m\u001b[33m encerrado.\u001b[39m\u001b[33m\"\u001b[39m)\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/requests/api.py:115\u001b[39m, in \u001b[36mpost\u001b[39m\u001b[34m(url, data, json, **kwargs)\u001b[39m\n\u001b[32m 103\u001b[39m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34mpost\u001b[39m(url, data=\u001b[38;5;28;01mNone\u001b[39;00m, json=\u001b[38;5;28;01mNone\u001b[39;00m, **kwargs):\n\u001b[32m 104\u001b[39m \u001b[38;5;250m \u001b[39m\u001b[33mr\u001b[39m\u001b[33;03m\"\"\"Sends a POST request.\u001b[39;00m\n\u001b[32m 105\u001b[39m \n\u001b[32m 106\u001b[39m \u001b[33;03m :param url: URL for the new :class:`Request` object.\u001b[39;00m\n\u001b[32m (...)\u001b[39m\u001b[32m 112\u001b[39m \u001b[33;03m :rtype: requests.Response\u001b[39;00m\n\u001b[32m 113\u001b[39m \u001b[33;03m \"\"\"\u001b[39;00m\n\u001b[32m--> \u001b[39m\u001b[32m115\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[43mrequest\u001b[49m\u001b[43m(\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43mpost\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mdata\u001b[49m\u001b[43m=\u001b[49m\u001b[43mdata\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mjson\u001b[49m\u001b[43m=\u001b[49m\u001b[43mjson\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m*\u001b[49m\u001b[43m*\u001b[49m\u001b[43mkwargs\u001b[49m\u001b[43m)\u001b[49m\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/requests/api.py:59\u001b[39m, in \u001b[36mrequest\u001b[39m\u001b[34m(method, url, **kwargs)\u001b[39m\n\u001b[32m 55\u001b[39m \u001b[38;5;66;03m# By using the 'with' statement we are sure the session is closed, thus we\u001b[39;00m\n\u001b[32m 56\u001b[39m \u001b[38;5;66;03m# avoid leaving sockets open which can trigger a ResourceWarning in some\u001b[39;00m\n\u001b[32m 57\u001b[39m \u001b[38;5;66;03m# cases, and look like a memory leak in others.\u001b[39;00m\n\u001b[32m 58\u001b[39m \u001b[38;5;28;01mwith\u001b[39;00m sessions.Session() \u001b[38;5;28;01mas\u001b[39;00m session:\n\u001b[32m---> \u001b[39m\u001b[32m59\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[43msession\u001b[49m\u001b[43m.\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m(\u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m=\u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43murl\u001b[49m\u001b[43m=\u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m*\u001b[49m\u001b[43m*\u001b[49m\u001b[43mkwargs\u001b[49m\u001b[43m)\u001b[49m\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/requests/sessions.py:589\u001b[39m, in \u001b[36mSession.request\u001b[39m\u001b[34m(self, method, url, params, data, headers, cookies, files, auth, timeout, allow_redirects, proxies, hooks, stream, verify, cert, json)\u001b[39m\n\u001b[32m 584\u001b[39m send_kwargs = {\n\u001b[32m 585\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mtimeout\u001b[39m\u001b[33m\"\u001b[39m: timeout,\n\u001b[32m 586\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mallow_redirects\u001b[39m\u001b[33m\"\u001b[39m: allow_redirects,\n\u001b[32m 587\u001b[39m }\n\u001b[32m 588\u001b[39m send_kwargs.update(settings)\n\u001b[32m--> \u001b[39m\u001b[32m589\u001b[39m resp = \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43msend\u001b[49m\u001b[43m(\u001b[49m\u001b[43mprep\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m*\u001b[49m\u001b[43m*\u001b[49m\u001b[43msend_kwargs\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 591\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m resp\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/requests/sessions.py:703\u001b[39m, in \u001b[36mSession.send\u001b[39m\u001b[34m(self, request, **kwargs)\u001b[39m\n\u001b[32m 700\u001b[39m start = preferred_clock()\n\u001b[32m 702\u001b[39m \u001b[38;5;66;03m# Send the request\u001b[39;00m\n\u001b[32m--> \u001b[39m\u001b[32m703\u001b[39m r = \u001b[43madapter\u001b[49m\u001b[43m.\u001b[49m\u001b[43msend\u001b[49m\u001b[43m(\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m*\u001b[49m\u001b[43m*\u001b[49m\u001b[43mkwargs\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 705\u001b[39m \u001b[38;5;66;03m# Total elapsed time of the request (approximately)\u001b[39;00m\n\u001b[32m 706\u001b[39m elapsed = preferred_clock() - start\n", - "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/requests/adapters.py:700\u001b[39m, in \u001b[36mHTTPAdapter.send\u001b[39m\u001b[34m(self, request, stream, timeout, verify, cert, proxies)\u001b[39m\n\u001b[32m 696\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(e.reason, _SSLError):\n\u001b[32m 697\u001b[39m \u001b[38;5;66;03m# This branch is for urllib3 v1.22 and later.\u001b[39;00m\n\u001b[32m 698\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m SSLError(e, request=request)\n\u001b[32m--> \u001b[39m\u001b[32m700\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mConnectionError\u001b[39;00m(e, request=request)\n\u001b[32m 702\u001b[39m \u001b[38;5;28;01mexcept\u001b[39;00m ClosedPoolError \u001b[38;5;28;01mas\u001b[39;00m e:\n\u001b[32m 703\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mConnectionError\u001b[39;00m(e, request=request)\n", - "\u001b[31mConnectionError\u001b[39m: HTTPConnectionPool(host='localhost', port=8090): Max retries exceeded with url: /druid/indexer/v1/supervisor/raw_scouter-opcua-orchestrated-pipeline-2/terminate (Caused by NewConnectionError(': Failed to establish a new connection: [Errno 111] Connection refused'))" - ] - } - ], - "source": [ - "import os\n", - "import requests\n", - "\n", - "DRUID = os.getenv(\"DRUID_URL\", \"http://localhost:8082\")\n", - "def terminate_supervisor(supervisor_id):\n", - " url = f\"{DRUID}/druid/indexer/v1/supervisor/{supervisor_id}/terminate\"\n", - " resp = requests.post(url)\n", - " if resp.status_code == 200:\n", - " print(f\"✅ Supervisor '{supervisor_id}' encerrado.\")\n", - " elif resp.status_code == 404:\n", - " print(f\"⚠️ Supervisor '{supervisor_id}' não encontrado.\")\n", - " else:\n", - " print(f\"❌ Erro {resp.status_code}: {resp.text}\")\n", - "\n", - "for i in range(1, 30):\n", - " terminate_supervisor(f\"raw_scouter-opcua-orchestrated-pipeline-{i+1}\")\n" - ] - }, - { - "cell_type": "code", - "execution_count": 3, - "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "✅ Document deleted successfully\n" - ] - } - ], - "source": [ - "from pymongo import MongoClient\n", - "import os\n", - "# Get MongoDB connection details from environment variables\n", - "MONGODB_USERNAME = os.getenv(\"MONGODB_USERNAME\", \"root\")\n", - "MONGODB_PASSWORD = os.getenv(\"MONGODB_PASSWORD\", \"wKZDbMNU1c\") \n", - "MONGODB_URL = os.getenv(\"MONGODB_URL\", \"localhost:27018\")\n", - "MONGODB_DATABASE = os.getenv(\"MONGODB_DATABASE\", \"sientia\")\n", - "\n", - "# Create MongoDB client\n", - "client = MongoClient(\n", - " f\"mongodb://{MONGODB_USERNAME}:{MONGODB_PASSWORD}@{MONGODB_URL}\"\n", - ")\n", - "\n", - "# Get database and collection\n", - "db = client[MONGODB_DATABASE]\n", - "collection = db[\"pipelines\"] # Replace with actual collection name\n", - "\n", - "for i in range(2, 31):\n", - " # Delete a document matching specific criteria\n", - " result = collection.delete_one({\"schedule_name\": f\"scouter-opcua-pipeline-{i}\"}) # Replace with actual query\n", - "\n", - "if result.deleted_count > 0:\n", - " print(\"✅ Document deleted successfully\")\n", - "else:\n", - " print(\"⚠️ No matching document found\")\n", - "\n", - "# Close the connection\n", - "client.close()\n" - ] - } - ], - "metadata": { - "kernelspec": { - "display_name": "venv", - "language": "python", - "name": "python3" - }, - "language_info": { - "codemirror_mode": { - "name": "ipython", - "version": 3 - }, - "file_extension": ".py", - "mimetype": "text/x-python", - "name": "python", - "nbconvert_exporter": "python", - "pygments_lexer": "ipython3", - "version": "3.11.13" - } - }, - "nbformat": 4, - "nbformat_minor": 2 -} diff --git a/values.yaml b/values.yaml index fe40615..9941b96 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.5.2" + tag: "1.0.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: @@ -150,7 +150,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git" - name: GITHUB_BRANCH - value: "fix/SIENTIAPDE-1273" + value: "release/SIENTIAPDE-1441" - name: PYTHON_APP value: "scouter.worker.worker" @@ -206,10 +206,8 @@ env: - name: MONGODB_DATABASE value: "sientia" - - name: DRUID_HOST - value: "druid-router.druid.svc.cluster.local" - - name: DRUID_PORT - value: "8888" + - name: PYPI_SERVER + value: "http://library-distribution-server.library.svc.cluster.local:5000" ssh: enabled: true From e879107053fd84f42349ea2471e8d0b8eed6c73d Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Dec 2025 10:10:42 -0300 Subject: [PATCH 02/11] SIENTIAPDE-1441 SIENTIAPDE-1442: Enhance Temporal worker configuration by adding environment variables for concurrency limits and poller behavior settings in values.yaml and worker.py for optimized performance. --- scouter/worker/worker.py | 40 ++++++++++++++++++++++++++++++++++------ values.yaml | 25 +++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 6 deletions(-) diff --git a/scouter/worker/worker.py b/scouter/worker/worker.py index 75e6715..228562c 100644 --- a/scouter/worker/worker.py +++ b/scouter/worker/worker.py @@ -26,6 +26,26 @@ POD_ID = os.getenv('HOSTNAME', 'localhost') SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091')) +# For optmized latency, Temporal docs recommends fixed slots, ensuring +# high concurency levels. + +MAX_CONCURRENT_WORKFLOW_TASKS = int(os.getenv('MAX_CONCURRENT_WORKFLOW_TASKS', '200')) +MAX_CONCURRENT_ACTIVITIES = int(os.getenv('MAX_CONCURRENT_ACTIVITIES', '200')) +MAX_CONCURRENT_LOCAL_ACTIVITIES = int(os.getenv('MAX_CONCURRENT_LOCAL_ACTIVITIES', '200')) +MAX_CACHED_WORKFLOWS = int(os.getenv('MAX_CACHED_WORKFLOWS', '200')) + + +# Temporal docs also recommends an autoscaling policy, with agrresive limits to prioritize latency over throughput. + +WORKFLOW_POLLER_BEHAVIUR_MINIMUM = int(os.getenv('WORKFLOW_POLLER_BEHAVIUR_MINIMUM', '10')) +WORKFLOW_POLLER_BEHAVIUR_INITIAL = int(os.getenv('WORKFLOW_POLLER_BEHAVIUR_INITIAL', '100')) +WORKFLOW_POLLER_BEHAVIUR_MAXIMUM = int(os.getenv('WORKFLOW_POLLER_BEHAVIUR_MAXIMUM', '200')) + +ACTIVITY_POLLER_BEHAVIUR_MINIMUM = int(os.getenv('ACTIVITY_POLLER_BEHAVIUR_MINIMUM', '10')) +ACTIVITY_POLLER_BEHAVIUR_INITIAL = int(os.getenv('ACTIVITY_POLLER_BEHAVIUR_INITIAL', '100')) +ACTIVITY_POLLER_BEHAVIUR_MAXIMUM = int(os.getenv('ACTIVITY_POLLER_BEHAVIUR_MAXIMUM', '200')) + + async def main(): """ Main entry point for the Scouter Temporal worker. @@ -116,12 +136,20 @@ async def main(): activities.write_metrics, activities.store_data_package, ], - max_concurrent_workflow_tasks=50, - max_concurrent_activities=50, - max_concurrent_local_activities=50, - max_cached_workflows=200, - workflow_task_poller_behavior=PollerBehaviorAutoscaling(), - activity_task_poller_behavior=PollerBehaviorAutoscaling(), + max_concurrent_workflow_tasks=MAX_CONCURRENT_WORKFLOW_TASKS, + max_concurrent_activities=MAX_CONCURRENT_ACTIVITIES, + max_concurrent_local_activities=MAX_CONCURRENT_LOCAL_ACTIVITIES, + max_cached_workflows=MAX_CACHED_WORKFLOWS, + workflow_task_poller_behavior=PollerBehaviorAutoscaling( + minimum=WORKFLOW_POLLER_BEHAVIUR_MINIMUM, + initial=WORKFLOW_POLLER_BEHAVIUR_INITIAL, + maximum=WORKFLOW_POLLER_BEHAVIUR_MAXIMUM, + ), + activity_task_poller_behavior=PollerBehaviorAutoscaling( + minimum=ACTIVITY_POLLER_BEHAVIUR_MINIMUM, + initial=ACTIVITY_POLLER_BEHAVIUR_INITIAL, + maximum=ACTIVITY_POLLER_BEHAVIUR_MAXIMUM, + ), ) ] diff --git a/values.yaml b/values.yaml index 9941b96..b88a2a2 100644 --- a/values.yaml +++ b/values.yaml @@ -209,6 +209,31 @@ env: - name: PYPI_SERVER value: "http://library-distribution-server.library.svc.cluster.local:5000" + # Temporal worker tuning + - name: MAX_CONCURRENT_WORKFLOW_TASKS + value: 200 + - name: MAX_CONCURRENT_ACTIVITIES + value: 200 + - name: MAX_CONCURRENT_LOCAL_ACTIVITIES + value: 200 + - name: MAX_CACHED_WORKFLOWS + value: 200 + + - name: WORKFLOW_POLLER_BEHAVIUR_MINIMUM + value: 10 + - name: WORKFLOW_POLLER_BEHAVIUR_INITIAL + value: 100 + - name: WORKFLOW_POLLER_BEHAVIUR_MAXIMUM + value: 200 + + - name: ACTIVITY_POLLER_BEHAVIUR_MINIMUM + value: 10 + - name: ACTIVITY_POLLER_BEHAVIUR_INITIAL + value: 100 + - name: ACTIVITY_POLLER_BEHAVIUR_MAXIMUM + value: 200 + + ssh: enabled: true secretName: git-ssh-key-sientia-scouter-worker From 5c4f9eca96e32dbeabb54c3d2b4733b439ced40b Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Dec 2025 10:40:21 -0300 Subject: [PATCH 03/11] SIENTIAPDE-1441 Update README.md and values.yaml to enhance system description and worker configuration. Added ML model orchestration to the system overview, introduced new debug mode features, and updated environment variable values to be strings for consistency. --- README.md | 140 +++++++++++++++++++++++++++++----------------------- values.yaml | 20 ++++---- 2 files changed, 87 insertions(+), 73 deletions(-) diff --git a/README.md b/README.md index 08853ab..0cad971 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Sientia DataOps Scouter -A high-performance, scalable data processing system built on Temporal.io for industrial data collection, processing, and analytics. The Scouter system provides enterprise-grade data ingestion from multiple sources with automatic data quality validation, aggregation, and export capabilities. +A high-performance, scalable data processing and ML model orchestration system built on Temporal.io for industrial data collection, processing, and analytics. The Scouter system provides enterprise-grade data ingestion from multiple sources with automatic data quality validation, aggregation, and export capabilities. ## Features @@ -17,7 +17,8 @@ A high-performance, scalable data processing system built on Temporal.io for ind - **Configurable Data Retention**: Redis-based temporary storage with TTL management - **Notification System**: Integrated alerting and notification management via MongoDB - **Scalable Architecture**: Kubernetes-ready deployment with horizontal scaling support -- **Test Data Generation**: Built-in fake data generation for development and testing +- **Debug Mode**: Optional data package storage for debugging and troubleshooting +- **Worker Autoscaling**: Configurable poller behavior with aggressive autoscaling policies # Architecture @@ -33,6 +34,8 @@ The Scouter system uses a Temporal-based workflow architecture with clear separa ## 🔄 Workflows +The Scouter system implements a parent-child workflow pattern for data processing orchestration. + ### 1. Scouter Workflow (`scouter.py`) The **Scouter** workflow is the main entry point for data processing pipelines. It orchestrates the complete data ingestion process and implements a robust incremental data processing pattern. @@ -130,6 +133,8 @@ The **CoreScouter** workflow implements the core data processing pipeline for in "schema": "sensor_data", "table_name": "temperature_readings", "retention_time": 3600, + "fill_missing_tags": false, + "debug_data_package": false, "model_tags": { "Temperature": { "data_range": [-50, 150], @@ -140,67 +145,31 @@ The **CoreScouter** workflow implements the core data processing pipeline for in } ``` +**Additional Parameters:** +- `fill_missing_tags` (bool): Enable filling of missing tag values with default data +- `debug_data_package` (bool): Store raw and processed data packages in MongoDB for debugging + #### Architecture ```mermaid flowchart LR A[1. data_quality_gate] --> B[2. aggregate_data] --> C[3. group_and_hold_data] --> D[4. export_data_to_postgres] --> E[5. write_metrics] + E --> F{debug_data_package?} + F -->|yes| G[6. store_data_package] B -.-> Redis1[(Redis)] C -.-> Redis2[(Redis)] D -.-> PostgreSQL[(PostgreSQL)] E -.-> Metrics[Prometheus] + G -.-> MongoDB[(MongoDB)] ``` -### 3. FakeData Workflow (`fake_data.py`) +#### Debug Mode -The **FakeData** workflow generates synthetic industrial sensor data for testing and development purposes. It's designed to simulate realistic data flows without requiring actual industrial data sources. - -#### Purpose -- **Test Data Generation**: Creates realistic sensor data for development and testing -- **Pipeline Validation**: Tests data processing workflows with known data -- **Load Testing**: Generates configurable data volumes for performance testing -- **Demonstration**: Shows data flow patterns and processing capabilities - -#### Execution Flow -1. **Data Generation**: Creates synthetic sensor readings with realistic values -2. **Kafka Publishing**: Sends generated data to specified Kafka topics -3. **Quality Assurance**: Ensures data format consistency and completeness -4. **Monitoring**: Tracks generation and publishing metrics - -#### Key Features -- **Realistic Data**: Generates data within realistic industrial ranges -- **Configurable Volume**: Adjustable message counts for different testing scenarios -- **Random Variation**: Includes realistic data variations and occasional null values -- **Kafka Integration**: Direct integration with Kafka for data streaming -- **Error Handling**: Comprehensive error handling and logging - -#### Input Parameters -```json -{ - "topic": "test_sensor_data", - "metadata": {...}, - "num_messages": 100 -} -``` - -#### Architecture - -```mermaid -flowchart TB - subgraph workflow [" "] - A[1. generate_and_send_data] - end - - subgraph services [" "] - Kafka[(Kafka)] - end - - A -.-> Kafka - - style workflow fill:none,stroke:none - style services fill:none,stroke:none -``` +When `debug_data_package` is set to `true`, the workflow stores both raw and processed data packages in MongoDB for debugging and troubleshooting purposes. This is useful for: +- Investigating data processing issues +- Validating data transformations +- Auditing data quality gate decisions ## 📋 Prerequisites @@ -398,6 +367,36 @@ The Scouter system exposes comprehensive Prometheus metrics: | `KAFKA_BOOTSTRAP_SERVERS` | Kafka broker addresses | `localhost:9092` | No | | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No | | `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No | +| `PROJECT_NAME` | Project identifier for notifications | `scouter` | No | + +### Worker Configuration + +The worker supports advanced configuration for optimizing performance and latency: + +| Variable | Description | Default | Recommended | +|----------|-------------|---------|-------------| +| `MAX_CONCURRENT_WORKFLOW_TASKS` | Maximum concurrent workflow tasks | `200` | 100-500 | +| `MAX_CONCURRENT_ACTIVITIES` | Maximum concurrent activities | `200` | 100-500 | +| `MAX_CONCURRENT_LOCAL_ACTIVITIES` | Maximum concurrent local activities | `200` | 100-500 | +| `MAX_CACHED_WORKFLOWS` | Maximum cached workflow instances | `200` | 100-500 | + +### Poller Autoscaling Configuration + +The worker implements aggressive autoscaling policies for workflow and activity pollers: + +**Workflow Poller Behavior:** +| Variable | Description | Default | +|----------|-------------|---------| +| `WORKFLOW_POLLER_BEHAVIUR_MINIMUM` | Minimum workflow pollers | `10` | +| `WORKFLOW_POLLER_BEHAVIUR_INITIAL` | Initial workflow pollers | `100` | +| `WORKFLOW_POLLER_BEHAVIUR_MAXIMUM` | Maximum workflow pollers | `200` | + +**Activity Poller Behavior:** +| Variable | Description | Default | +|----------|-------------|---------| +| `ACTIVITY_POLLER_BEHAVIUR_MINIMUM` | Minimum activity pollers | `10` | +| `ACTIVITY_POLLER_BEHAVIUR_INITIAL` | Initial activity pollers | `100` | +| `ACTIVITY_POLLER_BEHAVIUR_MAXIMUM` | Maximum activity pollers | `200` | ### Workflow Configuration @@ -461,13 +460,11 @@ MongoDB pipeline configuration: scouter/ ├── activities/ # Temporal activity implementations │ ├── activities.py # Main activities orchestrator -│ ├── redis.py # Redis operations -│ ├── gates.py # Data quality gates -│ ├── mongodb.py # MongoDB operations -│ └── faker.py # Test data generation +│ ├── redis.py # Redis operations (caching, timestamps) +│ ├── gates.py # Data quality gates and filtering +│ └── mongodb.py # MongoDB operations (data loading) ├── workflow/ # Temporal workflow definitions │ ├── scouter.py # Main data ingestion workflow -│ ├── fake_data.py # Test data workflow │ └── sub_workflows/ # Sub-workflow implementations │ └── core_scouter.py # Core data processing workflow ├── worker/ # Worker implementation @@ -479,6 +476,21 @@ scouter/ └── __init__.py ``` +### Activity Implementations + +The Activities class combines multiple service classes through multiple inheritance: + +- **Postgres** (from sientia-dataops-library): PostgreSQL data export and persistence +- **Redis**: Timestamp management, data caching, and temporary storage +- **Gates**: Data quality validation and filtering logic +- **MongoDB**: Data loading from raw collections + +All activities support: +- Comprehensive logging and error handling +- Notification integration for errors and alerts +- Prometheus metrics collection +- Graceful shutdown and resource cleanup + ### Adding New Features 1. **Follow Temporal patterns** for new workflows and activities @@ -522,17 +534,19 @@ export LOG_LEVEL=DEBUG ### Key Parameters -- **Worker Concurrency**: Adjust `max_concurrent_workflow_tasks` and `max_concurrent_activities` -- **Connection Pools**: Optimize database connection pool sizes -- **Data Retention**: Configure Redis TTL based on processing requirements -- **Batch Sizes**: Adjust data processing batch sizes for optimal throughput +- **Worker Concurrency**: Adjust `MAX_CONCURRENT_WORKFLOW_TASKS` and `MAX_CONCURRENT_ACTIVITIES` (default: 200) +- **Poller Autoscaling**: Configure minimum, initial, and maximum poller counts for optimal throughput +- **Connection Pools**: Optimize database connection pool sizes (configured in `build_*_config()` functions) +- **Data Retention**: Configure Redis TTL via `retention_time` parameter (in seconds) +- **Workflow Caching**: Set `MAX_CACHED_WORKFLOWS` to balance memory usage and performance ### Scaling Considerations -- **Horizontal Scaling**: Deploy multiple worker instances -- **Task Queue Distribution**: Use multiple task queues for different workflow types -- **Database Performance**: Optimize indexes and connection pooling -- **Kafka Partitioning**: Configure appropriate partition counts for data ingestion +- **Horizontal Scaling**: Deploy multiple worker instances (each registers to `scouter-queue`) +- **Poller Autoscaling**: Workers implement aggressive autoscaling (10-200 pollers) for latency optimization +- **Database Performance**: Connection pooling is configured in `utils/connectors_config.py` +- **Worker Placement**: Use pod anti-affinity rules in Kubernetes for optimal distribution +- **Resource Limits**: Configure appropriate CPU/memory limits based on concurrency settings ## 🤝 Contributing diff --git a/values.yaml b/values.yaml index b88a2a2..bb44c0d 100644 --- a/values.yaml +++ b/values.yaml @@ -211,27 +211,27 @@ env: # Temporal worker tuning - name: MAX_CONCURRENT_WORKFLOW_TASKS - value: 200 + value: "200" - name: MAX_CONCURRENT_ACTIVITIES - value: 200 + value: "200" - name: MAX_CONCURRENT_LOCAL_ACTIVITIES - value: 200 + value: "200" - name: MAX_CACHED_WORKFLOWS - value: 200 + value: "200" - name: WORKFLOW_POLLER_BEHAVIUR_MINIMUM - value: 10 + value: "10" - name: WORKFLOW_POLLER_BEHAVIUR_INITIAL - value: 100 + value: "100" - name: WORKFLOW_POLLER_BEHAVIUR_MAXIMUM - value: 200 + value: "200" - name: ACTIVITY_POLLER_BEHAVIUR_MINIMUM - value: 10 + value: "10" - name: ACTIVITY_POLLER_BEHAVIUR_INITIAL - value: 100 + value: "100" - name: ACTIVITY_POLLER_BEHAVIUR_MAXIMUM - value: 200 + value: "200" ssh: From c538f4348875c262cdc388ce28da73cb9657b3c5 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Dec 2025 10:45:01 -0300 Subject: [PATCH 04/11] SIENTIAPDE-1441 Update requirements.txt to remove unused dependencies: asyncua and aiokafka, streamlining the project requirements. --- requirements.txt | 3 --- 1 file changed, 3 deletions(-) diff --git a/requirements.txt b/requirements.txt index 1dc88da..20b3d68 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,10 +1,7 @@ temporalio psycopg2-binary sqlalchemy -asyncua redis -aiokafka pymongo git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3 -pydruid[pandas] prometheus-client \ No newline at end of file From d3555221cf22e176457c6014bc24f918f2807781 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Dec 2025 13:24:23 -0300 Subject: [PATCH 05/11] SIENTIAPDE-1441 Update values.yaml to define resource limits and requests for improved resource management in Kubernetes deployments. --- values.yaml | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/values.yaml b/values.yaml index bb44c0d..7d4109b 100644 --- a/values.yaml +++ b/values.yaml @@ -52,7 +52,7 @@ securityContext: {} # runAsUser: 1000 -resources: {} +resources: # We usually recommend not to specify default resources and to leave this as a conscious # choice for the user. This also increases chances charts run on environments with little # resources, such as Minikube. If you do want to specify resources, uncomment the following @@ -64,6 +64,13 @@ resources: {} # cpu: 100m # memory: 128Mi + limits: + cpu: 1000m + memory: 2048Mi + requests: + cpu: 300m + memory: 512Mi + # This is to setup the liveness and readiness probes more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/ livenessProbe: exec: From 39494b90ac3cc969a4bb58c5ce44bc8e3db2422e Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Dec 2025 13:40:54 -0300 Subject: [PATCH 06/11] SIENTIAPDE-1441 Update requirements.txt to upgrade sientia-dataops-library dependency from version 1.5.3 to 1.6.1 for improved functionality. --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 20b3d68..f9857c6 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,5 +3,5 @@ psycopg2-binary sqlalchemy redis pymongo -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.6.1 prometheus-client \ No newline at end of file From 14eea70090bed135c4cad763f6e95916b9bffae7 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Dec 2025 14:06:39 -0300 Subject: [PATCH 07/11] SIENTIAPDE-1441 Update values.yaml to modify liveness and readiness probe configurations, enhancing application health checks with improved metrics validation and adjusted timing parameters for better responsiveness. --- values.yaml | 28 ++++++++++++++++++++++------ 1 file changed, 22 insertions(+), 6 deletions(-) diff --git a/values.yaml b/values.yaml index 7d4109b..ce700c0 100644 --- a/values.yaml +++ b/values.yaml @@ -77,18 +77,34 @@ livenessProbe: command: - sh - -c - - pgrep -f "scouter.worker.worker" - initialDelaySeconds: 20 - periodSeconds: 30 + - | + APP_UP=$(curl -s http://localhost:9090/metrics | grep '^app_up{' | grep -o '[0-9]' | head -1) + if [ "$APP_UP" = "1" ]; then + exit 0 + else + exit 1 + fi + initialDelaySeconds: 30 + periodSeconds: 15 + timeoutSeconds: 5 + failureThreshold: 3 readinessProbe: exec: command: - sh - -c - - pgrep -f "scouter.worker.worker" - initialDelaySeconds: 10 - periodSeconds: 15 + - | + APP_UP=$(curl -s http://localhost:9090/metrics | grep '^app_up{' | grep -o '[0-9]' | head -1) + if [ "$APP_UP" = "1" ]; then + exit 0 + else + exit 1 + fi + initialDelaySeconds: 20 + periodSeconds: 10 + timeoutSeconds: 3 + failureThreshold: 2 # This section is for setting up autoscaling more information can be found here: https://kubernetes.io/docs/concepts/workloads/autoscaling/ From f1cdf7471d1c3a5a1eae2298b790ce7f92e5d901 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Dec 2025 15:39:54 -0300 Subject: [PATCH 08/11] SIENTIAPDE-1441 SIENTIAPDE-1441: Refactor worker configuration in worker.py to utilize a new prepare_worker function for cleaner setup. Update environment variable names in values.yaml for consistency and clarity, adjusting memory requests and enhancing health check commands for improved application monitoring. --- scouter/worker/prepare_worker.py | 60 ++++++++++++++++++++++++++++++++ scouter/worker/worker.py | 24 ++++--------- values.yaml | 36 +++++++------------ 3 files changed, 79 insertions(+), 41 deletions(-) create mode 100644 scouter/worker/prepare_worker.py diff --git a/scouter/worker/prepare_worker.py b/scouter/worker/prepare_worker.py new file mode 100644 index 0000000..6f1391f --- /dev/null +++ b/scouter/worker/prepare_worker.py @@ -0,0 +1,60 @@ +from typing import Sequence, Type, Any + +from temporalio.worker import Worker, PollerBehaviorAutoscaling +from temporalio.client import Client + +import os + + +parameters = [ + ('MAX_CONCURRENT_WORKFLOW_TASKS', '200'), + ('MAX_CONCURRENT_ACTIVITIES', '200'), + ('MAX_CONCURRENT_LOCAL_ACTIVITIES', '200'), + ('MAX_CACHED_WORKFLOWS', '200'), + + ('WORKFLOW_POLLER_BEHAVIUR_MINIMUM', '10'), + ('WORKFLOW_POLLER_BEHAVIUR_INITIAL', '100'), + ('WORKFLOW_POLLER_BEHAVIUR_MAXIMUM', '200'), + ('ACTIVITY_POLLER_BEHAVIUR_MINIMUM', '10'), + ('ACTIVITY_POLLER_BEHAVIUR_INITIAL', '100'), + ('ACTIVITY_POLLER_BEHAVIUR_MAXIMUM', '200'), +] + +def prepare_worker( + main_workflow: Type, + other_workflows: Sequence[Type], + activities: Sequence[Any], + temporal_client: Client, +): + + main_workflow_name = main_workflow.__name__.upper() + + local_workflow_parameters = {} + + for parameter in parameters: + local_workflow_parameters[parameter[0]] = int( + os.getenv(main_workflow_name + '_' + parameter[0], parameter[1])) + + + return [ + Worker( + temporal_client, + task_queue='scouter-queue', + workflows=[main_workflow, *other_workflows], + activities=[*activities], + max_concurrent_workflow_tasks=local_workflow_parameters['MAX_CONCURRENT_WORKFLOW_TASKS'], + max_concurrent_activities=local_workflow_parameters['MAX_CONCURRENT_ACTIVITIES'], + max_concurrent_local_activities=local_workflow_parameters['MAX_CONCURRENT_LOCAL_ACTIVITIES'], + max_cached_workflows=local_workflow_parameters['MAX_CACHED_WORKFLOWS'], + workflow_task_poller_behavior=PollerBehaviorAutoscaling( + minimum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MINIMUM'], + initial=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_INITIAL'], + maximum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MAXIMUM'], + ), + activity_task_poller_behavior=PollerBehaviorAutoscaling( + minimum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MINIMUM'], + initial=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_INITIAL'], + maximum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MAXIMUM'], + ), + ) + ] \ No newline at end of file diff --git a/scouter/worker/worker.py b/scouter/worker/worker.py index 228562c..d5f6be8 100644 --- a/scouter/worker/worker.py +++ b/scouter/worker/worker.py @@ -2,6 +2,8 @@ from temporalio import client, workflow from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig from temporalio.worker import PollerBehaviorAutoscaling, Worker +from scouter.worker.prepare_worker import prepare_worker + with workflow.unsafe.imports_passed_through(): import asyncio import os @@ -121,10 +123,10 @@ async def main(): logger.custom_info('Starting Workers...', metadata) workers = [ - Worker( - temporal_client, - task_queue='scouter-queue', - workflows=[Scouter, CoreScouter], + prepare_worker( + temporal_client=temporal_client, + main_workflow=Scouter, + other_workflows=[CoreScouter], activities=[ activities.load_latest_data, activities.get_last_data_timestamp, @@ -136,20 +138,6 @@ async def main(): activities.write_metrics, activities.store_data_package, ], - max_concurrent_workflow_tasks=MAX_CONCURRENT_WORKFLOW_TASKS, - max_concurrent_activities=MAX_CONCURRENT_ACTIVITIES, - max_concurrent_local_activities=MAX_CONCURRENT_LOCAL_ACTIVITIES, - max_cached_workflows=MAX_CACHED_WORKFLOWS, - workflow_task_poller_behavior=PollerBehaviorAutoscaling( - minimum=WORKFLOW_POLLER_BEHAVIUR_MINIMUM, - initial=WORKFLOW_POLLER_BEHAVIUR_INITIAL, - maximum=WORKFLOW_POLLER_BEHAVIUR_MAXIMUM, - ), - activity_task_poller_behavior=PollerBehaviorAutoscaling( - minimum=ACTIVITY_POLLER_BEHAVIUR_MINIMUM, - initial=ACTIVITY_POLLER_BEHAVIUR_INITIAL, - maximum=ACTIVITY_POLLER_BEHAVIUR_MAXIMUM, - ), ) ] diff --git a/values.yaml b/values.yaml index ce700c0..56d9b71 100644 --- a/values.yaml +++ b/values.yaml @@ -69,7 +69,7 @@ resources: memory: 2048Mi requests: cpu: 300m - memory: 512Mi + memory: 256Mi # This is to setup the liveness and readiness probes more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/ livenessProbe: @@ -78,12 +78,7 @@ livenessProbe: - sh - -c - | - APP_UP=$(curl -s http://localhost:9090/metrics | grep '^app_up{' | grep -o '[0-9]' | head -1) - if [ "$APP_UP" = "1" ]; then - exit 0 - else - exit 1 - fi + curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1' initialDelaySeconds: 30 periodSeconds: 15 timeoutSeconds: 5 @@ -95,12 +90,7 @@ readinessProbe: - sh - -c - | - APP_UP=$(curl -s http://localhost:9090/metrics | grep '^app_up{' | grep -o '[0-9]' | head -1) - if [ "$APP_UP" = "1" ]; then - exit 0 - else - exit 1 - fi + curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1' initialDelaySeconds: 20 periodSeconds: 10 timeoutSeconds: 3 @@ -233,27 +223,27 @@ env: value: "http://library-distribution-server.library.svc.cluster.local:5000" # Temporal worker tuning - - name: MAX_CONCURRENT_WORKFLOW_TASKS + - name: SCOUTER_MAX_CONCURRENT_WORKFLOW_TASKS value: "200" - - name: MAX_CONCURRENT_ACTIVITIES + - name: SCOUTER_MAX_CONCURRENT_ACTIVITIES value: "200" - - name: MAX_CONCURRENT_LOCAL_ACTIVITIES + - name: SCOUTER_MAX_CONCURRENT_LOCAL_ACTIVITIES value: "200" - - name: MAX_CACHED_WORKFLOWS + - name: SCOUTER_MAX_CACHED_WORKFLOWS value: "200" - - name: WORKFLOW_POLLER_BEHAVIUR_MINIMUM + - name: SCOUTER_WORKFLOW_POLLER_BEHAVIUR_MINIMUM value: "10" - - name: WORKFLOW_POLLER_BEHAVIUR_INITIAL + - name: SCOUTER_WORKFLOW_POLLER_BEHAVIUR_INITIAL value: "100" - - name: WORKFLOW_POLLER_BEHAVIUR_MAXIMUM + - name: SCOUTER_WORKFLOW_POLLER_BEHAVIUR_MAXIMUM value: "200" - - name: ACTIVITY_POLLER_BEHAVIUR_MINIMUM + - name: SCOUTER_ACTIVITY_POLLER_BEHAVIUR_MINIMUM value: "10" - - name: ACTIVITY_POLLER_BEHAVIUR_INITIAL + - name: SCOUTER_ACTIVITY_POLLER_BEHAVIUR_INITIAL value: "100" - - name: ACTIVITY_POLLER_BEHAVIUR_MAXIMUM + - name: SCOUTER_ACTIVITY_POLLER_BEHAVIUR_MAXIMUM value: "200" From 5e2403090b5716736d71bce15a0435acf3b3f36c Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Dec 2025 15:43:01 -0300 Subject: [PATCH 09/11] SIENTIAPDE-1441 Refactor prepare_worker function to return a single Worker instance instead of a list, streamlining worker configuration and improving code clarity. --- scouter/worker/prepare_worker.py | 44 +++++++++++++++----------------- 1 file changed, 21 insertions(+), 23 deletions(-) diff --git a/scouter/worker/prepare_worker.py b/scouter/worker/prepare_worker.py index 6f1391f..b261f68 100644 --- a/scouter/worker/prepare_worker.py +++ b/scouter/worker/prepare_worker.py @@ -25,7 +25,7 @@ def prepare_worker( other_workflows: Sequence[Type], activities: Sequence[Any], temporal_client: Client, -): +) -> Worker: main_workflow_name = main_workflow.__name__.upper() @@ -36,25 +36,23 @@ def prepare_worker( os.getenv(main_workflow_name + '_' + parameter[0], parameter[1])) - return [ - Worker( - temporal_client, - task_queue='scouter-queue', - workflows=[main_workflow, *other_workflows], - activities=[*activities], - max_concurrent_workflow_tasks=local_workflow_parameters['MAX_CONCURRENT_WORKFLOW_TASKS'], - max_concurrent_activities=local_workflow_parameters['MAX_CONCURRENT_ACTIVITIES'], - max_concurrent_local_activities=local_workflow_parameters['MAX_CONCURRENT_LOCAL_ACTIVITIES'], - max_cached_workflows=local_workflow_parameters['MAX_CACHED_WORKFLOWS'], - workflow_task_poller_behavior=PollerBehaviorAutoscaling( - minimum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MINIMUM'], - initial=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_INITIAL'], - maximum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MAXIMUM'], - ), - activity_task_poller_behavior=PollerBehaviorAutoscaling( - minimum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MINIMUM'], - initial=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_INITIAL'], - maximum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MAXIMUM'], - ), - ) - ] \ No newline at end of file + return Worker( + temporal_client, + task_queue='scouter-queue', + workflows=[main_workflow, *other_workflows], + activities=[*activities], + max_concurrent_workflow_tasks=local_workflow_parameters['MAX_CONCURRENT_WORKFLOW_TASKS'], + max_concurrent_activities=local_workflow_parameters['MAX_CONCURRENT_ACTIVITIES'], + max_concurrent_local_activities=local_workflow_parameters['MAX_CONCURRENT_LOCAL_ACTIVITIES'], + max_cached_workflows=local_workflow_parameters['MAX_CACHED_WORKFLOWS'], + workflow_task_poller_behavior=PollerBehaviorAutoscaling( + minimum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MINIMUM'], + initial=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_INITIAL'], + maximum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MAXIMUM'], + ), + activity_task_poller_behavior=PollerBehaviorAutoscaling( + minimum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MINIMUM'], + initial=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_INITIAL'], + maximum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MAXIMUM'], + ), + ) \ No newline at end of file From 26eed05f748df90b546325f95e17d588bd920b93 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Dec 2025 15:51:31 -0300 Subject: [PATCH 10/11] SIENTIAPDE-1441 Update sonar-project.properties to include prepare_worker.py in coverage exclusions, enhancing code quality metrics. --- sonar-project.properties | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sonar-project.properties b/sonar-project.properties index ca79a27..f98e3ab 100644 --- a/sonar-project.properties +++ b/sonar-project.properties @@ -5,7 +5,7 @@ sonar.tests=tests sonar.qualitygate.wait=true sonar.qualitygate.timeout=300 sonar.python.coverage.reportPaths=coverage.xml -sonar.coverage.exclusions=scouter/worker/worker.py +sonar.coverage.exclusions=scouter/worker/worker.py,scouter/worker/prepare_worker.py sonar.python.xunit.reportPath=pytest.xml sonar.python.version=3.11 sonar.projectVersion=1.0.0 From 666a0a6e13d76f2fca016c3c8795fb7872cdef56 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 16 Dec 2025 16:01:47 -0300 Subject: [PATCH 11/11] SIENTIAPDE-1441 Refactor prepare_worker.py to improve type annotations and code clarity. Update import statements and enhance formatting for better readability. Adjust comments in worker.py for consistency. --- scouter/worker/prepare_worker.py | 26 +++++++++++++------------- scouter/worker/worker.py | 3 +-- 2 files changed, 14 insertions(+), 15 deletions(-) diff --git a/scouter/worker/prepare_worker.py b/scouter/worker/prepare_worker.py index b261f68..09fced1 100644 --- a/scouter/worker/prepare_worker.py +++ b/scouter/worker/prepare_worker.py @@ -1,17 +1,15 @@ -from typing import Sequence, Type, Any - -from temporalio.worker import Worker, PollerBehaviorAutoscaling -from temporalio.client import Client - import os +from collections.abc import Sequence +from typing import Any +from temporalio.client import Client +from temporalio.worker import PollerBehaviorAutoscaling, Worker parameters = [ ('MAX_CONCURRENT_WORKFLOW_TASKS', '200'), ('MAX_CONCURRENT_ACTIVITIES', '200'), ('MAX_CONCURRENT_LOCAL_ACTIVITIES', '200'), ('MAX_CACHED_WORKFLOWS', '200'), - ('WORKFLOW_POLLER_BEHAVIUR_MINIMUM', '10'), ('WORKFLOW_POLLER_BEHAVIUR_INITIAL', '100'), ('WORKFLOW_POLLER_BEHAVIUR_MAXIMUM', '200'), @@ -20,21 +18,21 @@ parameters = [ ('ACTIVITY_POLLER_BEHAVIUR_MAXIMUM', '200'), ] + def prepare_worker( - main_workflow: Type, - other_workflows: Sequence[Type], + main_workflow: type, + other_workflows: Sequence[type], activities: Sequence[Any], temporal_client: Client, ) -> Worker: - main_workflow_name = main_workflow.__name__.upper() local_workflow_parameters = {} for parameter in parameters: local_workflow_parameters[parameter[0]] = int( - os.getenv(main_workflow_name + '_' + parameter[0], parameter[1])) - + os.getenv(main_workflow_name + '_' + parameter[0], parameter[1]) + ) return Worker( temporal_client, @@ -43,7 +41,9 @@ def prepare_worker( activities=[*activities], max_concurrent_workflow_tasks=local_workflow_parameters['MAX_CONCURRENT_WORKFLOW_TASKS'], max_concurrent_activities=local_workflow_parameters['MAX_CONCURRENT_ACTIVITIES'], - max_concurrent_local_activities=local_workflow_parameters['MAX_CONCURRENT_LOCAL_ACTIVITIES'], + max_concurrent_local_activities=local_workflow_parameters[ + 'MAX_CONCURRENT_LOCAL_ACTIVITIES' + ], max_cached_workflows=local_workflow_parameters['MAX_CACHED_WORKFLOWS'], workflow_task_poller_behavior=PollerBehaviorAutoscaling( minimum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MINIMUM'], @@ -55,4 +55,4 @@ def prepare_worker( initial=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_INITIAL'], maximum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MAXIMUM'], ), - ) \ No newline at end of file + ) diff --git a/scouter/worker/worker.py b/scouter/worker/worker.py index d5f6be8..cd333f4 100644 --- a/scouter/worker/worker.py +++ b/scouter/worker/worker.py @@ -1,6 +1,5 @@ from temporalio import client, workflow from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig -from temporalio.worker import PollerBehaviorAutoscaling, Worker from scouter.worker.prepare_worker import prepare_worker @@ -29,7 +28,7 @@ SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091')) # For optmized latency, Temporal docs recommends fixed slots, ensuring -# high concurency levels. +# high concurency levels. MAX_CONCURRENT_WORKFLOW_TASKS = int(os.getenv('MAX_CONCURRENT_WORKFLOW_TASKS', '200')) MAX_CONCURRENT_ACTIVITIES = int(os.getenv('MAX_CONCURRENT_ACTIVITIES', '200'))