xintongsong commented on code in PR #103:
URL: https://github.com/apache/flink-agents/pull/103#discussion_r2266460719
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -176,13 +184,26 @@ public void open() throws Exception {
mailboxProcessor = getMailboxProcessor();
+ // Initialize the event logger if it is set.
+ initEventLogger();
+
// Since an operator restart may change the key range it manages due
to changes in
// parallelism,
// and {@link tryProcessActionTaskForKey} mails might be lost,
// it is necessary to reprocess all keys to ensure correctness.
tryResumeProcessActionTasks();
}
+ private void initEventLogger() throws Exception {
+ if (eventLogger == null) {
+ return;
+ }
+ if (eventLogger.unwrap() instanceof FlinkStateEventLogger) {
+ ((FlinkStateEventLogger)
eventLogger).setStreamingRuntimeContext(getRuntimeContext());
+ }
Review Comment:
It seems this is the only place we need `unwrap()`. In this case, it might
be easier to pass the runtime context into `open()`. To make this extendable,
we can introduce an `EventLoggerOpenParams`, so that we can add more things in
it as needed by new event logger implementations, without modifying existing
loggers.
##########
api/src/main/java/org/apache/flink/agents/api/logger/EventLogger.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.api.logger;
+
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.EventContext;
+import org.apache.flink.agents.api.EventFilter;
+
+/**
+ * Interface for logging action events in the Flink Agents framework.
+ *
+ * <p>EventLogger provides a unified interface for capturing, filtering, and
persisting events as
+ * they flow through the agent execution pipeline. Implementations can target
different storage
+ * backends such as Flink state, external databases, or file systems.
+ */
+public interface EventLogger extends AutoCloseable {
+
+ /**
+ * Initialize the event logger.
+ *
+ * <p>This method is called once during setup to prepare the logging
infrastructure.
+ * Implementations should establish connections, initialize state, and
prepare for logging
+ * operations.
+ *
+ * @throws Exception if initialization fails
+ */
+ void open() throws Exception;
+
+ /**
+ * Appends an event along with its associated context to the logger.
+ *
+ * <p>This method is invoked for each event that passes the configured
filter. Implementations
+ * should perform the append operation efficiently, as it may be called
frequently during event
+ * processing.
+ *
+ * @param context metadata and other contextual information associated
with the event
+ * @param event the event to be logged
+ * @throws Exception if the append operation fails
+ */
+ void append(EventContext context, Event event) throws Exception;
+
+ /**
+ * Close the event logger and release resources.
+ *
+ * <p>This method is called during cleanup. Implementations should flush
any remaining records
+ * and release all resources.
+ *
+ * @throws Exception if closing fails
+ */
+ @Override
+ void close() throws Exception;
+
+ /**
+ * Unwraps the EventLogger if it is wrapped by another logger or filter.
+ *
+ * <p>This method can be overridden by implementations that wrap other
loggers or filters to
+ * return the underlying logger.
+ *
+ * @return the underlying EventLogger instance
+ */
+ default EventLogger unwrap() {
+ return this;
+ }
Review Comment:
Why do we need this?
##########
api/src/main/java/org/apache/flink/agents/api/EventContext.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.api;
+
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Context for events in the Flink Agents API.
+ *
+ * <p>EventContext provides metadata about the event, including event key,
timestamp, and additional
+ * attributes that can be used to filter or categorize events.
+ */
+public class EventContext {
+ private final Object key;
+ private final Instant timestamp;
+ private final Map<String, Object> attributes;
+
+ public EventContext(Object key) {
+ this.key = key;
+ this.timestamp = Instant.now();
+ this.attributes = new HashMap<>();
+ }
+
+ public Object getKey() {
+ return key;
+ }
+
+ public Instant getTimestamp() {
+ return timestamp;
+ }
+
+ public Map<String, Object> getAttributes() {
+ return attributes;
+ }
Review Comment:
Why do we need this?
##########
api/src/main/java/org/apache/flink/agents/api/logger/EventLogger.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.api.logger;
+
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.EventContext;
+import org.apache.flink.agents.api.EventFilter;
+
+/**
+ * Interface for logging action events in the Flink Agents framework.
+ *
+ * <p>EventLogger provides a unified interface for capturing, filtering, and
persisting events as
+ * they flow through the agent execution pipeline. Implementations can target
different storage
+ * backends such as Flink state, external databases, or file systems.
Review Comment:
I'm not sure about logging the events into the Flink state.
1. The purpose of adding event log is for users to understand what happened
in the agent. If the logs are in state, how does the user read them?
2. The size of the list state will keep increasing, since we never removed
anything from it. This could be problematic.
3. During failover, the state will be restored from a checkpoint. This means
events after the checkpoint and before the failure will be lost. (Not entirely
sure whether we need them or not.)
As the first event logger implementation, I would suggest to start with
logging the events into a file. (Just another log file of the Flink cluster).
Then in the next step we can apply rolling strategies on the files.
WDYT?
##########
api/src/main/java/org/apache/flink/agents/api/AgentsExecutionEnvironment.java:
##########
@@ -86,6 +91,64 @@ public static AgentsExecutionEnvironment
getExecutionEnvironment() {
return getExecutionEnvironment(null);
}
+ /**
+ * Enable an event logger for the execution environment.
+ *
+ * <p>This method allows users to set a custom EventLogger that will be
used to log events
+ * during agent execution.
+ *
+ * @param eventLogger Custom EventLogger instance.
+ * @return The current AgentsExecutionEnvironment instance for method
chaining.
+ */
+ public AgentsExecutionEnvironment enableEventLogger(EventLogger
eventLogger) {
+ return enableEventLogger(eventLogger, EventFilter.ACCEPT_ALL);
+ }
+
+ /**
+ * Enable an event logger with a filter for the execution environment.
+ *
+ * <p>This method allows users to set a custom EventLogger along with an
EventFilter to control
+ * which events are logged. This is useful for reducing noise and focusing
on relevant events.
+ *
+ * @param logger Custom EventLogger instance.
+ * @param filter EventFilter to apply to logged events.
+ * @return The current AgentsExecutionEnvironment instance for method
chaining.
+ */
+ public AgentsExecutionEnvironment enableEventLogger(EventLogger logger,
EventFilter filter) {
+ this.eventLogger = EventLogger.withFilter(logger, filter);
+ return this;
+ }
Review Comment:
I think we need someway to support users setting the event logger and filter
with configs. @GreatEugenius is working on the configuration mechanism of Flink
Agents. Once the mechanism is ready, I think it should be straightforward to
enable the event logger. But for the filters, we might need some further design
on how to express the filter rules with configurations.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -176,13 +184,26 @@ public void open() throws Exception {
mailboxProcessor = getMailboxProcessor();
+ // Initialize the event logger if it is set.
+ initEventLogger();
+
// Since an operator restart may change the key range it manages due
to changes in
// parallelism,
// and {@link tryProcessActionTaskForKey} mails might be lost,
// it is necessary to reprocess all keys to ensure correctness.
tryResumeProcessActionTasks();
}
+ private void initEventLogger() throws Exception {
+ if (eventLogger == null) {
+ return;
+ }
+ if (eventLogger.unwrap() instanceof FlinkStateEventLogger) {
+ ((FlinkStateEventLogger)
eventLogger).setStreamingRuntimeContext(getRuntimeContext());
+ }
Review Comment:
IIUC, the reason we need `EventFilter.ACCEPT_ALL` is also that we need to
make sure here the `eventLogger` is a wrapped one? Moving the arguments to
`open()`, we can remove `unwrap()`, as well as `ACCEPT_ALL`.
--
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]
