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

Merge pull request #7 from Aignosi/SIENTIAPDE-1110-criar-testes-e-2-e

Browse Source 
Sientiapde 1110 criar testes e 2 e
This commit is contained in:
Matheus Demoner
2025-07-04 16:12:34 -03:00
committed by GitHub
parent 99d80020d0 08240efc8b
commit e5ffcd0757
22 changed files with 6829 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1568
input_samples_30.json Normal file
View File

File diff suppressed because it is too large Load Diff

3
requirements.txt
View File

@@ -3,5 +3,8 @@ psycopg2-binary
sqlalchemy sqlalchemy
asyncua asyncua
redis redis
aiokafka
pymongo
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.2.0 git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.2.0
git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.1 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.1
pydruid[pandas]

29
scouter/activities/activities.py
View File

@@ -1,22 +1,22 @@
from temporalio import activity, workflow from temporalio import workflow
with workflow.unsafe.imports_passed_through(): with workflow.unsafe.imports_passed_through():
from sientia_do.temporal.activities.postgres import Postgres from sientia_do.temporal.activities.postgres import Postgres
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.temporal.utils.logger import Logger from sientia_do.temporal.utils.logger import Logger
from scouter.activities.redis import Redis from scouter.activities.redis import Redis
from scouter.activities.kafka import Kafka
from scouter.activities.gates import Gates from scouter.activities.gates import Gates
from scouter.activities.mongodb import MongoDB
from typing import Any from typing import Any
class Activities(Postgres, Redis, Kafka, Gates): class Activities(Postgres, Redis, Gates, MongoDB,):
"""Activities class that combines multiple services with proper initialization.""" """Activities class that combines multiple services with proper initialization."""
def __init__(self, def __init__(self,
postgres_config: dict[str, Any], postgres_config: dict[str, Any],
redis_config: dict[str, Any], redis_config: dict[str, Any],
kafka_config: dict[str, Any], mongodb_config: dict[str, Any],
logger: Logger, logger: Logger,
notification_handler: NotificationHandler): notification_handler: NotificationHandler):
@@ -45,16 +45,6 @@ class Activities(Postgres, Redis, Kafka, Gates):
password=redis_config['password'] password=redis_config['password']
) )
# Initialize Kafka
Kafka.__init__(
self,
bootstrap_servers=kafka_config['bootstrap_servers'],
polling_time=kafka_config['polling_time'],
group_id=kafka_config['group_id'],
logger=logger,
notification_handler=notification_handler
)
# Initialize Gates # Initialize Gates
Gates.__init__( Gates.__init__(
self, self,
@@ -62,6 +52,15 @@ class Activities(Postgres, Redis, Kafka, Gates):
notification_handler=notification_handler notification_handler=notification_handler
) )
# Initialize MongoDB
MongoDB.__init__(
self,
connection_string=mongodb_config['connection_string'],
database_name=mongodb_config['database_name'],
logger=logger,
notification_handler=notification_handler
)
def shutdown(self): def shutdown(self):
Postgres.close(self) Postgres.close(self)
Kafka.close(self) MongoDB.shutdown(self)

20
scouter/activities/gates.py
View File

