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

SIENTIAPDE-1005

Browse Source 
Implement workflows for fake data generation, scouter processing, and core scouter operations

- Added `FakeData` workflow to generate random data and send it to a Kafka topic.
- Implemented `Scouter` workflow to load data from Kafka and trigger the core scouter workflow.
- Created `CoreScouter` workflow to process data through quality gates, aggregation, and export to PostgreSQL.
- Developed comprehensive unit tests for activities and workflows, ensuring proper functionality and error handling.
- Enhanced Redis and Postgres activities with robust testing for data handling and error notifications.
- Introduced quality filters for data validation and implemented tests to verify their functionality.
This commit is contained in:
vitor-aignosi
2025-05-15 16:53:24 -03:00
parent 4e579dd5bd
commit b203b7d22c
29 changed files with 2070 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
scouter/activities/activities.py Normal file
View File

@@ -0,0 +1,65 @@
from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through():
from scouter.activities.postgres import Postgres
from scouter.activities.redis import Redis
from scouter.activities.kafka import Kafka
from scouter.activities.gates import Gates
from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler
from typing import Any
class Activities(Postgres, Redis, Kafka, Gates):
"""Activities class that combines multiple services with proper initialization."""
def __init__(self,
postgres_config: dict[str, Any],
redis_config: dict[str, Any],
kafka_config: dict[str, Any],
logger: Logger,
notification_handler: NotificationHandler):
# Initialize Postgres
Postgres.__init__(
self,
host=postgres_config['host'],
port=postgres_config['port'],
user=postgres_config['user'],
password=postgres_config['password'],
dbname=postgres_config['dbname'],
min_connections=postgres_config['min_connections'],
max_connections=postgres_config['max_connections'],
logger=logger,
notification_handler=notification_handler
)
# Initialize Redis
Redis.__init__(
self,
host=redis_config['host'],
port=redis_config['port'],
logger=logger,
notification_handler=notification_handler
)
# 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
Gates.__init__(
self,
logger=logger,
notification_handler=notification_handler
)
@activity.defn(name="prepare_activity")
def prepare_activity(self, input_data: dict[str, Any]):
super().prepare_activity(input_data)

80
scouter/activities/faker.py Normal file
View File

@@ -0,0 +1,80 @@
import random
from datetime import datetime, timezone
from typing import Any
import json
from logging import Logger
from kafka import KafkaProducer
from temporalio import activity
from sientia_do.notifications.handlers import NotificationHandler
from scouter.activities.base import BaseActivity
class Faker(BaseActivity):
def __init__(self, bootstrap_servers: str, logger: Logger,
notification_handler: NotificationHandler):
self.producer = KafkaProducer(
bootstrap_servers=bootstrap_servers,
value_serializer=lambda v: json.dumps(v).encode('utf-8')
)
# Predefined lists for tag and name
self.tags = {
'ns=1;i=1001': 'Temperature Sensor',
'ns=1;i=1002': 'Vibration Meter',
'ns=1;i=1003': 'Pressure Gauge',
'ns=1;i=1004': 'Flow Meter',
'ns=1;i=1005': 'Voltage Sensor',
'ns=1;i=1006': 'Current Sensor'
}
BaseActivity.__init__(self, logger, notification_handler)
@activity.defn(name="generate_and_send_data")
async def generate_and_send_data(self, input_data: dict[str, Any]):
"""
Generates random data and sends it to a Kafka topic.
Args:
input_data (dict[str, Any]): The input data containing:
topic (str): The Kafka topic to send data to
num_messages (int, optional): Number of messages to generate.
Defaults to random.randint(1, len(self.tags)).
"""
topic = input_data.get('topic')
num_messages = input_data.get(
'num_messages', random.randint(1, len(self.tags)))
if not topic:
raise ValueError("Topic must be specified in input_data")
self.logger.info(
f"Generating {num_messages} messages for topic {topic}")
for _ in range(num_messages):
# Select random tag and name
tag = random.choice(list(self.tags.keys()))
name = self.tags[tag]
# Generate random value between 0 and 100
if random.random() < 0.1:
value = None
else:
value = round(random.uniform(0, 100), 2)
# Create data dictionary
data = {
'tag': tag,
'name': name,
'timestamp': datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S'),
'value': value
}
# Send to Kafka
self.producer.send(topic, value=data)
# Ensure all messages are sent
self.producer.flush()
self.logger.info("Success")

