You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "weiqingy (via GitHub)" <gi...@apache.org> on 2026/04/07 06:53:35 UTC
[PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
weiqingy opened a new pull request, #609:
URL: https://github.com/apache/flink-agents/pull/609
Linked issue: #541 , https://github.com/apache/flink-agents/discussions/552
### Purpose of change
Add per-event-type configurable log levels (`OFF`, `STANDARD`, `VERBOSE`) to the event log system, as designed in https://github.com/apache/flink-agents/discussions/552.
**Key changes:**
- `EventLogLevel` enum (`OFF`, `STANDARD`, `VERBOSE`) with hierarchical config key inheritance (`event-log.type.<EVENT_TYPE>.level`)
- At `STANDARD` level, large content fields are truncated using configurable per-field thresholds (`max-string-length`, `max-array-elements`, `max-depth`) with JSON-parseable wrapper objects (`truncatedString`, `truncatedList`, `truncatedObject`)
- Top-level `logLevel` and `eventType` fields added to JSON log record schema for downstream filtering
- `EventFilter` removed — fully subsumed by the log level system
- `eventLogTruncatedEvents` counter metric for observability
- Python config options and E2E tests
**Config options:**
| Key | Default | Description |
|-----|---------|-------------|
| `event-log.level` | `STANDARD` | Root default log level |
| `event-log.type.<EVENT_TYPE>.level` | _(inherits)_ | Per-type override |
| `event-log.standard.max-string-length` | `2000` | Max string chars at STANDARD |
| `event-log.standard.max-array-elements` | `20` | Max array elements at STANDARD |
| `event-log.standard.max-depth` | `5` | Max nesting depth at STANDARD |
### Tests
- `EventLogLevelResolverTest` — 11 tests: exact match, hierarchy inheritance, root/built-in defaults, case insensitivity, edge cases
- `JsonTruncatorTest` — 8 tests: string/array/depth truncation, wrappers, protected fields, composition, disabled thresholds
- `FileEventLoggerTest` — 7 new tests: STANDARD truncation, VERBOSE preservation, OFF suppression, per-type override, hierarchical inheritance, JSON schema fields, backward-compatible deserialization
- Python E2E tests — 3 new: VERBOSE, OFF, STANDARD truncation
### API
- **New public API**: `EventLogLevel` enum in `api` module
- **Removed public API**: `EventFilter` interface (0.x pre-release, no known users)
- **Modified**: `EventLoggerConfig` — removed `eventFilter` field/builder method
### Documentation
- [x] `doc-needed`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3108615942
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/JsonTruncator.java:
##########
@@ -0,0 +1,269 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.agents.runtime.eventlog;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Truncates a Jackson {@link JsonNode} tree per configurable thresholds.
+ *
+ * <p>Three truncation strategies are applied in a single recursive pass:
+ *
+ * <ul>
+ * <li><b>String truncation</b>: Strings longer than {@code maxStringLength} are replaced with a
+ * wrapper: {@code {"truncatedString": "first N chars...", "omittedChars": M}}
+ * <li><b>Array truncation</b>: Arrays larger than {@code maxArrayElements} are replaced with a
+ * wrapper: {@code {"truncatedList": [first N elements], "omittedElements": M}}
+ * <li><b>Depth truncation</b>: At max depth, object nodes retain only scalar fields; nested
+ * objects/arrays are dropped: {@code {"truncatedObject": {scalars...}, "omittedFields": N}}
+ * </ul>
+ *
+ * <p>Setting any threshold to {@code 0} disables that specific truncation strategy. If all
+ * thresholds are {@code 0}, no truncation occurs.
+ *
+ * <p>Protected fields at the top level of the event node ({@code eventType}, {@code id}, {@code
+ * attributes}) are never truncated.
+ */
+public class JsonTruncator {
+
+ private static final Set<String> PROTECTED_FIELDS =
+ new HashSet<>(Arrays.asList("eventType", "id", "attributes"));
+
+ private final int maxStringLength;
+ private final int maxArrayElements;
+ private final int maxDepth;
+
+ /**
+ * Creates a new truncator with the given thresholds.
+ *
+ * @param maxStringLength maximum character length for string values; 0 to disable
+ * @param maxArrayElements maximum number of array elements retained; 0 to disable
+ * @param maxDepth maximum object nesting depth; 0 to disable
+ */
+ public JsonTruncator(int maxStringLength, int maxArrayElements, int maxDepth) {
+ this.maxStringLength = maxStringLength;
+ this.maxArrayElements = maxArrayElements;
+ this.maxDepth = maxDepth;
+ }
+
+ /**
+ * Truncates the given event node in place according to configured thresholds.
+ *
+ * <p>Protected fields ({@code eventType}, {@code id}, {@code attributes}) at the top level of
+ * the event node are never truncated.
+ *
+ * @param eventNode the top-level event JSON object to truncate
+ * @return {@code true} if any field was truncated, {@code false} if the node was unchanged
+ */
+ public boolean truncate(ObjectNode eventNode) {
+ if (eventNode == null) {
+ return false;
+ }
+ return truncateObject(eventNode, 1, true);
+ }
+
+ /**
+ * Recursively truncates an object node.
+ *
+ * @param node the object node to process
+ * @param depth current depth (1 = top-level event node)
+ * @param isTopLevel whether this is the top-level event node (for protected field checks)
+ * @return true if any truncation occurred
+ */
+ private boolean truncateObject(ObjectNode node, int depth, boolean isTopLevel) {
+ boolean truncated = false;
+
+ // At max depth, collapse the entire object to retain only scalars
+ if (maxDepth > 0 && depth >= maxDepth) {
+ return collapseAtMaxDepth(node, isTopLevel);
+ }
+
+ List<String> fieldNames = new ArrayList<>();
+ node.fieldNames().forEachRemaining(fieldNames::add);
+
+ for (String fieldName : fieldNames) {
+ if (isTopLevel && PROTECTED_FIELDS.contains(fieldName)) {
+ continue;
+ }
+
+ JsonNode child = node.get(fieldName);
+ if (child == null) {
+ continue;
+ }
+
+ if (child.isTextual()) {
+ JsonNode replacement = truncateString(child.textValue());
+ if (replacement != null) {
+ node.set(fieldName, replacement);
+ truncated = true;
+ }
+ } else if (child.isArray()) {
+ // First recurse into retained elements, then truncate the array if needed
+ truncated |= truncateArrayContents((ArrayNode) child, depth + 1);
+ JsonNode replacement = truncateArray((ArrayNode) child);
Review Comment:
Shall we truncate the array before truncating the array contents? If an element is omitted, we should not bother to truncate its content at all.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogRecord.java:
##########
@@ -31,23 +35,82 @@
*
* <p>The class uses custom JSON serialization/deserialization to handle polymorphic Event types by
* leveraging the eventType information stored in the EventContext.
+ *
+ * <p>Each record carries a {@link EventLogLevel} that controls truncation behavior during
+ * serialization, and a convenience copy of the event type string from the context.
*/
@JsonSerialize(using = EventLogRecordJsonSerializer.class)
@JsonDeserialize(using = EventLogRecordJsonDeserializer.class)
public class EventLogRecord {
private final EventContext context;
private final Event event;
+ private final EventLogLevel logLevel;
+ private final String eventType;
+ @Nullable private final transient JsonTruncator truncator;
+ @Nullable private final transient Counter truncatedEventsCounter;
Review Comment:
Why do we need the log-level and truncator in the record?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 merged PR #609:
URL: https://github.com/apache/flink-agents/pull/609
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3094868967
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogRecord.java:
##########
@@ -31,23 +33,82 @@
*
* <p>The class uses custom JSON serialization/deserialization to handle polymorphic Event types by
* leveraging the eventType information stored in the EventContext.
+ *
+ * <p>Each record carries a {@link EventLogLevel} that controls truncation behavior during
+ * serialization, and a convenience copy of the event type string from the context.
*/
@JsonSerialize(using = EventLogRecordJsonSerializer.class)
@JsonDeserialize(using = EventLogRecordJsonDeserializer.class)
public class EventLogRecord {
private final EventContext context;
private final Event event;
+ private final EventLogLevel logLevel;
+ private final String eventType;
+ private final transient JsonTruncator truncator;
+ private final transient Counter truncatedEventsCounter;
Review Comment:
Added `@Nullable` on both fields.
Regarding removing them from `EventLogRecord` — I agree they are serialization plumbing rather than content. However, Jackson's custom serializer (`EventLogRecordJsonSerializer`) receives only the record object via `serialize(EventLogRecord record, ...)`. Since the `ObjectMapper` is shared/static, there's no clean way to pass the truncator and counter to the serializer except through the record itself. The `transient` + `@Nullable` markers signal that these are not part of the logical content. Happy to explore alternatives if you have a preferred pattern in mind.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3162076092
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/JsonTruncator.java:
##########
@@ -0,0 +1,269 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.agents.runtime.eventlog;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Truncates a Jackson {@link JsonNode} tree per configurable thresholds.
+ *
+ * <p>Three truncation strategies are applied in a single recursive pass:
+ *
+ * <ul>
+ * <li><b>String truncation</b>: Strings longer than {@code maxStringLength} are replaced with a
+ * wrapper: {@code {"truncatedString": "first N chars...", "omittedChars": M}}
+ * <li><b>Array truncation</b>: Arrays larger than {@code maxArrayElements} are replaced with a
+ * wrapper: {@code {"truncatedList": [first N elements], "omittedElements": M}}
+ * <li><b>Depth truncation</b>: At max depth, object nodes retain only scalar fields; nested
+ * objects/arrays are dropped: {@code {"truncatedObject": {scalars...}, "omittedFields": N}}
+ * </ul>
+ *
+ * <p>Setting any threshold to {@code 0} disables that specific truncation strategy. If all
+ * thresholds are {@code 0}, no truncation occurs.
+ *
+ * <p>Protected fields at the top level of the event node ({@code eventType}, {@code id}, {@code
+ * attributes}) are never truncated.
+ */
+public class JsonTruncator {
+
+ private static final Set<String> PROTECTED_FIELDS =
+ new HashSet<>(Arrays.asList("eventType", "id", "attributes"));
+
+ private final int maxStringLength;
+ private final int maxArrayElements;
+ private final int maxDepth;
+
+ /**
+ * Creates a new truncator with the given thresholds.
+ *
+ * @param maxStringLength maximum character length for string values; 0 to disable
+ * @param maxArrayElements maximum number of array elements retained; 0 to disable
+ * @param maxDepth maximum object nesting depth; 0 to disable
+ */
+ public JsonTruncator(int maxStringLength, int maxArrayElements, int maxDepth) {
+ this.maxStringLength = maxStringLength;
+ this.maxArrayElements = maxArrayElements;
+ this.maxDepth = maxDepth;
+ }
+
+ /**
+ * Truncates the given event node in place according to configured thresholds.
+ *
+ * <p>Protected fields ({@code eventType}, {@code id}, {@code attributes}) at the top level of
+ * the event node are never truncated.
+ *
+ * @param eventNode the top-level event JSON object to truncate
+ * @return {@code true} if any field was truncated, {@code false} if the node was unchanged
+ */
+ public boolean truncate(ObjectNode eventNode) {
+ if (eventNode == null) {
+ return false;
+ }
+ return truncateObject(eventNode, 1, true);
+ }
+
+ /**
+ * Recursively truncates an object node.
+ *
+ * @param node the object node to process
+ * @param depth current depth (1 = top-level event node)
+ * @param isTopLevel whether this is the top-level event node (for protected field checks)
+ * @return true if any truncation occurred
+ */
+ private boolean truncateObject(ObjectNode node, int depth, boolean isTopLevel) {
+ boolean truncated = false;
+
+ // At max depth, collapse the entire object to retain only scalars
+ if (maxDepth > 0 && depth >= maxDepth) {
+ return collapseAtMaxDepth(node, isTopLevel);
+ }
+
+ List<String> fieldNames = new ArrayList<>();
+ node.fieldNames().forEachRemaining(fieldNames::add);
+
+ for (String fieldName : fieldNames) {
+ if (isTopLevel && PROTECTED_FIELDS.contains(fieldName)) {
+ continue;
+ }
+
+ JsonNode child = node.get(fieldName);
+ if (child == null) {
+ continue;
+ }
+
+ if (child.isTextual()) {
+ JsonNode replacement = truncateString(child.textValue());
+ if (replacement != null) {
+ node.set(fieldName, replacement);
+ truncated = true;
+ }
+ } else if (child.isArray()) {
+ // First recurse into retained elements, then truncate the array if needed
+ truncated |= truncateArrayContents((ArrayNode) child, depth + 1);
+ JsonNode replacement = truncateArray((ArrayNode) child);
Review Comment:
Agreed — applied in `4356979`. Both call sites now truncate the array first and only recurse into the elements that survive: the `truncateObject` array branch and the symmetric path in `truncateArrayContents` for nested arrays.
Added `testRecursionSkipsDroppedTailElements`, which holds references to the dropped tail elements and asserts their content remains unmutated, so the invariant is locked in against future regressions.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3094877516
##########
api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java:
##########
@@ -54,4 +54,32 @@ public class AgentConfigOptions {
/** The config parameter specifies the unique identifier of job. */
public static final ConfigOption<String> JOB_IDENTIFIER =
new ConfigOption<>("job-identifier", String.class, null);
+
+ /**
+ * The global event log level controlling the default verbosity for all event types. Valid
+ * values are "OFF", "STANDARD", and "VERBOSE". Defaults to "STANDARD".
+ */
+ public static final ConfigOption<String> EVENT_LOG_LEVEL =
+ new ConfigOption<>("event-log.level", String.class, "STANDARD");
Review Comment:
Yes. Updated the PR.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #609:
URL: https://github.com/apache/flink-agents/pull/609#issuecomment-4349214128
LGTM
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3061799790
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogLevelResolver.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.agents.runtime.eventlog;
+
+import org.apache.flink.agents.api.logger.EventLogLevel;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Resolves the effective {@link EventLogLevel} for a given event type string using hierarchical
+ * config key inheritance.
+ *
+ * <p>Resolution order for a given event type (e.g., {@code org.apache.flink.agents.api.event
+ * .ChatRequestEvent}):
+ *
+ * <ol>
+ * <li>Exact match in explicit per-type configuration
+ * <li>Walk up dot-separated segments (e.g., {@code org.apache.flink.agents.api.event}, then
+ * {@code org.apache.flink.agents.api}, etc.)
+ * <li>Root default from {@code event-log.level} config key
+ * <li>Built-in default: {@link EventLogLevel#STANDARD}
+ * </ol>
+ *
+ * <p>This mirrors Log4j's hierarchical logger configuration pattern. Resolved levels are cached in
+ * a {@link ConcurrentHashMap} for efficient repeated lookups.
+ *
+ * <p>Config keys are expected in the form:
+ *
+ * <pre>
+ * event-log.level = STANDARD (root default)
+ * event-log.type.<EVENT_TYPE>.level = OFF (per-type override)
+ * </pre>
+ */
+public class EventLogLevelResolver {
+
+ /** Config key for the root default log level. */
+ static final String ROOT_LEVEL_KEY = "event-log.level";
+
+ /** Prefix for per-event-type log level config keys. */
+ static final String TYPE_PREFIX = "event-log.type.";
+
+ /** Suffix for per-event-type log level config keys. */
+ static final String TYPE_SUFFIX = ".level";
+
+ /** Built-in default when no configuration is provided. */
+ static final EventLogLevel BUILT_IN_DEFAULT = EventLogLevel.STANDARD;
+
+ private final EventLogLevel rootDefault;
+ private final Map<String, EventLogLevel> explicitLevels;
+ private final ConcurrentHashMap<String, EventLogLevel> cache;
+
+ /**
+ * Creates a resolver from a configuration data map.
+ *
+ * <p>The map is scanned for keys matching {@code event-log.type.<EVENT_TYPE>.level} to build
+ * the explicit per-type level mappings, and {@code event-log.level} for the root default.
+ *
+ * @param confData the flat configuration key-value map (e.g., from {@code
+ * AgentConfiguration.getConfData()})
+ */
+ public EventLogLevelResolver(Map<String, Object> confData) {
+ Map<String, Object> data = confData != null ? confData : Collections.emptyMap();
+
+ // Parse root default
+ Object rootValue = data.get(ROOT_LEVEL_KEY);
Review Comment:
Shall we just use `EVENT_LOG_LEVEL.getKey()` here?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogLevelResolver.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.agents.runtime.eventlog;
+
+import org.apache.flink.agents.api.logger.EventLogLevel;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Resolves the effective {@link EventLogLevel} for a given event type string using hierarchical
+ * config key inheritance.
+ *
+ * <p>Resolution order for a given event type (e.g., {@code org.apache.flink.agents.api.event
+ * .ChatRequestEvent}):
+ *
+ * <ol>
+ * <li>Exact match in explicit per-type configuration
+ * <li>Walk up dot-separated segments (e.g., {@code org.apache.flink.agents.api.event}, then
+ * {@code org.apache.flink.agents.api}, etc.)
+ * <li>Root default from {@code event-log.level} config key
+ * <li>Built-in default: {@link EventLogLevel#STANDARD}
+ * </ol>
+ *
+ * <p>This mirrors Log4j's hierarchical logger configuration pattern. Resolved levels are cached in
+ * a {@link ConcurrentHashMap} for efficient repeated lookups.
+ *
+ * <p>Config keys are expected in the form:
+ *
+ * <pre>
+ * event-log.level = STANDARD (root default)
+ * event-log.type.<EVENT_TYPE>.level = OFF (per-type override)
+ * </pre>
+ */
+public class EventLogLevelResolver {
+
+ /** Config key for the root default log level. */
+ static final String ROOT_LEVEL_KEY = "event-log.level";
+
+ /** Prefix for per-event-type log level config keys. */
+ static final String TYPE_PREFIX = "event-log.type.";
+
+ /** Suffix for per-event-type log level config keys. */
+ static final String TYPE_SUFFIX = ".level";
+
+ /** Built-in default when no configuration is provided. */
+ static final EventLogLevel BUILT_IN_DEFAULT = EventLogLevel.STANDARD;
+
+ private final EventLogLevel rootDefault;
+ private final Map<String, EventLogLevel> explicitLevels;
+ private final ConcurrentHashMap<String, EventLogLevel> cache;
+
+ /**
+ * Creates a resolver from a configuration data map.
+ *
+ * <p>The map is scanned for keys matching {@code event-log.type.<EVENT_TYPE>.level} to build
+ * the explicit per-type level mappings, and {@code event-log.level} for the root default.
+ *
+ * @param confData the flat configuration key-value map (e.g., from {@code
+ * AgentConfiguration.getConfData()})
+ */
+ public EventLogLevelResolver(Map<String, Object> confData) {
+ Map<String, Object> data = confData != null ? confData : Collections.emptyMap();
+
+ // Parse root default
+ Object rootValue = data.get(ROOT_LEVEL_KEY);
+ this.rootDefault =
+ rootValue != null
+ ? EventLogLevel.fromString(rootValue.toString())
+ : BUILT_IN_DEFAULT;
Review Comment:
Shall we just use `EVENT_LOG_LEVEL.getDefaultValue()` here?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogRecordJsonSerializer.java:
##########
@@ -67,15 +70,37 @@ public void serialize(EventLogRecord record, JsonGenerator gen, SerializerProvid
gen.writeStartObject();
gen.writeStringField("timestamp", record.getContext().getTimestamp());
+ gen.writeStringField(
+ "logLevel", record.getLogLevel() != null ? record.getLogLevel().name() : "VERBOSE");
+ // For PythonEvent, use the Python module path as the top-level eventType
+ // (consistent with the eventType inside the event object)
+ String topLevelEventType = record.getContext().getEventType();
+ if (record.getEvent() instanceof PythonEvent) {
+ String pythonType = ((PythonEvent) record.getEvent()).getEventType();
+ if (pythonType != null) {
+ topLevelEventType = pythonType;
+ }
+ }
Review Comment:
The logic for extracting the python event type is the same as in `FileEventLogger`. We can extract a common util function.
##########
api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java:
##########
@@ -54,4 +54,32 @@ public class AgentConfigOptions {
/** The config parameter specifies the unique identifier of job. */
public static final ConfigOption<String> JOB_IDENTIFIER =
new ConfigOption<>("job-identifier", String.class, null);
+
+ /**
+ * The global event log level controlling the default verbosity for all event types. Valid
+ * values are "OFF", "STANDARD", and "VERBOSE". Defaults to "STANDARD".
+ */
+ public static final ConfigOption<String> EVENT_LOG_LEVEL =
+ new ConfigOption<>("event-log.level", String.class, "STANDARD");
Review Comment:
Maybe we can just use enum here
```
public static final ConfigOption<EventLogLevel> EVENT_LOG_LEVEL =
new ConfigOption<>("event-log.level", EventLogLevel.class, EventLogLevel.STANDARD);
```
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -342,6 +342,10 @@ private void initEventLogger(StreamingRuntimeContext runtimeContext) throws Exce
return;
}
eventLogger.open(new EventLoggerOpenParams(runtimeContext));
+ if (eventLogger instanceof FileEventLogger) {
Review Comment:
Why is it necessary to specifically check the type of `eventLogger` here?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogRecord.java:
##########
@@ -31,23 +33,82 @@
*
* <p>The class uses custom JSON serialization/deserialization to handle polymorphic Event types by
* leveraging the eventType information stored in the EventContext.
+ *
+ * <p>Each record carries a {@link EventLogLevel} that controls truncation behavior during
+ * serialization, and a convenience copy of the event type string from the context.
*/
@JsonSerialize(using = EventLogRecordJsonSerializer.class)
@JsonDeserialize(using = EventLogRecordJsonDeserializer.class)
public class EventLogRecord {
private final EventContext context;
private final Event event;
+ private final EventLogLevel logLevel;
+ private final String eventType;
+ private final transient JsonTruncator truncator;
+ private final transient Counter truncatedEventsCounter;
Review Comment:
Maybe we can mark truncator and truncatedEventsCounter as `@Nullable`.
Additionally, these two members are actually required variables for the serialization of `EventLogRecord`, rather than part of its content. During deserialization, we have to set them to `null`. I wonder if they could be removed from within `EventLogRecord` itself.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3094860436
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogRecordJsonSerializer.java:
##########
@@ -67,15 +70,37 @@ public void serialize(EventLogRecord record, JsonGenerator gen, SerializerProvid
gen.writeStartObject();
gen.writeStringField("timestamp", record.getContext().getTimestamp());
+ gen.writeStringField(
+ "logLevel", record.getLogLevel() != null ? record.getLogLevel().name() : "VERBOSE");
+ // For PythonEvent, use the Python module path as the top-level eventType
+ // (consistent with the eventType inside the event object)
+ String topLevelEventType = record.getContext().getEventType();
+ if (record.getEvent() instanceof PythonEvent) {
+ String pythonType = ((PythonEvent) record.getEvent()).getEventType();
+ if (pythonType != null) {
+ topLevelEventType = pythonType;
+ }
+ }
Review Comment:
Good catch. Extracted `EventUtil.resolveEventType(Event, EventContext)` and replaced the duplicated logic in both `FileEventLogger.append()` and `EventLogRecordJsonSerializer.serialize()`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3107948831
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogRecord.java:
##########
@@ -31,23 +33,82 @@
*
* <p>The class uses custom JSON serialization/deserialization to handle polymorphic Event types by
* leveraging the eventType information stored in the EventContext.
+ *
+ * <p>Each record carries a {@link EventLogLevel} that controls truncation behavior during
+ * serialization, and a convenience copy of the event type string from the context.
*/
@JsonSerialize(using = EventLogRecordJsonSerializer.class)
@JsonDeserialize(using = EventLogRecordJsonDeserializer.class)
public class EventLogRecord {
private final EventContext context;
private final Event event;
+ private final EventLogLevel logLevel;
+ private final String eventType;
+ private final transient JsonTruncator truncator;
+ private final transient Counter truncatedEventsCounter;
Review Comment:
A simple approach might be to move the truncation from the Record serialization process into FileEventLogger.append(), like:
```
@Override
public void append(EventContext context, Event event) throws Exception {
if (writer == null) {
throw new IllegalStateException("FileEventLogger not initialized. Call open() first.");
}
String eventTypeStr = EventUtil.resolveEventType(event, context);
// Resolve log level and skip OFF events
EventLogLevel level =
levelResolver != null ? levelResolver.resolve(eventTypeStr) : EventLogLevel.VERBOSE;
if (level == EventLogLevel.OFF) {
return;
}
EventLogRecord record = new EventLogRecord(context, event, level);
// All events should be JSON serializable, since we check it when sending events to context:
// RunnerContextImpl.sendEvent
JsonNode tree = MAPPER.valueToTree(record);
if (level == EventLogLevel.STANDARD && truncator != null) {
JsonNode eventNode = tree.get("event");
if (eventNode instanceof ObjectNode) {
boolean truncated = truncator.truncate((ObjectNode) eventNode);
if (truncated && truncatedEventsCounter != null) {
truncatedEventsCounter.inc();
}
}
}
String json =
prettyPrint
? MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(tree)
: MAPPER.writeValueAsString(tree);
writer.println(json);
}
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on PR #609:
URL: https://github.com/apache/flink-agents/pull/609#issuecomment-4261859834
@wenjin272 Thanks for the review! I’ve addressed your comments - could you take another look?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #609:
URL: https://github.com/apache/flink-agents/pull/609#issuecomment-4396480486
Hi, @weiqingy. Due to the refactoring of Event in #631, this PR needs to resolve some conflicts. Mainly about the SerDer for Event.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3162076092
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/JsonTruncator.java:
##########
@@ -0,0 +1,269 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.agents.runtime.eventlog;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Truncates a Jackson {@link JsonNode} tree per configurable thresholds.
+ *
+ * <p>Three truncation strategies are applied in a single recursive pass:
+ *
+ * <ul>
+ * <li><b>String truncation</b>: Strings longer than {@code maxStringLength} are replaced with a
+ * wrapper: {@code {"truncatedString": "first N chars...", "omittedChars": M}}
+ * <li><b>Array truncation</b>: Arrays larger than {@code maxArrayElements} are replaced with a
+ * wrapper: {@code {"truncatedList": [first N elements], "omittedElements": M}}
+ * <li><b>Depth truncation</b>: At max depth, object nodes retain only scalar fields; nested
+ * objects/arrays are dropped: {@code {"truncatedObject": {scalars...}, "omittedFields": N}}
+ * </ul>
+ *
+ * <p>Setting any threshold to {@code 0} disables that specific truncation strategy. If all
+ * thresholds are {@code 0}, no truncation occurs.
+ *
+ * <p>Protected fields at the top level of the event node ({@code eventType}, {@code id}, {@code
+ * attributes}) are never truncated.
+ */
+public class JsonTruncator {
+
+ private static final Set<String> PROTECTED_FIELDS =
+ new HashSet<>(Arrays.asList("eventType", "id", "attributes"));
+
+ private final int maxStringLength;
+ private final int maxArrayElements;
+ private final int maxDepth;
+
+ /**
+ * Creates a new truncator with the given thresholds.
+ *
+ * @param maxStringLength maximum character length for string values; 0 to disable
+ * @param maxArrayElements maximum number of array elements retained; 0 to disable
+ * @param maxDepth maximum object nesting depth; 0 to disable
+ */
+ public JsonTruncator(int maxStringLength, int maxArrayElements, int maxDepth) {
+ this.maxStringLength = maxStringLength;
+ this.maxArrayElements = maxArrayElements;
+ this.maxDepth = maxDepth;
+ }
+
+ /**
+ * Truncates the given event node in place according to configured thresholds.
+ *
+ * <p>Protected fields ({@code eventType}, {@code id}, {@code attributes}) at the top level of
+ * the event node are never truncated.
+ *
+ * @param eventNode the top-level event JSON object to truncate
+ * @return {@code true} if any field was truncated, {@code false} if the node was unchanged
+ */
+ public boolean truncate(ObjectNode eventNode) {
+ if (eventNode == null) {
+ return false;
+ }
+ return truncateObject(eventNode, 1, true);
+ }
+
+ /**
+ * Recursively truncates an object node.
+ *
+ * @param node the object node to process
+ * @param depth current depth (1 = top-level event node)
+ * @param isTopLevel whether this is the top-level event node (for protected field checks)
+ * @return true if any truncation occurred
+ */
+ private boolean truncateObject(ObjectNode node, int depth, boolean isTopLevel) {
+ boolean truncated = false;
+
+ // At max depth, collapse the entire object to retain only scalars
+ if (maxDepth > 0 && depth >= maxDepth) {
+ return collapseAtMaxDepth(node, isTopLevel);
+ }
+
+ List<String> fieldNames = new ArrayList<>();
+ node.fieldNames().forEachRemaining(fieldNames::add);
+
+ for (String fieldName : fieldNames) {
+ if (isTopLevel && PROTECTED_FIELDS.contains(fieldName)) {
+ continue;
+ }
+
+ JsonNode child = node.get(fieldName);
+ if (child == null) {
+ continue;
+ }
+
+ if (child.isTextual()) {
+ JsonNode replacement = truncateString(child.textValue());
+ if (replacement != null) {
+ node.set(fieldName, replacement);
+ truncated = true;
+ }
+ } else if (child.isArray()) {
+ // First recurse into retained elements, then truncate the array if needed
+ truncated |= truncateArrayContents((ArrayNode) child, depth + 1);
+ JsonNode replacement = truncateArray((ArrayNode) child);
Review Comment:
Good call. In `c4596bc` `EventLogRecord` is reduced to just `(context, event)`; `logLevel`, `eventType`, `truncator`, and `truncatedEventsCounter` are all gone from the record.
Truncation orchestration and the counter live in `FileEventLogger.append()` now, and the `logLevel` field in the on-disk JSON is emitted by the logger at write time rather than carried on the record.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3094860436
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogRecordJsonSerializer.java:
##########
@@ -67,15 +70,37 @@ public void serialize(EventLogRecord record, JsonGenerator gen, SerializerProvid
gen.writeStartObject();
gen.writeStringField("timestamp", record.getContext().getTimestamp());
+ gen.writeStringField(
+ "logLevel", record.getLogLevel() != null ? record.getLogLevel().name() : "VERBOSE");
+ // For PythonEvent, use the Python module path as the top-level eventType
+ // (consistent with the eventType inside the event object)
+ String topLevelEventType = record.getContext().getEventType();
+ if (record.getEvent() instanceof PythonEvent) {
+ String pythonType = ((PythonEvent) record.getEvent()).getEventType();
+ if (pythonType != null) {
+ topLevelEventType = pythonType;
+ }
+ }
Review Comment:
Extracted `EventUtil.resolveEventType(Event, EventContext)` and replaced the duplicated logic in both `FileEventLogger.append()` and `EventLogRecordJsonSerializer.serialize()`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3094888844
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -342,6 +342,10 @@ private void initEventLogger(StreamingRuntimeContext runtimeContext) throws Exce
return;
}
eventLogger.open(new EventLoggerOpenParams(runtimeContext));
+ if (eventLogger instanceof FileEventLogger) {
Review Comment:
The `instanceof` check is needed because `setTruncatedEventsCounter()` is specific to `FileEventLogger` — it's not on the `EventLogger` interface. The truncation counter is registered centrally in `BuiltInMetrics` (alongside `numOfEventProcessed`, `numOfActionsExecuted`, etc.) so all operator metrics live in one place under the same `FlinkAgentsMetricGroupImpl` hierarchy. The operator then passes the counter reference to `FileEventLogger` after `open()`.
We considered adding the counter to the `EventLogger` interface or having the logger self-register, but both options would either leak Flink metrics concerns into the public API or break the centralized metrics pattern. The `instanceof` check keeps the interface clean while allowing `FileEventLogger` to participate in the operator's metrics.
Open to suggestions if you'd prefer a different approach here.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3094884132
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogLevelResolver.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.agents.runtime.eventlog;
+
+import org.apache.flink.agents.api.logger.EventLogLevel;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Resolves the effective {@link EventLogLevel} for a given event type string using hierarchical
+ * config key inheritance.
+ *
+ * <p>Resolution order for a given event type (e.g., {@code org.apache.flink.agents.api.event
+ * .ChatRequestEvent}):
+ *
+ * <ol>
+ * <li>Exact match in explicit per-type configuration
+ * <li>Walk up dot-separated segments (e.g., {@code org.apache.flink.agents.api.event}, then
+ * {@code org.apache.flink.agents.api}, etc.)
+ * <li>Root default from {@code event-log.level} config key
+ * <li>Built-in default: {@link EventLogLevel#STANDARD}
+ * </ol>
+ *
+ * <p>This mirrors Log4j's hierarchical logger configuration pattern. Resolved levels are cached in
+ * a {@link ConcurrentHashMap} for efficient repeated lookups.
+ *
+ * <p>Config keys are expected in the form:
+ *
+ * <pre>
+ * event-log.level = STANDARD (root default)
+ * event-log.type.<EVENT_TYPE>.level = OFF (per-type override)
+ * </pre>
+ */
+public class EventLogLevelResolver {
+
+ /** Config key for the root default log level. */
+ static final String ROOT_LEVEL_KEY = "event-log.level";
+
+ /** Prefix for per-event-type log level config keys. */
+ static final String TYPE_PREFIX = "event-log.type.";
+
+ /** Suffix for per-event-type log level config keys. */
+ static final String TYPE_SUFFIX = ".level";
+
+ /** Built-in default when no configuration is provided. */
+ static final EventLogLevel BUILT_IN_DEFAULT = EventLogLevel.STANDARD;
+
+ private final EventLogLevel rootDefault;
+ private final Map<String, EventLogLevel> explicitLevels;
+ private final ConcurrentHashMap<String, EventLogLevel> cache;
+
+ /**
+ * Creates a resolver from a configuration data map.
+ *
+ * <p>The map is scanned for keys matching {@code event-log.type.<EVENT_TYPE>.level} to build
+ * the explicit per-type level mappings, and {@code event-log.level} for the root default.
+ *
+ * @param confData the flat configuration key-value map (e.g., from {@code
+ * AgentConfiguration.getConfData()})
+ */
+ public EventLogLevelResolver(Map<String, Object> confData) {
+ Map<String, Object> data = confData != null ? confData : Collections.emptyMap();
+
+ // Parse root default
+ Object rootValue = data.get(ROOT_LEVEL_KEY);
+ this.rootDefault =
+ rootValue != null
+ ? EventLogLevel.fromString(rootValue.toString())
+ : BUILT_IN_DEFAULT;
Review Comment:
Done. Replaced `BUILT_IN_DEFAULT` with `AgentConfigOptions.EVENT_LOG_LEVEL.getDefaultValue()`. Both removed constants are now derived from the single config option definition.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3094855905
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogLevelResolver.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.agents.runtime.eventlog;
+
+import org.apache.flink.agents.api.logger.EventLogLevel;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Resolves the effective {@link EventLogLevel} for a given event type string using hierarchical
+ * config key inheritance.
+ *
+ * <p>Resolution order for a given event type (e.g., {@code org.apache.flink.agents.api.event
+ * .ChatRequestEvent}):
+ *
+ * <ol>
+ * <li>Exact match in explicit per-type configuration
+ * <li>Walk up dot-separated segments (e.g., {@code org.apache.flink.agents.api.event}, then
+ * {@code org.apache.flink.agents.api}, etc.)
+ * <li>Root default from {@code event-log.level} config key
+ * <li>Built-in default: {@link EventLogLevel#STANDARD}
+ * </ol>
+ *
+ * <p>This mirrors Log4j's hierarchical logger configuration pattern. Resolved levels are cached in
+ * a {@link ConcurrentHashMap} for efficient repeated lookups.
+ *
+ * <p>Config keys are expected in the form:
+ *
+ * <pre>
+ * event-log.level = STANDARD (root default)
+ * event-log.type.<EVENT_TYPE>.level = OFF (per-type override)
+ * </pre>
+ */
+public class EventLogLevelResolver {
+
+ /** Config key for the root default log level. */
+ static final String ROOT_LEVEL_KEY = "event-log.level";
+
+ /** Prefix for per-event-type log level config keys. */
+ static final String TYPE_PREFIX = "event-log.type.";
+
+ /** Suffix for per-event-type log level config keys. */
+ static final String TYPE_SUFFIX = ".level";
+
+ /** Built-in default when no configuration is provided. */
+ static final EventLogLevel BUILT_IN_DEFAULT = EventLogLevel.STANDARD;
+
+ private final EventLogLevel rootDefault;
+ private final Map<String, EventLogLevel> explicitLevels;
+ private final ConcurrentHashMap<String, EventLogLevel> cache;
+
+ /**
+ * Creates a resolver from a configuration data map.
+ *
+ * <p>The map is scanned for keys matching {@code event-log.type.<EVENT_TYPE>.level} to build
+ * the explicit per-type level mappings, and {@code event-log.level} for the root default.
+ *
+ * @param confData the flat configuration key-value map (e.g., from {@code
+ * AgentConfiguration.getConfData()})
+ */
+ public EventLogLevelResolver(Map<String, Object> confData) {
+ Map<String, Object> data = confData != null ? confData : Collections.emptyMap();
+
+ // Parse root default
+ Object rootValue = data.get(ROOT_LEVEL_KEY);
Review Comment:
Good suggestion. Changed to `ConfigOption<EventLogLevel>` with `EventLogLevel.STANDARD` as the default. `AgentConfiguration.get()` already supports enum types via `ConfigurationUtils.convertValue()`, so this works end-to-end.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3094877516
##########
api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java:
##########
@@ -54,4 +54,32 @@ public class AgentConfigOptions {
/** The config parameter specifies the unique identifier of job. */
public static final ConfigOption<String> JOB_IDENTIFIER =
new ConfigOption<>("job-identifier", String.class, null);
+
+ /**
+ * The global event log level controlling the default verbosity for all event types. Valid
+ * values are "OFF", "STANDARD", and "VERBOSE". Defaults to "STANDARD".
+ */
+ public static final ConfigOption<String> EVENT_LOG_LEVEL =
+ new ConfigOption<>("event-log.level", String.class, "STANDARD");
Review Comment:
Updated the PR.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on PR #609:
URL: https://github.com/apache/flink-agents/pull/609#issuecomment-4423044191
Thanks @wenjin272 @xintongsong . I’ve updated the PR to resolve the branch conflicts.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3162068287
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogRecord.java:
##########
@@ -31,23 +33,82 @@
*
* <p>The class uses custom JSON serialization/deserialization to handle polymorphic Event types by
* leveraging the eventType information stored in the EventContext.
+ *
+ * <p>Each record carries a {@link EventLogLevel} that controls truncation behavior during
+ * serialization, and a convenience copy of the event type string from the context.
*/
@JsonSerialize(using = EventLogRecordJsonSerializer.class)
@JsonDeserialize(using = EventLogRecordJsonDeserializer.class)
public class EventLogRecord {
private final EventContext context;
private final Event event;
+ private final EventLogLevel logLevel;
+ private final String eventType;
+ private final transient JsonTruncator truncator;
+ private final transient Counter truncatedEventsCounter;
Review Comment:
Thanks for the concrete sketch — followed it in `c4596bc`, and combined it with Xintong's broader concern to drop `logLevel` from `EventLogRecord` as well, since it shares the same "not part of the event content" character as the truncator. `EventLogRecord` is now a pure `(context, event)` carrier, and `FileEventLogger.append()` owns truncation orchestration plus injection of the `logLevel` field on the way out.
The on-disk JSON layout is unchanged — entries still serialize as `timestamp, logLevel, eventType, event` — and the deserializer silently ignores any `logLevel` field present in older log files, so existing logs keep replaying without changes.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3162098193
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogRecord.java:
##########
@@ -31,23 +35,82 @@
*
* <p>The class uses custom JSON serialization/deserialization to handle polymorphic Event types by
* leveraging the eventType information stored in the EventContext.
+ *
+ * <p>Each record carries a {@link EventLogLevel} that controls truncation behavior during
+ * serialization, and a convenience copy of the event type string from the context.
*/
@JsonSerialize(using = EventLogRecordJsonSerializer.class)
@JsonDeserialize(using = EventLogRecordJsonDeserializer.class)
public class EventLogRecord {
private final EventContext context;
private final Event event;
+ private final EventLogLevel logLevel;
+ private final String eventType;
+ @Nullable private final transient JsonTruncator truncator;
+ @Nullable private final transient Counter truncatedEventsCounter;
Review Comment:
Good call. In `c4596bc` `EventLogRecord` is reduced to just `(context, event)`; `logLevel`, `eventType`, `truncator`, and `truncatedEventsCounter` are all gone from the record.
Truncation orchestration and the counter live in `FileEventLogger.append()` now, and the `logLevel` field in the on-disk JSON is emitted by the logger at write time rather than carried on the record.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on PR #609:
URL: https://github.com/apache/flink-agents/pull/609#issuecomment-4197100878
Hey @wenjin272 @xintongsong could you please take a look at the PR? Feedback is welcome. 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime] Support per-event-type configurable log levels for event log [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #609:
URL: https://github.com/apache/flink-agents/pull/609#discussion_r3094860436
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogRecordJsonSerializer.java:
##########
@@ -67,15 +70,37 @@ public void serialize(EventLogRecord record, JsonGenerator gen, SerializerProvid
gen.writeStartObject();
gen.writeStringField("timestamp", record.getContext().getTimestamp());
+ gen.writeStringField(
+ "logLevel", record.getLogLevel() != null ? record.getLogLevel().name() : "VERBOSE");
+ // For PythonEvent, use the Python module path as the top-level eventType
+ // (consistent with the eventType inside the event object)
+ String topLevelEventType = record.getContext().getEventType();
+ if (record.getEvent() instanceof PythonEvent) {
+ String pythonType = ((PythonEvent) record.getEvent()).getEventType();
+ if (pythonType != null) {
+ topLevelEventType = pythonType;
+ }
+ }
Review Comment:
Sounds good. Extracted `EventUtil.resolveEventType(Event, EventContext)` and replaced the duplicated logic in both `FileEventLogger.append()` and `EventLogRecordJsonSerializer.serialize()`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org