* log-exporter: export action and event in parquet format

Signed-off-by: neo <1100909+neowu@users.noreply.github.com>
neowu · Mar 3, 2025 · 7972e87 · 7972e87
1 parent 167d454
commit 7972e87
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 3 deletions.
diff --git a/ext/log-exporter/src/main/java/core/log/domain/ActionLogSchema.java b/ext/log-exporter/src/main/java/core/log/domain/ActionLogSchema.java
@@ -59,7 +59,7 @@ var record = new GenericData.Record(schema);
             if (stat.readEntries != null) perfStats.put(key + ".read_entries", (long) stat.readEntries);
             if (stat.writeEntries != null) perfStats.put(key + ".write_entries", (long) stat.writeEntries);
         }
-        record.put("perf_stats", message.performanceStats);
+        record.put("perf_stats", perfStats);
         return record;
     }
 }
diff --git a/ext/log-exporter/src/test/java/core/log/kafka/ActionLogMessageHandlerTest.java b/ext/log-exporter/src/test/java/core/log/kafka/ActionLogMessageHandlerTest.java
@@ -3,6 +3,7 @@
 import core.framework.inject.Inject;
 import core.framework.kafka.Message;
 import core.framework.log.message.ActionLogMessage;
+import core.framework.log.message.PerformanceStatMessage;
 import core.framework.util.Files;
 import core.log.IntegrationTest;
 import core.log.domain.ActionLogSchema;
@@ -20,9 +21,9 @@
  * @author neo
  */
 class ActionLogMessageHandlerTest extends IntegrationTest {
-    private ActionLogMessageHandler handler;
     @Inject
     ActionLogSchema schema;
+    private ActionLogMessageHandler handler;
 
     @BeforeEach
     void createActionLogMessageHandler() {
@@ -48,9 +49,19 @@ void handle() throws IOException {
         message.host = "host";
         message.elapsed = 1000L;
         message.context = Map.of("key", List.of("value1", "value2"));
-        message.performanceStats = Map.of();
+        message.performanceStats = Map.of("kafka", perfStats(1, 1000L, 10, 5),
+            "http", perfStats(1, 2000L, null, null));
         message.traceLog = "trace";
         handler.handle(List.of(new Message<>("key", message)));
         handler.handle(List.of(new Message<>("key", message)));
     }
+
+    private PerformanceStatMessage perfStats(int count, long totalElapsed, Integer readEntries, Integer writeEntries) {
+        final PerformanceStatMessage stats = new PerformanceStatMessage();
+        stats.count = count;
+        stats.totalElapsed = totalElapsed;
+        stats.readEntries = readEntries;
+        stats.writeEntries = writeEntries;
+        return stats;
+    }
 }