rosemarYuan commented on code in PR #923: URL: https://github.com/apache/flink-agents/pull/923#discussion_r3681276703
########## python/flink_agents/cli/trace_tree.py: ########## @@ -0,0 +1,333 @@ +#!/usr/bin/env python3 +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Reconstruct InputEvent-rooted Trace Trees from an Event Log.""" + +import argparse +import json +import sys +from collections import defaultdict +from pathlib import Path +from typing import Any, Iterator + +INPUT_EVENT_TYPE = "_input_event" + + +def _json_fingerprint(value: Any) -> str: + """Return a deterministic, type-sensitive representation of JSON data.""" + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def find_log_files(path: Path) -> list[Path]: + """Return Event Log files in deterministic file-name order.""" + if path.is_file(): + return [path] + return sorted(path.glob("events-*.log")) + + +def read_json_objects(path: Path, warnings: list[dict[str, Any]]) -> Iterator[Any]: + """Read consecutive JSON objects while retaining recoverable file warnings.""" + try: + content = path.read_text(encoding="utf-8") + except (OSError, UnicodeError): + warnings.append( + warning( + "UNREADABLE_FILE", + None, + f"Could not read Event Log file {path}.", + file_path=path, + ) + ) + return + + decoder = json.JSONDecoder() + position = 0 + while position < len(content): + while position < len(content) and content[position].isspace(): + position += 1 + if position == len(content): + return + try: + record, position = decoder.raw_decode(content, position) + except json.JSONDecodeError as error: + warnings.append( + warning( + "MALFORMED_RECORD", + None, + f"Could not decode an Event Log record in {path} " + f"at line {error.lineno}, column {error.colno}: {error.msg}.", + file_path=path, + line_number=error.lineno, + column_number=error.colno, + ) + ) + next_record = content.find("\n{", max(error.pos, position + 1)) + if next_record < 0: + return + position = next_record + 1 + continue + + yield record + + +def read_event_records( + path: Path, warnings: list[dict[str, Any]] +) -> Iterator[dict[str, Any]]: + """Read the fields needed to reconstruct Trace Trees.""" + log_files = find_log_files(path) + if not log_files: + message = f"No Event Log files found at {path}" + raise FileNotFoundError(message) + + for log_file in log_files: + for record in read_json_objects(log_file, warnings): + event_id: str | None = None + invalid_reason: str | None = None + if not isinstance(record, dict): + invalid_reason = "record must be a JSON object" + else: + event = record.get("event") + event_type = record.get("eventType") + if not isinstance(event, dict): + invalid_reason = "field 'event' must be a JSON object" + elif not isinstance(event.get("id"), str) or not event["id"]: + invalid_reason = "field 'event.id' must be a non-empty string" + elif not isinstance(event_type, str) or not event_type: + invalid_reason = "field 'eventType' must be a non-empty string" + else: + event_id = event["id"] + for field_name in ("upstreamEventId", "upstreamActionName"): + field_value = event.get(field_name) + if field_value is not None and not isinstance(field_value, str): + invalid_reason = ( + f"field 'event.{field_name}' must be a string or null" + ) + break + + if invalid_reason is not None: + warnings.append( + warning( + "MALFORMED_RECORD", + event_id, + f"Invalid Event Log record in {log_file}: {invalid_reason}.", + file_path=log_file, + ) + ) + continue + + assert isinstance(record, dict) + event = record["event"] + assert isinstance(event, dict) + event_content = dict(event) + event_content.pop("upstreamEventId", None) + event_content.pop("upstreamActionName", None) + yield { + "eventId": event["id"], + "eventType": record["eventType"], + "timestamp": record.get("timestamp"), + "upstreamEventId": event.get("upstreamEventId"), + "upstreamActionName": event.get("upstreamActionName"), + "eventContent": event_content, + } + + +def build_trace_forest( + records: Iterator[dict[str, Any]], + warnings: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """Build valid InputEvent trees while retaining auditable invalid nodes.""" + records_by_id: dict[str, list[dict[str, Any]]] = defaultdict(list) + for record in records: + records_by_id[record["eventId"]].append(record) + + if warnings is None: + warnings = [] + nodes: dict[str, dict[str, Any]] = {} + lineage_edges_by_id: dict[str, list[tuple[str | None, str | None]]] = {} + for event_id, matching_records in records_by_id.items(): + first_record = matching_records[0] + first_content_fingerprint = _json_fingerprint(first_record["eventContent"]) + if any( + record["eventType"] != first_record["eventType"] + or _json_fingerprint(record["eventContent"]) != first_content_fingerprint + for record in matching_records[1:] + ): + warnings.append( + warning( + "EVENT_ID_CONFLICT", + event_id, + f"Event ID {event_id} has inconsistent Event type or content " + f"across {len(matching_records)} records.", + ) + ) + continue Review Comment: Thanks @joeyutong and @weiqingy for the review. I have pushed the fixes. - The Trace Tree reader can now find multi-node cycles. It reports a `CYCLE_DETECTED` warning and removes only the edge that creates the cycle. Other valid branches are kept. - The document now explains that Python Event IDs cannot be changed. The old `result.id = event.id` code raises a `ValidationError`. Please use `reconstruct_from(event)` instead. - For `EVENT_ID_CONFLICT`, the reader now keeps the first record. Later conflicting records are counted, but their lineage links are not used. Child Events are no longer lost. I also added tests for these cases. Please take another look when convenient. Thanks! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
