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

SIENTIAPDE-1974

Browse Source 
SIENTIAPDE-1974-connectors-pi-web-api: Consolidate OPC server and tag data sourcing to the tag database. This change removes the legacy opc-servers collection and enforces tag_id as the sole reference for all read and write tags via connectors and connector_tags. The Orchestrator workflow now exclusively loads OPC server and tag information from the tag database, simplifying data flow and removing conditional logic for legacy data sources. Improved error reporting for pipeline timestamp updates and updated documentation, examples, and tests are also included.
This commit is contained in:
Vitor Pimentel dos Santos
2026-10-29 10:25:51 -03:00
parent 9ba2fc74c4
commit 0c9eaab3f5
30 changed files with 1082 additions and 3051 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
orchestrator/activities/formatters.py
View File

@@ -45,11 +45,6 @@ CONFIG_HASH_EXCLUDED_FIELDS = ('updated_at', 'config_hash')
# Read tags left out of the slots, by gather_read_tags issue kind
SLOT_ISSUES = (
(
'servers',
'ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
'Servers {} not found in opc_servers',
),
(
'tags',
'ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION',
@@ -63,9 +58,8 @@ SLOT_ISSUES = (
(
'invalid',
'ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION',
'Read tags {} have no integer frequency, an active_read without an integer '
'cycle_multiple, or lack server_id, tag_address or tag_name; or their pipeline lacks '
'schedule_name, model_id or model.name',
'Read tags {} have no integer tag_id, no integer frequency or an active_read without an '
'integer cycle_multiple; or their pipeline lacks schedule_name, model_id or model.name',
),
)
@@ -86,7 +80,7 @@ WRITE_TAG_ISSUES = (
(
'invalid',
'ORCHESTRATOR_INVALID_WRITE_TAG_DURING_SCHEDULE_CONFIGURATION',
'Write tags {} have no tag_id or a data_type the OPC Gateway does not accept',
'Write tags {} have no integer tag_id or a data_type the OPC Gateway does not accept',
),
)
@@ -335,18 +329,15 @@ class Formatters(SientiaMonitoring):
The division is by address: all the names of an address stay in the same slot
(SIENTIAPDE-1974, O5). Read tags that cannot be resolved are left out and
notified once per run: a tag_id missing from connector_tags, a connector missing
or inactive, a server_id missing from opc_servers, or a read tag without a usable
frequency (or, without tag_id, without server_id, tag_address or tag_name).
notified once per run: a read tag without tag_id or a usable frequency, a tag_id
missing from connector_tags, or a connector missing or inactive.
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[dict[str, Any]]): The OPC servers (opc-servers), for
read tags without tag_id.
- connectors (list[dict[str, Any]], optional): Active OPC UA connectors of
the tag database, for read tags with tag_id.
the tag database, the only source of the OPC servers.
- connector_tags (list[dict[str, Any]], optional): The connector_tags the
pipelines reference; only the active Read tags are used.
- active_ingestors (list[str]): The active ingestors to divide into slots.
@@ -361,11 +352,10 @@ class Formatters(SientiaMonitoring):
pipelines = input_data['pipelines']
active_ingestors = input_data['active_ingestors']
opc_servers = {str(server['id']): server for server in input_data['opc_servers']}
connectors = index_connectors(input_data.get('connectors', []))
tags_by_id = index_connector_tags(input_data.get('connector_tags', []))
entries, issues = gather_read_tags(pipelines, opc_servers, connectors, tags_by_id)
entries, issues = gather_read_tags(pipelines, connectors, tags_by_id)
self._notify_issues(issues, SLOT_ISSUES, metadata)
number_of_slots = len(active_ingestors) if active_ingestors else 1
@@ -484,10 +474,6 @@ class Formatters(SientiaMonitoring):
update_timestamp = schedule.get('updated_at', now())
current = current_schedules[schedule_name]
if not isinstance(current, dict):
# A run started before config_hash replays format_schedule_config's
# result, which held only the timestamp
current = {'updated_at': current, 'config_hash': None}
old_timestamp = current['updated_at']
self.debug(

25
orchestrator/activities/mongo_db.py
View File

@@ -247,9 +247,13 @@ class MongoDB(SientiaMonitoring):
self.info('No successful pipeline updates to timestamp', metadata=metadata)
return
try:
# One update per schedule: each has its own config_hash
for pipeline in succeeded:
# One update per schedule (each has its own config_hash). A failure does not stop
# the others: they are all tried, and the failures are notified together.
failures: dict[str, str] = {}
traces: list[str] = []
for pipeline in succeeded:
schedule = f'{pipeline["namespace"]}/{pipeline["schedule_name"]}'
try:
self.mongo_db_repository.update_one(
'orchestrated_schedules',
{
@@ -264,18 +268,25 @@ class MongoDB(SientiaMonitoring):
},
metadata,
)
except Exception as e:
trace = traceback.format_exc()
except Exception as e:
failures[schedule] = str(e)
traces.append(f'{schedule}\n{traceback.format_exc()}')
if failures:
trace = '\n'.join(traces)
details = '; '.join(f'{schedule}: {error}' for schedule, error in failures.items())
self.send_notification(
metadata=metadata,
notification_id='MONGODB_UPDATE_PIPELINES_ERROR',
message=f'Failed to update pipelines timestamps: {e}',
message=f'Failed to update pipelines timestamps: {details}',
block='update_pipelines_timestamps',
level=NotificationLevel.ERROR,
attachment_content=trace,
)
self.error(trace, metadata=metadata)
raise e
raise RuntimeError(
f'Failed to update {len(failures)} of {len(succeeded)} pipelines timestamps'
)
self.info(
f'Updated {len(updated_pipelines)} pipelines timestamps',

128
orchestrator/utils/opc_slots.py
View File

@@ -2,7 +2,7 @@
from typing import Any, cast
from orchestrator.utils.tag_database import is_complete, resolve_tag_names, tag_id_text
from orchestrator.utils.tag_database import is_complete, resolve_tag_names, tag_id_of
def _connector_server(connector: dict[str, Any]) -> dict[str, Any]:
@@ -14,26 +14,6 @@ def _connector_server(connector: dict[str, Any]) -> dict[str, Any]:
'server_uri': connector['server_uri'],
'use_security': bool(connector.get('use_security', False)),
'subscription_period_ms': connector.get('subscription_period_ms'),
'from_tag_database': True,
}
def _opc_server(server_id: str, opc_server: dict[str, Any]) -> dict[str, Any]:
"""
Slot server fields of a legacy opc-servers document.
The certificate now always comes from MinIO on the gateway side, so the file paths
are not carried: a cert_path only tells that the server uses security (RF4). The
subscription period is left to build_tag_config (half the lowest tag frequency).
"""
return {
'server_id': server_id,
'name': opc_server['server_name'],
'url': opc_server['url'],
'server_uri': opc_server['uri'],
'use_security': bool(opc_server.get('cert_path')),
'subscription_period_ms': None,
'from_tag_database': False,
}
@@ -46,39 +26,27 @@ def _frequency(tag: dict[str, Any]) -> int | None:
def _locate_read_tag(
tag: dict[str, Any],
opc_servers: dict[str, dict[str, Any]],
tag_id: int,
connectors: dict[str, dict[str, Any]],
tags_by_id: dict[str, dict[str, Any]],
tags_by_id: dict[int, dict[str, Any]],
issues: dict[str, set[str]],
) -> tuple[dict[str, Any], str] | None:
"""
Server and address of a read tag, or None when they cannot be resolved.
A tag with tag_id takes its address and connector from connector_tags and its server
from connectors (RF3). A tag without one keeps the current fields: server_id +
tag_address, with the server from opc-servers (RF4). What cannot be resolved is
recorded in issues ('tags', 'connectors' or 'servers').
The address and connector come from connector_tags and the server from connectors
(RF3). What cannot be resolved is recorded in issues ('tags' or 'connectors').
"""
tag_id = tag_id_text(tag)
if tag_id is not None:
connector_tag = tags_by_id.get(tag_id)
if connector_tag is None:
issues['tags'].add(tag_id)
return None
connector_id = str(connector_tag['connector_id'])
connector = connectors.get(connector_id)
if connector is None:
issues['connectors'].add(connector_id)
return None
return _connector_server(connector), connector_tag['address']
server_id = str(tag['server_id'])
opc_server = opc_servers.get(server_id)
if opc_server is None:
issues['servers'].add(server_id)
connector_tag = tags_by_id.get(tag_id)
if connector_tag is None:
issues['tags'].add(str(tag_id))
return None
return _opc_server(server_id, opc_server), tag['tag_address']
connector_id = str(connector_tag['connector_id'])
connector = connectors.get(connector_id)
if connector is None:
issues['connectors'].add(connector_id)
return None
return _connector_server(connector), connector_tag['address']
def _cycle_multiple(active_read: Any) -> int | None:
@@ -91,18 +59,22 @@ def _cycle_multiple(active_read: Any) -> int | None:
def _invalid_read_tag(tag: dict[str, Any]) -> bool:
"""
A read tag the slot cannot use: no integer frequency, an enabled active_read without
an integer cycle_multiple, or (without tag_id) no server_id, tag_address or tag_name.
A read tag the slot cannot use: no integer frequency, or an enabled active_read
without an integer cycle_multiple.
"""
if _frequency(tag) is None:
return True
active_read = tag.get('active_read')
if isinstance(active_read, dict) and active_read.get('enabled'):
if _cycle_multiple(active_read) is None:
return True
if tag_id_text(tag) is not None:
return False
return not is_complete(tag, ('server_id', 'tag_address', 'tag_name'))
return (
isinstance(active_read, dict)
and bool(active_read.get('enabled'))
and _cycle_multiple(active_read) is None
)
def _read_tag_ref(tag: dict[str, Any]) -> Any:
"""How an invalid read tag is reported: its tag_id, or its tag_name without one."""
return tag.get('tag_id') if tag.get('tag_id') not in (None, '') else tag.get('tag_name')
def _is_slot_pipeline(pipeline: dict[str, Any]) -> bool:
@@ -151,9 +123,7 @@ def _add_read_tag(
) -> None:
"""Merge one resolved read tag into its (server, address) entry and its server."""
server_id = server['server_id']
known = servers.get(server_id)
if known is None or (server['from_tag_database'] and not known['from_tag_database']):
servers[server_id] = server
servers[server_id] = server
# int, so the slot does not depend on how each pipeline stored the number
frequency = cast(int, _frequency(tag))
@@ -181,30 +151,24 @@ def _add_read_tag(
def gather_read_tags(
pipelines: list[dict[str, Any]],
opc_servers: dict[str, dict[str, Any]],
connectors: dict[str, dict[str, Any]],
tags_by_id: dict[str, dict[str, Any]],
tags_by_id: dict[int, dict[str, Any]],
) -> tuple[list[dict[str, Any]], dict[str, set[str]]]:
"""
Group the read tags of all scouter pipelines by (server, address) for the slot v2.
An address read by N pipelines becomes ONE entry with N items in names[]: the
gateway reads it once and writes it to N collections (SIENTIAPDE-1974, RF2). Each
read tag is resolved through the tag database when it has a tag_id (RF3), or through
the current server_id + tag_address + opc-servers fields when it has not (RF4).
A server known through both connectors and opc-servers under the same server_id is
the same server (the gateway keys its connections by server_id); the connectors
fields win.
read tag is resolved through the tag database by its tag_id (RF3): connectors is the
only source of the OPC servers.
Args:
pipelines (list[dict[str, Any]]): Pipeline configurations. Only workflow_type
'scouter' is read; each needs schedule_name, model_id, model.name and
read_tags (with frequency, and optionally tag_id and active_read)
opc_servers (dict[str, dict[str, Any]]): opc-servers documents keyed by str(id)
read_tags (with tag_id and frequency, and optionally active_read)
connectors (dict[str, dict[str, Any]]): Active OPC UA connectors indexed by
index_connectors
tags_by_id (dict[str, dict[str, Any]]): Active Read connector_tags indexed by
tags_by_id (dict[int, dict[str, Any]]): Active Read connector_tags indexed by
index_connector_tags
Returns:
@@ -218,15 +182,14 @@ def gather_read_tags(
- active_read (dict): {'enabled': False} unless a pipeline enables it (D3)
- names (list[dict]): collection ('raw_{schedule_name}'), tag_name,
model_id and model_name, one per pipeline name, sorted
- What was left out: 'tags' (tag_id), 'connectors' (connector_id), 'servers'
(opc-servers server_id) and 'invalid' (schedule_name/read tag without a
usable frequency, cycle_multiple or legacy fields, or the schedule_name of a
pipeline without schedule_name, model_id or model.name)
- What was left out: 'tags' (tag_id), 'connectors' (connector_id) and
'invalid' (schedule_name/read tag without an int tag_id, a usable frequency or
cycle_multiple, or the schedule_name of a pipeline without schedule_name,
model_id or model.name)
"""
issues: dict[str, set[str]] = {
'tags': set(),
'connectors': set(),
'servers': set(),
'invalid': set(),
}
servers: dict[str, dict[str, Any]] = {}
@@ -240,12 +203,12 @@ def gather_read_tags(
continue
for tag in resolve_tag_names(pipeline.get('read_tags') or [], tags_by_id):
if _invalid_read_tag(tag):
ref = tag_id_text(tag) or tag.get('tag_address') or tag.get('tag_name')
issues['invalid'].add(f'{pipeline["schedule_name"]}/{ref}')
tag_id = tag_id_of(tag)
if tag_id is None or _invalid_read_tag(tag):
issues['invalid'].add(f'{pipeline["schedule_name"]}/{_read_tag_ref(tag)}')
continue
located = _locate_read_tag(tag, opc_servers, connectors, tags_by_id, issues)
located = _locate_read_tag(tag_id, connectors, tags_by_id, issues)
if located is None:
continue
_add_read_tag(entries, servers, pipeline, tag, *located)
@@ -264,10 +227,8 @@ def gather_read_tags(
def build_tag_config(entries: list[dict[str, Any]]) -> dict[str, Any]:
"""
Build one slot in the slot v2 format (contract C1) from address entries.
The slot no longer carries cert_path, private_key_path or server_cert_path: the
gateway takes the certificate from MinIO (SIENTIAPDE-1974, O4).
Build one slot in the slot v2 format (contract C1) from address entries. The slot
carries no certificate: the gateway takes it from MinIO (SIENTIAPDE-1974, O4).
Args:
entries (list[dict[str, Any]]): Address entries from gather_read_tags for this slot
@@ -275,9 +236,8 @@ def build_tag_config(entries: list[dict[str, Any]]) -> dict[str, Any]:
Returns:
dict[str, Any]: The slot keyed by server key, each server with server_id, name,
url, server_uri, use_security, subscription_period_ms and tags. Each tag (keyed
by address) has frequency, active_read and names. A server without a
subscription period of its own (opc-servers) gets half its lowest tag
frequency in this slot, as before.
by address) has frequency, active_read and names. A connector without a
subscription_period_ms gets half its lowest tag frequency in this slot.
"""
slot_config: dict[str, Any] = {}

79
orchestrator/utils/tag_database.py
View File

@@ -36,24 +36,15 @@ def is_complete(document: dict[str, Any], fields: tuple[str, ...]) -> bool:
return all(document.get(field) not in (None, '') for field in fields)
def _is_integer_text(text: str) -> bool:
"""True for ASCII decimal text with an optional sign ('7', '007', '-3'), never raising."""
digits = text[1:] if text.startswith('-') else text
return digits.isascii() and digits.isdecimal()
def tag_id_text(tag: dict[str, Any]) -> str | None:
def tag_id_of(tag: Any) -> int | None:
"""
The tag's tag_id as text, or None when it has none (missing, null or empty).
The tag's tag_id, or None when it is not an integer.
Integer ids are canonical ('007' and 7 both give '7'), so a pipeline and the tag
database always agree on the key whether they store the id as int or text.
The connector_tags validator stores tag_id as int, so a pipeline references a tag
by the same int: text ('7'), bool, null or a missing tag_id is no reference.
"""
tag_id = tag.get('tag_id')
if tag_id in (None, ''):
return None
text = str(tag_id).strip()
return str(int(text)) if _is_integer_text(text) else text
tag_id = tag.get('tag_id') if isinstance(tag, dict) else None
return tag_id if isinstance(tag_id, int) and not isinstance(tag_id, bool) else None
def _is_active(document: dict[str, Any]) -> bool:
@@ -74,40 +65,31 @@ def pipeline_tag_refs(pipeline: dict[str, Any]) -> list[Any]:
return []
def referenced_tag_ids(pipelines: list[dict[str, Any]]) -> list[Any]:
def referenced_tag_ids(pipelines: list[dict[str, Any]]) -> list[int]:
"""
The tag_ids the pipelines reference (scouter read_tags and predictions_batch
write_tags), for the connector_tags query.
Each id goes in both forms, int and text, because the tag database stores int and
pipelines may hold either. Sorted, so the query is the same on every run. Runs in
the workflow, so it never raises on malformed references.
Sorted, so the query is the same on every run. Runs in the workflow, so it never
raises on malformed references.
Args:
pipelines (list[dict[str, Any]]): Pipeline configurations
Returns:
list[Any]: The referenced tag_ids (empty when no tag carries a tag_id)
list[int]: The referenced tag_ids (empty when no tag carries a tag_id)
"""
ids: set[Any] = set()
for pipeline in pipelines:
for tag in pipeline_tag_refs(pipeline):
tag_id = tag_id_text(tag) if isinstance(tag, dict) else None
if tag_id is None:
continue
ids.add(tag_id)
if _is_integer_text(tag_id):
ids.add(int(tag_id))
return sorted(ids, key=lambda value: (isinstance(value, str), str(value)))
ids = {tag_id_of(tag) for pipeline in pipelines for tag in pipeline_tag_refs(pipeline)}
return sorted(tag_id for tag_id in ids if tag_id is not None)
def filter_by_tag_ids(query: dict[str, Any], tag_ids: list[Any]) -> dict[str, Any]:
def filter_by_tag_ids(query: dict[str, Any], tag_ids: list[int]) -> dict[str, Any]:
"""
Restrict a connector_tags aggregation to the referenced tag_ids.
Args:
query (dict[str, Any]): The connector_tags query ({'collection', 'aggregation'})
tag_ids (list[Any]): The ids from referenced_tag_ids
tag_ids (list[int]): The ids from referenced_tag_ids
Returns:
dict[str, Any]: A new query whose aggregation starts with a tag_id $match. The
@@ -120,14 +102,14 @@ def filter_by_tag_ids(query: dict[str, Any], tag_ids: list[Any]) -> dict[str, An
}
def index_tag_documents(connector_tags: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
def index_tag_documents(connector_tags: list[dict[str, Any]]) -> dict[int, dict[str, Any]]:
"""
Every loaded connector_tags document keyed by tag_id text, whatever its status,
category or completeness.
Every loaded connector_tags document keyed by tag_id, whatever its status, category
or completeness.
"""
indexed = {}
for tag in connector_tags:
tag_id = tag_id_text(tag)
tag_id = tag_id_of(tag)
if tag_id is not None:
indexed[tag_id] = tag
return indexed
@@ -135,12 +117,11 @@ def index_tag_documents(connector_tags: list[dict[str, Any]]) -> dict[str, dict[
def index_connector_tags(
connector_tags: list[dict[str, Any]], categories: tuple[str, ...] = READ_CATEGORIES
) -> dict[str, dict[str, Any]]:
) -> dict[int, dict[str, Any]]:
"""
Index the tag database's usable connector_tags by tag_id.
The key is the tag_id as text (tag_id_text), because pipelines may store it as int
or str. Documents that are not active, have another category, or lack tag_id,
Documents that are not active, have another category, or lack an int tag_id,
connector_id, address or tag_name are left out.
Args:
@@ -150,7 +131,7 @@ def index_connector_tags(
without category is accepted.
Returns:
dict[str, dict[str, Any]]: connector_tags documents keyed by tag_id text
dict[int, dict[str, Any]]: connector_tags documents keyed by tag_id
"""
return {
tag_id: tag
@@ -182,7 +163,7 @@ def index_connectors(connectors: list[dict[str, Any]]) -> dict[str, dict[str, An
def resolve_tag_names(
read_tags: list[dict[str, Any]], tags_by_id: dict[str, dict[str, Any]]
read_tags: list[dict[str, Any]], tags_by_id: dict[int, dict[str, Any]]
) -> list[dict[str, Any]]:
"""
Fill in the tag_name of each read tag of a scouter pipeline.
@@ -194,7 +175,7 @@ def resolve_tag_names(
Args:
read_tags (list[dict[str, Any]]): The pipeline's read_tags
tags_by_id (dict[str, dict[str, Any]]): connector_tags indexed by index_connector_tags
tags_by_id (dict[int, dict[str, Any]]): connector_tags indexed by index_connector_tags
Returns:
list[dict[str, Any]]: The read tags, each with tag_name set when it could be
@@ -203,7 +184,7 @@ def resolve_tag_names(
resolved = []
for tag in read_tags:
name = tag.get('tag_name')
tag_id = tag_id_text(tag)
tag_id = tag_id_of(tag)
if not name and tag_id is not None:
name = tags_by_id.get(tag_id, {}).get('tag_name')
resolved.append({**tag, 'tag_name': name} if name else tag)
@@ -212,7 +193,7 @@ def resolve_tag_names(
def resolve_write_tags(
write_tags: list[Any],
tags_by_id: dict[str, dict[str, Any]],
tags_by_id: dict[int, dict[str, Any]],
connectors: dict[str, dict[str, Any]],
) -> tuple[dict[str, Any], dict[str, set[str]]]:
"""
@@ -226,7 +207,7 @@ def resolve_write_tags(
Args:
write_tags (list[Any]): The pipeline's write_tags ([{tag_id}])
tags_by_id (dict[str, dict[str, Any]]): Write connector_tags indexed by
tags_by_id (dict[int, dict[str, Any]]): Write connector_tags indexed by
index_connector_tags(..., WRITE_CATEGORIES)
connectors (dict[str, dict[str, Any]]): Active OPC UA connectors indexed by
index_connectors
@@ -235,20 +216,20 @@ def resolve_write_tags(
tuple: the opc_output_config ({connector_id: {prediction_tags | confidence_tags:
{address: {data_type}}}}) and what was left out, by kind: 'tags' (tag_id not
an active, complete Write / Prediction Confidence tag), 'connectors'
(connector_id) and 'invalid' (an entry without tag_id, or a tag whose
(connector_id) and 'invalid' (an entry without an int tag_id, or a tag whose
data_type the gateway does not accept)
"""
issues: dict[str, set[str]] = {'tags': set(), 'connectors': set(), 'invalid': set()}
output_config: dict[str, Any] = {}
for entry in write_tags:
tag_id = tag_id_text(entry) if isinstance(entry, dict) else None
tag_id = tag_id_of(entry)
if tag_id is None:
issues['invalid'].add(str(entry))
continue
connector_tag = tags_by_id.get(tag_id)
# The category is what tells a prediction from a confidence: it cannot be missing
if connector_tag is None or connector_tag.get('category') not in WRITE_TAG_GROUPS:
issues['tags'].add(tag_id)
issues['tags'].add(str(tag_id))
continue
connector_id = str(connector_tag['connector_id'])
if connector_id not in connectors:
@@ -256,7 +237,7 @@ def resolve_write_tags(
continue
data_type = connector_tag.get('data_type') or DEFAULT_DATA_TYPE
if data_type not in OPC_DATA_TYPES:
issues['invalid'].add(tag_id)
issues['invalid'].add(str(tag_id))
continue
group = WRITE_TAG_GROUPS[connector_tag['category']]
output_config.setdefault(connector_id, {}).setdefault(group, {})[

34
orchestrator/workflows/orchestrator.py
View File

@@ -57,8 +57,6 @@ DEFAULT_CONNECTOR_TAGS_QUERY = {
},
],
}
# Runs started before the tag database loads existed replay without them
TAG_DATABASE_PATCH = 'SIENTIAPDE-1974-tag-database'
@workflow.defn(name='orchestrator')
@@ -79,14 +77,12 @@ class Orchestrator:
pipelines: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""
Load the tag database (SIENTIAPDE-1974) for the read and write tags that carry a
tag_id.
Load the tag database (SIENTIAPDE-1974): the only source of the OPC servers and
tags of the read and write tags.
Only the referenced tags are loaded, with the fields used, so an environment
without a tag database never depends on its content (P4) and the payload stays
Only the referenced tags are loaded, with the fields used, so the payload stays
bounded by the pipelines. Nothing is loaded when no scouter read tag nor
predictions_batch write tag has a tag_id, or when replaying a run started before
TAG_DATABASE_PATCH.
predictions_batch write tag has a tag_id.
Args:
input_data (dict[str, Any]): Workflow input (optional connectors_query and
@@ -98,8 +94,7 @@ class Orchestrator:
tuple[list[dict[str, Any]], list[dict[str, Any]]]: connectors and connector_tags
"""
tag_ids = referenced_tag_ids(pipelines)
# patched() first: it must run on every execution to record its marker
if not workflow.patched(TAG_DATABASE_PATCH) or not tag_ids:
if not tag_ids:
return [], []
return await self._load_connectors(
@@ -145,9 +140,9 @@ class Orchestrator:
"""
Execute the orchestration workflow for pipeline and slot management.
This workflow retrieves pipeline configurations and OPC server data,
processes schedules and slot configurations, and deploys them to
the appropriate services. It handles creation, updates, and deletion
This workflow retrieves pipeline configurations and the tag database
(connectors and connector_tags), processes schedules and slot configurations,
and deploys them to the appropriate services. It handles creation, updates, and deletion
of schedules and slots based on current system state.
Args:
@@ -155,9 +150,7 @@ class Orchestrator:
Required fields:
- schedule_name (str): Name of the orchestration schedule
- pipelines_query (dict[str, Any]): MongoDB query for pipeline configurations
- opc_servers_query (dict[str, Any]): MongoDB query for OPC server data
Optional fields (used only when a scouter read tag or a predictions_batch
write tag carries a tag_id):
Optional fields:
- connectors_query (dict[str, Any]): MongoDB aggregation for the tag
database connectors (default: DEFAULT_CONNECTORS_QUERY)
- connector_tags_query (dict[str, Any]): MongoDB aggregation for the tag
@@ -192,13 +185,6 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60),
)
opc_servers_handler = workflow.start_local_activity_method(
Activities.find_documents_in_mongodb,
{**metadata, 'query': input_data['opc_servers_query']},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60),
)
orchestrated_schedules_handler = workflow.start_local_activity_method(
Activities.find_documents_in_mongodb,
{
@@ -236,7 +222,6 @@ class Orchestrator:
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
formatted_orchestrated_schedules_handler = workflow.start_local_activity_method(
@@ -262,7 +247,6 @@ class Orchestrator:
Activities.process_slots,
{
**metadata,
'opc_servers': opc_servers,
'connectors': connectors,
'connector_tags': connector_tags,
'active_ingestors': active_ingestors,