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 
Resolve PI Web API pipeline tags through the tag database

- Load active `pi_web_api` connectors and referenced tags behind the
  `connectors-pi-web-api` patch, so replays of older runs are unaffected.
- Resolve `pi_web_api_scouter` read tags and `predictions_batch` output
  tags by `tag_id`; legacy `webid` pipelines produce identical input.
- Add a `connection` block to scouter schedules with the secret reference
  only, plus connector defaults for period, max_count and api_timeout.
- Hold pipelines whose tags span several connectors and notify unresolved
  tags, wrong categories and missing connectors once per run.
- Share the issue notification logic between slots and schedules.
This commit is contained in:
Vitor Pimentel dos Santos
2026-10-28 15:46:09 -03:00
parent 744ca647f7
commit 7e6f1a8782
7 changed files with 1170 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
orchestrator/activities/formatters.py
View File

@@ -18,6 +18,7 @@ with workflow.unsafe.imports_passed_through():
)
from orchestrator.utils.orchestrator_functions import (
PI_OUTPUT_WORKFLOW_TYPES,
build_tag_config,
chunk_period_warning,
drift,
@@ -25,9 +26,12 @@ with workflow.unsafe.imports_passed_through():
gather_read_tags,
index_connector_tags,
index_connectors,
index_pi_connectors,
minimal_retrain,
pi_web_api_scouter,
predictions_batch,
resolve_pi_output_tags,
resolve_pi_read_tags,
resolve_tag_names,
scouter,
simple_metrics,
@@ -35,6 +39,65 @@ with workflow.unsafe.imports_passed_through():
topic_separator = '\n ========== \n'
# 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',
'Tags {} not found among the active, complete Read tags of connector_tags',
),
(
'connectors',
'ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION',
'Connectors {} not found, not active or incomplete in connectors',
),
(
'invalid',
'ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION',
'Read tags {} have no integer frequency or lack server_id, tag_address or tag_name',
),
)
# PI Web API tags left out of the schedules, by resolve_pi_read_tags / resolve_pi_output_tags
# issue kind; values are schedule_name/<tag_id or connector_id>
PI_SCHEDULE_ISSUES = (
(
'tags',
'ORCHESTRATOR_PI_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION',
'PI Web API tags {} not found among the active, complete tags of connector_tags',
),
(
'categories',
'ORCHESTRATOR_PI_TAG_WRONG_CATEGORY_DURING_SCHEDULE_CONFIGURATION',
'PI Web API tags {} have a category the pipeline cannot use (Read for read_tags, '
'Write for prediction_tags, Prediction Confidence for confidence_tags)',
),
(
'connectors',
'ORCHESTRATOR_PI_CONNECTOR_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION',
'PI Web API connectors {} not found, not active or incomplete in connectors',
),
(
'mixed',
'ORCHESTRATOR_PI_MIXED_CONNECTORS_DURING_SCHEDULE_CONFIGURATION',
'Pipelines {} read tags of several PI Web API connectors; their schedules were left '
'untouched',
),
)
def _merge_issues(
issues: dict[str, set[str]], pipeline_issues: dict[str, set[str]], schedule_name: str
) -> None:
"""Add one pipeline's issues to the run's, each value prefixed with schedule_name/."""
for kind, values in pipeline_issues.items():
issues.setdefault(kind, set()).update(f'{schedule_name}/{value}' for value in values)
class ScheduleType(TypedDict):
"""
@@ -180,6 +243,16 @@ class Formatters(SientiaMonitoring):
- pipelines (list[dict[str, Any]]): The schedules to process.
- connector_tags (list[dict[str, Any]], optional): Active Read tags of
the tag database, to name the scouter read tags that carry only a tag_id.
- pi_connectors (list[dict[str, Any]], optional): Active pi_web_api
connectors of the tag database.
- pi_connector_tags (list[dict[str, Any]], optional): Active tags of the
tag database referenced by the PI Web API pipelines (all categories).
PI Web API tags given by tag_id (pi_web_api_scouter read tags, predictions_batch
pi_web_api_output_config) are resolved through the tag database; what cannot be
resolved is left out and notified once per run. A pi_web_api_scouter whose tags
span several connectors is held: it gets {'held': True} instead of a config, so
create_schedule_config neither creates, updates nor deletes its schedule.
Returns:
- dict[str, Any]: The schedule configuration dictionary keyed by namespace
@@ -191,6 +264,9 @@ class Formatters(SientiaMonitoring):
pipelines = input_data['pipelines']
tags_by_id = index_connector_tags(input_data.get('connector_tags', []))
pi_connectors = index_pi_connectors(input_data.get('pi_connectors', []))
pi_tags = index_connector_tags(input_data.get('pi_connector_tags', []))
pi_issues: dict[str, set[str]] = {}
schedule_config: dict[str, dict[str, Any]] = {
self.scouter_namespace: {},
@@ -204,6 +280,11 @@ class Formatters(SientiaMonitoring):
self.error(f'Workflow type {workflow_type} not supported', metadata=metadata)
continue
schedule_type = schedule_types[workflow_type]
namespace = schedule_type['namespace']
function = schedule_type['function']
schedule_name = pipeline['schedule_name']
if workflow_type == 'scouter' and 'read_tags' in pipeline:
# Same names as the slot's names[] (process_slots): SIENTIAPDE-1974 SC1
pipeline = {
@@ -211,17 +292,38 @@ class Formatters(SientiaMonitoring):
'read_tags': resolve_tag_names(pipeline['read_tags'], tags_by_id),
}
schedule_type = schedule_types[workflow_type]
namespace = schedule_type['namespace']
function = schedule_type['function']
if workflow_type == 'pi_web_api_scouter' and isinstance(
pipeline.get('read_tags'), dict
):
read_tags, connector, issues = resolve_pi_read_tags(
pipeline['read_tags'], pi_tags, pi_connectors
)
_merge_issues(pi_issues, issues, schedule_name)
if issues['mixed']:
schedule_config[namespace][schedule_name] = {'held': True}
continue
pipeline = {**pipeline, 'read_tags': read_tags}
if connector is not None:
pipeline['pi_web_api_connector'] = connector
schedule_config[namespace][pipeline['schedule_name']] = {
if workflow_type in PI_OUTPUT_WORKFLOW_TYPES and isinstance(
pipeline.get('pi_web_api_output_config'), dict
):
output_config, issues = resolve_pi_output_tags(
pipeline['pi_web_api_output_config'], pi_tags, pi_connectors
)
_merge_issues(pi_issues, issues, schedule_name)
pipeline = {**pipeline, 'pi_web_api_output_config': output_config}
schedule_config[namespace][schedule_name] = {
**function(pipeline),
'updated_at': pipeline.get(
'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
),
}
self._notify_issues(pi_issues, PI_SCHEDULE_ISSUES, metadata)
self.info('Processed schedules', metadata=metadata)
self.debug(json.dumps(schedule_config, indent=4, sort_keys=True), metadata=metadata)
@@ -267,7 +369,7 @@ class Formatters(SientiaMonitoring):
tags_by_id = index_connector_tags(input_data.get('connector_tags', []))
entries, issues = gather_read_tags(pipelines, opc_servers, connectors, tags_by_id)
self._notify_slot_issues(issues, metadata)
self._notify_issues(issues, SLOT_ISSUES, metadata)
number_of_slots = len(active_ingestors) if active_ingestors else 1
entries_per_slot = ceil(len(entries) / number_of_slots)
@@ -282,38 +384,21 @@ class Formatters(SientiaMonitoring):
return slot_config
def _notify_slot_issues(self, issues: dict[str, set[str]], metadata: dict[str, Any]) -> None:
def _notify_issues(
self,
issues: dict[str, set[str]],
kinds: tuple[tuple[str, str, str], ...],
metadata: dict[str, Any],
) -> None:
"""
Notify the read tags left out of the slots, one notification per kind of issue.
Notify what was left out of the slots or schedules, one notification per kind.
Args:
issues (dict[str, set[str]]): What gather_read_tags left out, by kind:
'servers' (opc-servers server_id), 'tags' (tag_id), 'connectors'
(connector_id) and 'invalid' (schedule_name/read tag)
issues (dict[str, set[str]]): What was left out, by kind
kinds (tuple[tuple[str, str, str], ...]): (kind, notification_id, message with
one {} for the joined values), e.g. SLOT_ISSUES or PI_SCHEDULE_ISSUES
metadata (dict[str, Any]): Metadata for the notifications
"""
kinds = (
(
'servers',
'ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
'Servers {} not found in opc_servers',
),
(
'tags',
'ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION',
'Tags {} not found among the active, complete Read tags of connector_tags',
),
(
'connectors',
'ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION',
'Connectors {} not found, not active or incomplete in connectors',
),
(
'invalid',
'ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION',
'Read tags {} have no integer frequency or lack server_id, tag_address or tag_name',
),
)
for kind, notification_id, message in kinds:
if issues.get(kind):
self.send_notification(
@@ -417,7 +502,8 @@ class Formatters(SientiaMonitoring):
Args:
input_data (dict[str, Any]): The input data containing:
- current_schedule_config (dict[str, Any]): The current schedule config in Temporal server
- schedule_config (dict[str, Any]): The schedule config to process
- schedule_config (dict[str, Any]): The schedule config to process; an
entry {'held': True} keeps that schedule exactly as it is
- metadata (dict): Metadata for logging purposes
Returns:
@@ -449,6 +535,9 @@ class Formatters(SientiaMonitoring):
for namespace, schedules in schedule_config.items():
current_schedules = current_schedule_config.get(namespace, {})
# A held schedule (process_schedules) is neither created nor updated; it stays
# in schedule_config, so it is not deleted either
schedules = {name: s for name, s in schedules.items() if not s.get('held')}
self.compare_config_timestamps(
schedules, current_schedules, to_update, to_create, namespace, metadata
)

268
orchestrator/utils/orchestrator_functions.py
View File

@@ -304,21 +304,26 @@ def pi_web_api_scouter(config: dict[str, Any]):
Args:
config (dict[str, Any]): Pipeline configuration containing:
- read_tags (list[dict]): List of tag configurations with:
- tag_name (str): Name of the tag
- read_tags (dict[str, dict]): Tag configurations keyed by model tag name, already
passed through resolve_pi_read_tags (process_schedules does it), with:
- webid (str): PI Web API WebID for the tag
- aggr_func (str, optional): Aggregation function (default: 'lts')
- data_range (list[int], optional): Valid data range (default: [-100, 100])
- pi_web_api_config (dict): PI Web API connection settings with:
- pi_web_api_config (dict): PI Web API query settings with:
- endpoint (str): PI Web API endpoint URL
- period (str, optional): Time period for data retrieval (default: '*-1d')
- max_count (int, optional): Maximum number of values to retrieve (default: 1)
- api_timeout (int, optional): API request timeout in seconds
- pi_web_api_connector (dict, optional): The tag database connector of the
pipeline's resolved tags (set by process_schedules). Its pi_web_api period,
max_count and api_timeout are defaults under pi_web_api_config
- Additional fields from base_scouter
Returns:
dict[str, Any]: PI Web API scouter configuration with tag mappings and query settings.
API timeout is automatically adjusted to not exceed workflow frequency.
API timeout is automatically adjusted to not exceed workflow frequency. With a
connector, it also carries connection: {connector_id, base_url, auth} (auth holds
the secret reference only, never the credential).
"""
tags = {}
for tag, tag_config in config['read_tags'].items():
@@ -330,7 +335,18 @@ def pi_web_api_scouter(config: dict[str, Any]):
base_config = base_scouter(config)
connector = config.get('pi_web_api_connector')
pi_web_api_config = config['pi_web_api_config']
if connector:
connector_defaults = connector['pi_web_api']
pi_web_api_config = {
**{
key: connector_defaults[key]
for key in PI_QUERY_DEFAULTS
if key in connector_defaults
},
**pi_web_api_config,
}
config_timeout = pi_web_api_config.get('api_timeout', None)
frequency = parse_frequency(base_config['frequency'])
@@ -338,7 +354,7 @@ def pi_web_api_scouter(config: dict[str, Any]):
if config_timeout is None or config_timeout > frequency:
config_timeout = frequency
return {
output = {
**base_config,
'model_tags': tags,
'pi_web_api_query': {
@@ -348,6 +364,9 @@ def pi_web_api_scouter(config: dict[str, Any]):
'api_timeout': config_timeout,
},
}
if connector:
output['connection'] = _pi_connection(connector)
return output
def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[str, Any]]):
@@ -432,7 +451,8 @@ def predictions_batch(config: dict[str, Any]):
- model_retention_minutes (int, optional): Data retention time in minutes (default: 60)
- save_transform (bool, optional): Save transformed data to database (default: True)
- predictions_storage_policy (str, optional): Prediction storage policy (default: 'lts:1')
- pi_web_api_output_config (dict, optional): PI Web API output configuration for write-back (default: {})
- pi_web_api_output_config (dict, optional): PI Web API output configuration for write-back (default: {}),
already passed through resolve_pi_output_tags (process_schedules does it)
- Additional fields from common_config
Returns:
@@ -520,20 +540,65 @@ def referenced_tag_ids(pipelines: list[dict[str, Any]]) -> list[Any]:
Returns:
list[Any]: The referenced tag_ids (empty when no read tag carries a tag_id)
"""
tags = [
tag
for pipeline in pipelines
if pipeline.get('workflow_type') == 'scouter'
for tag in pipeline.get('read_tags') or []
]
return _query_tag_ids(tags)
def _query_tag_ids(tags: list[Any]) -> list[Any]:
"""The tag_ids of the tags in both forms (int and text), sorted for a stable query."""
ids: set[Any] = set()
for pipeline in pipelines:
if pipeline.get('workflow_type') != 'scouter':
for tag in tags:
tag_id = _tag_id(tag) if isinstance(tag, dict) else None
if tag_id is None:
continue
for tag in pipeline.get('read_tags') or []:
tag_id = _tag_id(tag)
if tag_id is None:
continue
ids.add(tag_id)
if tag_id.lstrip('-').isdigit():
ids.add(int(tag_id))
ids.add(tag_id)
if tag_id.lstrip('-').isdigit():
ids.add(int(tag_id))
return sorted(ids, key=lambda value: (isinstance(value, str), str(value)))
# Workflow types whose pi_web_api_output_config may reference the tag database
PI_OUTPUT_WORKFLOW_TYPES = ('predictions_batch', 'xgboost_predictions_batch')
# connector_tags category each PI Web API output mapping accepts
PI_OUTPUT_CATEGORIES = {'prediction_tags': 'Write', 'confidence_tags': 'Prediction Confidence'}
# Connection defaults a pi_web_api connector may carry for the scouter query
PI_QUERY_DEFAULTS = ('period', 'max_count', 'api_timeout')
def referenced_pi_tag_ids(pipelines: list[dict[str, Any]]) -> list[Any]:
"""
The tag_ids the PI Web API pipelines reference, for the PI connector_tags query.
Read tags of pi_web_api_scouter pipelines and the prediction_tags / confidence_tags
entries of a predictions_batch pi_web_api_output_config given as a list. Same forms
and order as referenced_tag_ids.
Args:
pipelines (list[dict[str, Any]]): Pipeline configurations
Returns:
list[Any]: The referenced tag_ids (empty when no PI Web API tag carries a tag_id)
"""
tags: list[Any] = []
for pipeline in pipelines:
workflow_type = pipeline.get('workflow_type')
if workflow_type == 'pi_web_api_scouter':
read_tags = pipeline.get('read_tags')
if isinstance(read_tags, dict):
tags.extend(read_tags.values())
elif workflow_type in PI_OUTPUT_WORKFLOW_TYPES:
output_config = pipeline.get('pi_web_api_output_config') or {}
for key in PI_OUTPUT_CATEGORIES:
if isinstance(output_config.get(key), list):
tags.extend(output_config[key])
return _query_tag_ids(tags)
def filter_by_tag_ids(query: dict[str, Any], tag_ids: list[Any]) -> dict[str, Any]:
"""
Restrict a connector_tags aggregation to the referenced tag_ids.
@@ -620,6 +685,179 @@ def resolve_tag_names(
return resolved
def _pi_auth(connector: dict[str, Any]) -> dict[str, Any] | None:
"""The connector's pi_web_api.auth when it names a type and a secret {name, key}."""
block = connector.get('pi_web_api')
auth = block.get('auth') if isinstance(block, dict) else None
if not isinstance(auth, dict) or auth.get('type') in (None, ''):
return None
secret = auth.get('secret')
if not isinstance(secret, dict) or not _complete(secret, ('name', 'key')):
return None
return auth
def index_pi_connectors(connectors: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
"""
Index the tag database's pi_web_api connectors by connector_id (as text).
Documents without connector_id, url or a pi_web_api.auth with type and secret
{name, key} are left out, so their tags are notified as connector not found.
Args:
connectors (list[dict[str, Any]]): Active pi_web_api connectors documents
Returns:
dict[str, dict[str, Any]]: connectors documents keyed by str(connector_id)
"""
return {
str(connector['connector_id']): connector
for connector in connectors
if _complete(connector, ('connector_id', 'url')) and _pi_auth(connector) is not None
}
def _pi_connection(connector: dict[str, Any]) -> dict[str, Any]:
"""
Connection block of a pi_web_api connector for the schedule input.
Only the secret reference is copied, field by field, so nothing else the document
may hold under auth ever reaches the schedule input or the Temporal history (D3).
"""
auth = cast(dict[str, Any], _pi_auth(connector))
connection = {
'connector_id': str(connector['connector_id']),
'base_url': connector['url'],
'auth': {
'type': auth['type'],
'secret': {'name': auth['secret']['name'], 'key': auth['secret']['key']},
},
}
data_server_webid = connector['pi_web_api'].get('data_server_webid')
if data_server_webid:
connection['data_server_webid'] = data_server_webid
return connection
def _new_pi_issues() -> dict[str, set[str]]:
return {'tags': set(), 'categories': set(), 'connectors': set(), 'mixed': set()}
def _locate_pi_tag(
tag_id: str,
category: str,
pi_tags: dict[str, dict[str, Any]],
pi_connectors: dict[str, dict[str, Any]],
issues: dict[str, set[str]],
) -> dict[str, Any] | None:
"""
The connector_tags document of a PI Web API tag_id, or None when it cannot be used:
unknown tag_id ('tags'), another category ('categories'), or a connector that is not
an active, complete pi_web_api connector ('connectors').
"""
connector_tag = pi_tags.get(tag_id)
if connector_tag is None:
issues['tags'].add(tag_id)
return None
if connector_tag.get('category') != category:
issues['categories'].add(tag_id)
return None
connector_id = str(connector_tag['connector_id'])
if connector_id not in pi_connectors:
issues['connectors'].add(connector_id)
return None
return connector_tag
def resolve_pi_read_tags(
read_tags: dict[str, dict[str, Any]],
pi_tags: dict[str, dict[str, Any]],
pi_connectors: dict[str, dict[str, Any]],
) -> tuple[dict[str, dict[str, Any]], dict[str, Any] | None, dict[str, set[str]]]:
"""
Resolve the read tags of a pi_web_api_scouter pipeline through the tag database.
An entry with tag_id gets its webid from connector_tags.address (Read category); its
key (the model tag name), aggr_func and data_range stay the pipeline's. An entry
without tag_id is kept as is (legacy webid). An entry that cannot be resolved is left
out. All the resolved tags must belong to one connector (D6).
Args:
read_tags (dict[str, dict[str, Any]]): The pipeline's read_tags
pi_tags (dict[str, dict[str, Any]]): PI connector_tags indexed by index_connector_tags
pi_connectors (dict[str, dict[str, Any]]): Indexed by index_pi_connectors
Returns:
tuple: the resolved read_tags; the connector of the resolved tags (None when no tag
was resolved through the tag database or they span several connectors); and
what was left out by kind: 'tags' / 'categories' (tag_id), 'connectors'
(connector_id) and 'mixed' (the connector_ids, joined, when there are several)
"""
issues = _new_pi_issues()
resolved: dict[str, dict[str, Any]] = {}
connector_ids: set[str] = set()
for name, tag in read_tags.items():
tag_id = _tag_id(tag)
if tag_id is None:
resolved[name] = tag
continue
connector_tag = _locate_pi_tag(tag_id, 'Read', pi_tags, pi_connectors, issues)
if connector_tag is None:
continue
connector_ids.add(str(connector_tag['connector_id']))
resolved[name] = {**tag, 'webid': connector_tag['address']}
if len(connector_ids) > 1:
issues['mixed'].add(', '.join(sorted(connector_ids)))
return resolved, None, issues
connector = pi_connectors[connector_ids.pop()] if connector_ids else None
return resolved, connector, issues
def resolve_pi_output_tags(
output_config: dict[str, Any],
pi_tags: dict[str, dict[str, Any]],
pi_connectors: dict[str, dict[str, Any]],
) -> tuple[dict[str, Any], dict[str, set[str]]]:
"""
Resolve the prediction_tags / confidence_tags of a pi_web_api_output_config.
A mapping given as a list is turned into {tag_name: WebID}: an entry with tag_id takes
tag_name and address from connector_tags ('Write' for prediction_tags, 'Prediction
Confidence' for confidence_tags); an entry without tag_id is read as a legacy
{tag_name: webid} fragment. A mapping already given as a dict passes through unchanged,
so the laborious input keeps today's shape.
Args:
output_config (dict[str, Any]): The pipeline's pi_web_api_output_config
pi_tags (dict[str, dict[str, Any]]): PI connector_tags indexed by index_connector_tags
pi_connectors (dict[str, dict[str, Any]]): Indexed by index_pi_connectors
Returns:
tuple: the resolved output config and what was left out, by kind ('tags',
'categories', 'connectors'), as in resolve_pi_read_tags
"""
issues = _new_pi_issues()
resolved = dict(output_config)
for key, category in PI_OUTPUT_CATEGORIES.items():
entries = output_config.get(key)
if not isinstance(entries, list):
continue
mapping: dict[str, Any] = {}
for entry in entries:
if not isinstance(entry, dict):
continue
tag_id = _tag_id(entry)
if tag_id is None:
mapping.update(entry)
continue
connector_tag = _locate_pi_tag(tag_id, category, pi_tags, pi_connectors, issues)
if connector_tag is not None:
mapping[connector_tag['tag_name']] = connector_tag['address']
resolved[key] = mapping
return resolved, issues
def _connector_server(connector: dict[str, Any]) -> dict[str, Any]:
"""Slot server fields of a tag database connector (connectors collection)."""
return {

105
orchestrator/workflows/orchestrator.py
View File

@@ -1,13 +1,18 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
import asyncio
from datetime import timedelta
from typing import Any
from sientia_do.temporal.policies import retry_policy
from orchestrator.activities.activities import Activities
from orchestrator.utils.orchestrator_functions import filter_by_tag_ids, referenced_tag_ids
from orchestrator.utils.orchestrator_functions import (
filter_by_tag_ids,
referenced_pi_tag_ids,
referenced_tag_ids,
)
# Tag database (SIENTIAPDE-1569) read by the slot v2 (SIENTIAPDE-1974). The workflow
# input may override these aggregations (connectors_query / connector_tags_query); the
@@ -41,6 +46,36 @@ DEFAULT_CONNECTOR_TAGS_QUERY = {
# Runs started before the tag database loads existed replay without them
TAG_DATABASE_PATCH = 'SIENTIAPDE-1974-tag-database'
# PI Web API side of the tag database, read by process_schedules only (never by the
# slot). The input may override them (pi_connectors_query / pi_connector_tags_query);
# the workflow puts a tag_id $match in front of pi_connector_tags_query. Tags of every
# category are loaded: predictions_batch writes to Write / Prediction Confidence tags.
DEFAULT_PI_CONNECTORS_QUERY = {
'collection': 'connectors',
'aggregation': [
{'$match': {'type': 'pi_web_api', 'status': 'Active'}},
{'$project': {'_id': 0, 'connector_id': 1, 'name': 1, 'url': 1, 'pi_web_api': 1}},
],
}
DEFAULT_PI_CONNECTOR_TAGS_QUERY = {
'collection': 'connector_tags',
'aggregation': [
{'$match': {'status': 'Active'}},
{
'$project': {
'_id': 0,
'tag_id': 1,
'connector_id': 1,
'address': 1,
'tag_name': 1,
'category': 1,
}
},
],
}
# Runs started before the PI Web API loads existed replay without them
PI_WEB_API_PATCH = 'connectors-pi-web-api'
@workflow.defn(name='orchestrator')
class Orchestrator:
@@ -81,9 +116,54 @@ class Orchestrator:
if not workflow.patched(TAG_DATABASE_PATCH) or not tag_ids:
return [], []
connectors_query = input_data.get('connectors_query', DEFAULT_CONNECTORS_QUERY)
connector_tags_query = input_data.get('connector_tags_query', DEFAULT_CONNECTOR_TAGS_QUERY)
return await self._load_connectors(
metadata,
input_data.get('connectors_query', DEFAULT_CONNECTORS_QUERY),
input_data.get('connector_tags_query', DEFAULT_CONNECTOR_TAGS_QUERY),
tag_ids,
)
async def _load_pi_tag_database(
self,
input_data: dict[str, Any],
metadata: dict[str, Any],
pipelines: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""
Load the PI Web API connectors and the tags the PI Web API pipelines reference.
Same bounds as _load_tag_database: nothing is loaded when no PI Web API tag has a
tag_id, or when replaying a run started before PI_WEB_API_PATCH.
Args:
input_data (dict[str, Any]): Workflow input (optional pi_connectors_query and
pi_connector_tags_query)
metadata (dict[str, Any]): Activity metadata ({'metadata': {...}})
pipelines (list[dict[str, Any]]): The loaded pipeline configurations
Returns:
tuple[list[dict[str, Any]], list[dict[str, Any]]]: pi connectors and their tags
"""
tag_ids = referenced_pi_tag_ids(pipelines)
# patched() first: it must run on every execution to record its marker
if not workflow.patched(PI_WEB_API_PATCH) or not tag_ids:
return [], []
return await self._load_connectors(
metadata,
input_data.get('pi_connectors_query', DEFAULT_PI_CONNECTORS_QUERY),
input_data.get('pi_connector_tags_query', DEFAULT_PI_CONNECTOR_TAGS_QUERY),
tag_ids,
)
async def _load_connectors(
self,
metadata: dict[str, Any],
connectors_query: dict[str, Any],
connector_tags_query: dict[str, Any],
tag_ids: list[Any],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""Run a connectors query and a connector_tags query restricted to tag_ids, in parallel."""
connectors_handler = workflow.start_local_activity_method(
Activities.aggregate_documents_in_mongodb,
{
@@ -125,6 +205,11 @@ class Orchestrator:
database connectors (default: DEFAULT_CONNECTORS_QUERY)
- connector_tags_query (dict[str, Any]): MongoDB aggregation for the tag
database tags (default: DEFAULT_CONNECTOR_TAGS_QUERY)
Optional fields (used only when a PI Web API tag carries a tag_id):
- pi_connectors_query (dict[str, Any]): MongoDB aggregation for the
pi_web_api connectors (default: DEFAULT_PI_CONNECTORS_QUERY)
- pi_connector_tags_query (dict[str, Any]): MongoDB aggregation for their
tags (default: DEFAULT_PI_CONNECTOR_TAGS_QUERY)
Returns:
None: Workflow completes without return value
@@ -193,8 +278,10 @@ class Orchestrator:
pipeline_config = await pipeline_config_handler
connectors, connector_tags = await self._load_tag_database(
input_data, metadata, pipeline_config
# Started in this order on every run, so commands stay deterministic
(connectors, connector_tags), (pi_connectors, pi_connector_tags) = await asyncio.gather(
self._load_tag_database(input_data, metadata, pipeline_config),
self._load_pi_tag_database(input_data, metadata, pipeline_config),
)
orchestrated_schedules = await orchestrated_schedules_handler
@@ -211,7 +298,13 @@ class Orchestrator:
schedules_config_handler = workflow.start_local_activity_method(
Activities.process_schedules,
{**metadata, 'pipelines': pipeline_config, 'connector_tags': connector_tags},
{
**metadata,
'pipelines': pipeline_config,
'connector_tags': connector_tags,
'pi_connectors': pi_connectors,
'pi_connector_tags': pi_connector_tags,
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60),
)