I need help

cli-mock/brain.proto

@@ -11,9 +11,10 @@ message NodePubkey {
 message RegisterNodeReq { 
   string node_pubkey = 1;
   string owner_pubkey = 2;
+  string ip = 3;
 }
 
-message NodeResourceReq {
+message NodeRes {
   string node_pubkey = 1;
   uint32 avail_ports = 2;
   uint32 avail_ipv4 = 3;
@@ -43,6 +44,7 @@ message NewVMReq {
 
 message UpdateVMReq {
   string uuid = 1;
+  string node_pubkey = 2;
   uint32 disk_size_gb = 3;
   uint32 vcpus = 4;
   uint32 memory_mb = 5;
@@ -79,21 +81,12 @@ message ListVMContractsReq {
   string node_pubkey = 2;
 }
 
-message NewVmRespIP {
-  uint32 nic_index = 1;
-  string address = 2;
-  string mask = 3;
-  string gateway = 4;
-}
-
 message NewVMResp {
   string uuid = 1;
   repeated uint32 exposed_ports = 2;
-  string ovmf_hash = 5;
+  string public_ipv4 = 3;
-  // This is needed to allow the CLI to build the kernel params from known data.
+  string public_ipv6 = 4;
-  // The CLI will use the kernel params to get the measurement.
+  string error = 5;
-  repeated NewVmRespIP ips = 6;
-  string error = 7;
 }
 
 message DeleteVMReq {
@@ -102,7 +95,7 @@ message DeleteVMReq {
 
 service BrainDaemonService {
   rpc RegisterNode (RegisterNodeReq) returns (Empty);
-  rpc SendNodeResources (stream NodeResourceReq) returns (Empty);
+  rpc NodeResUpdate (stream NodeRes) returns (Empty);
   rpc GetNewVMReqs (NodePubkey) returns (stream NewVMReq);
   rpc SendNewVMResp (stream NewVMResp) returns (Empty);
   rpc GetDeleteVMReq (NodePubkey) returns (stream DeleteVMReq);
@@ -124,11 +117,10 @@ message NodeFilters {
 message NodeListResp {
   string node_pubkey = 1;
   string country = 2;
-  string region = 3;
+  string city = 3;
-  string city = 4;
+  string ip = 4; // required for latency test
-  string ip = 5; // required for latency test
+  uint32 server_rating = 5;
-  uint32 server_rating = 6;
+  uint32 provider_rating = 6;
-  uint32 provider_rating = 7;
 }
 
 service BrainCliService {

cli-mock/src/main.rs

@@ -5,8 +5,8 @@ pub mod brain {
 
 use anyhow::Result;
 use brain::{
-    brain_cli_service_client::BrainCliServiceClient, DeleteVmReq, ListVmContractsReq, NewVmReq,
+    brain_cli_service_client::BrainCliServiceClient, DeleteVmReq, ListVmContractsReq,
-    NodeFilters, NodeListResp, UpdateVmReq, VmContract,
+    NewVmReq, NodeFilters, NodeListResp, VmContract, UpdateVmReq,
 };
 use lazy_static::lazy_static;
 use log::{debug, info, warn};
@@ -37,8 +37,8 @@ async fn get_node_list(mut client: BrainCliServiceClient<Channel>) -> Result<Vec
     let mut grpc_stream = client
         .list_nodes(NodeFilters {
             free_ports: 0,
-            offers_ipv4: false,
+            offers_ipv4: true,
-            offers_ipv6: false,
+            offers_ipv6: true,
             vcpus: 0,
             memory_mb: 0,
             storage_gb: 0,
@@ -84,12 +84,12 @@ async fn submit_vm_request(
     info!("Creating VM {req:?}");
     let result = client.create_vm_contract(req).await;
     match result {
-        Ok(resp) => {
+        Ok(confirmation) => {
-            let resp = resp.into_inner();
+            let confirmation = confirmation.into_inner();
-            if resp.error.is_empty() {
+            if confirmation.error.is_empty() {
-                info!("Got NewVMResp: {resp:?}");
+                info!("Got VM confirmation: {confirmation:?}");
             } else {
-                warn!("Got new VM error: {}", resp.error);
+                warn!("Got VM confirmation error: {}", confirmation.error);
             };
         }
         Err(e) => log::error!("Could not create vm: {e:?}"),
@@ -137,10 +137,12 @@ async fn delete_vm(mut client: BrainCliServiceClient<Channel>, uuid: &str) -> Re
 
 async fn update_vm_request(
     mut client: BrainCliServiceClient<Channel>,
-    uuid: &str,
+    node_pubkey: &str,
+    uuid: &str, 
 ) -> Result<()> {
     let req = UpdateVmReq {
         uuid: uuid.to_string(),
+        node_pubkey: node_pubkey.to_string(),
         vcpus: 4,
         memory_mb: 4096,
         disk_size_gb: 40,
@@ -152,12 +154,12 @@ async fn update_vm_request(
     info!("Updating VM {req:?}");
     let result = client.update_vm(req).await;
     match result {
-        Ok(resp) => {
+        Ok(confirmation) => {
-            let resp = resp.into_inner();
+            let confirmation = confirmation.into_inner();
-            if resp.error.is_empty() {
+            if confirmation.error.is_empty() {
-                info!("Got VM update response: {resp:?}");
+                info!("Got VM update confirmation: {confirmation:?}");
             } else {
-                warn!("Got VM update error: {}", resp.error);
+                warn!("Got VM update confirmation error: {}", confirmation.error);
             };
         }
         Err(e) => log::error!("Could not update vm: {e:?}"),
@@ -184,13 +186,16 @@ async fn main() -> Result<()> {
         }
     }
 
-    let contracts = list_contracts(client.clone()).await?;
+    let nodes = get_node_list(client.clone()).await?;
-    for contract in contracts {
+    for node in nodes {
-        if let Err(e) = update_vm_request(client.clone(), &contract.uuid).await {
+        let contracts = list_contracts(client.clone()).await?;
-            log::error!(
+        for contract in contracts {
-                "Received error when updating VM on node {}: {e:?}",
+            if let Err(e) = update_vm_request(client.clone(), &node.node_pubkey, &contract.uuid).await {
-                &contract.node_pubkey
+                log::error!(
-            );
+                    "Received error when updating VM on node {}: {e:?}",
+                    &node.node_pubkey
+                );
+            }
         }
     }
 

daemon-mock/brain.proto

@@ -11,9 +11,10 @@ message NodePubkey {
 message RegisterNodeReq { 
   string node_pubkey = 1;
   string owner_pubkey = 2;
+  string ip = 3;
 }
 
-message NodeResourceReq {
+message NodeRes {
   string node_pubkey = 1;
   uint32 avail_ports = 2;
   uint32 avail_ipv4 = 3;
@@ -43,6 +44,7 @@ message NewVMReq {
 
 message UpdateVMReq {
   string uuid = 1;
+  string node_pubkey = 2;
   uint32 disk_size_gb = 3;
   uint32 vcpus = 4;
   uint32 memory_mb = 5;
@@ -79,21 +81,12 @@ message ListVMContractsReq {
   string node_pubkey = 2;
 }
 
-message NewVmRespIP {
-  uint32 nic_index = 1;
-  string address = 2;
-  string mask = 3;
-  string gateway = 4;
-}
-
 message NewVMResp {
   string uuid = 1;
   repeated uint32 exposed_ports = 2;
-  string ovmf_hash = 5;
+  string public_ipv4 = 3;
-  // This is needed to allow the CLI to build the kernel params from known data.
+  string public_ipv6 = 4;
-  // The CLI will use the kernel params to get the measurement.
+  string error = 5;
-  repeated NewVmRespIP ips = 6;
-  string error = 7;
 }
 
 message DeleteVMReq {
@@ -102,7 +95,7 @@ message DeleteVMReq {
 
 service BrainDaemonService {
   rpc RegisterNode (RegisterNodeReq) returns (Empty);
-  rpc SendNodeResources (stream NodeResourceReq) returns (Empty);
+  rpc NodeResUpdate (stream NodeRes) returns (Empty);
   rpc GetNewVMReqs (NodePubkey) returns (stream NewVMReq);
   rpc SendNewVMResp (stream NewVMResp) returns (Empty);
   rpc GetDeleteVMReq (NodePubkey) returns (stream DeleteVMReq);
@@ -124,11 +117,10 @@ message NodeFilters {
 message NodeListResp {
   string node_pubkey = 1;
   string country = 2;
-  string region = 3;
+  string city = 3;
-  string city = 4;
+  string ip = 4; // required for latency test
-  string ip = 5; // required for latency test
+  uint32 server_rating = 5;
-  uint32 server_rating = 6;
+  uint32 provider_rating = 6;
-  uint32 provider_rating = 7;
 }
 
 service BrainCliService {
@@ -138,3 +130,4 @@ service BrainCliService {
   rpc DeleteVM (DeleteVMReq) returns (Empty);
   rpc UpdateVM (UpdateVMReq) returns (UpdateVMResp);
 }
+

daemon-mock/src/main.rs

@@ -5,8 +5,8 @@ pub mod brain {
 
 use anyhow::Result;
 use brain::{
-    brain_daemon_service_client::BrainDaemonServiceClient, DeleteVmReq, NewVmReq, NewVmResp,
+    brain_daemon_service_client::BrainDaemonServiceClient, DeleteVmReq, NewVmResp,
-    NewVmRespIp, NodePubkey, NodeResourceReq, RegisterNodeReq, UpdateVmReq, UpdateVmResp,
+    NewVmReq, NodePubkey, RegisterNodeReq, UpdateVmReq, UpdateVmResp, NodeRes,
 };
 use lazy_static::lazy_static;
 use log::{debug, error, info, warn};
@@ -65,32 +65,46 @@ async fn send_newvm_resp(
     Ok(())
 }
 
-async fn send_node_resources(
-    mut client: BrainDaemonServiceClient<Channel>,
-    rx: Receiver<NodeResourceReq>,
-) -> Result<()> {
-    debug!("starting send_newvm_resp stream");
-    let rx_stream = ReceiverStream::new(rx);
-    client.send_node_resources(rx_stream).await?;
-    debug!("send_newvm_resp is about to exit");
-    Ok(())
-}
-
 async fn register_node(mut client: BrainDaemonServiceClient<Channel>) {
     debug!("Starting node registration...");
     let req = RegisterNodeReq {
         node_pubkey: SECURE_PUBLIC_KEY.clone(),
         owner_pubkey: "IamTheOwnerOf".to_string() + &SECURE_PUBLIC_KEY,
+        ip: "10.0.10.1".to_string(),
     };
     match client.register_node(req).await {
         Ok(_) => info!(
-            "Registered as 10.0.10.1 from Bruma/Cyrodiil with ID {}",
+            "Registered as 10.0.10.1 {}",
             SECURE_PUBLIC_KEY.clone()
         ),
         Err(e) => error!("Could not register node data: {e:?}"),
     };
 }
 
+async fn update_node_res(
+    mut _client: BrainDaemonServiceClient<Channel>,
+    tx: Sender<NodeRes>,
+    ) -> Result<()> {
+    debug!("Starting node registration...");
+    let req = NodeRes {
+        node_pubkey: SECURE_PUBLIC_KEY.clone(),
+        avail_ports: 10000,
+        avail_ipv4: 10,
+        avail_ipv6: 100_000,
+        avail_vcpus: 16,
+        avail_memory_mb: 20_000,
+        avail_storage_gb: 700,
+        max_ports_per_vm: 5,
+    };
+
+    if let Err(e) = tx.send(req).await {
+        error!("Failed to send NodeRes: {e}");
+    } else {
+        info!("NodeRes sent successfully");
+    }
+    Ok(())
+}
+
 async fn listen_for_deleted_vms(
     mut client: BrainDaemonServiceClient<Channel>,
     tx: Sender<DeleteVmReq>,
@@ -143,17 +157,17 @@ async fn listen_for_update_vm_reqs(
 
 async fn handle_update_vm_requests(
     mut req: Receiver<UpdateVmReq>,
-    resp_chan: Sender<UpdateVmResp>,
+    resp: Sender<UpdateVmResp>
-) {
+    ) {
     info!("Started to handle update vm requests.");
     while let Some(update_vm) = req.recv().await {
-        let update_vm_resp = UpdateVmResp {
+        let confirmation = UpdateVmResp {
             uuid: update_vm.uuid,
             error: "".to_string(),
         };
-        info!("Sending UpdateVmResp: {update_vm_resp:?}");
+        info!("Sending UpdateVmResp: {confirmation:?}");
-        let _ = resp_chan.send(update_vm_resp).await;
+        let _ = resp.send(confirmation).await;
-    }
+    };
     warn!("update vm request handler is ending");
 }
 
@@ -168,11 +182,7 @@ async fn send_updatevm_resp(
     Ok(())
 }
 
-async fn handle_vm_requests(
+async fn handle_vm_requests(mut req: Receiver<NewVmReq>, resp: Sender<NewVmResp>) {
-    mut req: Receiver<NewVmReq>,
-    resp: Sender<NewVmResp>,
-    resource_tx: Sender<NodeResourceReq>,
-) {
     info!("Started to handle vm requests. 1 out of 5 requests will return error.");
     let mut i = 0;
     while let Some(new_vm) = req.recv().await {
@@ -180,53 +190,34 @@ async fn handle_vm_requests(
             true => Vec::new(),
             false => vec![20321, 20415, 25912],
         };
-        let mut ips = Vec::new();
+        let public_ipv4 = match new_vm.public_ipv4 {
-        ips.push(NewVmRespIp {
+            true => "10.0.100.5".to_string(),
-            nic_index: 0,
+            false => String::new(),
-            address: "190.0.100.5".to_string(),
+        };
-            gateway: "190.0.100.1".to_string(),
+        let public_ipv6 = match new_vm.public_ipv6 {
-            mask: "24".to_string(),
+            true => " 2a02:2f2d:d301:3100:afe8:a85e:54a0:dd28".to_string(),
-        });
+            false => String::new(),
-        ips.push(NewVmRespIp {
+        };
-            nic_index: 0,
-            address: "2a02:2f2d:d301:3100:afe8:a85e:54a0:dd28".to_string(),
-            gateway: "2a02:2f2d:d301:3100::1".to_string(),
-            mask: "64".to_string(),
-        });
         if i != 3 {
-            let confirmation = NewVmResp {
+            let confirmation = NewVmResp{
                 uuid: new_vm.uuid,
                 exposed_ports,
-                ovmf_hash: "YouAreNotGettingHacked".to_string(),
+                public_ipv4,
-                ips,
+                public_ipv6,
                 error: String::new(),
             };
             info!("Sending NewVmConfirmation: {confirmation:?}");
             let _ = resp.send(confirmation).await;
-            info!("Sending NodeResourceReq");
         } else {
             let confirmation = NewVmResp {
                 uuid: new_vm.uuid,
                 exposed_ports: Vec::new(),
-                ovmf_hash: "YouAreNotGettingHacked".to_string(),
+                public_ipv4: String::new(),
-                ips: Vec::new(),
+                public_ipv6: String::new(),
                 error: "No.".to_string(),
             };
             info!("Sending error for NewVmConfirmation: {confirmation:?}");
             let _ = resp.send(confirmation).await;
-            info!("Sending updated about used resources.");
-            let _ = resource_tx
-                .send(NodeResourceReq {
-                    node_pubkey: SECURE_PUBLIC_KEY.clone(),
-                    avail_ipv4: 5,
-                    avail_ipv6: 5,
-                    avail_memory_mb: 5000,
-                    avail_vcpus: 32,
-                    avail_storage_gb: 100,
-                    avail_ports: 5000,
-                    max_ports_per_vm: 5,
-                })
-                .await;
         }
         i += 1;
         if i == 5 {
@@ -241,31 +232,21 @@ async fn connect_and_run() -> Result<()> {
     let mut streaming_tasks = JoinSet::new();
 
     register_node(client.clone()).await;
+    
+    let update_node_res_client = client.clone();
+    let (node_tx, _node_rx) = tokio::sync::mpsc::channel(6);
+    streaming_tasks.spawn(update_node_res(update_node_res_client, node_tx));
 
     let newvm_client = client.clone();
     let (tx, newvm_rx) = tokio::sync::mpsc::channel(6);
     streaming_tasks.spawn(listen_for_new_vm_reqs(newvm_client, tx));
 
     let confirm_client = client.clone();
-    let resource_client = client.clone();
+    let (confirm_tx, rx) = tokio::sync::mpsc::channel(6);
-    let (resource_tx, resource_rx) = tokio::sync::mpsc::channel(6);
+    streaming_tasks.spawn(send_newvm_resp(confirm_client, rx));
-    let _ = resource_tx
+
-        .send(NodeResourceReq {
-            node_pubkey: SECURE_PUBLIC_KEY.clone(),
-            avail_ipv4: 10,
-            avail_ipv6: 10,
-            avail_memory_mb: 2500,
-            avail_vcpus: 60,
-            avail_storage_gb: 200,
-            avail_ports: 5002,
-            max_ports_per_vm: 5,
-        })
-        .await;
-    let (confirm_tx, confirm_rx) = tokio::sync::mpsc::channel(6);
-    streaming_tasks.spawn(send_newvm_resp(confirm_client, confirm_rx));
-    streaming_tasks.spawn(send_node_resources(resource_client, resource_rx));
     tokio::spawn(async move {
-        handle_vm_requests(newvm_rx, confirm_tx, resource_tx).await;
+        handle_vm_requests(newvm_rx, confirm_tx).await;
     });
 
     let updatevm_client = client.clone();
@@ -285,7 +266,7 @@ async fn connect_and_run() -> Result<()> {
     streaming_tasks.spawn(listen_for_deleted_vms(deletevms_client, tx));
 
     let task_output = streaming_tasks.join_next().await;
-    warn!("One stream exited: {task_output:?}");
+    warn!("One stream exited: {task_output:?} {streaming_tasks:?}");
     Ok(())
 }
 
@@ -306,4 +287,4 @@ async fn main() {
         .init();
     info!("Hello! My name is {}", SECURE_PUBLIC_KEY.clone());
     connection_wrapper().await;
-}
+}