weiqingy commented on code in PR #923:
URL: https://github.com/apache/flink-agents/pull/923#discussion_r3654008748
##########
docs/content/docs/operations/monitoring.md:
##########
@@ -184,14 +184,44 @@ Each record contains a top-level `timestamp`, the
resolved `logLevel`, and a top
{
"timestamp": "2024-01-15T10:30:00Z",
"logLevel": "STANDARD",
- "eventType": "_input_event",
+ "eventType": "MiddleEvent",
"event": {
- "eventType": "_input_event",
- "...": "..."
+ "eventType": "MiddleEvent",
+ "id": "39361629-4f1d-4b62-b734-a48b181cb6e0",
+ "attributes": {},
+ "type": "MiddleEvent",
+ "upstreamEventId": "dad5ed00-80e2-4746-8bdb-f126bad504b5",
+ "upstreamActionName": "action1"
}
}
```
+`upstreamEventId` identifies the Event consumed by the Action that emitted the
current Event, and `upstreamActionName` identifies that Action. The framework
maintains both fields directly on the `event` object, outside business
`attributes`. A root `InputEvent` omits both fields.
+
+### Trace Tree Reconstruction
+
+The local reader rebuilds InputEvent-rooted Trace Trees from a saved File
Event Log. From the repository root, pass either one log file for text output
or a log directory for Trace Tree JSON:
Review Comment:
This is an operations page, so the audience is people running jobs from an
installed wheel, but `tools/reconstruct_trace_tree.py` only exists in a git
checkout. `python/pyproject.toml` packages `flink_agents` only and has no
`[project.scripts]` block, so nothing installs the reader or puts it on `PATH`.
Everything else in `tools/` is a build or dev script (`build.sh`, `lint.sh`,
`ut.sh`, and so on), and this is the first user-facing program to land there.
The placement has two side effects worth weighing at the same time.
`tools/lint.sh` points ruff at `python/` only (lines 89-113), so the new
203-line file gets no ruff rules, no 88-column limit and no docstring checks,
which are the Python standards `AGENTS.md` states for this repo. And
`python/flink_agents/api/tests/test_trace_tree_reader.py:19-20` has to walk up
to `parents[4]` and `subprocess.run` a repo-root script from inside the `api`
module.
Should the reader live under `python/flink_agents/` with a
`[project.scripts]` entry (something like `flink-agents-trace-tree`), or would
you rather drop the docs pointer until it is packaged?
##########
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)
Review Comment:
The reader has a five-code warning framework and `monitoring.md:223`
promises that valid InputEvent-rooted trees are retained while warnings go to
stderr, but there is no `try`/`except` anywhere in the file. One malformed
record makes `raw_decode` raise `json.JSONDecodeError` and the whole run
aborts, so every valid tree in every file is lost, not just the bad record.
That is reachable while a job is still running: `FileEventLogger.append` writes
with `writer.println(json)` and exposes `flush()` as a separate method, so a
log read mid-run can genuinely end mid-record.
There is a second path at line 59. `event = record["event"]` raises a bare
`KeyError` on any JSON object that is not an `EventLogRecord`, and the fallback
glob at line 33 (`return log_files or sorted(path.glob("*.log"))`) sweeps in
any `.log` file in the directory when no `events-*.log` matches.
Could the per-file and per-record decoding be wrapped and routed through the
existing `warning()` helper, as something like `MALFORMED_RECORD` and
`UNREADABLE_FILE`?
##########
runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java:
##########
@@ -748,6 +781,16 @@ agentPlanWithStateStore, true, new
InMemoryActionStateStore(false)),
testHarness.processElement(new StreamRecord<>(inputValue));
operator.waitInFlightEventsFinished();
}
+
+ for (Map<String, ActionState> states :
actionStateStore.getKeyedActionStates().values()) {
Review Comment:
This pre-null loop manufactures a state the runtime can no longer produce.
`setOutputEventLineage` now runs before `maybePersistTaskResult` (which is why
`outputEvents = actionTaskResult.getOutputEvents();` moved up), and
`testCompletedActionStatePersistsOutputEventLineage` asserts that persisted
states always carry lineage. So the only way to reach null lineage in a stored
state is to write it by hand here.
The assertion at line 827 is `isNotNull()`. What the re-stamp on the reuse
branch actually guards is that the replayed trigger event is a different object
with a different UUID, so a value check is what would catch a regression: as
written, the test still passes if the re-stamp writes the wrong id. Asserting
that the re-stamped `upstreamEventId` equals the id of the InputEvent processed
in the second harness would pin the behavior that matters.
This is also riding on `testActionStateStoreReplayIncurNoFunctionCall`,
whose stated concern is that functions are not re-executed. Would a separate
test for the replay re-stamp, without the pre-nulling, read more clearly?
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -81,6 +87,30 @@ public Map<String, Object> getAttributes() {
return attributes;
}
+ /** Returns the ID of the Event consumed by the Action that emitted this
Event. */
+ @Nullable
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public UUID getUpstreamEventId() {
+ return upstreamEventId;
+ }
+
+ /** Sets the ID of the Event consumed by the Action that emitted this
Event. */
+ public void setUpstreamEventId(@Nullable UUID upstreamEventId) {
Review Comment:
Python declares both fields with `AliasChoices` (`event.py:89-98`), so they
are ordinary constructor kwargs, and the PR's own
`test_typed_from_event_preserves_lineage` builds a fully-formed Event in a
single `Event(type=..., attributes=..., upstreamEventId=...,
upstreamActionName=...)` call.
On the Java side the `@JsonCreator` constructor stays `(id, type,
attributes)`, so these setters are the only way in, which is why `EventTest`
has to do `new Event("ChildEvent")` followed by two `set...` calls. `AGENTS.md`
asks that public API changes keep Java and Python semantically aligned. This is
the construction surface specifically, separate from the reconstruction-helper
contract already under discussion on this PR.
Which way do you see this settling: a lineage-carrying constructor on the
Java side, or keeping both sides framework-write-only and treating the Python
kwargs as incidental to pydantic?
--
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]