@@ -16,7 +16,8 @@ quality_gate_filters = {
class Gates(BaseActivity): class Gates(BaseActivity):
def apply_aggregation(self, group: DataFrame, aggr_function: str) -> float | None | str: def apply_aggregation(self, group: DataFrame, aggr_function: str,
metadata: dict[str, Any]) -> float | None | str:
""" """
Apply aggregation function to a group of data. Apply aggregation function to a group of data.
@@ -48,7 +49,8 @@ class Gates(BaseActivity):
elif aggr_function == 'min': elif aggr_function == 'min':
return group['value'].min() return group['value'].min()
else: else:
self.notification_handler.build_and_send_notification( self.send_notification(
metadata=metadata,
notification_id="AGGREGATION_ISSUES", notification_id="AGGREGATION_ISSUES",
message=f"Invalid aggregation function: {aggr_function}", message=f"Invalid aggregation function: {aggr_function}",
block="aggregate_data", block="aggregate_data",
@@ -93,14 +95,15 @@ class Gates(BaseActivity):
for (tag, name), group in grouped: for (tag, name), group in grouped:
# Get the aggregation function from model_tags # Get the aggregation function from model_tags
aggr_function = input_data['model_tags'].get( aggr_function = input_data['model_tags'].get(
name, {}).get('aggr_function', 'lts') name, {}).get('aggr_func', 'lts')
group.sort_values(by='timestamp', inplace=True) group.sort_values(by='timestamp', inplace=True)
# Get the latest timestamp # Get the latest timestamp
latest_timestamp = group['timestamp'].max() latest_timestamp = group['timestamp'].max()
aggr_value = self.apply_aggregation(group, aggr_function) aggr_value = self.apply_aggregation(
group, aggr_function, metadata)
if aggr_value == 'continue': if aggr_value == 'continue':
continue continue
@@ -145,7 +148,8 @@ class Gates(BaseActivity):
except Exception as e: except Exception as e:
trace = traceback.format_exc() trace = traceback.format_exc()
self.notification_handler.build_and_send_notification( self.send_notification(
metadata=metadata,
notification_id="AGGREGATION_ISSUES", notification_id="AGGREGATION_ISSUES",
message=f"Error aggregating data: {e}", message=f"Error aggregating data: {e}",
block="aggregate_data", block="aggregate_data",
@@ -202,7 +206,8 @@ class Gates(BaseActivity):
except Exception as e: except Exception as e:
trace = traceback.format_exc() trace = traceback.format_exc()
self.notification_handler.build_and_send_notification( self.send_notification(
metadata=metadata,
notification_id="DATA_QUALITY_GATE_ISSUES", notification_id="DATA_QUALITY_GATE_ISSUES",
message=f"Error applying filter {filter_name}: {e}", message=f"Error applying filter {filter_name}: {e}",
block="data_quality_gate", block="data_quality_gate",
@@ -219,7 +224,8 @@ class Gates(BaseActivity):
message = f"{len(filtered_data)} rows has quality issues: {filter_name}: {policy}" message = f"{len(filtered_data)} rows has quality issues: {filter_name}: {policy}"
attachment = filtered_data.to_string() attachment = filtered_data.to_string()
self.notification_handler.build_and_send_notification( self.send_notification(
metadata=metadata,
notification_id=f"DATA_QUALITY_GATE_ISSUES__{filter_name}", notification_id=f"DATA_QUALITY_GATE_ISSUES__{filter_name}",
message=message, message=message,
block="data_quality_gate", block="data_quality_gate",

91
scouter/activities/kafka.py
View File

@@ -1,91 +0,0 @@
from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.temporal.utils.logger import Logger
from typing import Any
from kafka import KafkaConsumer
from pandas import DataFrame
import json
class Kafka(BaseActivity):
def __init__(self, bootstrap_servers: str, polling_time: int,
group_id: str, logger: Logger, notification_handler: NotificationHandler):
self.polling_time = polling_time
self.kafka_connector = KafkaConsumer(
bootstrap_servers=bootstrap_servers,
auto_offset_reset="earliest",
enable_auto_commit=True,
group_id=group_id,
value_deserializer=lambda x: json.loads(x.decode("utf-8"))
)
BaseActivity.__init__(self, logger, notification_handler)
def close(self):
"""Closes the connector connection."""
self.info("Closing Kafka connector...")
self.kafka_connector.close()
def __del__(self):
self.close()
@activity.defn(name="load_from_kafka")
async def load_from_kafka(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Loads data from a kafka topic. Polls the topic for a given time and returns the data.
Args:
input_data (dict[str, Any]): The data to load. Contains:
topic (str): The topic to load data from.
Returns:
dict[str, Any]: The data loaded from the topic.
"""
metadata = input_data['metadata']
self.debug(
f"Loading data from topic: {input_data['topic']}",
metadata=metadata
)
topic = input_data["topic"]
# Subscribe to the specified topic
self.kafka_connector.subscribe([topic])
# List to store message values
message_values = []
# Poll for messages
records = self.kafka_connector.poll(timeout_ms=self.polling_time)
self.debug(
f"Polled {len(records)} records from topic: {topic}",
metadata=metadata
)
# Process the polled records
for _topic_partition, msgs in records.items():
for msg in msgs:
message_values.append(msg.value)
# Return empty dict if no messages were received
if not message_values:
return {}
self.debug(
f"Loaded {len(message_values)} messages from topic: {topic}",
metadata=metadata
)
self.debug(
f"Loaded data: {message_values}",
metadata=metadata
)
return DataFrame(message_values).to_dict()

140
scouter/activities/mongodb.py Normal file
View File

@@ -0,0 +1,140 @@
from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
from typing import Any
import traceback
from datetime import datetime
from pymongo import MongoClient
from pandas import DataFrame
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.temporal.utils.logger import Logger
def clear_mongo_id(docs: list) -> list:
"""
Remove the MongoDB internal `_id` field from the document.
Args:
docs (list): The document to clear.
Returns:
list: The documents without the `_id` field.
"""
for doc in docs:
if isinstance(doc, list):
clear_mongo_id(doc)
elif isinstance(doc, dict):
if "_id" in doc:
del doc["_id"]
for key, value in doc.items():
if isinstance(value, list):
clear_mongo_id(value)
elif isinstance(value, dict):
clear_mongo_id([value])
return docs
class MongoDB(BaseActivity):
def __init__(self, connection_string: str, database_name: str,
logger: Logger,
notification_handler: NotificationHandler):
self.connection_string = connection_string
self.database_name = database_name
self.client = MongoClient(
self.connection_string, serverSelectionTimeoutMS=5000)
self.client.server_info() # Trigger an exception if connection fails
self.database = self.client[self.database_name]
# Initialize MongoDB client here (omitted for brevity)
logger.info("MongoDB connection initialized")
BaseActivity.__init__(self,
logger=logger,
notification_handler=notification_handler)
def shutdown(self):
"""
Close the MongoDB client connection.
"""
try:
if self.client:
self.logger.info("Closing MongoDB connection...")
self.client.close()
self.logger.info("MongoDB connection closed successfully")
except Exception as e:
self.logger.error(f"Failed to close MongoDB connection: {e}")
def __del__(self):
"""
Destructor to ensure MongoDB client is closed when the object is deleted.
"""
self.shutdown()
@activity.defn(name="load_latest_data")
async def load_latest_data(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Loads the latest data from MongoDB.
"""
metadata = input_data['metadata']
collection_name = input_data['collection_name']
last_data_timestamp = input_data['last_data_timestamp']
self.debug(
f"Loading data from MongoDB: {input_data}",
metadata=metadata
)
try:
if last_data_timestamp is None:
data_filter = {}
else:
data_filter = {
"inserted_at": {
"$gt": datetime.strptime(last_data_timestamp, "%Y-%m-%d %H:%M:%S.%f")
}
}
self.debug(
f"Data filter: {data_filter}",
metadata=metadata
)
data = list(self.database[collection_name].find(
data_filter, {"_id": 0}))
data = clear_mongo_id(data)
for item in data:
item['inserted_at'] = item['inserted_at'].strftime(
"%Y-%m-%d %H:%M:%S.%f")
self.info(
f"Loaded {len(data)} documents from MongoDB",
metadata=metadata
)
self.debug(
f"Loaded data: {data}",
metadata=metadata
)
return DataFrame(data).to_dict()
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=metadata,
notification_id="MONGO_LOAD_ERROR",
message=f"Error loading data from MongoDB: {e}",
block="load_latest_data",
level=NotificationLevel.ERROR,
attachment_content=trace
)
raise e

120
scouter/activities/redis.py
View File

@@ -1,8 +1,10 @@
import traceback
from temporalio import workflow, activity from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through(): with workflow.unsafe.imports_passed_through():
from logging import Logger from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.redis_base import Redis as RedisBase from sientia_do.temporal.activities.redis_base import Redis as RedisBase
from sientia_do.temporal.utils.logger import Logger from sientia_do.temporal.utils.logger import Logger
from typing import Any from typing import Any
@@ -18,6 +20,76 @@ class Redis(RedisBase):
RedisBase.__init__(self, host, port, username, RedisBase.__init__(self, host, port, username,
password, logger, notification_handler) password, logger, notification_handler)
@activity.defn(name="get_last_data_timestamp")
async def get_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None:
"""
Gets the last data timestamp from redis.
"""
metadata = input_data['metadata']
key = f"last_data_timestamp_{input_data['workflow_name']}_{input_data['schedule_name']}"
try:
data_hold = self.get(key)
except Exception as e:
self.send_notification(
metadata=metadata,
notification_id="REDIS_GET_ERROR",
message=f"Error getting last data timestamp: {e}",
block="get_last_data_timestamp",
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc()
)
raise e
self.debug(
f"Last collected timestamp: {data_hold}",
metadata=metadata
)
if not data_hold:
return None
return data_hold
@activity.defn(name="put_last_data_timestamp")
async def put_last_data_timestamp(self, input_data: dict[str, Any]):
"""
Puts the last data timestamp into redis.
"""
metadata = input_data['metadata']
key = f"last_data_timestamp_{input_data['workflow_name']}_{input_data['schedule_name']}"
data = DataFrame(input_data['data'])
if data.empty:
self.warning("No data to insert",
metadata=metadata
)
return None
last_data_timestamp = data['inserted_at'].max()
self.debug(
f"Last collected timestamp to insert: {last_data_timestamp}",
metadata=metadata
)
try:
self.set(key, last_data_timestamp, ttl=None)
except Exception as e:
self.send_notification(
metadata=metadata,
notification_id="REDIS_SET_ERROR",
message=f"Error setting last data timestamp: {e}",
block="put_last_data_timestamp",
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc()
)
raise e
return last_data_timestamp
@activity.defn(name="group_and_hold_data") @activity.defn(name="group_and_hold_data")
async def group_and_hold_data(self, input_data: dict[str, Any]): async def group_and_hold_data(self, input_data: dict[str, Any]):
""" """
@@ -40,9 +112,20 @@ class Redis(RedisBase):
data = DataFrame(input_data['data']) data = DataFrame(input_data['data'])
retention_time = input_data['retention_time'] retention_time = input_data['retention_time']
key = f"{input_data['workflow_name']}_{input_data['schedule_name']}" key = f"held_data_{input_data['workflow_name']}_{input_data['schedule_name']}"
data_hold = self.get(key) try:
data_hold = self.get(key)
except Exception as e:
self.send_notification(
metadata=metadata,
notification_id="REDIS_GET_ERROR",
message=f"Error getting held data: {e}",
block="group_and_hold_data",
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc()
)
raise e
if not data_hold: if not data_hold:
data_hold = {} data_hold = {}
@@ -52,22 +135,33 @@ class Redis(RedisBase):
) )
return data_hold return data_hold
for _, row in data.iterrows(): try:
value = row['value'] for _, row in data.iterrows():
value = row['value']
data_hold[row['name']] = value data_hold[row['name']] = value
data_hold['timestamp'] = data['timestamp'].max() if not data.empty else \ data_hold['timestamp'] = data['timestamp'].max() if not data.empty else \
datetime.now().strftime("%Y-%m-%d %H:%M:%S") datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.set(key, data_hold, ttl=retention_time) self.set(key, data_hold, ttl=retention_time)
data_hold_df = DataFrame(data_hold, index=[0]) data_hold_df = DataFrame(data_hold, index=[0])
data_hold_melted = data_hold_df.melt( data_hold_melted = data_hold_df.melt(
id_vars='timestamp', var_name='variable', value_name='value') id_vars='timestamp', var_name='variable', value_name='value')
data_hold_melted['model_id'] = input_data['model_id'] data_hold_melted['model_id'] = input_data['model_id']
data_hold_melted.reset_index(drop=True, inplace=True) data_hold_melted.reset_index(drop=True, inplace=True)
except Exception as e:
self.send_notification(
metadata=metadata,
notification_id="REDIS_SET_ERROR",
message=f"Error setting held data: {e}",
block="group_and_hold_data",
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc()
)
raise e
self.debug( self.debug(
f"Data grouped and held successfully:\n {data_hold_melted.to_string()}", f"Data grouped and held successfully:\n {data_hold_melted.to_string()}",

19
scouter/utils/connectors_config.py
View File

@@ -28,3 +28,22 @@ def build_redis_config():
'username': getenv('REDIS_USERNAME', None), 'username': getenv('REDIS_USERNAME', None),
'password': getenv('REDIS_PASSWORD', None) 'password': getenv('REDIS_PASSWORD', None)
} }
def build_mongodb_config():
username = getenv('MONGODB_USERNAME', 'sientia')
password = getenv('MONGODB_PASSWORD', 'sientia')
uri = getenv('MONGODB_URL', 'localhost:27017')
connection_string = f'mongodb://{username}:{password}@{uri}'
return {
'connection_string': connection_string,
'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia')
}
def build_druid_config():
return {
'host': getenv('DRUID_HOST', 'localhost'),
'port': int(getenv('DRUID_PORT', '8082')),
}

13
scouter/worker/worker.py
View File

@@ -14,8 +14,8 @@ with workflow.unsafe.imports_passed_through():
import asyncio import asyncio
from scouter.utils.connectors_config import ( from scouter.utils.connectors_config import (
build_postgres_config, build_postgres_config,
build_kafka_config, build_redis_config,
build_redis_config build_mongodb_config
) )
@@ -40,8 +40,8 @@ async def main():
logger=logger, logger=logger,
notification_handler=notification_handler, notification_handler=notification_handler,
postgres_config=build_postgres_config(), postgres_config=build_postgres_config(),
kafka_config=build_kafka_config(), redis_config=build_redis_config(),
redis_config=build_redis_config() mongodb_config=build_mongodb_config()
) )
logger.info('Starting Faker Activities...') logger.info('Starting Faker Activities...')
@@ -68,12 +68,13 @@ async def main():
task_queue='scouter-queue', task_queue='scouter-queue',
workflows=[Scouter, CoreScouter], workflows=[Scouter, CoreScouter],
activities=[ activities=[
activities.load_from_kafka, activities.load_latest_data,
activities.get_last_data_timestamp,
activities.put_last_data_timestamp,
activities.data_quality_gate, activities.data_quality_gate,
activities.aggregate_data, activities.aggregate_data,
activities.group_and_hold_data, activities.group_and_hold_data,
activities.export_data_to_postgres, activities.export_data_to_postgres,
activities.prepare_activity,
] ]
), ),
Worker( Worker(

35
scouter/workflow/scouter.py
View File

@@ -41,20 +41,45 @@ class Scouter:
} }
} }
data = await workflow.execute_activity_method( last_data_timestamp = await workflow.execute_local_activity_method(
Activities.load_from_kafka, Activities.get_last_data_timestamp,
{ {
**metadata, **metadata,
'topic': input_data['topic'] 'workflow_name': input_data['workflow_name'],
'schedule_name': input_data['schedule_name']
}, },
retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60),
start_to_close_timeout=timedelta(seconds=60) retry_policy=retry_policy
)
data = await workflow.execute_local_activity_method(
Activities.load_latest_data,
{
**metadata,
'collection_name': f"raw_{input_data['schedule_name']}",
'last_data_timestamp': last_data_timestamp
},
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
) )
if data == {}: if data == {}:
return return
await workflow.execute_activity_method(
Activities.put_last_data_timestamp,
{
**metadata,
'data': data,
'workflow_name': input_data['workflow_name'],
'schedule_name': input_data['schedule_name']
},
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
input_data['data'] = data input_data['data'] = data
input_data['metadata'] = metadata
await workflow.execute_child_workflow( await workflow.execute_child_workflow(
'core_scouter', 'core_scouter',

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

@@ -32,14 +32,7 @@ class CoreScouter:
retention_time (int): The retention time for data in redis in seconds. retention_time (int): The retention time for data in redis in seconds.
""" """
metadata = { metadata = input_data['metadata']
'metadata': {
'model_id': input_data['model_id'],
'model_name': input_data['model_name'],
'schedule_name': input_data['schedule_name'],
'workflow_name': input_data['workflow_name']
}
}
filtered_data = await workflow.execute_local_activity_method( filtered_data = await workflow.execute_local_activity_method(
Activities.data_quality_gate, Activities.data_quality_gate,

3743
specs_30.json Normal file
View File

File diff suppressed because it is too large Load Diff

724
test.ipynb Normal file
View File

@@ -0,0 +1,724 @@
{
"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 <class 'pydruid.db.sqlalchemy.DruidDialect'> 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": [
"<div>\n",
"<style scoped>\n",
" .dataframe tbody tr th:only-of-type {\n",
" vertical-align: middle;\n",
" }\n",
"\n",
" .dataframe tbody tr th {\n",
" vertical-align: top;\n",
" }\n",
"\n",
" .dataframe thead th {\n",
" text-align: right;\n",
" }\n",
"</style>\n",
"<table border=\"1\" class=\"dataframe\">\n",
" <thead>\n",
" <tr style=\"text-align: right;\">\n",
" <th></th>\n",
" <th>name</th>\n",
" <th>kafka.topic</th>\n",
" <th>tag</th>\n",
" <th>value</th>\n",
" <th>timestamp</th>\n",
" <th>inserted_at</th>\n",
" </tr>\n",
" </thead>\n",
" <tbody>\n",
" <tr>\n",
" <th>0</th>\n",
" <td>Counter</td>\n",
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
" <td>ns=2;i=2</td>\n",
" <td>-50.132</td>\n",
" <td>2025-07-02 13:05:19</td>\n",
" <td>2025-07-02 13:05:19.729000</td>\n",
" </tr>\n",
" <tr>\n",
" <th>1</th>\n",
" <td>Rollout</td>\n",
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
" <td>ns=2;i=3</td>\n",
" <td>70.767</td>\n",
" <td>2025-07-02 13:05:19</td>\n",
" <td>2025-07-02 13:05:19.731000</td>\n",
" </tr>\n",
" <tr>\n",
" <th>2</th>\n",
" <td>Square</td>\n",
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
" <td>ns=2;i=4</td>\n",
" <td>-58.448</td>\n",
" <td>2025-07-02 13:05:19</td>\n",
" <td>2025-07-02 13:05:19.732000</td>\n",
" </tr>\n",
" <tr>\n",
" <th>3</th>\n",
" <td>Counter</td>\n",
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
" <td>ns=2;i=2</td>\n",
" <td>-50.126</td>\n",
" <td>2025-07-02 13:05:24</td>\n",
" <td>2025-07-02 13:05:24.728000</td>\n",
" </tr>\n",
" <tr>\n",
" <th>4</th>\n",
" <td>Rollout</td>\n",
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
" <td>ns=2;i=3</td>\n",
" <td>69.199</td>\n",
" <td>2025-07-02 13:05:24</td>\n",
" <td>2025-07-02 13:05:24.730000</td>\n",
" </tr>\n",
" <tr>\n",
" <th>...</th>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" </tr>\n",
" <tr>\n",
" <th>373</th>\n",
" <td>Rollout</td>\n",
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
" <td>ns=2;i=3</td>\n",
" <td>85.921</td>\n",
" <td>2025-07-02 13:15:40</td>\n",
" <td>2025-07-02 13:15:40.230000</td>\n",
" </tr>\n",
" <tr>\n",
" <th>374</th>\n",
" <td>Square</td>\n",
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
" <td>ns=2;i=4</td>\n",
" <td>-69.296</td>\n",
" <td>2025-07-02 13:15:40</td>\n",
" <td>2025-07-02 13:15:40.232000</td>\n",
" </tr>\n",
" <tr>\n",
" <th>375</th>\n",
" <td>Counter</td>\n",
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
" <td>ns=2;i=2</td>\n",
" <td>-71.207</td>\n",
" <td>2025-07-02 13:15:45</td>\n",
" <td>2025-07-02 13:15:45.228000</td>\n",
" </tr>\n",
" <tr>\n",
" <th>376</th>\n",
" <td>Rollout</td>\n",
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
" <td>ns=2;i=3</td>\n",
" <td>84.665</td>\n",
" <td>2025-07-02 13:15:45</td>\n",
" <td>2025-07-02 13:15:45.231000</td>\n",
" </tr>\n",
" <tr>\n",
" <th>377</th>\n",
" <td>Square</td>\n",
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
" <td>ns=2;i=4</td>\n",
" <td>-67.592</td>\n",
" <td>2025-07-02 13:15:45</td>\n",
" <td>2025-07-02 13:15:45.233000</td>\n",
" </tr>\n",
" </tbody>\n",
"</table>\n",
"<p>378 rows × 6 columns</p>\n",
"</div>"
],
"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('<urllib3.connection.HTTPConnection object at 0x72128094cc50>: 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: <urllib3.connection.HTTPConnection object at 0x72128094cc50>: 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('<urllib3.connection.HTTPConnection object at 0x72128094cc50>: 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('<urllib3.connection.HTTPConnection object at 0x72128094cc50>: 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
}

43
tests/activities/test_activities.py
View File

@@ -2,16 +2,16 @@ from unittest.mock import patch, MagicMock, ANY
from pytest import mark from pytest import mark
from sientia_do.temporal.activities.postgres import Postgres from sientia_do.temporal.activities.postgres import Postgres
from scouter.activities.activities import Activities from scouter.activities.activities import Activities
from scouter.activities.mongodb import MongoDB
from scouter.activities.redis import Redis from scouter.activities.redis import Redis
from scouter.activities.kafka import Kafka
from scouter.activities.gates import Gates from scouter.activities.gates import Gates
@patch('scouter.activities.activities.MongoDB.__init__')
@patch('scouter.activities.activities.Postgres.__init__') @patch('scouter.activities.activities.Postgres.__init__')
@patch('scouter.activities.activities.Redis.__init__') @patch('scouter.activities.activities.Redis.__init__')
@patch('scouter.activities.activities.Kafka.__init__')
@patch('scouter.activities.activities.Gates.__init__') @patch('scouter.activities.activities.Gates.__init__')
def test___init__(mock_gates_init, mock_kafka_init, mock_redis_init, mock_postgres_init): def test___init__(mock_gates_init, mock_redis_init, mock_postgres_init, mock_mongodb_init):
postgres_config = { postgres_config = {
'host': 'localhost', 'host': 'localhost',
@@ -30,10 +30,9 @@ def test___init__(mock_gates_init, mock_kafka_init, mock_redis_init, mock_postgr
'password': 'redis' 'password': 'redis'
} }
kafka_config = { mongodb_config = {
'bootstrap_servers': 'localhost:9092', 'connection_string': 'mongodb://localhost:27017',
'polling_time': 1000, 'database_name': 'test_database'
'group_id': 'test-group'
} }
logger = MagicMock() logger = MagicMock()
@@ -42,7 +41,7 @@ def test___init__(mock_gates_init, mock_kafka_init, mock_redis_init, mock_postgr
activities = Activities( activities = Activities(
postgres_config=postgres_config, postgres_config=postgres_config,
redis_config=redis_config, redis_config=redis_config,
kafka_config=kafka_config, mongodb_config=mongodb_config,
logger=logger, logger=logger,
notification_handler=notification_handler notification_handler=notification_handler
) )
@@ -50,7 +49,7 @@ def test___init__(mock_gates_init, mock_kafka_init, mock_redis_init, mock_postgr
assert isinstance(activities, Activities) assert isinstance(activities, Activities)
assert isinstance(activities, Postgres) assert isinstance(activities, Postgres)
assert isinstance(activities, Redis) assert isinstance(activities, Redis)
assert isinstance(activities, Kafka) assert isinstance(activities, MongoDB)
assert isinstance(activities, Gates) assert isinstance(activities, Gates)
mock_postgres_init.assert_called_once_with( mock_postgres_init.assert_called_once_with(
@@ -76,11 +75,10 @@ def test___init__(mock_gates_init, mock_kafka_init, mock_redis_init, mock_postgr
notification_handler=notification_handler notification_handler=notification_handler
) )
mock_kafka_init.assert_called_once_with( mock_mongodb_init.assert_called_once_with(
ANY, ANY,
bootstrap_servers=kafka_config['bootstrap_servers'], connection_string=mongodb_config['connection_string'],
polling_time=kafka_config['polling_time'], database_name=mongodb_config['database_name'],
group_id=kafka_config['group_id'],
logger=logger, logger=logger,
notification_handler=notification_handler notification_handler=notification_handler
) )
@@ -94,12 +92,12 @@ def test___init__(mock_gates_init, mock_kafka_init, mock_redis_init, mock_postgr
@patch('scouter.activities.activities.Postgres.__init__') @patch('scouter.activities.activities.Postgres.__init__')
@patch('scouter.activities.activities.Redis.__init__') @patch('scouter.activities.activities.Redis.__init__')
@patch('scouter.activities.activities.Kafka.__init__')
@patch('scouter.activities.activities.Gates.__init__') @patch('scouter.activities.activities.Gates.__init__')
@patch('scouter.activities.activities.MongoDB.__init__')
@patch('scouter.activities.activities.Postgres.close') @patch('scouter.activities.activities.Postgres.close')
@patch('scouter.activities.activities.Kafka.close') @patch('scouter.activities.activities.MongoDB.shutdown')
def test_shutdown(mock_kafka_close, mock_postgres_close, def test_shutdown(mock_mongodb_close, mock_postgres_close, _mock_mongodb_init,
_mock_gates_init, _mock_redis_init, _mock_kafka_init, _mock_postgres_init): _mock_gates_init, _mock_redis_init, _mock_postgres_init):
postgres_config = { postgres_config = {
'host': 'localhost', 'host': 'localhost',
'port': 5432, 'port': 5432,
@@ -117,10 +115,9 @@ def test_shutdown(mock_kafka_close, mock_postgres_close,
'password': 'redis' 'password': 'redis'
} }
kafka_config = { mongodb_config = {
'bootstrap_servers': 'localhost:9092', 'connection_string': 'mongodb://localhost:27017',
'polling_time': 1000, 'database_name': 'test_database'
'group_id': 'test-group'
} }
logger = MagicMock() logger = MagicMock()
@@ -129,7 +126,7 @@ def test_shutdown(mock_kafka_close, mock_postgres_close,
activities = Activities( activities = Activities(
postgres_config=postgres_config, postgres_config=postgres_config,
redis_config=redis_config, redis_config=redis_config,
kafka_config=kafka_config, mongodb_config=mongodb_config,
logger=logger, logger=logger,
notification_handler=notification_handler notification_handler=notification_handler
) )
@@ -137,4 +134,4 @@ def test_shutdown(mock_kafka_close, mock_postgres_close,
activities.shutdown() activities.shutdown()
mock_postgres_close.assert_called_once() mock_postgres_close.assert_called_once()
mock_kafka_close.assert_called_once() mock_mongodb_close.assert_called_once()

36
tests/activities/test_gates.py
View File

@@ -11,7 +11,9 @@ def gates_fixture():
"""Fixture to create a Gates instance with mocked dependencies.""" """Fixture to create a Gates instance with mocked dependencies."""
logger = Mock() logger = Mock()
notification_handler = MagicMock() notification_handler = MagicMock()
return Gates(logger=logger, notification_handler=notification_handler) gates = Gates(logger=logger, notification_handler=notification_handler)
gates.send_notification = MagicMock()
return gates
metadata = { metadata = {
@@ -51,7 +53,7 @@ async def test_data_quality_gate_with_null_values_filter_discard(gates_fixture):
# Verify # Verify
assert len(result['tag']) == 2 assert len(result['tag']) == 2
assert 'tag2' not in result['tag'] assert 'tag2' not in result['tag']
gates_fixture.notification_handler.build_and_send_notification.assert_called_once() gates_fixture.send_notification.assert_called_once()
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -84,7 +86,7 @@ async def test_data_quality_gate_with_out_of_bounds_filter_keep(gates_fixture):
# Verify data is kept but notification is sent # Verify data is kept but notification is sent
assert len(result['tag']) == 3 # All rows kept assert len(result['tag']) == 3 # All rows kept
gates_fixture.notification_handler.build_and_send_notification.assert_called_once() gates_fixture.send_notification.assert_called_once()
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -117,7 +119,7 @@ async def test_data_quality_gate_with_multiple_filters(gates_fixture):
assert result == {'tag': {0: 'tag1', 3: 'tag4'}, 'name': {0: 'tag1', 3: 'tag4'}, 'value': { assert result == {'tag': {0: 'tag1', 3: 'tag4'}, 'name': {0: 'tag1', 3: 'tag4'}, 'value': {
0: 1.0, 3: 4.0}, 'timestamp': {0: '2023-01-01', 3: '2023-01-04'}} 0: 1.0, 3: 4.0}, 'timestamp': {0: '2023-01-01', 3: '2023-01-04'}}
# Should be called twice (once for each filter) # Should be called twice (once for each filter)
assert gates_fixture.notification_handler.build_and_send_notification.call_count == 2 assert gates_fixture.send_notification.call_count == 2
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -184,8 +186,8 @@ async def test_data_quality_gate_with_filter_error(gates_fixture):
# Verify error notification is sent and data is unchanged # Verify error notification is sent and data is unchanged
assert len(result['tag']) == 1 assert len(result['tag']) == 1
gates_fixture.notification_handler.build_and_send_notification.assert_called_once() gates_fixture.send_notification.assert_called_once()
call_args = gates_fixture.notification_handler.build_and_send_notification.call_args[1] call_args = gates_fixture.send_notification.call_args[1]
assert call_args['notification_id'] == "DATA_QUALITY_GATE_ISSUES" assert call_args['notification_id'] == "DATA_QUALITY_GATE_ISSUES"
assert call_args['level'] == NotificationLevel.ERROR assert call_args['level'] == NotificationLevel.ERROR
assert "Filter error" in call_args['message'] assert "Filter error" in call_args['message']
@@ -214,7 +216,7 @@ async def test_data_quality_gate_with_empty_data(gates_fixture):
# Verify empty result and no notifications # Verify empty result and no notifications
assert len(result['tag']) == 0 assert len(result['tag']) == 0
gates_fixture.notification_handler.build_and_send_notification.assert_not_called() gates_fixture.send_notification.assert_not_called()
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -240,7 +242,7 @@ async def test_data_quality_gate_with_no_filters(gates_fixture):
# Verify data is unchanged and no notifications # Verify data is unchanged and no notifications
assert len(result['tag']) == 1 assert len(result['tag']) == 1
gates_fixture.notification_handler.build_and_send_notification.assert_not_called() gates_fixture.send_notification.assert_not_called()
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -265,14 +267,15 @@ async def test_data_quality_gate_with_no_filters(gates_fixture):
) )
def test_apply_aggregation(gates_fixture, group_data, aggr_function, expected_result): def test_apply_aggregation(gates_fixture, group_data, aggr_function, expected_result):
"""Test apply_aggregation method with various scenarios.""" """Test apply_aggregation method with various scenarios."""
result = gates_fixture.apply_aggregation(group_data, aggr_function) result = gates_fixture.apply_aggregation(
group_data, aggr_function, metadata)
assert result == expected_result assert result == expected_result
# Check notification was sent for invalid function # Check notification was sent for invalid function
if aggr_function == 'invalid': if aggr_function == 'invalid':
gates_fixture.notification_handler.build_and_send_notification.assert_called_once() gates_fixture.send_notification.assert_called_once()
else: else:
gates_fixture.notification_handler.build_and_send_notification.assert_not_called() gates_fixture.send_notification.assert_not_called()
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -290,8 +293,8 @@ async def test_aggregate_data(gates_fixture):
'value': None, 'timestamp': '2023-01-04'}, 'value': None, 'timestamp': '2023-01-04'},
], ],
'model_tags': { 'model_tags': {
'name1': {'aggr_function': 'avg'}, 'name1': {'aggr_func': 'avg'},
'name2': {'aggr_function': 'max'}, 'name2': {'aggr_func': 'max'},
}, },
**metadata **metadata
} }
@@ -308,7 +311,7 @@ async def test_aggregate_data(gates_fixture):
# Verify # Verify
assert result == expected_result assert result == expected_result
gates_fixture.notification_handler.build_and_send_notification.assert_not_called() gates_fixture.send_notification.assert_not_called()
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -342,7 +345,7 @@ async def test_aggregate_data_with_continue(gates_fixture):
# Verify # Verify
assert result == expected_result assert result == expected_result
gates_fixture.notification_handler.build_and_send_notification.assert_not_called() gates_fixture.send_notification.assert_not_called()
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -373,7 +376,8 @@ async def test_aggregate_data_raise_exception(gates_fixture):
await gates_fixture.aggregate_data(input_data) await gates_fixture.aggregate_data(input_data)
except Exception as e: except Exception as e:
assert str(e) == "Test exception" assert str(e) == "Test exception"
gates_fixture.notification_handler.build_and_send_notification.assert_called_once_with( gates_fixture.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="AGGREGATION_ISSUES", notification_id="AGGREGATION_ISSUES",
message="Error aggregating data: Test exception", message="Error aggregating data: Test exception",
block="aggregate_data", block="aggregate_data",

92
tests/activities/test_kafka.py
View File

@@ -1,92 +0,0 @@
from unittest.mock import MagicMock, patch, ANY
from pytest import fixture, mark
from pandas import DataFrame
from scouter.activities.kafka import Kafka
@fixture
@patch("scouter.activities.kafka.KafkaConsumer")
def kafka(_kafka_consumer):
return Kafka(
bootstrap_servers="localhost:9092",
polling_time=1000,
group_id="test-group",
logger=MagicMock(),
notification_handler=MagicMock()
)
@patch("scouter.activities.kafka.KafkaConsumer")
def test___init__(kafka_consumer):
kafka = Kafka(
bootstrap_servers="localhost:9092",
polling_time=1000,
group_id="test-group",
logger=MagicMock(),
notification_handler=MagicMock()
)
assert kafka.polling_time == 1000
assert kafka.kafka_connector == kafka_consumer.return_value
kafka_consumer.assert_called_once_with(
bootstrap_servers="localhost:9092",
auto_offset_reset="earliest",
enable_auto_commit=True,
group_id="test-group",
value_deserializer=ANY
)
metadata = {
'metadata': {
'model_id': 'test_model_id',
'model_name': 'test_model',
'schedule_name': 'test_schedule',
'workflow_name': 'scouter'
}
}
@mark.asyncio
async def test_load_from_kafka(kafka):
input_data = {"topic": "test-topic", **metadata}
data = [
("test-topic", [
MagicMock(
value=f"test-value-{i}"
) for i in range(10)
])
]
kafka.kafka_connector.poll.return_value = MagicMock(
items=MagicMock(return_value=data)
)
expected = DataFrame([d.value for d in data[0][1]]).to_dict()
result = await kafka.load_from_kafka(input_data)
assert result == expected
kafka.kafka_connector.subscribe.assert_called_once_with(["test-topic"])
kafka.kafka_connector.poll.assert_called_once_with(timeout_ms=1000)
@mark.asyncio
async def test_load_from_kafka_empty(kafka):
input_data = {"topic": "test-topic", **metadata}
kafka.kafka_connector.poll.return_value = MagicMock(
items=MagicMock(return_value=[])
)
result = await kafka.load_from_kafka(input_data)
assert result == {}
kafka.kafka_connector.subscribe.assert_called_once_with(["test-topic"])
kafka.kafka_connector.poll.assert_called_once_with(timeout_ms=1000)

192
tests/activities/test_mongo.py Normal file
View File

@@ -0,0 +1,192 @@
from datetime import datetime
from unittest.mock import ANY, MagicMock, patch
from pytest import fixture, mark
from sientia_do.notifications.models import NotificationLevel
from scouter.activities.mongodb import MongoDB, clear_mongo_id
def test_clear_mongo_id():
"""Test clear_mongo_id"""
data = [
{'_id': '1', 'name': 'test1'},
{'_id': '2', 'name': [{
'_id': '3',
'name': 'test3'
}]}
]
result = clear_mongo_id(data)
assert result == [{'name': 'test1'}, {'name': [{'name': 'test3'}]}]
@patch('scouter.activities.mongodb.MongoClient')
def test_mongodb___init__(mock_mongo_client):
"""Test MongoDB __init__"""
mongo = MongoDB(
connection_string='mongodb://localhost:27017',
database_name='test_db',
logger=MagicMock(),
notification_handler=MagicMock()
)
mock_mongo_client.assert_called_once_with(
'mongodb://localhost:27017',
serverSelectionTimeoutMS=5000
)
mock_mongo_client.return_value.server_info.assert_called_once()
mock_mongo_client.return_value.__getitem__.assert_called_once_with(
'test_db')
assert mongo.client is not None
assert mongo.database is not None
@fixture
@patch('scouter.activities.mongodb.MongoClient')
def mongodb_activity(mock_mongo_client):
"""Test MongoDB activity"""
mongo = MongoDB(
connection_string='mongodb://localhost:27017',
database_name='test_db',
logger=MagicMock(),
notification_handler=MagicMock()
)
return mongo
def test_shutdown_success(mongodb_activity):
"""Test shutdown"""
mongodb_activity.shutdown()
mongodb_activity.client.close.assert_called_once()
def test_shutdown_error(mongodb_activity):
"""Test shutdown"""
mongodb_activity.client.close = MagicMock(side_effect=Exception('test'))
mongodb_activity.shutdown()
mongodb_activity.client.close.assert_called_once()
@mark.asyncio
async def test_load_latest_data_none_last_data_timestamp(mongodb_activity):
"""Test load_latest_data"""
collection = MagicMock()
mongodb_activity.database.__getitem__.return_value = collection
collection.find.return_value = [
{
'name': 'test1',
'value': 1,
'inserted_at': datetime.strptime(
'2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f')
}
]
result = await mongodb_activity.load_latest_data({
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection',
'last_data_timestamp': None
})
mongodb_activity.database.__getitem__.assert_called_once_with(
'test_collection')
collection.find.assert_called_once_with(
{},
{"_id": 0}
)
assert result == {
'name': {
0: 'test1'
},
'value': {
0: 1
},
'inserted_at': {
0: '2023-01-01 12:00:00.000000'
}
}
@mark.asyncio
async def test_load_latest_data_not_none_last_data_timestamp(mongodb_activity):
"""Test load_latest_data"""
collection = MagicMock()
mongodb_activity.database.__getitem__.return_value = collection
collection.find.return_value = [
{
'name': 'test1',
'value': 1,
'inserted_at': datetime.strptime(
'2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f')
}
]
result = await mongodb_activity.load_latest_data({
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection',
'last_data_timestamp': '2023-01-01 12:00:00.000000'
})
mongodb_activity.database.__getitem__.assert_called_once_with(
'test_collection')
collection.find.assert_called_once_with(
{
'inserted_at': {
'$gt': datetime.strptime(
'2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f')
}
},
{"_id": 0}
)
assert result == {
'name': {
0: 'test1'
},
'value': {
0: 1
},
'inserted_at': {
0: '2023-01-01 12:00:00.000000'
}
}
@mark.asyncio
async def test_load_latest_data_error(mongodb_activity):
"""Test load_latest_data"""
collection = MagicMock()
mongodb_activity.send_notification = MagicMock()
mongodb_activity.database.__getitem__.return_value = collection
collection.find.side_effect = Exception('test')
try:
await mongodb_activity.load_latest_data({
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection',
'last_data_timestamp': '2023-01-01 12:00:00.000000'
})
except Exception as e:
assert str(e) == 'test'
mongodb_activity.send_notification.assert_called_once_with(
metadata={'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule'},
notification_id='MONGO_LOAD_ERROR',
message='Error loading data from MongoDB: test',
block='load_latest_data',
level=NotificationLevel.ERROR,
attachment_content=ANY
)

88
tests/activities/test_redis.py
View File

@@ -51,6 +51,90 @@ metadata = {
} }
@pytest.mark.asyncio
async def test_get_last_data_timestamp_none(redis_activity):
"""Test get_last_data_timestamp"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule'
}
redis_activity.get = MagicMock(return_value=None)
result = await redis_activity.get_last_data_timestamp(test_data)
assert result is None
@pytest.mark.asyncio
async def test_get_last_data_timestamp_not_none(redis_activity):
"""Test get_last_data_timestamp"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule'
}
redis_activity.get = MagicMock(return_value='2023-01-01 12:00:00')
result = await redis_activity.get_last_data_timestamp(test_data)
redis_activity.get.assert_called_once_with(
'last_data_timestamp_test_pipeline_test_schedule'
)
assert result == '2023-01-01 12:00:00'
@pytest.mark.asyncio
async def test_put_last_data_timestamp_empty_dataframe(redis_activity):
"""Test put_last_data_timestamp with empty dataframe"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule',
'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records')
}
redis_activity.set = MagicMock()
result = await redis_activity.put_last_data_timestamp(test_data)
assert result is None
redis_activity.set.assert_not_called()
@pytest.mark.asyncio
async def test_put_last_data_timestamp_not_empty_dataframe(redis_activity):
"""Test put_last_data_timestamp with not empty dataframe"""
data = DataFrame({
'name': ['sensor1', 'sensor2'],
'value': [25.5, 30.0],
'inserted_at': ['2023-01-01 12:00:00', '2023-01-01 12:00:01']
})
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule',
'data': data.to_dict('records')
}
redis_activity.set = MagicMock()
result = await redis_activity.put_last_data_timestamp(test_data)
assert result == '2023-01-01 12:00:01'
redis_activity.set.assert_called_once_with(
'last_data_timestamp_test_pipeline_test_schedule',
'2023-01-01 12:00:01',
ttl=None
)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_group_and_hold_data_new_key(redis_activity): async def test_group_and_hold_data_new_key(redis_activity):
"""Test group_and_hold_data with a new key""" """Test group_and_hold_data with a new key"""
@@ -87,7 +171,7 @@ async def test_group_and_hold_data_new_key(redis_activity):
# Verify set was called with correct arguments # Verify set was called with correct arguments
redis_activity.set.assert_called_once() redis_activity.set.assert_called_once()
args, kwargs = redis_activity.set.call_args args, kwargs = redis_activity.set.call_args
assert args[0] == 'test_pipeline_test_schedule' assert args[0] == 'held_data_test_pipeline_test_schedule'
assert args[1] == { assert args[1] == {
'sensor1': 25.5, 'sensor1': 25.5,
'sensor2': 30.0, 'sensor2': 30.0,
@@ -139,7 +223,7 @@ async def test_group_and_hold_data_update_existing(redis_activity):
# Verify set was called with correct arguments # Verify set was called with correct arguments
redis_activity.set.assert_called_once() redis_activity.set.assert_called_once()
args, kwargs = redis_activity.set.call_args args, kwargs = redis_activity.set.call_args
assert args[0] == 'test_workflow_test_schedule' assert args[0] == 'held_data_test_workflow_test_schedule'
assert args[1] == { assert args[1] == {
'sensor1': 25.5, 'sensor1': 25.5,
'sensor2': 28.0, 'sensor2': 28.0,

52
tests/utils/test_connectors_config.py
View File

@@ -2,6 +2,8 @@ import os
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
from scouter.utils.connectors_config import ( from scouter.utils.connectors_config import (
build_druid_config,
build_mongodb_config,
build_postgres_config, build_postgres_config,
build_kafka_config, build_kafka_config,
build_redis_config build_redis_config
@@ -113,3 +115,53 @@ def test_build_redis_config_with_env_vars():
'username': 'test', 'username': 'test',
'password': 'test' 'password': 'test'
} }
def test_build_mongodb_config_defaults():
"""Test that build_mongodb_config returns default values when no env vars are set"""
config = build_mongodb_config()
assert config == {
'connection_string': 'mongodb://sientia:sientia@localhost:27017', # NOSONAR
'database_name': 'sientia'
}
def test_build_mongodb_config_with_env_vars():
"""Test that build_mongodb_config uses env vars when set"""
with patch.dict(os.environ, {
'MONGODB_URL': 'mongodb.example.com:27017',
'MONGODB_DATABASE_NAME': 'test_db',
'MONGODB_USERNAME': 'test',
'MONGODB_PASSWORD': 'test'
}):
config = build_mongodb_config()
assert config == {
'connection_string': 'mongodb://test:test@mongodb.example.com:27017',
'database_name': 'test_db'
}
def test_build_druid_config_defaults():
"""Test that build_druid_config returns default values when no env vars are set"""
config = build_druid_config()
assert config == {
'host': 'localhost',
'port': 8082
}
def test_build_druid_config_with_env_vars():
"""Test that build_druid_config uses env vars when set"""
with patch.dict(os.environ, {
'DRUID_HOST': 'druid.example.com',
'DRUID_PORT': '8083'
}):
config = build_druid_config()
assert config == {
'host': 'druid.example.com',
'port': 8083
}

16
tests/workflow/sub_workflows/test_core_scouter.py
View File

@@ -16,6 +16,14 @@ async def test_core_scouter_workflow_success(mock_workflow, core_scouter):
'filtered_data', 'grouped_data', 'held_data'] 'filtered_data', 'grouped_data', 'held_data']
await core_scouter.run( await core_scouter.run(
input_data={ input_data={
'metadata': {
'metadata': {
'model_id': 'test_model_id',
'model_name': 'test_model',
'schedule_name': 'test_schedule',
'workflow_name': 'test_workflow'
}
},
'workflow_name': 'test_workflow', 'workflow_name': 'test_workflow',
'schedule_name': 'test_schedule', 'schedule_name': 'test_schedule',
'model_name': 'test_model', 'model_name': 'test_model',
@@ -97,6 +105,14 @@ async def test_core_scouter_workflow_with_empty_data(mock_workflow, core_scouter
mock_workflow.execute_local_activity_method.return_value = {} mock_workflow.execute_local_activity_method.return_value = {}
await core_scouter.run( await core_scouter.run(
input_data={ input_data={
'metadata': {
'metadata': {
'model_id': 'test_model_id',
'model_name': 'test_model',
'schedule_name': 'test_schedule',
'workflow_name': 'test_workflow'
}
},
'workflow_name': 'test_workflow', 'workflow_name': 'test_workflow',
'schedule_name': 'test_schedule', 'schedule_name': 'test_schedule',
'model_name': 'test_model', 'model_name': 'test_model',

70
tests/workflow/test_scouter.py
View File

@@ -1,4 +1,4 @@
from unittest.mock import AsyncMock, patch, ANY from unittest.mock import AsyncMock, patch, ANY, call
from pytest import fixture, mark from pytest import fixture, mark
from scouter.workflow.scouter import Scouter from scouter.workflow.scouter import Scouter
from scouter.activities.activities import Activities from scouter.activities.activities import Activities
@@ -13,7 +13,10 @@ def scouter():
@patch('scouter.workflow.scouter.workflow', new_callable=AsyncMock) @patch('scouter.workflow.scouter.workflow', new_callable=AsyncMock)
async def test_scouter_workflow(mock_workflow, scouter): async def test_scouter_workflow(mock_workflow, scouter):
mock_workflow.execute_activity_method.return_value = 'test_data' mock_workflow.execute_local_activity_method.side_effect = [
'test_last_data_timestamp',
'test_data'
]
await scouter.run( await scouter.run(
input_data={ input_data={
'topic': 'test_topic', 'topic': 'test_topic',
@@ -32,11 +35,41 @@ async def test_scouter_workflow(mock_workflow, scouter):
} }
} }
mock_workflow.execute_local_activity_method.assert_has_calls(
[
call(
Activities.get_last_data_timestamp,
{
**expected_metadata,
'workflow_name': 'scouter',
'schedule_name': 'test_schedule'
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
mock_workflow.execute_local_activity_method.assert_has_calls(
[
call(
Activities.load_latest_data,
{
**expected_metadata,
'collection_name': "raw_test_schedule",
'last_data_timestamp': 'test_last_data_timestamp'
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
mock_workflow.execute_activity_method.assert_called_once_with( mock_workflow.execute_activity_method.assert_called_once_with(
Activities.load_from_kafka, Activities.put_last_data_timestamp,
{ {
**expected_metadata, **expected_metadata,
'topic': 'test_topic' 'data': 'test_data',
'workflow_name': 'scouter',
'schedule_name': 'test_schedule'
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
@@ -45,6 +78,7 @@ async def test_scouter_workflow(mock_workflow, scouter):
mock_workflow.execute_child_workflow.assert_called_once_with( mock_workflow.execute_child_workflow.assert_called_once_with(
'core_scouter', 'core_scouter',
{ {
'metadata': expected_metadata,
'topic': 'test_topic', 'topic': 'test_topic',
'data': 'test_data', 'data': 'test_data',
'workflow_name': 'scouter', 'workflow_name': 'scouter',
@@ -58,7 +92,10 @@ async def test_scouter_workflow(mock_workflow, scouter):
@mark.asyncio @mark.asyncio
@patch('scouter.workflow.scouter.workflow', new_callable=AsyncMock) @patch('scouter.workflow.scouter.workflow', new_callable=AsyncMock)
async def test_scouter_workflow_empty(mock_workflow, scouter): async def test_scouter_workflow_empty(mock_workflow, scouter):
mock_workflow.execute_activity_method.return_value = {} mock_workflow.execute_local_activity_method.side_effect = [
'test_last_data_timestamp',
{}
]
await scouter.run( await scouter.run(
input_data={ input_data={
'topic': 'test_topic', 'topic': 'test_topic',
@@ -77,14 +114,19 @@ async def test_scouter_workflow_empty(mock_workflow, scouter):
} }
} }
mock_workflow.execute_activity_method.assert_called_once_with( mock_workflow.execute_local_activity_method.assert_has_calls(
Activities.load_from_kafka, [
{ call(
**expected_metadata, Activities.load_latest_data,
'topic': 'test_topic' {
}, **expected_metadata,
retry_policy=ANY, 'collection_name': "raw_test_schedule",
start_to_close_timeout=ANY 'last_data_timestamp': 'test_last_data_timestamp'
) },
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
mock_workflow.execute_activity_method.assert_not_called()
mock_workflow.execute_child_workflow.assert_not_called() mock_workflow.execute_child_workflow.assert_not_called()

22
values.yaml
View File

@@ -11,7 +11,7 @@ image:
# This sets the pull policy for images. # This sets the pull policy for images.
pullPolicy: Always pullPolicy: Always
# Overrides the image tag whose default is the chart appVersion. # Overrides the image tag whose default is the chart appVersion.
tag: "0.2.2" tag: "0.2.3"
# 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/ # 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: imagePullSecrets:
@@ -123,7 +123,7 @@ env:
- name: GITHUB_REPO_URL - name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git" value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git"
- name: GITHUB_BRANCH - name: GITHUB_BRANCH
value: "main" value: "SIENTIAPDE-1110-criar-testes-e-2-e"
- name: PYTHON_APP - name: PYTHON_APP
value: "scouter.worker.worker" value: "scouter.worker.worker"
@@ -141,12 +141,12 @@ env:
- name: POSTGRES_MIN_CONNECTIONS - name: POSTGRES_MIN_CONNECTIONS
value: "10" value: "10"
- name: POSTGRES_MAX_CONNECTIONS - name: POSTGRES_MAX_CONNECTIONS
value: "20" value: "40"
- name: KAFKA_BOOTSTRAP_SERVERS - name: KAFKA_BOOTSTRAP_SERVERS
value: "kafka.kafka.svc.cluster.local:9092" value: "kafka.kafka.svc.cluster.local:9092"
- name: KAFKA_POLLING_TIME - name: KAFKA_POLLING_TIME
value: "1000" value: "10000"
- name: REDIS_HOST - name: REDIS_HOST
value: "redis-master.redis.svc.cluster.local" value: "redis-master.redis.svc.cluster.local"
@@ -173,6 +173,20 @@ env:
- name: TEMPORAL_NAMESPACE - name: TEMPORAL_NAMESPACE
value: "default" value: "default"
- name: MONGODB_USERNAME
value: "root"
- name: MONGODB_PASSWORD
value: "wKZDbMNU1c"
- name: MONGODB_URL
value: "my-release-mongodb.mongodb.svc.cluster.local:27017"
- name: MONGODB_DATABASE
value: "sientia"
- name: DRUID_HOST
value: "druid-router.druid.svc.cluster.local"
- name: DRUID_PORT
value: "8888"
ssh: ssh:
enabled: true enabled: true
secretName: git-ssh-key-sientia-scouter-worker secretName: git-ssh-key-sientia-scouter-worker