diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..d6ac8dc --- /dev/null +++ b/Dockerfile @@ -0,0 +1,37 @@ +# Stage 1: Build the application +FROM rust:1-slim as builder + +# Install cargo-chef for dependency caching +RUN apt-get update && apt-get install -y protobuf-compiler libssl-dev pkg-config +RUN cargo install cargo-chef + +# Set the working directory +WORKDIR /app + +# Copy the dependency manifests +COPY Cargo.toml Cargo.lock ./ +COPY control.toml . +COPY . . + +# Build the release binary +RUN cargo build --release --bin server-db + +# Stage 2: Create the runtime image +FROM gcr.io/distroless/cc-debian12 + +WORKDIR /app + +# Copy the binary from the builder stage +COPY --from=builder /app/target/release/server-db /usr/local/bin/server-db + +# Copy the control file from the builder stage +COPY --from=builder /app/control.toml /app/control.toml + +# Set the environment variable for the control file path +ENV CONTROL_FILE_PATH="/app/control.toml" + +# Create and set up a volume for data persistence +VOLUME /app/data + +# Set the command to run the server +CMD ["/usr/local/bin/server-db"] diff --git a/control.toml b/control.toml new file mode 100644 index 0000000..afa16ba --- /dev/null +++ b/control.toml @@ -0,0 +1,13 @@ +host = "localhost" +port = 8766 +last_wal_timeline = 0 +last_checkpoint_id = 0 +checkpoint_directory_path = "/app/data/checkpoints" +wal_directory_path = "/app/data/wal" +current_leader_value = 12 +self_identifier = 12 +send_addr = "0.0.0.0:8080" +consume_addr = "0.0.0.0:8081" +checkpoint_timer_interval = 1 +paxos_timer_interval = 1 +gossip_timeout = 200 diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..19e0f39 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,35 @@ +version: '3.8' +services: + lokikv-1: + build: . + container_name: lokikv-1 + ports: + - "7878:7878" + networks: + - lokikv-net + volumes: + - ./control.toml:/app/control.toml + - lokikv-data-1:/app/data + environment: + - CONTROL_FILE_PATH=/app/control.toml + + lokikv-2: + build: . + container_name: lokikv-2 + ports: + - "7879:7878" + networks: + - lokikv-net + volumes: + - ./control.toml:/app/control.toml + - lokikv-data-2:/app/data + environment: + - CONTROL_FILE_PATH=/app/control.toml + +networks: + lokikv-net: + driver: bridge + +volumes: + lokikv-data-1: + lokikv-data-2: diff --git a/src/db/loki_kv/control.rs b/src/db/loki_kv/control.rs index efecec4..6bce68e 100644 --- a/src/db/loki_kv/control.rs +++ b/src/db/loki_kv/control.rs @@ -22,14 +22,14 @@ pub struct ControlFile { consume_addr: String, checkpoint_timer_interval: Option, paxos_timer_interval: Option, - gossip_timeout: Option + gossip_timeout: Option, } impl ControlFile { - pub fn get_hostname(&self) -> String{ + pub fn get_hostname(&self) -> String { return self.host.clone(); } - pub fn get_port(&self) -> u16{ + pub fn get_port(&self) -> u16 { return self.port; } pub fn get_next_checkpoint_id(&self) -> u64 { @@ -39,11 +39,11 @@ impl ControlFile { self.last_wal_timeline + 1 } - pub fn get_send_addr(&self) -> &str{ + pub fn get_send_addr(&self) -> &str { &self.send_addr } - pub fn get_consume_addr(&self) -> &str{ + pub fn get_consume_addr(&self) -> &str { &self.consume_addr } @@ -64,23 +64,23 @@ impl ControlFile { } pub fn get_checkpoint_timer_interval(&self) -> u64 { - match self.checkpoint_timer_interval{ + match self.checkpoint_timer_interval { Some(val) => return val, - None => return 1 + None => return 1, } } pub fn get_paxos_timer_interval(&self) -> u64 { - match self.paxos_timer_interval{ + match self.paxos_timer_interval { Some(val) => return val, - None => return 5 + None => return 5, } } - pub fn get_gossip_timeout(&self) -> u64{ - match self.gossip_timeout{ + pub fn get_gossip_timeout(&self) -> u64 { + match self.gossip_timeout { Some(val) => return val, - None => return 300 + None => return 300, } } @@ -112,7 +112,7 @@ impl ControlFile { consume_addr: Option, checkpoint_timer_interval: Option, paxos_timer_interval: Option, - gossip_timeout: Option + gossip_timeout: Option, ) -> Result { // Create the WAL and checkpoint directories let wal_dir = Path::new(&wal_directory_path); @@ -129,15 +129,19 @@ impl ControlFile { let final_send_addr: String = match send_addr { Some(addr) => addr, None => { - info_string("no listening address provided.. defaulting to 0.0.0.0:8080".to_string()); + info_string( + "no listening address provided.. defaulting to 0.0.0.0:8080".to_string(), + ); "0.0.0.0:8080".to_string() } }; - let final_consume_addr: String = match consume_addr{ + let final_consume_addr: String = match consume_addr { Some(addr) => addr, None => { - info_string("no listening address provided.. defaulting to 0.0.0.0:8081".to_string()); + info_string( + "no listening address provided.. defaulting to 0.0.0.0:8081".to_string(), + ); "0.0.0.0:8081".to_string() } }; @@ -155,7 +159,7 @@ impl ControlFile { consume_addr: final_consume_addr, checkpoint_timer_interval, paxos_timer_interval, - gossip_timeout + gossip_timeout, }; // Take lock on control file diff --git a/src/db/loki_kv/wal.rs b/src/db/loki_kv/wal.rs index 4714708..0cb0b53 100644 --- a/src/db/loki_kv/wal.rs +++ b/src/db/loki_kv/wal.rs @@ -65,7 +65,7 @@ impl WALManager { Some("0.0.0.0:8081".to_string()), None, None, - None + None, ) .unwrap(); let timeline = control_file.get_next_timeline_id(); diff --git a/src/db/server_multithread/paxos.rs b/src/db/server_multithread/paxos.rs index de870a8..42661bd 100644 --- a/src/db/server_multithread/paxos.rs +++ b/src/db/server_multithread/paxos.rs @@ -1,10 +1,17 @@ -use std::{collections::{HashMap, HashSet}, io::Split, net::SocketAddr}; use local_ip_address::local_ip; +use std::{ + collections::{HashMap, HashSet}, + io::Split, + net::{Ipv4Addr, SocketAddr}, +}; -use tokio::{net::UdpSocket, time::timeout}; use tokio::time::Duration; +use tokio::{net::UdpSocket, time::timeout}; -use crate::{loki_kv::{control::ControlFile, loki_kv::get_control_file_path}, utils::{info_string, warning, warning_string}}; +use crate::{ + loki_kv::{control::ControlFile, loki_kv::get_control_file_path}, + utils::{info_string, warning, warning_string}, +}; // ---------------------------- SERVICE MANAGER ------------------------------------------- @@ -12,7 +19,7 @@ pub struct ServiceManager { udp_socket_send: UdpSocket, udp_socket_recv: UdpSocket, node_directory: HashSet<(String, String)>, // Hashset of node_id, address(ip + port) - BROADCAST_ADDRESS: SocketAddr, + multicast_addr: SocketAddr, } impl ServiceManager { @@ -26,34 +33,38 @@ impl ServiceManager { let mut node_directory: HashSet<(String, String)> = HashSet::new(); - std_socket.set_broadcast(true); + let recv_socket = UdpSocket::from_std(std_consumer_socket).unwrap(); + _ = recv_socket.join_multicast_v4(Ipv4Addr::new(239, 1, 1, 1), Ipv4Addr::UNSPECIFIED); ServiceManager { udp_socket_send: UdpSocket::from_std(std_socket).unwrap(), - udp_socket_recv: UdpSocket::from_std(std_consumer_socket).unwrap(), + udp_socket_recv: recv_socket, node_directory: node_directory, - BROADCAST_ADDRESS: "255.255.255.255:8080".parse().unwrap(), + multicast_addr: format!("239.1.1.1:{}", consume_addr.port()) + .parse() + .unwrap(), } } pub async fn broadcast_message(&self, msg: &str) -> Result<(), String> { - self.udp_socket_send.send_to(msg.as_bytes(), self.BROADCAST_ADDRESS).await.unwrap(); + self.udp_socket_send + .send_to(msg.as_bytes(), self.multicast_addr) + .await + .unwrap(); Ok(()) } pub async fn start_consumption(&self) -> Result<(), ()> { - loop{ + loop { // TODO: Add consumption logic // Somehitng like a go-routine treatment here? let mut msg_bytes: Vec = vec![]; - self.udp_socket_recv.recv_from(&mut msg_bytes); + _ = self.udp_socket_recv.recv_from(&mut msg_bytes); - tokio::spawn( - async move { - // Log message - info_string(format!("Recieved the following message: {:?}", msg_bytes)); - } - ); + tokio::spawn(async move { + // Log message + info_string(format!("Recieved the following message: {:?}", msg_bytes)); + }); break; } @@ -67,10 +78,14 @@ impl ServiceManager { // ---------------- PAXOS NODE ---------------------------- -fn get_ip_addr(addr: String) -> String{ +fn get_ip_addr(addr: String) -> String { let my_local_ip = local_ip().unwrap(); let mut tks = addr.split(":"); - let ip = format!("{}:{}", my_local_ip.to_string(), tks.nth(1).unwrap().to_string()); + let ip = format!( + "{}:{}", + my_local_ip.to_string(), + tks.nth(1).unwrap().to_string() + ); return ip; } @@ -80,20 +95,23 @@ pub struct PaxosNode { } impl PaxosNode { - pub fn new_node() -> Self{ + pub fn new_node() -> Self { let control_file = ControlFile::read_from_file_path(get_control_file_path()).unwrap(); - PaxosNode { ctrl_file: control_file, service_manager: ServiceManager::new()} + PaxosNode { + ctrl_file: control_file, + service_manager: ServiceManager::new(), + } } pub async fn propose(&self) { let value = self.ctrl_file.get_self_identifier(); // Broadcast to network - match value{ - Some(val) => { - let msg = format!("PROPOSE {}", val); - self.service_manager.broadcast_message(msg.as_str()); - }, - None => panic!("No value to broadcast!") + match value { + Some(val) => { + let msg = format!("PROPOSE {}", val); + self.service_manager.broadcast_message(msg.as_str()); + } + None => panic!("No value to broadcast!"), } } @@ -103,7 +121,11 @@ impl PaxosNode { // Gossip for node discovery pub async fn gossip(&self) -> Result<(), String> { let node_id = self.ctrl_file.get_self_identifier().unwrap(); - let data = format!("{}~{}", node_id.to_string(), get_ip_addr(self.ctrl_file.get_consume_addr().to_string())); + let data = format!( + "{}~{}", + node_id.to_string(), + get_ip_addr(self.ctrl_file.get_consume_addr().to_string()) + ); info_string(format!("Sending -> {}", data)); let result = self.service_manager.broadcast_message(data.as_str()).await; return result; @@ -111,17 +133,24 @@ impl PaxosNode { pub async fn gossip_consume(&mut self) { let MAX_GOSSIP_CONSUMPTION = 10; - for i in 0..MAX_GOSSIP_CONSUMPTION{ + for i in 0..MAX_GOSSIP_CONSUMPTION { info_string(format!("{} gossip trial", i)); // Somehitng like a go-routine treatment here? - let mut msg_bytes: Vec = vec![]; - match timeout(Duration::from_secs(self.ctrl_file.get_gossip_timeout()), self.service_manager.udp_socket_recv.recv_from(&mut msg_bytes)).await{ + let mut msg_bytes: Vec = vec![0u8; 2048]; + match timeout( + Duration::from_secs(self.ctrl_file.get_gossip_timeout()), + self.service_manager + .udp_socket_recv + .recv_from(&mut msg_bytes), + ) + .await + { Ok(Ok((_, _))) => { let data = String::from_utf8(msg_bytes).unwrap(); let mut tokens; - if data.contains("~"){ + if data.contains("~") { tokens = data.as_str().split("~"); - }else{ + } else { let msg = format!("Token does not contain any ~ {:?}, skipping..", data); warning(msg.as_str()); continue; @@ -129,12 +158,14 @@ impl PaxosNode { // Log message info_string(format!("Recieved the following message: {:?}", tokens)); - self.service_manager.update_node_directory(tokens.nth(0).unwrap().to_string(), tokens.nth(1).unwrap().to_string()); - }, + self.service_manager.update_node_directory( + tokens.nth(0).unwrap().to_string(), + tokens.nth(1).unwrap().to_string(), + ); + } Ok(Err(e)) => panic!("{}", e), - Err(e) => panic!("{}", e) + Err(e) => panic!("{}", e), }; - } } } diff --git a/src/db/server_multithread/server.rs b/src/db/server_multithread/server.rs index fa6a6e8..e7cc6af 100644 --- a/src/db/server_multithread/server.rs +++ b/src/db/server_multithread/server.rs @@ -1,16 +1,16 @@ use crate::loki_kv::control::ControlFile; -use crate::loki_kv::loki_kv::{LokiKV, ValueObject, get_control_file_path}; +use crate::loki_kv::loki_kv::{get_control_file_path, LokiKV, ValueObject}; use crate::parser::executor::Executor; use crate::parser::parser::parse_lokiql; use crate::server_multithread::paxos::PaxosNode; use crate::utils::{error_string, info, info_string, warning}; +use rand; use std::env; use std::time::{Duration, Instant}; use std::{ ops::{Deref, DerefMut}, sync::{Arc, RwLock}, }; -use rand; use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::select; use tokio::time::{interval, sleep}; @@ -29,7 +29,7 @@ pub struct LokiServer { control_file: ControlFile, } -fn decide_random_bool() -> bool{ +fn decide_random_bool() -> bool { let x: u8 = rand::random(); x % 2 == 0 } @@ -60,10 +60,10 @@ async fn handle_connection( let mut resp_str = String::new(); let asts = parse_lokiql(&request_line); - if asts.len() == 0{ + if asts.len() == 0 { // Query was wrong.. lets tell it to the user resp_str += "Invalid command.. Pls try again\n"; - }else{ + } else { let mut ast_exector = Executor::new(db_instance.clone(), asts); let responses = ast_exector.execute(); @@ -77,31 +77,41 @@ async fn handle_connection( resp_str += "\n"; - wr.write_all(resp_str.as_bytes()).await - .map_err(|e| format!("Failed to write response: {}", e))?; + wr.write_all(resp_str.as_bytes()) + .await + .map_err(|e| format!("Failed to write response: {}", e))?; - wr.flush().await - .map_err(|e| format!("Failed to flush writer: {}", e))?; + wr.flush() + .await + .map_err(|e| format!("Failed to flush writer: {}", e))?; info_string(format!("Sent response: {} bytes", resp_str)); - } } impl LokiServer { pub async fn new(thread_count: usize) -> Self { - let control_file = ControlFile::read_from_file_path(get_control_file_path()).unwrap(); - let host: String = control_file.get_hostname(); - let port: u16 = control_file.get_port(); - let addr = format!("{}:{}", control_file.get_hostname(), control_file.get_port()); + let control_file = match ControlFile::read_from_file_path(get_control_file_path()) { + Ok(cf) => cf, + Err(e) => { + error_string(format!("Failed to read control file: {}", e)); + // Exit gracefully if the control file is essential for operation + std::process::exit(1); + } + }; + + let addr = format!( + "{}:{}", + control_file.get_hostname(), + control_file.get_port() + ); info_string(format!("Trying to start server at -> {}", addr)); let tcp_listener = TcpListener::bind(addr).await; match tcp_listener { Ok(tcp_list) => { - info_string(format!("Started Sevrer at {}:{}", host, port)); - let db_instance = LokiKV::new(); - LokiServer { + let info_string = format!("new connection from {} on {}", addr, thread::current().name().unwrap_or("unnamed")); + info!("{}", info_string); tcp_listener: tcp_list, host, port, @@ -120,11 +130,12 @@ impl LokiServer { let checkpoint_itr: u64 = self.control_file.get_checkpoint_timer_interval(); let paxos_itr: u64 = self.control_file.get_paxos_timer_interval(); - let mut checkpoint_timer = interval(Duration::from_secs(checkpoint_itr*60)); - let mut paxos_gossip_broadcast_timer = interval(Duration::from_secs(paxos_itr*30)); + let mut checkpoint_timer = interval(Duration::from_secs(checkpoint_itr * 60)); + let mut paxos_gossip_broadcast_timer = interval(Duration::from_secs(paxos_itr * 30)); // let mut paxos_gossip_consumer_timer = interval(Duration::from_secs(paxos_itr*60)); - let paxos_node: Arc> = Arc::new(tokio::sync::RwLock::new(PaxosNode::new_node())); + let paxos_node: Arc> = + Arc::new(tokio::sync::RwLock::new(PaxosNode::new_node())); let mut should_broadcast = decide_random_bool();