diff --git a/server/docker-compose.yml b/server/docker-compose.yml index 5681211..6408902 100644 --- a/server/docker-compose.yml +++ b/server/docker-compose.yml @@ -13,6 +13,7 @@ services: context: kafka ports: - '9092:9092' + - '7243:7243' environment: - KAFKA_BROKER_ID=1 - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=DOCKER diff --git a/server/kafka/Dockerfile b/server/kafka/Dockerfile index 8b284fb..76c78f2 100644 --- a/server/kafka/Dockerfile +++ b/server/kafka/Dockerfile @@ -1,4 +1,4 @@ -FROM rust:latest as build-stage +FROM rust:1.54 as build-stage RUN mkdir -p /opt/pong/kafka-script-proxy WORKDIR /opt/pong/kafka-script-proxy @@ -10,5 +10,8 @@ FROM bitnami/kafka:latest ADD custom-entrypoint.sh / COPY --from=build-stage /opt/pong/kafka-script-proxy/target/release/kafka-script-proxy /bin/ +USER root +RUN mkdir -p /var/lib/kafka-script-proxy + ENTRYPOINT [] CMD [ "/custom-entrypoint.sh" ] diff --git a/server/kafka/kafka-script-proxy/.dockerignore b/server/kafka/kafka-script-proxy/.dockerignore new file mode 100644 index 0000000..f419dc7 --- /dev/null +++ b/server/kafka/kafka-script-proxy/.dockerignore @@ -0,0 +1,2 @@ +/.idea +/target diff --git a/server/kafka/kafka-script-proxy/Cargo.toml b/server/kafka/kafka-script-proxy/Cargo.toml index 3813d85..5e03ca8 100644 --- a/server/kafka/kafka-script-proxy/Cargo.toml +++ b/server/kafka/kafka-script-proxy/Cargo.toml @@ -1,7 +1,8 @@ + [package] name = "kafka-script-proxy" version = "0.1.0" -edition = "2021" +edition = "2018" [workspace] diff --git a/server/kafka/kafka-script-proxy/src/main.rs b/server/kafka/kafka-script-proxy/src/main.rs index 665d0f4..eacce13 100644 --- a/server/kafka/kafka-script-proxy/src/main.rs +++ b/server/kafka/kafka-script-proxy/src/main.rs @@ -1,7 +1,11 @@ -use std::convert::Infallible; -use hyper::{Body, Method, Request, Response, Server}; use hyper::server::conn::AddrStream; use hyper::service::{make_service_fn, service_fn}; +use hyper::{Body, Method, Request, Response, Server}; +use std::convert::Infallible; +use std::fs::write; +use std::process::Output; +use tokio::fs::OpenOptions; +use tokio::io::AsyncWriteExt; use tokio::process::Command; const TOPICS: [&str; 3] = ["move", "status", "input"]; @@ -18,34 +22,39 @@ pub async fn run() -> Result<(), ()> { })) }); - let host = ([0,0,0,0], 7243).into(); + let host = ([0, 0, 0, 0], 7243).into(); let server = Server::bind(&host).serve(make_svc); - println!("Listening on http://{}", host); + write_to_log(&format!("Listening on http://{}", host)).await; let graceful = server.with_graceful_shutdown(shutdown_signal()); graceful.await; Ok(()) } async fn handle_request(req: Request) -> Result, Infallible> { - println!("req to {} with method {}", req.uri().path(), req.method()); + write_to_log(&format!( + "req to {} with method {}", + req.uri().path(), + req.method() + )) + .await; match (req.method(), req.uri().path()) { (&Method::POST, "/add_partition") => handle_add_partition().await, - _ => Ok(Response::new("unknown".into())) + _ => Ok(Response::new("unknown".into())), } } -async fn handle_add_partition() -> Result, Infallible> { +async fn handle_add_partition() -> Result, Infallible> { let current_count = get_highest_partition_count().await; let next_partition = current_count + 1; for topic in TOPICS { - let output = Command::new("/bin/bash").arg(format!("-c /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic {} --partitions {}", topic, next_partition)).output().await; + let output = run_command(&format!("/opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic {} --partitions {}", topic, next_partition)).await; if let Err(e) = output { - println!("{}", e); + write_to_log(&format!("{}", e)).await; return Ok(Response::new(Body::empty())); } let output = output.unwrap(); if !output.status.success() { - println!("{:?}", std::str::from_utf8(&*output.stderr)); + write_to_log(&format!("{:?}", std::str::from_utf8(&*output.stderr))).await; return Ok(Response::new(Body::empty())); } } @@ -53,14 +62,43 @@ async fn handle_add_partition() -> Result, Infallible> { } async fn get_highest_partition_count() -> u32 { - let output = Command::new("/bin/bash").arg("-c /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe | grep -Po '(?<=Partition: )(\\d+)' | sort -r | head -1").output().await.unwrap(); + let output = run_command("/opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe | grep -Po '(?<=PartitionCount: )(\\d+)' | sort -r | head -1").await.unwrap(); let stdout = std::str::from_utf8(&*output.stdout).unwrap().to_owned(); stdout.parse::().unwrap() } +async fn run_command(command: &str) -> std::io::Result { + write_to_log(&format!("Running command: {}", command)).await; + let output = Command::new("/bin/bash") + .arg(format!("-c {}", command)) + .output() + .await.unwrap(); + let stdout = std::str::from_utf8(&*output.stdout).unwrap(); + let stderr = std::str::from_utf8(&*output.stderr).unwrap(); + write_to_log(&format!("Command returned stdout: {}", stdout)); + write_to_log(&format!("Command returned stdout: {}", stderr)); + Ok(output) +} + pub async fn shutdown_signal() { // Wait for the CTRL+C signal tokio::signal::ctrl_c() .await .expect("failed to install CTRL+C signal handler"); } + +async fn write_to_log(value: &str) -> std::io::Result<()> { + let content = format!("{}\n", value); + let mut options = OpenOptions::new(); + let mut file = options + .create(true) + .write(true) + .append(true) + .open("/var/lib/kafka-script-proxy/log.txt") + .await; + if let Err(e) = file { + println!("Failed to open file: {}", e); + return Ok(()); + } + file.unwrap().write_all(content.as_bytes()).await +}