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

SIENTIAPDE-1316

Browse Source 
Update .gitignore and refactor metrics.py, activities.py, and gates.py for improved clarity and consistency. Added coverage.xml and cache directories to .gitignore. Standardized string formatting and parameter handling in metrics and activities classes, enhancing code readability. Removed the deprecated faker.py file and adjusted related tests accordingly.
This commit is contained in:
vitor-aignosi
2025-10-16 13:31:12 -03:00
parent 8bdbf049b8
commit 97eb5bc904
27 changed files with 1101 additions and 1336 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
scouter/activities/activities.py
View File

@@ -1,14 +1,16 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from sientia_do.temporal.activities.postgres import Postgres
from os import getenv
from typing import Any
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.observability.logger import Logger
from scouter.activities.redis import Redis
from sientia_do.temporal.activities.postgres import Postgres
from scouter.activities.gates import Gates
from scouter.activities.mongodb import MongoDB
from typing import Any
from os import getenv
from scouter.activities.redis import Redis
class Activities(Postgres, Redis, Gates, MongoDB):
@@ -27,12 +29,14 @@ class Activities(Postgres, Redis, Gates, MongoDB):
interface while maintaining separation of concerns across different data services.
"""
def __init__(self,
postgres_config: dict[str, Any],
redis_config: dict[str, Any],
mongodb_config: dict[str, Any],
logger: Logger,
notification_handler: NotificationHandler):
def __init__(
self,
postgres_config: dict[str, Any],
redis_config: dict[str, Any],
mongodb_config: dict[str, Any],
logger: Logger,
notification_handler: NotificationHandler,
):
"""
Initialize the Activities class with all required services.
@@ -57,7 +61,7 @@ class Activities(Postgres, Redis, Gates, MongoDB):
min_connections=postgres_config['min_connections'],
max_connections=postgres_config['max_connections'],
logger=logger,
notification_handler=notification_handler
notification_handler=notification_handler,
)
# Initialize Redis
@@ -68,15 +72,11 @@ class Activities(Postgres, Redis, Gates, MongoDB):
logger=logger,
notification_handler=notification_handler,
username=redis_config['username'],
password=redis_config['password']
password=redis_config['password'],
)
# Initialize Gates
Gates.__init__(
self,
logger=logger,
notification_handler=notification_handler
)
Gates.__init__(self, logger=logger, notification_handler=notification_handler)
# Initialize MongoDB
MongoDB.__init__(
@@ -84,10 +84,10 @@ class Activities(Postgres, Redis, Gates, MongoDB):
connection_string=mongodb_config['connection_string'],
database_name=mongodb_config['database_name'],
logger=logger,
notification_handler=notification_handler
notification_handler=notification_handler,
)
self.pod_id = getenv("HOSTNAME", "localhost")
self.pod_id = getenv('HOSTNAME', 'localhost')
def shutdown(self):
"""

117
scouter/activities/faker.py
View File

@@ -1,117 +0,0 @@
import random
from datetime import datetime, timezone
from typing import Any
import json
from kafka import KafkaProducer
from temporalio import activity
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.observability.logger import Logger
class Faker(BaseActivity):
"""
Synthetic data generation for testing and development.
This class generates realistic industrial sensor data for testing purposes.
It provides:
- Configurable sensor tag simulation
- Realistic data value generation
- Kafka integration for data publishing
- Comprehensive error handling and logging
The class is designed for development, testing, and demonstration of
data processing pipelines without requiring real industrial data sources.
"""
def __init__(self, bootstrap_servers: str, logger: Logger,
notification_handler: NotificationHandler):
"""
Initialize the Faker class with Kafka producer and sensor configuration.
Args:
bootstrap_servers (str): Kafka bootstrap servers configuration
logger (Logger): Logger instance for operation logging
notification_handler (NotificationHandler): Handler for system notifications
"""
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]) -> None:
"""
Generate synthetic sensor data and publish to Kafka topic.
This activity creates realistic industrial sensor readings and publishes
them to the specified Kafka topic. The data includes sensor tags, names,
timestamps, and values with configurable message counts.
Args:
input_data (dict[str, Any]): Activity input parameters.
Required fields:
- topic (str): Kafka topic name for data publication
- metadata (dict[str, Any], optional): Workflow execution metadata
- num_messages (int, optional): Number of messages to generate.
Defaults to random count between 1 and available sensor tags
Returns:
None: This activity publishes data but doesn't return results
Raises:
ValueError: If topic is not specified
Exception: If data generation or Kafka publishing fails
"""
metadata = input_data['metadata']
topic = input_data.get('topic')
num_messages = input_data.get(
'num_messages', random.randint(1, len(self.tags))) # NOSONAR
if not topic:
raise ValueError("Topic must be specified in input_data")
self.info(
f"Generating {num_messages} messages for topic {topic}",
metadata=metadata
)
for _ in range(num_messages):
# Select random tag and name
tag = random.choice(list(self.tags.keys())) # NOSONAR
name = self.tags[tag]
# Generate random value between 0 and 100
if random.random() < 0.1: # NOSONAR
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.info("Success", metadata=metadata)

157
scouter/activities/gates.py
View File

@@ -1,20 +1,23 @@
from temporalio import workflow, activity
from collections.abc import Hashable
from scouter import metrics
from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through():
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.observability.logger import Logger
from scouter.utils.quality.filters import null_values_filter, out_of_bounds_filter
from typing import Any
import traceback
from typing import Any
from pandas import DataFrame
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.observability.logger import Logger
from sientia_do.temporal.activities.base import BaseActivity
from scouter import metrics
from scouter.utils.quality.filters import null_values_filter, out_of_bounds_filter
quality_gate_filters = {
'NULL_VALUES_FILTER': null_values_filter,
'OUT_OF_BOUNDS_FILTER': out_of_bounds_filter
'OUT_OF_BOUNDS_FILTER': out_of_bounds_filter,
}
@@ -41,11 +44,11 @@ class Gates(BaseActivity):
logger (Logger): Logger instance for operation logging
notification_handler (NotificationHandler): Handler for system notifications
"""
BaseActivity.__init__(
self, logger, notification_handler, set_error_counter=True)
BaseActivity.__init__(self, logger, notification_handler, set_error_counter=True)
def apply_aggregation(self, values: DataFrame, aggr_function: str,
metadata: dict[str, Any]) -> float | None | str:
def apply_aggregation(
self, values: DataFrame, aggr_function: str, metadata: dict[str, Any]
) -> float | None | str:
"""
Apply aggregation function to a group of time-series data.
@@ -84,7 +87,7 @@ class Gates(BaseActivity):
'avg': lambda x: x.mean(),
'mdn': lambda x: x.median(),
'max': lambda x: x.max(),
'min': lambda x: x.min()
'min': lambda x: x.min(),
}
if aggr_function in aggregation_map:
@@ -92,16 +95,16 @@ class Gates(BaseActivity):
else:
self.send_notification(
metadata=metadata,
notification_id="AGGREGATION_ISSUES",
message=f"Invalid aggregation function: {aggr_function}",
block="aggregate_data",
notification_id='AGGREGATION_ISSUES',
message=f'Invalid aggregation function: {aggr_function}',
block='aggregate_data',
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc()
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]:
@activity.defn(name='aggregate_data')
async def aggregate_data(self, input_data: dict[str, Any]) -> dict[Hashable, Any]:
"""
Aggregate time-series data by tag and name using specified functions.
@@ -116,7 +119,7 @@ class Gates(BaseActivity):
- model_tags (dict[str, Any]): Tag configuration with aggregation functions
Returns:
dict[str, Any]: Aggregated data organized by tag and name
dict[Hashable, Any]: Aggregated data organized by tag and name
Raises:
Exception: If aggregation operation fails
@@ -128,10 +131,7 @@ class Gates(BaseActivity):
# Convert input data to DataFrame
df = DataFrame(input_data['data'])
self.info(
f"Aggregating time series data for {len(df)} rows",
metadata=metadata
)
self.info(f'Aggregating time series data for {len(df)} rows', metadata=metadata)
# Sort once by timestamp for all data (more efficient than sorting each group)
df = df.sort_values(['tag', 'name', 'timestamp'])
@@ -147,49 +147,34 @@ class Gates(BaseActivity):
results = []
for (tag, name), group in grouped:
# Get the aggregation function from model_tags
aggr_function = model_tags.get(
name, {}).get('aggr_func', 'lts')
aggr_function = model_tags.get(name, {}).get('aggr_func', 'lts')
# Get the latest timestamp (last row since data is sorted)
latest_timestamp = group['timestamp'].iloc[-1]
aggr_value = self.apply_aggregation(
group, aggr_function, metadata)
aggr_value = self.apply_aggregation(group, aggr_function, metadata)
if aggr_value == 'continue':
continue
# Batch debug logging to reduce overhead
if self.logger.level <= 10: # DEBUG level
self.debug(
f"Processed {tag}_{name}: value={aggr_value}, "
f"timestamp={latest_timestamp}, func={aggr_function}",
metadata=metadata
)
# Store the result directly in list for better performance
results.append({
'tag': tag,
'name': name,
'value': aggr_value,
'timestamp': latest_timestamp,
'aggregation_function': aggr_function
})
results.append(
{
'tag': tag,
'name': name,
'value': aggr_value,
'timestamp': latest_timestamp,
'aggregation_function': aggr_function,
}
)
self.info(
f"Aggregated data has {len(results)} rows",
metadata=metadata
)
self.info(f'Aggregated data has {len(results)} rows', metadata=metadata)
# Convert to DataFrame only once at the end if we have results
if results:
result_df = DataFrame(results)
if self.logger.level <= 10: # DEBUG level
self.debug(
f"Final aggregated data:\n{result_df.to_string()}",
metadata=metadata
)
self.debug(f'Final aggregated data:\n{result_df.to_string()}', metadata=metadata)
return result_df.to_dict()
else:
@@ -201,18 +186,18 @@ class Gates(BaseActivity):
self.send_notification(
metadata=metadata,
notification_id="AGGREGATION_ISSUES",
message=f"Error aggregating data: {e}",
block="aggregate_data",
notification_id='AGGREGATION_ISSUES',
message=f'Error aggregating data: {e}',
block='aggregate_data',
level=NotificationLevel.ERROR,
attachment_content=trace
attachment_content=trace,
)
self.error(trace, metadata=metadata)
raise e
@activity.defn(name="data_quality_gate")
async def data_quality_gate(self, input_data: dict[str, Any]) -> dict[str, Any]:
@activity.defn(name='data_quality_gate')
async def data_quality_gate(self, input_data: dict[str, Any]) -> dict[Hashable, Any]:
"""
Apply data quality filters to incoming data.
@@ -228,7 +213,7 @@ class Gates(BaseActivity):
- model_tags (dict[str, Any]): Tag-specific validation rules
Returns:
dict[str, Any]: Filtered data that passes quality validation
dict[Hashable, Any]: Filtered data that passes quality validation
Raises:
Exception: If quality validation fails
@@ -240,10 +225,7 @@ class Gates(BaseActivity):
data = DataFrame(input_data['data'])
model_tags = input_data['model_tags']
self.info(
f"Applying quality gate to data to {len(data)} rows",
metadata=metadata
)
self.info(f'Applying quality gate to data to {len(data)} rows', metadata=metadata)
tags = list(model_tags.keys())
@@ -252,25 +234,21 @@ class Gates(BaseActivity):
for filter_name, config in filters.items():
policy = config['policy']
if filter_name not in quality_gate_filters:
self.warning(
f"Filter {filter_name} not found",
metadata=metadata
)
self.warning(f'Filter {filter_name} not found', metadata=metadata)
continue
try:
filtered_data = quality_gate_filters[filter_name](
data, model_tags)
filtered_data = quality_gate_filters[filter_name](data, model_tags)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=metadata,
notification_id="DATA_QUALITY_GATE_ISSUES",
message=f"Error applying filter {filter_name}: {e}",
block="data_quality_gate",
notification_id='DATA_QUALITY_GATE_ISSUES',
message=f'Error applying filter {filter_name}: {e}',
block='data_quality_gate',
level=NotificationLevel.ERROR,
attachment_content=trace
attachment_content=trace,
)
self.error(trace, metadata=metadata)
@@ -279,30 +257,27 @@ class Gates(BaseActivity):
if filtered_data.empty:
continue
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()
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,
block="data_quality_gate",
block='data_quality_gate',
level=NotificationLevel.WARNING,
attachment_content=attachment
attachment_content=attachment,
)
if policy == "DISCARD":
if policy == 'DISCARD':
data = data[~data.index.isin(filtered_data.index)]
self.info(
f"Data quality gate applied, final data has {len(data)} rows",
metadata=metadata
)
self.info(f'Data quality gate applied, final data has {len(data)} rows', metadata=metadata)
return data.to_dict()
@activity.defn(name="write_metrics")
async def write_metrics(self, input_data: dict[str, Any]):
@activity.defn(name='write_metrics')
async def write_metrics(self, input_data: dict[str, Any]) -> None:
"""
Write metrics to the database.
input_data:
@@ -310,18 +285,12 @@ class Gates(BaseActivity):
"""
metadata = input_data['metadata']
self.info(
f"Writing metrics for {metadata['model_name']}",
metadata=metadata
)
self.info(f'Writing metrics for {metadata["model_name"]}', metadata=metadata)
metrics.LABORIOUS_DATA_WRITTEN_COUNT.labels(
pod_id=self.pod_id,
model_name=metadata['model_name'],
pipeline_name=metadata['workflow_name']
pipeline_name=metadata['workflow_name'],
).inc()
self.info(
f"Metrics written for {metadata['model_name']}",
metadata=metadata
)
self.info(f'Metrics written for {metadata["model_name"]}', metadata=metadata)

100
scouter/activities/mongodb.py
View File

@@ -1,15 +1,19 @@
from temporalio import workflow, activity
from collections.abc import Hashable
from datetime import UTC
from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through():
from typing import Any
import traceback
from datetime import datetime, timezone
from pymongo import MongoClient
from datetime import datetime
from typing import Any
from pandas import DataFrame
from pymongo import MongoClient
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.observability.logger import Logger
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ
@@ -35,10 +39,10 @@ def clear_mongo_id(docs: list) -> list:
clear_mongo_id(doc)
elif isinstance(doc, dict):
if "_id" in doc:
del doc["_id"]
if '_id' in doc:
del doc['_id']
for key, value in doc.items():
for _key, value in doc.items():
if isinstance(value, list):
clear_mongo_id(value)
elif isinstance(value, dict):
@@ -62,9 +66,13 @@ class MongoDB(BaseActivity):
distributed data processing with fault tolerance and monitoring.
"""
def __init__(self, connection_string: str, database_name: str,
logger: Logger,
notification_handler: NotificationHandler):
def __init__(
self,
connection_string: str,
database_name: str,
logger: Logger,
notification_handler: NotificationHandler,
):
"""
Initialize MongoDB connection and services.
@@ -80,19 +88,19 @@ class MongoDB(BaseActivity):
self.connection_string = connection_string
self.database_name = database_name
self.client = MongoClient(
self.connection_string, serverSelectionTimeoutMS=5000)
self.client: MongoClient = 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")
logger.info('MongoDB connection initialized')
BaseActivity.__init__(self,
logger=logger,
notification_handler=notification_handler,
set_error_counter=True)
BaseActivity.__init__(
self, logger=logger, notification_handler=notification_handler, set_error_counter=True
)
def shutdown(self):
"""
@@ -103,11 +111,11 @@ class MongoDB(BaseActivity):
"""
try:
if self.client:
self.logger.info("Closing MongoDB connection...")
self.logger.info('Closing MongoDB connection...')
self.client.close()
self.logger.info("MongoDB connection closed successfully")
self.logger.info('MongoDB connection closed successfully')
except Exception as e:
self.logger.error(f"Failed to close MongoDB connection: {e}")
self.logger.error(f'Failed to close MongoDB connection: {e}')
def __del__(self):
"""
@@ -118,8 +126,8 @@ class MongoDB(BaseActivity):
"""
self.shutdown()
@activity.defn(name="load_latest_data")
async def load_latest_data(self, input_data: dict[str, Any]) -> dict[str, Any]:
@activity.defn(name='load_latest_data')
async def load_latest_data(self, input_data: dict[str, Any]) -> dict[Hashable, Any]:
"""
Load the latest data from MongoDB collection since a specified timestamp.
@@ -144,60 +152,44 @@ class MongoDB(BaseActivity):
collection_name = input_data['collection_name']
last_data_timestamp = input_data['last_data_timestamp']
self.info(
f"Loading data from MongoDB: {input_data}",
metadata=metadata
)
self.info(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, DATETIME_FORMAT_MS_WITH_TZ)
'inserted_at': {
'$gt': datetime.strptime(last_data_timestamp, DATETIME_FORMAT_MS_WITH_TZ)
}
}
self.debug(
f"Data filter: {data_filter}",
metadata=metadata
)
self.debug(f'Data filter: {data_filter}', metadata=metadata)
data = list(self.database[collection_name].find(
data_filter, {"_id": 0}))
data = list(self.database[collection_name].find(data_filter, {'_id': 0}))
data = clear_mongo_id(data)
self.debug(
f"Collected: {data}",
metadata=metadata
)
self.debug(f'Collected: {data}', metadata=metadata)
for item in data:
item['inserted_at'] = item['inserted_at'].replace(
tzinfo=timezone.utc).strftime(DATETIME_FORMAT_MS_WITH_TZ)
item['inserted_at'] = (
item['inserted_at'].replace(tzinfo=UTC).strftime(DATETIME_FORMAT_MS_WITH_TZ)
)
self.info(
f"Loaded {len(data)} documents from MongoDB",
metadata=metadata
)
self.info(f'Loaded {len(data)} documents from MongoDB', metadata=metadata)
self.debug(
f"Loaded data: {data}",
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",
notification_id='MONGO_LOAD_ERROR',
message=f'Error loading data from MongoDB: {e}',
block='load_latest_data',
level=NotificationLevel.ERROR,
attachment_content=trace
attachment_content=trace,
)
raise e

