diff --git a/README.md b/README.md index 08853ab..0cad971 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Sientia DataOps Scouter -A high-performance, scalable data processing system built on Temporal.io for industrial data collection, processing, and analytics. The Scouter system provides enterprise-grade data ingestion from multiple sources with automatic data quality validation, aggregation, and export capabilities. +A high-performance, scalable data processing and ML model orchestration system built on Temporal.io for industrial data collection, processing, and analytics. The Scouter system provides enterprise-grade data ingestion from multiple sources with automatic data quality validation, aggregation, and export capabilities. ## Features @@ -17,7 +17,8 @@ A high-performance, scalable data processing system built on Temporal.io for ind - **Configurable Data Retention**: Redis-based temporary storage with TTL management - **Notification System**: Integrated alerting and notification management via MongoDB - **Scalable Architecture**: Kubernetes-ready deployment with horizontal scaling support -- **Test Data Generation**: Built-in fake data generation for development and testing +- **Debug Mode**: Optional data package storage for debugging and troubleshooting +- **Worker Autoscaling**: Configurable poller behavior with aggressive autoscaling policies # Architecture @@ -33,6 +34,8 @@ The Scouter system uses a Temporal-based workflow architecture with clear separa ## 🔄 Workflows +The Scouter system implements a parent-child workflow pattern for data processing orchestration. + ### 1. Scouter Workflow (`scouter.py`) The **Scouter** workflow is the main entry point for data processing pipelines. It orchestrates the complete data ingestion process and implements a robust incremental data processing pattern. @@ -130,6 +133,8 @@ The **CoreScouter** workflow implements the core data processing pipeline for in "schema": "sensor_data", "table_name": "temperature_readings", "retention_time": 3600, + "fill_missing_tags": false, + "debug_data_package": false, "model_tags": { "Temperature": { "data_range": [-50, 150], @@ -140,67 +145,31 @@ The **CoreScouter** workflow implements the core data processing pipeline for in } ``` +**Additional Parameters:** +- `fill_missing_tags` (bool): Enable filling of missing tag values with default data +- `debug_data_package` (bool): Store raw and processed data packages in MongoDB for debugging + #### Architecture ```mermaid flowchart LR A[1. data_quality_gate] --> B[2. aggregate_data] --> C[3. group_and_hold_data] --> D[4. export_data_to_postgres] --> E[5. write_metrics] + E --> F{debug_data_package?} + F -->|yes| G[6. store_data_package] B -.-> Redis1[(Redis)] C -.-> Redis2[(Redis)] D -.-> PostgreSQL[(PostgreSQL)] E -.-> Metrics[Prometheus] + G -.-> MongoDB[(MongoDB)] ``` -### 3. FakeData Workflow (`fake_data.py`) +#### Debug Mode -The **FakeData** workflow generates synthetic industrial sensor data for testing and development purposes. It's designed to simulate realistic data flows without requiring actual industrial data sources. - -#### Purpose -- **Test Data Generation**: Creates realistic sensor data for development and testing -- **Pipeline Validation**: Tests data processing workflows with known data -- **Load Testing**: Generates configurable data volumes for performance testing -- **Demonstration**: Shows data flow patterns and processing capabilities - -#### Execution Flow -1. **Data Generation**: Creates synthetic sensor readings with realistic values -2. **Kafka Publishing**: Sends generated data to specified Kafka topics -3. **Quality Assurance**: Ensures data format consistency and completeness -4. **Monitoring**: Tracks generation and publishing metrics - -#### Key Features -- **Realistic Data**: Generates data within realistic industrial ranges -- **Configurable Volume**: Adjustable message counts for different testing scenarios -- **Random Variation**: Includes realistic data variations and occasional null values -- **Kafka Integration**: Direct integration with Kafka for data streaming -- **Error Handling**: Comprehensive error handling and logging - -#### Input Parameters -```json -{ - "topic": "test_sensor_data", - "metadata": {...}, - "num_messages": 100 -} -``` - -#### Architecture - -```mermaid -flowchart TB - subgraph workflow [" "] - A[1. generate_and_send_data] - end - - subgraph services [" "] - Kafka[(Kafka)] - end - - A -.-> Kafka - - style workflow fill:none,stroke:none - style services fill:none,stroke:none -``` +When `debug_data_package` is set to `true`, the workflow stores both raw and processed data packages in MongoDB for debugging and troubleshooting purposes. This is useful for: +- Investigating data processing issues +- Validating data transformations +- Auditing data quality gate decisions ## 📋 Prerequisites @@ -398,6 +367,36 @@ The Scouter system exposes comprehensive Prometheus metrics: | `KAFKA_BOOTSTRAP_SERVERS` | Kafka broker addresses | `localhost:9092` | No | | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No | | `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No | +| `PROJECT_NAME` | Project identifier for notifications | `scouter` | No | + +### Worker Configuration + +The worker supports advanced configuration for optimizing performance and latency: + +| Variable | Description | Default | Recommended | +|----------|-------------|---------|-------------| +| `MAX_CONCURRENT_WORKFLOW_TASKS` | Maximum concurrent workflow tasks | `200` | 100-500 | +| `MAX_CONCURRENT_ACTIVITIES` | Maximum concurrent activities | `200` | 100-500 | +| `MAX_CONCURRENT_LOCAL_ACTIVITIES` | Maximum concurrent local activities | `200` | 100-500 | +| `MAX_CACHED_WORKFLOWS` | Maximum cached workflow instances | `200` | 100-500 | + +### Poller Autoscaling Configuration + +The worker implements aggressive autoscaling policies for workflow and activity pollers: + +**Workflow Poller Behavior:** +| Variable | Description | Default | +|----------|-------------|---------| +| `WORKFLOW_POLLER_BEHAVIUR_MINIMUM` | Minimum workflow pollers | `10` | +| `WORKFLOW_POLLER_BEHAVIUR_INITIAL` | Initial workflow pollers | `100` | +| `WORKFLOW_POLLER_BEHAVIUR_MAXIMUM` | Maximum workflow pollers | `200` | + +**Activity Poller Behavior:** +| Variable | Description | Default | +|----------|-------------|---------| +| `ACTIVITY_POLLER_BEHAVIUR_MINIMUM` | Minimum activity pollers | `10` | +| `ACTIVITY_POLLER_BEHAVIUR_INITIAL` | Initial activity pollers | `100` | +| `ACTIVITY_POLLER_BEHAVIUR_MAXIMUM` | Maximum activity pollers | `200` | ### Workflow Configuration @@ -461,13 +460,11 @@ MongoDB pipeline configuration: scouter/ ├── activities/ # Temporal activity implementations │ ├── activities.py # Main activities orchestrator -│ ├── redis.py # Redis operations -│ ├── gates.py # Data quality gates -│ ├── mongodb.py # MongoDB operations -│ └── faker.py # Test data generation +│ ├── redis.py # Redis operations (caching, timestamps) +│ ├── gates.py # Data quality gates and filtering +│ └── mongodb.py # MongoDB operations (data loading) ├── workflow/ # Temporal workflow definitions │ ├── scouter.py # Main data ingestion workflow -│ ├── fake_data.py # Test data workflow │ └── sub_workflows/ # Sub-workflow implementations │ └── core_scouter.py # Core data processing workflow ├── worker/ # Worker implementation @@ -479,6 +476,21 @@ scouter/ └── __init__.py ``` +### Activity Implementations + +The Activities class combines multiple service classes through multiple inheritance: + +- **Postgres** (from sientia-dataops-library): PostgreSQL data export and persistence +- **Redis**: Timestamp management, data caching, and temporary storage +- **Gates**: Data quality validation and filtering logic +- **MongoDB**: Data loading from raw collections + +All activities support: +- Comprehensive logging and error handling +- Notification integration for errors and alerts +- Prometheus metrics collection +- Graceful shutdown and resource cleanup + ### Adding New Features 1. **Follow Temporal patterns** for new workflows and activities @@ -522,17 +534,19 @@ export LOG_LEVEL=DEBUG ### Key Parameters -- **Worker Concurrency**: Adjust `max_concurrent_workflow_tasks` and `max_concurrent_activities` -- **Connection Pools**: Optimize database connection pool sizes -- **Data Retention**: Configure Redis TTL based on processing requirements -- **Batch Sizes**: Adjust data processing batch sizes for optimal throughput +- **Worker Concurrency**: Adjust `MAX_CONCURRENT_WORKFLOW_TASKS` and `MAX_CONCURRENT_ACTIVITIES` (default: 200) +- **Poller Autoscaling**: Configure minimum, initial, and maximum poller counts for optimal throughput +- **Connection Pools**: Optimize database connection pool sizes (configured in `build_*_config()` functions) +- **Data Retention**: Configure Redis TTL via `retention_time` parameter (in seconds) +- **Workflow Caching**: Set `MAX_CACHED_WORKFLOWS` to balance memory usage and performance ### Scaling Considerations -- **Horizontal Scaling**: Deploy multiple worker instances -- **Task Queue Distribution**: Use multiple task queues for different workflow types -- **Database Performance**: Optimize indexes and connection pooling -- **Kafka Partitioning**: Configure appropriate partition counts for data ingestion +- **Horizontal Scaling**: Deploy multiple worker instances (each registers to `scouter-queue`) +- **Poller Autoscaling**: Workers implement aggressive autoscaling (10-200 pollers) for latency optimization +- **Database Performance**: Connection pooling is configured in `utils/connectors_config.py` +- **Worker Placement**: Use pod anti-affinity rules in Kubernetes for optimal distribution +- **Resource Limits**: Configure appropriate CPU/memory limits based on concurrency settings ## 🤝 Contributing diff --git a/values.yaml b/values.yaml index b88a2a2..bb44c0d 100644 --- a/values.yaml +++ b/values.yaml @@ -211,27 +211,27 @@ env: # Temporal worker tuning - name: MAX_CONCURRENT_WORKFLOW_TASKS - value: 200 + value: "200" - name: MAX_CONCURRENT_ACTIVITIES - value: 200 + value: "200" - name: MAX_CONCURRENT_LOCAL_ACTIVITIES - value: 200 + value: "200" - name: MAX_CACHED_WORKFLOWS - value: 200 + value: "200" - name: WORKFLOW_POLLER_BEHAVIUR_MINIMUM - value: 10 + value: "10" - name: WORKFLOW_POLLER_BEHAVIUR_INITIAL - value: 100 + value: "100" - name: WORKFLOW_POLLER_BEHAVIUR_MAXIMUM - value: 200 + value: "200" - name: ACTIVITY_POLLER_BEHAVIUR_MINIMUM - value: 10 + value: "10" - name: ACTIVITY_POLLER_BEHAVIUR_INITIAL - value: 100 + value: "100" - name: ACTIVITY_POLLER_BEHAVIUR_MAXIMUM - value: 200 + value: "200" ssh: