Client side code done for dupl, loss

ChaosManager at Actor TBD

Co-authored-by: Brian Sajeev <Brian-1402@users.noreply.github.com>

Author: Aman-Hassan

Cargo.lock

@@ -25,4 +25,9 @@ kill_node:
 
 node: kill_node
 	@echo "Running demo server..."
-	cargo run --features swagger --bin reactor_nctrl -- --port 3000 /tmp
+	cargo run --features swagger --bin reactor_nctrl -- --port 3000 /tmp
+
+generate:
+	@echo "Generating client"
+	openapi-generator-cli generate -i http://localhost:3000/api-doc/openapi.json \
+	-g rust -o rpc_client/ --additional-properties=packageName=reactor-client

Makefile

@@ -27,6 +27,8 @@ pub enum ControlInst {
     StartLocalRecv(LocalChannelRx),
     StartTcpRecv(u16),
     Stop,
+    SetMsgLoss { probability: f32 },
+    SetMsgDuplication { factor: i32, probability: f32 },
 }
 
 #[derive(Debug)]

actor/src/node_comm.rs

@@ -135,6 +135,23 @@ where
                     .map_err(|_| ActorError::R2PErr)?;
                 break;
             }
+            ControlInst::SetMsgDuplication {
+                factor,
+                probability,
+            } => {
+                // cancel_token.cancel();
+                // p_tx.send(R2PMsg::Exit)
+                //     .await
+                //     .map_err(|_| ActorError::R2PErr)?;
+                break;
+            }
+            ControlInst::SetMsgLoss { probability } => {
+                // cancel_token.cancel();
+                // p_tx.send(R2PMsg::Exit)
+                //     .await
+                //     .map_err(|_| ActorError::R2PErr)?;
+                break;
+            }
         }
     }
     tcp_server_set.abort_all();

actor/src/recv.rs

@@ -211,7 +211,8 @@ async fn sender_task<M, E>(
             Connection::CouldntResolve => {
                 log::warn!("[ACTOR] Failed to resolve {}", send_addr);
                 tokio::time::sleep(Duration::from_millis(100)).await;
-                continue;
+                // continue;
+                break;
             }
         };
     }

actor/src/send.rs

@@ -109,7 +109,9 @@ port = 3000
                         actor_name: "pinger".into(),
                         payload: HashMap::from([("other".to_string(), json!("ponger"))]),
                         replicas: None,
