Merge pull request #26 from Aignosi/release/SIENTIAPDE-1441
Enhance Temporal Worker Configuration, Refine Helm Chart, and Streamline Dependencies
This commit is contained in:
vitor-aignosi
140
README.md
140
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
|
||||
|
||||
|
||||
113
encrypt.py
113
encrypt.py
@@ -1,113 +0,0 @@
|
||||
import os
|
||||
import argparse
|
||||
from pathspec import PathSpec
|
||||
import yaml # type: ignore
|
||||
from typing import Any
|
||||
|
||||
'''
|
||||
Usage:
|
||||
python .\encrypt.py path_to_dir output_file --ignore ignore_file --chunk-size 100000
|
||||
'''
|
||||
|
||||
|
||||
def load_ignore_patterns(ignore_file, include_library):
|
||||
# Ensure the .gitignore file exists
|
||||
if not os.path.exists(ignore_file):
|
||||
raise FileNotFoundError(f"Ignore file not found at {ignore_file}")
|
||||
|
||||
# Load and parse the .gitignore patterns
|
||||
with open(ignore_file, 'r') as file:
|
||||
patterns = file.readlines()
|
||||
if not include_library:
|
||||
patterns.append('**/deploy/library/')
|
||||
|
||||
spec = PathSpec.from_lines('gitwildmatch', patterns)
|
||||
return spec
|
||||
|
||||
|
||||
def is_ignored(file_path, spec):
|
||||
"""Check if a file should be ignored based on the ignore patterns."""
|
||||
return spec.match_file(file_path) if spec else False
|
||||
|
||||
|
||||
def encode_file_tree_to_yaml(directory, ignore_file, include_library):
|
||||
"""Encode the file tree into a single YAML file."""
|
||||
ignore_patterns = load_ignore_patterns(
|
||||
ignore_file, include_library) if ignore_file else None
|
||||
file_tree: dict[str, Any] = {}
|
||||
|
||||
for root, dirs, files in os.walk(directory):
|
||||
# Skip ignored directories
|
||||
dirs[:] = [d for d in dirs if not is_ignored(
|
||||
os.path.join(root, d), ignore_patterns)]
|
||||
|
||||
for file in files:
|
||||
file_path = os.path.join(root, file)
|
||||
|
||||
# Skip ignored files
|
||||
if is_ignored(file_path, ignore_patterns):
|
||||
continue
|
||||
|
||||
# Read file content
|
||||
try:
|
||||
with open(file_path, 'r', encoding='utf-8') as f:
|
||||
content = f.read()
|
||||
except Exception as e:
|
||||
print(f"Error reading file {file_path}: {e}")
|
||||
raise
|
||||
|
||||
# Create nested dictionary structure
|
||||
path_parts = os.path.relpath(file_path, directory).split(os.sep)
|
||||
current_level = file_tree
|
||||
|
||||
# all except the last part (the file name)
|
||||
for part in path_parts[:-1]:
|
||||
current_level = current_level.setdefault(part, {})
|
||||
|
||||
# Add the file and its content
|
||||
current_level[path_parts[-1]] = content
|
||||
return yaml.dump(file_tree, default_flow_style=False)
|
||||
|
||||
|
||||
def chunk_and_write_file_tree_to_yaml(yaml_content, output_file, chunk_size=None):
|
||||
"""Chunk the YAML content and write it to the output file."""
|
||||
|
||||
chunks = [yaml_content] if chunk_size is None else [
|
||||
yaml_content[i:i + chunk_size] for i in range(0, len(yaml_content), chunk_size)]
|
||||
|
||||
for i, chunk in enumerate(chunks):
|
||||
chunk_file = f"{output_file}_{i}.yaml"
|
||||
# Write the file tree to the output YAML file
|
||||
with open(chunk_file, 'w', encoding='utf-8') as yaml_file:
|
||||
yaml_file.write(chunk)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Encrypts file tree to yaml file")
|
||||
parser.add_argument("input_directory", help="Directory to encode")
|
||||
parser.add_argument("output_yaml_file", help="Output YAML file")
|
||||
parser.add_argument("--ignore", default=None,
|
||||
help="Path to the ignore file")
|
||||
parser.add_argument("--chunk-size", type=int, default=None,
|
||||
help="Chunk size for the output YAML file")
|
||||
parser.add_argument("--library", type=bool, default=False,
|
||||
help="Incude the library in the output YAML file")
|
||||
|
||||
# Parse arguments
|
||||
args = parser.parse_args()
|
||||
|
||||
# Example usage
|
||||
directory_to_encode = args.input_directory
|
||||
ignore_file_path = args.ignore
|
||||
output_yaml_file = args.output_yaml_file
|
||||
include_library = args.library
|
||||
|
||||
content = encode_file_tree_to_yaml(
|
||||
directory_to_encode, ignore_file_path, include_library)
|
||||
chunk_and_write_file_tree_to_yaml(
|
||||
content, output_yaml_file, args.chunk_size)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
1
git-requirements-mapping.txt
Normal file
1
git-requirements-mapping.txt
Normal file
@@ -0,0 +1 @@
|
||||
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git:sientia-do
|
||||
@@ -1,10 +1,7 @@
|
||||
temporalio
|
||||
psycopg2-binary
|
||||
sqlalchemy
|
||||
asyncua
|
||||
redis
|
||||
aiokafka
|
||||
pymongo
|
||||
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3
|
||||
pydruid[pandas]
|
||||
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.6.1
|
||||
prometheus-client
|
||||
58
scouter/worker/prepare_worker.py
Normal file
58
scouter/worker/prepare_worker.py
Normal file
@@ -0,0 +1,58 @@
|
||||
import os
|
||||
from collections.abc import Sequence
|
||||
from typing import Any
|
||||
|
||||
from temporalio.client import Client
|
||||
from temporalio.worker import PollerBehaviorAutoscaling, Worker
|
||||
|
||||
parameters = [
|
||||
('MAX_CONCURRENT_WORKFLOW_TASKS', '200'),
|
||||
('MAX_CONCURRENT_ACTIVITIES', '200'),
|
||||
('MAX_CONCURRENT_LOCAL_ACTIVITIES', '200'),
|
||||
('MAX_CACHED_WORKFLOWS', '200'),
|
||||
('WORKFLOW_POLLER_BEHAVIUR_MINIMUM', '10'),
|
||||
('WORKFLOW_POLLER_BEHAVIUR_INITIAL', '100'),
|
||||
('WORKFLOW_POLLER_BEHAVIUR_MAXIMUM', '200'),
|
||||
('ACTIVITY_POLLER_BEHAVIUR_MINIMUM', '10'),
|
||||
('ACTIVITY_POLLER_BEHAVIUR_INITIAL', '100'),
|
||||
('ACTIVITY_POLLER_BEHAVIUR_MAXIMUM', '200'),
|
||||
]
|
||||
|
||||
|
||||
def prepare_worker(
|
||||
main_workflow: type,
|
||||
other_workflows: Sequence[type],
|
||||
activities: Sequence[Any],
|
||||
temporal_client: Client,
|
||||
) -> Worker:
|
||||
main_workflow_name = main_workflow.__name__.upper()
|
||||
|
||||
local_workflow_parameters = {}
|
||||
|
||||
for parameter in parameters:
|
||||
local_workflow_parameters[parameter[0]] = int(
|
||||
os.getenv(main_workflow_name + '_' + parameter[0], parameter[1])
|
||||
)
|
||||
|
||||
return Worker(
|
||||
temporal_client,
|
||||
task_queue='scouter-queue',
|
||||
workflows=[main_workflow, *other_workflows],
|
||||
activities=[*activities],
|
||||
max_concurrent_workflow_tasks=local_workflow_parameters['MAX_CONCURRENT_WORKFLOW_TASKS'],
|
||||
max_concurrent_activities=local_workflow_parameters['MAX_CONCURRENT_ACTIVITIES'],
|
||||
max_concurrent_local_activities=local_workflow_parameters[
|
||||
'MAX_CONCURRENT_LOCAL_ACTIVITIES'
|
||||
],
|
||||
max_cached_workflows=local_workflow_parameters['MAX_CACHED_WORKFLOWS'],
|
||||
workflow_task_poller_behavior=PollerBehaviorAutoscaling(
|
||||
minimum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MINIMUM'],
|
||||
initial=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_INITIAL'],
|
||||
maximum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MAXIMUM'],
|
||||
),
|
||||
activity_task_poller_behavior=PollerBehaviorAutoscaling(
|
||||
minimum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MINIMUM'],
|
||||
initial=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_INITIAL'],
|
||||
maximum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MAXIMUM'],
|
||||
),
|
||||
)
|
||||
@@ -1,6 +1,7 @@
|
||||
from temporalio import client, workflow
|
||||
from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
|
||||
from temporalio.worker import PollerBehaviorAutoscaling, Worker
|
||||
|
||||
from scouter.worker.prepare_worker import prepare_worker
|
||||
|
||||
with workflow.unsafe.imports_passed_through():
|
||||
import asyncio
|
||||
@@ -26,6 +27,26 @@ POD_ID = os.getenv('HOSTNAME', 'localhost')
|
||||
SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091'))
|
||||
|
||||
|
||||
# For optmized latency, Temporal docs recommends fixed slots, ensuring
|
||||
# high concurency levels.
|
||||
|
||||
MAX_CONCURRENT_WORKFLOW_TASKS = int(os.getenv('MAX_CONCURRENT_WORKFLOW_TASKS', '200'))
|
||||
MAX_CONCURRENT_ACTIVITIES = int(os.getenv('MAX_CONCURRENT_ACTIVITIES', '200'))
|
||||
MAX_CONCURRENT_LOCAL_ACTIVITIES = int(os.getenv('MAX_CONCURRENT_LOCAL_ACTIVITIES', '200'))
|
||||
MAX_CACHED_WORKFLOWS = int(os.getenv('MAX_CACHED_WORKFLOWS', '200'))
|
||||
|
||||
|
||||
# Temporal docs also recommends an autoscaling policy, with agrresive limits to prioritize latency over throughput.
|
||||
|
||||
WORKFLOW_POLLER_BEHAVIUR_MINIMUM = int(os.getenv('WORKFLOW_POLLER_BEHAVIUR_MINIMUM', '10'))
|
||||
WORKFLOW_POLLER_BEHAVIUR_INITIAL = int(os.getenv('WORKFLOW_POLLER_BEHAVIUR_INITIAL', '100'))
|
||||
WORKFLOW_POLLER_BEHAVIUR_MAXIMUM = int(os.getenv('WORKFLOW_POLLER_BEHAVIUR_MAXIMUM', '200'))
|
||||
|
||||
ACTIVITY_POLLER_BEHAVIUR_MINIMUM = int(os.getenv('ACTIVITY_POLLER_BEHAVIUR_MINIMUM', '10'))
|
||||
ACTIVITY_POLLER_BEHAVIUR_INITIAL = int(os.getenv('ACTIVITY_POLLER_BEHAVIUR_INITIAL', '100'))
|
||||
ACTIVITY_POLLER_BEHAVIUR_MAXIMUM = int(os.getenv('ACTIVITY_POLLER_BEHAVIUR_MAXIMUM', '200'))
|
||||
|
||||
|
||||
async def main():
|
||||
"""
|
||||
Main entry point for the Scouter Temporal worker.
|
||||
@@ -101,10 +122,10 @@ async def main():
|
||||
logger.custom_info('Starting Workers...', metadata)
|
||||
|
||||
workers = [
|
||||
Worker(
|
||||
temporal_client,
|
||||
task_queue='scouter-queue',
|
||||
workflows=[Scouter, CoreScouter],
|
||||
prepare_worker(
|
||||
temporal_client=temporal_client,
|
||||
main_workflow=Scouter,
|
||||
other_workflows=[CoreScouter],
|
||||
activities=[
|
||||
activities.load_latest_data,
|
||||
activities.get_last_data_timestamp,
|
||||
@@ -116,12 +137,6 @@ async def main():
|
||||
activities.write_metrics,
|
||||
activities.store_data_package,
|
||||
],
|
||||
max_concurrent_workflow_tasks=50,
|
||||
max_concurrent_activities=50,
|
||||
max_concurrent_local_activities=50,
|
||||
max_cached_workflows=200,
|
||||
workflow_task_poller_behavior=PollerBehaviorAutoscaling(),
|
||||
activity_task_poller_behavior=PollerBehaviorAutoscaling(),
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ sonar.tests=tests
|
||||
sonar.qualitygate.wait=true
|
||||
sonar.qualitygate.timeout=300
|
||||
sonar.python.coverage.reportPaths=coverage.xml
|
||||
sonar.coverage.exclusions=scouter/worker/worker.py
|
||||
sonar.coverage.exclusions=scouter/worker/worker.py,scouter/worker/prepare_worker.py
|
||||
sonar.python.xunit.reportPath=pytest.xml
|
||||
sonar.python.version=3.11
|
||||
sonar.projectVersion=1.0.0
|
||||
|
||||
724
test.ipynb
724
test.ipynb
@@ -1,724 +0,0 @@
|
||||
{
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"from sqlalchemy.engine import create_engine\n",
|
||||
"\n",
|
||||
"engine = create_engine('druid://localhost:8082/druid/v2/sql/')"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"from sqlalchemy import MetaData, Table\n",
|
||||
"\n",
|
||||
"metadata = MetaData()\n",
|
||||
"places = Table('raw_scouter-opcua-orchestrated-pipeline', metadata, autoload_with=engine)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"from sqlalchemy import select\n",
|
||||
"\n",
|
||||
"stmt = select(places)\n",
|
||||
"with engine.connect() as conn:\n",
|
||||
" result = conn.execute(stmt)\n",
|
||||
" for row in result:\n",
|
||||
" print(row)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 5,
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stderr",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"/tmp/ipykernel_30057/1086103244.py:5: SADeprecationWarning: The dbapi() classmethod on dialect classes has been renamed to import_dbapi(). Implement an import_dbapi() classmethod directly on class <class 'pydruid.db.sqlalchemy.DruidDialect'> to remove this warning; the old .dbapi() classmethod may be maintained for backwards compatibility.\n",
|
||||
" engine = create_engine('druid://localhost:8082/druid/v2/sql/')\n",
|
||||
"/home/grezewave/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/pydruid/db/sqlalchemy.py:188: SAWarning: Dialect druid:rest will not make use of SQL compilation caching as it does not set the 'supports_statement_cache' attribute to ``True``. This can have significant performance implications including some performance degradations in comparison to prior SQLAlchemy versions. Dialect maintainers should seek to set this attribute to True after appropriate development and testing for SQLAlchemy 1.4 caching support. Alternatively, this attribute may be set to False which will disable this warning. (Background on this warning at: https://sqlalche.me/e/20/cprf)\n",
|
||||
" result = connection.execute(text(query))\n"
|
||||
]
|
||||
},
|
||||
{
|
||||
"data": {
|
||||
"text/html": [
|
||||
"<div>\n",
|
||||
"<style scoped>\n",
|
||||
" .dataframe tbody tr th:only-of-type {\n",
|
||||
" vertical-align: middle;\n",
|
||||
" }\n",
|
||||
"\n",
|
||||
" .dataframe tbody tr th {\n",
|
||||
" vertical-align: top;\n",
|
||||
" }\n",
|
||||
"\n",
|
||||
" .dataframe thead th {\n",
|
||||
" text-align: right;\n",
|
||||
" }\n",
|
||||
"</style>\n",
|
||||
"<table border=\"1\" class=\"dataframe\">\n",
|
||||
" <thead>\n",
|
||||
" <tr style=\"text-align: right;\">\n",
|
||||
" <th></th>\n",
|
||||
" <th>name</th>\n",
|
||||
" <th>kafka.topic</th>\n",
|
||||
" <th>tag</th>\n",
|
||||
" <th>value</th>\n",
|
||||
" <th>timestamp</th>\n",
|
||||
" <th>inserted_at</th>\n",
|
||||
" </tr>\n",
|
||||
" </thead>\n",
|
||||
" <tbody>\n",
|
||||
" <tr>\n",
|
||||
" <th>0</th>\n",
|
||||
" <td>Counter</td>\n",
|
||||
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
|
||||
" <td>ns=2;i=2</td>\n",
|
||||
" <td>-50.132</td>\n",
|
||||
" <td>2025-07-02 13:05:19</td>\n",
|
||||
" <td>2025-07-02 13:05:19.729000</td>\n",
|
||||
" </tr>\n",
|
||||
" <tr>\n",
|
||||
" <th>1</th>\n",
|
||||
" <td>Rollout</td>\n",
|
||||
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
|
||||
" <td>ns=2;i=3</td>\n",
|
||||
" <td>70.767</td>\n",
|
||||
" <td>2025-07-02 13:05:19</td>\n",
|
||||
" <td>2025-07-02 13:05:19.731000</td>\n",
|
||||
" </tr>\n",
|
||||
" <tr>\n",
|
||||
" <th>2</th>\n",
|
||||
" <td>Square</td>\n",
|
||||
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
|
||||
" <td>ns=2;i=4</td>\n",
|
||||
" <td>-58.448</td>\n",
|
||||
" <td>2025-07-02 13:05:19</td>\n",
|
||||
" <td>2025-07-02 13:05:19.732000</td>\n",
|
||||
" </tr>\n",
|
||||
" <tr>\n",
|
||||
" <th>3</th>\n",
|
||||
" <td>Counter</td>\n",
|
||||
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
|
||||
" <td>ns=2;i=2</td>\n",
|
||||
" <td>-50.126</td>\n",
|
||||
" <td>2025-07-02 13:05:24</td>\n",
|
||||
" <td>2025-07-02 13:05:24.728000</td>\n",
|
||||
" </tr>\n",
|
||||
" <tr>\n",
|
||||
" <th>4</th>\n",
|
||||
" <td>Rollout</td>\n",
|
||||
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
|
||||
" <td>ns=2;i=3</td>\n",
|
||||
" <td>69.199</td>\n",
|
||||
" <td>2025-07-02 13:05:24</td>\n",
|
||||
" <td>2025-07-02 13:05:24.730000</td>\n",
|
||||
" </tr>\n",
|
||||
" <tr>\n",
|
||||
" <th>...</th>\n",
|
||||
" <td>...</td>\n",
|
||||
" <td>...</td>\n",
|
||||
" <td>...</td>\n",
|
||||
" <td>...</td>\n",
|
||||
" <td>...</td>\n",
|
||||
" <td>...</td>\n",
|
||||
" </tr>\n",
|
||||
" <tr>\n",
|
||||
" <th>373</th>\n",
|
||||
" <td>Rollout</td>\n",
|
||||
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
|
||||
" <td>ns=2;i=3</td>\n",
|
||||
" <td>85.921</td>\n",
|
||||
" <td>2025-07-02 13:15:40</td>\n",
|
||||
" <td>2025-07-02 13:15:40.230000</td>\n",
|
||||
" </tr>\n",
|
||||
" <tr>\n",
|
||||
" <th>374</th>\n",
|
||||
" <td>Square</td>\n",
|
||||
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
|
||||
" <td>ns=2;i=4</td>\n",
|
||||
" <td>-69.296</td>\n",
|
||||
" <td>2025-07-02 13:15:40</td>\n",
|
||||
" <td>2025-07-02 13:15:40.232000</td>\n",
|
||||
" </tr>\n",
|
||||
" <tr>\n",
|
||||
" <th>375</th>\n",
|
||||
" <td>Counter</td>\n",
|
||||
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
|
||||
" <td>ns=2;i=2</td>\n",
|
||||
" <td>-71.207</td>\n",
|
||||
" <td>2025-07-02 13:15:45</td>\n",
|
||||
" <td>2025-07-02 13:15:45.228000</td>\n",
|
||||
" </tr>\n",
|
||||
" <tr>\n",
|
||||
" <th>376</th>\n",
|
||||
" <td>Rollout</td>\n",
|
||||
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
|
||||
" <td>ns=2;i=3</td>\n",
|
||||
" <td>84.665</td>\n",
|
||||
" <td>2025-07-02 13:15:45</td>\n",
|
||||
" <td>2025-07-02 13:15:45.231000</td>\n",
|
||||
" </tr>\n",
|
||||
" <tr>\n",
|
||||
" <th>377</th>\n",
|
||||
" <td>Square</td>\n",
|
||||
" <td>raw_scouter-opcua-orchestrated-pipeline</td>\n",
|
||||
" <td>ns=2;i=4</td>\n",
|
||||
" <td>-67.592</td>\n",
|
||||
" <td>2025-07-02 13:15:45</td>\n",
|
||||
" <td>2025-07-02 13:15:45.233000</td>\n",
|
||||
" </tr>\n",
|
||||
" </tbody>\n",
|
||||
"</table>\n",
|
||||
"<p>378 rows × 6 columns</p>\n",
|
||||
"</div>"
|
||||
],
|
||||
"text/plain": [
|
||||
" name kafka.topic tag value \\\n",
|
||||
"0 Counter raw_scouter-opcua-orchestrated-pipeline ns=2;i=2 -50.132 \n",
|
||||
"1 Rollout raw_scouter-opcua-orchestrated-pipeline ns=2;i=3 70.767 \n",
|
||||
"2 Square raw_scouter-opcua-orchestrated-pipeline ns=2;i=4 -58.448 \n",
|
||||
"3 Counter raw_scouter-opcua-orchestrated-pipeline ns=2;i=2 -50.126 \n",
|
||||
"4 Rollout raw_scouter-opcua-orchestrated-pipeline ns=2;i=3 69.199 \n",
|
||||
".. ... ... ... ... \n",
|
||||
"373 Rollout raw_scouter-opcua-orchestrated-pipeline ns=2;i=3 85.921 \n",
|
||||
"374 Square raw_scouter-opcua-orchestrated-pipeline ns=2;i=4 -69.296 \n",
|
||||
"375 Counter raw_scouter-opcua-orchestrated-pipeline ns=2;i=2 -71.207 \n",
|
||||
"376 Rollout raw_scouter-opcua-orchestrated-pipeline ns=2;i=3 84.665 \n",
|
||||
"377 Square raw_scouter-opcua-orchestrated-pipeline ns=2;i=4 -67.592 \n",
|
||||
"\n",
|
||||
" timestamp inserted_at \n",
|
||||
"0 2025-07-02 13:05:19 2025-07-02 13:05:19.729000 \n",
|
||||
"1 2025-07-02 13:05:19 2025-07-02 13:05:19.731000 \n",
|
||||
"2 2025-07-02 13:05:19 2025-07-02 13:05:19.732000 \n",
|
||||
"3 2025-07-02 13:05:24 2025-07-02 13:05:24.728000 \n",
|
||||
"4 2025-07-02 13:05:24 2025-07-02 13:05:24.730000 \n",
|
||||
".. ... ... \n",
|
||||
"373 2025-07-02 13:15:40 2025-07-02 13:15:40.230000 \n",
|
||||
"374 2025-07-02 13:15:40 2025-07-02 13:15:40.232000 \n",
|
||||
"375 2025-07-02 13:15:45 2025-07-02 13:15:45.228000 \n",
|
||||
"376 2025-07-02 13:15:45 2025-07-02 13:15:45.231000 \n",
|
||||
"377 2025-07-02 13:15:45 2025-07-02 13:15:45.233000 \n",
|
||||
"\n",
|
||||
"[378 rows x 6 columns]"
|
||||
]
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"from sqlalchemy import create_engine, MetaData, Table, select, func, text\n",
|
||||
"import pandas as pd\n",
|
||||
"from datetime import datetime\n",
|
||||
"\n",
|
||||
"engine = create_engine('druid://localhost:8082/druid/v2/sql/')\n",
|
||||
"metadata = MetaData()\n",
|
||||
"places = Table('raw_scouter-opcua-orchestrated-pipeline', metadata, autoload_with=engine)\n",
|
||||
"date_str = '2025-01-01'\n",
|
||||
"stmt = select(places).where(text(f'\"__time\" > TIMESTAMP \\'{date_str}\\''))\n",
|
||||
"\n",
|
||||
"result = pd.read_sql(stmt, engine)\n",
|
||||
"\n",
|
||||
"result[\"inserted_at\"] = pd.to_datetime(result[\"__time\"]).dt.strftime(\n",
|
||||
" \"%Y-%m-%d %H:%M:%S.%f\")\n",
|
||||
"\n",
|
||||
"result.drop(columns=[\"__time\"], inplace=True)\n",
|
||||
"\n",
|
||||
"display(result)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 1,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"data = {\n",
|
||||
" \"id\": \"1\",\n",
|
||||
" \"schedule_name\": \"scouter-opcua-orchestrated-pipeline\",\n",
|
||||
" \"model_id\": \"1\",\n",
|
||||
" \"workflow_type\": \"scouter\",\n",
|
||||
" \"frequency\": \"30s\",\n",
|
||||
" \"max_retry_policy\": 1,\n",
|
||||
" \"read_tags\": [\n",
|
||||
" {\n",
|
||||
" \"tag_name\": \"Counter\",\n",
|
||||
" \"server_id\": \"1\",\n",
|
||||
" \"aggr_func\": \"avg\",\n",
|
||||
" \"tag_address\": \"ns=2;i=2\",\n",
|
||||
" \"frequency\": \"15000\",\n",
|
||||
" \"data_range\": [\n",
|
||||
" -100,\n",
|
||||
" 100\n",
|
||||
" ]\n",
|
||||
" },\n",
|
||||
" {\n",
|
||||
" \"tag_name\": \"Rollout\",\n",
|
||||
" \"server_id\": \"1\",\n",
|
||||
" \"aggr_func\": \"mdn\",\n",
|
||||
" \"tag_address\": \"ns=2;i=3\",\n",
|
||||
" \"frequency\": \"15000\",\n",
|
||||
" \"data_range\": [\n",
|
||||
" -100,\n",
|
||||
" 100\n",
|
||||
" ]\n",
|
||||
" },\n",
|
||||
" {\n",
|
||||
" \"tag_name\": \"Square\",\n",
|
||||
" \"server_id\": \"1\",\n",
|
||||
" \"aggr_func\": \"lts\",\n",
|
||||
" \"tag_address\": \"ns=2;i=4\",\n",
|
||||
" \"frequency\": \"15000\",\n",
|
||||
" \"data_range\": [\n",
|
||||
" -100,\n",
|
||||
" 100\n",
|
||||
" ]\n",
|
||||
" }\n",
|
||||
" ],\n",
|
||||
" \"filters\": [\n",
|
||||
" {\n",
|
||||
" \"filter_name\": \"OUT_OF_BOUNDS_FILTER\",\n",
|
||||
" \"policy\": \"DISCARD\"\n",
|
||||
" },\n",
|
||||
" {\n",
|
||||
" \"filter_name\": \"NULL_VALUES_FILTER\",\n",
|
||||
" \"policy\": \"DISCARD\"\n",
|
||||
" }\n",
|
||||
" ],\n",
|
||||
" \"tag_retention_minutes\": 60\n",
|
||||
"}"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 2,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"import json\n",
|
||||
"\n",
|
||||
"\n",
|
||||
"# Generate 29 more, changing only id and schedule_name\n",
|
||||
"json_list = []\n",
|
||||
"for i in range(30):\n",
|
||||
" obj = data.copy()\n",
|
||||
" obj['id'] = i + 1 # or any other unique id logic\n",
|
||||
" obj['schedule_name'] = f\"scouter-opcua-pipeline-{i+1}\"\n",
|
||||
" json_list.append(obj)\n",
|
||||
"\n",
|
||||
"# Save to a new file\n",
|
||||
"with open('input_samples_30.json', 'w') as f:\n",
|
||||
" json.dump(json_list, f, indent=2)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 9,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"spec = {\n",
|
||||
" \"type\": \"kafka\",\n",
|
||||
" \"spec\": {\n",
|
||||
" \"dataSchema\": {\n",
|
||||
" \"dataSource\": \"raw_scouter-opcua-orchestrated-pipeline\",\n",
|
||||
" \"timestampSpec\": {\n",
|
||||
" \"column\": \"kafka.timestamp\",\n",
|
||||
" \"format\": \"millis\",\n",
|
||||
" \"missingValue\": None\n",
|
||||
" },\n",
|
||||
" \"dimensionsSpec\": {\n",
|
||||
" \"dimensions\": [],\n",
|
||||
" \"dimensionExclusions\": [\n",
|
||||
" \"__time\",\n",
|
||||
" \"kafka.timestamp\"\n",
|
||||
" ],\n",
|
||||
" \"includeAllDimensions\": False,\n",
|
||||
" \"useSchemaDiscovery\": True\n",
|
||||
" },\n",
|
||||
" \"metricsSpec\": [],\n",
|
||||
" \"granularitySpec\": {\n",
|
||||
" \"type\": \"uniform\",\n",
|
||||
" \"segmentGranularity\": \"DAY\",\n",
|
||||
" \"queryGranularity\": {\n",
|
||||
" \"type\": \"none\"\n",
|
||||
" },\n",
|
||||
" \"rollup\": False,\n",
|
||||
" \"intervals\": []\n",
|
||||
" },\n",
|
||||
" \"transformSpec\": {\n",
|
||||
" \"filter\": None,\n",
|
||||
" \"transforms\": []\n",
|
||||
" }\n",
|
||||
" },\n",
|
||||
" \"ioConfig\": {\n",
|
||||
" \"topic\": \"raw_scouter-opcua-orchestrated-pipeline\",\n",
|
||||
" \"topicPattern\": None,\n",
|
||||
" \"inputFormat\": {\n",
|
||||
" \"type\": \"kafka\",\n",
|
||||
" \"headerFormat\": None,\n",
|
||||
" \"keyFormat\": None,\n",
|
||||
" \"valueFormat\": {\n",
|
||||
" \"type\": \"json\",\n",
|
||||
" \"keepNoneColumns\": False,\n",
|
||||
" \"assumeNewlineDelimited\": False,\n",
|
||||
" \"useJsonNodeReader\": False\n",
|
||||
" },\n",
|
||||
" \"headerColumnPrefix\": \"kafka.header.\",\n",
|
||||
" \"keyColumnName\": \"kafka.key\",\n",
|
||||
" \"timestampColumnName\": \"kafka.timestamp\",\n",
|
||||
" \"topicColumnName\": \"kafka.topic\"\n",
|
||||
" },\n",
|
||||
" \"replicas\": 1,\n",
|
||||
" \"taskCount\": 1,\n",
|
||||
" \"taskDuration\": \"PT3600S\",\n",
|
||||
" \"consumerProperties\": {\n",
|
||||
" \"bootstrap.servers\": \"kafka.kafka.svc.cluster.local:9092\"\n",
|
||||
" },\n",
|
||||
" \"autoScalerConfig\": None,\n",
|
||||
" \"pollTimeout\": 100,\n",
|
||||
" \"startDelay\": \"PT5S\",\n",
|
||||
" \"period\": \"PT30S\",\n",
|
||||
" \"useEarliestOffset\": True,\n",
|
||||
" \"completionTimeout\": \"PT1800S\",\n",
|
||||
" \"lateMessageRejectionPeriod\": None,\n",
|
||||
" \"earlyMessageRejectionPeriod\": None,\n",
|
||||
" \"lateMessageRejectionStartDateTime\": None,\n",
|
||||
" \"configOverrides\": None,\n",
|
||||
" \"idleConfig\": None,\n",
|
||||
" \"stopTaskCount\": None,\n",
|
||||
" \"stream\": \"raw_scouter-opcua-orchestrated-pipeline\",\n",
|
||||
" \"useEarliestSequenceNumber\": True\n",
|
||||
" },\n",
|
||||
" \"tuningConfig\": {\n",
|
||||
" \"type\": \"kafka\",\n",
|
||||
" \"appendableIndexSpec\": {\n",
|
||||
" \"type\": \"onheap\",\n",
|
||||
" \"preserveExistingMetrics\": False\n",
|
||||
" },\n",
|
||||
" \"maxRowsInMemory\": 150000,\n",
|
||||
" \"maxBytesInMemory\": 0,\n",
|
||||
" \"skipBytesInMemoryOverheadCheck\": False,\n",
|
||||
" \"maxRowsPerSegment\": 5000000,\n",
|
||||
" \"maxTotalRows\": None,\n",
|
||||
" \"intermediatePersistPeriod\": \"PT10M\",\n",
|
||||
" \"maxPendingPersists\": 0,\n",
|
||||
" \"indexSpec\": {\n",
|
||||
" \"bitmap\": {\n",
|
||||
" \"type\": \"roaring\"\n",
|
||||
" },\n",
|
||||
" \"dimensionCompression\": \"lz4\",\n",
|
||||
" \"stringDictionaryEncoding\": {\n",
|
||||
" \"type\": \"utf8\"\n",
|
||||
" },\n",
|
||||
" \"metricCompression\": \"lz4\",\n",
|
||||
" \"longEncoding\": \"longs\"\n",
|
||||
" },\n",
|
||||
" \"indexSpecForIntermediatePersists\": {\n",
|
||||
" \"bitmap\": {\n",
|
||||
" \"type\": \"roaring\"\n",
|
||||
" },\n",
|
||||
" \"dimensionCompression\": \"lz4\",\n",
|
||||
" \"stringDictionaryEncoding\": {\n",
|
||||
" \"type\": \"utf8\"\n",
|
||||
" },\n",
|
||||
" \"metricCompression\": \"lz4\",\n",
|
||||
" \"longEncoding\": \"longs\"\n",
|
||||
" },\n",
|
||||
" \"reportParseExceptions\": False,\n",
|
||||
" \"handoffConditionTimeout\": 900000,\n",
|
||||
" \"resetOffsetAutomatically\": False,\n",
|
||||
" \"segmentWriteOutMediumFactory\": None,\n",
|
||||
" \"workerThreads\": None,\n",
|
||||
" \"chatRetries\": 8,\n",
|
||||
" \"httpTimeout\": \"PT10S\",\n",
|
||||
" \"shutdownTimeout\": \"PT80S\",\n",
|
||||
" \"offsetFetchPeriod\": \"PT30S\",\n",
|
||||
" \"intermediateHandoffPeriod\": \"P2147483647D\",\n",
|
||||
" \"logParseExceptions\": False,\n",
|
||||
" \"maxParseExceptions\": 2147483647,\n",
|
||||
" \"maxSavedParseExceptions\": 0,\n",
|
||||
" \"numPersistThreads\": 1,\n",
|
||||
" \"skipSequenceNumberAvailabilityCheck\": False,\n",
|
||||
" \"repartitionTransitionDuration\": \"PT120S\"\n",
|
||||
" }\n",
|
||||
" },\n",
|
||||
" \"context\": None,\n",
|
||||
" \"suspended\": False\n",
|
||||
"}"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 18,
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"raw_scouter-opcua-pipeline-2\n",
|
||||
"raw_scouter-opcua-pipeline-3\n",
|
||||
"raw_scouter-opcua-pipeline-4\n",
|
||||
"raw_scouter-opcua-pipeline-5\n",
|
||||
"raw_scouter-opcua-pipeline-6\n",
|
||||
"raw_scouter-opcua-pipeline-7\n",
|
||||
"raw_scouter-opcua-pipeline-8\n",
|
||||
"raw_scouter-opcua-pipeline-9\n",
|
||||
"raw_scouter-opcua-pipeline-10\n",
|
||||
"raw_scouter-opcua-pipeline-11\n",
|
||||
"raw_scouter-opcua-pipeline-12\n",
|
||||
"raw_scouter-opcua-pipeline-13\n",
|
||||
"raw_scouter-opcua-pipeline-14\n",
|
||||
"raw_scouter-opcua-pipeline-15\n",
|
||||
"raw_scouter-opcua-pipeline-16\n",
|
||||
"raw_scouter-opcua-pipeline-17\n",
|
||||
"raw_scouter-opcua-pipeline-18\n",
|
||||
"raw_scouter-opcua-pipeline-19\n",
|
||||
"raw_scouter-opcua-pipeline-20\n",
|
||||
"raw_scouter-opcua-pipeline-21\n",
|
||||
"raw_scouter-opcua-pipeline-22\n",
|
||||
"raw_scouter-opcua-pipeline-23\n",
|
||||
"raw_scouter-opcua-pipeline-24\n",
|
||||
"raw_scouter-opcua-pipeline-25\n",
|
||||
"raw_scouter-opcua-pipeline-26\n",
|
||||
"raw_scouter-opcua-pipeline-27\n",
|
||||
"raw_scouter-opcua-pipeline-28\n",
|
||||
"raw_scouter-opcua-pipeline-29\n",
|
||||
"raw_scouter-opcua-pipeline-30\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"import json\n",
|
||||
"from copy import deepcopy\n",
|
||||
"\n",
|
||||
"json_list = []\n",
|
||||
"i = 0\n",
|
||||
"for i in range(1, 30):\n",
|
||||
" topic = f\"raw_scouter-opcua-pipeline-{i+1}\"\n",
|
||||
" print(topic)\n",
|
||||
" obj = deepcopy(spec)\n",
|
||||
" obj['spec']['dataSchema']['dataSource'] = topic\n",
|
||||
" obj['spec']['ioConfig']['topic'] = topic\n",
|
||||
" obj['spec']['ioConfig']['stream'] = topic\n",
|
||||
"\n",
|
||||
" json_list.append(obj)\n",
|
||||
"\n",
|
||||
"# Save to a new file\n",
|
||||
"with open('specs_30.json', 'w') as f:\n",
|
||||
" json.dump(json_list, f, indent=2)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 19,
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"[OK] raw_scouter-opcua-pipeline-2 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-3 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-4 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-5 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-6 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-7 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-8 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-9 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-10 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-11 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-12 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-13 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-14 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-15 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-16 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-17 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-18 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-19 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-20 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-21 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-22 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-23 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-24 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-25 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-26 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-27 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-28 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-29 enviado.\n",
|
||||
"[OK] raw_scouter-opcua-pipeline-30 enviado.\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"import os\n",
|
||||
"import requests\n",
|
||||
"\n",
|
||||
"DRUID_OVERLORD = os.getenv(\"DRUID_OVERLORD\", \"http://localhost:8082\")\n",
|
||||
"SUPERVISOR_ENDPOINT = f\"{DRUID_OVERLORD}/druid/indexer/v1/supervisor\"\n",
|
||||
"\n",
|
||||
"def enviar_supervisores(specs):\n",
|
||||
" for spec in specs:\n",
|
||||
" resp = requests.post(\n",
|
||||
" SUPERVISOR_ENDPOINT,\n",
|
||||
" headers={\"Content-Type\": \"application/json\"},\n",
|
||||
" json=spec\n",
|
||||
" )\n",
|
||||
" if resp.status_code == 200:\n",
|
||||
" print(f\"[OK] {spec['spec']['dataSchema']['dataSource']} enviado.\")\n",
|
||||
" else:\n",
|
||||
" print(f\"[ERRO] {spec['spec']['dataSchema']['dataSource']}: {resp.status_code} → {resp.text}\")\n",
|
||||
"\n",
|
||||
"with open(\"./specs_30.json\", \"r\") as f:\n",
|
||||
" specs = json.load(f)\n",
|
||||
"enviar_supervisores(specs)\n"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 20,
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"ename": "ConnectionError",
|
||||
"evalue": "HTTPConnectionPool(host='localhost', port=8090): Max retries exceeded with url: /druid/indexer/v1/supervisor/raw_scouter-opcua-orchestrated-pipeline-2/terminate (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x72128094cc50>: Failed to establish a new connection: [Errno 111] Connection refused'))",
|
||||
"output_type": "error",
|
||||
"traceback": [
|
||||
"\u001b[31m---------------------------------------------------------------------------\u001b[39m",
|
||||
"\u001b[31mConnectionRefusedError\u001b[39m Traceback (most recent call last)",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connection.py:198\u001b[39m, in \u001b[36mHTTPConnection._new_conn\u001b[39m\u001b[34m(self)\u001b[39m\n\u001b[32m 197\u001b[39m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[32m--> \u001b[39m\u001b[32m198\u001b[39m sock = \u001b[43mconnection\u001b[49m\u001b[43m.\u001b[49m\u001b[43mcreate_connection\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m 199\u001b[39m \u001b[43m \u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_dns_host\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43mport\u001b[49m\u001b[43m)\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 200\u001b[39m \u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43mtimeout\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 201\u001b[39m \u001b[43m \u001b[49m\u001b[43msource_address\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43msource_address\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 202\u001b[39m \u001b[43m \u001b[49m\u001b[43msocket_options\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43msocket_options\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 203\u001b[39m \u001b[43m \u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 204\u001b[39m \u001b[38;5;28;01mexcept\u001b[39;00m socket.gaierror \u001b[38;5;28;01mas\u001b[39;00m e:\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/util/connection.py:85\u001b[39m, in \u001b[36mcreate_connection\u001b[39m\u001b[34m(address, timeout, source_address, socket_options)\u001b[39m\n\u001b[32m 84\u001b[39m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[32m---> \u001b[39m\u001b[32m85\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m err\n\u001b[32m 86\u001b[39m \u001b[38;5;28;01mfinally\u001b[39;00m:\n\u001b[32m 87\u001b[39m \u001b[38;5;66;03m# Break explicitly a reference cycle\u001b[39;00m\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/util/connection.py:73\u001b[39m, in \u001b[36mcreate_connection\u001b[39m\u001b[34m(address, timeout, source_address, socket_options)\u001b[39m\n\u001b[32m 72\u001b[39m sock.bind(source_address)\n\u001b[32m---> \u001b[39m\u001b[32m73\u001b[39m \u001b[43msock\u001b[49m\u001b[43m.\u001b[49m\u001b[43mconnect\u001b[49m\u001b[43m(\u001b[49m\u001b[43msa\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 74\u001b[39m \u001b[38;5;66;03m# Break explicitly a reference cycle\u001b[39;00m\n",
|
||||
"\u001b[31mConnectionRefusedError\u001b[39m: [Errno 111] Connection refused",
|
||||
"\nThe above exception was the direct cause of the following exception:\n",
|
||||
"\u001b[31mNewConnectionError\u001b[39m Traceback (most recent call last)",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connectionpool.py:787\u001b[39m, in \u001b[36mHTTPConnectionPool.urlopen\u001b[39m\u001b[34m(self, method, url, body, headers, retries, redirect, assert_same_host, timeout, pool_timeout, release_conn, chunked, body_pos, preload_content, decode_content, **response_kw)\u001b[39m\n\u001b[32m 786\u001b[39m \u001b[38;5;66;03m# Make the request on the HTTPConnection object\u001b[39;00m\n\u001b[32m--> \u001b[39m\u001b[32m787\u001b[39m response = \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_make_request\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m 788\u001b[39m \u001b[43m \u001b[49m\u001b[43mconn\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 789\u001b[39m \u001b[43m \u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 790\u001b[39m \u001b[43m \u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 791\u001b[39m \u001b[43m \u001b[49m\u001b[43mtimeout\u001b[49m\u001b[43m=\u001b[49m\u001b[43mtimeout_obj\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 792\u001b[39m \u001b[43m \u001b[49m\u001b[43mbody\u001b[49m\u001b[43m=\u001b[49m\u001b[43mbody\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 793\u001b[39m \u001b[43m \u001b[49m\u001b[43mheaders\u001b[49m\u001b[43m=\u001b[49m\u001b[43mheaders\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 794\u001b[39m \u001b[43m \u001b[49m\u001b[43mchunked\u001b[49m\u001b[43m=\u001b[49m\u001b[43mchunked\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 795\u001b[39m \u001b[43m \u001b[49m\u001b[43mretries\u001b[49m\u001b[43m=\u001b[49m\u001b[43mretries\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 796\u001b[39m \u001b[43m \u001b[49m\u001b[43mresponse_conn\u001b[49m\u001b[43m=\u001b[49m\u001b[43mresponse_conn\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 797\u001b[39m \u001b[43m \u001b[49m\u001b[43mpreload_content\u001b[49m\u001b[43m=\u001b[49m\u001b[43mpreload_content\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 798\u001b[39m \u001b[43m \u001b[49m\u001b[43mdecode_content\u001b[49m\u001b[43m=\u001b[49m\u001b[43mdecode_content\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 799\u001b[39m \u001b[43m \u001b[49m\u001b[43m*\u001b[49m\u001b[43m*\u001b[49m\u001b[43mresponse_kw\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 800\u001b[39m \u001b[43m\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 802\u001b[39m \u001b[38;5;66;03m# Everything went great!\u001b[39;00m\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connectionpool.py:493\u001b[39m, in \u001b[36mHTTPConnectionPool._make_request\u001b[39m\u001b[34m(self, conn, method, url, body, headers, retries, timeout, chunked, response_conn, preload_content, decode_content, enforce_content_length)\u001b[39m\n\u001b[32m 492\u001b[39m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[32m--> \u001b[39m\u001b[32m493\u001b[39m \u001b[43mconn\u001b[49m\u001b[43m.\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m 494\u001b[39m \u001b[43m \u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 495\u001b[39m \u001b[43m \u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 496\u001b[39m \u001b[43m \u001b[49m\u001b[43mbody\u001b[49m\u001b[43m=\u001b[49m\u001b[43mbody\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 497\u001b[39m \u001b[43m \u001b[49m\u001b[43mheaders\u001b[49m\u001b[43m=\u001b[49m\u001b[43mheaders\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 498\u001b[39m \u001b[43m \u001b[49m\u001b[43mchunked\u001b[49m\u001b[43m=\u001b[49m\u001b[43mchunked\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 499\u001b[39m \u001b[43m \u001b[49m\u001b[43mpreload_content\u001b[49m\u001b[43m=\u001b[49m\u001b[43mpreload_content\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 500\u001b[39m \u001b[43m \u001b[49m\u001b[43mdecode_content\u001b[49m\u001b[43m=\u001b[49m\u001b[43mdecode_content\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 501\u001b[39m \u001b[43m \u001b[49m\u001b[43menforce_content_length\u001b[49m\u001b[43m=\u001b[49m\u001b[43menforce_content_length\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 502\u001b[39m \u001b[43m \u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 504\u001b[39m \u001b[38;5;66;03m# We are swallowing BrokenPipeError (errno.EPIPE) since the server is\u001b[39;00m\n\u001b[32m 505\u001b[39m \u001b[38;5;66;03m# legitimately able to close the connection after sending a valid response.\u001b[39;00m\n\u001b[32m 506\u001b[39m \u001b[38;5;66;03m# With this behaviour, the received response is still readable.\u001b[39;00m\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connection.py:494\u001b[39m, in \u001b[36mHTTPConnection.request\u001b[39m\u001b[34m(self, method, url, body, headers, chunked, preload_content, decode_content, enforce_content_length)\u001b[39m\n\u001b[32m 493\u001b[39m \u001b[38;5;28mself\u001b[39m.putheader(header, value)\n\u001b[32m--> \u001b[39m\u001b[32m494\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43mendheaders\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 496\u001b[39m \u001b[38;5;66;03m# If we're given a body we start sending that in chunks.\u001b[39;00m\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m/usr/lib/python3.11/http/client.py:1298\u001b[39m, in \u001b[36mHTTPConnection.endheaders\u001b[39m\u001b[34m(self, message_body, encode_chunked)\u001b[39m\n\u001b[32m 1297\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m CannotSendHeader()\n\u001b[32m-> \u001b[39m\u001b[32m1298\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_send_output\u001b[49m\u001b[43m(\u001b[49m\u001b[43mmessage_body\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mencode_chunked\u001b[49m\u001b[43m=\u001b[49m\u001b[43mencode_chunked\u001b[49m\u001b[43m)\u001b[49m\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m/usr/lib/python3.11/http/client.py:1058\u001b[39m, in \u001b[36mHTTPConnection._send_output\u001b[39m\u001b[34m(self, message_body, encode_chunked)\u001b[39m\n\u001b[32m 1057\u001b[39m \u001b[38;5;28;01mdel\u001b[39;00m \u001b[38;5;28mself\u001b[39m._buffer[:]\n\u001b[32m-> \u001b[39m\u001b[32m1058\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43msend\u001b[49m\u001b[43m(\u001b[49m\u001b[43mmsg\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 1060\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m message_body \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[32m 1061\u001b[39m \n\u001b[32m 1062\u001b[39m \u001b[38;5;66;03m# create a consistent interface to message_body\u001b[39;00m\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m/usr/lib/python3.11/http/client.py:996\u001b[39m, in \u001b[36mHTTPConnection.send\u001b[39m\u001b[34m(self, data)\u001b[39m\n\u001b[32m 995\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m.auto_open:\n\u001b[32m--> \u001b[39m\u001b[32m996\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43mconnect\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 997\u001b[39m \u001b[38;5;28;01melse\u001b[39;00m:\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connection.py:325\u001b[39m, in \u001b[36mHTTPConnection.connect\u001b[39m\u001b[34m(self)\u001b[39m\n\u001b[32m 324\u001b[39m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34mconnect\u001b[39m(\u001b[38;5;28mself\u001b[39m) -> \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[32m--> \u001b[39m\u001b[32m325\u001b[39m \u001b[38;5;28mself\u001b[39m.sock = \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_new_conn\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 326\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m._tunnel_host:\n\u001b[32m 327\u001b[39m \u001b[38;5;66;03m# If we're tunneling it means we're connected to our proxy.\u001b[39;00m\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connection.py:213\u001b[39m, in \u001b[36mHTTPConnection._new_conn\u001b[39m\u001b[34m(self)\u001b[39m\n\u001b[32m 212\u001b[39m \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mOSError\u001b[39;00m \u001b[38;5;28;01mas\u001b[39;00m e:\n\u001b[32m--> \u001b[39m\u001b[32m213\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m NewConnectionError(\n\u001b[32m 214\u001b[39m \u001b[38;5;28mself\u001b[39m, \u001b[33mf\u001b[39m\u001b[33m\"\u001b[39m\u001b[33mFailed to establish a new connection: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00me\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m\"\u001b[39m\n\u001b[32m 215\u001b[39m ) \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34;01me\u001b[39;00m\n\u001b[32m 217\u001b[39m sys.audit(\u001b[33m\"\u001b[39m\u001b[33mhttp.client.connect\u001b[39m\u001b[33m\"\u001b[39m, \u001b[38;5;28mself\u001b[39m, \u001b[38;5;28mself\u001b[39m.host, \u001b[38;5;28mself\u001b[39m.port)\n",
|
||||
"\u001b[31mNewConnectionError\u001b[39m: <urllib3.connection.HTTPConnection object at 0x72128094cc50>: Failed to establish a new connection: [Errno 111] Connection refused",
|
||||
"\nThe above exception was the direct cause of the following exception:\n",
|
||||
"\u001b[31mMaxRetryError\u001b[39m Traceback (most recent call last)",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/requests/adapters.py:667\u001b[39m, in \u001b[36mHTTPAdapter.send\u001b[39m\u001b[34m(self, request, stream, timeout, verify, cert, proxies)\u001b[39m\n\u001b[32m 666\u001b[39m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[32m--> \u001b[39m\u001b[32m667\u001b[39m resp = \u001b[43mconn\u001b[49m\u001b[43m.\u001b[49m\u001b[43murlopen\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m 668\u001b[39m \u001b[43m \u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m=\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m.\u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 669\u001b[39m \u001b[43m \u001b[49m\u001b[43murl\u001b[49m\u001b[43m=\u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 670\u001b[39m \u001b[43m \u001b[49m\u001b[43mbody\u001b[49m\u001b[43m=\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m.\u001b[49m\u001b[43mbody\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 671\u001b[39m \u001b[43m \u001b[49m\u001b[43mheaders\u001b[49m\u001b[43m=\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m.\u001b[49m\u001b[43mheaders\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 672\u001b[39m \u001b[43m \u001b[49m\u001b[43mredirect\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43;01mFalse\u001b[39;49;00m\u001b[43m,\u001b[49m\n\u001b[32m 673\u001b[39m \u001b[43m \u001b[49m\u001b[43massert_same_host\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43;01mFalse\u001b[39;49;00m\u001b[43m,\u001b[49m\n\u001b[32m 674\u001b[39m \u001b[43m \u001b[49m\u001b[43mpreload_content\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43;01mFalse\u001b[39;49;00m\u001b[43m,\u001b[49m\n\u001b[32m 675\u001b[39m \u001b[43m \u001b[49m\u001b[43mdecode_content\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43;01mFalse\u001b[39;49;00m\u001b[43m,\u001b[49m\n\u001b[32m 676\u001b[39m \u001b[43m \u001b[49m\u001b[43mretries\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43mmax_retries\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 677\u001b[39m \u001b[43m \u001b[49m\u001b[43mtimeout\u001b[49m\u001b[43m=\u001b[49m\u001b[43mtimeout\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 678\u001b[39m \u001b[43m \u001b[49m\u001b[43mchunked\u001b[49m\u001b[43m=\u001b[49m\u001b[43mchunked\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 679\u001b[39m \u001b[43m \u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 681\u001b[39m \u001b[38;5;28;01mexcept\u001b[39;00m (ProtocolError, \u001b[38;5;167;01mOSError\u001b[39;00m) \u001b[38;5;28;01mas\u001b[39;00m err:\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/connectionpool.py:841\u001b[39m, in \u001b[36mHTTPConnectionPool.urlopen\u001b[39m\u001b[34m(self, method, url, body, headers, retries, redirect, assert_same_host, timeout, pool_timeout, release_conn, chunked, body_pos, preload_content, decode_content, **response_kw)\u001b[39m\n\u001b[32m 839\u001b[39m new_e = ProtocolError(\u001b[33m\"\u001b[39m\u001b[33mConnection aborted.\u001b[39m\u001b[33m\"\u001b[39m, new_e)\n\u001b[32m--> \u001b[39m\u001b[32m841\u001b[39m retries = \u001b[43mretries\u001b[49m\u001b[43m.\u001b[49m\u001b[43mincrement\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m 842\u001b[39m \u001b[43m \u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43merror\u001b[49m\u001b[43m=\u001b[49m\u001b[43mnew_e\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m_pool\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m_stacktrace\u001b[49m\u001b[43m=\u001b[49m\u001b[43msys\u001b[49m\u001b[43m.\u001b[49m\u001b[43mexc_info\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\u001b[43m[\u001b[49m\u001b[32;43m2\u001b[39;49m\u001b[43m]\u001b[49m\n\u001b[32m 843\u001b[39m \u001b[43m\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 844\u001b[39m retries.sleep()\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/urllib3/util/retry.py:519\u001b[39m, in \u001b[36mRetry.increment\u001b[39m\u001b[34m(self, method, url, response, error, _pool, _stacktrace)\u001b[39m\n\u001b[32m 518\u001b[39m reason = error \u001b[38;5;129;01mor\u001b[39;00m ResponseError(cause)\n\u001b[32m--> \u001b[39m\u001b[32m519\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m MaxRetryError(_pool, url, reason) \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34;01mreason\u001b[39;00m \u001b[38;5;66;03m# type: ignore[arg-type]\u001b[39;00m\n\u001b[32m 521\u001b[39m log.debug(\u001b[33m\"\u001b[39m\u001b[33mIncremented Retry for (url=\u001b[39m\u001b[33m'\u001b[39m\u001b[38;5;132;01m%s\u001b[39;00m\u001b[33m'\u001b[39m\u001b[33m): \u001b[39m\u001b[38;5;132;01m%r\u001b[39;00m\u001b[33m\"\u001b[39m, url, new_retry)\n",
|
||||
"\u001b[31mMaxRetryError\u001b[39m: HTTPConnectionPool(host='localhost', port=8090): Max retries exceeded with url: /druid/indexer/v1/supervisor/raw_scouter-opcua-orchestrated-pipeline-2/terminate (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x72128094cc50>: Failed to establish a new connection: [Errno 111] Connection refused'))",
|
||||
"\nDuring handling of the above exception, another exception occurred:\n",
|
||||
"\u001b[31mConnectionError\u001b[39m Traceback (most recent call last)",
|
||||
"\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[20]\u001b[39m\u001b[32m, line 16\u001b[39m\n\u001b[32m 13\u001b[39m \u001b[38;5;28mprint\u001b[39m(\u001b[33mf\u001b[39m\u001b[33m\"\u001b[39m\u001b[33m❌ Erro \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mresp.status_code\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mresp.text\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m\"\u001b[39m)\n\u001b[32m 15\u001b[39m \u001b[38;5;28;01mfor\u001b[39;00m i \u001b[38;5;129;01min\u001b[39;00m \u001b[38;5;28mrange\u001b[39m(\u001b[32m1\u001b[39m, \u001b[32m30\u001b[39m):\n\u001b[32m---> \u001b[39m\u001b[32m16\u001b[39m \u001b[43mterminate_supervisor\u001b[49m\u001b[43m(\u001b[49m\u001b[33;43mf\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43mraw_scouter-opcua-orchestrated-pipeline-\u001b[39;49m\u001b[38;5;132;43;01m{\u001b[39;49;00m\u001b[43mi\u001b[49m\u001b[43m+\u001b[49m\u001b[32;43m1\u001b[39;49m\u001b[38;5;132;43;01m}\u001b[39;49;00m\u001b[33;43m\"\u001b[39;49m\u001b[43m)\u001b[49m\n",
|
||||
"\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[20]\u001b[39m\u001b[32m, line 7\u001b[39m, in \u001b[36mterminate_supervisor\u001b[39m\u001b[34m(supervisor_id)\u001b[39m\n\u001b[32m 5\u001b[39m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34mterminate_supervisor\u001b[39m(supervisor_id):\n\u001b[32m 6\u001b[39m url = \u001b[33mf\u001b[39m\u001b[33m\"\u001b[39m\u001b[38;5;132;01m{\u001b[39;00mDRUID\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m/druid/indexer/v1/supervisor/\u001b[39m\u001b[38;5;132;01m{\u001b[39;00msupervisor_id\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m/terminate\u001b[39m\u001b[33m\"\u001b[39m\n\u001b[32m----> \u001b[39m\u001b[32m7\u001b[39m resp = \u001b[43mrequests\u001b[49m\u001b[43m.\u001b[49m\u001b[43mpost\u001b[49m\u001b[43m(\u001b[49m\u001b[43murl\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 8\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m resp.status_code == \u001b[32m200\u001b[39m:\n\u001b[32m 9\u001b[39m \u001b[38;5;28mprint\u001b[39m(\u001b[33mf\u001b[39m\u001b[33m\"\u001b[39m\u001b[33m✅ Supervisor \u001b[39m\u001b[33m'\u001b[39m\u001b[38;5;132;01m{\u001b[39;00msupervisor_id\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m'\u001b[39m\u001b[33m encerrado.\u001b[39m\u001b[33m\"\u001b[39m)\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/requests/api.py:115\u001b[39m, in \u001b[36mpost\u001b[39m\u001b[34m(url, data, json, **kwargs)\u001b[39m\n\u001b[32m 103\u001b[39m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34mpost\u001b[39m(url, data=\u001b[38;5;28;01mNone\u001b[39;00m, json=\u001b[38;5;28;01mNone\u001b[39;00m, **kwargs):\n\u001b[32m 104\u001b[39m \u001b[38;5;250m \u001b[39m\u001b[33mr\u001b[39m\u001b[33;03m\"\"\"Sends a POST request.\u001b[39;00m\n\u001b[32m 105\u001b[39m \n\u001b[32m 106\u001b[39m \u001b[33;03m :param url: URL for the new :class:`Request` object.\u001b[39;00m\n\u001b[32m (...)\u001b[39m\u001b[32m 112\u001b[39m \u001b[33;03m :rtype: requests.Response\u001b[39;00m\n\u001b[32m 113\u001b[39m \u001b[33;03m \"\"\"\u001b[39;00m\n\u001b[32m--> \u001b[39m\u001b[32m115\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[43mrequest\u001b[49m\u001b[43m(\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43mpost\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mdata\u001b[49m\u001b[43m=\u001b[49m\u001b[43mdata\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mjson\u001b[49m\u001b[43m=\u001b[49m\u001b[43mjson\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m*\u001b[49m\u001b[43m*\u001b[49m\u001b[43mkwargs\u001b[49m\u001b[43m)\u001b[49m\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/requests/api.py:59\u001b[39m, in \u001b[36mrequest\u001b[39m\u001b[34m(method, url, **kwargs)\u001b[39m\n\u001b[32m 55\u001b[39m \u001b[38;5;66;03m# By using the 'with' statement we are sure the session is closed, thus we\u001b[39;00m\n\u001b[32m 56\u001b[39m \u001b[38;5;66;03m# avoid leaving sockets open which can trigger a ResourceWarning in some\u001b[39;00m\n\u001b[32m 57\u001b[39m \u001b[38;5;66;03m# cases, and look like a memory leak in others.\u001b[39;00m\n\u001b[32m 58\u001b[39m \u001b[38;5;28;01mwith\u001b[39;00m sessions.Session() \u001b[38;5;28;01mas\u001b[39;00m session:\n\u001b[32m---> \u001b[39m\u001b[32m59\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[43msession\u001b[49m\u001b[43m.\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m(\u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m=\u001b[49m\u001b[43mmethod\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43murl\u001b[49m\u001b[43m=\u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m*\u001b[49m\u001b[43m*\u001b[49m\u001b[43mkwargs\u001b[49m\u001b[43m)\u001b[49m\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/requests/sessions.py:589\u001b[39m, in \u001b[36mSession.request\u001b[39m\u001b[34m(self, method, url, params, data, headers, cookies, files, auth, timeout, allow_redirects, proxies, hooks, stream, verify, cert, json)\u001b[39m\n\u001b[32m 584\u001b[39m send_kwargs = {\n\u001b[32m 585\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mtimeout\u001b[39m\u001b[33m\"\u001b[39m: timeout,\n\u001b[32m 586\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mallow_redirects\u001b[39m\u001b[33m\"\u001b[39m: allow_redirects,\n\u001b[32m 587\u001b[39m }\n\u001b[32m 588\u001b[39m send_kwargs.update(settings)\n\u001b[32m--> \u001b[39m\u001b[32m589\u001b[39m resp = \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43msend\u001b[49m\u001b[43m(\u001b[49m\u001b[43mprep\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m*\u001b[49m\u001b[43m*\u001b[49m\u001b[43msend_kwargs\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 591\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m resp\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/requests/sessions.py:703\u001b[39m, in \u001b[36mSession.send\u001b[39m\u001b[34m(self, request, **kwargs)\u001b[39m\n\u001b[32m 700\u001b[39m start = preferred_clock()\n\u001b[32m 702\u001b[39m \u001b[38;5;66;03m# Send the request\u001b[39;00m\n\u001b[32m--> \u001b[39m\u001b[32m703\u001b[39m r = \u001b[43madapter\u001b[49m\u001b[43m.\u001b[49m\u001b[43msend\u001b[49m\u001b[43m(\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m*\u001b[49m\u001b[43m*\u001b[49m\u001b[43mkwargs\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 705\u001b[39m \u001b[38;5;66;03m# Total elapsed time of the request (approximately)\u001b[39;00m\n\u001b[32m 706\u001b[39m elapsed = preferred_clock() - start\n",
|
||||
"\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/requests/adapters.py:700\u001b[39m, in \u001b[36mHTTPAdapter.send\u001b[39m\u001b[34m(self, request, stream, timeout, verify, cert, proxies)\u001b[39m\n\u001b[32m 696\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(e.reason, _SSLError):\n\u001b[32m 697\u001b[39m \u001b[38;5;66;03m# This branch is for urllib3 v1.22 and later.\u001b[39;00m\n\u001b[32m 698\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m SSLError(e, request=request)\n\u001b[32m--> \u001b[39m\u001b[32m700\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mConnectionError\u001b[39;00m(e, request=request)\n\u001b[32m 702\u001b[39m \u001b[38;5;28;01mexcept\u001b[39;00m ClosedPoolError \u001b[38;5;28;01mas\u001b[39;00m e:\n\u001b[32m 703\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mConnectionError\u001b[39;00m(e, request=request)\n",
|
||||
"\u001b[31mConnectionError\u001b[39m: HTTPConnectionPool(host='localhost', port=8090): Max retries exceeded with url: /druid/indexer/v1/supervisor/raw_scouter-opcua-orchestrated-pipeline-2/terminate (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x72128094cc50>: Failed to establish a new connection: [Errno 111] Connection refused'))"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"import os\n",
|
||||
"import requests\n",
|
||||
"\n",
|
||||
"DRUID = os.getenv(\"DRUID_URL\", \"http://localhost:8082\")\n",
|
||||
"def terminate_supervisor(supervisor_id):\n",
|
||||
" url = f\"{DRUID}/druid/indexer/v1/supervisor/{supervisor_id}/terminate\"\n",
|
||||
" resp = requests.post(url)\n",
|
||||
" if resp.status_code == 200:\n",
|
||||
" print(f\"✅ Supervisor '{supervisor_id}' encerrado.\")\n",
|
||||
" elif resp.status_code == 404:\n",
|
||||
" print(f\"⚠️ Supervisor '{supervisor_id}' não encontrado.\")\n",
|
||||
" else:\n",
|
||||
" print(f\"❌ Erro {resp.status_code}: {resp.text}\")\n",
|
||||
"\n",
|
||||
"for i in range(1, 30):\n",
|
||||
" terminate_supervisor(f\"raw_scouter-opcua-orchestrated-pipeline-{i+1}\")\n"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 3,
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"✅ Document deleted successfully\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"from pymongo import MongoClient\n",
|
||||
"import os\n",
|
||||
"# Get MongoDB connection details from environment variables\n",
|
||||
"MONGODB_USERNAME = os.getenv(\"MONGODB_USERNAME\", \"root\")\n",
|
||||
"MONGODB_PASSWORD = os.getenv(\"MONGODB_PASSWORD\", \"wKZDbMNU1c\") \n",
|
||||
"MONGODB_URL = os.getenv(\"MONGODB_URL\", \"localhost:27018\")\n",
|
||||
"MONGODB_DATABASE = os.getenv(\"MONGODB_DATABASE\", \"sientia\")\n",
|
||||
"\n",
|
||||
"# Create MongoDB client\n",
|
||||
"client = MongoClient(\n",
|
||||
" f\"mongodb://{MONGODB_USERNAME}:{MONGODB_PASSWORD}@{MONGODB_URL}\"\n",
|
||||
")\n",
|
||||
"\n",
|
||||
"# Get database and collection\n",
|
||||
"db = client[MONGODB_DATABASE]\n",
|
||||
"collection = db[\"pipelines\"] # Replace with actual collection name\n",
|
||||
"\n",
|
||||
"for i in range(2, 31):\n",
|
||||
" # Delete a document matching specific criteria\n",
|
||||
" result = collection.delete_one({\"schedule_name\": f\"scouter-opcua-pipeline-{i}\"}) # Replace with actual query\n",
|
||||
"\n",
|
||||
"if result.deleted_count > 0:\n",
|
||||
" print(\"✅ Document deleted successfully\")\n",
|
||||
"else:\n",
|
||||
" print(\"⚠️ No matching document found\")\n",
|
||||
"\n",
|
||||
"# Close the connection\n",
|
||||
"client.close()\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"display_name": "venv",
|
||||
"language": "python",
|
||||
"name": "python3"
|
||||
},
|
||||
"language_info": {
|
||||
"codemirror_mode": {
|
||||
"name": "ipython",
|
||||
"version": 3
|
||||
},
|
||||
"file_extension": ".py",
|
||||
"mimetype": "text/x-python",
|
||||
"name": "python",
|
||||
"nbconvert_exporter": "python",
|
||||
"pygments_lexer": "ipython3",
|
||||
"version": "3.11.13"
|
||||
}
|
||||
},
|
||||
"nbformat": 4,
|
||||
"nbformat_minor": 2
|
||||
}
|
||||
62
values.yaml
62
values.yaml
@@ -11,7 +11,7 @@ image:
|
||||
# This sets the pull policy for images.
|
||||
pullPolicy: Always
|
||||
# Overrides the image tag whose default is the chart appVersion.
|
||||
tag: "0.5.2"
|
||||
tag: "1.0.0"
|
||||
|
||||
# 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:
|
||||
@@ -52,7 +52,7 @@ securityContext: {}
|
||||
# runAsUser: 1000
|
||||
|
||||
|
||||
resources: {}
|
||||
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
|
||||
@@ -64,24 +64,37 @@ resources: {}
|
||||
# cpu: 100m
|
||||
# memory: 128Mi
|
||||
|
||||
limits:
|
||||
cpu: 1000m
|
||||
memory: 2048Mi
|
||||
requests:
|
||||
cpu: 300m
|
||||
memory: 256Mi
|
||||
|
||||
# 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
|
||||
- |
|
||||
curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1'
|
||||
initialDelaySeconds: 30
|
||||
periodSeconds: 15
|
||||
timeoutSeconds: 5
|
||||
failureThreshold: 3
|
||||
|
||||
readinessProbe:
|
||||
exec:
|
||||
command:
|
||||
- sh
|
||||
- -c
|
||||
- pgrep -f "scouter.worker.worker"
|
||||
initialDelaySeconds: 10
|
||||
periodSeconds: 15
|
||||
- |
|
||||
curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1'
|
||||
initialDelaySeconds: 20
|
||||
periodSeconds: 10
|
||||
timeoutSeconds: 3
|
||||
failureThreshold: 2
|
||||
|
||||
|
||||
# This section is for setting up autoscaling more information can be found here: https://kubernetes.io/docs/concepts/workloads/autoscaling/
|
||||
@@ -150,7 +163,7 @@ env:
|
||||
- name: GITHUB_REPO_URL
|
||||
value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git"
|
||||
- name: GITHUB_BRANCH
|
||||
value: "fix/SIENTIAPDE-1273"
|
||||
value: "release/SIENTIAPDE-1441"
|
||||
- name: PYTHON_APP
|
||||
value: "scouter.worker.worker"
|
||||
|
||||
@@ -206,10 +219,33 @@ env:
|
||||
- name: MONGODB_DATABASE
|
||||
value: "sientia"
|
||||
|
||||
- name: DRUID_HOST
|
||||
value: "druid-router.druid.svc.cluster.local"
|
||||
- name: DRUID_PORT
|
||||
value: "8888"
|
||||
- name: PYPI_SERVER
|
||||
value: "http://library-distribution-server.library.svc.cluster.local:5000"
|
||||
|
||||
# Temporal worker tuning
|
||||
- name: SCOUTER_MAX_CONCURRENT_WORKFLOW_TASKS
|
||||
value: "200"
|
||||
- name: SCOUTER_MAX_CONCURRENT_ACTIVITIES
|
||||
value: "200"
|
||||
- name: SCOUTER_MAX_CONCURRENT_LOCAL_ACTIVITIES
|
||||
value: "200"
|
||||
- name: SCOUTER_MAX_CACHED_WORKFLOWS
|
||||
value: "200"
|
||||
|
||||
- name: SCOUTER_WORKFLOW_POLLER_BEHAVIUR_MINIMUM
|
||||
value: "10"
|
||||
- name: SCOUTER_WORKFLOW_POLLER_BEHAVIUR_INITIAL
|
||||
value: "100"
|
||||
- name: SCOUTER_WORKFLOW_POLLER_BEHAVIUR_MAXIMUM
|
||||
value: "200"
|
||||
|
||||
- name: SCOUTER_ACTIVITY_POLLER_BEHAVIUR_MINIMUM
|
||||
value: "10"
|
||||
- name: SCOUTER_ACTIVITY_POLLER_BEHAVIUR_INITIAL
|
||||
value: "100"
|
||||
- name: SCOUTER_ACTIVITY_POLLER_BEHAVIUR_MAXIMUM
|
||||
value: "200"
|
||||
|
||||
|
||||
ssh:
|
||||
enabled: true
|
||||
|
||||
Reference in New Issue
Block a user