diff --git a/.env b/.env new file mode 100644 index 0000000..432cc6e --- /dev/null +++ b/.env @@ -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 diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index 028daff..cfb497b 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -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 diff --git a/.gitignore b/.gitignore index a34260d..a446ab3 100644 --- a/.gitignore +++ b/.gitignore @@ -32,4 +32,8 @@ __pycache__/ *.tmp *.bak *.old -.secret \ No newline at end of file +.secret + +# Ignorar coverage +htmlcov/ +.coverage \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..c7c1621 --- /dev/null +++ b/Dockerfile @@ -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"] \ No newline at end of file diff --git a/README.md b/README.md index de42781..c16d207 100644 --- a/README.md +++ b/README.md @@ -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 +``` diff --git a/client-schedule.py b/client-schedule.py new file mode 100644 index 0000000..984fbf8 --- /dev/null +++ b/client-schedule.py @@ -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()) diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..91d3569 --- /dev/null +++ b/docker-compose.yml @@ -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 \ No newline at end of file diff --git a/input_sample.json b/input_sample.json new file mode 100644 index 0000000..c3d55ce --- /dev/null +++ b/input_sample.json @@ -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" + } + } +} \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 1ccf40c..90e049b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -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 diff --git a/scouter/activities/activities.py b/scouter/activities/activities.py new file mode 100644 index 0000000..fa4d9c1 --- /dev/null +++ b/scouter/activities/activities.py @@ -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) diff --git a/scouter/activities/base.py b/scouter/activities/base.py new file mode 100644 index 0000000..3adb0e4 --- /dev/null +++ b/scouter/activities/base.py @@ -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'] diff --git a/scouter/activities/faker.py b/scouter/activities/faker.py new file mode 100644 index 0000000..deb89ad --- /dev/null +++ b/scouter/activities/faker.py @@ -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") diff --git a/scouter/activities/gates.py b/scouter/activities/gates.py new file mode 100644 index 0000000..c9f1395 --- /dev/null +++ b/scouter/activities/gates.py @@ -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() diff --git a/scouter/activities/kafka.py b/scouter/activities/kafka.py new file mode 100644 index 0000000..80ddd70 --- /dev/null +++ b/scouter/activities/kafka.py @@ -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() diff --git a/scouter/activities/postgres.py b/scouter/activities/postgres.py new file mode 100644 index 0000000..53ca32e --- /dev/null +++ b/scouter/activities/postgres.py @@ -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() diff --git a/scouter/activities/redis.py b/scouter/activities/redis.py new file mode 100644 index 0000000..6af1225 --- /dev/null +++ b/scouter/activities/redis.py @@ -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() diff --git a/scouter/utils/connectors_config.py b/scouter/utils/connectors_config.py new file mode 100644 index 0000000..6acc4ae --- /dev/null +++ b/scouter/utils/connectors_config.py @@ -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) + } diff --git a/scouter/utils/logger.py b/scouter/utils/logger.py new file mode 100644 index 0000000..42a9cfd --- /dev/null +++ b/scouter/utils/logger.py @@ -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 diff --git a/scouter/utils/policies.py b/scouter/utils/policies.py new file mode 100644 index 0000000..9230f31 --- /dev/null +++ b/scouter/utils/policies.py @@ -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 +) diff --git a/scouter/utils/quality/filters.py b/scouter/utils/quality/filters.py new file mode 100644 index 0000000..50dd2e6 --- /dev/null +++ b/scouter/utils/quality/filters.py @@ -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()] diff --git a/scouter/worker/worker.py b/scouter/worker/worker.py new file mode 100644 index 0000000..bc2eaab --- /dev/null +++ b/scouter/worker/worker.py @@ -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()) diff --git a/scouter/workflow/fake_data.py b/scouter/workflow/fake_data.py new file mode 100644 index 0000000..6ff347a --- /dev/null +++ b/scouter/workflow/fake_data.py @@ -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) + ) diff --git a/scouter/workflow/scouter.py b/scouter/workflow/scouter.py new file mode 100644 index 0000000..72596db --- /dev/null +++ b/scouter/workflow/scouter.py @@ -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 + ) diff --git a/scouter/workflow/sub_workflows/core_scouter.py b/scouter/workflow/sub_workflows/core_scouter.py new file mode 100644 index 0000000..f1d3dd6 --- /dev/null +++ b/scouter/workflow/sub_workflows/core_scouter.py @@ -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 diff --git a/tests/activities/__init__.py b/tests/activities/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/activities/test_activities.py b/tests/activities/test_activities.py new file mode 100644 index 0000000..49d555f --- /dev/null +++ b/tests/activities/test_activities.py @@ -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'] diff --git a/tests/activities/test_base.py b/tests/activities/test_base.py new file mode 100644 index 0000000..02e50fc --- /dev/null +++ b/tests/activities/test_base.py @@ -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" diff --git a/tests/activities/test_faker.py b/tests/activities/test_faker.py new file mode 100644 index 0000000..5a35abf --- /dev/null +++ b/tests/activities/test_faker.py @@ -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 diff --git a/tests/activities/test_gates.py b/tests/activities/test_gates.py new file mode 100644 index 0000000..9569d57 --- /dev/null +++ b/tests/activities/test_gates.py @@ -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 diff --git a/tests/activities/test_kafka.py b/tests/activities/test_kafka.py new file mode 100644 index 0000000..d38abfd --- /dev/null +++ b/tests/activities/test_kafka.py @@ -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) diff --git a/tests/activities/test_postgres.py b/tests/activities/test_postgres.py new file mode 100644 index 0000000..fac1120 --- /dev/null +++ b/tests/activities/test_postgres.py @@ -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() diff --git a/tests/activities/test_redis.py b/tests/activities/test_redis.py new file mode 100644 index 0000000..ea9e453 --- /dev/null +++ b/tests/activities/test_redis.py @@ -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 == {} diff --git a/tests/utils/__init__.py b/tests/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/utils/quality/test_filters.py b/tests/utils/quality/test_filters.py new file mode 100644 index 0000000..c08ab2c --- /dev/null +++ b/tests/utils/quality/test_filters.py @@ -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'] diff --git a/tests/utils/test_connectors_config.py b/tests/utils/test_connectors_config.py new file mode 100644 index 0000000..bd1694b --- /dev/null +++ b/tests/utils/test_connectors_config.py @@ -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' + } diff --git a/tests/utils/test_logger.py b/tests/utils/test_logger.py new file mode 100644 index 0000000..b931714 --- /dev/null +++ b/tests/utils/test_logger.py @@ -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 diff --git a/tests/worker/__init__.py b/tests/worker/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/workflow/__init__.py b/tests/workflow/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/workflow/sub_workflows/test_core_scouter.py b/tests/workflow/sub_workflows/test_core_scouter.py new file mode 100644 index 0000000..a87d063 --- /dev/null +++ b/tests/workflow/sub_workflows/test_core_scouter.py @@ -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 diff --git a/tests/workflow/test_fake_data.py b/tests/workflow/test_fake_data.py new file mode 100644 index 0000000..f55e942 --- /dev/null +++ b/tests/workflow/test_fake_data.py @@ -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 + ) diff --git a/tests/workflow/test_scouter.py b/tests/workflow/test_scouter.py new file mode 100644 index 0000000..f1402c7 --- /dev/null +++ b/tests/workflow/test_scouter.py @@ -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() diff --git a/values.yaml b/values.yaml index e69de29..ca9f0bd 100644 --- a/values.yaml +++ b/values.yaml @@ -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 \ No newline at end of file