weiqingy commented on code in PR #923: URL: https://github.com/apache/flink-agents/pull/923#discussion_r3697195854
########## 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: Confirmed on the log from this thread: `k` logged twice with different content, then `c <- k`, `d <- c`, `e <- d`. One `EVENT_ID_CONFLICT`, and the full tree renders. One case I hit while poking at it. When the discarded observation is the only record linking a parent to the conflicting Event, that parent loses its child edge and nothing names it: ``` records: rootA, rootB, k<-rootA(actA, content "full"), k<-rootB(actB, content "trunc"), c<-k warnings: [EVENT_ID_CONFLICT k "...across 2 records"] rootB.actions: [] Trace Tree 2 _input_event (rootB) ``` `rootB` reads as a root that produced nothing. Reachable via the cross-restart case from earlier: a reused Event ID is re-stamped with the current trigger's lineage, so a record logged either side of an `event-log.standard.max-string-length` change differs in content, and the replay parent is the one that drops. `CYCLE_DETECTED` names its omitted edge with `upstreamEventId` and `upstreamActionName`, and `warning()` takes both as of this commit. Carry the discarded edge the same way, or keep the warning at the Event level and let `observationCount` versus `upstreamEdges` tell the story? -- 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]
