feat(kubegraph): begin implementation of visualizer

Cargo.toml

@@ -36,6 +36,7 @@ members = [
     "crates/kiss/monitor",
     "crates/kiss/operator",
     "crates/kubegraph/analyzer/llm",
+    "crates/kubegraph/annotator",
     "crates/kubegraph/api",
     "crates/kubegraph/connector/fake",
     "crates/kubegraph/connector/local",
@@ -47,8 +48,11 @@ members = [
     "crates/kubegraph/operator",
     "crates/kubegraph/parser",
     "crates/kubegraph/runner/simulator",
+    "crates/kubegraph/simulator",
     "crates/kubegraph/solver/ortools",
+    "crates/kubegraph/visualizer",
     "crates/kubegraph/vm/http",
+    "crates/kubegraph/vm/lazy",
     "crates/kubegraph/vm/local",
     "crates/straw/api",
     "crates/straw/provider",
@@ -113,6 +117,9 @@ deltalake = { version = "0.17", features = [
 ] }
 derivative = { version = "2.2" }
 dirs = { version = "5.0" }
+eframe = { version = "0.27" }
+egui = { version = "0.27" }
+egui_graphs = { version = "0.20" }
 email_address = { version = "0.2" }
 futures = { version = "0.3" }
 gethostname = { version = "0.4" }
@@ -176,6 +183,7 @@ ordered-float = { version = "4.2", default-features = false, features = [
     "serde",
     "std",
 ] }
+petgraph = { version = "0.6" }
 polars = { version = "0.40", features = [
     "diagonal_concat",
     "diff",

crates/kubegraph/annotator/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "kubegraph-annotator"
+
+authors = { workspace = true }
+description = { workspace = true }
+documentation = { workspace = true }
+edition = { workspace = true }
+include = { workspace = true }
+keywords = { workspace = true }
+license = { workspace = true }
+readme = { workspace = true }
+rust-version = { workspace = true }
+homepage = { workspace = true }
+repository = { workspace = true }
+version = { workspace = true }
+
+[lints]
+workspace = true
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+kubegraph-api = { path = "../api" }
+kubegraph-vm-lazy = { path = "../vm/lazy" }
+
+anyhow = { workspace = true }
+async-trait = { workspace = true }

crates/kubegraph/annotator/src/lib.rs

@@ -0,0 +1,22 @@
+use anyhow::Result;
+use async_trait::async_trait;
+use kubegraph_api::{annotator::NetworkAnnotationSpec, vm::Script};
+
+pub struct NetworkAnnotator {}
+
+#[async_trait]
+impl<G> ::kubegraph_api::annotator::NetworkAnnotator<G> for NetworkAnnotator
+where
+    G: Send,
+{
+    async fn annotate(
+        &self,
+        graph: G,
+        spec: &NetworkAnnotationSpec,
+    ) -> Result<NetworkAnnotationSpec<Script>>
+    where
+        G: 'async_trait,
+    {
+        todo!()
+    }
+}

crates/kubegraph/api/Cargo.toml

@@ -42,7 +42,14 @@ df-polars = ["dep:polars"]
 function-full = ["function-dummy"]
 function-dummy = []
 
+# Virtual Machines
+vm-entrypoint = ["ark-core"]
+
 [dependencies]
+ark-core = { path = "../../ark/core", optional = true, features = [
+    "actix-web",
+    "signal",
+] }
 ark-core-k8s = { path = "../../ark/core/k8s", features = ["data"] }
 
 anyhow = { workspace = true }

crates/kubegraph/api/src/annotator.rs

@@ -0,0 +1,49 @@
+use anyhow::Result;
+use async_trait::async_trait;
+use kube::CustomResource;
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use crate::vm::Script;
+
+#[async_trait]
+pub trait NetworkAnnotator<G>
+where
+    G: Send,
+{
+    async fn annotate(
+        &self,
+        graph: G,
+        spec: &NetworkAnnotationSpec,
+    ) -> Result<NetworkAnnotationSpec<Script>>
+    where
+        G: 'async_trait;
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema, CustomResource)]
+#[kube(
+    group = "kubegraph.ulagbulag.io",
+    version = "v1alpha1",
+    kind = "NetworkAnnotation",
+    root = "NetworkAnnotationCrd",
+    shortname = "nf",
+    namespaced,
+    printcolumn = r#"{
+        "name": "created-at",
+        "type": "date",
+        "description": "created time",
+        "jsonPath": ".metadata.creationTimestamp"
+    }"#,
+    printcolumn = r#"{
+        "name": "version",
+        "type": "integer",
+        "description": "annotation version",
+        "jsonPath": ".metadata.generation"
+    }"#
+)]
+#[serde(rename_all = "camelCase")]
+pub struct NetworkAnnotationSpec<Script = String> {
+    #[serde(default)]
+    pub filter: Option<Script>,
+    pub script: Script,
+}

