From cf7f9fb63cf001d06005cbe192616551e1205cc6 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 8 Jan 2026 12:41:53 -0300 Subject: [PATCH] SIENTIAPDE-1478 Refactor PIWebAPIClient and Update Configuration Imports - Removed outdated PostgreSQL, Redis, MongoDB, and API configuration functions from connectors_config.py. - Deleted the pi_web_api_client.py file as part of the refactor. - Updated imports in worker.py and api.py to use the new repository structure. - Cleaned up scenarios.md by removing obsolete scenarios related to unique constraint violations. - Commented out the specific version of the sientia-dataops-library in requirements.txt for flexibility. --- e2e/scenarios.md | 20 -- requirements.txt | 3 +- scouter/activities/api.py | 3 +- scouter/metrics.py | 22 +- scouter/utils/clients/pi_web_api_client.py | 340 ------------------ scouter/utils/connectors_config.py | 96 ----- scouter/worker/worker.py | 2 +- .../workflow/sub_workflows/core_scouter.py | 4 +- tests/utils/clients/__init__.py | 0 9 files changed, 6 insertions(+), 484 deletions(-) delete mode 100644 scouter/utils/clients/pi_web_api_client.py delete mode 100644 tests/utils/clients/__init__.py diff --git a/e2e/scenarios.md b/e2e/scenarios.md index c5a9260..253c19e 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -338,25 +338,6 @@ The `pi_web_api_scouter` workflow: --- -#### Scenario 2.3.2: PostgreSQL Unique Constraint Violation -**Description**: Duplicate data violates unique constraint - -**Input**: -- Data with duplicate `model_id`, `timestamp`, `variable` combination -- `on_conflict: 'ignore'` configured - -**Expected Behavior**: -- PostgreSQL handles conflict with `ON CONFLICT DO NOTHING` -- `affected_rows` may be 0 for duplicates -- Workflow continues normally - -**Assertions**: -- No exception raised -- Duplicates ignored -- Workflow continues - ---- - ## 3. Activity-Specific Scenarios > **Note**: Activity-specific scenarios are better suited for unit tests rather than e2e tests. @@ -457,7 +438,6 @@ For each scenario, verify: 2. Scenario 2.1.2: Quality Filters 3. Scenario 2.1.3: Different Aggregations 4. Scenario 3.3.5: Invalid Aggregation -5. Scenario 3.5.2: Conflict Ignore ### Low Priority (Nice to Have) 1. Scenario 4.1.2: Retry Success diff --git a/requirements.txt b/requirements.txt index 541df15..3ffda95 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,6 +3,7 @@ psycopg2-binary sqlalchemy redis pymongo -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.7.1 +#git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.7.1 +/home/grezewave/Documents/projects/sientia/sientia-dataops-library/ prometheus-client pycurl \ No newline at end of file diff --git a/scouter/activities/api.py b/scouter/activities/api.py index fb65850..1b1c282 100644 --- a/scouter/activities/api.py +++ b/scouter/activities/api.py @@ -10,8 +10,7 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.observability.metrics_controller import MetricsController from sientia_do.observability.sientia_monitoring import SientiaMonitoring from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ - - from scouter.utils.clients.pi_web_api_client import PIWebAPIClient + from sientia_do.repository.pi_web_api_client import PIWebAPIClient class API(SientiaMonitoring): diff --git a/scouter/metrics.py b/scouter/metrics.py index 3ef2288..a89be12 100644 --- a/scouter/metrics.py +++ b/scouter/metrics.py @@ -23,24 +23,4 @@ TAG_CHANGES_MONITOR = Gauge( 'scouter_tag_changes_monitor', '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, -) +) \ No newline at end of file diff --git a/scouter/utils/clients/pi_web_api_client.py b/scouter/utils/clients/pi_web_api_client.py deleted file mode 100644 index a8ba79a..0000000 --- a/scouter/utils/clients/pi_web_api_client.py +++ /dev/null @@ -1,340 +0,0 @@ -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, - ) - - 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]}") - - await self.emit_metric( - metric_object=metrics.GENERIC_REST_READ_COUNT, - tags=core_labels, - ) - 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()] - # Invert startTime and endTime to get descending order (most recent first) - # PI Web API returns descending order when endTime < startTime - params.extend( - [ - ('startTime', end_time), # Use end_time as startTime (inverted) - ('endtime', start_time), # Use start_time as endTime (inverted) - ('selectedFields', 'Items.Name;Items.Items.Timestamp;Items.Items.Value'), - ] - ) - - if max_count is not None: - params.append(('maxCount', str(max_count))) - - data = await self._curl_get_json( - url=url, - params=params, - timeout=timeout, - verify=False, - metadata=metadata, - ) - - self.debug(f'Raw data from PI Web API: {data}', 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[tag_name]['webid'] - 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 diff --git a/scouter/utils/connectors_config.py b/scouter/utils/connectors_config.py index da712d9..64737df 100644 --- a/scouter/utils/connectors_config.py +++ b/scouter/utils/connectors_config.py @@ -2,31 +2,6 @@ from os import getenv from typing import Any -def build_postgres_config() -> dict[str, Any]: - """ - Build PostgreSQL connection configuration from environment variables. - - Returns: - dict[str, Any]: PostgreSQL configuration dictionary with keys: - - host: Database hostname (default: localhost) - - port: Database port (default: 5432) - - user: Database username (default: sientia) - - password: Database password (default: sientia) - - dbname: Database name (default: sientia) - - min_connections: Minimum connection pool size (default: 5) - - max_connections: Maximum connection pool size (default: 20) - """ - return { - 'host': getenv('POSTGRES_HOST', 'localhost'), - 'port': int(getenv('POSTGRES_PORT', '5432')), - 'user': getenv('POSTGRES_USER', 'sientia'), - 'password': getenv('POSTGRES_PASSWORD', 'sientia'), - 'dbname': getenv('POSTGRES_DBNAME', 'sientia'), - 'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')), - 'max_connections': int(getenv('POSTGRES_MAX_CONNECTIONS', '20')), - } - - def build_kafka_config() -> dict[str, Any]: """ Build Kafka configuration from environment variables. @@ -42,74 +17,3 @@ def build_kafka_config() -> dict[str, Any]: 'polling_time': int(getenv('KAFKA_POLLING_TIME', '1000')), 'group_id': 'scouter-group', } - - -def build_redis_config() -> dict[str, Any]: - """ - Build Redis connection configuration from environment variables. - - Returns: - dict[str, Any]: Redis configuration dictionary with keys: - - host: Redis server hostname (default: localhost) - - port: Redis server port (default: 6379) - - username: Redis authentication username (default: None) - - password: Redis authentication password (default: None) - """ - return { - 'host': getenv('REDIS_HOST', 'localhost'), - 'port': int(getenv('REDIS_PORT', '6379')), - 'username': getenv('REDIS_USERNAME', None), - 'password': getenv('REDIS_PASSWORD', None), - } - - -def build_mongodb_config() -> dict[str, Any]: - """ - Build MongoDB connection configuration from environment variables. - - Returns: - dict[str, Any]: MongoDB configuration dictionary with keys: - - connection_string: Complete MongoDB connection URI - - database_name: Target database name (default: sientia) - """ - username = getenv('MONGODB_USERNAME', 'sientia') - password = getenv('MONGODB_PASSWORD', 'sientia') - uri = getenv('MONGODB_URL', 'localhost:27017') - - connection_string = f'mongodb://{username}:{password}@{uri}' - return { - 'connection_string': connection_string, - 'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'), - } - - -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('PI_WEB_API_BASE_URL', 'https://pi.example.com'), - 'auth_type': getenv('PI_WEB_API_AUTH_TYPE', 'basic'), - 'auth_token': getenv('PI_WEB_API_AUTH_TOKEN', None), - } - - -def build_druid_config() -> dict[str, Any]: - """ - Build Apache Druid connection configuration from environment variables. - - Returns: - dict[str, Any]: Druid configuration dictionary with keys: - - host: Druid server hostname (default: localhost) - - port: Druid server port (default: 8082) - """ - return { - 'host': getenv('DRUID_HOST', 'localhost'), - 'port': int(getenv('DRUID_PORT', '8082')), - } diff --git a/scouter/worker/worker.py b/scouter/worker/worker.py index 4de701a..d518f66 100644 --- a/scouter/worker/worker.py +++ b/scouter/worker/worker.py @@ -14,7 +14,7 @@ with workflow.unsafe.imports_passed_through(): from scouter import metrics from scouter.activities.activities import Activities - from scouter.utils.connectors_config import ( + from sientia_do.connectors_config import ( build_api_config, build_mongodb_config, build_postgres_config, diff --git a/scouter/workflow/sub_workflows/core_scouter.py b/scouter/workflow/sub_workflows/core_scouter.py index 2e9b1c1..13b7942 100644 --- a/scouter/workflow/sub_workflows/core_scouter.py +++ b/scouter/workflow/sub_workflows/core_scouter.py @@ -109,9 +109,7 @@ class CoreScouter: 'schema': input_data['schema'], 'table_name': input_data['table_name'], 'data': held_data, - 'timestamp_conversion': {'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ}, - 'on_conflict': 'ignore', - 'unique_columns': ['model_id', 'timestamp', 'variable'], + 'timestamp_conversion': {'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ} }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60), diff --git a/tests/utils/clients/__init__.py b/tests/utils/clients/__init__.py deleted file mode 100644 index e69de29..0000000