149
scouter/activities/gates.py
View File

@@ -2,11 +2,11 @@ from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
from sientia_do.notifications.models import NotificationLevel
from pandas import DataFrame
from scouter.activities.base import BaseActivity
from scouter.utils.quality.filters import null_values_filter, out_of_bounds_filter
from typing import Any
import traceback
from pandas import DataFrame
quality_gate_filters = {
'NULL_VALUES_FILTER': null_values_filter,
@@ -15,6 +15,133 @@ quality_gate_filters = {
class Gates(BaseActivity):
def apply_aggregation(self, group: DataFrame, aggr_function: str) -> float | None | str:
"""
Apply aggregation function to a group of data.
Args:
group (DataFrame): The group of data to apply the aggregation function to.
aggr_function (str): The aggregation function to apply.
Returns:
float | None | str: The result of the aggregation function.
"""
if len(group) == 1:
return group['value'].item()
# Apply aggregation function to value
if aggr_function == 'lts':
return group['value'].iloc[-1]
else:
group.dropna(inplace=True, subset=['value'])
if group.empty:
return None
if aggr_function == 'avg':
return group['value'].mean()
elif aggr_function == 'mdn':
return group['value'].median()
elif aggr_function == 'max':
return group['value'].max()
elif aggr_function == 'min':
return group['value'].min()
else:
self.notification_handler.build_and_send_notification(
notification_id="AGGREGATION_ISSUES",
message=f"Invalid aggregation function: {aggr_function}",
block="aggregate_data",
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc()
)
return 'continue'
@activity.defn(name="aggregate_data")
async def aggregate_data(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Aggregates time series data by tag and name, applying specified
aggregation functions and taking the latest timestamp.
Args:
input_data (dict[str, Any]): The data to aggregate. Contains:
data (dict[str, Any]): The time series data.
model_tags (dict[str, Any]): The tags configuration
containing aggregation functions.
Returns:
dict[str, Any]: The aggregated data.
"""
try:
# Convert input data to DataFrame
df = DataFrame(input_data['data'])
self.logger.debug(
f"Aggregating time series data: {df.to_string()}")
# Initialize result dictionary
result = {}
# Group by tag and name
grouped = df.groupby(['tag', 'name'])
for (tag, name), group in grouped:
# Get the aggregation function from model_tags
aggr_function = input_data['model_tags'].get(
name, {}).get('aggr_function', 'lts')
group.sort_values(by='timestamp', inplace=True)
# Get the latest timestamp
latest_timestamp = group['timestamp'].max()
if group.empty:
continue
aggr_value = self.apply_aggregation(group, aggr_function)
if aggr_value == 'continue':
continue
self.logger.debug(
f"Aggregated data: {aggr_value}")
self.logger.debug(
f"Latest timestamp: {latest_timestamp}")
self.logger.debug(
f"Groups: {group.to_string()}")
self.logger.debug(
f"group name: {name}")
self.logger.debug(
f"group tag: {tag}")
# Store the result
result[f"{tag}_{name}"] = {
'tag': tag,
'name': name,
'value': aggr_value,
'timestamp': latest_timestamp,
'aggregation_function': aggr_function
}
result_df = DataFrame(list(result.values()))
self.logger.debug(f"Aggregated data:\n {result_df.to_string()}")
return result_df.to_dict()
except Exception as e:
trace = traceback.format_exc()
self.notification_handler.build_and_send_notification(
notification_id="AGGREGATION_ISSUES",
message=f"Error aggregating data: {e}",
block="aggregate_data",
level=NotificationLevel.ERROR,
attachment_content=trace
)
self.logger.error(trace)
raise
@activity.defn(name="data_quality_gate")
async def data_quality_gate(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
@@ -29,6 +156,8 @@ class Gates(BaseActivity):
filter_name: The name of the filter.
policy: The policy to apply. Can be "DISCARD" or "KEEP".
data (dict[str, Any]): The data to validate.
model_tags (dict[str, Any]): The tags of the model.
And it's respective configuration.
Returns:
dict[str, Any]: The data validated.
@@ -36,6 +165,10 @@ class Gates(BaseActivity):
filters = input_data['filters']
data = DataFrame(input_data['data'])
model_tags = input_data['model_tags']
self.logger.debug(
f"Applying quality gate to data: {data.to_string()}")
for filter_name, policy in filters.items():
if filter_name not in quality_gate_filters:
@@ -43,17 +176,21 @@ class Gates(BaseActivity):
continue
try:
filtered_data = quality_gate_filters[filter_name](data)
filtered_data = quality_gate_filters[filter_name](
data, model_tags)
except Exception as e:
trace = traceback.format_exc()
self.notification_handler.build_and_send_notification(
notification_id="DATA_QUALITY_GATE_ISSUES",
message=f"Error applying filter {filter_name}: {e}",
block="data_quality_gate",
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc()
attachment_content=trace
)
self.logger.error(trace)
else:
if filtered_data.empty:
continue
@@ -62,7 +199,7 @@ class Gates(BaseActivity):
attachment = filtered_data.to_string()
self.notification_handler.build_and_send_notification(
notification_id="DATA_QUALITY_GATE_ISSUES",
notification_id=f"DATA_QUALITY_GATE_ISSUES__{filter_name}",
message=message,
block="data_quality_gate",
level=NotificationLevel.WARNING,
@@ -70,6 +207,8 @@ class Gates(BaseActivity):
)
if policy == "DISCARD":
data = data[not data.isin(filtered_data).all(axis=1)]
data = data[~data.index.isin(filtered_data.index)]
self.logger.debug("Data quality gate applied")
return data.to_dict()

18
scouter/activities/kafka.py
View File

@@ -7,6 +7,7 @@ with workflow.unsafe.imports_passed_through():
from typing import Any
from kafka import KafkaConsumer
from pandas import DataFrame
import json
class Kafka(BaseActivity):
@@ -19,13 +20,13 @@ class Kafka(BaseActivity):
auto_offset_reset="earliest",
enable_auto_commit=True,
group_id=group_id,
value_deserializer=lambda x: x.decode("utf-8")
value_deserializer=lambda x: json.loads(x.decode("utf-8"))
)
super().__init__(logger, notification_handler)
BaseActivity.__init__(self, logger, notification_handler)
@activity.defn(name="load_from_kafka")
def load_from_kafka(self, input_data: dict[str, Any]) -> dict[str, Any]:
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.
@@ -35,6 +36,9 @@ class Kafka(BaseActivity):
Returns:
dict[str, Any]: The data loaded from the topic.
"""
self.logger.debug(f"Loading data from topic: {input_data['topic']}")
topic = input_data["topic"]
# Subscribe to the specified topic
@@ -46,6 +50,8 @@ class Kafka(BaseActivity):
# Poll for messages
records = self.kafka_connector.poll(timeout_ms=self.polling_time)
self.logger.debug(f"Polled {len(records)} records from topic: {topic}")
# Process the polled records
for _topic_partition, msgs in records.items():
for msg in msgs:
@@ -55,4 +61,10 @@ class Kafka(BaseActivity):
if not message_values:
return {}
self.logger.debug(
f"Loaded {len(message_values)} messages from topic: {topic}")
self.logger.debug(
f"Loaded data: {message_values}")
return DataFrame(message_values).to_dict()

65
scouter/activities/postgres.py
View File

@@ -2,7 +2,9 @@ import traceback
from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
from psycopg2.pool import ThreadedConnectionPool
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool
from pandas import DataFrame
from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler
@@ -22,19 +24,20 @@ class Postgres(BaseActivity):
self.password = password
self.dbname = dbname
self.pool = ThreadedConnectionPool(
minconn=min_connections,
maxconn=max_connections,
host=self.host,
port=self.port,
user=self.user,
password=self.password,
dbname=self.dbname)
# Create SQLAlchemy engine with connection pooling
self.engine = create_engine(
f'postgresql://{user}:{password}@{host}:{port}/{dbname}',
poolclass=QueuePool,
pool_size=min_connections,
max_overflow=max_connections - min_connections,
pool_pre_ping=True
)
self.session_factory = sessionmaker(bind=self.engine)
super().__init__(logger, notification_handler)
BaseActivity.__init__(self, logger, notification_handler)
def close(self):
self.pool.closeall()
self.engine.dispose()
def __del__(self):
self.close()
@@ -51,28 +54,32 @@ class Postgres(BaseActivity):
data (DataFrame): The data to export.
"""
self.logger.debug(
f"Exporting data to postgres: {input_data['data']}")
schema = input_data["schema"]
table_name = input_data["table_name"]
data = DataFrame(input_data["data"])
conn = self.pool.getconn()
with self.session_factory() as session:
try:
data.to_sql(table_name, self.engine, schema=schema,
if_exists="append", index=False)
session.commit()
try:
data.to_sql(table_name, conn, schema=schema,
if_exists="append", index=False)
conn.commit()
except Exception as e:
trace = traceback.format_exc()
self.notification_handler.build_and_send_notification(
notification_id="ERROR_EXPORTING_DATA_TO_POSTGRES",
message=f"Error exporting data to postgres: {e}",
block="export_data_to_postgres",
level=NotificationLevel.ERROR,
attachment_content=trace
)
except Exception as e:
trace = traceback.format_exc()
self.notification_handler.build_and_send_notification(
notification_id="ERROR_EXPORTING_DATA_TO_POSTGRES",
message=f"Error exporting data to postgres: {e}",
block="export_data_to_postgres",
level=NotificationLevel.ERROR,
attachment_content=trace
)
self.logger.error(trace)
self.logger.error(trace)
finally:
self.pool.putconn(conn)
else:
self.logger.debug("Data exported to postgres")
finally:
session.close()

78
scouter/activities/redis.py
View File

@@ -4,15 +4,85 @@ with workflow.unsafe.imports_passed_through():
from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler
from scouter.activities.base import BaseActivity
import redis
import json
from typing import Any
from kafka import KafkaConsumer
from pandas import DataFrame
import numpy as np
from datetime import datetime
class Redis(BaseActivity):
def __init__(self, host: str, port: int, db: int, logger: Logger, notification_handler: NotificationHandler):
def __init__(self, host: str, port: int,
logger: Logger, notification_handler: NotificationHandler):
self.host = host
self.port = port
self.db = db
super().__init__(logger, notification_handler)f
self.redis_client = redis.Redis(
host=self.host,
port=self.port,
decode_responses=True
)
BaseActivity.__init__(self, logger, notification_handler)
def get(self, key: str):
history = self.redis_client.get(key)
return json.loads(history) if history else None
def set(self, key: str, data: dict, ttl=600):
self.redis_client.set(key, json.dumps(data), ex=ttl)
@activity.defn(name="group_and_hold_data")
async def group_and_hold_data(self, input_data: dict[str, Any]):
"""
Groups and holds data in redis. Keep a copy of the most recent
received data for a given pipeline and schedule. This activity updates
the data in redis and return the full keeped data.
Args:
input_data (dict[str, Any]): The data to group and hold.
workflow_name (str): The name of the workflow.
schedule_name (str): The name of the schedule.
data (dict[str, Any]): The data to group and hold.
retention_time (int): The retention time for data in redis in seconds.
"""
self.logger.debug("Grouping and holding data...")
data = DataFrame(input_data['data'])
retention_time = input_data['retention_time']
key = f"{input_data['workflow_name']}_{input_data['schedule_name']}"
data_hold = self.get(key)
if not data_hold:
data_hold = {}
if data.empty:
self.logger.warning("No data to export")
return data_hold
for _, row in data.iterrows():
value = row['value']
if value is None:
data_hold[row['name']] = np.nan
else:
data_hold[row['name']] = value
data_hold['timestamp'] = data['timestamp'].max() if not data.empty else \
datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.set(key, data_hold, ttl=retention_time)
data_hold_df = DataFrame(data_hold, index=[0])
data_hold_melted = data_hold_df.melt(
id_vars='timestamp', var_name='variable', value_name='value')
data_hold_melted['model_id'] = input_data['model_id']
data_hold_melted.reset_index(drop=True, inplace=True)
self.logger.debug(
f"Data grouped and held successfully:\n {data_hold_melted.to_string()}")
return data_hold_melted.to_dict()