Compare commits
17 Commits
fcc8920a8b
...
fix/SIENTI
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
695e4c07a6 | ||
|
|
6a8c41b328 | ||
|
|
773c980fc3 | ||
|
|
35efb67c89 | ||
|
|
acd925cf2a | ||
|
|
310ceea0d8 | ||
|
|
2d73ec9ec2 | ||
|
|
2142143ab9 | ||
|
|
c07457bfbe | ||
|
|
086b12492e | ||
|
|
4ea0754f0c | ||
|
|
ddb1618209 | ||
|
|
90f8bdda61 | ||
|
|
2ccda3e440 | ||
|
|
d856150e24 | ||
|
|
6569810756 | ||
|
|
7153f1da0d |
18
.github/workflows/quality-gate.yml
vendored
18
.github/workflows/quality-gate.yml
vendored
@@ -1,15 +1,29 @@
|
||||
name: Quality gate
|
||||
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- main
|
||||
- 'release/**'
|
||||
- 'feature/**'
|
||||
pull_request:
|
||||
branches:
|
||||
- main
|
||||
- 'release/**'
|
||||
- 'feature/**'
|
||||
types: [ opened, synchronize, reopened ]
|
||||
|
||||
concurrency:
|
||||
group: ${{ github.workflow }}-${{ github.ref }}
|
||||
cancel-in-progress: true
|
||||
|
||||
jobs:
|
||||
quality-gate:
|
||||
uses: Aignosi/github_workflow_templates/.github/workflows/dataops-module-quality-gate.yml@main
|
||||
permissions: write-all
|
||||
uses: Aignosi/github_workflow_templates/.github/workflows/python-quality-gate.yml@main
|
||||
permissions:
|
||||
contents: read
|
||||
pull-requests: write
|
||||
issues: write
|
||||
with:
|
||||
project_name: 'laborious'
|
||||
repositories: 'sientia-dataops-library, sientia-mlops-library'
|
||||
|
||||
15
README.md
15
README.md
@@ -126,10 +126,15 @@ Laborious uses a Temporal-based architecture with strong separation of concerns
|
||||
### Key Components
|
||||
|
||||
#### **Worker (`laborious/worker/worker.py`)**
|
||||
- Temporal client setup, worker lifecycle, task queues
|
||||
- Temporal client setup, four workers via `sientia_do.temporal.worker.prepare_worker`
|
||||
- Runtime-scoped task queues: `{workflow}-{RUNTIME}-queue` for all workflows
|
||||
- Metrics server initialization, notification handler setup
|
||||
- Graceful shutdown and autoscaling-friendly behavior
|
||||
|
||||
**Breaking (schedulers):** drift and simple_metrics queues are no longer `drift-queue` /
|
||||
`simple_metrics-queue`. Use `drift-{RUNTIME}-queue` and `simple_metrics-{RUNTIME}-queue`
|
||||
matching the worker pod `RUNTIME` env (same as `predictions_batch` / `minimal_retrain`).
|
||||
|
||||
#### **Workflows (`laborious/workflows/`)**
|
||||
- `predictions_batch.py`: Batch prediction entry point
|
||||
- `sub_workflows/prediction_process.py`: Core prediction pipeline
|
||||
@@ -802,6 +807,7 @@ See [OPC UA Communication](#opc-ua-communication) for semantics, concurrency, an
|
||||
|----------|-------------|---------|----------|
|
||||
| `TEMPORAL_HOST` | Temporal server address | `localhost:7233` | Yes |
|
||||
| `TEMPORAL_NAMESPACE` | Temporal namespace | `laborious` | No |
|
||||
| `RUNTIME` | Task queue suffix for all workflows (`{workflow}-{RUNTIME}-queue`) | _(none)_ | Yes |
|
||||
| `POSTGRES_HOST` | PostgreSQL hostname | `localhost` | Yes |
|
||||
| `POSTGRES_PORT` | PostgreSQL port | `5432` | Yes |
|
||||
| `POSTGRES_USER` | PostgreSQL username | `sientia` | Yes |
|
||||
@@ -1097,8 +1103,7 @@ laborious/
|
||||
│ ├── prediction_process.py # Core prediction workflow
|
||||
│ └── format_and_export_prediction.py # Export workflow
|
||||
├── worker/ # Worker implementation
|
||||
│ ├── worker.py # Main worker orchestrator
|
||||
│ └── prepare_worker.py # Worker factory with autoscaling config
|
||||
│ └── worker.py # Main worker orchestrator (uses sientia_do prepare_worker)
|
||||
├── utils/ # Utility functions
|
||||
│ ├── connectors_config.py # Environment-driven config builders
|
||||
│ ├── models/ # Data models
|
||||
@@ -1179,7 +1184,9 @@ export LOG_LEVEL=DEBUG
|
||||
### Scaling Considerations
|
||||
|
||||
- **Horizontal Scaling**: Deploy multiple worker instances
|
||||
- **Task Queue Distribution**: Use multiple task queues for different workflow types
|
||||
- **Task Queue Distribution**: One worker pod per `RUNTIME`; queues are
|
||||
`predictions_batch-{RUNTIME}-queue`, `minimal_retrain-{RUNTIME}-queue`,
|
||||
`drift-{RUNTIME}-queue`, `simple_metrics-{RUNTIME}-queue`
|
||||
- **Database Performance**: Optimize indexes and connection pooling
|
||||
- **MLFlow Performance**: Configure appropriate model serving resources
|
||||
|
||||
|
||||
@@ -164,6 +164,6 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
|
||||
Storage.close(self)
|
||||
MLFlow.close(self)
|
||||
Gates.close(self)
|
||||
await OPC.close(self)
|
||||
await OPC.aclose(self)
|
||||
ModelMetrics.close(self)
|
||||
API.close(self)
|
||||
|
||||
@@ -491,7 +491,7 @@ class OPC(SientiaMonitoring):
|
||||
|
||||
return data.to_dict()
|
||||
|
||||
async def close(self):
|
||||
async def aclose(self):
|
||||
"""
|
||||
Gracefully shutdown all OPC server connections and cleanup resources.
|
||||
|
||||
|
||||
@@ -202,7 +202,8 @@ class Storage(Postgres, MinioManager):
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close Storage resources (MinIO client and Postgres engine)."""
|
||||
Postgres.close(self)
|
||||
if hasattr(self, 'engine'):
|
||||
Postgres.close(self)
|
||||
MinioManager.close(self)
|
||||
|
||||
def __del__(self):
|
||||
|
||||
@@ -170,8 +170,19 @@ class MLFlowRepository(SientiaMonitoring):
|
||||
|
||||
# Sort by version number to get the latest
|
||||
latest_version = max(stage_versions, key=lambda v: int(v.version))
|
||||
run_id = latest_version.source.split('/')
|
||||
return run_id[2]
|
||||
source = latest_version.source
|
||||
if source is None:
|
||||
raise mlflow.exceptions.MlflowException(
|
||||
f"Model '{model_name}' version '{latest_version.version}' in stage '{stage}' "
|
||||
'has no source URI to resolve run ID.'
|
||||
)
|
||||
parts = source.split('/')
|
||||
if len(parts) <= 2 or not parts[2]:
|
||||
raise mlflow.exceptions.MlflowException(
|
||||
f"Model '{model_name}' version '{latest_version.version}' in stage '{stage}' "
|
||||
f"has invalid source URI '{source}' for run ID resolution."
|
||||
)
|
||||
return parts[2]
|
||||
|
||||
def get_next_run_name(self, model_name: str) -> str:
|
||||
"""
|
||||
|
||||
@@ -1,73 +0,0 @@
|
||||
import os
|
||||
import re
|
||||
from collections.abc import Sequence
|
||||
from typing import Any
|
||||
|
||||
from sientia_do.observability.logger import Logger
|
||||
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_BEHAVIOUR_MINIMUM', '10'),
|
||||
('WORKFLOW_POLLER_BEHAVIOUR_INITIAL', '100'),
|
||||
('WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM', '200'),
|
||||
('ACTIVITY_POLLER_BEHAVIOUR_MINIMUM', '10'),
|
||||
('ACTIVITY_POLLER_BEHAVIOUR_INITIAL', '100'),
|
||||
('ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM', '200'),
|
||||
]
|
||||
|
||||
|
||||
def camel_to_snake(text: str) -> str:
|
||||
"""Convert camelCase or PascalCase to snake_case."""
|
||||
text = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', text)
|
||||
text = re.sub('([a-z0-9])([A-Z])', r'\1_\2', text)
|
||||
return text.lower()
|
||||
|
||||
|
||||
def prepare_worker(
|
||||
main_workflow: type,
|
||||
other_workflows: Sequence[type],
|
||||
activities: Sequence[Any],
|
||||
temporal_client: Client,
|
||||
logger: Logger,
|
||||
) -> Worker:
|
||||
main_workflow_name = main_workflow.__name__.upper()
|
||||
|
||||
queue_name = f'{camel_to_snake(main_workflow.__name__)}-queue'
|
||||
|
||||
local_workflow_parameters = {}
|
||||
|
||||
for parameter in parameters:
|
||||
local_workflow_parameters[parameter[0]] = int(
|
||||
os.getenv(main_workflow_name + '_' + parameter[0], parameter[1])
|
||||
)
|
||||
|
||||
logger.info(f'Preparing worker for {main_workflow_name} with queue {queue_name}')
|
||||
logger.info(f'Worker runtime config: {local_workflow_parameters}')
|
||||
|
||||
return Worker(
|
||||
temporal_client,
|
||||
task_queue=queue_name,
|
||||
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_BEHAVIOUR_MINIMUM'],
|
||||
initial=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIOUR_INITIAL'],
|
||||
maximum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM'],
|
||||
),
|
||||
activity_task_poller_behavior=PollerBehaviorAutoscaling(
|
||||
minimum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIOUR_MINIMUM'],
|
||||
initial=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIOUR_INITIAL'],
|
||||
maximum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM'],
|
||||
),
|
||||
)
|
||||
@@ -5,12 +5,14 @@ This module provides the main worker implementation for the Sientia DataOps Labo
|
||||
It orchestrates Temporal workers, manages task queues, and handles the lifecycle of
|
||||
prediction and retraining workflows.
|
||||
|
||||
The worker supports multiple task queues:
|
||||
- predictions_batch-queue: Handles batch prediction workflows (heavy workload)
|
||||
Includes activities for MLFlow, data quality gates, OPC export, PI Web API export, and PostgreSQL
|
||||
- minimal_retrain-queue: Handles model retraining workflows
|
||||
- drift-queue: Handles drift detection workflows
|
||||
- simple_metrics-queue: Handles simple metrics calculation workflows
|
||||
The worker supports multiple runtime-scoped task queues (via ``sientia_do.temporal.worker.prepare_worker``):
|
||||
- predictions_batch-{runtime}-queue: Batch prediction workflows (heavy workload)
|
||||
- minimal_retrain-{runtime}-queue: Model retraining workflows
|
||||
- drift-{runtime}-queue: Drift detection workflows
|
||||
- simple_metrics-{runtime}-queue: Simple metrics workflows
|
||||
|
||||
``RUNTIME`` must be set; it is passed to every ``prepare_worker`` call. Schedulers must use the
|
||||
same queue names (breaking change vs legacy ``drift-queue`` / ``simple_metrics-queue``).
|
||||
|
||||
Key Features:
|
||||
- Resource-based scaling with WorkerTuner (CPU and memory aware)
|
||||
@@ -21,6 +23,7 @@ Key Features:
|
||||
- Multiple worker instances for different workflow types
|
||||
|
||||
Environment Variables:
|
||||
- RUNTIME: Required non-empty string; suffix for all task queue names
|
||||
- TEMPORAL_HOST: Temporal server address (default: localhost:7233)
|
||||
- TEMPORAL_NAMESPACE: Temporal namespace (default: laborious)
|
||||
- POD_ID: Kubernetes pod identifier for metrics
|
||||
@@ -40,6 +43,7 @@ with workflow.unsafe.imports_passed_through():
|
||||
from prometheus_client import start_http_server
|
||||
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
|
||||
from sientia_do.observability.logger import get_logger
|
||||
from sientia_do.temporal.worker.prepare_worker import prepare_worker
|
||||
from sientia_do.utils.connectors_config import (
|
||||
build_api_config,
|
||||
build_mongodb_config,
|
||||
@@ -53,7 +57,6 @@ with workflow.unsafe.imports_passed_through():
|
||||
build_mlflow_config,
|
||||
build_opc_config,
|
||||
)
|
||||
from laborious.worker.prepare_worker import prepare_worker
|
||||
from laborious.workflows.drift import Drift
|
||||
from laborious.workflows.minimal_retrain import MinimalRetrain
|
||||
from laborious.workflows.predictions_batch import PredictionsBatch
|
||||
@@ -100,10 +103,21 @@ async def main():
|
||||
|
||||
logger.custom_info(f'Starting Worker with POD_ID: {POD_ID}', metadata)
|
||||
|
||||
logger.custom_info('Starting prometheus client...', metadata)
|
||||
runtime = os.getenv('RUNTIME', '').strip()
|
||||
if not runtime:
|
||||
logger.custom_critical(
|
||||
'RUNTIME environment variable is required and must be non-empty',
|
||||
metadata,
|
||||
)
|
||||
metrics.APP_UP.labels(pod_id=POD_ID).set(0)
|
||||
sys.exit(1)
|
||||
|
||||
metadata_runtime = {**metadata, 'runtime': runtime}
|
||||
|
||||
logger.custom_info('Starting prometheus client...', metadata_runtime)
|
||||
start_prometheus_server()
|
||||
|
||||
logger.custom_info('Starting Notification Handler...', metadata)
|
||||
logger.custom_info('Starting Notification Handler...', metadata_runtime)
|
||||
|
||||
mongo_config = build_mongodb_config()
|
||||
notification_handler = NotificationHandler(
|
||||
@@ -113,7 +127,7 @@ async def main():
|
||||
project_name=os.getenv('PROJECT_NAME', 'laborious'),
|
||||
)
|
||||
|
||||
logger.custom_info('Starting Activities...', metadata)
|
||||
logger.custom_info('Starting Activities...', metadata_runtime)
|
||||
|
||||
activities = Activities(
|
||||
postgres_config=build_postgres_config(),
|
||||
@@ -125,10 +139,13 @@ async def main():
|
||||
notification_handler=notification_handler,
|
||||
)
|
||||
|
||||
logger.custom_info('Initializing OPC...', metadata)
|
||||
logger.custom_info('Initializing OPC...', metadata_runtime)
|
||||
await activities.init_opc()
|
||||
|
||||
logger.custom_info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata)
|
||||
logger.custom_info(
|
||||
f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...',
|
||||
metadata_runtime,
|
||||
)
|
||||
|
||||
new_runtime = Runtime(
|
||||
telemetry=TelemetryConfig(
|
||||
@@ -136,7 +153,7 @@ async def main():
|
||||
)
|
||||
)
|
||||
|
||||
logger.custom_info(f'Starting Temporal Client at {host}...', metadata)
|
||||
logger.custom_info(f'Starting Temporal Client at {host}...', metadata_runtime)
|
||||
|
||||
temporal_client = await client.Client.connect(
|
||||
target_host=host,
|
||||
@@ -144,7 +161,7 @@ async def main():
|
||||
runtime=new_runtime,
|
||||
)
|
||||
|
||||
logger.custom_info('Starting Workers...', metadata)
|
||||
logger.custom_info(f'Starting Workers (runtime={runtime})...', metadata_runtime)
|
||||
|
||||
workers = [
|
||||
prepare_worker(
|
||||
@@ -159,6 +176,7 @@ async def main():
|
||||
activities.export_data_to_postgres,
|
||||
],
|
||||
logger=logger,
|
||||
runtime=runtime,
|
||||
),
|
||||
prepare_worker(
|
||||
temporal_client=temporal_client,
|
||||
@@ -170,6 +188,7 @@ async def main():
|
||||
activities.export_data_to_postgres,
|
||||
],
|
||||
logger=logger,
|
||||
runtime=runtime,
|
||||
),
|
||||
prepare_worker(
|
||||
temporal_client=temporal_client,
|
||||
@@ -182,6 +201,7 @@ async def main():
|
||||
activities.export_data_to_postgres,
|
||||
],
|
||||
logger=logger,
|
||||
runtime=runtime,
|
||||
),
|
||||
prepare_worker(
|
||||
temporal_client=temporal_client,
|
||||
@@ -211,6 +231,7 @@ async def main():
|
||||
activities.write_pi_web_api_data,
|
||||
],
|
||||
logger=logger,
|
||||
runtime=runtime,
|
||||
),
|
||||
]
|
||||
|
||||
@@ -218,7 +239,7 @@ async def main():
|
||||
for w in workers:
|
||||
handlers.append(w.run())
|
||||
|
||||
logger.custom_info('Workers started successfully', metadata)
|
||||
logger.custom_info('Workers started successfully', metadata_runtime)
|
||||
|
||||
exit_code = 0
|
||||
try:
|
||||
|
||||
@@ -116,6 +116,12 @@ python_functions = ["test_*"]
|
||||
addopts = [
|
||||
"-v",
|
||||
"--strict-markers",
|
||||
# pytest>=9.1 has a known bug where its unraisableexception plugin crashes
|
||||
# (tracemalloc partially-initialized AttributeError) when 2+ unraisable
|
||||
# exceptions land close together — e.g. "coroutine was never awaited" from
|
||||
# AsyncMock-mocked sync methods (metrics_controller, minio_repository) being
|
||||
# GC'd. Harmless mock artifacts turned into a hard ERROR by the plugin itself.
|
||||
"-p", "no:unraisableexception",
|
||||
]
|
||||
markers = [
|
||||
"asyncio: marks tests as async",
|
||||
|
||||
@@ -1,12 +1,18 @@
|
||||
temporalio
|
||||
psycopg2-binary
|
||||
sqlalchemy
|
||||
asyncua
|
||||
asyncua==1.0.6
|
||||
redis
|
||||
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.0
|
||||
sientia_do>=1.12.2
|
||||
mlflow
|
||||
prometheus-client
|
||||
botocore
|
||||
boto3
|
||||
s3fs
|
||||
pyarrow
|
||||
mlflow
|
||||
kaleido
|
||||
hyperopt
|
||||
shap
|
||||
pycurl
|
||||
scipy<1.14.0
|
||||
scikit-learn==1.5.2
|
||||
|
||||
18
requirements-local.txt
Normal file
18
requirements-local.txt
Normal file
@@ -0,0 +1,18 @@
|
||||
temporalio
|
||||
psycopg2-binary
|
||||
sqlalchemy
|
||||
asyncua==1.0.6
|
||||
redis
|
||||
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2
|
||||
git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.10.0
|
||||
prometheus-client
|
||||
botocore
|
||||
boto3
|
||||
s3fs
|
||||
pyarrow
|
||||
kaleido
|
||||
hyperopt
|
||||
shap
|
||||
pycurl
|
||||
scipy<1.14.0
|
||||
scikit-learn==1.5.2
|
||||
@@ -1,10 +1,10 @@
|
||||
temporalio
|
||||
psycopg2-binary
|
||||
sqlalchemy
|
||||
asyncua
|
||||
asyncua==1.0.6
|
||||
redis
|
||||
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.10.4
|
||||
git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.41.0
|
||||
sientia_do>=1.12.2
|
||||
sientia>0.40.0
|
||||
prometheus-client
|
||||
botocore
|
||||
boto3
|
||||
@@ -15,4 +15,4 @@ hyperopt
|
||||
shap
|
||||
pycurl
|
||||
scipy<1.14.0
|
||||
scikit-learn==1.5.2
|
||||
scikit-learn==1.5.2
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
sonar.projectKey=Aignosi_sientia-dataops-laborious_temporal_beaec423-6c42-4f26-8134-b676287b499d
|
||||
sonar.projectKey=Aignosi_sientia-dataops-laborious_temporal_ca1a7039-6db9-49e5-be78-54d29bc93e4f
|
||||
sonar.projectName=sientia-dataops-laborious_temporal
|
||||
sonar.sources=laborious
|
||||
sonar.tests=tests
|
||||
|
||||
@@ -173,7 +173,7 @@ async def test_shutdown(
|
||||
mock_mlflow_init,
|
||||
mock_storage_init,
|
||||
):
|
||||
mock_opc_init.close = AsyncMock()
|
||||
mock_opc_init.aclose = AsyncMock()
|
||||
postgres_config = {
|
||||
'host': 'localhost',
|
||||
'port': 5432,
|
||||
@@ -221,7 +221,7 @@ async def test_shutdown(
|
||||
)
|
||||
|
||||
await activities.shutdown()
|
||||
mock_opc_init.close.assert_called_once()
|
||||
mock_opc_init.aclose.assert_called_once()
|
||||
mock_storage_init.close.assert_called_once()
|
||||
mock_mlflow_init.close.assert_called_once()
|
||||
mock_gates_init.close.assert_called_once()
|
||||
|
||||
@@ -734,5 +734,5 @@ async def test_validate_server(opc):
|
||||
@mark.asyncio
|
||||
async def test_close(opc):
|
||||
opc.opc_repository['server1'].disconnect = AsyncMock(return_value=True)
|
||||
await opc.close()
|
||||
await opc.aclose()
|
||||
opc.opc_repository['server1'].disconnect.assert_called_once()
|
||||
|
||||
@@ -142,6 +142,41 @@ def test_get_model_run_id_success(mlflow_repository):
|
||||
assert output == '1'
|
||||
|
||||
|
||||
def test_get_model_run_id_missing_source(mlflow_repository):
|
||||
mlflow_repository.client.search_registered_models.return_value = [MagicMock(name='test')]
|
||||
|
||||
mlflow_repository.client.search_model_versions.return_value = [
|
||||
MagicMock(current_stage='Production', version='1', source='runs/test/0'),
|
||||
MagicMock(current_stage='Production', version='2', source=None),
|
||||
]
|
||||
|
||||
with pytest.raises(mlflow_lib.exceptions.MlflowException) as exc_info:
|
||||
mlflow_repository.get_model_run_id('test')
|
||||
|
||||
assert (
|
||||
str(exc_info.value)
|
||||
== "Model 'test' version '2' in stage 'Production' has no source URI to resolve run ID."
|
||||
)
|
||||
|
||||
|
||||
def test_get_model_run_id_invalid_source(mlflow_repository):
|
||||
mlflow_repository.client.search_registered_models.return_value = [MagicMock(name='test')]
|
||||
|
||||
mlflow_repository.client.search_model_versions.return_value = [
|
||||
MagicMock(current_stage='Production', version='1', source='runs/test/0'),
|
||||
MagicMock(current_stage='Production', version='2', source='runs/test'),
|
||||
]
|
||||
|
||||
with pytest.raises(mlflow_lib.exceptions.MlflowException) as exc_info:
|
||||
mlflow_repository.get_model_run_id('test')
|
||||
|
||||
assert (
|
||||
str(exc_info.value)
|
||||
== "Model 'test' version '2' in stage 'Production' has invalid source URI "
|
||||
"'runs/test' for run ID resolution."
|
||||
)
|
||||
|
||||
|
||||
def test_get_next_run_name(mlflow, mlflow_repository):
|
||||
mlflow.search_runs.return_value = [1, 2, 3]
|
||||
output = mlflow_repository.get_next_run_name('run')
|
||||
|
||||
14
tests/laborious/worker/test_runtime_task_queues.py
Normal file
14
tests/laborious/worker/test_runtime_task_queues.py
Normal file
@@ -0,0 +1,14 @@
|
||||
from sientia_do.temporal.worker.prepare_worker import build_queue_name
|
||||
|
||||
from laborious.workflows.drift import Drift
|
||||
from laborious.workflows.minimal_retrain import MinimalRetrain
|
||||
from laborious.workflows.predictions_batch import PredictionsBatch
|
||||
from laborious.workflows.simple_metrics import SimpleMetrics
|
||||
|
||||
|
||||
def test_runtime_scoped_queue_names():
|
||||
runtime = 'prod-a'
|
||||
assert build_queue_name(PredictionsBatch.__name__, runtime) == 'predictions_batch-prod-a-queue'
|
||||
assert build_queue_name(MinimalRetrain.__name__, runtime) == 'minimal_retrain-prod-a-queue'
|
||||
assert build_queue_name(Drift.__name__, runtime) == 'drift-prod-a-queue'
|
||||
assert build_queue_name(SimpleMetrics.__name__, runtime) == 'simple_metrics-prod-a-queue'
|
||||
315
values.yaml
315
values.yaml
@@ -1,315 +0,0 @@
|
||||
# Default values for sientia-module.
|
||||
# This is a YAML-formatted file.
|
||||
# Declare variables to be passed into your templates.
|
||||
|
||||
# This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/
|
||||
replicaCount: 1
|
||||
|
||||
# This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
|
||||
image:
|
||||
repository: aignosi.azurecr.io/sientia-module
|
||||
# This sets the pull policy for images.
|
||||
pullPolicy: Always
|
||||
# Overrides the image tag whose default is the chart appVersion.
|
||||
tag: "1.1.2"
|
||||
|
||||
# This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
|
||||
imagePullSecrets:
|
||||
- name: docker-hub-secret
|
||||
# This is to override the chart name.
|
||||
nameOverride: "sientia-laborious-legacy-worker"
|
||||
fullnameOverride: "sientia-laborious-legacy-worker"
|
||||
namespace: sientia
|
||||
|
||||
# This section builds out the service account more information can be found here: https://kubernetes.io/docs/concepts/security/service-accounts/
|
||||
serviceAccount:
|
||||
# Specifies whether a service account should be created
|
||||
create: true
|
||||
# Automatically mount a ServiceAccount's API credentials?
|
||||
automount: true
|
||||
# Annotations to add to the service account
|
||||
annotations: {}
|
||||
# The name of the service account to use.
|
||||
# If not set and create is true, a name is generated using the fullname template
|
||||
name: "sientia-laborious-legacy-worker"
|
||||
|
||||
# This is for setting Kubernetes Annotations to a Pod.
|
||||
# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
|
||||
podAnnotations: {}
|
||||
# This is for setting Kubernetes Labels to a Pod.
|
||||
# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
|
||||
podLabels: {}
|
||||
|
||||
podSecurityContext: {}
|
||||
# fsGroup: 2000
|
||||
|
||||
securityContext: {}
|
||||
# capabilities:
|
||||
# drop:
|
||||
# - ALL
|
||||
# readOnlyRootFilesystem: true
|
||||
# runAsNonRoot: true
|
||||
# runAsUser: 1000
|
||||
|
||||
|
||||
resources:
|
||||
# Resource limits and requests are important for ResourceBasedTuner to work correctly.
|
||||
# The tuner monitors system CPU and memory usage, so proper resource limits must be set.
|
||||
limits:
|
||||
cpu: 2000m # 2 CPU cores
|
||||
memory: 20Gi # 20 GB memory
|
||||
requests:
|
||||
cpu: 1000m # 1 CPU core
|
||||
memory: 2Gi # 2 GB memory
|
||||
|
||||
# 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/
|
||||
# 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
|
||||
- |
|
||||
curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1'
|
||||
initialDelaySeconds: 1260
|
||||
periodSeconds: 15
|
||||
timeoutSeconds: 5
|
||||
failureThreshold: 3
|
||||
|
||||
readinessProbe:
|
||||
exec:
|
||||
command:
|
||||
- sh
|
||||
- -c
|
||||
- |
|
||||
curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1'
|
||||
initialDelaySeconds: 1200
|
||||
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/
|
||||
autoscaling:
|
||||
enabled: false
|
||||
minReplicas: 1
|
||||
maxReplicas: 100
|
||||
targetCPUUtilizationPercentage: 80
|
||||
# targetMemoryUtilizationPercentage: 80
|
||||
|
||||
# Additional volumes on the output Deployment definition.
|
||||
volumes: []
|
||||
# - name: foo
|
||||
# secret:
|
||||
# secretName: mysecret
|
||||
# optional: false
|
||||
|
||||
# Additional volumeMounts on the output Deployment definition.
|
||||
volumeMounts: []
|
||||
# - name: foo
|
||||
# mountPath: "/etc/foo"
|
||||
# readOnly: true
|
||||
|
||||
nodeSelector: {}
|
||||
|
||||
tolerations: []
|
||||
|
||||
affinity: {}
|
||||
|
||||
services:
|
||||
sdk-metrics:
|
||||
enabled: true
|
||||
type: ClusterIP
|
||||
port: 9091
|
||||
targetPort: 9091
|
||||
name: sdk-metrics
|
||||
|
||||
metrics:
|
||||
enabled: true
|
||||
type: ClusterIP
|
||||
port: 9090
|
||||
targetPort: 9090
|
||||
name: metrics
|
||||
|
||||
# Configuração do ServiceMonitor para o Prometheus Operator
|
||||
# ref: https://github.com/prometheus-operator/prometheus-operator
|
||||
serviceMonitor:
|
||||
# Se true, um recurso ServiceMonitor será criado.
|
||||
enabled: true
|
||||
# O intervalo no qual as métricas devem ser coletadas (ex: 30s, 1m).
|
||||
endpoints:
|
||||
- port: metrics
|
||||
path: /metrics
|
||||
interval: 30s
|
||||
relabelings: []
|
||||
- port: sdk-metrics
|
||||
path: /metrics
|
||||
interval: 30s
|
||||
relabelings: []
|
||||
|
||||
additionalLabels:
|
||||
release: kube-prometheus-stack
|
||||
|
||||
|
||||
env:
|
||||
# Entrypoint variables
|
||||
- name: GITHUB_REPO_URL
|
||||
value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git"
|
||||
- name: GITHUB_BRANCH
|
||||
value: "fix/SIENTIAPDE-1811"
|
||||
- name: PYTHON_APP
|
||||
value: "laborious.worker.worker"
|
||||
|
||||
# Application variables
|
||||
- name: POSTGRES_HOST
|
||||
value: "paradedb-rw.paradedb.svc.cluster.local"
|
||||
- name: POSTGRES_PORT
|
||||
value: "5432"
|
||||
- name: POSTGRES_USER
|
||||
value: "postgres"
|
||||
- name: POSTGRES_PASSWORD
|
||||
value: "nFqc81y6kwmr2zuAIx43DhiOosFCVPpeEfTtTWZflkNjB2j1KtEeIANkhFR9mAX3"
|
||||
- name: POSTGRES_DBNAME
|
||||
value: "sientia"
|
||||
- name: POSTGRES_MIN_CONNECTIONS
|
||||
value: "20"
|
||||
# max_connections = number_of_workers * max_concurrent_activities * safety_factor
|
||||
# Example: 4 workers * 50 activities * 0.5 = 100 connections
|
||||
- name: POSTGRES_MAX_CONNECTIONS
|
||||
value: "100"
|
||||
|
||||
- name: MLFLOW_HOST
|
||||
value: "http://sientia-tracker-mlflow-tracking.sientia-tracker.svc.cluster.local"
|
||||
- name: MLFLOW_PORT
|
||||
value: "80"
|
||||
- name: MLFLOW_USERNAME
|
||||
value: "aignosi"
|
||||
- name: MLFLOW_PASSWORD
|
||||
value: "1L0FP50j3ncp123"
|
||||
|
||||
- name: OPC_ID
|
||||
value: "1"
|
||||
- name: OPC_SERVER_NAME
|
||||
value: "default_server"
|
||||
- name: OPC_URL
|
||||
value: "opc.tcp://sientia-opc-simulator-opc.sientia.svc.cluster.local:4840"
|
||||
|
||||
|
||||
- name: LOG_LEVEL
|
||||
value: "DEBUG"
|
||||
- name: HTTP_METRICS_PORT
|
||||
value: "9090"
|
||||
- name: HTTP_SDK_METRICS_PORT
|
||||
value: "9091"
|
||||
- name: PROJECT_NAME
|
||||
value: "sientia-laborious"
|
||||
|
||||
- name: TEMPORAL_HOST
|
||||
value: "temporal-frontend.temporal.svc.cluster.local:7233"
|
||||
- name: TEMPORAL_NAMESPACE
|
||||
value: "laborious"
|
||||
|
||||
- name: MONGODB_USERNAME
|
||||
value: "root"
|
||||
- name: MONGODB_PASSWORD
|
||||
value: "wKZDbMNU1c"
|
||||
- name: MONGODB_URL
|
||||
value: "my-release-mongodb.mongodb.svc.cluster.local:27017"
|
||||
- name: MONGODB_DATABASE
|
||||
value: "sientia"
|
||||
- name: MONGODB_TTL_INDEX_HOURS
|
||||
value: "1"
|
||||
|
||||
- name: MINIO_ENDPOINT_URL
|
||||
value: "minio.minio.svc.cluster.local:9000"
|
||||
- name: MINIO_ACCESS_KEY
|
||||
value: "admin"
|
||||
- name: MINIO_SECRET_KEY
|
||||
value: "LiArt4eNmJ"
|
||||
- name: MINIO_DEFAULT_BUCKET
|
||||
value: "sientia"
|
||||
- name: MINIO_RETENTION_HOURS
|
||||
value: "24"
|
||||
- name: SIENTIA_MINIO_OFFLOAD_THRESHOLD_MEGABYTES
|
||||
value: "0.5"
|
||||
|
||||
# Temporal worker tuning for PredictionsBatch.
|
||||
# IMPORTANT: prefix must be PREDICTIONSBATCH_ (from class name PredictionsBatch).
|
||||
# Keep workflow-task concurrency moderate to reduce task completion races under load.
|
||||
- name: PREDICTIONSBATCH_MAX_CONCURRENT_WORKFLOW_TASKS
|
||||
value: "20"
|
||||
# Allow higher activity parallelism because most activities are I/O-bound, but keep headroom.
|
||||
- name: PREDICTIONSBATCH_MAX_CONCURRENT_ACTIVITIES
|
||||
value: "60"
|
||||
# Keep local activities controlled so they do not monopolize the event loop.
|
||||
- name: PREDICTIONSBATCH_MAX_CONCURRENT_LOCAL_ACTIVITIES
|
||||
value: "20"
|
||||
# Cache enough workflows for reuse without excessive memory growth.
|
||||
- name: PREDICTIONSBATCH_MAX_CACHED_WORKFLOWS
|
||||
value: "200"
|
||||
# Start with one workflow poller to avoid burst contention at startup.
|
||||
- name: PREDICTIONSBATCH_WORKFLOW_POLLER_BEHAVIOUR_MINIMUM
|
||||
value: "3"
|
||||
# Small initial poller count warms up gradually instead of spiking task fetches.
|
||||
- name: PREDICTIONSBATCH_WORKFLOW_POLLER_BEHAVIOUR_INITIAL
|
||||
value: "5"
|
||||
# Cap workflow pollers to limit scheduling pressure and avoid over-polling.
|
||||
- name: PREDICTIONSBATCH_WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM
|
||||
value: "15"
|
||||
# Keep at least two activity pollers so activity queues do not starve during spikes.
|
||||
- name: PREDICTIONSBATCH_ACTIVITY_POLLER_BEHAVIOUR_MINIMUM
|
||||
value: "3"
|
||||
# Moderate initial activity pollers for faster ramp-up with controlled pressure.
|
||||
- name: PREDICTIONSBATCH_ACTIVITY_POLLER_BEHAVIOUR_INITIAL
|
||||
value: "10"
|
||||
# Limit max activity pollers to preserve CPU for workflow-task completion.
|
||||
- name: PREDICTIONSBATCH_ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM
|
||||
value: "30"
|
||||
|
||||
- name: MINIMALRETRAIN_MAX_CONCURRENT_ACTIVITIES
|
||||
value: "1"
|
||||
- name: MINIMALRETRAIN_MAX_CONCURRENT_LOCAL_ACTIVITIES
|
||||
value: "1"
|
||||
- name: MINIMALRETRAIN_MAX_CACHED_WORKFLOWS
|
||||
value: "1"
|
||||
- name: MINIMALRETRAIN_WORKFLOW_POLLER_BEHAVIOUR_MINIMUM
|
||||
value: "1"
|
||||
- name: MINIMALRETRAIN_WORKFLOW_POLLER_BEHAVIOUR_INITIAL
|
||||
value: "1"
|
||||
- name: MINIMALRETRAIN_WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM
|
||||
value: "1"
|
||||
- name: MINIMALRETRAIN_ACTIVITY_POLLER_BEHAVIOUR_MINIMUM
|
||||
value: "1"
|
||||
- name: MINIMALRETRAIN_ACTIVITY_POLLER_BEHAVIOUR_INITIAL
|
||||
value: "1"
|
||||
- name: MINIMALRETRAIN_ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM
|
||||
value: "1"
|
||||
|
||||
- name: PI_WEB_API_BASE_URL
|
||||
value: "https://pivision.votorantimcimentos.com/piwebapi"
|
||||
- name: PI_WEB_API_AUTH_TYPE
|
||||
value: "basic"
|
||||
- name: PI_WEB_API_AUTH_TOKEN
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: pi-web-api-auth-token
|
||||
key: token
|
||||
|
||||
- name: PYPI_SERVER
|
||||
value: "http://library-distribution-server.library.svc.cluster.local:5000"
|
||||
|
||||
ssh:
|
||||
enabled: true
|
||||
secretName: git-ssh-key-sientia-laborious-worker
|
||||
sshPath: /mnt/.ssh
|
||||
knownHostsPath: /mnt/known_hosts
|
||||
|
||||
# kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp
|
||||
|
||||
# helm upgrade --install sientia-laborious-legacy-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.6.0
|
||||
|
||||
# kubectl create secret generic git-ssh-key-sientia-laborious-worker \
|
||||
# --namespace sientia \
|
||||
# --from-file=ssh-privatekey=git_key \
|
||||
# --type=kubernetes.io/ssh-auth
|
||||
Reference in New Issue
Block a user