diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5d7bd52..5782c82 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,4 +14,4 @@ jobs: - uses: Swatinem/rust-cache@v2 - run: cargo build --release - run: cargo test - - run: cargo clippy -- -D warnings + - run: cargo clippy --all-targets -- -D warnings diff --git a/src/discovery.rs b/src/discovery.rs index 149f0f5..0793451 100644 --- a/src/discovery.rs +++ b/src/discovery.rs @@ -97,8 +97,8 @@ mod tests { #[test] fn test_discover_tools_returns_vec() { let tools = discover_tools(); - // We can't guarantee which tools are installed, but it should not panic - assert!(tools.len() >= 0); + // We can't guarantee which tools are installed, but it should return a valid vec + let _ = tools.len(); } #[test] diff --git a/src/executor.rs b/src/executor.rs index c76fd76..dfd13b0 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -1,3 +1,5 @@ +#![allow(dead_code)] + use serde::{Deserialize, Serialize}; use std::process::Command as SysCommand; use std::time::Instant; @@ -37,6 +39,7 @@ pub fn execute_task(cmd: &TaskCommand) -> TaskResult { let result = match cmd.tool_name.as_str() { "sigops.restart" => execute_restart(cmd), "sigops.http" => execute_http_sync(cmd), + "sigops.notify_slack" => execute_notify_slack(cmd), "sigops.condition" => execute_condition(cmd), "sigops.wait" => execute_wait(cmd), _ => Err(format!("unknown tool: {}", cmd.tool_name)), @@ -120,6 +123,56 @@ fn execute_http_sync(cmd: &TaskCommand) -> Result { })) } +fn execute_notify_slack(cmd: &TaskCommand) -> Result { + let channel = cmd.input["channel"] + .as_str() + .ok_or("missing 'channel' field")?; + let message = cmd.input["message"] + .as_str() + .ok_or("missing 'message' field")?; + + let webhook_url = cmd.input["webhookUrl"] + .as_str() + .unwrap_or("https://hooks.slack.com/services/placeholder"); + + let payload = serde_json::json!({ + "channel": channel, + "text": message, + }); + + let payload_str = + serde_json::to_string(&payload).map_err(|e| format!("json encode failed: {}", e))?; + + let output = SysCommand::new("curl") + .args([ + "-s", + "-o", + "/dev/null", + "-w", + "%{http_code}", + "-X", + "POST", + "-H", + "Content-Type: application/json", + "-d", + &payload_str, + webhook_url, + ]) + .output() + .map_err(|e| format!("curl failed: {}", e))?; + + let status_code = String::from_utf8_lossy(&output.stdout) + .trim() + .parse::() + .unwrap_or(0); + + Ok(serde_json::json!({ + "ok": (200..300).contains(&status_code), + "channel": channel, + "status": status_code, + })) +} + fn execute_condition(cmd: &TaskCommand) -> Result { let expr = cmd.input["expression"] .as_str() @@ -276,4 +329,28 @@ mod tests { assert_eq!(result.status, TaskStatus::Failed); assert!(result.error.unwrap().contains("invalid service name")); } + + #[test] + fn test_notify_slack_missing_channel() { + let cmd = TaskCommand { + task_id: "t1".to_string(), + tool_name: "sigops.notify_slack".to_string(), + input: serde_json::json!({"message": "hello"}), + }; + let result = execute_task(&cmd); + assert_eq!(result.status, TaskStatus::Failed); + assert!(result.error.unwrap().contains("channel")); + } + + #[test] + fn test_notify_slack_missing_message() { + let cmd = TaskCommand { + task_id: "t1".to_string(), + tool_name: "sigops.notify_slack".to_string(), + input: serde_json::json!({"channel": "#alerts"}), + }; + let result = execute_task(&cmd); + assert_eq!(result.status, TaskStatus::Failed); + assert!(result.error.unwrap().contains("message")); + } } diff --git a/src/health.rs b/src/health.rs new file mode 100644 index 0000000..91c180f --- /dev/null +++ b/src/health.rs @@ -0,0 +1,103 @@ +#![allow(dead_code)] +//! Minimal HTTP health endpoint for the SigOps Agent. +//! +//! Runs on a configurable port (default 9100) and responds +//! to `GET /health` with a JSON status object. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use tokio::io::AsyncWriteExt; +use tokio::net::TcpListener; +use tracing::{debug, error, info}; + +/// Starts the health HTTP server. Returns when `shutdown` is set to true. +pub async fn serve_health(port: u16, shutdown: Arc) { + let addr = format!("0.0.0.0:{}", port); + let listener = match TcpListener::bind(&addr).await { + Ok(l) => { + info!(port = port, "health endpoint listening"); + l + } + Err(e) => { + error!(error = %e, port = port, "failed to bind health endpoint"); + return; + } + }; + + loop { + if shutdown.load(Ordering::Relaxed) { + debug!("health server shutting down"); + break; + } + + let accept = tokio::time::timeout( + std::time::Duration::from_secs(1), + listener.accept(), + ) + .await; + + let (mut stream, _addr) = match accept { + Ok(Ok(pair)) => pair, + Ok(Err(e)) => { + error!(error = %e, "accept error"); + continue; + } + Err(_) => continue, // timeout — check shutdown flag + }; + + let body = serde_json::json!({ + "status": "ok", + "agent": "sigops-agent", + "version": env!("CARGO_PKG_VERSION"), + }); + let body_str = serde_json::to_string(&body).unwrap_or_default(); + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body_str.len(), + body_str, + ); + + if let Err(e) = stream.write_all(response.as_bytes()).await { + debug!(error = %e, "failed to write health response"); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_health_server_starts_and_stops() { + let shutdown = Arc::new(AtomicBool::new(false)); + let shutdown_clone = shutdown.clone(); + + let handle = tokio::spawn(async move { + serve_health(0, shutdown_clone).await; // port 0 = OS-assigned + }); + + // Give it a moment to bind, then signal shutdown + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + shutdown.store(true, Ordering::Relaxed); + + // Should complete within 2 seconds (timeout loop) + let result = tokio::time::timeout( + std::time::Duration::from_secs(3), + handle, + ) + .await; + assert!(result.is_ok(), "health server should shut down"); + } + + #[test] + fn test_health_response_format() { + let body = serde_json::json!({ + "status": "ok", + "agent": "sigops-agent", + "version": env!("CARGO_PKG_VERSION"), + }); + let s = serde_json::to_string(&body).unwrap(); + assert!(s.contains("\"status\":\"ok\"")); + assert!(s.contains("sigops-agent")); + } +} diff --git a/src/main.rs b/src/main.rs index 64c6fde..7dd066d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,14 +1,18 @@ mod config; mod discovery; mod executor; +mod health; mod heartbeat; use clap::Parser; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; use std::time::Duration; -use tracing::{error, info}; +use tracing::{error, info, warn}; use config::Config; use discovery::{collect_host_info, discover_tools}; +use health::serve_health; use heartbeat::HeartbeatClient; #[tokio::main] @@ -41,24 +45,73 @@ async fn main() { "Host discovery complete" ); + // Shared shutdown flag + let shutdown = Arc::new(AtomicBool::new(false)); + + // Start health endpoint + let health_shutdown = shutdown.clone(); + let health_handle = tokio::spawn(async move { + serve_health(9100, health_shutdown).await; + }); + // Start heartbeat loop let hb_client = HeartbeatClient::new(&config.server_url, &config.api_token); let interval = Duration::from_secs(config.heartbeat_interval); - loop { - match hb_client - .send_heartbeat(&agent_id, &config.tenant_id, &host_info, &tools) - .await - { - Ok(resp) => { - info!(status = %resp.status, "heartbeat acknowledged"); + let hb_shutdown = shutdown.clone(); + let hb_handle = tokio::spawn(async move { + while !hb_shutdown.load(Ordering::Relaxed) { + match hb_client + .send_heartbeat(&agent_id, &config.tenant_id, &host_info, &tools) + .await + { + Ok(resp) => { + info!(status = %resp.status, "heartbeat acknowledged"); + } + Err(e) => { + error!(error = %e, "heartbeat failed — will retry"); + } } - Err(e) => { - error!(error = %e, "heartbeat failed — will retry"); + tokio::time::sleep(interval).await; + } + info!("heartbeat loop stopped"); + }); + + // Wait for shutdown signal (SIGTERM or SIGINT) + tokio::select! { + _ = tokio::signal::ctrl_c() => { + warn!("received SIGINT, shutting down gracefully"); + } + _ = async { + #[cfg(unix)] + { + let mut sigterm = tokio::signal::unix::signal( + tokio::signal::unix::SignalKind::terminate(), + ).expect("failed to register SIGTERM handler"); + sigterm.recv().await; } + #[cfg(not(unix))] + { + // On non-unix platforms, just wait forever (ctrl_c will fire) + std::future::pending::<()>().await; + } + } => { + warn!("received SIGTERM, shutting down gracefully"); } - tokio::time::sleep(interval).await; } + + // Signal all tasks to stop + shutdown.store(true, Ordering::Relaxed); + info!("waiting for tasks to finish..."); + + // Give tasks time to finish + let _ = tokio::time::timeout(Duration::from_secs(5), async { + let _ = hb_handle.await; + let _ = health_handle.await; + }) + .await; + + info!("SigOps Agent stopped"); } #[cfg(test)]