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

SIENTIAPDE-1445

Browse Source 
Enhance Activities and API Integration

- Updated Activities class to include API operations for external data ingestion.
- Added API configuration builder to connectors_config.py for environment variable management.
- Integrated API configuration into worker setup.
- Expanded unit tests to cover new API functionality and configuration handling.
- Updated requirements.txt to include pycurl and prometheus-client for enhanced metrics support.
This commit is contained in:
vitor-aignosi
2025-12-17 15:16:00 -03:00
parent ab99695702
commit e87c5d2da6
14 changed files with 1709 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
scouter/activities/activities.py
View File

@@ -9,12 +9,13 @@ with workflow.unsafe.imports_passed_through():
from sientia_do.observability.logger import Logger
from sientia_do.temporal.activities.postgres import Postgres
from scouter.activities.api import API
from scouter.activities.gates import Gates
from scouter.activities.mongodb import MongoDB
from scouter.activities.redis import Redis
class Activities(Postgres, Redis, Gates, MongoDB):
class Activities(Postgres, Redis, Gates, MongoDB, API):
"""
Unified activities class that combines multiple data processing services.
@@ -24,6 +25,7 @@ class Activities(Postgres, Redis, Gates, MongoDB):
- Redis operations for caching and temporary storage
- Data quality gates and filtering
- MongoDB operations for data retrieval
- PI Web API operations for external data ingestion
- Notification handling and logging
The class implements the multiple inheritance pattern to provide a unified
@@ -35,6 +37,7 @@ class Activities(Postgres, Redis, Gates, MongoDB):
postgres_config: dict[str, Any],
redis_config: dict[str, Any],
mongodb_config: dict[str, Any],
api_config: dict[str, Any],
logger: Logger,
notification_handler: NotificationHandler,
):
@@ -48,6 +51,8 @@ class Activities(Postgres, Redis, Gates, MongoDB):
Required fields: host, port, username, password
mongodb_config (dict[str, Any]): MongoDB connection configuration.
Required fields: connection_string, database_name
api_config (dict[str, Any]): PI Web API configuration.
Required fields: base_url, auth_type, auth_token
logger (Logger): Logger instance for application logging
notification_handler (NotificationHandler): Handler for system notifications
"""
@@ -100,6 +105,17 @@ class Activities(Postgres, Redis, Gates, MongoDB):
metrics_controller=metrics_controller,
)
# Initialize API
API.__init__(
self,
base_url=api_config['base_url'],
auth_type=api_config['auth_type'],
auth_token=api_config['auth_token'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
self.pod_id = getenv('HOSTNAME', 'localhost')
def shutdown(self):
@@ -113,3 +129,4 @@ class Activities(Postgres, Redis, Gates, MongoDB):
MongoDB.close(self)
Redis.close(self)
Gates.close(self)
API.close(self)

133
scouter/activities/api.py Normal file
View File

@@ -0,0 +1,133 @@
from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through():
import traceback
from typing import Any
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.observability.logger import Logger
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from scouter.utils.clients.pi_web_api_client import PIWebAPIClient
class API(SientiaMonitoring):
"""
PI Web API operations for data retrieval.
This class provides Temporal activities for interacting with the PI Web API
to retrieve tag values and historical data. It implements:
- Tag value retrieval from PI Web API endpoints
- Data quality filtering and validation
- Error handling with notifications
- Metrics collection for monitoring
The class wraps the PIWebAPIClient to provide Temporal-aware activity methods
that can be used in workflow orchestration.
"""
def __init__(
self,
base_url: str,
auth_type: str,
auth_token: str,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
) -> None:
"""
Initialize API activity with PI Web API client.
Args:
base_url (str): Base URL of the PI Web API server
auth_type (str): Authentication type ('basic' or 'bearer')
auth_token (str): Authentication token
logger (Logger): Logger instance for operation logging
notification_handler (NotificationHandler): Handler for system notifications
metrics_controller (MetricsController): Controller for metrics collection
"""
SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
self.pi_web_api_client = PIWebAPIClient(
base_url=base_url,
auth_config={
'type': auth_type,
'token': auth_token,
},
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
def close(self) -> None:
"""
Close the PI Web API client and shutdown monitoring services.
"""
self.pi_web_api_client.close()
SientiaMonitoring.shutdown(self)
@activity.defn(name='get_tag_values')
async def get_tag_values(self, input_data: dict[str, Any]) -> list[dict]:
"""
Retrieve tag values from PI Web API for specified WebIds.
This activity fetches historical or real-time data from the PI Web API
for a set of configured tags. It returns the data as a list of dictionaries
suitable for further processing in the workflow.
Args:
input_data (dict[str, Any]): Activity input parameters.
Required fields:
- metadata (dict[str, Any]): Workflow execution metadata
- endpoint (str): PI Web API endpoint path
- web_ids (dict[str, str | None]): Tag names mapped to WebIds
- period (dict[str, str]): Time period with 'start_time' field
- api_timeout (int): Request timeout in seconds
- max_count (int, optional): Maximum data points per tag. Defaults to 1
Returns:
list[dict]: List of data records, each containing:
- timestamp: Data point timestamp
- name: Tag name
- value: Numeric value
- tag: WebId
Raises:
PIMSRequestError: If API request fails
Exception: If data retrieval or processing fails
"""
metadata = input_data['metadata']
endpoint = input_data['endpoint']
web_ids = input_data['web_ids']
period = input_data['period']
max_count = input_data.get('max_count', 1)
api_timeout = input_data['api_timeout']
self.info(f'Getting tag values from {endpoint}', metadata=metadata)
self.debug(f'Web IDs: {web_ids}', metadata=metadata)
try:
latest_values = await self.pi_web_api_client.get_latest_values_df(
endpoint=endpoint,
web_ids=web_ids,
start_time=period,
max_count=max_count,
metadata=metadata,
timeout=api_timeout,
)
except Exception as e:
await self.send_notification_async(
metadata=metadata,
notification_id='PI_WEB_API_REQUEST_ERROR',
message=f'Error getting tag values from PI Web API: {e}',
block='get_tag_values',
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc(),
)
raise e
self.debug(f'Latest values: {latest_values}', metadata=metadata)
self.info(f'Gathered {len(latest_values)} tag values', metadata=metadata)
return latest_values.to_dict(orient='records')

23
scouter/metrics.py
View File

@@ -1,4 +1,5 @@
from prometheus_client import Counter, Gauge
from prometheus_client import Counter, Gauge, Histogram
from sientia_do.observability.metrics import CORE_LABELS as SIENTIA_CORE_LABELS
# Application health and status metrics
APP_UP = Gauge(
@@ -23,3 +24,23 @@ TAG_CHANGES_MONITOR = Gauge(
'Current value change of each tag',
[*CORE_LABELS, 'tag_name'],
)
# Generic REST client metrics
GENERIC_REST_CLIENT_LAG = Histogram(
'scouter_generic_rest_client_lag',
'Lag time for a request to a generic REST client',
SIENTIA_CORE_LABELS,
)
GENERIC_REST_READ_COUNT = Counter(
'scouter_generic_rest_client_read_count',
'Number of reads from a generic REST client',
SIENTIA_CORE_LABELS,
)
GENERIC_REST_READ_ERROR_COUNT = Counter(
'scouter_generic_rest_client_read_error_count',
'Number of read errors from a generic REST client',
SIENTIA_CORE_LABELS,
)

335
scouter/utils/clients/pi_web_api_client.py Normal file
View File

@@ -0,0 +1,335 @@
import io
import json
import time
import warnings
from typing import Any
from urllib.parse import urlencode
import pandas as pd
import pycurl
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.observability.logger import Logger
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from scouter import metrics
warnings.simplefilter('ignore') # Ignore warnings such as 'verify=False'
class PIMSRequestError(Exception):
"""Generic error for failed requests to PI Web API using pycurl."""
pass
class PIWebAPIClient(SientiaMonitoring):
"""
Client for interacting with the PI Web API.
This class provides a robust interface for querying historical and real-time
data from OSIsoft PI systems through the PI Web API. It implements:
- Asynchronous HTTP requests using pycurl
- Authentication support (Basic and Bearer)
- Automatic data normalization and timestamp handling
- Comprehensive error handling and monitoring
- Metrics collection for observability
The client is designed for high-performance data retrieval with proper
connection management and error recovery mechanisms.
"""
def __init__(
self,
base_url: str,
auth_config: dict[str, Any],
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
headers_config: dict[str, Any] | None = None,
) -> None:
"""
Initialize the PI Web API client with connection parameters.
Args:
base_url (str): Base URL of the PI Web API server
auth_config (dict[str, Any]): Authentication configuration.
Required fields:
- type (str): Authentication type ('basic' or 'bearer')
- token (str): Authentication token
logger (Logger): Logger instance for operation logging
notification_handler (NotificationHandler): Handler for system notifications
metrics_controller (MetricsController): Controller for metrics collection
headers_config (dict[str, Any], optional): HTTP headers configuration.
Default headers include content-type, accept, and x-requested-with
max_concurrency (int, optional): Maximum number of concurrent requests. Defaults to 8
"""
if headers_config is None:
headers_config = {
'Content-Type': 'application/json',
'Accept': 'application/json',
'x-requested-with': 'XMLHttpRequest',
}
SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
self.base_url = base_url.rstrip('/')
# auth_config spec:
# 'type': 'basic' or 'bearer',
# 'token': 'token',
self.auth_config = auth_config
self.auth_config['type'] = self.auth_config['type'].lower()
self.headers: dict[str, str] = headers_config
self.authenticate()
def close(self) -> None:
"""
Close the client and shutdown monitoring services.
"""
SientiaMonitoring.shutdown(self)
def _to_clean_timestamp(self, series: pd.Series) -> pd.Series:
"""
Convert a Series of timestamps to datetime, UTC, and round to the nearest second.
Args:
series (pd.Series): Series containing timestamp values
Returns:
pd.Series: Cleaned timestamp series in UTC, floored to seconds
"""
series = pd.to_datetime(series, utc=True, errors='coerce')
return series.dt.floor('s')
def _extract_numeric(self, value: Any) -> float | None:
"""
Normalize a value (potentially nested) to float.
This method handles PI Web API response values that may be nested
in dictionaries or other structures, extracting the numeric value.
Args:
value (Any): Value to extract and normalize
Returns:
float | None: Numeric value as float, or None if conversion fails
"""
if isinstance(value, dict):
value = value.get('Value', value)
return pd.to_numeric(value, errors='coerce')
def authenticate(self):
"""
Configure authentication headers based on auth_config.
This method sets up the Authorization header using either Basic or Bearer
authentication based on the configured authentication type.
Raises:
ValueError: If authentication type is not 'basic' or 'bearer'
"""
self.logger.info(f'Authenticating with {self.auth_config["type"]} authentication')
if self.auth_config['type'] == 'basic':
self.headers['Authorization'] = f'Basic {self.auth_config["token"]}'
elif self.auth_config['type'] == 'bearer':
self.headers['Authorization'] = f'Bearer {self.auth_config["token"]}'
else:
raise ValueError(f'Invalid authentication type: {self.auth_config["type"]}')
async def _curl_get_json(
self,
url: str,
params: list[tuple[str, str]] | None = None,
timeout: int = 30,
verify: bool = True,
metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""
Perform a GET request using pycurl and return the decoded JSON response.
This method executes an asynchronous HTTP GET request with proper error handling,
metrics collection, and timeout management. It automatically tracks request
latency and emits monitoring metrics.
Args:
url (str): Target URL for the GET request
params (list[tuple[str, str]], optional): Query parameters as list of tuples.
Each tuple contains (parameter_name, parameter_value)
timeout (int, optional): Request timeout in seconds. Defaults to 30
verify (bool, optional): Verify SSL certificates. Defaults to True
metadata (dict[str, Any], optional): Workflow execution metadata for tracking
Returns:
dict[str, Any]: Parsed JSON response body
Raises:
PIMSRequestError: If HTTP error, connection error, or JSON parsing error occurs
"""
if metadata is None:
metadata = {}
buffer = io.BytesIO()
c = pycurl.Curl()
core_labels = self.get_core_labels(metadata=metadata, operation_type='get_json')
try:
if params:
query_string = urlencode(params, doseq=True)
full_url = f'{url}?{query_string}'
else:
full_url = url
c.setopt(pycurl.URL, full_url.encode('utf-8'))
c.setopt(pycurl.WRITEDATA, buffer)
# Configure HTTP headers
header_list = [f'{k}: {v}' for k, v in self.headers.items()]
if header_list:
c.setopt(pycurl.HTTPHEADER, header_list)
# Set request timeout
c.setopt(pycurl.TIMEOUT, timeout)
# Configure SSL verification
if not verify:
c.setopt(pycurl.SSL_VERIFYPEER, 0)
c.setopt(pycurl.SSL_VERIFYHOST, 0)
start_time = time.time()
try:
c.perform()
except Exception as e:
await self.emit_metric(
metric_object=metrics.GENERIC_REST_READ_ERROR_COUNT,
tags=core_labels,
)
raise e
await self.observe_lag(
start_time=start_time,
metric_object=metrics.GENERIC_REST_CLIENT_LAG,
tags=core_labels,
)
await self.emit_metric(
metric_object=metrics.GENERIC_REST_READ_COUNT,
tags=core_labels,
)
status_code = c.getinfo(pycurl.RESPONSE_CODE)
body = buffer.getvalue().decode('utf-8', errors='replace')
if status_code >= 400:
await self.emit_metric(
metric_object=metrics.GENERIC_REST_READ_ERROR_COUNT,
tags=core_labels,
)
raise PIMSRequestError(f"HTTP {status_code} calling '{full_url}': {body[:200]}")
try:
return json.loads(body)
except json.JSONDecodeError as e:
raise PIMSRequestError(
f"Error decoding JSON response from '{full_url}': {e}; body: {body[:200]}"
) from e
except pycurl.error as e:
raise PIMSRequestError(f"Connection error calling '{url}': {e}") from e
finally:
c.close()
async def get_latest_values_df(
self,
web_ids: dict[str, dict[str, str]],
endpoint: str,
timeout: int = 30,
start_time: str = '*-1d',
end_time: str = '*',
max_count: int | None = 1,
metadata: dict[str, Any] | None = None,
) -> pd.DataFrame:
"""
Retrieve historical values for multiple WebIds using PI Web API streamsets.
This method queries the PI Web API's /streamsets/recorded endpoint to fetch
historical data for multiple tags simultaneously. It returns a normalized
DataFrame with timestamps, tag names, values, and WebIds.
Args:
web_ids (dict[str, str | None]): Dictionary mapping tag names to their WebIds.
None values are filtered out before querying
endpoint (str): PI Web API endpoint path (e.g., '/streamsets/recorded')
timeout (int, optional): Request timeout in seconds. Defaults to 30
start_time (str, optional): Start time in PI Web API format (e.g., "*-50d").
Defaults to "*-1d" (1 day ago)
end_time (str, optional): End time in PI Web API format (e.g., "*").
Defaults to "*" (current time)
max_count (int, optional): Maximum number of data points per series.
Defaults to 1. If None, maxCount parameter is not sent
metadata (dict[str, Any], optional): Workflow execution metadata for tracking
Returns:
pd.DataFrame: DataFrame with columns:
- timestamp: Cleaned timestamp (UTC, floored to seconds)
- name: Tag name
- value: Numeric value (normalized)
- tag: WebId of the tag
Returns empty DataFrame if no data is found
Raises:
PIMSRequestError: If API request fails or returns invalid data
"""
if metadata is None:
metadata = {}
url = f'{self.base_url}{endpoint}'
# Build query parameters with WebIds (filtering out None values)
params: list[tuple[str, str]] = [('webid', web_id['webid']) for web_id in web_ids.values()]
params.extend(
[
('startTime', start_time),
('endtime', end_time),
('selectedFields', 'Items.Name;Items.Items.Timestamp;Items.Items.Value'),
]
)
params.append(('maxCount', str(max_count)))
data = await self._curl_get_json(
url=url,
params=params,
timeout=timeout,
verify=False,
metadata=metadata,
)
raw_data = data.get('Items', [])
records = []
for entry in raw_data:
tag_name = entry.get('Name')
series_items = entry.get('Items', [])
for it in series_items:
if isinstance(it, dict) and 'Timestamp' in it and 'Value' in it:
ts = it.get('Timestamp')
val = it.get('Value')
web_id = web_ids.get(tag_name)
if ts is not None:
records.append(
{
'timestamp': ts,
'name': tag_name,
'value': self._extract_numeric(val),
'tag': web_id,
}
)
if not records:
return pd.DataFrame()
df = pd.DataFrame.from_records(records)
df['timestamp'] = self._to_clean_timestamp(df['timestamp'])
return df

17
scouter/utils/connectors_config.py
View File

@@ -83,6 +83,23 @@ def build_mongodb_config() -> dict[str, Any]:
}
def build_api_config() -> dict[str, Any]:
"""
Build API connection configuration from environment variables.
Returns:
dict[str, Any]: API configuration dictionary with keys:
- base_url: API base URL (default: https://pi.example.com)
- auth_type: API authentication type (default: basic)
- auth_token: API authentication token (default: None)
"""
return {
'base_url': getenv('API_BASE_URL', 'https://pi.example.com'),
'auth_type': getenv('API_AUTH_TYPE', 'basic'),
'auth_token': getenv('API_AUTH_TOKEN', None),
}
def build_druid_config() -> dict[str, Any]:
"""
Build Apache Druid connection configuration from environment variables.

2
scouter/worker/worker.py
View File

@@ -15,6 +15,7 @@ with workflow.unsafe.imports_passed_through():
from scouter import metrics
from scouter.activities.activities import Activities
from scouter.utils.connectors_config import (
build_api_config,
build_mongodb_config,
build_postgres_config,
build_redis_config,
@@ -103,6 +104,7 @@ async def main():
postgres_config=build_postgres_config(),
redis_config=build_redis_config(),
mongodb_config=build_mongodb_config(),
api_config=build_api_config(),
)
logger.custom_info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata)

101
scouter/workflow/pi_web_api_scouter.py Normal file
View File

@@ -0,0 +1,101 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from datetime import timedelta
from typing import Any
from sientia_do.temporal.policies import retry_policy
from scouter.activities.activities import Activities
@workflow.defn(name='pi_web_api_scouter')
class PIWebAPIScouter:
"""
PI Web API Scouter workflow that orchestrates data ingestion from PI systems.
This workflow serves as the entry point for PI Web API data processing pipelines.
Unlike the standard Scouter that loads from MongoDB, this workflow directly queries
PI Web API endpoints to retrieve tag values and processes them for downstream use.
The workflow implements a direct API ingestion pattern with:
- Real-time data retrieval from PI Web API
- Configurable time periods and data point limits
- Error handling and retry policies
- Child workflow orchestration for data processing
- Integration with CoreScouter for standardized processing
"""
@workflow.run
async def run(self, input_data: dict[str, Any]) -> None:
"""
Execute the PI Web API Scouter workflow.
This method orchestrates the complete data ingestion process from PI Web API:
1. Retrieves tag values from PI Web API using configured WebIds
2. Validates and normalizes the retrieved data
3. Delegates data processing to the CoreScouter workflow
Args:
input_data (dict[str, Any]): Configuration and parameters for the workflow execution.
Required fields:
- model_name (str): Name of the data model being processed
- model_id (str): Unique identifier for the data model
- schedule_name (str): Unique identifier for the data collection schedule
- endpoint (str): PI Web API endpoint path (e.g., '/streamsets/recorded')
- web_ids (dict[str, str | None]): Mapping of tag names to WebIds
- period (dict[str, str]): Time period configuration with 'start_time'
- api_timeout (int): Request timeout in seconds for PI Web API calls
- max_count (int, optional): Maximum data points per tag. Defaults to 1
- trigger_laborious (bool): Flag to enable intensive data processing
- filters (dict[str, str]): Data quality filters configuration
- schema (str): Target database schema for data export
- table_name (str): Target table name for data export
- retention_time (int): Data retention period in Redis (seconds)
- model_tags (dict[str, Any]): Tag-specific configuration including:
- data_range: [min, max] values for data validation
- aggr_function: Aggregation method (avg, mdn, max, min, lts)
- frequency: Data collection frequency in milliseconds
- topics: List of Kafka topics for data routing
Returns:
None: This workflow doesn't return data, it orchestrates data processing
Raises:
WorkflowExecutionError: If workflow execution fails
ActivityExecutionError: If any activity fails after retry attempts
PIMSRequestError: If PI Web API request fails
"""
input_data['workflow_name'] = 'scouter'
metadata = {
'metadata': {
'model_id': input_data['model_id'],
'model_name': input_data['model_name'],
'schedule_name': input_data['schedule_name'],
'workflow_name': input_data['workflow_name'],
}
}
data = await workflow.execute_local_activity_method(
Activities.get_tag_values,
{
**metadata,
'endpoint': input_data['endpoint'],
'web_ids': input_data['model_tags'],
'period': input_data['period'],
'api_timeout': input_data['api_timeout'],
'max_count': input_data.get('max_count', 1),
},
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy,
)
if not data:
return
input_data['data'] = data
input_data['metadata'] = metadata
await workflow.execute_child_workflow('subworkflow.core_scouter', input_data)