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

Merge pull request #2 from Aignosi/SIENTIAPDE-1005-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas

Browse Source 
Sientiapde 1005 implementar os workflows mapeados utilizando as workers e activities apropriadas
This commit is contained in:
vitor-aignosi
2025-05-20 14:37:24 -03:00
committed by GitHub
parent 3622d6b68c 7822a41040
commit deafb335a1
42 changed files with 3151 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
.env Normal file
View File

@@ -0,0 +1,20 @@
POSTGRES_HOST=postgres
POSTGRES_PORT=5432
POSTGRES_USER=sientia
POSTGRES_PASSWORD=sientia
POSTGRES_DB=sientia
POSTGRES_MIN_CONNECTIONS=5
POSTGRES_MAX_CONNECTIONS=20
KAFKA_BOOTSTRAP_SERVERS=kafka:29092
KAFKA_POLLING_TIME=1000
REDIS_HOST=redis
REDIS_PORT=6379
TEMPORAL_HOST=host.docker.internal:7233
TEMPORAL_NAMESPACE=default
LOG_LEVEL=INFO
PROJECT_NAME=scouter

2
.github/workflows/quality-gate.yml vendored
View File

@@ -63,7 +63,7 @@ jobs:
run: |
python -m pip install --upgrade pip
pip install -r ${{ steps.prepare-requirements.outputs.PROCESSED_REQUIREMENTS_FILE }}
pip install pytest pytest-cov
pip install pytest pytest-cov pytest-asyncio
- name: ⬇️ Setup Node.js 18
uses: actions/setup-node@v4

4
.gitignore vendored
View File

@@ -33,3 +33,7 @@ __pycache__/
*.bak
*.old
.secret
# Ignorar coverage
htmlcov/
.coverage

31
Dockerfile Normal file
View File

