fix: also switch to duration in nanos in frontend

Signed-off-by: Felix Hilgers <[email protected]>

frontend/app/src/main/java/de/amosproj3/ziofa/api/WriteEvent.kt

@@ -23,6 +23,6 @@ sealed class WriteEvent(
         val pid: UInt,
         val tid: UInt,
         val beginTimestamp: ULong,
-        val durationMicros: ULong,
-    ) : WriteEvent(fd, pid, beginTimestamp, durationMicros)
+        val durationNanos: ULong,
+    ) : WriteEvent(fd, pid, beginTimestamp, durationNanos)
 }

frontend/app/src/main/java/de/amosproj3/ziofa/bl/DataStreamManager.kt

@@ -10,10 +10,12 @@ import de.amosproj3.ziofa.api.WriteEvent
 import de.amosproj3.ziofa.client.ClientFactory
 import de.amosproj3.ziofa.client.Event
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.mapNotNull
 import timber.log.Timber
 
+// TODO: use a single sharedFlow and then different filters on top of that
+// otherwise we are sending all the data multiple times from server to client
 class DataStreamManager(private val clientFactory: ClientFactory) : DataStreamProvider {
 
     override suspend fun counter(ebpfProgramName: String): Flow<UInt> {
@@ -36,27 +38,28 @@ class DataStreamManager(private val clientFactory: ClientFactory) : DataStreamPr
         clientFactory
             .connect()
             .initStream()
-            .filter { it is Event.VfsWrite }
+            .mapNotNull { it as? Event.VfsWrite }
             .map {
-                if (it is Event.VfsWrite) {
-                    WriteEvent.VfsWriteEvent(it.fp, it.pid, it.bytesWritten, it.beginTimeStamp)
-                } else throw Exception("only for the compiler")
+                WriteEvent.VfsWriteEvent(
+                    fd = it.fp,
+                    pid = it.pid,
+                    size = it.bytesWritten,
+                    timestampMillis = it.beginTimeStamp
+                )
             }
 
     override suspend fun sendMessageEvents(pids: List<UInt>): Flow<WriteEvent.SendMessageEvent> =
         clientFactory
             .connect()
             .initStream()
-            .filter { it is Event.SysSendmsg }
+            .mapNotNull { it as? Event.SysSendmsg }
             .map {
-                if (it is Event.SysSendmsg) {
-                    WriteEvent.SendMessageEvent(
-                        it.fd.toULong(),
-                        it.pid,
-                        it.tid,
-                        it.beginTimeStamp,
-                        it.durationMicroSec,
-                    )
-                } else throw Exception("only for the compiler")
+                WriteEvent.SendMessageEvent(
+                    fd = it.fd,
+                    pid = it.pid,
+                    tid = it.tid,
+                    beginTimestamp = it.beginTimeStamp,
+                    durationNanos = it.durationNanoSecs,
+                )
             }
 }

frontend/app/src/main/java/de/amosproj3/ziofa/ui/visualization/composables/EventList.kt

@@ -39,7 +39,7 @@ fun EventList(events: List<WriteEvent>) {
                 when (event) {
                     is WriteEvent.SendMessageEvent -> {
                         Text(
-                            text = (event.durationMicros / 1_000u).toString(),
+                            text = (event.durationNanos / 1_000_000u).toString(),
                             modifier = Modifier.weight(1f),
                         )
                     }

frontend/client/src/main/java/de/amosproj3/ziofa/client/Client.kt

@@ -31,8 +31,8 @@ sealed class Event {
         val pid: UInt,
         val tid: UInt,
         val beginTimeStamp: ULong,
-        val fd: Int,
-        val durationMicroSec: ULong,
+        val fd: ULong,
+        val durationNanoSecs: ULong,
     ) : Event()
 }
 

frontend/client/src/mock/java/de/amosproj3/ziofa/client/RustClient.kt

@@ -94,8 +94,8 @@ object RustClient : Client {
                 Event.SysSendmsg(
                     pid = 12345u,
                     tid = 1234u,
-                    fd = 125123123,
-                    durationMicroSec =
+                    fd = 125123123u,
+                    durationNanoSecs =
                         (System.currentTimeMillis() + Random.nextLong(1000)).toULong(),
                     beginTimeStamp = System.currentTimeMillis().toULong(),
                 )

rust/backend/daemon/src/collector.rs

@@ -54,7 +54,7 @@ impl CollectFromMap for SysSendmsgCollect {
                 tid: data.tid,
                 begin_time_stamp: data.begin_time_stamp,
                 fd: data.fd,
-                duration_micro_sec: data.duration_micro_sec
+                duration_nano_sec: data.duration_nano_sec
             }))
         })
     }

rust/backend/daemon/src/features.rs

@@ -65,7 +65,7 @@ impl SysSendmsgFeature {
     }
 
     fn update_pids(&mut self, ebpf: &mut Ebpf, pids: &[u32]) -> Result<(), EbpfError> {
-        let mut pids_to_track: HashMap<_, u32, u32> = ebpf.map_mut("PIDS_TO_TRACK")
+        let mut pids_to_track: HashMap<_, u32, u64> = ebpf.map_mut("PIDS_TO_TRACK")
             .ok_or(EbpfError::MapError(
                 aya::maps::MapError::InvalidName {
                     name: "PIDS_TO_TRACK".to_string(),
@@ -242,4 +242,4 @@ impl VfsFeature {
 
         Ok(())
     }
-}
+}

rust/shared/proto/ziofa.proto

@@ -68,5 +68,5 @@ message SysSendmsgEvent {
     uint32 tid = 2;
     uint64 begin_time_stamp = 3;
     uint64 fd = 4;
-    uint64 duration_micro_sec = 5;
+    uint64 duration_nano_sec = 5;
 }