weiqingy commented on code in PR #923: URL: https://github.com/apache/flink-agents/pull/923#discussion_r3671366825
########## 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: `continue` skips building a node for the conflicting id, and every descendant leaves the rendered trees with it, not just the conflicting Event. Six records in (a root, `K` logged twice with different content, then `C <- K`, `D <- C`, `E <- D`): ``` warnings: [('EVENT_ID_CONFLICT', 'k'), ('MISSING_PARENT', 'c')] nodes: ['c', 'd', 'e', 'r'] text: Trace Tree 1 _input_event (r) ``` `C` gets a `MISSING_PARENT`, but `D` and `E` get nothing, since they link cleanly to nodes that are themselves unreachable. Four Events drop out of the rendered tree. They stay in `nodes`, but nothing links them to a root, and two warnings name two of them. The path I can see reaching this: `FileEventLogger` truncates at write time from `event-log.standard.max-string-length` and friends, so a reused Event ID logged before and after a threshold change differs in content and reads as a conflict. That is the cross-restart case the new hint block describes. Could the conflicting id keep its first observation as the node, so the branch survives and the inconsistency still surfaces as a warning? -- 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]