@@ -0,0 +1,31 @@
# Use uma imagem base Python
FROM python:3.11-slim
# Instale git e outras dependências do sistema
RUN apt-get update && apt-get install -y git \
&& apt-get install -y build-essential python3-dev \
&& apt-get install -y vim \
&& rm -rf /var/lib/apt/lists/*
# Defina o diretório de trabalho
WORKDIR /app
# Copie os arquivos do projeto
COPY . /app
RUN pip install --upgrade pip setuptools wheel
# Install the required packages
# Add github to known hosts
# This is needed for SSH to work
# The SSH key will NOT remain in the image
# IMPORTANT: this block requires BuildKit
# and the --ssh flag during docker build
RUN --mount=type=ssh \
mkdir -p ~/.ssh && \
ssh-keyscan github.com >> ~/.ssh/known_hosts && \
pip install --no-cache-dir -r requirements.txt
# Defina o comando para executar o worker
CMD ["python", "-m", "scouter.worker.worker"]

96
README.md
View File

@@ -1,2 +1,94 @@
# sientia-dataops-scouter_temporal
Scouter version in Temporal
# Sientia DataOps Scouter
The Sientia DataOps Scouter is a Temporal-based workflow application that processes industrial data from OPC collectors. It aggregates and filters data received from OPC collectors via Kafka topics, direct access to the OPC server, or active trigger (real time applications). The processed data is then stored or forwarded for further analysis.
## Key Features
- Data ingestion from multiple sources:
- OPC collectors through Kafka
- Direct access to OPC servers
- Real-time triggers for immediate processing
- Data aggregation and filtering
- Workflow orchestration using Temporal.io
- Integration with Redis for caching and PostgreSQL for storage
- Scalable deployment using Kubernetes
## Workflows
### Core Scouter
The Core Scouter is the main workflow processes the received data. Steps:
- data_quality_gate: Filters the data received from the OPC collector.
- aggregate_data: Aggregates the data received from the OPC collector.
- group_and_hold_data: Groups the data received from the OPC collector.
- export_data_to_postgres: Exports the data received from the OPC collector to PostgreSQL.
### Scouter
The Scouter is the batch basic workflow that extracts data from the source and processes it using the Core Scouter workflow. Steps:
- load_from_kafka: Loads data from a kafka topic.
- core_scouter: Processes the data using the Core Scouter workflow.
#### Workflow inputs:
- `topic` (str): Kafka topic name where data is received
- `schedule_name` (str): Name of the schedule that triggers the workflow
- `model_name` (str): Name of the model being used for processing
- `model_id` (int): Unique identifier for the model
- `trigger_laborious` (bool): Flag indicating if laborious direct processing is required (real time applications)
- `filters` (dict): Dictionary containing data filtering rules
- `NULL_VALUES_FILTER`: Configuration for handling null values
- `policy`: Policy for null values ("KEEP" or "DISCARD")
- `OUT_OF_BOUNDS_FILTER`: Configuration for handling out-of-bounds values
- `policy`: Policy for out-of-bounds values ("KEEP" or "DISCARD")
- `schema` (str): Database schema name where data will be stored
- `table_name` (str): Name of the table where data will be stored
- `retention_time` (int): Time in seconds that data will be retained in Redis
- `model_tags` (dict): Configuration for different OPC tags
- The key is the tag name and the value is a dictionary containing:
- `data_range`: List of two numbers [min, max] defining valid data range
- `aggr_function`: Aggregation function to use ("lts", "mdn", "avg", "max", "min")
## Fake Data
The Fake Data activity is used to generate fake data for testing purposes. Steps:
- generate_and_send_data: Generates fake data and sends it to a kafka topic.
#### Workflow inputs:
- `topic` (str): Kafka topic name where data is received
## Environment variables
- `POSTGRES_HOST`
- `POSTGRES_PORT`
- `POSTGRES_USER`
- `POSTGRES_PASSWORD`
- `POSTGRES_DBNAME`
- `POSTGRES_MIN_CONNECTIONS`
- `POSTGRES_MAX_CONNECTIONS`
- `KAFKA_BOOTSTRAP_SERVERS`
- `KAFKA_POLLING_TIME`
- `REDIS_HOST`
- `REDIS_PORT`
- `REDIS_USERNAME`
- `REDIS_PASSWORD`
- `LOG_LEVEL`
- `PROJECT_NAME`
- `TEMPORAL_HOST`
- `TEMPORAL_NAMESPACE`
## Application deployment
The application can be deployed using the following command:
```bash
helm upgrade --install sientia-dataops-opc-ingestor sientia/sientia-module -n sientia-opc --create-namespace -f ./values.yaml
```

42
client-schedule.py Normal file
View File

@@ -0,0 +1,42 @@
from temporalio.client import Client, Schedule, ScheduleActionStartWorkflow, ScheduleSpec, ScheduleIntervalSpec
import asyncio
from datetime import timedelta
seconds = 5
async def main():
# Conecta ao servidor Temporal
client = await Client.connect("http://localhost:7233")
for i in range(1, 2):
# Define o agendamento para rodar a cada 5 segundos
schedule = Schedule(
action=ScheduleActionStartWorkflow(
'fake_data', # Nome da classe do workflow no worker.py
# Argumento de entrada (ajuste conforme seu workflow)
{
'topic': 'fake_data'
},
# ID que você define aqui
id=f"test-{i}-{seconds}",
task_queue="fake_data-queue", # Deve coincidir com o worker
),
spec=ScheduleSpec(
intervals=[ScheduleIntervalSpec(
every=timedelta(seconds=seconds))]
),
)
# Cria ou atualiza o schedule no Temporal
# ID único para o schedule
schedule_id = f"test-schedule-c-{i}-every-o-{seconds}s"
try:
await client.create_schedule(schedule_id, schedule)
print(
f"Schedule '{schedule_id}' criado com sucesso. Workflow rodará a cada {seconds} segundos.")
except Exception as e:
print(f"Erro ao criar o schedule: {e}")
if __name__ == "__main__":
asyncio.run(main())

98
docker-compose.yml Normal file
View File

@@ -0,0 +1,98 @@
version: '3.8'
services:
postgres:
image: postgres:15
container_name: postgres
environment:
POSTGRES_USER: sientia
POSTGRES_PASSWORD: sientia
POSTGRES_DB: sientia
ports:
- "5432:5432"
volumes:
- ./postgres_data:/var/lib/postgresql/data
networks:
- sientia-network
zookeeper:
image: confluentinc/cp-zookeeper:7.5.1
container_name: zookeeper
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
ports:
- "2181:2181"
networks:
- sientia-network
kafka:
image: confluentinc/cp-kafka:7.5.1
container_name: kafka
depends_on:
- zookeeper
ports:
- "9092:9092"
- "29092:29092"
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
# Message retention settings (5 minutes)
KAFKA_LOG_RETENTION_MINUTES: 5 # 5 minutes
KAFKA_LOG_RETENTION_MS: 300000 # 5 minutes in milliseconds
KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: 30000 # Check every 30 seconds
networks:
- sientia-network
kafka-ui:
image: provectuslabs/kafka-ui:latest
container_name: kafka-ui
ports:
- "8080:8080"
environment:
KAFKA_CLUSTERS_0_NAME: local
KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
networks:
- sientia-network
redis:
image: redis:latest
ports:
- "${REDIS_PORT}:6379"
networks:
- sientia-network
redis-ui:
image: redislabs/redisinsight:latest
container_name: redis-ui
ports:
- "8001:8001"
networks:
- sientia-network
depends_on:
- redis
# scouter:
# build: .
# container_name: scouter
# networks:
# - sientia-network
# env_file:
# - .env
# depends_on:
# - postgres
# - kafka
# - redis
networks:
sientia-network:
driver: bridge
volumes:
postgres_data:
driver: local

32
input_sample.json Normal file
View File

@@ -0,0 +1,32 @@
{
"topic": "opcua",
"schedule_name": "scouter-opcua-pipeline",
"model_name": "Demo Model",
"model_id": 1,
"trigger_laborious": false,
"filters": {
"NULL_VALUES_FILTER": {
"policy": "KEEP"
},
"OUT_OF_BOUNDS_FILTER": {
"policy": "DISCARD"
}
},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"model_tags": {
"Counter": {
"data_range": [0, 50],
"aggr_function": "lts"
},
"Rollout": {
"data_range": [0, 50],
"aggr_function": "mdn"
},
"Square": {
"data_range": [0, 100],
"aggr_function": "avg"
}
}
}

2
requirements.txt
View File

@@ -1,5 +1,7 @@
temporalio
psycopg2-binary
sqlalchemy
asyncua
redis
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git
git+ssh://git@github.com/Aignosi/sientia-mlops-library.git

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

@@ -0,0 +1,67 @@
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,
username=redis_config['username'],
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
Gates.__init__(
self,
logger=logger,
notification_handler=notification_handler
)
@activity.defn(name="prepare_activity")
async def prepare_activity(self, input_data: dict[str, Any]):
await super().prepare_activity(input_data)

26
scouter/activities/base.py Normal file
View File

@@ -0,0 +1,26 @@
from typing import Any
from logging import Logger
from temporalio import activity
from sientia_do.notifications.handlers import NotificationHandler
class BaseActivity:
def __init__(self, logger: Logger, notification_handler: NotificationHandler):
self.logger = logger
self.notification_handler = notification_handler
@activity.defn(name="prepare_activity")
async def prepare_activity(self, input_data: dict[str, Any]):
"""
Prepare the activity for the notification handler.
Args:
workflow_name (str): The name of the workflow.
schedule_name (str): The name of the schedule.
model_name (str): The name of the model.
model_id (str): The id of the model.
"""
self.notification_handler.base_notification.pipeline_name = input_data['workflow_name']
self.notification_handler.base_notification.schedule_name = input_data['schedule_name']
self.notification_handler.base_notification.model_name = input_data['model_name']
self.notification_handler.base_notification.model_id = input_data['model_id']

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))) # NOSONAR
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())) # 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.logger.info("Success")

211
scouter/activities/gates.py Normal file
View File

@@ -0,0 +1,211 @@
from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
from sientia_do.notifications.models import NotificationLevel
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,
'OUT_OF_BOUNDS_FILTER': out_of_bounds_filter
}
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()
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]:
"""
Data quality gate activity. for each selected filter,
extracts filtered data, discards or keeps filtered data
based on the filter.
Args:
input_data (dict[str, Any]): The data to validate. Contains:
filters (dict[str, str]): The filters to apply. In format:
{filter_name: policy}.
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.
"""
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:
self.logger.warning(f"Filter {filter_name} not found")
continue
try:
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=trace
)
self.logger.error(trace)
else:
if filtered_data.empty:
continue
message = f"{len(filtered_data)} rows has quality issues: {filter_name}: {policy}"
attachment = filtered_data.to_string()
self.notification_handler.build_and_send_notification(
notification_id=f"DATA_QUALITY_GATE_ISSUES__{filter_name}",
message=message,
block="data_quality_gate",
level=NotificationLevel.WARNING,
attachment_content=attachment
)
if policy == "DISCARD":
data = data[~data.index.isin(filtered_data.index)]
self.logger.debug("Data quality gate applied")
return data.to_dict()

70
scouter/activities/kafka.py Normal file
View File

@@ -0,0 +1,70 @@
from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler
from scouter.activities.base import BaseActivity
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)
@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.
"""
self.logger.debug(f"Loading data from topic: {input_data['topic']}")
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.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:
message_values.append(msg.value)
# Return empty dict if no messages were received
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()

85
scouter/activities/postgres.py Normal file
View File

@@ -0,0 +1,85 @@
import traceback
from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
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
from sientia_do.notifications.models import NotificationLevel
from scouter.activities.base import BaseActivity
from typing import Any
class Postgres(BaseActivity):
def __init__(self, host: str, port: int,
user: str, password: str, dbname: str,
min_connections: int, max_connections: int,
logger: Logger, notification_handler: NotificationHandler):
self.host = host
self.port = port
self.user = user
self.password = password
self.dbname = 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)
BaseActivity.__init__(self, logger, notification_handler)
def close(self):
self.engine.dispose()
def __del__(self):
self.close()
@activity.defn(name="export_data_to_postgres")
async def export_data_to_postgres(self, input_data: dict[str, Any]):
"""
Exports data to a postgres table.
Args:
input_data (dict[str, Any]): The data to export. Contains:
schema (str): The schema of the table.
table_name (str): The name of the table.
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"])
with self.session_factory() as session:
try:
data.to_sql(table_name, self.engine, schema=schema,
if_exists="append", index=False)
session.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
)
self.logger.error(trace)
else:
self.logger.debug("Data exported to postgres")
finally:
session.close()

88
scouter/activities/redis.py Normal file
View File

@@ -0,0 +1,88 @@
from temporalio import workflow, activity
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 pandas import DataFrame
from datetime import datetime
class Redis(BaseActivity):
def __init__(self, host: str, port: int,
username: str, password: str,
logger: Logger, notification_handler: NotificationHandler):
self.host = host
self.port = port
self.username = username
self.password = password
self.redis_client = redis.Redis(
host=self.host,
port=self.port,
decode_responses=True,
username=self.username,
password=self.password
)
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']
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()

30
scouter/utils/connectors_config.py Normal file
View File

@@ -0,0 +1,30 @@
from os import getenv
def build_postgres_config():
return {
'host': getenv('POSTGRES_HOST', 'localhost'),
'port': int(getenv('POSTGRES_PORT', '5432')),
'user': getenv('POSTGRES_USER', 'sientia'),
'password': getenv('POSTGRES_PASSWORD', 'sientia'),
'dbname': getenv('POSTGRES_DBNAME', 'sientia'),
'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')),
'max_connections': int(getenv('POSTGRES_MAX_CONNECTIONS', '20'))
}
def build_kafka_config():
return {
'bootstrap_servers': getenv('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092'),
'polling_time': int(getenv('KAFKA_POLLING_TIME', '1000')),
'group_id': 'scouter-group'
}
def build_redis_config():
return {
'host': getenv('REDIS_HOST', 'localhost'),
'port': int(getenv('REDIS_PORT', '6379')),
'username': getenv('REDIS_USERNAME', None),
'password': getenv('REDIS_PASSWORD', None)
}

22
scouter/utils/logger.py Normal file
View File

@@ -0,0 +1,22 @@
from os import getenv
import logging
import sys
def get_logger(name: str):
log_level = getenv('LOG_LEVEL', 'INFO').upper()
logger = logging.getLogger(name)
logger.setLevel(log_level)
stream_handler = logging.StreamHandler(sys.stdout)
stream_handler.setLevel(log_level)
stream_handler.setFormatter(
logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
)
logger.addHandler(stream_handler)
return logger

9
scouter/utils/policies.py Normal file
View File

@@ -0,0 +1,9 @@
from temporalio.common import RetryPolicy
from datetime import timedelta
retry_policy = RetryPolicy(
initial_interval=timedelta(seconds=1),
backoff_coefficient=2.0,
maximum_interval=timedelta(minutes=1),
maximum_attempts=1
)

54
scouter/utils/quality/filters.py Normal file
View File

@@ -0,0 +1,54 @@
from pandas import DataFrame
import numpy as np
from typing import Any
def check_data_range(value: float | int | None, val_range: list) -> bool:
"""
Check if a value is out of a given range.
Args:
value (float | int | None): The value to check.
val_range (list): The range to check against.
Returns:
bool: True if the value is out of the range, False otherwise.
"""
if value is None or np.isnan(value):
return True
bottom = val_range[0]
up = val_range[-1]
return value < bottom or value > up
def out_of_bounds_filter(df: DataFrame, model_tags: dict[str, Any]):
"""
Filter out rows where the value is out of the range.
Args:
df (DataFrame): The DataFrame to filter.
model_tags (dict[str, Any]): The model tags. Contains
the data_range for each tag. If the tag does not have a data_range,
it will be considered as (-inf, inf).
Returns:
DataFrame: The filtered DataFrame.
"""
return df[df.apply(lambda x: check_data_range(
x['value'], model_tags[x['name']].get('data_range', (-np.inf, np.inf))),
axis=1)]
def null_values_filter(df: DataFrame, _model_tags: dict[str, Any]):
"""
Filter out rows where the value is null.
Args:
df (DataFrame): The DataFrame to filter.
Returns:
DataFrame: The filtered DataFrame.
"""
return df[df['value'].isnull()]

101
scouter/worker/worker.py Normal file
View File

@@ -0,0 +1,101 @@
from temporalio import workflow, client
from temporalio.worker import Worker
with workflow.unsafe.imports_passed_through():
import os
from sientia_do.notifications.handlers import NotificationHandler
from scouter.activities.activities import Activities
from scouter.workflow.scouter import Scouter
from scouter.workflow.sub_workflows.core_scouter import CoreScouter
from scouter.workflow.fake_data import FakeData
from scouter.activities.faker import Faker
import asyncio
from scouter.utils.logger import get_logger
from scouter.utils.connectors_config import (
build_postgres_config,
build_kafka_config,
build_redis_config
)
async def main():
host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
logger = get_logger(__name__)
logger.info('Starting Worker...')
logger.info('Starting Notification Handler...')
notification_handler = NotificationHandler(
servers=os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'http://localhost:9092'),
logger=logger,
project_name=os.getenv('PROJECT_NAME', 'scouter'),
pipeline_name='-',
trigger_name='-',
model_name='-',
model='-'
)
logger.info('Starting Activities...')
activities = Activities(
logger=logger,
notification_handler=notification_handler,
postgres_config=build_postgres_config(),
kafka_config=build_kafka_config(),
redis_config=build_redis_config()
)
logger.info('Starting Faker Activities...')
faker_activities = Faker(
logger=logger,
notification_handler=notification_handler,
bootstrap_servers=os.getenv(
'KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092')
)
logger.info('Starting Temporal Client...')
temporal_client = await client.Client.connect(
target_host=host,
namespace=os.getenv('TEMPORAL_NAMESPACE', 'default')
)
logger.info('Starting Workers...')
workers = [
Worker(
temporal_client,
task_queue='scouter-queue',
workflows=[Scouter, CoreScouter],
activities=[
activities.load_from_kafka,
activities.data_quality_gate,
activities.aggregate_data,
activities.group_and_hold_data,
activities.export_data_to_postgres,
activities.prepare_activity,
]
),
Worker(
temporal_client,
task_queue='fake_data-queue',
workflows=[FakeData],
activities=[
faker_activities.generate_and_send_data,
]
)
]
handlers = []
for w in workers:
handlers.append(w.run())
logger.info('Workers started successfully')
await asyncio.gather(*handlers)
if __name__ == '__main__':
asyncio.run(main())

28
scouter/workflow/fake_data.py Normal file
View File

@@ -0,0 +1,28 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from scouter.activities.faker import Faker
from datetime import timedelta
from typing import Dict, Any
from scouter.utils.policies import retry_policy
@workflow.defn(name="fake_data")
class FakeData:
@workflow.run
async def run(self, workflow_input: Dict[str, Any]) -> str:
"""
Generates random data and sends it to a Kafka topic.
Args:
workflow_input (dict[str, Any]): The input data containing:
topic (str): The Kafka topic to send data to
"""
await workflow.execute_activity_method(
Faker.generate_and_send_data,
{
'topic': workflow_input['topic']
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)

64
scouter/workflow/scouter.py Normal file
View File

@@ -0,0 +1,64 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from scouter.activities.activities import Activities
from typing import Any
from datetime import timedelta
from scouter.utils.policies import retry_policy
@workflow.defn(name="scouter")
class Scouter:
@workflow.run
async def run(self, input_data: dict[str, Any]):
"""
Scouter workflow. Loads data from kafka and sends it to the core_scouter
workflow.
Args:
input_data (dict[str, Any]): The data to process. Contains:
topic (str): The topic to load data from.
schedule_name (str): The name of the schedule.
model_name (str): The name of the model.
model_id (str): The id of the model.
trigger_laborious (bool): Whether to trigger laborious.
filters (dict[str, str]): The filters to apply.
schema (str): The schema of the table to export data to.
table_name (str): The name of the table to export data to.
retention_time (int): The retention time for data in redis in seconds.
model_tags (dict[str, Any]): The tags of the model.
And it's respective configuration.
"""
input_data['workflow_name'] = 'scouter'
await workflow.execute_local_activity_method(
Activities.prepare_activity,
{
'workflow_name': input_data['workflow_name'],
'schedule_name': input_data['schedule_name'],
'model_name': input_data['model_name'],
'model_id': input_data['model_id']
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
data = await workflow.execute_activity_method(
Activities.load_from_kafka,
{
'topic': input_data['topic']
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
if data == {}:
return
input_data['data'] = data
await workflow.execute_child_workflow(
'core_scouter',
input_data
)

82
scouter/workflow/sub_workflows/core_scouter.py Normal file
View File

@@ -0,0 +1,82 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from scouter.activities.activities import Activities
from typing import Any
from datetime import timedelta
from scouter.utils.policies import retry_policy
@workflow.defn(name="core_scouter")
class CoreScouter:
@workflow.run
async def run(self, input_data: dict[str, Any]):
"""
Core scouter workflow. Passes data through data_quality_gate,
group_and_hold_data, and then asynchronously exports data to postgres
using export_data_to_postgres and in the future will trigger_laborious
if needed.
Args:
input_data (dict[str, Any]): The data to process. Contains:
workflow_name (str): The name of the workflow.
schedule_name (str): The name of the schedule.
model_name (str): The name of the model.
model_id (str): The id of the model.
data (dict[str, Any]): The data to process.
trigger_laborious (bool): Whether to trigger laborious.
filters (dict[str, str]): The filters to apply.
schema (str): The schema of the table to export data to.
table_name (str): The name of the table to export data to.
retention_time (int): The retention time for data in redis in seconds.
"""
filtered_data = await workflow.execute_local_activity_method(
Activities.data_quality_gate,
{
'filters': input_data['filters'],
'data': input_data['data'],
'model_tags': input_data['model_tags']
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
grouped_data = await workflow.execute_local_activity_method(
Activities.aggregate_data,
{
'data': filtered_data,
'model_tags': input_data['model_tags']
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
held_data = await workflow.execute_local_activity_method(
Activities.group_and_hold_data,
{
'workflow_name': input_data['workflow_name'],
'schedule_name': input_data['schedule_name'],
'data': grouped_data,
'model_id': input_data['model_id'],
'retention_time': input_data['retention_time']
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
if held_data == {}:
return
async_export = workflow.execute_activity_method(
Activities.export_data_to_postgres,
{
'schema': input_data['schema'],
'table_name': input_data['table_name'],
'data': held_data
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
await async_export

0
tests/activities/__init__.py Normal file
View File

151
tests/activities/test_activities.py Normal file
View File

@@ -0,0 +1,151 @@
from unittest.mock import patch, MagicMock, ANY
from scouter.activities.activities import Activities
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 pytest import mark
@patch('scouter.activities.activities.Postgres.__init__')
@patch('scouter.activities.activities.Redis.__init__')
@patch('scouter.activities.activities.Kafka.__init__')
@patch('scouter.activities.activities.Gates.__init__')
def test___init__(mock_gates_init, mock_kafka_init, mock_redis_init, mock_postgres_init):
postgres_config = {
'host': 'localhost',
'port': 5432,
'user': 'postgres',
'password': 'postgres',
'dbname': 'postgres',
'min_connections': 1,
'max_connections': 10
}
redis_config = {
'host': 'localhost',
'port': 6379,
'username': 'redis',
'password': 'redis'
}
kafka_config = {
'bootstrap_servers': 'localhost:9092',
'polling_time': 1000,
'group_id': 'test-group'
}
logger = MagicMock()
notification_handler = MagicMock()
activities = Activities(
postgres_config=postgres_config,
redis_config=redis_config,
kafka_config=kafka_config,
logger=logger,
notification_handler=notification_handler
)
assert isinstance(activities, Activities)
assert isinstance(activities, Postgres)
assert isinstance(activities, Redis)
assert isinstance(activities, Kafka)
assert isinstance(activities, Gates)
mock_postgres_init.assert_called_once_with(
ANY,
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
)
mock_redis_init.assert_called_once_with(
ANY,
host=redis_config['host'],
port=redis_config['port'],
username=redis_config['username'],
password=redis_config['password'],
logger=logger,
notification_handler=notification_handler
)
mock_kafka_init.assert_called_once_with(
ANY,
bootstrap_servers=kafka_config['bootstrap_servers'],
polling_time=kafka_config['polling_time'],
group_id=kafka_config['group_id'],
logger=logger,
notification_handler=notification_handler
)
mock_gates_init.assert_called_once_with(
ANY,
logger=logger,
notification_handler=notification_handler
)
@mark.asyncio
@patch('scouter.activities.activities.Postgres.__init__')
@patch('scouter.activities.activities.Redis.__init__')
@patch('scouter.activities.activities.Kafka.__init__')
async def test_prepare_activity(_mock_kafka_init,
_mock_redis_init, _mock_postgres_init):
postgres_config = {
'host': 'localhost',
'port': 5432,
'user': 'postgres',
'password': 'postgres',
'dbname': 'postgres',
'min_connections': 1,
'max_connections': 10
}
redis_config = {
'host': 'localhost',
'port': 6379,
'username': 'redis',
'password': 'redis'
}
kafka_config = {
'bootstrap_servers': 'localhost:9092',
'polling_time': 1000,
'group_id': 'test-group'
}
logger = MagicMock()
notification_handler = MagicMock()
activities = Activities(
postgres_config=postgres_config,
redis_config=redis_config,
kafka_config=kafka_config,
logger=logger,
notification_handler=notification_handler
)
input_data = {
'workflow_name': 'test-workflow-name',
'schedule_name': 'test-schedule-name',
'model_name': 'test-model-name',
'model_id': 'test-model-id'
}
await activities.prepare_activity(input_data)
assert activities.notification_handler.base_notification.pipeline_name == input_data[
'workflow_name']
assert activities.notification_handler.base_notification.schedule_name == input_data[
'schedule_name']
assert activities.notification_handler.base_notification.model_name == input_data[
'model_name']
assert activities.notification_handler.base_notification.model_id == input_data[
'model_id']

37
tests/activities/test_base.py Normal file
View File

@@ -0,0 +1,37 @@
from unittest.mock import MagicMock
from pytest import fixture, mark
from sientia_do.notifications.models import Notification
from scouter.activities.base import BaseActivity
@fixture
def base_activity():
return BaseActivity(
logger=MagicMock(),
notification_handler=MagicMock(),
)
@mark.asyncio
async def test_prepare_activity(base_activity):
base_activity.notification_handler.base_notification = Notification(
project="project",
pipeline="pipeline",
trigger="-",
model_name="-",
model_id="-",
)
await base_activity.prepare_activity(
{
'workflow_name': 'test_workflow',
'schedule_name': 'test_schedule',
'model_name': 'test_model',
'model_id': 'test_model_id',
}
)
assert base_activity.notification_handler.base_notification.schedule_name == "test_schedule"
assert base_activity.notification_handler.base_notification.model_name == "test_model"
assert base_activity.notification_handler.base_notification.model_id == "test_model_id"
assert base_activity.notification_handler.base_notification.pipeline_name == "test_workflow"

143
tests/activities/test_faker.py Normal file
View File

@@ -0,0 +1,143 @@
from logging import Logger
from unittest.mock import MagicMock, patch, call
import pytest
from sientia_do.notifications.handlers import NotificationHandler
from scouter.activities.faker import Faker
@pytest.fixture
def mock_kafka_producer():
with patch('scouter.activities.faker.KafkaProducer') as mock:
producer = MagicMock()
mock.return_value = producer
yield producer
@pytest.fixture
def mock_datetime():
with patch('scouter.activities.faker.datetime') as mock_dt:
mock_dt.now.return_value.strftime.return_value = '2025-05-14 14:54:24'
yield mock_dt
@pytest.fixture
def faker_instance(mock_kafka_producer):
logger = MagicMock(spec=Logger)
notification_handler = MagicMock(spec=NotificationHandler)
return Faker(
bootstrap_servers='localhost:9092',
logger=logger,
notification_handler=notification_handler
)
@pytest.mark.asyncio
async def test_faker_init(faker_instance, mock_kafka_producer):
"""Test Faker initialization with correct parameters"""
assert faker_instance.producer is not None
assert len(faker_instance.tags) == 6
@pytest.mark.asyncio
async def test_generate_and_send_data_default_count(faker_instance,
mock_kafka_producer, mock_datetime):
"""Test generating data with default message count"""
# Mock random.choice to control the output
with patch('random.choice') as mock_choice, \
patch('random.uniform', return_value=42.5), \
patch('random.randint', return_value=3), \
patch('random.random', return_value=0.5):
# Setup mock for tag and name selection
mock_choice.side_effect = [
'ns=1;i=1001',
'ns=1;i=1002',
'ns=1;i=1003'
]
# Call the method
await faker_instance.generate_and_send_data({'topic': 'test_topic'})
# Verify the producer was called 3 times (default count)
assert mock_kafka_producer.send.call_count == 3
mock_kafka_producer.flush.assert_called_once()
# Verify the message format
expected_data = [{
'tag': 'ns=1;i=1001',
'name': 'Temperature Sensor',
'timestamp': '2025-05-14 14:54:24',
'value': 42.5
}, {
'tag': 'ns=1;i=1002',
'name': 'Vibration Meter',
'timestamp': '2025-05-14 14:54:24',
'value': 42.5
}, {
'tag': 'ns=1;i=1003',
'name': 'Pressure Gauge',
'timestamp': '2025-05-14 14:54:24',
'value': 42.5
}]
mock_kafka_producer.send.assert_has_calls([
call('test_topic', value=expected_data[0]),
call('test_topic', value=expected_data[1]),
call('test_topic', value=expected_data[2])
])
@pytest.mark.asyncio
async def test_generate_and_send_data_custom_count(faker_instance, mock_kafka_producer):
"""Test generating data with custom message count"""
# Call the method with custom count
await faker_instance.generate_and_send_data({
'topic': 'test_topic',
'num_messages': 2
})
# Verify the producer was called 2 times
assert mock_kafka_producer.send.call_count == 2
mock_kafka_producer.flush.assert_called_once()
@pytest.mark.asyncio
async def test_generate_and_send_data_no_topic(faker_instance):
"""Test that ValueError is raised when no topic is provided"""
with pytest.raises(ValueError, match="Topic must be specified in input_data"):
await faker_instance.generate_and_send_data({})
@pytest.mark.asyncio
@patch('scouter.activities.faker.random.random', return_value=0.5)
async def test_generate_and_send_data_random_values(_random, faker_instance, mock_kafka_producer):
"""Test that random values are within expected ranges"""
# Call the method
await faker_instance.generate_and_send_data({'topic': 'test_topic'})
# Get the call arguments
call_args = mock_kafka_producer.send.call_args[1]['value']
# Verify the data structure
assert 'tag' in call_args
assert call_args['tag'] in faker_instance.tags
assert 'value' in call_args
assert 0 <= call_args['value'] <= 100
@pytest.mark.asyncio
@patch('scouter.activities.faker.random.random', return_value=0.05)
async def test_generate_and_send_data_generate_null_values(
_random_mock,
faker_instance,
mock_kafka_producer):
# Call the method
await faker_instance.generate_and_send_data({'topic': 'test_topic',
'num_messages': 1})
# Get the call arguments
call_args = mock_kafka_producer.send.call_args[1]['value']
assert call_args['value'] is None
assert call_args['tag'] in faker_instance.tags
assert 'name' in call_args
assert 'timestamp' in call_args

361
tests/activities/test_gates.py Normal file
View File

@@ -0,0 +1,361 @@
from unittest.mock import Mock, patch, MagicMock, ANY
import numpy as np
import pandas as pd
import pytest
from sientia_do.notifications.models import NotificationLevel
from scouter.activities.gates import Gates
@pytest.fixture
def gates_fixture():
"""Fixture to create a Gates instance with mocked dependencies."""
logger = Mock()
notification_handler = MagicMock()
return Gates(logger=logger, notification_handler=notification_handler)
@pytest.mark.asyncio
async def test_data_quality_gate_with_null_values_filter_discard(gates_fixture):
"""Test data_quality_gate with NULL_VALUES_FILTER and DISCARD policy."""
# Setup test data
input_data = {
'filters': {
'NULL_VALUES_FILTER': 'DISCARD'
},
'data': {
'tag': ['tag1', 'tag2', 'tag3'],
'value': [1.0, None, 3.0],
'timestamp': ['2023-01-01', '2023-01-02', '2023-01-03'],
},
'model_tags': {
'tag1': {'data_range': [0, 100]},
'tag2': {'data_range': [0, 100]},
'tag3': {'data_range': [0, 100]}
}
}
# Execute
result = await gates_fixture.data_quality_gate(input_data)
# Verify
assert len(result['tag']) == 2
assert 'tag2' not in result['tag']
gates_fixture.notification_handler.build_and_send_notification.assert_called_once()
@pytest.mark.asyncio
async def test_data_quality_gate_with_out_of_bounds_filter_keep(gates_fixture):
"""Test data_quality_gate with OUT_OF_BOUNDS_FILTER and KEEP policy."""
# Setup test data with out of bounds values
input_data = {
'filters': {
'OUT_OF_BOUNDS_FILTER': 'KEEP'
},
'data': {
'tag': ['tag1', 'tag2', 'tag3'],
'value': [1.0, 200.0, 3.0],
'timestamp': ['2023-01-01', '2023-01-02', '2023-01-03']
},
'model_tags': {
'tag1': {'data_range': [0, 100]},
'tag2': {'data_range': [0, 100]},
'tag3': {'data_range': [0, 100]}
}
}
# Mock the out_of_bounds_filter to return rows with out of bounds values
with patch('scouter.activities.gates.quality_gate_filters', {
'OUT_OF_BOUNDS_FILTER': lambda df: df[df['tag'] == 'tag2']
}):
# Execute
result = await gates_fixture.data_quality_gate(input_data)
# Verify data is kept but notification is sent
assert len(result['tag']) == 3 # All rows kept
gates_fixture.notification_handler.build_and_send_notification.assert_called_once()
@pytest.mark.asyncio
async def test_data_quality_gate_with_multiple_filters(gates_fixture):
"""Test data_quality_gate with multiple filters."""
# Setup test data
input_data = {
'filters': {
'NULL_VALUES_FILTER': 'DISCARD',
'OUT_OF_BOUNDS_FILTER': 'DISCARD'
},
'data': {
'tag': ['tag1', 'tag2', 'tag3', 'tag4'],
'name': ['tag1', 'tag2', 'tag3', 'tag4'],
'value': [1.0, None, 300.0, 4.0],
'timestamp': ['2023-01-01', '2023-01-02', '2023-01-03', '2023-01-04']
},
'model_tags': {
'tag1': {'data_range': [0, 100]},
'tag2': {'data_range': [0, 100]},
'tag3': {'data_range': [0, 100]},
'tag4': {'data_range': [0, 100]}
}
}
result = await gates_fixture.data_quality_gate(input_data)
# Verify only tag1 and tag4 remain (tag2 has null, tag3 is out of bounds)
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'}}
# Should be called twice (once for each filter)
assert gates_fixture.notification_handler.build_and_send_notification.call_count == 2
@pytest.mark.asyncio
async def test_data_quality_gate_with_unknown_filter(gates_fixture):
"""Test data_quality_gate with an unknown filter."""
# Setup test data with unknown filter
input_data = {
'filters': {
'UNKNOWN_FILTER': 'DISCARD'
},
'data': {
'tag': ['tag1'],
'name': ['tag1'],
'value': [1.0],
'timestamp': ['2023-01-01']
},
'model_tags': {
'tag1': {'data_range': [0, 100]}
}
}
# Execute
result = await gates_fixture.data_quality_gate(input_data)
# Verify data is unchanged and warning is logged
assert len(result['tag']) == 1
gates_fixture.logger.warning.assert_called_once_with(
"Filter UNKNOWN_FILTER not found")
@pytest.mark.asyncio
async def test_data_quality_gate_with_filter_error(gates_fixture):
"""Test data_quality_gate when a filter raises an exception."""
# Setup test data
input_data = {
'filters': {
'NULL_VALUES_FILTER': 'DISCARD'
},
'data': {
'tag': ['tag1'],
'name': ['tag1'],
'value': [1.0],
'timestamp': ['2023-01-01']
},
'model_tags': {
'tag1': {'data_range': [0, 100]}
}
}
# Mock the filter to raise an exception
def failing_filter(_, _model_tags):
raise ValueError("Filter error")
with patch('scouter.activities.gates.quality_gate_filters', {
'NULL_VALUES_FILTER': failing_filter
}):
# Execute
result = await gates_fixture.data_quality_gate(input_data)
# Verify error notification is sent and data is unchanged
assert len(result['tag']) == 1
gates_fixture.notification_handler.build_and_send_notification.assert_called_once()
call_args = gates_fixture.notification_handler.build_and_send_notification.call_args[1]
assert call_args['notification_id'] == "DATA_QUALITY_GATE_ISSUES"
assert call_args['level'] == NotificationLevel.ERROR
assert "Filter error" in call_args['message']
@pytest.mark.asyncio
async def test_data_quality_gate_with_empty_data(gates_fixture):
"""Test data_quality_gate with empty input data."""
# Setup empty input data
input_data = {
'filters': {
'NULL_VALUES_FILTER': 'DISCARD'
},
'data': {
'tag': [],
'name': [],
'value': [],
'timestamp': []
},
'model_tags': {}
}
# Execute
result = await gates_fixture.data_quality_gate(input_data)
# Verify empty result and no notifications
assert len(result['tag']) == 0
gates_fixture.notification_handler.build_and_send_notification.assert_not_called()
@pytest.mark.asyncio
async def test_data_quality_gate_with_no_filters(gates_fixture):
"""Test data_quality_gate with no filters specified."""
# Setup test data with no filters
input_data = {
'filters': {},
'data': {
'tag': ['tag1'],
'name': ['tag1'],
'value': [1.0],
'timestamp': ['2023-01-01']
},
'model_tags': {
'tag1': {'data_range': [0, 100]}
}
}
# Execute
result = await gates_fixture.data_quality_gate(input_data)
# Verify data is unchanged and no notifications
assert len(result['tag']) == 1
gates_fixture.notification_handler.build_and_send_notification.assert_not_called()
@pytest.mark.parametrize(
"group_data, aggr_function, expected_result",
[
# Single value case
(pd.DataFrame({'value': [10.0]}), 'avg', 10.0),
# Multiple values with different aggregation functions
(pd.DataFrame({'value': [1.0, 2.0, 3.0, 4.0]}), 'avg', 2.5),
(pd.DataFrame({'value': [1.0, 2.0, 3.0, 4.0]}), 'mdn', 2.5),
(pd.DataFrame({'value': [1.0, 2.0, 3.0, 4.0]}), 'max', 4.0),
(pd.DataFrame({'value': [1.0, 2.0, 3.0, 4.0]}), 'min', 1.0),
(pd.DataFrame({'value': [1.0, 2.0, 3.0, 4.0]}), 'lts', 4.0),
# With NaN values
(pd.DataFrame({'value': [1.0, np.nan, 3.0, 4.0]}),
'avg', 2.6666666666666665),
# Empty group after dropping NaN
(pd.DataFrame({'value': [np.nan, np.nan]}), 'avg', None),
# Invalid aggregation function
(pd.DataFrame({'value': [1.0, 2.0]}), 'invalid', 'continue'),
]
)
def test_apply_aggregation(gates_fixture, group_data, aggr_function, expected_result):
"""Test apply_aggregation method with various scenarios."""
result = gates_fixture.apply_aggregation(group_data, aggr_function)
assert result == expected_result
# Check notification was sent for invalid function
if aggr_function == 'invalid':
gates_fixture.notification_handler.build_and_send_notification.assert_called_once()
else:
gates_fixture.notification_handler.build_and_send_notification.assert_not_called()
@pytest.mark.asyncio
async def test_aggregate_data(gates_fixture):
"""Test aggregate_data method with multiple groups and aggregation functions."""
input_data = {
'data': [
{'tag': 'tag1', 'name': 'name1', 'value': 1.0, 'timestamp': '2023-01-01'},
{'tag': 'tag1', 'name': 'name1', 'value': 2.0, 'timestamp': '2023-01-02'},
{'tag': 'tag1', 'name': 'name1', 'value': 3.0, 'timestamp': '2023-01-03'},
{'tag': 'tag2', 'name': 'name2', 'value': 4.0, 'timestamp': '2023-01-01'},
{'tag': 'tag2', 'name': 'name2', 'value': 5.0, 'timestamp': '2023-01-02'},
{'tag': 'tag2', 'name': 'name2', 'value': 6.0, 'timestamp': '2023-01-03'},
{'tag': 'tag1', 'name': 'name1',
'value': None, 'timestamp': '2023-01-04'},
],
'model_tags': {
'name1': {'aggr_function': 'avg'},
'name2': {'aggr_function': 'max'},
}
}
# Expected result
expected_result = {'tag': {0: 'tag1', 1: 'tag2'},
'name': {0: 'name1', 1: 'name2'},
'value': {0: 2.0, 1: 6.0},
'timestamp': {0: '2023-01-04', 1: '2023-01-03'},
'aggregation_function': {0: 'avg', 1: 'max'}}
# Execute
result = await gates_fixture.aggregate_data(input_data)
# Verify
assert result == expected_result
gates_fixture.notification_handler.build_and_send_notification.assert_not_called()
@pytest.mark.asyncio
async def test_aggregate_data_with_continue(gates_fixture):
gates_fixture.apply_aggregation = MagicMock(return_value='continue')
input_data = {
'data': [
{'tag': 'tag1', 'name': 'name1', 'value': 1.0, 'timestamp': '2023-01-01'},
{'tag': 'tag1', 'name': 'name1', 'value': 2.0, 'timestamp': '2023-01-02'},
{'tag': 'tag1', 'name': 'name1', 'value': 3.0, 'timestamp': '2023-01-03'},
{'tag': 'tag2', 'name': 'name2', 'value': 4.0, 'timestamp': '2023-01-01'},
{'tag': 'tag2', 'name': 'name2', 'value': 5.0, 'timestamp': '2023-01-02'},
{'tag': 'tag2', 'name': 'name2', 'value': 6.0, 'timestamp': '2023-01-03'},
{'tag': 'tag1', 'name': 'name1',
'value': None, 'timestamp': '2023-01-04'},
],
'model_tags': {
'name1': {'aggr_function': 'avg'},
'name2': {'aggr_function': 'max'},
}
}
# Expected result
expected_result = {}
# Execute
result = await gates_fixture.aggregate_data(input_data)
# Verify
assert result == expected_result
gates_fixture.notification_handler.build_and_send_notification.assert_not_called()
@pytest.mark.asyncio
async def test_aggregate_data_raise_exception(gates_fixture):
gates_fixture.apply_aggregation = MagicMock(
side_effect=Exception("Test exception"))
input_data = {
'data': [
{'tag': 'tag1', 'name': 'name1', 'value': 1.0, 'timestamp': '2023-01-01'},
{'tag': 'tag1', 'name': 'name1', 'value': 2.0, 'timestamp': '2023-01-02'},
{'tag': 'tag1', 'name': 'name1', 'value': 3.0, 'timestamp': '2023-01-03'},
{'tag': 'tag2', 'name': 'name2', 'value': 4.0, 'timestamp': '2023-01-01'},
{'tag': 'tag2', 'name': 'name2', 'value': 5.0, 'timestamp': '2023-01-02'},
{'tag': 'tag2', 'name': 'name2', 'value': 6.0, 'timestamp': '2023-01-03'},
{'tag': 'tag1', 'name': 'name1',
'value': None, 'timestamp': '2023-01-04'},
],
'model_tags': {
'name1': {'aggr_function': 'avg'},
'name2': {'aggr_function': 'max'},
}
}
try:
await gates_fixture.aggregate_data(input_data)
except Exception as e:
assert str(e) == "Test exception"
gates_fixture.notification_handler.build_and_send_notification.assert_called_once_with(
notification_id="AGGREGATION_ISSUES",
message="Error aggregating data: Test exception",
block="aggregate_data",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
else:
assert False

82
tests/activities/test_kafka.py Normal file
View File

@@ -0,0 +1,82 @@
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
)
@mark.asyncio
async def test_load_from_kafka(kafka):
input_data = {"topic": "test-topic"}
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"}
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)

90
tests/activities/test_postgres.py Normal file
View File

@@ -0,0 +1,90 @@
from unittest.mock import ANY, MagicMock, patch
from pytest import fixture
from pytest import mark
from sientia_do.notifications.models import NotificationLevel
from scouter.activities.postgres import Postgres
@fixture
@patch("scouter.activities.postgres.create_engine")
@patch("scouter.activities.postgres.sessionmaker")
def postgres_client(mock_sessionmaker, mock_engine):
# Create a mock session
mock_session = MagicMock()
mock_session.commit = MagicMock()
mock_session.close = MagicMock()
# Configure the session to work with context management
mock_session.__enter__ = MagicMock(return_value=mock_session)
mock_session.__exit__ = MagicMock(return_value=None)
# Configure the sessionmaker to return our mock session
mock_sessionmaker.return_value = mock_session
# Configure the engine to return our mock sessionmaker
mock_engine.return_value = MagicMock()
mock_engine.return_value.dispose = MagicMock()
# Create the Postgres client
client = Postgres(
host="localhost",
port=5432,
user="postgres",
password="postgres",
dbname="postgres",
min_connections=1,
max_connections=10,
logger=MagicMock(),
notification_handler=MagicMock(),
)
# Set up the session factory
client.session_factory = mock_sessionmaker
return client
@mark.asyncio
@patch("scouter.activities.postgres.DataFrame")
async def test_export_data_to_postgres_success(mock_dataframe, postgres_client):
data = {"schema": "test", "table_name": "test",
"data": {"a": [1, 2, 3], "b": [4, 5, 6]}}
await postgres_client.export_data_to_postgres(data)
# Verify notification handler wasn't called
postgres_client.notification_handler.build_and_send_notification.assert_not_called()
# Verify session handling
mock_dataframe.assert_called_once_with(data["data"])
mock_dataframe.return_value.to_sql.assert_called_once_with(
data["table_name"],
postgres_client.engine,
schema=data["schema"],
if_exists="append",
index=False
)
postgres_client.session_factory.return_value.commit.assert_called_once()
postgres_client.session_factory.return_value.close.assert_called_once()
@mark.asyncio
@patch("scouter.activities.postgres.DataFrame", return_value=MagicMock(
to_sql=MagicMock(side_effect=Exception("Error exporting data to postgres"))
))
async def test_export_data_to_postgres_error(_mock_dataframe, postgres_client):
data = {"schema": "test", "table_name": "test",
"data": {"a": [1, 2, 3], "b": [4, 5, 6]}}
await postgres_client.export_data_to_postgres(data)
# Verify error notification was sent
postgres_client.notification_handler.build_and_send_notification.assert_called_once_with(
notification_id="ERROR_EXPORTING_DATA_TO_POSTGRES",
message="Error exporting data to postgres: Error exporting data to postgres",
block="export_data_to_postgres",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
# Verify session handling
postgres_client.session_factory.return_value.close.assert_called_once()

214
tests/activities/test_redis.py Normal file
View File

@@ -0,0 +1,214 @@
import json
from unittest.mock import MagicMock, patch
from datetime import datetime
import pytest
import numpy as np
from pandas import DataFrame
from sientia_do.notifications.handlers import NotificationHandler
from scouter.activities.redis import Redis
@pytest.fixture
@patch('scouter.activities.redis.redis.Redis')
def redis_activity(_mock_redis_client):
logger = MagicMock()
notification_handler = MagicMock(spec=NotificationHandler)
return Redis(host='localhost', port=6379,
logger=logger, notification_handler=notification_handler,
username='test', password='test')
@patch('scouter.activities.redis.redis.Redis')
def test_redis_initialization(mock_redis_client):
"""Test Redis activity initialization"""
redis_activity = Redis(host='localhost', port=6379,
logger=MagicMock(), notification_handler=MagicMock(),
username='test', password='test')
assert redis_activity.host == 'localhost'
assert redis_activity.port == 6379
assert redis_activity.username == 'test'
assert redis_activity.password == 'test'
mock_redis_client.assert_called_once_with(
host='localhost',
port=6379,
decode_responses=True,
username='test',
password='test'
)
def test_get_existing_key(redis_activity):
"""Test getting an existing key from Redis"""
test_data = {'key': 'value'}
redis_activity.redis_client.get.return_value = json.dumps(test_data)
result = redis_activity.get('test_key')
assert result == test_data
redis_activity.redis_client.get.assert_called_once_with('test_key')
def test_get_nonexistent_key(redis_activity):
"""Test getting a non-existent key from Redis"""
redis_activity.redis_client.get.return_value = None
result = redis_activity.get('nonexistent_key')
assert result is None
redis_activity.redis_client.get.assert_called_once_with('nonexistent_key')
def test_set_key(redis_activity):
"""Test setting a key in Redis"""
test_data = {'key': 'value'}
redis_activity.set('test_key', test_data, ttl=300)
redis_activity.redis_client.set.assert_called_once_with(
'test_key',
json.dumps(test_data),
ex=300
)
@pytest.mark.asyncio
async def test_group_and_hold_data_new_key(redis_activity):
"""Test group_and_hold_data with a new key"""
# Setup
test_data = {
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule',
'retention_time': 3600,
'model_id': 1,
'data': DataFrame({
'name': ['sensor1', 'sensor2'],
'value': [25.5, 30.0],
'timestamp': ['2023-01-01 12:00:00'] * 2
}).to_dict('records')
}
# Mock get to return None for new key
redis_activity.get = MagicMock(return_value=None)
redis_activity.set = MagicMock()
# Call the method
result = await redis_activity.group_and_hold_data(test_data)
# Verify the result
expected_result = {
'timestamp': {0: '2023-01-01 12:00:00', 1: '2023-01-01 12:00:00'},
'variable': {0: 'sensor1', 1: 'sensor2'},
'value': {0: 25.5, 1: 30.0},
'model_id': {0: 1, 1: 1}
}
assert result == expected_result
# Verify set was called with correct arguments
redis_activity.set.assert_called_once()
args, kwargs = redis_activity.set.call_args
assert args[0] == 'test_pipeline_test_schedule'
assert args[1] == {
'sensor1': 25.5,
'sensor2': 30.0,
'timestamp': '2023-01-01 12:00:00'
}
assert kwargs['ttl'] == 3600
@pytest.mark.asyncio
async def test_group_and_hold_data_update_existing(redis_activity):
"""Test updating existing data with group_and_hold_data"""
# Setup initial data in Redis
existing_data = {
'sensor1': 20.0,
'sensor2': 28.0,
'timestamp': '2023-01-01 11:00:00'
}
# New data to update with
test_data = {
'workflow_name': 'test_workflow',
'schedule_name': 'test_schedule',
'retention_time': 3600,
'model_id': 1,
'data': DataFrame({
'name': ['sensor1', 'sensor3'],
'value': [25.5, 42.0],
'timestamp': ['2023-01-01 12:00:00'] * 2
}).to_dict('records')
}
# Mock get to return existing data
redis_activity.get = MagicMock(return_value=existing_data)
redis_activity.set = MagicMock()
# Call the method
result = await redis_activity.group_and_hold_data(test_data)
# Verify the result
expected_result = {
'timestamp': {0: '2023-01-01 12:00:00', 1: '2023-01-01 12:00:00', 2: '2023-01-01 12:00:00'},
'variable': {0: 'sensor1', 1: 'sensor2', 2: 'sensor3'},
'value': {0: 25.5, 1: 28.0, 2: 42.0},
'model_id': {0: 1, 1: 1, 2: 1}
}
assert result == expected_result
# Verify set was called with correct arguments
redis_activity.set.assert_called_once()
args, kwargs = redis_activity.set.call_args
assert args[0] == 'test_workflow_test_schedule'
assert args[1] == {
'sensor1': 25.5,
'sensor2': 28.0,
'sensor3': 42.0,
'timestamp': '2023-01-01 12:00:00'
}
assert kwargs['ttl'] == 3600
@pytest.mark.asyncio
async def test_group_and_hold_data_with_none_values(redis_activity):
"""Test handling of None values in group_and_hold_data"""
# Setup test data with None values
test_data = {
'workflow_name': 'test_workflow',
'schedule_name': 'test_schedule',
'retention_time': 3600,
'model_id': 1,
'data': DataFrame({
'name': ['sensor1', 'sensor2'],
'value': [None, 30.0],
'timestamp': [datetime(2023, 1, 1, 12, 0, 0)] * 2
}).to_dict('records')
}
# Mock get to return None for new key
redis_activity.get = MagicMock(return_value=None)
redis_activity.set = MagicMock()
# Call the method
result = await redis_activity.group_and_hold_data(test_data)
# Verify None was converted to np.nan and values are as expected
assert np.isnan(result['value'][0])
assert result['value'][1] == pytest.approx(30.0)
@pytest.mark.asyncio
async def test_group_and_hold_data_empty_dataframe(redis_activity):
"""Test group_and_hold_data with empty DataFrame"""
# Setup test with empty data
test_data = {
'workflow_name': 'test_workflow',
'schedule_name': 'test_schedule',
'retention_time': 3600,
'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records')
}
redis_activity.get = MagicMock(return_value=None)
# Call the method
result = await redis_activity.group_and_hold_data(test_data)
assert result == {}

0
tests/utils/__init__.py Normal file
View File

132
tests/utils/quality/test_filters.py Normal file
View File

@@ -0,0 +1,132 @@
import pytest
import pandas as pd
import numpy as np
from pandas.testing import assert_frame_equal
from scouter.utils.quality.filters import check_data_range, out_of_bounds_filter, null_values_filter
# Fixtures
@pytest.fixture
def sample_dataframe():
"""Fixture providing a sample DataFrame for testing."""
return pd.DataFrame({
'tag': ['temp', 'temp', 'pressure', 'pressure', 'humidity', 'wind_speed'],
'name': ['temp', 'temp', 'pressure', 'pressure', 'humidity', 'wind_speed'],
'value': [25, 35, 95, 105, 60, None],
'timestamp': pd.date_range(start='2023-01-01', periods=6)
})
@pytest.fixture
def nodes_data_range():
"""Fixture providing data ranges for different tags."""
return {
'temp': {'data_range': [10, 30]},
'pressure': {'data_range': [90, 100]},
'humidity': {'data_range': [40, 80]},
'wind_speed': {'data_range': [0, 50]}
}
# Parameterized test data
CHECK_DATA_RANGE_CASES = [
# (value, val_range, expected)
# Values within range
(5, [0, 10], False),
(0, [0, 10], False), # Edge case: value equals lower bound
(10, [0, 10], False), # Edge case: value equals upper bound
# Values outside range
(-1, [0, 10], True),
(11, [0, 10], True),
# Single value range
(5, [5, 5], False),
(4, [5, 5], True),
# Empty or None value
(None, [0, 10], True),
(np.nan, [0, 10], True),
]
# Tests for check_data_range
@pytest.mark.parametrize('value,val_range,expected', CHECK_DATA_RANGE_CASES)
def test_check_data_range(value, val_range, expected):
"""Test the check_data_range function with various input scenarios."""
result = check_data_range(value, val_range)
if isinstance(value, float) and np.isnan(value):
assert result is True
else:
assert result == expected
# Tests for out_of_bounds_filter
def test_out_of_bounds_filter(sample_dataframe, nodes_data_range):
"""Test filtering out-of-bounds values from a DataFrame."""
# Expected result: rows where value is outside the defined range
expected_data = {
'tag': ['temp', 'pressure', 'wind_speed'],
'name': ['temp', 'pressure', 'wind_speed'],
'value': [35, 105, None],
'timestamp': [
pd.Timestamp('2023-01-02'),
pd.Timestamp('2023-01-04'),
pd.Timestamp('2023-01-06')
]
}
expected_df = pd.DataFrame(expected_data)
result = out_of_bounds_filter(sample_dataframe, nodes_data_range)
result = result.reset_index(drop=True)
expected_df = expected_df.reset_index(drop=True)
assert_frame_equal(result, expected_df)
def test_out_of_bounds_filter_empty_df(nodes_data_range):
"""Test with an empty DataFrame."""
df = pd.DataFrame(columns=['tag', 'name', 'value', 'timestamp'])
result = out_of_bounds_filter(df, nodes_data_range)
assert result.empty
assert list(result.columns) == ['tag', 'name', 'value', 'timestamp']
# Tests for null_values_filter
def test_null_values_filter(sample_dataframe, nodes_data_range):
"""Test filtering null values from a DataFrame."""
expected_data = {
'tag': ['wind_speed'],
'name': ['wind_speed'],
'value': [None],
'timestamp': [pd.Timestamp('2023-01-06')]
}
expected_df = pd.DataFrame(expected_data)
result = null_values_filter(sample_dataframe, nodes_data_range)
result = result.reset_index(drop=True)
expected_df = expected_df.reset_index(drop=True)
assert_frame_equal(result, expected_df, check_dtype=False)
def test_null_values_filter_no_nulls(nodes_data_range):
"""Test with a DataFrame containing no null values."""
df = pd.DataFrame({
'tag': ['temp', 'pressure'],
'name': ['temp', 'pressure'],
'value': [25, 100],
'timestamp': pd.date_range(start='2023-01-01', periods=2)
})
result = null_values_filter(df, nodes_data_range)
assert result.empty
assert list(result.columns) == ['tag', 'name', 'value', 'timestamp']
def test_null_values_filter_empty_df(nodes_data_range):
"""Test with an empty DataFrame."""
df = pd.DataFrame(columns=['tag', 'name', 'value', 'timestamp'])
result = null_values_filter(df, nodes_data_range)
assert result.empty
assert list(result.columns) == ['tag', 'name', 'value', 'timestamp']

115
tests/utils/test_connectors_config.py Normal file
View File

@@ -0,0 +1,115 @@
import os
from unittest.mock import patch
import pytest
from scouter.utils.connectors_config import (
build_postgres_config,
build_kafka_config,
build_redis_config
)
@pytest.fixture
def mock_env_vars():
with patch.dict(os.environ, {}, clear=True):
yield
@pytest.mark.usefixtures("mock_env_vars")
def test_build_postgres_config_defaults():
"""Test that build_postgres_config returns default values when no env vars are set"""
config = build_postgres_config()
assert config == {
'host': 'localhost',
'port': 5432,
'user': 'sientia',
'password': 'sientia',
'dbname': 'sientia',
'min_connections': 5,
'max_connections': 20
}
@pytest.mark.usefixtures("mock_env_vars")
def test_build_postgres_config_with_env_vars():
"""Test that build_postgres_config uses env vars when set"""
with patch.dict(os.environ, {
'POSTGRES_HOST': 'db.example.com',
'POSTGRES_PORT': '5433',
'POSTGRES_USER': 'admin',
'POSTGRES_PASSWORD': 'secret',
'POSTGRES_DBNAME': 'test_db',
'POSTGRES_MIN_CONNECTIONS': '3',
'POSTGRES_MAX_CONNECTIONS': '15'
}):
config = build_postgres_config()
assert config == {
'host': 'db.example.com',
'port': 5433,
'user': 'admin',
'password': 'secret',
'dbname': 'test_db',
'min_connections': 3,
'max_connections': 15
}
@pytest.mark.usefixtures("mock_env_vars")
def test_build_kafka_config_defaults():
"""Test that build_kafka_config returns default values when no env vars are set"""
config = build_kafka_config()
assert config == {
'bootstrap_servers': 'localhost:9092',
'polling_time': 1000,
'group_id': 'scouter-group'
}
@pytest.mark.usefixtures("mock_env_vars")
def test_build_kafka_config_with_env_vars():
"""Test that build_kafka_config uses env vars when set"""
with patch.dict(os.environ, {
'KAFKA_BOOTSTRAP_SERVERS': 'kafka.example.com:9092',
'KAFKA_POLLING_TIME': '5000'
}):
config = build_kafka_config()
assert config == {
'bootstrap_servers': 'kafka.example.com:9092',
'polling_time': 5000,
'group_id': 'scouter-group'
}
@pytest.mark.usefixtures("mock_env_vars")
def test_build_redis_config_defaults():
"""Test that build_redis_config returns default values when no env vars are set"""
config = build_redis_config()
assert config == {
'host': 'localhost',
'port': 6379,
'username': None,
'password': None
}
@pytest.mark.usefixtures("mock_env_vars")
def test_build_redis_config_with_env_vars():
"""Test that build_redis_config uses env vars when set"""
with patch.dict(os.environ, {
'REDIS_HOST': 'redis.example.com',
'REDIS_PORT': '6380',
'REDIS_USERNAME': 'test',
'REDIS_PASSWORD': 'test'
}):
config = build_redis_config()
assert config == {
'host': 'redis.example.com',
'port': 6380,
'username': 'test',
'password': 'test'
}

37
tests/utils/test_logger.py Normal file
View File

@@ -0,0 +1,37 @@
import os
from unittest.mock import patch
import logging
import pytest
from scouter.utils.logger import get_logger
@pytest.fixture
def mock_env_vars():
with patch.dict(os.environ, {}, clear=True):
yield
@pytest.mark.usefixtures("mock_env_vars")
@patch('scouter.utils.logger.logging.Formatter')
@patch('scouter.utils.logger.logging.StreamHandler')
def test_get_logger_defaults(mock_stream_handler, mock_formatter):
"""Test logger creation with default settings"""
# Mock the StreamHandler and Formatter
logger = get_logger('test_logger')
# Verify logger settings
assert logger.name == 'test_logger'
assert logger.level == logging.INFO
# Verify handler configuration
mock_stream_handler.return_value.setLevel.assert_called_once_with('INFO')
mock_stream_handler.return_value.setFormatter.assert_called_once()
# Verify formatter configuration
mock_formatter.assert_called_once_with(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# Verify handler was added to logger
assert len(logger.handlers) == 1

0
tests/worker/__init__.py Normal file
View File

0
tests/workflow/__init__.py Normal file
View File

136
tests/workflow/sub_workflows/test_core_scouter.py Normal file
View File

@@ -0,0 +1,136 @@
from unittest.mock import AsyncMock, patch, call, ANY
import pytest
from scouter.workflow.sub_workflows.core_scouter import CoreScouter
from scouter.activities.activities import Activities
@pytest.fixture
def core_scouter():
return CoreScouter()
@pytest.mark.asyncio
@patch('scouter.workflow.sub_workflows.core_scouter.workflow', new_callable=AsyncMock)
async def test_core_scouter_workflow_success(mock_workflow, core_scouter):
mock_workflow.execute_local_activity_method.side_effect = [
'filtered_data', 'grouped_data', 'held_data']
await core_scouter.run(
input_data={
'workflow_name': 'test_workflow',
'schedule_name': 'test_schedule',
'model_name': 'test_model',
'model_id': 'test_model_id',
'data': 'test_data',
'trigger_laborious': False,
'filters': {'test_filter': 'test_value'},
'schema': 'test_schema',
'table_name': 'test_table',
'retention_time': 3600,
'model_tags': {}
}
)
mock_workflow.execute_local_activity_method.assert_has_calls([
call(
Activities.data_quality_gate,
{
'filters': {'test_filter': 'test_value'},
'data': 'test_data',
'model_tags': {}
},
retry_policy=ANY,
start_to_close_timeout=ANY
)])
mock_workflow.execute_local_activity_method.assert_has_calls([
call(
Activities.aggregate_data,
{
'data': 'filtered_data',
'model_tags': {}
},
retry_policy=ANY,
start_to_close_timeout=ANY
)])
mock_workflow.execute_local_activity_method.assert_has_calls([
call(
Activities.group_and_hold_data,
{
'workflow_name': 'test_workflow',
'schedule_name': 'test_schedule',
'data': 'grouped_data',
'model_id': 'test_model_id',
'retention_time': 3600
},
retry_policy=ANY,
start_to_close_timeout=ANY
)])
mock_workflow.execute_activity_method.assert_has_calls([
call(
Activities.export_data_to_postgres,
{
'schema': 'test_schema',
'table_name': 'test_table',
'data': 'held_data'},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
@pytest.mark.asyncio
@patch('scouter.workflow.sub_workflows.core_scouter.workflow', new_callable=AsyncMock)
async def test_core_scouter_workflow_with_empty_data(mock_workflow, core_scouter):
mock_workflow.execute_local_activity_method.return_value = {}
await core_scouter.run(
input_data={
'workflow_name': 'test_workflow',
'schedule_name': 'test_schedule',
'model_name': 'test_model',
'model_id': 'test_model_id',
'data': 'test_data',
'trigger_laborious': False,
'filters': {'test_filter': 'test_value'},
'schema': 'test_schema',
'table_name': 'test_table',
'retention_time': 3600,
'model_tags': {}
}
)
mock_workflow.execute_local_activity_method.assert_has_calls([
call(
Activities.data_quality_gate,
{
'filters': {'test_filter': 'test_value'},
'data': 'test_data',
'model_tags': {}
},
retry_policy=ANY,
start_to_close_timeout=ANY
)])
mock_workflow.execute_local_activity_method.assert_has_calls([
call(
Activities.aggregate_data,
{
'data': {},
'model_tags': {}
},
retry_policy=ANY,
start_to_close_timeout=ANY
)])
mock_workflow.execute_local_activity_method.assert_has_calls([
call(
Activities.group_and_hold_data,
{
'workflow_name': 'test_workflow',
'schedule_name': 'test_schedule',
'data': {},
'model_id': 'test_model_id',
'retention_time': 3600
},
retry_policy=ANY,
start_to_close_timeout=ANY
)])
assert mock_workflow.execute_local_activity_method.call_count == 3

29
tests/workflow/test_fake_data.py Normal file
View File

@@ -0,0 +1,29 @@
from unittest.mock import AsyncMock, patch, ANY
from pytest import fixture, mark
from scouter.workflow.fake_data import FakeData
from scouter.activities.faker import Faker
@fixture
def fake_data():
return FakeData()
@mark.asyncio
@patch('scouter.workflow.fake_data.workflow', new_callable=AsyncMock)
async def test_fake_data_workflow(mock_workflow, fake_data):
mock_workflow.execute_activity_method.return_value = None
await fake_data.run(
{
'topic': 'test_topic'
}
)
mock_workflow.execute_activity_method.assert_called_once_with(
Faker.generate_and_send_data,
{
'topic': 'test_topic'
},
retry_policy=ANY,
start_to_close_timeout=ANY
)

94
tests/workflow/test_scouter.py Normal file
View File

@@ -0,0 +1,94 @@
from unittest.mock import AsyncMock, patch, ANY
from pytest import fixture, mark
from scouter.workflow.scouter import Scouter
from scouter.activities.activities import Activities
@fixture
def scouter():
return Scouter()
@mark.asyncio
@patch('scouter.workflow.scouter.workflow', new_callable=AsyncMock)
async def test_scouter_workflow(mock_workflow, scouter):
mock_workflow.execute_activity_method.return_value = 'test_data'
await scouter.run(
input_data={
'topic': 'test_topic',
'schedule_name': 'test_schedule',
'model_name': 'test_model',
'model_id': 'test_model_id'
}
)
mock_workflow.execute_local_activity_method.assert_called_once_with(
Activities.prepare_activity,
{
'workflow_name': 'scouter',
'schedule_name': 'test_schedule',
'model_name': 'test_model',
'model_id': 'test_model_id'
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
mock_workflow.execute_activity_method.assert_called_once_with(
Activities.load_from_kafka,
{
'topic': 'test_topic'
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
mock_workflow.execute_child_workflow.assert_called_once_with(
'core_scouter',
{
'topic': 'test_topic',
'data': 'test_data',
'workflow_name': 'scouter',
'schedule_name': 'test_schedule',
'model_name': 'test_model',
'model_id': 'test_model_id'
}
)
@mark.asyncio
@patch('scouter.workflow.scouter.workflow', new_callable=AsyncMock)
async def test_scouter_workflow_empty(mock_workflow, scouter):
mock_workflow.execute_activity_method.return_value = {}
await scouter.run(
input_data={
'topic': 'test_topic',
'schedule_name': 'test_schedule',
'model_name': 'test_model',
'model_id': 'test_model_id'
}
)
mock_workflow.execute_local_activity_method.assert_called_once_with(
Activities.prepare_activity,
{
'workflow_name': 'scouter',
'schedule_name': 'test_schedule',
'model_name': 'test_model',
'model_id': 'test_model_id'
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
mock_workflow.execute_activity_method.assert_called_once_with(
Activities.load_from_kafka,
{
'topic': 'test_topic'
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
mock_workflow.execute_child_workflow.assert_not_called()

188
values.yaml
View File

@@ -0,0 +1,188 @@
# Default values for sientia-module.
# This is a YAML-formatted file.
# Declare variables to be passed into your templates.
# This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/
replicaCount: 1
# This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
image:
repository: aignosi.azurecr.io/sientia-module
# This sets the pull policy for images.
pullPolicy: Always
# Overrides the image tag whose default is the chart appVersion.
tag: "0.0.2"
# 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:
- name: docker-hub-secret
# This is to override the chart name.
nameOverride: "sientia-scouter-worker"
fullnameOverride: "sientia-scouter-worker"
namespace: sientia
# This section builds out the service account more information can be found here: https://kubernetes.io/docs/concepts/security/service-accounts/
serviceAccount:
# Specifies whether a service account should be created
create: true
# Automatically mount a ServiceAccount's API credentials?
automount: true
# Annotations to add to the service account
annotations: {}
# The name of the service account to use.
# If not set and create is true, a name is generated using the fullname template
name: "sientia-scouter-worker"
# This is for setting Kubernetes Annotations to a Pod.
# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
podAnnotations: {}
# This is for setting Kubernetes Labels to a Pod.
# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
podLabels: {}
podSecurityContext: {}
# fsGroup: 2000
securityContext: {}
# capabilities:
# drop:
# - ALL
# readOnlyRootFilesystem: true
# runAsNonRoot: true
# runAsUser: 1000
resources: {}
# We usually recommend not to specify default resources and to leave this as a conscious
# choice for the user. This also increases chances charts run on environments with little
# resources, such as Minikube. If you do want to specify resources, uncomment the following
# lines, adjust them as necessary, and remove the curly braces after 'resources:'.
# limits:
# cpu: 100m
# memory: 128Mi
# requests:
# cpu: 100m
# memory: 128Mi
# This is to setup the liveness and readiness probes more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/
livenessProbe:
exec:
command:
- sh
- -c
- pgrep -f "scouter.worker.worker"
initialDelaySeconds: 20
periodSeconds: 30
readinessProbe:
exec:
command:
- sh
- -c
- pgrep -f "scouter.worker.worker"
initialDelaySeconds: 10
periodSeconds: 15
# This section is for setting up autoscaling more information can be found here: https://kubernetes.io/docs/concepts/workloads/autoscaling/
autoscaling:
enabled: false
minReplicas: 1
maxReplicas: 100
targetCPUUtilizationPercentage: 80
# targetMemoryUtilizationPercentage: 80
# Additional volumes on the output Deployment definition.
volumes: []
# - name: foo
# secret:
# secretName: mysecret
# optional: false
# Additional volumeMounts on the output Deployment definition.
volumeMounts: []
# - name: foo
# mountPath: "/etc/foo"
# readOnly: true
nodeSelector: {}
tolerations: []
affinity: {}
service:
enabled: false
type: ClusterIP
port: 4840
targetPort: 4840
env:
# Entrypoint variables
- name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git"
- name: GITHUB_BRANCH
value: "SIENTIAPDE-1005-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas"
- name: PYTHON_APP
value: "scouter.worker.worker"
# Application variables
- name: POSTGRES_HOST
value: "paradedb-rw.paradedb.svc.cluster.local"
- name: POSTGRES_PORT
value: "5432"
- name: POSTGRES_USER
value: "sientia"
- name: POSTGRES_PASSWORD
value: "sientia"
- name: POSTGRES_DBNAME
value: "sientia"
- name: POSTGRES_MIN_CONNECTIONS
value: "10"
- name: POSTGRES_MAX_CONNECTIONS
value: "20"
- name: KAFKA_BOOTSTRAP_SERVERS
value: "kafka.kafka.svc.cluster.local:9092"
- name: KAFKA_POLLING_TIME
value: "1000"
- name: REDIS_HOST
value: "redis-master.redis.svc.cluster.local"
- name: REDIS_PORT
value: "6379"
- name: REDIS_USERNAME
valueFrom:
secretKeyRef:
name: redis
key: redis-username
- name: REDIS_PASSWORD
valueFrom:
secretKeyRef:
name: redis
key: redis-password
- name: LOG_LEVEL
value: "INFO"
- name: PROJECT_NAME
value: "sientia-scouter"
- name: TEMPORAL_HOST
value: "temporal-frontend.temporal.svc.cluster.local:7233"
- name: TEMPORAL_NAMESPACE
value: "default"
ssh:
enabled: true
secretName: git-ssh-key-sientia-scouter-worker
sshPath: /mnt/.ssh
knownHostsPath: /mnt/known_hosts
# kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp
# helm upgrade --install sientia-scouter-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.1.0-uat
# kubectl create secret generic git-ssh-key-sientia-scouter-worker \
# --namespace sientia \
# --from-file=ssh-privatekey=git_key \
# --type=kubernetes.io/ssh-auth