crates/kubegraph/api/src/function/mod.rs

@@ -5,7 +5,7 @@ use kube::CustomResource;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 
-use crate::{graph::GraphScope, resource::NetworkResource};
+use crate::{annotator::NetworkAnnotationSpec, graph::GraphScope, resource::NetworkResource};
 
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema, CustomResource)]
 #[kube(
@@ -33,23 +33,13 @@ pub struct NetworkFunctionSpec {
     #[serde(flatten)]
     pub kind: NetworkFunctionKind,
     #[serde(flatten)]
-    pub metadata: NetworkFunctionMetadata,
+    pub metadata: NetworkAnnotationSpec,
 }
 
 impl NetworkResource for NetworkFunctionCrd {
     type Filter = ();
 }
 
-#[derive(
-    Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, JsonSchema,
-)]
-#[serde(rename_all = "camelCase")]
-pub struct NetworkFunctionMetadata<Script = String> {
-    #[serde(default)]
-    pub filter: Option<Script>,
-    pub script: Script,
-}
-
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
 #[serde(rename_all = "camelCase")]
 #[non_exhaustive]

crates/kubegraph/api/src/lib.rs

@@ -2,6 +2,7 @@
 extern crate polars as pl;
 
 pub mod analyzer;
+pub mod annotator;
 pub mod connector;
 pub mod frame;
 pub mod function;

crates/kubegraph/api/src/vm.rs

