fix: remove timestamp normalization instead of gating it by max_count

Review feedback (Vitor): gating the batch-max normalization behind
max_count == 1 just hid the same collapsing behavior for the single-point
case, where it's redundant anyway (aggregation/export don't need it) while
still being the wrong fix conceptually. Removed the normalization block
entirely so every record always keeps its own real timestamp from the PI
Web API, regardless of max_count.

Updated unit tests accordingly and confirmed no e2e scenario needs
adjustment: aggregate_data groups by (tag, name) and always collapses to
one output row per tag per execution, independent of per-record timestamp
values, so pi_web_api_scouter_multiple_tags' aggregated-value assertions
are unaffected.

SIENTIAPDE-1993

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Eduardo Rios
2026-09-11 16:05:37 -03:00
parent fc3660433f
commit f79d51336f
3 changed files with 13 additions and 28 deletions

View File

@@ -48,7 +48,7 @@ Parent/child Temporal workflow pattern, two independent worker task queues shari
``` ```
Scouter (scouter/workflow/scouter.py) PIWebAPIScouter (scouter/workflow/pi_web_api_scouter.py) Scouter (scouter/workflow/scouter.py) PIWebAPIScouter (scouter/workflow/pi_web_api_scouter.py)
reads raw_{schedule_name} from MongoDB queries PI Web API directly (get_tag_values) reads raw_{schedule_name} from MongoDB queries PI Web API directly (get_tag_values)
tracks last-processed timestamp in Redis normalizes all timestamps to the batch max tracks last-processed timestamp in Redis preserves each record's real timestamp
│ │ │ │
└───────────────► CoreScouter (scouter/workflow/sub_workflows/core_scouter.py) ◄───────┘ └───────────────► CoreScouter (scouter/workflow/sub_workflows/core_scouter.py) ◄───────┘
1. data_quality_gate (NULL_VALUES_FILTER / OUT_OF_BOUNDS_FILTER, DISCARD policy) 1. data_quality_gate (NULL_VALUES_FILTER / OUT_OF_BOUNDS_FILTER, DISCARD policy)

View File

@@ -86,13 +86,8 @@ class API(SientiaMonitoring):
for a set of configured tags. It returns the data as a list of dictionaries for a set of configured tags. It returns the data as a list of dictionaries
suitable for further processing in the workflow. suitable for further processing in the workflow.
Timestamps are converted to string format. When max_count is 1 (one point per Timestamps are converted to string format and each record keeps its own real
tag), timestamps are additionally normalized to the maximum timestamp value timestamp as returned by the PI Web API.
(lexicographically) found in the dataset, to align minor sampling differences
across tags. When max_count > 1 (multiple historical points per tag), each
record's real timestamp is preserved instead, since collapsing them would make
distinct historical points collide on the (model_id, timestamp, variable)
unique constraint downstream.
Args: Args:
input_data (dict[str, Any]): Activity input parameters. input_data (dict[str, Any]): Activity input parameters.
@@ -106,8 +101,7 @@ class API(SientiaMonitoring):
Returns: Returns:
list[dict]: List of data records, each containing: list[dict]: List of data records, each containing:
- timestamp: Timestamp string (normalized to a shared value only when - timestamp: Each record's real timestamp string
max_count is 1; otherwise each record's real timestamp)
- name: Tag name - name: Tag name
- value: Numeric value - value: Numeric value
- tag: WebId - tag: WebId
@@ -152,15 +146,6 @@ class API(SientiaMonitoring):
self.debug(f'Latest values: {latest_values.to_string()}', metadata=metadata) self.debug(f'Latest values: {latest_values.to_string()}', metadata=metadata)
if max_count == 1:
# Normalize the package timestamp - only meaningful when a single point per tag
# is retrieved, to align minor sampling differences across tags. With
# max_count > 1 each historical point's real timestamp must be preserved,
# otherwise distinct records collide on the (model_id, timestamp, variable)
# unique constraint and are silently dropped downstream.
valid_timestamp_values = latest_values['timestamp'].dropna()
latest_values['timestamp'] = valid_timestamp_values.max()
self.info(f'Gathered {len(latest_values)} tag values', metadata=metadata) self.info(f'Gathered {len(latest_values)} tag values', metadata=metadata)
return latest_values.to_dict(orient='records') return latest_values.to_dict(orient='records')

View File

@@ -87,7 +87,7 @@ def test_close(mock_sientia_monitoring, api_activity):
def test_get_tag_values_success(api_activity): def test_get_tag_values_success(api_activity):
"""Test get_tag_values with max_count=1 normalizes timestamps across tags.""" """Test get_tag_values with successful data retrieval preserves each record's real timestamp."""
# Setup test data # Setup test data
test_data = { test_data = {
**metadata, **metadata,
@@ -110,7 +110,7 @@ def test_get_tag_values_success(api_activity):
}, },
}, },
'period': '*-1d', 'period': '*-1d',
'max_count': 1, 'max_count': 10,
'api_timeout': 30, 'api_timeout': 30,
} }
@@ -145,7 +145,7 @@ def test_get_tag_values_success(api_activity):
}, },
start_time='*-1d', start_time='*-1d',
end_time='*', end_time='*',
max_count=1, max_count=10,
metadata=metadata['metadata'], metadata=metadata['metadata'],
request_timeout=30, request_timeout=30,
) )
@@ -155,17 +155,17 @@ def test_get_tag_values_success(api_activity):
assert result[0]['value'] == pytest.approx(10.5) assert result[0]['value'] == pytest.approx(10.5)
assert result[1]['name'] == 'tag2' assert result[1]['name'] == 'tag2'
assert result[2]['name'] == 'tag3' assert result[2]['name'] == 'tag3'
# With max_count=1, timestamps are normalized to the batch max across tags # Each record keeps its own real timestamp, no cross-tag normalization
assert result[0]['timestamp'] == '2023-01-01 12:02:00+0000' assert result[0]['timestamp'] == '2023-01-01 12:00:00+0000'
assert result[1]['timestamp'] == '2023-01-01 12:02:00+0000' assert result[1]['timestamp'] == '2023-01-01 12:01:00+0000'
assert result[2]['timestamp'] == '2023-01-01 12:02:00+0000' assert result[2]['timestamp'] == '2023-01-01 12:02:00+0000'
def test_get_tag_values_with_max_count_greater_than_one_preserves_timestamps(api_activity): def test_get_tag_values_with_max_count_greater_than_one_preserves_timestamps(api_activity):
"""Regression test for SIENTIAPDE-1993: with max_count > 1, each historical point's """Regression test for SIENTIAPDE-1993: with max_count > 1, each historical point's
real timestamp must be preserved instead of being collapsed to the batch max, real timestamp must be preserved, otherwise distinct records collide on the
otherwise distinct records collide on the (model_id, timestamp, variable) unique (model_id, timestamp, variable) unique constraint downstream and are silently
constraint downstream and are silently dropped.""" dropped."""
test_data = { test_data = {
**metadata, **metadata,
'endpoint': '/streamsets/recorded', 'endpoint': '/streamsets/recorded',