alnzng commented on code in PR #103: URL: https://github.com/apache/flink-agents/pull/103#discussion_r2281398780
########## examples/src/main/java/org/apache/flink/agents/examples/EventLoggingAndListenerExample.java: ########## @@ -0,0 +1,293 @@ +/* + * 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.examples; + +import org.apache.flink.agents.api.Agent; +import org.apache.flink.agents.api.AgentsExecutionEnvironment; +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.EventContext; +import org.apache.flink.agents.api.EventFilter; +import org.apache.flink.agents.api.InputEvent; +import org.apache.flink.agents.api.OutputEvent; +import org.apache.flink.agents.api.annotation.Action; +import org.apache.flink.agents.api.context.RunnerContext; +import org.apache.flink.agents.api.listener.EventListener; +import org.apache.flink.agents.runtime.logger.FileEventLoggerConfig; +import org.apache.flink.api.java.functions.KeySelector; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; + +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Example demonstrating EventLogger and EventListener usage with Flink Agents. + * + * <p>This example shows how to: + * + * <ul> + * <li>Configure file-based event logging with filtering + * <li>Register custom event listeners for monitoring and metrics + * <li>Process streaming data through agents + * <li>Track event processing statistics + * <li>Log events to files for debugging and analysis + * </ul> + * + * <p>The example processes a stream of order data, applies business logic through an agent, and + * demonstrates comprehensive event monitoring and logging capabilities. + */ +public class EventLoggingAndListenerExample { + + /** Order data class for the example. */ + public static class OrderData { + public final int orderId; + public final String customerId; + public final double amount; + public final String status; + + public OrderData(int orderId, String customerId, double amount, String status) { + this.orderId = orderId; + this.customerId = customerId; + this.amount = amount; + this.status = status; + } + + @Override + public String toString() { + return String.format( + "OrderData{orderId=%d, customerId='%s', amount=%.2f, status='%s'}", + orderId, customerId, amount, status); + } + } + + /** Key selector for extracting customer ID from OrderData. */ + public static class OrderKeySelector implements KeySelector<OrderData, String> { + @Override + public String getKey(OrderData order) { + return order.customerId; + } + } + + /** Custom event for order processing results. */ + public static class OrderProcessedEvent extends Event { + private final String orderId; + private final String result; + private final double processedAmount; + + public OrderProcessedEvent(String orderId, String result, double processedAmount) { + this.orderId = orderId; + this.result = result; + this.processedAmount = processedAmount; + } + + public String getOrderId() { + return orderId; + } + + public String getResult() { + return result; + } + + public double getProcessedAmount() { + return processedAmount; + } + + @Override + public String toString() { + return String.format( + "OrderProcessedEvent{orderId='%s', result='%s', processedAmount=%.2f}", + orderId, result, processedAmount); + } + } + + /** Agent that processes orders with business logic. */ + public static class OrderProcessingAgent extends Agent { + + /** + * Action that processes incoming order events. + * + * @param event The input event containing order data + * @param ctx The runner context for sending events + */ + @Action(listenEvents = {InputEvent.class}) + public void processOrder(Event event, RunnerContext ctx) { + InputEvent inputEvent = (InputEvent) event; + OrderData order = (OrderData) inputEvent.getInput(); + + // Business logic: validate and process the order + String result; + double processedAmount = order.amount; + + if (order.amount > 1000.0) { + result = "HIGH_VALUE_ORDER"; + processedAmount = order.amount * 0.95; // 5% discount for high-value orders + } else if (order.amount > 100.0) { + result = "STANDARD_ORDER"; + processedAmount = order.amount * 0.98; // 2% discount for standard orders + } else { + result = "SMALL_ORDER"; + // No discount for small orders + } + + // Send processed event for further handling + ctx.sendEvent( + new OrderProcessedEvent( + String.valueOf(order.orderId), result, processedAmount)); + } + + /** + * Action that handles processed orders and generates final output. + * + * @param event The processed order event + * @param ctx The runner context for sending events + */ + @Action(listenEvents = {OrderProcessedEvent.class}) + public void generateOrderOutput(Event event, RunnerContext ctx) { + OrderProcessedEvent processedEvent = (OrderProcessedEvent) event; + + // Generate final output with order summary + String output = + String.format( + "Order %s processed: %s - Final amount: $%.2f", + processedEvent.getOrderId(), + processedEvent.getResult(), + processedEvent.getProcessedAmount()); + + ctx.sendEvent(new OutputEvent(output)); + } + } + + /** Custom event listener for monitoring and metrics collection. */ + public static class OrderMetricsListener implements EventListener { + private final AtomicInteger totalEvents = new AtomicInteger(0); + private final AtomicInteger inputEvents = new AtomicInteger(0); + private final AtomicInteger processedEvents = new AtomicInteger(0); + private final AtomicInteger outputEvents = new AtomicInteger(0); + + @Override + public void onEventProcessed(EventContext context, Event event) { + totalEvents.incrementAndGet(); + + if (event instanceof InputEvent) { + inputEvents.incrementAndGet(); + System.out.printf( + "[METRICS] Input Event processed: %s (Total inputs: %d)%n", + event, inputEvents.get()); + } else if (event instanceof OrderProcessedEvent) { + processedEvents.incrementAndGet(); + OrderProcessedEvent orderEvent = (OrderProcessedEvent) event; + System.out.printf( + "[METRICS] Order processed: %s with result %s (Total processed: %d)%n", + orderEvent.getOrderId(), orderEvent.getResult(), processedEvents.get()); + } else if (event instanceof OutputEvent) { + outputEvents.incrementAndGet(); + System.out.printf( + "[METRICS] Output Event generated: %s (Total outputs: %d)%n", + event, outputEvents.get()); + } + + // Print summary every 5 events + if (totalEvents.get() % 5 == 0) { + printSummary(); + } + } + + private void printSummary() { + System.out.printf( + "[METRICS SUMMARY] Total: %d | Inputs: %d | Processed: %d | Outputs: %d%n", + totalEvents.get(), + inputEvents.get(), + processedEvents.get(), + outputEvents.get()); + } + } + + /** Event listener for debugging and detailed event tracking. */ + public static class DebugEventListener implements EventListener { + @Override + public void onEventProcessed(EventContext context, Event event) { + System.out.printf( + "[DEBUG] Event processed at %s: %s (Type: %s, ID: %s)%n", + context.getTimestamp(), event, event.getClass().getSimpleName(), event.getId()); + + // Log context information + System.out.printf("[DEBUG] Event context type: %s%n", context.getEventType()); + } + } + + public static void main(String[] args) throws Exception { + // Create the execution environment + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + + // Create input DataStream with sample order data + DataStream<OrderData> inputStream = + env.fromElements( + new OrderData(1001, "customer_A", 150.0, "pending"), + new OrderData(1002, "customer_B", 75.0, "pending"), + new OrderData(1003, "customer_A", 1200.0, "pending"), + new OrderData(1004, "customer_C", 300.0, "pending"), + new OrderData(1005, "customer_B", 50.0, "pending"), + new OrderData(1006, "customer_A", 800.0, "pending"), + new OrderData(1007, "customer_D", 2000.0, "pending")); + + // Configure file-based event logging with filtering + FileEventLoggerConfig loggerConfig = + FileEventLoggerConfig.builder() + .baseEventLogDir("/tmp/flink-agents-logs") + .eventFilter( + EventFilter.byEventType( + InputEvent.class, + OrderProcessedEvent.class, + OutputEvent.class)) + .build(); Review Comment: I'm ok with not exposing Event Logger/Listener as programming APIs, since I can't see any reason we have to keep it as programming APIs. And one potential benefit is this makes flink-agent programming APIs as (agent features) focused as possible. And yes, let's enable this by default, and expose it to end users once we have the new configuration mechanism. ########## api/src/main/java/org/apache/flink/agents/api/logger/EventLogRecord.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.logger; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.EventContext; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize; + +/** + * Represents a record in the event log, containing the event context and the event itself. + * + * <p>This class is used to encapsulate the details of an event as it is logged, allowing for + * structured logging and retrieval of event information. + * + * <p>The class uses custom JSON serialization/deserialization to handle polymorphic Event types by + * leveraging the eventType information stored in the EventContext. + */ +@JsonSerialize(using = EventLogRecordJsonSerializer.class) +@JsonDeserialize(using = EventLogRecordJsonDeserializer.class) +public class EventLogRecord { Review Comment: Sounds good. Moving this to `runtime` module, but I would prefer adding it into a dedicated folder `eventlog` to group all the classes related to `event log` together. ########## api/src/main/java/org/apache/flink/agents/api/logger/EventLoggerFactory.java: ########## @@ -0,0 +1,192 @@ +/* + * 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 java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Function; + +/** + * Factory for creating EventLogger instances based on EventLoggerConfig types. + * + * <p>This factory uses a registry-based approach that allows both built-in and custom EventLogger + * implementations to be created in a type-safe manner. Built-in loggers are automatically + * registered, while custom loggers can be registered using the {@link #registerFactory(Class, + * Function)} method. + * + * <h3>Thread Safety</h3> + * + * <p>This factory is thread-safe and can be used concurrently from multiple threads. Factory + * registration and logger creation operations are atomic and do not require external + * synchronization. + * + * <h3>Usage Examples</h3> + * + * <pre>{@code + * // Create built-in file logger + * FileEventLoggerConfig fileConfig = FileEventLoggerConfig.builder() + * .logFileDir("/tmp/flink-agents") + * .build(); + * EventLogger fileLogger = EventLoggerFactory.createLogger(fileConfig); + * + * // Register and use custom logger + * EventLoggerFactory.registerFactory(MyCustomConfig.class, + * config -> new MyCustomLogger((MyCustomConfig) config)); + * + * MyCustomConfig customConfig = new MyCustomConfig(...); + * EventLogger customLogger = EventLoggerFactory.createLogger(customConfig); + * }</pre> + * + * @see EventLogger + * @see EventLoggerConfig + */ +public final class EventLoggerFactory { + + /** Thread-safe registry of factory functions keyed by config class type. */ + private static final Map< + Class<? extends EventLoggerConfig>, Function<EventLoggerConfig, EventLogger>> + FACTORIES = new ConcurrentHashMap<>(); Review Comment: Great suggestion. I like this idea "offerring an unitified EventLoggerConfig". ########## runtime/src/main/java/org/apache/flink/agents/runtime/logger/FileEventLogger.java: ########## @@ -0,0 +1,134 @@ +/* + * 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.logger; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.EventContext; +import org.apache.flink.agents.api.EventFilter; +import org.apache.flink.agents.api.logger.EventLogRecord; +import org.apache.flink.agents.api.logger.EventLogger; +import org.apache.flink.agents.api.logger.EventLoggerOpenParams; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; + +import java.io.BufferedWriter; +import java.io.FileWriter; +import java.io.PrintWriter; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; + +/** + * A file-based event logger that logs events to a file specific to each subtask. + * + * <p>This logger creates a unique directory structure for each subtask based on the job name and + * subtask ID, ensuring no file conflicts in multi-TaskManager deployments. Events are appended to a + * log file named "events.log" within that directory in JSON Lines format. + * + * <h3>Thread Safety</h3> + * + * <p>This class is <strong>thread-safe at the Flink subtask level</strong>. Flink's execution model + * guarantees that each subtask instance processes events in a single-threaded manner within the + * operator's mailbox thread. This means: + * + * <ul> + * <li>No synchronization is needed for concurrent access within a subtask + * <li>Each subtask instance gets its own logger instance and unique log file + * <li>Multiple subtasks can run concurrently without file conflicts + * </ul> + * + * <h3>File Structure</h3> + * + * <p>The logger creates the following directory structure: + * + * <pre> + * {baseLogDir}/ + * └── {jobId}/ + * ├── 0/ + * │ └── events.log (subtask 0) + * ├── 1/ + * │ └── events.log (subtask 1) + * └── 2/ + * └── events.log (subtask 2) Review Comment: IIUC, the existing implementation doesn't support multiple agents in a same Flink job: https://github.com/apache/flink-agents/blob/d282e6e334ad785ec42ddead0e9ce492e942ed13/runtime/src/main/java/org/apache/flink/agents/runtime/env/RemoteExecutionEnvironment.java#L108C9-L115C10 ```java public AgentBuilder apply(Agent agent) { try { this.agentPlan = new AgentPlan(agent); return this; } catch (Exception e) { throw new RuntimeException("Failed to create agent plan from agent", e); } } ``` but I think you are right on that we should support multiple agents in one Flink job, this seems to be a gap in current implementation. Let me add task name to differeniate the logs from different tasks. And thanks for the suggestion that using flatten way to organize the log files. -- 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]