165
scouter/activities/redis.py
View File

@@ -1,17 +1,20 @@
from temporalio import workflow, activity
from collections.abc import Hashable
from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through():
from logging import Logger
import traceback
from typing import Any
from pandas import DataFrame
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.redis_base import Redis as RedisBase
from sientia_do.observability.logger import Logger
from typing import Any
from pandas import DataFrame
from scouter import metrics
from sientia_do.temporal.activities.redis_base import Redis as RedisBase
from sientia_do.temporal.constants import DATETIME_FORMAT, now
from scouter import metrics
class Redis(RedisBase):
"""
@@ -28,9 +31,15 @@ class Redis(RedisBase):
distributed data processing with fault tolerance and monitoring.
"""
def __init__(self, host: str, port: int,
username: str, password: str,
logger: Logger, notification_handler: NotificationHandler):
def __init__(
self,
host: str,
port: int,
username: str,
password: str,
logger: Logger,
notification_handler: NotificationHandler,
):
"""
Initialize Redis connection and services.
@@ -42,10 +51,9 @@ class Redis(RedisBase):
logger (Logger): Logger instance for operation logging
notification_handler (NotificationHandler): Handler for system notifications
"""
RedisBase.__init__(self, host, port, username,
password, logger, notification_handler)
RedisBase.__init__(self, host, port, username, password, logger, notification_handler)
@activity.defn(name="get_last_data_timestamp")
@activity.defn(name='get_last_data_timestamp')
async def get_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None:
"""
Retrieve the last processed data timestamp from Redis.
@@ -68,34 +76,31 @@ class Redis(RedisBase):
Exception: If Redis operation fails
"""
metadata = input_data['metadata']
key = f"last_data_timestamp:{input_data['workflow_name']}:{input_data['schedule_name']}"
key = f'last_data_timestamp:{input_data["workflow_name"]}:{input_data["schedule_name"]}'
self.info(f"Getting last data timestamp for {key}")
self.info(f'Getting last data timestamp for {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 last data timestamp: {e}",
block="get_last_data_timestamp",
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()
attachment_content=traceback.format_exc(),
)
raise e
self.info(
f"Last collected timestamp: {data_hold}",
metadata=metadata
)
self.info(f'Last collected timestamp: {data_hold}', metadata=metadata)
if not data_hold:
return None
return data_hold
@activity.defn(name="put_last_data_timestamp")
@activity.defn(name='put_last_data_timestamp')
async def put_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None:
"""
Store the last processed data timestamp in Redis.
@@ -119,43 +124,37 @@ class Redis(RedisBase):
Exception: If Redis operation fails
"""
metadata = input_data['metadata']
key = f"last_data_timestamp:{input_data['workflow_name']}:{input_data['schedule_name']}"
key = f'last_data_timestamp:{input_data["workflow_name"]}:{input_data["schedule_name"]}'
self.info(f"Putting last data timestamp for {key}")
self.info(f'Putting last data timestamp for {key}')
data = DataFrame(input_data['data'])
if data.empty:
self.warning("No data to insert",
metadata=metadata
)
self.warning('No data to insert', metadata=metadata)
return None
last_data_timestamp = data['inserted_at'].max()
self.info(
f"Last collected timestamp to insert: {last_data_timestamp}",
metadata=metadata
)
self.info(f'Last collected timestamp to insert: {last_data_timestamp}', metadata=metadata)
try:
self.set(key, last_data_timestamp, ttl=60*60*5)
self.set(key, last_data_timestamp, ttl=60 * 60 * 5)
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",
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()
attachment_content=traceback.format_exc(),
)
raise e
return last_data_timestamp
@activity.defn(name="group_and_hold_data")
async def group_and_hold_data(self, input_data: dict[str, Any]) -> dict[str, Any]:
@activity.defn(name='group_and_hold_data')
async def group_and_hold_data(self, input_data: dict[str, Any]) -> dict[Hashable, Any]:
"""
Group data by tags and store temporarily in Redis with TTL.
@@ -182,109 +181,92 @@ class Redis(RedisBase):
"""
metadata = input_data['metadata']
self.debug("Grouping and holding data...",
metadata=metadata
)
self.debug('Grouping and holding data...', metadata=metadata)
data = DataFrame(input_data['data'])
model_tags = input_data['model_tags']
retention_time = input_data['retention_time']
key = f"held_data_{input_data['workflow_name']}_{input_data['schedule_name']}"
key = f'held_data_{input_data["workflow_name"]}_{input_data["schedule_name"]}'
self.info(f"Getting held data for {key}")
self.info(f'Getting held data for {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",
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()
attachment_content=traceback.format_exc(),
)
raise e
if not data_hold:
data_hold = {}
if data.empty:
self.warning("No data to export",
metadata=metadata
)
self.warning('No data to export', metadata=metadata)
return data_hold
self.info(f"Grouping and holding data for {len(data)} rows")
self.info(f'Grouping and holding data for {len(data)} rows')
try:
# Remove possibly removed tags
tags = list(model_tags.keys())
tags.append('timestamp')
self.debug(
f"Tags to keep: {tags}",
metadata=metadata
)
data_hold = {tag: content for tag,
content in data_hold.items() if tag in tags}
self.debug(
f"Data hold after removing removed tags: {data_hold}",
metadata=metadata
)
self.debug(f'Tags to keep: {tags}', metadata=metadata)
data_hold = {tag: content for tag, content in data_hold.items() if tag in tags}
self.debug(f'Data hold after removing removed tags: {data_hold}', metadata=metadata)
to_register_metrics = []
for _, row in data.iterrows():
value = row['value']
data_hold[row['name']] = value
to_register_metrics.append(
(row['name'], value))
to_register_metrics.append((row['name'], value))
data_hold['timestamp'] = data['timestamp'].max() if not data.empty else \
data_hold['timestamp']
data_hold['timestamp'] = (
data['timestamp'].max() if not data.empty else data_hold['timestamp']
)
self.set(key, data_hold, ttl=retention_time)
# Register metrics
self.debug(
f"Metrics to register: {to_register_metrics}",
metadata=metadata
)
self.debug(f'Metrics to register: {to_register_metrics}', metadata=metadata)
for metric in to_register_metrics:
metrics.TAG_CHANGES_MONITOR.labels(
pod_id=self.pod_id,
model_name=metadata['model_name'],
pipeline_name=metadata['workflow_name'],
tag_name=metric[0]
tag_name=metric[0],
).set(metric[1])
data_hold_df = DataFrame(data_hold, index=[0])
data_hold_melted = data_hold_df.melt(
id_vars='timestamp', var_name='variable', value_name='value')
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)
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",
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()
attachment_content=traceback.format_exc(),
)
raise e
self.info(f"Data held and melted has {len(data_hold_melted)} rows")
self.info(f'Data held and melted has {len(data_hold_melted)} rows')
self.debug(
f"Data held and melted:\n {data_hold_melted.to_string()}",
metadata=metadata
)
self.debug(f'Data held and melted:\n {data_hold_melted.to_string()}', metadata=metadata)
return data_hold_melted.to_dict()
@activity.defn(name="store_data_package")
@activity.defn(name='store_data_package')
async def store_data_package(self, input_data: dict[str, Any]):
"""
Stores the data package in redis. It's a debug feature and must be toggled on.
@@ -296,25 +278,22 @@ class Redis(RedisBase):
data: The data used to collect the data.
"""
metadata = input_data['metadata']
key = f"data_package_{input_data['workflow_name']}_{input_data['schedule_name']}_{now().strftime(DATETIME_FORMAT)}"
key = f'data_package_{input_data["workflow_name"]}_{input_data["schedule_name"]}_{now().strftime(DATETIME_FORMAT)}'
data = DataFrame(input_data['data'])
held_data = DataFrame(input_data['held_data'])
cache = {
'data': data.to_dict(),
'held_data': held_data.to_dict()
}
cache = {'data': data.to_dict(), 'held_data': held_data.to_dict()}
try:
self.set(key, cache, ttl=120)
except Exception as e:
self.send_notification(
metadata=metadata,
notification_id="REDIS_SET_ERROR",
message=f"Error setting data package: {e}",
block="store_data_package",
notification_id='REDIS_SET_ERROR',
message=f'Error setting data package: {e}',
block='store_data_package',
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc()
attachment_content=traceback.format_exc(),
)
raise e