joeyutong commented on code in PR #923: URL: https://github.com/apache/flink-agents/pull/923#discussion_r3643414802
########## tools/reconstruct_trace_tree.py: ########## @@ -0,0 +1,203 @@ +#!/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 find_log_files(path: Path) -> list[Path]: + """Return Event Log files in deterministic file-name order.""" + if path.is_file(): + return [path] + log_files = sorted(path.glob("events-*.log")) + return log_files or sorted(path.glob("*.log")) + + +def read_json_objects(path: Path) -> Iterator[dict[str, Any]]: + """Read consecutive JSON objects separated by whitespace.""" + content = path.read_text(encoding="utf-8") + decoder = json.JSONDecoder() + position = 0 + while position < len(content): + while position < len(content) and content[position].isspace(): + position += 1 + if position == len(content): + return + record, position = decoder.raw_decode(content, position) + yield record + + +def read_event_records(path: Path) -> 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): + event = record["event"] + yield { + "eventId": str(event["id"]), + "eventType": record["eventType"], + "timestamp": record.get("timestamp"), + "upstreamEventId": event.get("upstreamEventId"), + "upstreamActionName": event.get("upstreamActionName"), + } + + +def build_trace_forest(records: Iterator[dict[str, Any]]) -> 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) + + warnings: list[dict[str, str]] = [] + nodes: dict[str, dict[str, Any]] = {} + for event_id, matching_records in records_by_id.items(): + if len(matching_records) > 1: Review Comment: ActionStateStore reuse can legitimately log the same Event ID again, possibly with different upstream lineage. Treating every repeated ID as invalid and dropping all matching records loses valid replay branches. Could reconstruction model this as a DAG instead: deduplicate repeated `(event ID, lineage edge)` observations, retain distinct lineage edges for the same Event ID, and report an ID conflict only when the Event type or content is inconsistent? ########## api/src/main/java/org/apache/flink/agents/api/Event.java: ########## @@ -105,18 +135,24 @@ public void setSourceTimestamp(long timestamp) { } /** - * Creates a base Event from another Event, copying id, type, and attributes. Subclasses - * override this to reconstruct typed event objects with proper field deserialization. + * Creates a base Event from another Event, copying its identity, data, and framework metadata. + * Subclasses override this to reconstruct typed event objects with proper field + * deserialization. */ public static Event fromEvent(Event event) { Event copy = new Event(event.getId(), event.getType(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - copy.setSourceTimestamp(event.getSourceTimestamp()); - } + copy.copyFrameworkMetadataFrom(event); return copy; } + /** Copies framework-managed metadata when reconstructing a typed Event. */ + protected void copyFrameworkMetadataFrom(Event event) { Review Comment: Typed reconstruction represents the same Event occurrence, but preserving that occurrence currently depends on every custom `fromEvent` implementation remembering both to pass the ID separately and to call this helper. Could the framework own a reconstruction template/factory so a custom subtype only maps its payload? Otherwise an omitted helper call silently loses `sourceTimestamp` and lineage. -- 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]