@@ -33,6 +33,49 @@ pub trait NetworkVirtualMachineExt
 where
     Self: NetworkVirtualMachine,
 {
+    #[cfg(feature = "vm-entrypoint")]
+    async fn main<F>(handlers: F)
+    where
+        Self: 'static + Sized,
+        F: Send + FnOnce(&Self) -> Vec<::tokio::task::JoinHandle<()>>,
+    {
+        ::ark_core::tracer::init_once();
+        info!("Welcome to kubegraph!");
+
+        let signal = ::ark_core::signal::FunctionSignal::default().trap_on_panic();
+        if let Err(error) = signal.trap_on_sigint() {
+            error!("{error}");
+            return;
+        }
+
+        info!("Booting...");
+        let vm = match <Self as NetworkVirtualMachine>::try_default().await {
+            Ok(vm) => vm,
+            Err(error) => {
+                signal
+                    .panic(anyhow!("failed to init network virtual machine: {error}"))
+                    .await
+            }
+        };
+
+        info!("Registering side workers...");
+        let handlers = handlers(&vm);
+
+        info!("Ready");
+        signal.wait_to_terminate().await;
+
+        info!("Terminating...");
+        for handler in handlers {
+            handler.abort();
+        }
+
+        if let Err(error) = vm.close().await {
+            error!("{error}");
+        };
+
+        signal.exit().await
+    }
+
     #[instrument(level = Level::INFO, skip(self))]
     async fn loop_forever(&self) {
         let fallback_interval = self.fallback_interval();
@@ -200,6 +243,10 @@ where
 
     fn interval(&self) -> Option<Duration>;
 
+    async fn try_default() -> Result<Self>
+    where
+        Self: Sized;
+
     async fn infer_edges(
         &self,
         problem: &VirtualProblem,
@@ -244,6 +291,13 @@ where
         <T as NetworkVirtualMachine>::interval(&**self)
     }
 
+    #[instrument(level = Level::INFO)]
+    async fn try_default() -> Result<Self> {
+        <T as NetworkVirtualMachine>::try_default()
+            .await
+            .map(Self::new)
+    }
+
     #[instrument(level = Level::INFO, skip(self, problem, nodes))]
     async fn infer_edges(
         &self,

crates/kubegraph/gateway/Cargo.toml

@@ -39,8 +39,8 @@ full = [
 # Configure Analyzers
 analyzer-full = ["analyzer-llm"]
 analyzer-llm = [
-    "kubegraph-vm-local?/analyzer-llm",
     "kubegraph-api/analyzer-llm",
+    "kubegraph-vm-local?/analyzer-llm",
 ]
 
 # Configure Connectors
@@ -87,8 +87,8 @@ vm-full = ["vm-local"]
 vm-local = ["kubegraph-vm-local"]
 
 [dependencies]
-ark-core = { path = "../../ark/core", features = ["actix-web", "signal"] }
-kubegraph-api = { path = "../api" }
+ark-core = { path = "../../ark/core" }
+kubegraph-api = { path = "../api", features = ["vm-entrypoint"] }
 kubegraph-vm-local = { path = "../vm/local", optional = true, default-features = false }
 
 actix-web = { workspace = true }

crates/kubegraph/gateway/problems/warehouse.yaml

@@ -16,8 +16,6 @@ metadata:
   name: warehouse-sample-a
   namespace: kubegraph
 spec:
-  metadata:
-    supply: payload
   fake:
     nodes:
       count: 1
@@ -45,8 +43,6 @@ metadata:
   name: warehouse-sample-b
   namespace: kubegraph
 spec:
-  metadata:
-    supply: payload
   fake:
     nodes:
       count: 1
@@ -69,6 +65,16 @@ spec:
           valueType: I64
 ---
 apiVersion: kubegraph.ulagbulag.io/v1alpha1
+kind: NetworkAnnotation
+metadata:
+  name: warehouse
+  namespace: kubegraph
+spec:
+  filter: src != sink and src.supply > 0 and src.supply > sink.supply + 1
+  script: |
+    supply = payload;
+---
+apiVersion: kubegraph.ulagbulag.io/v1alpha1
 kind: NetworkProblem
 metadata:
   name: warehouse
@@ -85,7 +91,7 @@ metadata:
   namespace: kubegraph
 spec:
   dummy: {}
-  filter: src != sink and src.supply > 0 and src.supply > sink.supply
+  filter: src != sink and src.supply > 0 and src.supply > sink.supply + 1
   script: |
     capacity = min(50, (src.supply - sink.supply) / 2);
     unit_cost = 1;

crates/kubegraph/gateway/src/main.rs

@@ -2,46 +2,11 @@ mod actix;
 mod routes;
 mod vm;
 
-use anyhow::anyhow;
 use kubegraph_api::vm::NetworkVirtualMachineExt;
 use tokio::spawn;
-use tracing::{error, info};
 
 #[tokio::main]
 async fn main() {
-    ::ark_core::tracer::init_once();
-    info!("Welcome to kubegraph!");
-
-    let signal = ::ark_core::signal::FunctionSignal::default().trap_on_panic();
-    if let Err(error) = signal.trap_on_sigint() {
-        error!("{error}");
-        return;
-    }
-
-    info!("Booting...");
-    let vm = match self::vm::try_init().await {
-        Ok(vm) => vm,
-        Err(error) => {
-            signal
-                .panic(anyhow!("failed to init network virtual machine: {error}"))
-                .await
-        }
-    };
-
-    info!("Registering side workers...");
-    let handlers = vec![spawn(crate::actix::loop_forever(vm.clone()))];
-
-    info!("Ready");
-    signal.wait_to_terminate().await;
-
-    info!("Terminating...");
-    for handler in handlers {
-        handler.abort();
-    }
-
-    if let Err(error) = vm.close().await {
-        error!("{error}");
-    };
-
-    signal.exit().await
+    self::vm::NetworkVirtualMachine::main(|vm| vec![spawn(crate::actix::loop_forever(vm.clone()))])
+        .await
 }

crates/kubegraph/gateway/src/vm.rs

@@ -1,6 +1,2 @@
-use anyhow::Result;
-
 #[cfg(feature = "vm-local")]
-pub async fn try_init() -> Result<::kubegraph_vm_local::NetworkVirtualMachine> {
-    ::kubegraph_vm_local::NetworkVirtualMachine::try_default().await
-}
+pub type NetworkVirtualMachine = ::kubegraph_vm_local::NetworkVirtualMachine;