17 Commits

Author SHA1 Message Date
Eduardo Rios
695e4c07a6 SIENTIAPDE-2072: bump sientia_do pin to 1.12.2
Picks up the notification timestamp -> native datetime fix.
2026-08-17 16:42:31 -03:00
Bruno Domingues
6a8c41b328 fix(pytest): Workaround unraisableexception plugin crash
Disables the unraisableexception plugin in pytest due to a known bug in
pytest>=9.1 where it crashes with tracemalloc errors when multiple
unraisable exceptions occur close together.
2026-08-04 15:21:14 -03:00
Bruno Domingues
773c980fc3 fix(storage): Prevent AttributeError when closing uninitialized Postgres engine 2026-08-04 14:57:52 -03:00
Bruno Domingues
35efb67c89 fix(test_activities): Use aclose for async mock shutdown 2026-08-04 14:43:15 -03:00
Bruno Domingues
acd925cf2a refactor(opc): Rename async close method to aclose
Renamed the OPC.close asynchronous method to OPC.aclose to align with common Python conventions for asynchronous context managers and methods, improving clarity. All call sites and tests have been updated accordingly.
2026-08-04 12:01:06 -03:00
Bruno Domingues
310ceea0d8 ci(quality-gate): Configure push triggers, concurrency, and granular permissions 2026-08-04 11:41:23 -03:00
Bruno Domingues
2d73ec9ec2 Merge pull request #42 from Aignosi/feature/SIENTIAPDE-1945
SIENTIAPDE-1945: Remove values.yaml from sientia-module Helm chart
2026-07-10 11:49:36 -03:00
Bruno Domingues
2142143ab9 SIENTIAPDE-1945: Delete values.yaml configuration file for sientia-module Helm chart. 2026-07-08 22:12:50 -03:00
Bruno Domingues
c07457bfbe chore(sonar): update project key 2026-07-01 20:59:29 -03:00
vitor-aignosi
086b12492e Merge pull request #41 from Aignosi/feature/SIENTIAPDE-1646-legacy-laborious-worker
SIENTIAPDE-1646: Refactor Worker Task Queue Management and Update Dependencies
2026-05-21 15:30:09 -03:00
vitor-aignosi
4ea0754f0c SIENTIAPDE-1646
Enhance MLFlow run ID resolution with error handling for missing and invalid source URIs

- Added checks in `get_model_run_id` method to raise exceptions for models with missing or invalid source URIs.
- Introduced new test cases to validate error handling for these scenarios.
- Updated `requirements-light.txt` to include `mlflow` as a dependency.
2026-05-20 09:53:53 -03:00
vitor-aignosi
ddb1618209 SIENTIAPDE-1646
Update quality-gate workflow to use python-quality-gate template
2026-05-20 09:30:05 -03:00
vitor-aignosi
90f8bdda61 SIENTIAPDE-1646
Update requirements.txt to align with recent dependency changes and ensure compatibility across the project.
2026-05-20 09:26:35 -03:00
vitor-aignosi
2ccda3e440 SIENTIAPDE-1646
Refactor worker task queue management and update README

- Introduced runtime-scoped task queues for workflows, replacing legacy queue names.
- Updated worker implementation to utilize `sientia_do.temporal.worker.prepare_worker`.
- Added `RUNTIME` environment variable to configure task queue suffixes.
- Enhanced README documentation to reflect changes in task queue structure and worker setup.
2026-05-19 17:07:18 -03:00
vitor-aignosi
d856150e24 Update requirements.txt 2026-05-19 14:45:26 -03:00
vitor-aignosi
6569810756 Update requirements.txt 2026-05-19 14:41:58 -03:00
vitor-aignosi
7153f1da0d SIENTIAPDE-1646
Remove requirements-light.txt and update requirements.txt to specify versions for asyncua and new sientia dependencies.
2026-05-19 14:13:23 -03:00
18 changed files with 170 additions and 425 deletions

View File

@@ -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'

View File

@@ -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

View File

@@ -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)

View File

@@ -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.

View File

@@ -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):

View File

@@ -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:
"""

View File

@@ -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'],
),
)

View File

@@ -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:

View File

@@ -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",

View File

@@ -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
View 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

View File

@@ -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

View File

@@ -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

View File

@@ -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()

View File

@@ -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()

View File

@@ -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')

View 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'

View File

@@ -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