diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml new file mode 100644 index 0000000..c1073c9 --- /dev/null +++ b/.github/workflows/quality-gate.yml @@ -0,0 +1,109 @@ +name: Quality gate + +on: + push: + branches: + - '**' + pull_request: + branches: + - '**' + types: [ opened, synchronize, reopened ] + +jobs: + sonar: + name: SonarQube Analysis + runs-on: ubuntu-latest + permissions: + contents: read + + steps: + - name: ⬇️ Checkout Code + uses: actions/checkout@v4 + with: + fetch-depth: 0 + persist-credentials: false + + - name: Generate App Token + id: generate-app-token + uses: actions/create-github-app-token@v1 + with: + app-id: ${{ secrets.APP_ID }} + private-key: ${{ secrets.APP_PRIVATE_KEY }} + owner: 'Aignosi' + repositories: 'sientia-dataops-library' + + - name: Prepare requirements.txt + id: prepare-requirements + run: | + sed -e "s|git+ssh://git@github.com/|git+https://github.com/|g" \ + -e "s|git@github.com:|git+https://github.com/|g" \ + requirements.txt > requirements_prepared.txt + echo "PROCESSED_REQUIREMENTS_FILE=requirements_prepared.txt" >> $GITHUB_OUTPUT + + - name: Configure Git to use App Token + env: + GH_APP_TOKEN: ${{ steps.generate-app-token.outputs.token }} + run: | + git config --global url."https://oauth2:${GH_APP_TOKEN}@github.com/".insteadOf "https://github.com/" + + - name: 🔧 Setup Python + uses: actions/setup-python@v4 + with: + python-version: "3.11" + + - name: 🗄️ Cache Python dependencies + uses: actions/cache@v3 + with: + path: ~/.cache/pip + key: ${{ runner.os }}-pip-${{ hashFiles(steps.prepare-requirements.outputs.PROCESSED_REQUIREMENTS_FILE) }} + restore-keys: | + ${{ runner.os }}-pip- + + - name: 📦 Install Dependencies + run: | + python -m pip install --upgrade pip + pip install -r ${{ steps.prepare-requirements.outputs.PROCESSED_REQUIREMENTS_FILE }} + pip install pytest pytest-cov pytest-asyncio + + - name: ⬇️ Setup Node.js 18 + uses: actions/setup-node@v4 + with: + node-version: 18 + + - name: 📥 Setup SonarScanner + uses: warchant/setup-sonar-scanner@v7 + + - name: 🧪 Run Tests with Pytest + run: | + set +e + pytest tests --junitxml=pytest.xml --cov=orchestrator --cov-report=xml --cov-report=term + PYTEST_EXIT_CODE=$? + set -e + + if [ $PYTEST_EXIT_CODE -eq 0 ]; then + echo "Pytest executado com sucesso." + elif [ $PYTEST_EXIT_CODE -eq 5 ]; then + echo "Pytest finalizado com código 5 (Nenhum teste encontrado). Tratando como sucesso para este workflow." + exit 0 + else + echo "Pytest falhou com código de saída $PYTEST_EXIT_CODE." + exit $PYTEST_EXIT_CODE + fi + + - name: 📊 Run SonarQube Analysis + env: + SONAR_PROJECT_KEY: ${{ secrets.SONAR_PROJECT_KEY }} + SONAR_HOST_URL: ${{ secrets.SONAR_HOST_URL }} + SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }} # Este é o token do SonarQube + run: | + sonar-scanner \ + -Dsonar.projectKey=$SONAR_PROJECT_KEY \ + -Dsonar.sources=orchestrator \ + -Dsonar.tests=tests \ + -Dsonar.python.coverage.reportPaths=coverage.xml \ + -Dsonar.python.xunit.reportPath=pytest.xml \ + -Dsonar.host.url=$SONAR_HOST_URL \ + -Dsonar.token=$SONAR_TOKEN \ + -Dsonar.python.version=3.11 \ + -Dsonar.projectVersion=1.0.0 \ + -Dsonar.coverage.exclusions=orchestrator/worker/worker.py diff --git a/.gitignore b/.gitignore index 0a19790..fd3c51f 100644 --- a/.gitignore +++ b/.gitignore @@ -1,174 +1,44 @@ -# Byte-compiled / optimized / DLL files +# Ignorar volumes do Docker +docker-compose.override.yml +**/db_data/ +**/kafka-volume/ +**/zookeeper-volume/ +**/mage_data/ +**/minio_data/ +**/venv/ +**/certs/*.pem +**/certs/*.der +**/certs/*.csr +**/deploy/*.yaml +scouter/.file_versions/ +scouter/pipelines/**/triggers.yaml +**/postgres_data/** +**/couchbase_data/** +**/redis_data/** +# Ignorar arquivos e diretórios de cache do Python __pycache__/ -*.py[cod] -*$py.class +*.pyc +*.pyo +*.pyd -# C extensions -*.so - -# Distribution / packaging -.Python -build/ -develop-eggs/ -dist/ -downloads/ -eggs/ -.eggs/ -lib/ -lib64/ -parts/ -sdist/ -var/ -wheels/ -share/python-wheels/ -*.egg-info/ -.installed.cfg -*.egg -MANIFEST - -# PyInstaller -# Usually these files are written by a python script from a template -# before PyInstaller builds the exe, so as to inject date/other infos into it. -*.manifest -*.spec - -# Installer logs -pip-log.txt -pip-delete-this-directory.txt - -# Unit test / coverage reports -htmlcov/ -.tox/ -.nox/ -.coverage -.coverage.* -.cache -nosetests.xml -coverage.xml -*.cover -*.py,cover -.hypothesis/ -.pytest_cache/ -cover/ - -# Translations -*.mo -*.pot - -# Django stuff: +# Ignorar logs *.log -local_settings.py -db.sqlite3 -db.sqlite3-journal -# Flask stuff: -instance/ -.webassets-cache +# Ignorar arquivos de configuração locais +.vscode/ +.pytest_cache/ +.idea/ +*.swp -# Scrapy stuff: -.scrapy +# Ignorar arquivos temporários +*.tmp +*.bak +*.old +.secret -# Sphinx documentation -docs/_build/ +# Ignorar coverage +htmlcov/ +.coverage -# PyBuilder -.pybuilder/ -target/ - -# Jupyter Notebook -.ipynb_checkpoints - -# IPython -profile_default/ -ipython_config.py - -# pyenv -# For a library or package, you might want to ignore these files since the code is -# intended to run in multiple environments; otherwise, check them in: -# .python-version - -# pipenv -# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. -# However, in case of collaboration, if having platform-specific dependencies or dependencies -# having no cross-platform support, pipenv may install dependencies that don't work, or not -# install all needed dependencies. -#Pipfile.lock - -# UV -# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -#uv.lock - -# poetry -# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control -#poetry.lock - -# pdm -# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. -#pdm.lock -# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it -# in version control. -# https://pdm.fming.dev/latest/usage/project/#working-with-version-control -.pdm.toml -.pdm-python -.pdm-build/ - -# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm -__pypackages__/ - -# Celery stuff -celerybeat-schedule -celerybeat.pid - -# SageMath parsed files -*.sage.py - -# Environments -.env -.venv -env/ -venv/ -ENV/ -env.bak/ -venv.bak/ - -# Spyder project settings -.spyderproject -.spyproject - -# Rope project settings -.ropeproject - -# mkdocs documentation -/site - -# mypy -.mypy_cache/ -.dmypy.json -dmypy.json - -# Pyre type checker -.pyre/ - -# pytype static type analyzer -.pytype/ - -# Cython debug symbols -cython_debug/ - -# PyCharm -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ - -# Ruff stuff: -.ruff_cache/ - -# PyPI configuration file -.pypirc +# git keys +git_key* \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..95984aa --- /dev/null +++ b/Dockerfile @@ -0,0 +1,83 @@ +FROM python:3.11-bookworm +LABEL description="Deploy Mage on ECS" +ARG FEATURE_BRANCH +USER root +SHELL ["/bin/bash", "-o", "pipefail", "-c"] + +# Definir Python 3.11 como padrão +ENV PATH="/usr/local/bin/python3.11:$PATH" +RUN update-alternatives --install /usr/bin/python python /usr/local/bin/python3.11 1 && \ + update-alternatives --install /usr/bin/python3 python3 /usr/local/bin/python3.11 1 && \ + update-alternatives --config python3 <<< '1' && \ + update-alternatives --config python <<< '1' + +## System Packages +RUN \ + curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - && \ + curl https://packages.microsoft.com/config/debian/11/prod.list > /etc/apt/sources.list.d/mssql-release.list && \ + apt-get -y update && \ + ACCEPT_EULA=Y apt-get -y install --no-install-recommends \ + # NFS dependencies + nfs-common \ + # odbc dependencies + msodbcsql18 \ + unixodbc-dev \ + graphviz \ + # postgres dependencies + postgresql-client \ + # R + r-base && \ + apt-get clean && \ + rm -rf /var/lib/apt/lists/* + +## R Packages +RUN \ + R -e "install.packages('pacman', repos='http://cran.us.r-project.org')" && \ + R -e "install.packages('renv', repos='http://cran.us.r-project.org')" + +## Python Packages +RUN \ + pip3 install --no-cache-dir sparkmagic && \ + mkdir ~/.sparkmagic && \ + curl https://raw.githubusercontent.com/jupyter-incubator/sparkmagic/master/sparkmagic/example_config.json > ~/.sparkmagic/config.json && \ + sed -i 's/localhost:8998/host.docker.internal:9999/g' ~/.sparkmagic/config.json && \ + jupyter-kernelspec install --user "$(pip3 show sparkmagic | grep Location | cut -d' ' -f2)/sparkmagic/kernels/pysparkkernel" + +# Mage integrations and other related packages +RUN \ + pip3 install --no-cache-dir "git+https://github.com/wbond/oscrypto.git@d5f3437ed24257895ae1edd9e503cfb352e635a8" && \ + pip3 install --no-cache-dir "git+https://github.com/dremio-hub/arrow-flight-client-examples.git#egg=dremio-flight&subdirectory=python/dremio-flight" && \ + pip3 install --no-cache-dir "git+https://github.com/mage-ai/singer-python.git#egg=singer-python" && \ + pip3 install --no-cache-dir "git+https://github.com/mage-ai/dbt-mysql.git#egg=dbt-mysql" && \ + pip3 install --no-cache-dir "git+https://github.com/mage-ai/sqlglot#egg=sqlglot" && \ + pip3 install --no-cache-dir faster-fifo && \ + if [ -z "$FEATURE_BRANCH" ] || [ "$FEATURE_BRANCH" = "null" ]; then \ + pip3 install --no-cache-dir "git+https://github.com/mage-ai/mage-ai.git#egg=mage-integrations&subdirectory=mage_integrations"; \ + else \ + pip3 install --no-cache-dir "git+https://github.com/mage-ai/mage-ai.git@$FEATURE_BRANCH#egg=mage-integrations&subdirectory=mage_integrations"; \ + fi + +# Mage +COPY ./mage_ai/server/constants.py /tmp/constants.py +RUN if [ -z "$FEATURE_BRANCH" ] || [ "$FEATURE_BRANCH" = "null" ] ; then \ + tag=$(tail -n 1 /tmp/constants.py) && \ + VERSION=$(echo "$tag" | tr -d "'") && \ + pip3 install --no-cache-dir "mage-ai[all]==$VERSION"; \ + else \ + pip3 install --no-cache-dir "git+https://github.com/mage-ai/mage-ai.git@$FEATURE_BRANCH#egg=mage-ai[all]"; \ + fi + +## Startup Script +COPY --chmod=0755 ./scripts/install_other_dependencies.py ./scripts/run_app.sh /app/ +ENV MAGE_DATA_DIR="/home/src/mage_data" +ENV PYTHONPATH="${PYTHONPATH}:/home/src" +WORKDIR /home/src +EXPOSE 6789 +EXPOSE 7789 + +# Copia o arquivo requirements.txt para o contêiner +COPY requirements.txt /app/requirements.txt +RUN pip3 install --no-cache-dir -r /app/requirements.txt + + +CMD ["/bin/sh", "-c", "/app/run_app.sh"] \ No newline at end of file diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..66aae81 --- /dev/null +++ b/Makefile @@ -0,0 +1,7 @@ +VERSION = 1.0.8 +name = sientia-laborious +# ENVIRONMENT = production + +docker-hub: + @docker build --no-cache -t aignosi.azurecr.io/$(name):$(VERSION) . + @docker push aignosi.azurecr.io/$(name):$(VERSION) \ No newline at end of file diff --git a/README.md b/README.md index a183764..e69de29 100644 --- a/README.md +++ b/README.md @@ -1,2 +0,0 @@ -# sientia-dataops-orchestrator_temporal -Orchestrator for SIENTIA at Temporal frameworker diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..baab896 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,133 @@ +version: '3.8' + +services: + postgres: + image: postgres:15 + container_name: postgres + environment: + POSTGRES_USER: sientia + POSTGRES_PASSWORD: sientia + POSTGRES_DB: sientia + ports: + - "5432:5432" + volumes: + - ./postgres_data:/var/lib/postgresql/data + networks: + - sientia-network + + + + couchbase: + image: couchbase/server:7.2.0 + container_name: couchbase + ports: + - "8091:8091" # Admin UI and REST API + - "8092:8092" # Query Service (N1QL) + - "8093:8093" # Index Service + - "8094:8094" # Search Service + - "11210:11210" # Data Service (KV) + - "18091:18091" # Analytics Service (if enabled) + environment: + CB_CLUSTER_USERNAME: sientia + CB_CLUSTER_PASSWORD: sientia + CB_CLUSTER_RAMSIZE: 256 + CB_CLUSTER_INDEX_RAMSIZE: 256 + volumes: + - ./couchbase_data:/opt/couchbase/var + networks: + - sientia-network + healthcheck: + test: ["CMD-SHELL", "curl -f http://localhost:8091/pools/default || exit 1"] + interval: 10s + timeout: 10s + retries: 5 + + redis: + image: redis:7-alpine # Using a lightweight Redis image + container_name: redis + ports: + - "6379:6379" + volumes: + - ./redis_data:/data # Persist Redis data + networks: + - sientia-network + + redis-commander: + image: rediscommander/redis-commander:latest + container_name: redis-commander + environment: + REDIS_HOSTS: local:redis:6379 # Connects to the 'redis' service within the Docker network + ports: + - "8081:8081" # Access the Redis Commander UI on this port + depends_on: + - redis # Ensures Redis starts before Redis Commander + networks: + - sientia-network + + kafka: + image: bitnami/kafka:3.7 # Using a specific Kafka version for stability + container_name: kafka + ports: + - "9092:9092" # For clients connecting from the host machine or outside Docker network + environment: + # KRaft (Kafka Raft without Zookeeper) settings + KAFKA_CFG_NODE_ID: '0' + KAFKA_CFG_PROCESS_ROLES: 'broker,controller' + KAFKA_CFG_CONTROLLER_LISTENER_NAMES: 'CONTROLLER' + # Listeners: ://: + # PLAINTEXT_EXTERNAL for host access, INTERNAL for container-to-container communication + KAFKA_CFG_LISTENERS: 'PLAINTEXT_EXTERNAL://0.0.0.0:9092,INTERNAL://0.0.0.0:19092,CONTROLLER://0.0.0.0:9093' + # Advertised Listeners: How clients (including Kafka-UI) will connect + KAFKA_CFG_ADVERTISED_LISTENERS: 'PLAINTEXT_EXTERNAL://localhost:9092,INTERNAL://kafka:19092' + KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT_EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT' + KAFKA_CFG_INTER_BROKER_LISTENER_NAME: 'INTERNAL' + KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: '0@kafka:9093' # Node 0 is at kafka:9093 for controller comms + + # Single node cluster settings (important for KRaft single node) + KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: '1' + KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: '1' + KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR: '1' + KAFKA_CFG_DEFAULT_REPLICATION_FACTOR: '1' # For auto-created topics + KAFKA_CFG_NUM_PARTITIONS: '1' # Default partitions for auto-created topics + + KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: 'true' # Convenient for development + volumes: + - kafka_data:/bitnami/kafka # Bitnami Kafka data directory + networks: + - sientia-network + healthcheck: + # Checks if Kafka is ready by trying to list topics using the internal listener + test: ["CMD-SHELL", "/opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server kafka:19092 --list > /dev/null || exit 1"] + interval: 30s + timeout: 10s + retries: 5 + + kafka-ui: + image: provectuslabs/kafka-ui:latest + container_name: kafka-ui + ports: + - "8082:8080" # Kafka UI will be accessible on host's port 8082 + environment: + KAFKA_CLUSTERS_0_NAME: sientia-local-kafka + KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:19092 # Connects to Kafka's internal listener + # DYNAMIC_CONFIG_ENABLED: 'true' # Optional: To allow config changes through UI + depends_on: + kafka: # Ensures Kafka starts and is healthy before Kafka UI + condition: service_healthy + networks: + - sientia-network + + +networks: + sientia-network: + driver: bridge + +volumes: + postgres_data: + driver: local + couchbase_data: + driver: local + redis_data: + driver: local + kafka_data: + driver: local \ No newline at end of file diff --git a/input_sample.json b/input_sample.json new file mode 100644 index 0000000..4a89997 --- /dev/null +++ b/input_sample.json @@ -0,0 +1,5 @@ +{ + "schedule_name": "orchestrator-test", + "pipelines_query": "SELECT pipelines.*, models FROM `pipelines` JOIN `models` ON KEYS pipelines.model_id;", + "opc_servers_query": "select META().id, opc_servers.* from `opc_servers`;" +} \ No newline at end of file diff --git a/orchestrator/__init__.py b/orchestrator/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/orchestrator/activities/__init__.py b/orchestrator/activities/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/orchestrator/activities/activities.py b/orchestrator/activities/activities.py new file mode 100644 index 0000000..1265539 --- /dev/null +++ b/orchestrator/activities/activities.py @@ -0,0 +1,52 @@ +from temporalio import activity, workflow +from temporalio.client import Client + +with workflow.unsafe.imports_passed_through(): + from orchestrator.activities.couchbase import Couchbase + from orchestrator.activities.temporal_manager import TemporalManager + from orchestrator.activities.slot_manager import SlotManager + from orchestrator.activities.formatters import Formatters + from typing import Any + from logging import Logger + from sientia_do.notifications.handlers import NotificationHandler + + +class Activities(Couchbase, TemporalManager, SlotManager, Formatters): + + def __init__(self, + temporal_client: Client, + couchbase_config: dict[str, Any], + redis_config: dict[str, Any], + logger: Logger, + notification_handler: NotificationHandler): + + # Initialize parent classes + Couchbase.__init__(self, connection_string=couchbase_config['connection_string'], + username=couchbase_config['username'], + password=couchbase_config['password'], + logger=logger, + notification_handler=notification_handler) + + TemporalManager.__init__(self, + temporal_client=temporal_client, + logger=logger, + notification_handler=notification_handler) + + SlotManager.__init__(self, + host=redis_config['host'], + port=redis_config['port'], + username=redis_config['username'], + password=redis_config['password'], + logger=logger, + notification_handler=notification_handler) + + Formatters.__init__(self, + logger=logger, + notification_handler=notification_handler) + + @activity.defn(name="prepare_activity") + async def prepare_activity(self, input_data: dict[str, Any]): + await super().prepare_activity(input_data) + + def shutdown(self): + Couchbase.shutdown(self) diff --git a/orchestrator/activities/couchbase.py b/orchestrator/activities/couchbase.py new file mode 100644 index 0000000..4aeec5b --- /dev/null +++ b/orchestrator/activities/couchbase.py @@ -0,0 +1,94 @@ +from temporalio import activity, workflow + +with workflow.unsafe.imports_passed_through(): + from typing import Any + from logging import Logger + from datetime import timedelta + import json + import traceback + from couchbase.auth import PasswordAuthenticator + from couchbase.cluster import Cluster + from couchbase.options import ClusterOptions + from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.notifications.models import NotificationLevel + from sientia_do.temporal.activities.base import BaseActivity + + +class Couchbase(BaseActivity): + def __init__(self, connection_string: str, username: str, + password: str, logger: Logger, + notification_handler: NotificationHandler): + + self.connection_string = connection_string + self.username = username + self.password = password + + logger.info("Initializing Couchbase connection...") + self.cluster = Cluster( + connection_string, + ClusterOptions( + authenticator=PasswordAuthenticator( + username=username, + password=password + ) + ) + ) + + logger.info("Awaiting Couchbase connection...") + self.cluster.wait_until_ready(timeout=timedelta(seconds=10)) + + logger.info("Couchbase connection ready") + + BaseActivity.__init__(self, + logger=logger, + notification_handler=notification_handler) + + def shutdown(self): + try: + self.cluster.close() + except Exception as e: + self.logger.error("Failed to close Couchbase connection: %s", e) + + def __del__(self): + self.shutdown() + + @activity.defn(name="load_query_from_couchbase") + async def load_query_from_couchbase(self, input_data: dict[str, Any]) -> list[dict[str, Any]]: + """ + Load a query from couchbase + + Args: + input_data (dict[str, Any]): The input data containing the query to execute + + Returns: + list[dict[str, Any]]: The result of the query + """ + query = input_data['query'] + + self.logger.info("Executing couchbase query: %s", query) + + try: + result = self.cluster.query(query) + except Exception as e: + trace = traceback.format_exc() + self.notification_handler.build_and_send_notification( + notification_id="COUCHBASE_LOAD_QUERY_ERROR", + message=f"Failed to execute couchbase query: {e}", + block="load_query_from_couchbase", + level=NotificationLevel.ERROR, + attachment_content=trace, + ) + + self.logger.error(trace) + raise e + + rows = [] + + for row in result.rows(): + rows.append(row) + + self.logger.info("Fetched %d rows from couchbase", len(rows)) + self.logger.debug("Rows: \n %s", + json.dumps(rows, indent=4, sort_keys=True)) + + return rows diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py new file mode 100644 index 0000000..8a06865 --- /dev/null +++ b/orchestrator/activities/formatters.py @@ -0,0 +1,362 @@ +from sientia_do.notifications.models import NotificationLevel +from temporalio import activity, workflow + +with workflow.unsafe.imports_passed_through(): + import json + from typing import Any + from logging import Logger + from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.temporal.activities.base import BaseActivity + from orchestrator.utils.orchestrator_functions import ( + scouter, predictions_batch, gather_read_tags, build_tag_config + ) + from math import ceil + + +class Formatters(BaseActivity): + def __init__(self, logger: Logger, notification_handler: NotificationHandler): + BaseActivity.__init__(self, logger=logger, + notification_handler=notification_handler) + + @activity.defn(name="process_schedules") + async def process_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Process schedules. Generates a schedule config dictionary + based on the input data workflow type. + + Args: + - input_data (dict[str, Any]): The input data containing + the schedules to process. + - pipelines (list[dict[str, Any]]): The schedules to process. + + Returns: + - dict[str, Any]: The schedule config dictionary + """ + + self.logger.info("Processing schedules...") + + pipelines = input_data['pipelines'] + + schedule_config = {} + + for pipeline in pipelines: + if pipeline['workflow_type'] == 'scouter': + schedule_config[pipeline['schedule_name']] = scouter(pipeline) + elif pipeline['workflow_type'] == 'predictions_batch': + schedule_config[pipeline['schedule_name'] + ] = predictions_batch(pipeline) + + self.logger.info("Processed schedules") + + self.logger.debug(json.dumps( + schedule_config, indent=4, sort_keys=True)) + + return schedule_config + + @activity.defn(name="process_slots") + async def process_slots(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Extracts all read tags from input pipelines, divides them into slots and + returns a slot config dictionary. If no ingestor is available, only one slot + is created. + + Args: + - input_data (dict[str, Any]): The input data containing + the schedules to process. + - pipelines (list[dict[str, Any]]): The schedules to process. + - opc_servers (list[str]): The OPC servers to create ingestor config. + - active_ingestors (list[str]): The active ingestors to divide into slots. + + Returns: + - dict[str, Any]: The slot config dictionary + """ + + self.logger.info("Processing slots...") + + pipelines = input_data['pipelines'] + opc_servers_list = input_data['opc_servers'] + active_ingestors = input_data['active_ingestors'] + + opc_servers = {} + for server in opc_servers_list: + opc_servers[server['id']] = { + **server, + } + + tags = list(gather_read_tags(pipelines).values()) + + number_of_tags = len(tags) + number_of_slots = len(active_ingestors) if active_ingestors else 1 + tags_per_slot = ceil(number_of_tags / number_of_slots) + + slot_config = {} + last_index = 0 + + for i in range(1, number_of_slots): + slot_config[f"{i}"] = {} + for tag in tags[last_index:last_index + tags_per_slot]: + slot_config = build_tag_config( + tag, slot_config.copy(), opc_servers, i) + last_index += tags_per_slot + + slot_config[f"{number_of_slots}"] = {} + for tag in tags[last_index:]: + slot_config = build_tag_config( + tag, slot_config.copy(), opc_servers, number_of_slots) + + self.logger.info("Processed slots") + self.logger.debug(json.dumps( + slot_config, indent=4, sort_keys=True)) + + return slot_config + + @activity.defn(name="create_schedule_config") + async def create_schedule_config(self, + input_data: dict[str, Any]) -> dict[str, Any]: + """ + Creates a schedule config dictionary based on the input data. + Checks the existing schedule config and updates it with the new schedule config, + deleting unnecessary schedules, creating new schedules and updating existing schedules. + + Args: + - input_data (dict[str, Any]): The input data containing + the schedules to process. + - current_schedule_config (dict[str, Any]): The current schedule + config in Temporal server. + - schedule_config (dict[str, Any]): The schedule config to process. + + Returns: + - dict[str, Any]: The schedule config dictionary + """ + + self.logger.info("Creating schedule config...") + + current_schedule_config = input_data['current_schedule_config'] + schedule_config = input_data['schedule_config'] + + to_update = {} + to_create = {} + to_delete = [] + + for schedule_name, schedule in schedule_config.items(): + if schedule_name in current_schedule_config: + self.logger.debug(f"{current_schedule_config[schedule_name]}") + old_config = current_schedule_config[schedule_name]['data'] + + self.logger.debug(f"Comparing {schedule_name}:") + self.logger.debug(json.dumps( + old_config, indent=4, sort_keys=True)) + self.logger.debug(json.dumps( + schedule, indent=4, sort_keys=True)) + + if schedule != old_config: + to_update[schedule_name] = schedule + + elif schedule_name not in current_schedule_config: + to_create[schedule_name] = schedule + + for schedule_name in current_schedule_config: + if schedule_name not in schedule_config: + to_delete.append(schedule_name) + + output = { + "to_update": to_update, + "to_create": to_create, + "to_delete": to_delete + } + + self.logger.info("Created schedule config") + self.logger.debug(json.dumps( + output, indent=4, sort_keys=True)) + + return output + + @activity.defn(name="create_slot_config") + async def create_slot_config(self, + input_data: dict[str, Any]) -> dict[str, Any]: + """ + Creates a slot config dictionary based on the input data. + Checks the existing slot config and updates it with the new slot config, + deleting unnecessary slots. + + Args: + - input_data (dict[str, Any]): The input data containing + the slots to process. + - current_slot_config (dict[str, Any]): The current slot + config in Temporal server. + - slot_config (dict[str, Any]): The slot config to process. + + Returns: + - dict[str, Any]: The slot config dictionary + """ + + self.logger.info("Creating slot config...") + + current_slot_config = input_data['current_slot_config'] + slot_config = input_data['slot_config'] + to_delete = [] + + number_of_current_slots = len(current_slot_config) + number_of_slots = len(slot_config) + + if number_of_current_slots > number_of_slots: + to_delete = [str(i) for i in range( + number_of_slots + 1, number_of_current_slots + 1)] + + output = { + "to_delete": to_delete, + "to_insert": slot_config + } + + self.logger.info("Created slot config") + self.logger.debug(json.dumps( + output, indent=4, sort_keys=True)) + + return output + + def send_success_report(self, message: str, notification_id: str) -> None: + self.notification_handler.build_and_send_notification( + notification_id, + message, + "report_orchestration", + NotificationLevel.INFO + ) + + def send_error_report(self, message: str, notification_id: str, + attachment: dict[str, Any]) -> None: + self.notification_handler.build_and_send_notification( + notification_id, + message, + "report_orchestration", + NotificationLevel.ERROR, + attachment_content=json.dumps(attachment, indent=4, sort_keys=True) + ) + + def parse_report(self, input_data: dict[str, Any]) -> tuple[list[str], list[str]]: + success_keys = [key for key, value + in input_data.items() if value['success']] + + error_keys = [key for key, value + in input_data.items() if not value['success']] + + return success_keys, error_keys + + @activity.defn(name="report_schedule_orchestration") + async def report_schedule_orchestration(self, + input_data: dict[str, Any]) -> None: + """ + Reports the orchestration result to the notification handler. + + Args: + - input_data (dict[str, Any]): The input data containing + the orchestration result. + - created_schedules (dict[str, Any]): The created schedules. + - updated_schedules (dict[str, Any]): The updated schedules. + - deleted_schedules (list[str]): The deleted schedules. + """ + + self.logger.info("Reporting orchestration...") + + created_schedules = input_data['created_schedules'] + updated_schedules = input_data['updated_schedules'] + deleted_schedules = input_data['deleted_schedules'] + + # Send report for created schedules + if len(created_schedules) > 0: + success_keys, error_keys = self.parse_report(created_schedules) + + if len(success_keys) > 0: + self.send_success_report( + f"Created schedules: \n {', '.join(success_keys)}", + "REPORT_ORCHESTRATION_CREATED_SCHEDULES" + ) + + if len(error_keys) > 0: + self.send_error_report( + f"Failed to create schedules: \n {', '.join(error_keys)}", + "REPORT_ORCHESTRATION_CREATED_SCHEDULES", + created_schedules + ) + + # Send report for updated schedules + if len(updated_schedules) > 0: + success_keys, error_keys = self.parse_report(updated_schedules) + + if len(success_keys) > 0: + self.send_success_report( + f"Updated schedules: \n {', '.join(success_keys)}", + "REPORT_ORCHESTRATION_UPDATED_SCHEDULES" + ) + + if len(error_keys) > 0: + self.send_error_report( + f"Failed to update schedules: \n {', '.join(error_keys)}", + "REPORT_ORCHESTRATION_UPDATED_SCHEDULES", + updated_schedules + ) + + if len(deleted_schedules) > 0: + success_keys, error_keys = self.parse_report(deleted_schedules) + + if len(success_keys) > 0: + self.send_success_report( + f"Deleted schedules: \n {', '.join(success_keys)}", + "REPORT_ORCHESTRATION_DELETED_SCHEDULES" + ) + + if len(error_keys) > 0: + self.send_error_report( + f"Failed to delete schedules: \n {', '.join(error_keys)}", + "REPORT_ORCHESTRATION_DELETED_SCHEDULES", + deleted_schedules + ) + + @activity.defn(name="report_slot_orchestration") + async def report_slot_orchestration(self, + input_data: dict[str, Any]) -> None: + """ + Reports the orchestration result to the notification handler. + + Args: + - input_data (dict[str, Any]): The input data containing + the orchestration result. + - inserted_slots (dict[str, Any]): The inserted slots. + - deleted_slots (list[str]): The deleted slots. + """ + + self.logger.info("Reporting orchestration...") + + inserted_slots = input_data['inserted_slots'] + deleted_slots = input_data['deleted_slots'] + + if len(inserted_slots) > 0: + success_keys, error_keys = self.parse_report(inserted_slots) + + if len(success_keys) > 0: + self.send_success_report( + f"Inserted slots: \n {', '.join(success_keys)}", + "REPORT_ORCHESTRATION_INSERTED_SLOTS" + ) + + if len(error_keys) > 0: + self.send_error_report( + f"Failed to insert slots: \n {', '.join(error_keys)}", + "REPORT_ORCHESTRATION_INSERTED_SLOTS", + inserted_slots + ) + + if len(deleted_slots) > 0: + success_keys, error_keys = self.parse_report(deleted_slots) + + if len(success_keys) > 0: + self.send_success_report( + f"Deleted slots: \n {', '.join(success_keys)}", + "REPORT_ORCHESTRATION_DELETED_SLOTS" + ) + + if len(error_keys) > 0: + self.send_error_report( + f"Failed to delete slots: \n {', '.join(error_keys)}", + "REPORT_ORCHESTRATION_DELETED_SLOTS", + deleted_slots + ) diff --git a/orchestrator/activities/notification.py b/orchestrator/activities/notification.py new file mode 100644 index 0000000..e69de29 diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py new file mode 100644 index 0000000..45d77d4 --- /dev/null +++ b/orchestrator/activities/slot_manager.py @@ -0,0 +1,153 @@ +from temporalio import activity, workflow + +with workflow.unsafe.imports_passed_through(): + from typing import Any + import json + from logging import Logger + from sientia_do.temporal.activities.redis_base import Redis + from sientia_do.notifications.handlers import NotificationHandler + + +class SlotManager(Redis): + + def __init__(self, host: str, port: int, + username: str, password: str, + logger: Logger, notification_handler: NotificationHandler): + + Redis.__init__(self, host, port, username, + password, logger, notification_handler) + + @activity.defn(name="load_opc_slots") + async def load_opc_slots(self) -> dict[str, Any]: + """ + Load all OPC slots from Redis + + Returns: + dict[str, Any]: A dictionary of OPC slots + """ + + self.logger.info("Loading OPC slots...") + + opc_slots = {} + + slot_keys = self.redis_client.keys("slot:opc_tags:*") + + self.logger.debug("Slot keys: %s", slot_keys) + + if slot_keys: + if isinstance(slot_keys[0], bytes): + decoded_keys = [key.decode('utf-8') for key in slot_keys] + else: + decoded_keys = slot_keys + + for key in decoded_keys: + opc_slots[key] = self.get(key) + + self.logger.info(f"Loaded {len(opc_slots)} OPC slots") + + self.logger.debug("Loaded: \n %s", + json.dumps(opc_slots, indent=4, sort_keys=True)) + + return opc_slots + + @activity.defn(name="load_active_ingestors") + async def load_active_ingestors(self) -> list[str]: + """ + Load all active ingestors from Redis + + Returns: + list[str]: A list of active ingestors + """ + + self.logger.info("Loading active ingestors...") + + active_ingestors = self.redis_client.keys("heartbeat:ingestor:*") + + self.logger.info(f"Loaded {len(active_ingestors)} active ingestors") + + self.logger.debug("Active ingestors: \n %s", active_ingestors) + + return [ingestor.decode('utf-8') for ingestor in active_ingestors] + + @activity.defn(name="update_slots") + async def update_slots(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Update OPC slots in Redis + + Args: + - input_data (dict[str, Any]): The input data containing + the slots to update. + - to_insert (dict[str, Any]): The slots to insert. + + Returns: + - report (dict[str, Any]): A report of the updated slots. + """ + + to_insert = input_data['to_insert'] + self.logger.info("Updating OPC slots...") + + report = {} + + for slot in to_insert: + try: + self.set(f"slot:opc_tags:{slot}", + to_insert[slot], ttl=None) + report[slot] = { + "success": True, + "message": "Slot updated successfully" + } + except Exception as e: + self.logger.error("Failed to update slot %s: %s", + slot, str(e)) + report[slot] = { + "success": False, + "message": str(e) + } + + self.logger.info(f"Updated {len(to_insert)} OPC slots") + + self.logger.debug("Report: \n %s", + json.dumps(report, indent=4, sort_keys=True)) + + return report + + @activity.defn(name="delete_slots") + async def delete_slots(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Delete OPC slots from Redis + + Args: + - input_data (dict[str, Any]): The input data containing + the slots to delete. + - to_delete (list[str]): The slots to delete. + + Returns: + - report (dict[str, Any]): A report of the deleted slots. + """ + + to_delete = input_data['to_delete'] + self.logger.info("Deleting OPC slots...") + + report = {} + + for slot in to_delete: + try: + self.redis_client.delete(f"slot:opc_tags:{slot}") + report[slot] = { + "success": True, + "message": "Slot deleted successfully" + } + except Exception as e: + self.logger.error("Failed to delete slot %s: %s", + slot, str(e)) + report[slot] = { + "success": False, + "message": str(e) + } + + self.logger.info(f"Deleted {len(to_delete)} OPC slots") + + self.logger.debug("Report: \n %s", + json.dumps(report, indent=4, sort_keys=True)) + + return report diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py new file mode 100644 index 0000000..7fd520e --- /dev/null +++ b/orchestrator/activities/temporal_manager.py @@ -0,0 +1,275 @@ +from temporalio import activity, workflow +from temporalio.client import ( + Client, Schedule, ScheduleActionStartWorkflow, ScheduleIntervalSpec, ScheduleSpec, ScheduleUpdate, ScheduleUpdateInput) +from temporalio.common import SearchAttributeKey, SearchAttributePair, TypedSearchAttributes + +with workflow.unsafe.imports_passed_through(): + from typing import Any + from logging import Logger + from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.temporal.activities.base import BaseActivity + from google.protobuf.json_format import MessageToDict + import base64 + from datetime import timedelta + import json + from orchestrator.utils.converters import parse_frequency + + +class TemporalManager(BaseActivity): + def __init__(self, temporal_client: Client, logger: Logger, + notification_handler: NotificationHandler): + + self.temporal_client = temporal_client + + self.schedule_handles = {} + + self.model_id_id_key = SearchAttributeKey.for_keyword("model_id") + self.model_name_id_key = SearchAttributeKey.for_keyword("model_name") + self.orchestrated_id_key = SearchAttributeKey.for_keyword( + "orchestrated") + + BaseActivity.__init__(self, + logger=logger, + notification_handler=notification_handler) + + @activity.defn(name="load_schedule") + async def load_schedule(self) -> dict[str, Any]: + """ + Load all orchestrated schedules from Temporal. Filters by search attribute + "Orchestrated" set to "true" and returns a dictionary of schedule_id: + {frequency, data, handle} + + Returns: + dict[str, Any]: A dictionary of orchestrated schedules + """ + + self.logger.info("Getting orchestrated schedules...") + + orchestrated_schedules = {} + + async for schedule in await self.temporal_client.list_schedules(): + search_attrs = getattr(schedule, "search_attributes", {}) + if search_attrs.get("orchestrated", ["false"]) == ["true"]: + schedule_id = schedule.id + + handle = self.temporal_client.get_schedule_handle(schedule_id) + + self.schedule_handles[schedule_id] = handle + + desc = await handle.describe() + + for arg in desc.schedule.action.args: + data = MessageToDict(arg)['data'] + data = base64.b64decode(data).decode('utf-8') + + frequency = desc.schedule.spec.intervals[0].every.seconds + + orchestrated_schedules[schedule_id] = { + 'frequency': frequency, + 'data': json.loads(data), + } + + self.logger.info("Found %d orchestrated schedules", + len(orchestrated_schedules)) + + self.logger.debug("Orchestrated schedules: %s", + orchestrated_schedules) + + return orchestrated_schedules + + @activity.defn(name="create_schedules") + async def create_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Create schedules in Temporal + + Args: + - input_data (dict[str, Any]): The input data containing + the schedules to create. + - schedules (dict[str, Any]): The schedules to create. + + Returns: + - dict[str, Any]: A report of the created schedules. + """ + + schedules = input_data['schedules'] + + report = {} + + for schedule_name, schedule in schedules.items(): + search_attributes = TypedSearchAttributes([ + SearchAttributePair( + key=self.model_id_id_key, + value=schedule['model_id'] + ), + SearchAttributePair( + key=self.model_name_id_key, + value=schedule['model_name'] + ), + SearchAttributePair( + key=self.orchestrated_id_key, + value="true" + ) + ]) + workflow_type = schedule['workflow_type'] + + try: + self.logger.debug(f"Creating schedule {schedule_name}:") + self.logger.debug(json.dumps( + schedule, indent=4, sort_keys=True)) + await self.temporal_client.create_schedule( + schedule_name, + Schedule( + action=ScheduleActionStartWorkflow( + workflow_type, + schedule, + id=schedule_name, + task_queue=f"{workflow_type}-queue", + execution_timeout=timedelta(minutes=2) + ), + spec=ScheduleSpec( + intervals=[ + ScheduleIntervalSpec( + every=timedelta(seconds=parse_frequency( + schedule.get('frequency', '1m'))) + ) + ] + ) + ), + search_attributes=search_attributes + ) + + report[schedule_name] = { + "success": True, + "message": "Schedule created successfully" + } + except Exception as e: + self.logger.error("Failed to create schedule %s: %s", + schedule_name, str(e)) + report[schedule_name] = { + "success": False, + "message": str(e) + } + + self.logger.info(f"Processed {len(schedules)} schedules") + + self.logger.debug("\n %s", + json.dumps(report, indent=4, sort_keys=True)) + + return report + + @activity.defn(name="update_schedules") + async def update_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Update schedules in Temporal + + Args: + - input_data (dict[str, Any]): The input data containing + the schedules to update. + - schedules (dict[str, Any]): The schedules to update. + + Returns: + - dict[str, Any]: A report of the updated schedules. + """ + + schedules = input_data['schedules'] + + report = {} + + for schedule_name, schedule in schedules.items(): + try: + handler = self.schedule_handles.get(schedule_name) + + if not handler: + raise ValueError(f"Schedule {schedule_name} not found") + + async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: + schedule_action = input_data.description.schedule.action + + self.logger.debug("Updating schedule:") + + if hasattr(schedule_action, "args"): + self.logger.debug("New schedule:") + self.logger.debug(json.dumps( + schedule, indent=4, sort_keys=True)) + + schedule_action.args = [schedule] + + input_data.description.schedule.spec.intervals = [ + ScheduleIntervalSpec( + every=timedelta(seconds=parse_frequency( + schedule.get('frequency', '1m'))) + ) + ] + + return ScheduleUpdate(schedule=input_data.description.schedule) + + await handler.update(update_schedule) + + del update_schedule + + report[schedule_name] = { + "success": True, + "message": "Schedule updated successfully" + } + except Exception as e: + self.logger.error("Failed to update schedule %s: %s", + schedule_name, str(e)) + report[schedule_name] = { + "success": False, + "message": str(e) + } + + self.logger.info(f"Processed {len(schedules)} schedules") + + self.logger.debug("\n %s", + json.dumps(report, indent=4, sort_keys=True)) + + return report + + @activity.defn(name="delete_schedules") + async def delete_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Delete schedules in Temporal + + Args: + - input_data (dict[str, Any]): The input data containing + the schedules to delete. + - schedules (list[str]): The schedules to delete. + + Returns: + - dict[str, Any]: A report of the deleted schedules. + """ + + schedules = input_data['schedules'] + + report = {} + + for schedule_name in schedules: + try: + handler = self.schedule_handles.get(schedule_name) + + if not handler: + raise ValueError(f"Schedule {schedule_name} not found") + + await handler.delete() + + del self.schedule_handles[schedule_name] + + report[schedule_name] = { + "success": True, + "message": "Schedule deleted successfully" + } + except Exception as e: + self.logger.error("Failed to delete schedule %s: %s", + schedule_name, str(e)) + report[schedule_name] = { + "success": False, + "message": str(e) + } + + self.logger.info(f"Processed {len(schedules)} schedules") + + self.logger.debug("\n %s", + json.dumps(report, indent=4, sort_keys=True)) + + return report diff --git a/orchestrator/utils/__init__.py b/orchestrator/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/orchestrator/utils/connectors_config.py b/orchestrator/utils/connectors_config.py new file mode 100644 index 0000000..2e5e80f --- /dev/null +++ b/orchestrator/utils/connectors_config.py @@ -0,0 +1,18 @@ +from os import getenv + + +def build_redis_config(): + return { + 'host': getenv('REDIS_HOST', 'localhost'), + 'port': int(getenv('REDIS_PORT', '6379')), + 'username': getenv('REDIS_USERNAME', None), + 'password': getenv('REDIS_PASSWORD', None) + } + + +def build_couchbase_config(): + return { + 'connection_string': getenv('COUCHBASE_CONNECTION_STRING', 'couchbase://localhost'), + 'username': getenv('COUCHBASE_USERNAME', 'sientia'), + 'password': getenv('COUCHBASE_PASSWORD', 'sientia') + } diff --git a/orchestrator/utils/converters.py b/orchestrator/utils/converters.py new file mode 100644 index 0000000..45bd436 --- /dev/null +++ b/orchestrator/utils/converters.py @@ -0,0 +1,14 @@ +def parse_frequency(frequency: str) -> int: + """ + Parse frequency string to seconds + """ + if frequency.endswith("s"): + return int(frequency[:-1]) + elif frequency.endswith("m"): + return int(frequency[:-1]) * 60 + elif frequency.endswith("h"): + return int(frequency[:-1]) * 60 * 60 + elif frequency.endswith("d"): + return int(frequency[:-1]) * 60 * 60 * 24 + else: + raise ValueError("Invalid frequency") diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py new file mode 100644 index 0000000..22e5444 --- /dev/null +++ b/orchestrator/utils/orchestrator_functions.py @@ -0,0 +1,164 @@ +from typing import Any + + +def common_config(config: dict[str, Any]): + return { + "workflow_type": config['workflow_type'], + "schedule_name": config['schedule_name'], + "frequency": config.get('frequency', '1m'), + "max_retry_policy": config.get('max_retry_policy', 1), + + "model_id": config['model_id'], + "model_name": config['models']['name'], + } + + +def scouter(config: dict[str, Any]): + filters = {} + for f in config.get('filters', []): + filters[f['filter_name']] = { + "policy": f['policy'] + } + + tags = {} + for tag in config['read_tags']: + tags[tag['tag_name']] = { + "aggr_func": tag.get('aggr_func', 'lts'), + "data_range": tag.get('data_range', [-100, 100]) + } + + return { + **common_config(config), + + "topic": f"raw_{config['schedule_name']}", + "trigger_laborious": False, + "filters": filters, + "schema": "sientia_data", + "table_name": "laborious_data", + "retention_time": config.get('tag_retention_minutes', 60) * 60, + "model_tags": tags + } + + +def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[str, Any]]): + for fil in config: + base_filter_config[fil['filter_name']] = { + "policy": fil['policy'], + "config": fil.get('config', {}) + } + + return base_filter_config + + +def process_path_priority(path_priority: list[str]): + for priority in path_priority[:]: + if priority not in ["STOP", "CONTINUE", "REPEAT"]: + path_priority.remove(priority) + + for priority in ["STOP", "CONTINUE", "REPEAT"]: + if priority not in path_priority: + path_priority.append(priority) + + return path_priority[0:3] + + +def predictions_batch(config: dict[str, Any]): + tags = {} + for tag in config['write_tags']: + if tag['server_name'] not in tags: + tags[tag['server_name']] = {} + + tag_type = tag['type'] + + if tag_type == 'prediction' or tag_type == 'confidence': + tag_type_str = f"{tag_type}_tags" + + if tag_type_str not in tags[tag['server_name']]: + tags[tag['server_name']][tag_type_str] = {} + + tags[tag['server_name']][tag_type_str][tag['addr']] = { + "data_type": tag.get('data_type', 'float'), + } + + path_priority = process_path_priority(config.get( + 'path_priority', ["STOP", "CONTINUE", "REPEAT"])) + + return { + **common_config(config), + + "query": config['query'], + "schema": "sientia_data", + "table_name": "predictions", + "retention_time": config.get('model_retention_minutes', 60) * 60, + "opc_output_config": tags, + "input_filters": overlap_filter_config({ + "EMPTY_DATA": { + "policy": "STOP", + "config": {} + } + }, config['input_filters']), + "mlflow_transform_filters": overlap_filter_config({ + "API_ERROR": { + "policy": "STOP", + "config": {} + } + }, config['mlflow_transform_filters']), + "mlflow_predict_filters": overlap_filter_config({ + "API_ERROR": { + "policy": "STOP", + "config": {} + } + }, config['mlflow_predict_filters']), + "path_priority": path_priority + } + + +def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: + """ + Gathers all read tags from input pipelines. + + Args: + - pipelines (list[dict[str, Any]]): The schedules to process. + + Returns: + - dict[str, Any]: The read tags dictionary + """ + + tags = {} + + # Get all read tags from pipelines + for pipeline in pipelines: + for tag in pipeline['read_tags']: + tag_string = f"{tag['server_id']}:{tag['tag_address']}" + if tag_string not in tags: + tags[tag_string] = { + **tag, + "topics": [] + } + + tags[tag_string]['topics'].append( + f"raw_{pipeline['schedule_name']}") + + return tags + + +def build_tag_config(tag: dict[str, Any], slot_config: dict[str, Any], + opc_servers: dict[str, Any], i: int): + server_id = tag['server_id'] + server_name = opc_servers[server_id]['server_name'] + if server_name not in slot_config[f"{i}"]: + slot_config[f"{i}"][server_name] = { + "server_id": server_id, + "name": server_name, + "url": opc_servers[server_id]['url'], + "server_uri": opc_servers[server_id]['uri'], + "tags": {} + } + for name, spec in opc_servers[server_id].get('security_spec', {}).items(): + slot_config[f"{i}"][server_name][name] = spec + + slot_config[f"{i}"][server_name]["tags"][tag['tag_address']] = { + **tag, + } + + return slot_config diff --git a/orchestrator/worker/__init__.py b/orchestrator/worker/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py new file mode 100644 index 0000000..0814b25 --- /dev/null +++ b/orchestrator/worker/worker.py @@ -0,0 +1,103 @@ +from temporalio import workflow, client +from temporalio.worker import Worker + +with workflow.unsafe.imports_passed_through(): + import os + import sys + import asyncio + from orchestrator.workflows.orchestrator import Orchestrator + from orchestrator.activities.activities import Activities + from orchestrator.utils.connectors_config import ( + build_couchbase_config, + build_redis_config, + ) + from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.temporal.utils.logger import get_logger + + +async def main(): + host = os.getenv('TEMPORAL_HOST', 'localhost:7233') + logger = get_logger(__name__) + + logger.info('Starting Worker...') + + logger.info('Starting Notification Handler...') + + notification_handler = NotificationHandler( + servers=os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'http://localhost:9092'), + logger=logger, + project_name=os.getenv('PROJECT_NAME', 'orchestrator'), + ) + + logger.info('Starting Temporal Client...') + + temporal_client = await client.Client.connect( + target_host=host, + namespace=os.getenv('TEMPORAL_NAMESPACE', 'default') + ) + + logger.info('Starting Activities...') + + activities = Activities( + temporal_client=temporal_client, + couchbase_config=build_couchbase_config(), + redis_config=build_redis_config(), + logger=logger, + notification_handler=notification_handler + ) + + logger.info('Starting Workers...') + + workers = [ + Worker( + temporal_client, + task_queue='orchestrator-queue', + workflows=[Orchestrator], + activities=[ + # Base + activities.prepare_activity, + # Redis + activities.load_active_ingestors, + activities.load_opc_slots, + activities.update_slots, + activities.delete_slots, + # Couchbase + activities.load_query_from_couchbase, + # Temporal + activities.load_schedule, + activities.create_schedules, + activities.update_schedules, + activities.delete_schedules, + # Formatters + activities.process_schedules, + activities.process_slots, + activities.create_schedule_config, + activities.create_slot_config, + activities.report_schedule_orchestration, + activities.report_slot_orchestration, + ] + ) + ] + + handlers = [] + for w in workers: + handlers.append(w.run()) + + logger.info('Workers started successfully') + + try: + # This will run the workers and wait for them to complete. + # If an exception occurs in any of the worker handlers, it will be propagated here. + await asyncio.gather(*handlers) + except BaseException as e: + logger.error("An unhandled exception occurred: %s", e, exc_info=True) + finally: + if notification_handler: + notification_handler.shutdown() + if activities: + activities.shutdown() + # Exit with a non-zero status code to indicate failure to Kubernetes + sys.exit(1) + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/orchestrator/workflows/__init__.py b/orchestrator/workflows/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py new file mode 100644 index 0000000..12550d5 --- /dev/null +++ b/orchestrator/workflows/orchestrator.py @@ -0,0 +1,190 @@ +from temporalio import workflow + +with workflow.unsafe.imports_passed_through(): + from orchestrator.activities.activities import Activities + from typing import Any + from datetime import timedelta + from sientia_do.temporal.utils.policies import retry_policy + + +@workflow.defn(name="orchestrator") +class Orchestrator: + @workflow.run + async def run(self, input_data: dict[str, Any]): + + input_data['workflow_name'] = 'orchestrator' + + await workflow.execute_local_activity_method( + Activities.prepare_activity, + { + 'workflow_name': input_data['workflow_name'], + 'schedule_name': input_data['schedule_name'], + 'model_name': '-', + 'model_id': '-' + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + pipeline_config_handler = workflow.execute_local_activity_method( + Activities.load_query_from_couchbase, + { + 'query': input_data['pipelines_query'] + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + opc_servers_handler = workflow.execute_local_activity_method( + Activities.load_query_from_couchbase, + { + 'query': input_data['opc_servers_query'] + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + orchestrated_schedules_handler = workflow.execute_local_activity_method( + Activities.load_schedule, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + current_slot_config_handler = workflow.execute_local_activity_method( + Activities.load_opc_slots, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + active_ingestors_handler = workflow.execute_local_activity_method( + Activities.load_active_ingestors, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + pipeline_config = await pipeline_config_handler + orchestrated_schedules = await orchestrated_schedules_handler + current_slot_config = await current_slot_config_handler + opc_servers = await opc_servers_handler + active_ingestors = await active_ingestors_handler + + schedules_config_handler = workflow.execute_local_activity_method( + Activities.process_schedules, + { + 'pipelines': pipeline_config + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + slot_config_handler = workflow.execute_local_activity_method( + Activities.process_slots, + { + 'opc_servers': opc_servers, + 'active_ingestors': active_ingestors, + 'pipelines': pipeline_config, + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + schedules_config = await schedules_config_handler + slot_config = await slot_config_handler + + schedule_actions_handler = workflow.execute_local_activity_method( + Activities.create_schedule_config, + { + 'current_schedule_config': orchestrated_schedules, + 'schedule_config': schedules_config + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + slot_actions_handler = workflow.execute_local_activity_method( + Activities.create_slot_config, + { + 'current_slot_config': current_slot_config, + 'slot_config': slot_config + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + schedule_actions = await schedule_actions_handler + slot_actions = await slot_actions_handler + + slot_deletion_report_handler = workflow.execute_activity_method( + Activities.delete_slots, + { + 'to_delete': slot_actions['to_delete'] + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + slot_insertion_report_handler = workflow.execute_activity_method( + Activities.update_slots, + { + 'to_insert': slot_actions['to_insert'] + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + schedule_deletion_report_handler = workflow.execute_activity_method( + Activities.delete_schedules, + { + 'schedules': schedule_actions['to_delete'] + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + schedule_insertion_report_handler = workflow.execute_activity_method( + Activities.create_schedules, + { + 'schedules': schedule_actions['to_create'] + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + schedule_update_report_handler = workflow.execute_activity_method( + Activities.update_schedules, + { + 'schedules': schedule_actions['to_update'] + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + slot_deletion_report = await slot_deletion_report_handler + slot_insertion_report = await slot_insertion_report_handler + schedule_deletion_report = await schedule_deletion_report_handler + schedule_insertion_report = await schedule_insertion_report_handler + schedule_update_report = await schedule_update_report_handler + + schedule_report_handler = workflow.execute_activity_method( + Activities.report_schedule_orchestration, + { + 'created_schedules': schedule_insertion_report, + 'updated_schedules': schedule_update_report, + 'deleted_schedules': schedule_deletion_report + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + slot_report_handler = workflow.execute_activity_method( + Activities.report_slot_orchestration, + { + 'inserted_slots': slot_insertion_report, + 'deleted_slots': slot_deletion_report + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + await schedule_report_handler + await slot_report_handler diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..07d06ae --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +temporalio +psycopg2-binary +sqlalchemy +redis +couchbase +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.14 diff --git a/samples.json b/samples.json new file mode 100644 index 0000000..f38194e --- /dev/null +++ b/samples.json @@ -0,0 +1,69 @@ +{ + "models": { + "1": { + "name": "Demo Model-Demo2" + } + }, + "pipelines": { + "1": { + "schedule_name": "scouter-opcua-orchestrated-pipeline", + "model_id": "1", + "workflow_type": "scouter", + "frequency": "5s", + "max_retry_policy": 1, + "read_tags": [ + { + "tag_name": "Counter", + "server_id": "default_server", + "aggr_func": "avg", + "tag_address": "ns=2;i=2", + "frequency": "15000", + "data_range": [ + -100, + 100 + ] + }, + { + "tag_name": "Rollout", + "server_id": "default_server", + "aggr_func": "mdn", + "tag_address": "ns=2;i=3", + "frequency": "15000", + "data_range": [ + -100, + 100 + ] + }, + { + "tag_name": "Square", + "server_id": "default_server", + "aggr_func": "lts", + "tag_address": "ns=2;i=4", + "frequency": "15000", + "data_range": [ + -100, + 100 + ] + } + ], + "filters": [ + { + "filter_name": "OUT_OF_BOUNDS_FILTER", + "policy": "DISCARD" + }, + { + "filter_name": "NULL_VALUES_FILTER", + "policy": "DISCARD" + } + ], + "tag_retention_minutes": 60 + } + }, + "opc-servers": { + "1": { + "server_name": "default_server", + "url": "opc.tcp://sientia-opc-simulator.sientia.svc.cluster.local:4840", + "uri": "http://opcua-server.simulator" + } + } +} \ No newline at end of file diff --git a/test.ipynb b/test.ipynb new file mode 100644 index 0000000..ecf11d1 --- /dev/null +++ b/test.ipynb @@ -0,0 +1,408 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": 4, + "id": "a287fa45", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Initializing Couchbase connection...\n", + "Awaiting Couchbase connection...\n", + "Couchbase connection ready\n" + ] + } + ], + "source": [ + "from orchestrator.activities.couchbase import Couchbase\n", + "from unittest.mock import MagicMock\n", + "logger = MagicMock(info=MagicMock(side_effect=print), debug=MagicMock(side_effect=print))\n", + "couchbase = Couchbase(\n", + " connection_string=\"couchbase://localhost\",\n", + " username=\"sientia\",\n", + " password=\"sientia\",\n", + " logger=logger,\n", + " notification_handler=MagicMock()\n", + ")\n" + ] + }, + { + "cell_type": "code", + "execution_count": 3, + "id": "d9a0f9c4", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Executing couchbase query: %s \n", + "SELECT\n", + " pipelines.*,\n", + " models as model\n", + "FROM\n", + " `pipelines`\n", + "JOIN\n", + " `models` ON KEYS pipelines.model_id;\n", + "\n", + "Fetched %d rows from couchbase 1\n", + "Rows: \n", + " %s [\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", + " \"frequency\": \"5s\",\n", + " \"max_retry_policy\": 1,\n", + " \"model\": {\n", + " \"name\": \"Demo Model-Demo2\"\n", + " },\n", + " \"model_id\": \"1\",\n", + " \"name\": \"scouter-opcua-pipeline\",\n", + " \"read_tags\": [\n", + " {\n", + " \"aggr_func\": \"avg\",\n", + " \"data_range\": [\n", + " -100,\n", + " 100\n", + " ],\n", + " \"tag_name\": \"Counter\"\n", + " }\n", + " ],\n", + " \"tag_retention_minutes\": 60,\n", + " \"workflow_type\": \"scouter\"\n", + " }\n", + "]\n", + "[{'filters': [{'filter_name': 'OUT_OF_BOUNDS_FILTER', 'policy': 'DISCARD'}, {'filter_name': 'NULL_VALUES_FILTER', 'policy': 'DISCARD'}], 'frequency': '5s', 'max_retry_policy': 1, 'model': {'name': 'Demo Model-Demo2'}, 'model_id': '1', 'name': 'scouter-opcua-pipeline', 'read_tags': [{'aggr_func': 'avg', 'data_range': [-100, 100], 'tag_name': 'Counter'}], 'tag_retention_minutes': 60, 'workflow_type': 'scouter'}]\n" + ] + } + ], + "source": [ + "query = \"\"\"\n", + "SELECT\n", + " pipelines.*,\n", + " models as model\n", + "FROM\n", + " `pipelines`\n", + "JOIN\n", + " `models` ON KEYS pipelines.model_id;\n", + "\"\"\"\n", + "if __name__ == \"__main__\":\n", + "\n", + "\n", + " result = await couchbase.load_query_from_couchbase({\n", + " \"query\": query\n", + " })\n", + "\n", + " print(result)\n", + " " + ] + }, + { + "cell_type": "code", + "execution_count": 1, + "id": "7d01f160", + "metadata": {}, + "outputs": [], + "source": [ + "from temporalio import client\n", + "from orchestrator.activities.temporal_manager import TemporalManager\n", + "import os\n", + "from unittest.mock import MagicMock\n", + "\n", + "host = \"localhost:7233\"\n", + "logger = MagicMock(info=MagicMock(side_effect=print), debug=MagicMock(side_effect=print))\n", + "\n", + "temporal_client = await client.Client.connect(\n", + " target_host=host,\n", + " namespace=os.getenv('TEMPORAL_NAMESPACE', 'default')\n", + ")\n", + "\n", + "manager = TemporalManager(temporal_client=temporal_client,\n", + " logger=logger,\n", + " notification_handler=MagicMock())\n", + " " + ] + }, + { + "cell_type": "code", + "execution_count": 6, + "id": "bb750ae6", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "" + ] + }, + "execution_count": 6, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "import asyncio\n", + "from datetime import timedelta\n", + "from temporalio.client import (\n", + " Client,\n", + " Schedule,\n", + " ScheduleActionStartWorkflow,\n", + " ScheduleIntervalSpec,\n", + " ScheduleSpec,\n", + ")\n", + "from temporalio.common import TypedSearchAttributes, SearchAttributeKey, SearchAttributePair\n", + "\n", + "\n", + "customer_id_key = SearchAttributeKey.for_keyword(\"orchestrated\")\n", + "search_attributes = TypedSearchAttributes([\n", + " SearchAttributePair(customer_id_key, \"true\")\n", + "])\n", + "await temporal_client.create_schedule(\n", + " \"meu-schedule-id5\",\n", + " Schedule(\n", + " action=ScheduleActionStartWorkflow(\n", + " 'scouter-test2',\n", + " {\n", + " 'args': {\n", + " 'arg1': 'value1'\n", + " }\n", + " },\n", + " id=\"workflow-id-unico\",\n", + " task_queue=\"nome-da-task-queue\",\n", + " ),\n", + " spec=ScheduleSpec(\n", + " intervals=[ScheduleIntervalSpec(every=timedelta(minutes=10))]\n", + " )\n", + " ),\n", + " search_attributes=search_attributes,\n", + ")\n" + ] + }, + { + "cell_type": "code", + "execution_count": 2, + "id": "1bd82225", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Getting orchestrated schedules...\n" + ] + } + ], + "source": [ + "schedules = await manager.load_schedule()" + ] + }, + { + "cell_type": "code", + "execution_count": 24, + "id": "1948670e", + "metadata": {}, + "outputs": [], + "source": [ + "from google.protobuf.json_format import MessageToDict\n", + "import base64\n", + "import json\n", + "for arg in schedules.schedule.action.args:\n", + " data = MessageToDict(arg)\n", + " data = base64.b64decode(data['data']).decode('utf-8')" + ] + }, + { + "cell_type": "code", + "execution_count": 12, + "id": "a4c777dd", + "metadata": {}, + "outputs": [], + "source": [ + "from google.protobuf.json_format import MessageToDict\n", + "import base64\n", + "import json\n", + "\n", + "schedules_config = {}\n", + "async for schedule in await temporal_client.list_schedules():\n", + " id = schedule.id\n", + "\n", + " handle = temporal_client.get_schedule_handle(id)\n", + "\n", + " desc = await handle.describe()\n", + "\n", + " for arg in desc.schedule.action.args:\n", + " data = MessageToDict(arg)['data']\n", + " data = base64.b64decode(data).decode('utf-8')\n", + "\n", + " frequency = desc.schedule.spec.intervals[0].every.seconds\n", + "\n", + " schedules_config[id] = {\n", + " 'frequency': frequency,\n", + " 'data': json.loads(data),\n", + " 'handle': handle\n", + " }\n", + " \n", + " \n", + " \n", + " " + ] + }, + { + "cell_type": "code", + "execution_count": 13, + "id": "988d1718", + "metadata": {}, + "outputs": [], + "source": [ + "schedules_config = {\n", + " \"scouter-opcua-orchestrated-pipeline\": schedules_config['scouter-opcua-orchestrated-pipeline'],\n", + "}" + ] + }, + { + "cell_type": "code", + "execution_count": 14, + "id": "c12f5e75", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "scouter-opcua-orchestrated-pipeline {'frequency': 5, 'data': {'filters': {'NULL_VALUES_FILTER': {'policy': 'DISCARD'}, 'OUT_OF_BOUNDS_FILTER': {'policy': 'DISCARD'}}, 'frequency': '5s', 'max_retry_policy': 1, 'model_id': '1', 'model_name': 'Demo Model-Demo2', 'model_tags': {'Counter': {'aggr_func': 'avg', 'data_range': [-100, 100]}}, 'retention_time': 3600, 'schedule_name': 'scouter-opcua-orchestrated-pipeline', 'schema': 'sientia_data', 'table_name': 'laborious_data', 'topic': 'raw_scouter-opcua-orchestrated-pipeline', 'trigger_laborious': False, 'workflow_type': 'scouter'}, 'handle': }\n", + "scouter-opcua-orchestrated-pipeline\n", + "{'workflow': 'scouter', 'args': [metadata {\n", + " key: \"encoding\"\n", + " value: \"json/plain\"\n", + "}\n", + "data: \"{\\\"filters\\\":{\\\"NULL_VALUES_FILTER\\\":{\\\"policy\\\":\\\"DISCARD\\\"},\\\"OUT_OF_BOUNDS_FILTER\\\":{\\\"policy\\\":\\\"DISCARD\\\"}},\\\"frequency\\\":\\\"5s\\\",\\\"max_retry_policy\\\":1,\\\"model_id\\\":\\\"1\\\",\\\"model_name\\\":\\\"Demo Model-Demo2\\\",\\\"model_tags\\\":{\\\"Counter\\\":{\\\"aggr_func\\\":\\\"avg\\\",\\\"data_range\\\":[-100,100]}},\\\"retention_time\\\":3600,\\\"schedule_name\\\":\\\"scouter-opcua-orchestrated-pipeline\\\",\\\"schema\\\":\\\"sientia_data\\\",\\\"table_name\\\":\\\"laborious_data\\\",\\\"topic\\\":\\\"raw_scouter-opcua-orchestrated-pipeline\\\",\\\"trigger_laborious\\\":false,\\\"workflow_type\\\":\\\"scouter\\\"}\"\n", + "], 'id': 'scouter-opcua-orchestrated-pipeline', 'task_queue': 'scouter-queue', 'execution_timeout': datetime.timedelta(seconds=120), 'run_timeout': None, 'task_timeout': None, 'retry_policy': None, 'memo': None, 'typed_search_attributes': TypedSearchAttributes(search_attributes=[]), 'headers': None, 'untyped_search_attributes': {}, 'static_summary': None, 'static_details': None, 'priority': Priority(priority_key=None)}\n" + ] + } + ], + "source": [ + "from temporalio.client import Client, ScheduleUpdateInput, ScheduleUpdate, ScheduleSpec\n", + "\n", + "\n", + "for id, schedule in schedules_config.items():\n", + " print(id, schedule)\n", + " \n", + " if schedule['frequency'] != 600:\n", + " print(id)\n", + "\n", + " handle = schedule['handle']\n", + " \n", + " async def update_schedule(input: ScheduleUpdateInput) -> ScheduleUpdate:\n", + " schedule_action = input.description.schedule.action\n", + "\n", + " print(schedule_action.__dict__)\n", + " \n", + " if hasattr(schedule_action, 'args'):\n", + " schedule_action.args = [{}]\n", + " \n", + " # Atualiza o intervalo de execução\n", + " input.description.schedule.spec.intervals = [\n", + " ScheduleIntervalSpec(every=timedelta(minutes=10))\n", + " ]\n", + " \n", + " return ScheduleUpdate(schedule=input.description.schedule)\n", + "\n", + " await handle.update(update_schedule)\n", + " \n" + ] + }, + { + "cell_type": "code", + "execution_count": 1, + "id": "87eb10c8", + "metadata": {}, + "outputs": [], + "source": [ + "from redis import Redis\n", + "\n", + "redis = Redis(host='localhost', port=6379)" + ] + }, + { + "cell_type": "code", + "execution_count": 6, + "id": "fe4ad9dc", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Chave: slot:opc_tags:2, Valor: {\n", + "\"slot2\": \"value\"\n", + "}\n", + "Chave: slot:opc_tags:1, Valor: {\n", + "\"slot1\": \"value\"\n", + "}\n" + ] + } + ], + "source": [ + "matching_keys = redis.keys(\"slot:opc_tags:*\")\n", + "\n", + "if matching_keys:\n", + " decoded_keys = [key.decode('utf-8') for key in matching_keys]\n", + " values = redis.mget(decoded_keys)\n", + "\n", + " items = {}\n", + " for i, key in enumerate(decoded_keys):\n", + " value = values[i]\n", + " if value is not None:\n", + " try:\n", + " items[key] = value.decode('utf-8')\n", + " except (UnicodeDecodeError, AttributeError):\n", + " items[key] = value\n", + " else:\n", + " items[key] = None\n", + "\n", + " for key, value in items.items():\n", + " print(f\"Chave: {key}, Valor: {value}\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "87a9dc89", + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "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": 5 +} diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/orchestrator/__init__.py b/tests/orchestrator/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/orchestrator/activities/__init__.py b/tests/orchestrator/activities/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/orchestrator/activities/test_activities.py b/tests/orchestrator/activities/test_activities.py new file mode 100644 index 0000000..997a55a --- /dev/null +++ b/tests/orchestrator/activities/test_activities.py @@ -0,0 +1,126 @@ +from unittest.mock import patch, MagicMock, ANY +from pytest import mark +from orchestrator.activities.activities import Activities +from orchestrator.activities.couchbase import Couchbase +from orchestrator.activities.temporal_manager import TemporalManager +from orchestrator.activities.slot_manager import SlotManager +from orchestrator.activities.formatters import Formatters + + +@patch('orchestrator.activities.couchbase.Couchbase.__init__') +@patch('orchestrator.activities.temporal_manager.TemporalManager.__init__') +@patch('orchestrator.activities.slot_manager.SlotManager.__init__') +@patch('orchestrator.activities.formatters.Formatters.__init__') +def test___init__(mock_formatters_init, mock_slot_manager_init, + mock_temporal_manager_init, + mock_couchbase_init): + + couchbase_config = { + 'connection_string': 'couchbase://localhost', + 'username': 'admin', + 'password': 'password' + } + + redis_config = { + 'host': 'localhost', + 'port': 6379, + 'username': 'admin', + 'password': 'password' + } + + temporal_client = MagicMock() + logger = MagicMock() + notification_handler = MagicMock() + + activities = Activities( + temporal_client=temporal_client, + couchbase_config=couchbase_config, + redis_config=redis_config, + logger=logger, + notification_handler=notification_handler + ) + + assert isinstance(activities, Activities) + assert isinstance(activities, Couchbase) + assert isinstance(activities, TemporalManager) + assert isinstance(activities, SlotManager) + assert isinstance(activities, Formatters) + + mock_slot_manager_init.assert_called_once_with( + ANY, + host="localhost", + port=6379, + username="admin", + password="password", + logger=logger, + notification_handler=notification_handler + ) + + mock_couchbase_init.assert_called_once_with( + ANY, + connection_string=couchbase_config['connection_string'], + username=couchbase_config['username'], + password=couchbase_config['password'], + logger=logger, + notification_handler=notification_handler + ) + + mock_temporal_manager_init.assert_called_once_with( + ANY, + temporal_client=temporal_client, + logger=logger, + notification_handler=notification_handler + ) + + mock_formatters_init.assert_called_once_with( + ANY, + logger=logger, + notification_handler=notification_handler + ) + + +@mark.asyncio +@patch('orchestrator.activities.couchbase.Cluster') +async def test_prepare_activity(_mock_cluster): + couchbase_config = { + 'connection_string': 'couchbase://localhost', + 'username': 'admin', + 'password': 'password' + } + + redis_config = { + 'host': 'localhost', + 'port': 6379, + 'username': 'admin', + 'password': 'password' + } + + temporal_client = MagicMock() + logger = MagicMock() + notification_handler = MagicMock() + + activities = Activities( + temporal_client=temporal_client, + couchbase_config=couchbase_config, + redis_config=redis_config, + logger=logger, + notification_handler=notification_handler + ) + + input_data = { + 'workflow_name': 'test-workflow-name', + 'schedule_name': 'test-schedule-name', + 'model_name': 'test-model-name', + 'model_id': 'test-model-id' + } + + await activities.prepare_activity(input_data) + + assert activities.notification_handler.base_notification.pipeline == input_data[ + 'workflow_name'] + assert activities.notification_handler.base_notification.trigger == input_data[ + 'schedule_name'] + assert activities.notification_handler.base_notification.model_name == input_data[ + 'model_name'] + assert activities.notification_handler.base_notification.model_id == input_data[ + 'model_id'] diff --git a/tests/orchestrator/activities/test_couchbase.py b/tests/orchestrator/activities/test_couchbase.py new file mode 100644 index 0000000..84f4da7 --- /dev/null +++ b/tests/orchestrator/activities/test_couchbase.py @@ -0,0 +1,56 @@ +from unittest.mock import MagicMock, patch, ANY +from pytest import fixture, mark, raises +from sientia_do.notifications.models import NotificationLevel +from orchestrator.activities.couchbase import Couchbase + + +@fixture +@patch("orchestrator.activities.couchbase.Cluster") +def couchbase(_cluster_mock): + return Couchbase( + connection_string="couchbase://localhost", + username="admin", + password="password", + logger=MagicMock(), + notification_handler=MagicMock(), + ) + + +@mark.asyncio +async def test_load_query_from_couchbase_success(couchbase): + couchbase.cluster.query.return_value.rows.return_value = [ + {"id": "1", "name": "test"}, + {"id": "2", "name": "test2"}, + ] + query = "SELECT * FROM bucket" + + result = await couchbase.load_query_from_couchbase({ + "query": query, + }) + + assert result == [ + {"id": "1", "name": "test"}, + {"id": "2", "name": "test2"}, + ] + couchbase.cluster.query.assert_called_once_with(query) + couchbase.notification_handler.build_and_send_notification.assert_not_called() + + +@mark.asyncio +async def test_load_query_from_couchbase_failure(couchbase): + couchbase.cluster.query.side_effect = Exception("Test error") + query = "SELECT * FROM bucket" + + with raises(Exception): + await couchbase.load_query_from_couchbase({ + "query": query, + }) + + couchbase.cluster.query.assert_called_once_with(query) + couchbase.notification_handler.build_and_send_notification.assert_called_once_with( + notification_id="COUCHBASE_LOAD_QUERY_ERROR", + message="Failed to execute couchbase query: Test error", + block="load_query_from_couchbase", + level=NotificationLevel.ERROR, + attachment_content=ANY, + ) diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py new file mode 100644 index 0000000..73f29a7 --- /dev/null +++ b/tests/orchestrator/activities/test_formatters.py @@ -0,0 +1,500 @@ +from unittest.mock import MagicMock, patch, call, ANY +import json +from pytest import fixture, mark +from sientia_do.notifications.models import NotificationLevel +from orchestrator.activities.formatters import Formatters +from orchestrator.utils.orchestrator_functions import build_tag_config + + +@fixture +def formatters(): + return Formatters( + logger=MagicMock(), + notification_handler=MagicMock() + ) + + +@mark.asyncio +@patch("orchestrator.activities.formatters.scouter", + return_value="test_scouter") +@patch("orchestrator.activities.formatters.predictions_batch", + return_value="test_predictions_batch") +async def test_process_schedules(mock_predictions_batch, mock_scouter, formatters): + input_data = { + "pipelines": [ + { + "schedule_name": "test_schedule_name", + "workflow_type": "scouter", + "model_name": "test_model_name", + "model_id": "test_model_id" + }, + { + "schedule_name": "test_schedule_name2", + "workflow_type": "predictions_batch", + "model_name": "test_model_name", + "model_id": "test_model_id" + } + ] + } + + result = await formatters.process_schedules(input_data) + + assert result == { + "test_schedule_name": "test_scouter", + "test_schedule_name2": "test_predictions_batch" + } + + mock_scouter.assert_called_once_with(input_data['pipelines'][0]) + mock_predictions_batch.assert_called_once_with(input_data['pipelines'][1]) + + +@mark.asyncio +@patch("orchestrator.activities.formatters.gather_read_tags", + return_value={ + "1:test_tag_address": { + "server_id": "1", + "server_name": "test_server_name", + "tag_address": "test_tag_address", + "topics": ["raw_test_schedule"] + }, + "2:test_tag_address2": { + "server_id": "2", + "server_name": "test_server_name2", + "tag_address": "test_tag_address2", + "topics": ["raw_test_schedule2"] + }, + "2:test_tag_address3": { + "server_id": "2", + "server_name": "test_server_name2", + "tag_address": "test_tag_address3", + "topics": ["raw_test_schedule2"] + } + }) +@patch("orchestrator.activities.formatters.build_tag_config", side_effect=build_tag_config) +async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, formatters): + input_data = { + "opc_servers": [ + { + "id": "1", + "server_name": "test_server_name", + "url": "test_url", + "uri": "test_uri", + "security_spec": { + "test_name": "test_spec" + } + }, + { + "id": "2", + "server_name": "test_server_name2", + "url": "test_url2", + "uri": "test_uri2" + } + ], + "active_ingestors": [ + "test_active_ingestor1", + "test_active_ingestor2" + ], + "pipelines": "test_gather_read_tags" + } + + result = await formatters.process_slots(input_data) + + mock_gather_read_tags.assert_called_once_with(input_data['pipelines']) + mock_build_tag_config.assert_has_calls([ + call( + { + "server_id": "1", + "server_name": "test_server_name", + "tag_address": "test_tag_address", + "topics": ["raw_test_schedule"] + }, + ANY, + { + "1": { + "id": "1", + "server_name": "test_server_name", + "url": "test_url", + "uri": "test_uri", + "security_spec": { + "test_name": "test_spec" + } + }, + "2": { + "id": "2", + "server_name": "test_server_name2", + "url": "test_url2", + "uri": "test_uri2" + } + }, + 1 + ) + ]) + mock_build_tag_config.assert_has_calls([ + call( + { + "server_id": "2", + "server_name": "test_server_name2", + "tag_address": "test_tag_address2", + "topics": ["raw_test_schedule2"] + }, + ANY, + { + "1": { + "id": "1", + "server_name": "test_server_name", + "url": "test_url", + "uri": "test_uri", + "security_spec": { + "test_name": "test_spec" + } + }, + "2": { + "id": "2", + "server_name": "test_server_name2", + "url": "test_url2", + "uri": "test_uri2" + } + }, + 1 + ) + ]) + mock_build_tag_config.assert_has_calls([ + call( + { + "server_id": "2", + "server_name": "test_server_name2", + "tag_address": "test_tag_address3", + "topics": ["raw_test_schedule2"] + }, + ANY, + { + "1": { + "id": "1", + "server_name": "test_server_name", + "url": "test_url", + "uri": "test_uri", + "security_spec": { + "test_name": "test_spec" + } + }, + "2": { + "id": "2", + "server_name": "test_server_name2", + "url": "test_url2", + "uri": "test_uri2" + } + }, + 2 + ) + ]) + + assert result == { + "1": { + "test_server_name": { + "server_id": "1", + "name": "test_server_name", + "url": "test_url", + "server_uri": "test_uri", + "test_name": "test_spec", + "tags": { + "test_tag_address": { + "server_id": "1", + "server_name": "test_server_name", + "tag_address": "test_tag_address", + "topics": ["raw_test_schedule"] + } + } + }, + "test_server_name2": { + "server_id": "2", + "name": "test_server_name2", + "url": "test_url2", + "server_uri": "test_uri2", + "tags": { + "test_tag_address2": { + "server_id": "2", + "server_name": "test_server_name2", + "tag_address": "test_tag_address2", + "topics": ["raw_test_schedule2"] + } + } + } + }, + "2": { + "test_server_name2": { + "server_id": "2", + "name": "test_server_name2", + "url": "test_url2", + "server_uri": "test_uri2", + "tags": { + "test_tag_address3": { + "server_id": "2", + "server_name": "test_server_name2", + "tag_address": "test_tag_address3", + "topics": ["raw_test_schedule2"] + } + } + } + } + } + + +@mark.asyncio +async def test_create_schedule_config(formatters): + input_data = { + "current_schedule_config": { + "test_schedule_name_to_delete": { + "frequency": 60, + "data": {"test": "test"} + }, + "test_schedule_name_to_update": { + "frequency": 60, + "data": {"test": "test"} + } + }, + "schedule_config": { + "test_schedule_name_to_create": { + "frequency": 60, + "data": {"test": "test"} + }, + "test_schedule_name_to_update": { + "frequency": 60, + "data": {"test": "test2"} + } + } + } + + result = await formatters.create_schedule_config(input_data) + + assert result == { + "to_create": { + "test_schedule_name_to_create": { + "frequency": 60, + "data": {"test": "test"} + } + }, + "to_update": { + "test_schedule_name_to_update": { + "frequency": 60, + "data": {"test": "test2"} + } + }, + "to_delete": [ + "test_schedule_name_to_delete" + ] + } + + +@mark.asyncio +async def test_create_slot_config(formatters): + input_data = { + "current_slot_config": { + "1": { + "frequency": 60, + "data": {"test": "test"} + }, + "2": { + "frequency": 60, + "data": {"test": "test"} + } + }, + "slot_config": { + "1": { + "frequency": 60, + "data": {"test": "test2"} + } + } + } + + result = await formatters.create_slot_config(input_data) + + assert result == { + "to_delete": [ + "2" + ], + "to_insert": { + "1": { + "frequency": 60, + "data": {"test": "test2"} + } + } + } + + +def test_send_success_report(formatters): + formatters.send_success_report("test_message", "test_notification_id") + formatters.notification_handler.build_and_send_notification.assert_called_once_with( + "test_notification_id", + "test_message", + "report_orchestration", + NotificationLevel.INFO + ) + + +def test_send_error_report(formatters): + formatters.send_error_report( + "test_message", "test_notification_id", {"test": "test"}) + formatters.notification_handler.build_and_send_notification.assert_called_once_with( + "test_notification_id", + "test_message", + "report_orchestration", + NotificationLevel.ERROR, + attachment_content=json.dumps( + {"test": "test"}, indent=4, sort_keys=True) + ) + + +def test_parse_report(formatters): + input_data = { + "test_schedule_name_to_create": { + "success": True + }, + "test_schedule_name_to_create_error": { + "success": False, + "error": "test_error" + } + } + + result = formatters.parse_report(input_data) + + assert result == ( + ["test_schedule_name_to_create"], + ["test_schedule_name_to_create_error"] + ) + + +@mark.asyncio +async def test_report_schedule_orchestration(formatters): + formatters.parse_report = MagicMock( + side_effect=formatters.parse_report + ) + formatters.send_success_report = MagicMock() + formatters.send_error_report = MagicMock() + + input_data = { + "created_schedules": { + "test_schedule_name_to_create": { + "success": True + }, + "test_schedule_name_to_create_error": { + "success": False, + "error": "test_error" + } + }, + "updated_schedules": { + "test_schedule_name_to_update": { + "success": True + }, + "test_schedule_name_to_update_error": { + "success": False, + "error": "test_error" + } + }, + "deleted_schedules": { + "test_schedule_name_to_delete": { + "success": True + }, + "test_schedule_name_to_delete_error": { + "success": False, + "error": "test_error" + } + } + } + + await formatters.report_schedule_orchestration(input_data) + + formatters.parse_report.assert_has_calls([ + call(input_data['created_schedules']), + call(input_data['updated_schedules']), + call(input_data['deleted_schedules']) + ]) + formatters.send_success_report.assert_has_calls([ + call( + "Created schedules: \n test_schedule_name_to_create", + "REPORT_ORCHESTRATION_CREATED_SCHEDULES" + ), + call( + "Updated schedules: \n test_schedule_name_to_update", + "REPORT_ORCHESTRATION_UPDATED_SCHEDULES" + ), + call( + "Deleted schedules: \n test_schedule_name_to_delete", + "REPORT_ORCHESTRATION_DELETED_SCHEDULES" + ) + ]) + formatters.send_error_report.assert_has_calls([ + call( + "Failed to create schedules: \n test_schedule_name_to_create_error", + "REPORT_ORCHESTRATION_CREATED_SCHEDULES", + input_data['created_schedules'] + ), + call( + "Failed to update schedules: \n test_schedule_name_to_update_error", + "REPORT_ORCHESTRATION_UPDATED_SCHEDULES", + input_data['updated_schedules'] + ), + call( + "Failed to delete schedules: \n test_schedule_name_to_delete_error", + "REPORT_ORCHESTRATION_DELETED_SCHEDULES", + input_data['deleted_schedules'] + ) + ]) + + +@mark.asyncio +async def test_report_slot_orchestration(formatters): + formatters.parse_report = MagicMock( + side_effect=formatters.parse_report + ) + formatters.send_success_report = MagicMock() + formatters.send_error_report = MagicMock() + + input_data = { + "inserted_slots": { + "test_slot_name_to_create": { + "success": True + }, + "test_slot_name_to_create_error": { + "success": False, + "error": "test_error" + } + }, + "deleted_slots": { + "test_slot_name_to_delete": { + "success": True + }, + "test_slot_name_to_delete_error": { + "success": False, + "error": "test_error" + } + } + } + + await formatters.report_slot_orchestration(input_data) + + formatters.parse_report.assert_has_calls([ + call(input_data['inserted_slots']), + call(input_data['deleted_slots']) + ]) + formatters.send_success_report.assert_has_calls([ + call( + "Inserted slots: \n test_slot_name_to_create", + "REPORT_ORCHESTRATION_INSERTED_SLOTS" + ), + call( + "Deleted slots: \n test_slot_name_to_delete", + "REPORT_ORCHESTRATION_DELETED_SLOTS" + ) + ]) + formatters.send_error_report.assert_has_calls([ + call( + "Failed to insert slots: \n test_slot_name_to_create_error", + "REPORT_ORCHESTRATION_INSERTED_SLOTS", + input_data['inserted_slots'] + ), + call( + "Failed to delete slots: \n test_slot_name_to_delete_error", + "REPORT_ORCHESTRATION_DELETED_SLOTS", + input_data['deleted_slots'] + ) + ]) diff --git a/tests/orchestrator/activities/test_slot_manager.py b/tests/orchestrator/activities/test_slot_manager.py new file mode 100644 index 0000000..ca2adfa --- /dev/null +++ b/tests/orchestrator/activities/test_slot_manager.py @@ -0,0 +1,147 @@ +from unittest.mock import MagicMock, patch, call +from pytest import mark, fixture +from orchestrator.activities.slot_manager import SlotManager + + +@fixture +@patch("orchestrator.activities.slot_manager.Redis.__init__") +def slot_manager(_redis_mock): + + slot_manager = SlotManager( + host="localhost", + port=6379, + username="admin", + password="password", + logger=MagicMock(), + notification_handler=MagicMock() + ) + + slot_manager.redis_client = MagicMock() + slot_manager.logger = MagicMock() + slot_manager.notification_handler = MagicMock() + + return slot_manager + + +@mark.asyncio +async def test_load_opc_slots_no_slot_keys(slot_manager): + slot_manager.redis_client.keys.return_value = [] + assert await slot_manager.load_opc_slots() == {} + + +@mark.asyncio +async def test_load_opc_slots(slot_manager): + slot_manager.redis_client.keys.return_value = [ + b"slot:opc_tags:1", b"slot:opc_tags:2", b"slot:opc_tags:3"] + + slot_manager.get = MagicMock( + side_effect=[ + "value1", + "value2", + None + ] + ) + + response = await slot_manager.load_opc_slots() + + assert response == { + "slot:opc_tags:1": "value1", + "slot:opc_tags:2": "value2", + "slot:opc_tags:3": None + } + + +@mark.asyncio +async def test_load_opc_slots_no_decode(slot_manager): + slot_manager.redis_client.keys.return_value = [ + "slot:opc_tags:1", "slot:opc_tags:2", "slot:opc_tags:3"] + + slot_manager.get = MagicMock( + side_effect=[ + "value1", + "value2", + None + ] + ) + + response = await slot_manager.load_opc_slots() + + assert response == { + "slot:opc_tags:1": "value1", + "slot:opc_tags:2": "value2", + "slot:opc_tags:3": None + } + + +@mark.asyncio +async def test_load_active_ingestors(slot_manager): + slot_manager.redis_client.keys.return_value = [ + b"heartbeat:ingestor:1", b"heartbeat:ingestor:2", b"heartbeat:ingestor:3"] + + response = await slot_manager.load_active_ingestors() + + assert response == ["heartbeat:ingestor:1", + "heartbeat:ingestor:2", "heartbeat:ingestor:3"] + + +@mark.asyncio +async def test_update_slots(slot_manager): + slot_manager.set = MagicMock( + side_effect=[ + None, + Exception("Test exception") + ] + ) + + response = await slot_manager.update_slots({ + "to_insert": { + "1": "value1", + "2": "value2" + } + }) + + slot_manager.set.assert_has_calls([ + call("slot:opc_tags:1", "value1", ttl=None), + call("slot:opc_tags:2", "value2", ttl=None) + ]) + + assert response == { + "1": { + "success": True, + "message": "Slot updated successfully" + }, + "2": { + "success": False, + "message": "Test exception" + } + } + + +@mark.asyncio +async def test_delete_slots(slot_manager): + slot_manager.redis_client.delete = MagicMock( + side_effect=[ + None, + Exception("Test exception") + ] + ) + + response = await slot_manager.delete_slots({ + "to_delete": ["1", "2"] + }) + + slot_manager.redis_client.delete.assert_has_calls([ + call("slot:opc_tags:1"), + call("slot:opc_tags:2") + ]) + + assert response == { + "1": { + "success": True, + "message": "Slot deleted successfully" + }, + "2": { + "success": False, + "message": "Test exception" + } + } diff --git a/tests/orchestrator/activities/test_temporal_manager.py b/tests/orchestrator/activities/test_temporal_manager.py new file mode 100644 index 0000000..7e504ef --- /dev/null +++ b/tests/orchestrator/activities/test_temporal_manager.py @@ -0,0 +1,282 @@ +from unittest.mock import MagicMock, patch, AsyncMock, call, ANY +from datetime import timedelta +import base64 +import json +from pytest import fixture, mark +from orchestrator.activities.temporal_manager import TemporalManager +from orchestrator.utils.converters import parse_frequency + + +@fixture +def temporal_manager(): + return TemporalManager( + temporal_client=MagicMock(), + logger=MagicMock(), + notification_handler=MagicMock() + ) + + +@mark.asyncio +@patch("orchestrator.activities.temporal_manager.MessageToDict", + return_value={"data": base64.b64encode(json.dumps({"test": "test"}).encode('utf-8'))}) +async def test_load_schedule(_mock_message_to_dict, temporal_manager): + # Create async iterator mock + async def async_iter(): + yield MagicMock( + id="test-schedule-id", + search_attributes={ + "orchestrated": ["true"] + } + ) + yield MagicMock( + id="test-schedule-id-2", + search_attributes={ + "Attr": ["false"] + } + ) + yield MagicMock( + id="test-schedule-id-3", + search_attributes={ + "Attr": ["false"] + } + ) + + temporal_manager.temporal_client.list_schedules = AsyncMock( + return_value=async_iter()) + temporal_manager.temporal_client.get_schedule_handle.return_value = MagicMock( + describe=AsyncMock( + return_value=MagicMock( + schedule=MagicMock( + action=MagicMock( + args=[ + MagicMock( + data=base64.b64encode(json.dumps( + {"test": "test"}).encode('utf-8')) + ) + ] + ) + ) + ) + ) + ) + temporal_manager.temporal_client.get_schedule_handle.return_value.describe \ + .return_value.schedule.spec = MagicMock( + intervals=[ + MagicMock( + every=MagicMock( + seconds=60 + ) + ) + ] + ) + + response = await temporal_manager.load_schedule() + + temporal_manager.temporal_client.list_schedules.assert_called_once() + assert response == { + "test-schedule-id": { + "frequency": 60, + "data": {"test": "test"} + } + } + + +@mark.asyncio +@patch("orchestrator.activities.temporal_manager.parse_frequency", + side_effect=parse_frequency) +@patch("orchestrator.activities.temporal_manager.Schedule") +@patch("orchestrator.activities.temporal_manager.ScheduleActionStartWorkflow") +@patch("orchestrator.activities.temporal_manager.ScheduleIntervalSpec") +@patch("orchestrator.activities.temporal_manager.ScheduleSpec") +@patch("orchestrator.activities.temporal_manager.TypedSearchAttributes") +@patch("orchestrator.activities.temporal_manager.SearchAttributePair") +async def test_create_schedule( + mock_search_attribute_pair, + mock_typed_search_attributes, + mock_schedule_spec, + mock_schedule_interval_spec, + mock_schedule_action_start_workflow, + mock_schedule, + mock_parse_frequency, + temporal_manager): + + input_data = { + "schedules": { + "test-schedule": { + "model_id": 1, + "model_name": "test-model-name", + "workflow_type": "test-workflow", + "frequency": "1m", + "data": {"test": "test"} + }, + "test-schedule-invalid-frequency": { + "model_id": 2, + "model_name": "test-model-name", + "workflow_type": "test-workflow", + "frequency": "10y", + "data": {"test": "test"} + } + } + } + + temporal_manager.temporal_client.create_schedule = AsyncMock() + + report = await temporal_manager.create_schedules(input_data) + + temporal_manager.temporal_client.create_schedule.assert_called_once_with( + "test-schedule", + mock_schedule.return_value, + search_attributes=mock_typed_search_attributes.return_value + ) + + mock_schedule.assert_called_once_with( + action=mock_schedule_action_start_workflow.return_value, + spec=mock_schedule_spec.return_value + ) + + mock_schedule_action_start_workflow.assert_has_calls([ + call( + "test-workflow", + input_data['schedules']['test-schedule'], + id="test-schedule", + task_queue="test-workflow-queue", + execution_timeout=ANY + ), + call( + "test-workflow", + input_data['schedules']['test-schedule-invalid-frequency'], + id="test-schedule-invalid-frequency", + task_queue="test-workflow-queue", + execution_timeout=ANY + ) + ]) + + mock_schedule_spec.assert_called_once_with( + intervals=[ + mock_schedule_interval_spec.return_value + ] + ) + + mock_schedule_interval_spec.assert_called_once_with( + every=timedelta(seconds=60) + ) + + mock_parse_frequency.assert_has_calls([ + call("1m"), + call("10y") + ]) + + mock_typed_search_attributes.assert_has_calls([ + call([ + mock_search_attribute_pair.return_value, + mock_search_attribute_pair.return_value, + mock_search_attribute_pair.return_value + ]), + call([ + mock_search_attribute_pair.return_value, + mock_search_attribute_pair.return_value, + mock_search_attribute_pair.return_value + ]) + ]) + + mock_search_attribute_pair.assert_has_calls([ + call( + key=temporal_manager.model_id_id_key, + value=1 + ), + call( + key=temporal_manager.model_name_id_key, + value="test-model-name" + ), + call( + key=temporal_manager.orchestrated_id_key, + value="true" + ) + ]) + + assert report == { + "test-schedule": { + "success": True, + "message": "Schedule created successfully" + }, + "test-schedule-invalid-frequency": { + "success": False, + "message": "Invalid frequency" + } + } + + +@mark.asyncio +@patch("orchestrator.activities.temporal_manager.parse_frequency", + side_effect=parse_frequency) +@patch("orchestrator.activities.temporal_manager.ScheduleIntervalSpec") +async def test_update_schedules( + _mock_schedule_interval_spec, + _mock_parse_frequency, + temporal_manager): + input_mock = MagicMock( + args=MagicMock() + ) + temporal_manager.schedule_handles = { + "test-schedule": MagicMock( + update=AsyncMock( + update=AsyncMock( + side_effect=lambda f: f(input_mock) + ) + ) + ) + } + input_data = { + "schedules": { + "test-schedule": { + "frequency": "1m", + "data": {"test": "test"} + }, + "test-schedule_no_handler": { + "frequency": "1m", + "data": {"test": "test"} + } + } + } + + report = await temporal_manager.update_schedules(input_data) + + temporal_manager.schedule_handles['test-schedule'].update.assert_called_once() + + assert report == { + "test-schedule": { + "success": True, + "message": "Schedule updated successfully" + }, + "test-schedule_no_handler": { + "success": False, + "message": "Schedule test-schedule_no_handler not found" + } + } + + +@mark.asyncio +async def test_delete_schedules(temporal_manager): + temporal_manager.schedule_handles = { + "test-schedule": MagicMock( + delete=AsyncMock() + ) + } + input_data = { + "schedules": [ + "test-schedule", "test-schedule_no_handler" + ] + } + + report = await temporal_manager.delete_schedules(input_data) + + assert report == { + "test-schedule": { + "success": True, + "message": "Schedule deleted successfully" + }, + "test-schedule_no_handler": { + "success": False, + "message": "Schedule test-schedule_no_handler not found" + } + } diff --git a/tests/orchestrator/utils/__init__.py b/tests/orchestrator/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/orchestrator/utils/test_connectors_config.py b/tests/orchestrator/utils/test_connectors_config.py new file mode 100644 index 0000000..3cf4dac --- /dev/null +++ b/tests/orchestrator/utils/test_connectors_config.py @@ -0,0 +1,51 @@ +from os import environ +from orchestrator.utils.connectors_config import (build_redis_config, + build_couchbase_config) + + +def test_build_redis_config_with_env_vars(): + environ['REDIS_HOST'] = 'localhost' + environ['REDIS_PORT'] = '6379' + environ['REDIS_USERNAME'] = 'sientia' + environ['REDIS_PASSWORD'] = 'sientia' + assert build_redis_config() == { + 'host': 'localhost', + 'port': 6379, + 'username': 'sientia', + 'password': 'sientia' + } + + +def test_build_couchbase_config_with_env_vars(): + environ['COUCHBASE_CONNECTION_STRING'] = 'couchbase://localhost' + environ['COUCHBASE_USERNAME'] = 'sientia' + environ['COUCHBASE_PASSWORD'] = 'sientia' + assert build_couchbase_config() == { + 'connection_string': 'couchbase://localhost', + 'username': 'sientia', + 'password': 'sientia' + } + + +def test_build_redis_config_with_defaults(): + environ.pop('REDIS_HOST', None) + environ.pop('REDIS_PORT', None) + environ.pop('REDIS_USERNAME', None) + environ.pop('REDIS_PASSWORD', None) + assert build_redis_config() == { + 'host': 'localhost', + 'port': 6379, + 'username': None, + 'password': None + } + + +def test_build_couchbase_config_with_defaults(): + environ.pop('COUCHBASE_CONNECTION_STRING', None) + environ.pop('COUCHBASE_USERNAME', None) + environ.pop('COUCHBASE_PASSWORD', None) + assert build_couchbase_config() == { + 'connection_string': 'couchbase://localhost', + 'username': 'sientia', + 'password': 'sientia' + } diff --git a/tests/orchestrator/utils/test_converters.py b/tests/orchestrator/utils/test_converters.py new file mode 100644 index 0000000..430316e --- /dev/null +++ b/tests/orchestrator/utils/test_converters.py @@ -0,0 +1,15 @@ +from orchestrator.utils.converters import parse_frequency + + +def test_parse_frequency(): + assert parse_frequency("1s") == 1 + assert parse_frequency("1m") == 60 + assert parse_frequency("1h") == 60 * 60 + assert parse_frequency("1d") == 60 * 60 * 24 + + try: + parse_frequency("1") + except ValueError as e: + assert str(e) == "Invalid frequency" + else: + assert False diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py new file mode 100644 index 0000000..789c374 --- /dev/null +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -0,0 +1,343 @@ +from unittest.mock import patch, call +from orchestrator.utils.orchestrator_functions import ( + common_config, + scouter, + predictions_batch, + overlap_filter_config, + process_path_priority, + gather_read_tags, + build_tag_config +) + + +def test_common_config(): + config = { + "workflow_type": "scouter", + "schedule_name": "test_schedule", + "model_id": "test_model_id", + "models": { + "name": "test_model_name" + } + } + result = common_config(config) + expected = { + "workflow_type": "scouter", + "schedule_name": "test_schedule", + "frequency": "1m", + "max_retry_policy": 1, + "model_id": "test_model_id", + "model_name": "test_model_name" + } + assert result == expected + + +def test_scouter(): + config = { + "workflow_type": "scouter", + "schedule_name": "test_schedule", + "model_id": "test_model_id", + "models": { + "name": "test_model_name" + }, + "filters": [ + { + "filter_name": "test_filter_name", + "policy": "test_policy" + } + ], + "read_tags": [ + { + "tag_name": "test_tag_name", + "aggr_func": "test_aggr_func", + "data_range": [1, 2] + } + ], + "tag_retention_minutes": 10 + } + result = scouter(config) + expected = { + "workflow_type": "scouter", + "schedule_name": "test_schedule", + "frequency": "1m", + "max_retry_policy": 1, + "model_id": "test_model_id", + "model_name": "test_model_name", + "topic": "raw_test_schedule", + "trigger_laborious": False, + "filters": { + "test_filter_name": { + "policy": "test_policy" + } + }, + "schema": "sientia_data", + "table_name": "laborious_data", + "retention_time": 10 * 60, + "model_tags": { + "test_tag_name": { + "aggr_func": "test_aggr_func", + "data_range": [1, 2] + } + } + } + assert result == expected + + +def test_overlap_filter_config(): + config = [ + { + "filter_name": "test_filter_name", + "policy": "test_policy" + }, + { + "filter_name": "test_filter_name2", + "policy": "test_policy2" + } + ] + result = overlap_filter_config({ + "test_filter_name": { + "policy": "test_policy" + } + }, config) + expected = { + "test_filter_name": { + "policy": "test_policy", + "config": {} + }, + "test_filter_name2": { + "policy": "test_policy2", + "config": {} + } + } + assert result == expected + + +def test_process_path_priority(): + config = ["OTHER", "STOP", "CONTINUE"] + result = process_path_priority(config) + expected = ["STOP", "CONTINUE", "REPEAT"] + assert result == expected + + +@patch('orchestrator.utils.orchestrator_functions.overlap_filter_config', + return_value={ + "test_filter_name": { + "policy": "test_policy", + "config": {} + } + }) +@patch('orchestrator.utils.orchestrator_functions.process_path_priority', + return_value=["STOP", "CONTINUE", "REPEAT"]) +def test_predictions_batch(mock_process_path_priority, + mock_overlap_filter_config): + config = { + "schedule_name": "test_schedule", + "workflow_type": "predictions_batch", + "model_id": "test_model_id", + "models": { + "name": "test_model_name" + }, + "query": "test_query", + "write_tags": [ + { + "server_name": "test_server_name", + "type": "prediction", + "addr": "test_addr" + }, + { + "server_name": "test_server_name", + "type": "confidence", + "addr": "test_addr" + } + ], + "input_filters": [ + { + "filter_name": "test_filter_name", + "policy": "test_policy" + } + ], + "mlflow_transform_filters": [ + { + "filter_name": "test_filter_name", + "policy": "test_policy" + } + ], + "mlflow_predict_filters": [ + { + "filter_name": "test_filter_name", + "policy": "test_policy" + } + ], + "path_priority": ["STOP", "CONTINUE", "REPEAT"], + } + + result = predictions_batch(config) + + mock_overlap_filter_config.assert_has_calls([ + call({ + "EMPTY_DATA": { + "policy": "STOP", + "config": {} + } + }, config['input_filters']) + ]) + mock_overlap_filter_config.assert_has_calls([ + call({ + "API_ERROR": { + "policy": "STOP", + "config": {} + } + }, config['mlflow_transform_filters']) + ]) + mock_overlap_filter_config.assert_has_calls([ + call({ + "API_ERROR": { + "policy": "STOP", + "config": {} + } + }, config['mlflow_predict_filters']) + ]) + mock_process_path_priority.assert_called_once_with(config['path_priority']) + + expected = { + "workflow_type": "predictions_batch", + "schedule_name": "test_schedule", + "frequency": "1m", + "max_retry_policy": 1, + "model_id": "test_model_id", + "model_name": "test_model_name", + "query": "test_query", + "schema": "sientia_data", + "table_name": "predictions", + "retention_time": 60 * 60, + "opc_output_config": { + "test_server_name": { + "prediction_tags": { + "test_addr": { + "data_type": "float" + } + }, + "confidence_tags": { + "test_addr": { + "data_type": "float" + } + } + } + }, + "input_filters": { + "test_filter_name": { + "policy": "test_policy", + "config": {} + } + }, + "mlflow_transform_filters": { + "test_filter_name": { + "policy": "test_policy", + "config": {} + } + }, + "mlflow_predict_filters": { + "test_filter_name": { + "policy": "test_policy", + "config": {} + } + }, + "path_priority": ["STOP", "CONTINUE", "REPEAT"] + } + assert result == expected + + +def test_gather_read_tags(): + pipelines = [ + { + "schedule_name": "test_schedule", + "read_tags": [ + { + "server_id": "1", + "server_name": "test_server_name", + "tag_address": "test_tag_address" + } + ] + }, + { + "schedule_name": "test_schedule2", + "read_tags": [ + { + "server_id": "2", + "server_name": "test_server_name2", + "tag_address": "test_tag_address2" + }, + { + "server_id": "2", + "server_name": "test_server_name2", + "tag_address": "test_tag_address3" + } + ] + } + ] + + result = gather_read_tags(pipelines) + + expected = { + "1:test_tag_address": { + "server_id": "1", + "server_name": "test_server_name", + "tag_address": "test_tag_address", + "topics": ["raw_test_schedule"] + }, + "2:test_tag_address2": { + "server_id": "2", + "server_name": "test_server_name2", + "tag_address": "test_tag_address2", + "topics": ["raw_test_schedule2"] + }, + "2:test_tag_address3": { + "server_id": "2", + "server_name": "test_server_name2", + "tag_address": "test_tag_address3", + "topics": ["raw_test_schedule2"] + } + } + + assert result == expected + + +def test_build_tag_config(): + tag = { + "server_id": "1", + "server_name": "test_server_name", + "tag_address": "test_tag_address" + } + opc_servers = { + "1": { + "server_name": "test_server_name", + "url": "test_url", + "uri": "test_uri", + "security_spec": { + "test_name": "test_spec" + } + } + } + slot_config = { + "1": {} + } + i = 1 + result = build_tag_config(tag, slot_config, opc_servers, i) + expected = { + "1": { + "test_server_name": { + "server_id": "1", + "name": "test_server_name", + "url": "test_url", + "server_uri": "test_uri", + "tags": { + "test_tag_address": { + "server_id": "1", + "server_name": "test_server_name", + "tag_address": "test_tag_address" + } + }, + "test_name": "test_spec" + } + } + } + assert result == expected diff --git a/tests/orchestrator/workflows/__init__.py b/tests/orchestrator/workflows/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/orchestrator/workflows/test_orchestrator.py b/tests/orchestrator/workflows/test_orchestrator.py new file mode 100644 index 0000000..0b1d4f8 --- /dev/null +++ b/tests/orchestrator/workflows/test_orchestrator.py @@ -0,0 +1,189 @@ +from unittest.mock import AsyncMock, patch, ANY, call +from pytest import fixture, mark +from orchestrator.workflows.orchestrator import Orchestrator +from orchestrator.activities.activities import Activities + + +@fixture +def orchestrator(): + return Orchestrator() + + +@mark.asyncio +@patch("orchestrator.workflows.orchestrator.workflow", new_callable=AsyncMock) +async def test_run(workflow_mock, orchestrator): + input_data = { + "pipelines_query": "SELECT * FROM bucket", + "opc_servers_query": "SELECT * FROM servers", + "schedule_name": "test-schedule-name", + } + + await orchestrator.run(input_data) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.prepare_activity, + { + "workflow_name": "orchestrator", + "schedule_name": "test-schedule-name", + "model_name": "-", + "model_id": "-" + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.load_query_from_couchbase, + { + "query": input_data["pipelines_query"] + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.load_query_from_couchbase, + { + "query": input_data["opc_servers_query"] + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.load_schedule, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.load_opc_slots, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.load_active_ingestors, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.process_schedules, + { + 'pipelines': workflow_mock.execute_local_activity_method.return_value + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.process_slots, + { + 'opc_servers': workflow_mock.execute_local_activity_method.return_value, + 'active_ingestors': workflow_mock.execute_local_activity_method.return_value, + 'pipelines': workflow_mock.execute_local_activity_method.return_value, + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.create_schedule_config, + { + 'current_schedule_config': workflow_mock.execute_local_activity_method.return_value, + 'schedule_config': workflow_mock.execute_local_activity_method.return_value + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.create_slot_config, + { + 'current_slot_config': workflow_mock.execute_local_activity_method.return_value, + 'slot_config': workflow_mock.execute_local_activity_method.return_value + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_activity_method.assert_has_calls([ + call( + Activities.delete_slots, + { + 'to_delete': + workflow_mock.execute_local_activity_method.return_value['to_delete'] + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_activity_method.assert_has_calls([ + call( + Activities.update_slots, + { + 'to_insert': + workflow_mock.execute_local_activity_method.return_value['to_insert'] + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_activity_method.assert_has_calls([ + call( + Activities.delete_schedules, + { + 'schedules': + workflow_mock.execute_local_activity_method.return_value['to_delete'] + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_activity_method.assert_has_calls([ + call( + Activities.create_schedules, + { + 'schedules': + workflow_mock.execute_local_activity_method.return_value['to_create'] + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_activity_method.assert_has_calls([ + call( + Activities.update_schedules, + { + 'schedules': + workflow_mock.execute_local_activity_method.return_value['to_update'] + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) diff --git a/values.yaml b/values.yaml new file mode 100644 index 0000000..016b99a --- /dev/null +++ b/values.yaml @@ -0,0 +1,179 @@ +# 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: "0.1.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: +- name: docker-hub-secret +# This is to override the chart name. +nameOverride: "sientia-orchestrator-worker" +fullnameOverride: "sientia-orchestrator-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-orchestrator-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: {} + # 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 + # lines, adjust them as necessary, and remove the curly braces after 'resources:'. + # limits: + # cpu: 100m + # memory: 128Mi + # requests: + # cpu: 100m + # memory: 128Mi + +# 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 "orchestrator.worker.worker" + initialDelaySeconds: 20 + periodSeconds: 30 + +readinessProbe: + exec: + command: + - sh + - -c + - pgrep -f "orchestrator.worker.worker" + initialDelaySeconds: 10 + periodSeconds: 15 + + +# 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: {} + +service: + enabled: false + type: ClusterIP + port: 4840 + targetPort: 4840 + + +env: + # Entrypoint variables + - name: GITHUB_REPO_URL + value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" + - name: GITHUB_BRANCH + value: "SIENTIAPDE-1030-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas" + - name: PYTHON_APP + value: "orchestrator.worker.worker" + + # Application variables + - name: REDIS_HOST + value: "redis-master.redis.svc.cluster.local" + - name: REDIS_PORT + value: "6379" + - name: REDIS_USERNAME + valueFrom: + secretKeyRef: + name: redis + key: redis-username + - name: REDIS_PASSWORD + valueFrom: + secretKeyRef: + name: redis + key: redis-password + + - name: COUCHBASE_CONNECTION_STRING + value: "couchbase://sientia.couchbase.svc.cluster.local" + - name: COUCHBASE_USERNAME + value: "sientia" + - name: COUCHBASE_PASSWORD + value: "sientia" + + - name: KAFKA_BOOTSTRAP_SERVERS + value: "kafka.kafka.svc.cluster.local:9092" + + - name: LOG_LEVEL + value: "DEBUG" + - name: PROJECT_NAME + value: "sientia-orchestrator" + + - name: TEMPORAL_HOST + value: "temporal-frontend.temporal.svc.cluster.local:7233" + - name: TEMPORAL_NAMESPACE + value: "default" + +ssh: + enabled: true + secretName: git-ssh-key-sientia-orchestrator-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-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.1.0-uat + +# kubectl create secret generic git-ssh-key-sientia-orchestrator-worker \ +# --namespace sientia \ +# --from-file=ssh-privatekey=git_key \ +# --type=kubernetes.io/ssh-auth \ No newline at end of file