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

SIENTIAPDE-1222

Browse Source 
Update README.md to reflect new features and configuration changes

- Added details about two dedicated task queues: `predictions_batch-queue` and `minimal_retrain-queue`.
- Enhanced descriptions of activities and workflows, including multiple inheritance patterns and configurable MLFlow model serving.
- Updated monitoring metrics section to include new labels and metrics for prediction and OPC export operations.
- Revised configuration section with updated default values and added new environment variables for Kubernetes pod identification.
- Improved clarity in the Predictions Batch Workflow configuration example, including structured input filters and updated retention policies.
This commit is contained in:
vitor-aignosi
2025-09-16 17:07:44 -03:00
parent a5fc526f61
commit 25bc4d06ca
1 changed files with 115 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

198
README.md
View File

@@ -66,6 +66,7 @@ The Laborious system uses a Temporal-based workflow architecture with clear sepa
- Health check endpoints for Kubernetes liveness/readiness probes
- Graceful shutdown with cleanup procedures
- Multi-instance deployment support
- Two dedicated task queues: `predictions_batch-queue` and `minimal_retrain-queue`
#### **Workflows (`laborious/workflows/`)**
- **PredictionsBatch**: Main entry point for batch prediction pipelines
@@ -79,25 +80,32 @@ The Laborious system uses a Temporal-based workflow architecture with clear sepa
- Configurable timeout and retry strategies
#### **Activities (`laborious/activities/`)**
- **Activities**: Main activity orchestrator combining all functionality through multiple inheritance
- **Gates**: Data quality validation and filtering mechanisms
- **MLFlow**: Model transformation and prediction operations
- **OPC**: Real-time data export to industrial OPC servers
- **Activities**: Main activity orchestrator and coordination
- **Key Features**:
- Multiple inheritance pattern for unified activity interface
- Configurable filter policies and validation rules
- MLFlow model serving integration
- MLFlow model serving integration with configurable flavors
- OPC UA client with certificate-based authentication
- Comprehensive error handling and notification
- Comprehensive error handling and notification integration
- Support for multiple OPC servers with independent configurations
#### **Data Services (`laborious/utils/`)**
- **Connectors**: Database and external service configuration management
- **Connectors Config**: Environment variable-based configuration management
- **Repository**: Data access layer for MLFlow and OPC operations
- `model_repository.py`: MLFlow model operations and retraining
- `opc_repository.py`: OPC server communication and data writing
- **Filters**: Data quality validation and MLFlow response filtering
- `conditional_filters.py`: Input data validation filters
- `mlflow_filters.py`: MLFlow API response validation filters
- **Key Features**:
- Environment variable-based configuration
- Environment variable-based configuration with sensible defaults
- Connection pool management and optimization
- Security credential management
- Configuration validation and error handling
- Support for multiple OPC servers and MLFlow model flavors
### Data Flow Architecture
@@ -506,23 +514,32 @@ pytest tests/workflow/test_predictions_batch.py
## 📊 Monitoring and Metrics
The Laborious system exposes comprehensive Prometheus metrics:
The Laborious system exposes comprehensive Prometheus metrics for operational visibility and performance monitoring:
### Application Metrics
### Application Health Metrics
- `app_up`: Application health status (1=healthy, 0=unhealthy)
- `laborious_predictions_written_count`: Prediction export operation count
- `laborious_prediction_confidence_monitor`: Prediction confidence monitoring
- `laborious_prediction_response_time_monitor`: Prediction response time monitoring
- Labels: `pod_id`
### MLFlow Metrics
- Model transformation and prediction success rates
- API response times and error rates
- Model retention and versioning metrics
### Prediction Operation Metrics
- `laborious_predictions_written_count`: Counter for successful prediction exports
- Labels: `pod_id`, `model_name`, `pipeline_name`
- `laborious_prediction_confidence_monitor`: Gauge for current prediction confidence levels
- Labels: `pod_id`, `model_name`, `pipeline_name`
- `laborious_prediction_response_time_monitor`: Histogram for prediction response times
- Labels: `pod_id`, `model_name`, `pipeline_name`
- Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0]
### Export Metrics
- PostgreSQL export operation counts and response times
- OPC server write operations and performance
- Data quality filter pass/fail rates
### OPC Export Metrics
- `laborious_prediction_opc_writing_count`: Counter for OPC server write operations
- Labels: `pod_id`, `model_name`, `pipeline_name`, `opc_server_id`
- `laborious_prediction_opc_writing_response_time_monitor`: Histogram for OPC write response times
- Labels: `pod_id`, `model_name`, `pipeline_name`, `opc_server_id`
- Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0]
### Data Quality Metrics
- Filter pass/fail rates through notification system
- MLFlow API response validation metrics
- Data quality gate performance tracking
## ⚙️ Configuration
@@ -537,31 +554,30 @@ The Laborious system exposes comprehensive Prometheus metrics:
| `POSTGRES_USER` | PostgreSQL username | `sientia` | Yes |
| `POSTGRES_PASSWORD` | PostgreSQL password | `sientia` | Yes |
| `POSTGRES_DBNAME` | PostgreSQL database | `sientia` | Yes |
| `POSTGRES_MIN_CONNECTIONS` | Minimum PostgreSQL connections | `10` | No |
| `POSTGRES_MAX_CONNECTIONS` | Maximum PostgreSQL connections | `30` | No |
| `MLFLOW_HOST` | MLFlow server hostname | `localhost` | Yes |
| `MLFLOW_PORT` | MLFlow server port | `5000` | Yes |
| `MLFLOW_USERNAME` | MLFlow username | `admin` | Yes |
| `MLFLOW_PASSWORD` | MLFlow password | `admin` | Yes |
| `POSTGRES_MIN_CONNECTIONS` | Minimum PostgreSQL connections | `5` | No |
| `POSTGRES_MAX_CONNECTIONS` | Maximum PostgreSQL connections | `20` | No |
| `MLFLOW_HOST` | MLFlow server hostname | `http://localhost` | Yes |
| `MLFLOW_PORT` | MLFlow server port | `5080` | Yes |
| `MLFLOW_USERNAME` | MLFlow username | `aignosi` | Yes |
| `MLFLOW_PASSWORD` | MLFlow password | `aignosi` | Yes |
| `OPC_CONFIG` | OPC server configuration (JSON) | `{}` | No |
| `OPC_ID` | OPC server identifier | `1` | No |
| `OPC_URL` | OPC server URL | `opc.tcp://localhost:4840` | No |
| `OPC_NAME` | OPC server name | `OPC_Server` | No |
| `OPC_SERVER_URI` | OPC server URI | `urn:opcserver:opcua` | No |
| `OPC_CERT_PATH` | OPC client certificate path | `/path/to/cert.pem` | No |
| `OPC_PRIVATE_KEY_PATH` | OPC private key path | `/path/to/key.pem` | No |
| `OPC_SERVER_CERT_PATH` | OPC server certificate path | `/path/to/server_cert.pem` | No |
| `OPC_RECONNECTION_INTERVAL` | OPC reconnection interval (ms) | `5000` | No |
| `MONGODB_URL` | MongoDB connection URI | `localhost:27017` | Yes |
| `OPC_SERVER_URI` | OPC server URI | `opc.tcp://localhost:4840` | No |
| `OPC_CERT_PATH` | OPC client certificate path | `None` | No |
| `OPC_PRIVATE_KEY_PATH` | OPC private key path | `None` | No |
| `OPC_SERVER_CERT_PATH` | OPC server certificate path | `None` | No |
| `OPC_RECONNECTION_INTERVAL` | OPC reconnection interval (ms) | `120` | No |
| `MONGODB_URL` | MongoDB connection URI | `localhost:27018` | Yes |
| `MONGODB_USERNAME` | MongoDB username | `root` | Yes |
| `MONGODB_PASSWORD` | MongoDB password | `password` | Yes |
| `MONGODB_DATABASE` | MongoDB database name | `sientia` | Yes |
| `MONGODB_PASSWORD` | MongoDB password | `wKZDbMNU1c` | Yes |
| `MONGODB_DATABASE_NAME` | MongoDB database name | `sientia` | Yes |
| `MONGODB_TTL_INDEX_HOURS` | MongoDB TTL index hours | `1` | No |
| `KAFKA_BOOTSTRAP_SERVERS` | Kafka bootstrap servers | `localhost:9092` | No |
| `LOG_LEVEL` | Application log level | `INFO` | No |
| `PROJECT_NAME` | Project name for metrics | `sientia-laborious` | No |
| `PROJECT_NAME` | Project name for metrics | `laborious` | No |
| `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No |
| `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No |
| `POD_ID` | Kubernetes pod identifier | `None` | No |
@@ -606,74 +622,88 @@ For single OPC server, use individual environment variables:
MongoDB pipeline configuration:
#### Predictions Batch Workflow
#### Predictions Batch Workflow configuration sample
This is the configuration for the Predictions Batch Workflow, to be inserted into the MongoDB pipeline collection.
```json
{
"schedule_name": "laborious-orchestrated-pipeline",
"model_id": "1",
"workflow_type": "predictions_batch",
"frequency": "30s", # Workflow execution frequency
"max_retry_policy": 1, # Maximum number of retries for the workflow
"frequency": "30s",
"max_retry_policy": 1,
"query": "select * from sientia_data.laborious_data where model_id = 1 and \"timestamp\" > NOW() - INTERVAL '5 minutes' order by \"timestamp\" desc limit 30;",
"retention_time": 60, # Retention time for models in minutes
"write_tags": [
{
"server_id": "1",
"type": "prediction", # Type of tag to write, can be prediction or confidence
"server_id": "server1",
"type": "prediction",
"addr": "ns=2;i=5",
"data_type": "double"
},
{
"server_id": "1",
"server_id": "server1",
"type": "confidence",
"addr": "ns=2;i=5",
"addr": "ns=2;i=6",
"data_type": "double"
}
],
"input_filters": [
{
"filter_name": "EMPTY_DATA", # Required filter
"policy": "STOP"
"input_filters": {
"EMPTY_DATA": {"POLICY": "STOP"},
"SPECIFIC_VARIABLES_NULL_VALUES": {
"POLICY": "CONTINUE",
"config": {"variables": ["Counter"]}
}
},
{
"filter_name": "SPECIFIC_VARIABLES_NULL_VALUES",
"policy": "CONTINUE",
"config": {
"variables": [
"Counter"
]
}
}
],
"mlflow_transform_filters": [
{
"filter_name": "API_ERROR", # Required filter
"policy": "REPEAT"
"mlflow_transform_filters": {
"API_ERROR": {"POLICY": "REPEAT"},
"NAN_VALUES": {"POLICY": "STOP"}
},
{
"filter_name": "NAN_VALUES",
"policy": "STOP"
}
],
"mlflow_predict_filters": [
{
"filter_name": "API_ERROR", # Required filter
"policy": "CONTINUE"
}
],
"path_priority": [ # In case of multiple filters catch problems, this will determine the path to take
"STOP",
"CONTINUE",
"REPEAT"
],
"mlflow_predict_filters": {
"API_ERROR": {"POLICY": "CONTINUE"}
},
"path_priority": ["STOP", "CONTINUE", "REPEAT"],
"active": true,
"datetime_columns": [ # Columns in data comming from query that are datetime
"timestamp",
"created_at"
],
"updated_at": {
"$date": "2025-08-27T18:35:01.600Z"
}
"$date": "2025-09-16T10:00:00.000Z"
},
"datetime_columns": ["timestamp", "created_at"],
"predictions_storage_policy": "lts:1"
}
```
This is the configuration created by the Orchestrator in Temporal.
```json
{
"datetime_columns":["timestamp","created_at"],
"frequency":"15m",
"input_filters":{"EMPTY_DATA":{"config":{},"policy":"STOP"}},
"max_retry_policy":1,
"mlflow_predict_filters":{"API_ERROR":{"config":{},"policy":"CONTINUE"}},
"mlflow_transform_filters":{
"API_ERROR":{"config":{},"policy":"CONTINUE"},
"EMPTY_DATA":{"config":{},"policy":"STOP"}
},
"model_config":{
"is_compressed":true,
"predict_flavor":"pyfunc",
"retention_minutes":60,
"retention_target":"artifact",
"transform_function_keyword":"transform"
},
"model_id":"352",
"model_name":"courier",
"opc_output_config":{},
"path_priority":["STOP","CONTINUE","REPEAT"],
"predictions_storage_policy":"lts:1",
"query":"select * from sientia_data.laborious_data where model_id = 352 order by \"timestamp\" desc limit 300;",
"retention_time":3600,
"schedule_name":"laborious-courier",
"schema":"sientia_data",
"table_name":"predictions",
"updated_at":"2025-09-12 19:35:01.600000+0000",
"workflow_type":"predictions_batch"
}
```
@@ -687,7 +717,7 @@ laborious/
│ ├── gates.py # Data quality gates and filtering
│ ├── mlflow.py # MLFlow model operations
│ └── opc.py # OPC server operations
├── workflow/ # Temporal workflow definitions
├── workflows/ # Temporal workflow definitions
│ ├── predictions_batch.py # Main batch prediction workflow
│ ├── minimal_retrain.py # Model retraining workflow
│ └── sub_workflows/ # Sub-workflow implementations