-                        chaos: Some(ChaosOp::Crash { start_ms: 10000 }),
+                        chaos: Some(vec![ChaosOp::Crash {
+                            start_ms: Some(10000),
+                        }]),
                     }],
                 ),
                 (

generic_jctrl/src/main.rs

@@ -20,6 +20,7 @@ serde_json = "1.0"
 cargo_toml = "0.22"
 thiserror = "2.0"
 tempfile = "3.20.0"
+ordered-float = {version = "5", features = ["serde"]}
 reactor-client = {path="../rpc_client"}
 
 

job_manager/Cargo.toml

@@ -7,7 +7,7 @@ use tokio::time::sleep;
 use placement::{ChaosOp, Hostname, LibInfo, LogicalOp, PhysicalOp, PlacementManager};
 use reactor_client::{
     self,
-    models::{RemoteActorInfo, SpawnArgs},
+    models::{ChaosConfig, ChaosType, RemoteActorInfo, SpawnArgs},
 };
 
 pub mod placement;
@@ -17,7 +17,7 @@ struct NodeHandle {
     client_config: reactor_client::apis::configuration::Configuration,
     actors: Vec<RemoteActorInfo>,
     loaded_libs: Vec<String>,
-    crash_schedule: Vec<(RemoteActorInfo, Instant)>,
+    chaos_schedule: Vec<(RemoteActorInfo, ChaosOp, Instant, Option<Instant>)>,
 }
 
 impl NodeHandle {
@@ -52,6 +52,7 @@ impl NodeHandle {
     async fn place(&mut self, logical_op: &LogicalOp, physical_op: &PhysicalOp) -> RemoteActorInfo {
         let mut remote_actor_info = reactor_client::apis::default_api::start_actor(
             &self.client_config,
+            //? Might need to add choas info here (instead of just payload) -> useful for msg_loss, msg_duplicate chaos ops
             SpawnArgs {
                 actor_name: physical_op.actor_name.clone(),
                 operator_name: logical_op.name.clone(),
@@ -63,11 +64,18 @@ impl NodeHandle {
         .unwrap();
         remote_actor_info.hostname = self.hostname.to_string();
         self.actors.push(remote_actor_info.clone());
-        if let Some(ChaosOp::Crash { start_ms }) = &physical_op.chaos {
+        for chaos_op in &physical_op.chaos.clone().unwrap_or_default() {
             let now = Instant::now();
-            let delay = Duration::from_millis(*start_ms as u64);
-            self.crash_schedule
-                .push((remote_actor_info.clone(), now + delay));
+            let start_time = now + Duration::from_millis(chaos_op.start_ms() as u64);
+            let stop_time = chaos_op
+                .stop_ms()
+                .map(|stop_ms| now + Duration::from_millis(stop_ms as u64));
+            self.chaos_schedule.push((
+                remote_actor_info.clone(),
+                chaos_op.clone(),
+                start_time,
+                stop_time,
+            ));
         }
         remote_actor_info
     }
@@ -78,22 +86,59 @@ impl NodeHandle {
             .unwrap();
     }
 
-    async fn schedule_actor_crash(&self) {
+    async fn schedule_actor_chaos(&self) {
         // also needs some logic on, what if ctrlc pressed prematurely
         // and this function keeps running and sends requests to non-existent nodes
-        for (actor, when) in self.crash_schedule.clone() {
-            // let this = self.clone();
+        for (actor, op, start, stop_opt) in self.chaos_schedule.clone() {
             let client_config = self.client_config.clone();
+
+            let chaos_config = match op {
+                ChaosOp::Crash { .. } => ChaosConfig {
+                    kind: ChaosType::Crash,
+                    actor_name: actor.name.clone(),
+                    factor: None,
+                    probability: None,
+                },
+
+                ChaosOp::MsgLoss { probability, .. } => ChaosConfig {
+                    kind: ChaosType::MsgLoss,
+                    actor_name: actor.name.clone(),
+                    factor: None,
+                    probability: Some(probability.into_inner()),
+                },
+
+                ChaosOp::MsgDuplication {
+                    factor,
+                    probability,
+                    ..
+                } => ChaosConfig {
+                    kind: ChaosType::MsgDuplication,
+                    actor_name: actor.name.clone(),
+                    factor: Some(factor),
+                    probability: Some(probability.into_inner()),
+                },
+            };
             tokio::spawn(async move {
                 let now = Instant::now();
-                if when > now {
-                    sleep(when - now).await;
+                if start > now {
+                    sleep(start - now).await;
                 }
-                // this.stop_actor(&actor).await;
-                reactor_client::apis::default_api::stop_actor(&client_config, actor.clone())
+                reactor_client::apis::default_api::add_chaos(&client_config, chaos_config)
                     .await
                     .unwrap();
             });
+            //? This is for later
+            // if let Some(stop) = stop_opt {
+            //     tokio::spawn(async move {
+            //         let now = Instant::now();
+            //         if stop > now {
+            //             sleep(stop - now).await;
+            //         }
+            //         // reactor_client::apis::default_api::stop_actor(&client_config, actor.clone())
+            //         //     .await
+            //         //     .unwrap();
+            //     });
+            // }
         }
     }
 
@@ -124,7 +169,7 @@ impl<PM: PlacementManager> JobController<PM> {
                 client_config: self.client_config(hostname, port),
                 actors: Vec::new(),
                 loaded_libs: Vec::new(),
-                crash_schedule: Vec::new(),
+                chaos_schedule: Vec::new(),
             },
         );
     }
@@ -170,7 +215,7 @@ impl<PM: PlacementManager> JobController<PM> {
 
     pub async fn chaos_scheduler(&self) {
         for (_, node_handle) in self.nodes.iter() {
-            node_handle.schedule_actor_crash().await;
+            node_handle.schedule_actor_chaos().await;
         }
     }
 

job_manager/src/lib.rs

@@ -1,3 +1,4 @@
+use ordered_float::OrderedFloat;
 use serde::Deserialize;
 use std::{
     collections::{BTreeMap, HashMap},
@@ -44,24 +45,47 @@ pub struct LogicalOp {
 #[serde(tag = "type", rename_all = "SCREAMING_SNAKE_CASE")]
 pub enum ChaosOp {
     Crash {
-        start_ms: u32,
+        start_ms: Option<u32>,
     },
     MsgLoss {
-        start_ms: u32,
-        probability: i32, // should change to f32, but issue with derive(Eq)
+        start_ms: Option<u32>,
+        stop_ms: Option<u32>,
+        probability: OrderedFloat<f32>, // should change to f32, but issue with derive(Eq)
     },
+    //? Add probabiltity field for this?
+    //? duplication factor instead of rate? (confusing terms)
     MsgDuplication {
-        start_ms: u32,
-        rate: i32, // should change to f32, but issue with derive(Eq)
+        start_ms: Option<u32>,
+        stop_ms: Option<u32>,
+        factor: i32,
+        probability: OrderedFloat<f32>,
     },
 }
 
+impl ChaosOp {
+    pub fn start_ms(&self) -> u32 {
+        match self {
+            ChaosOp::Crash { start_ms, .. }
+            | ChaosOp::MsgLoss { start_ms, .. }
+            | ChaosOp::MsgDuplication { start_ms, .. } => start_ms.unwrap_or(0),
+        }
+    }
+
+    pub fn stop_ms(&self) -> Option<u32> {
+        match self {
+            ChaosOp::Crash { .. } => None,
+            ChaosOp::MsgLoss { stop_ms, .. } | ChaosOp::MsgDuplication { stop_ms, .. } => *stop_ms,
+        }
+    }
+}
+
 #[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
 pub struct PhysicalOp {
     pub nodename: String,
     pub actor_name: String,
     pub replicas: Option<u32>,
-    pub chaos: Option<ChaosOp>,
+    //? Change to list (/map) of chaos ops for multiple choas ops per actor
+    pub chaos: Option<Vec<ChaosOp>>,
     #[serde(flatten)]
     pub payload: HashMap<String, serde_json::Value>,
 }

job_manager/src/placement.rs

@@ -18,6 +18,7 @@ tempfile = "3.20.0"
 reactor-actor = { path = "../actor"}
 # env_logger = "0.11"
 thiserror.workspace = true
+ordered-float = {version = "5", features = ["serde"]}
 
 tracing = "0.1.41"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }