diff --git a/Cargo.lock b/Cargo.lock index 39055b2..431da80 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3309,12 +3309,15 @@ dependencies = [ "serde_json", "sha1", "sha2 0.10.9", + "tempfile", "thiserror 1.0.69", "tokio", + "tokio-test", "tokio-util", "tracing", "uuid", "wasmparser 0.215.0", + "wat", "xmas-elf", ] @@ -3351,6 +3354,7 @@ dependencies = [ "tower 0.4.13", "tower-http 0.5.2", "tracing", + "tracing-subscriber", "utoipa", "utoipa-swagger-ui", "uuid", @@ -3388,6 +3392,7 @@ dependencies = [ "thiserror 1.0.69", "tokio", "toml 0.8.23", + "tracing", "urlencoding", "uuid", "wasmparser 0.215.0", diff --git a/crates/openre-api/Cargo.toml b/crates/openre-api/Cargo.toml index 9714cca..8f7ff9f 100644 --- a/crates/openre-api/Cargo.toml +++ b/crates/openre-api/Cargo.toml @@ -8,6 +8,10 @@ description = "API server (HTTP/gRPC) for open-re" categories = { workspace = true } keywords = { workspace = true } +[[bin]] +name = "openre-api" +path = "src/main.rs" + [dependencies] openre-core = { path = "../openre-core" } openre-config = { path = "../openre-config" } @@ -35,6 +39,7 @@ validator = { workspace = true } utoipa = { workspace = true } utoipa-swagger-ui = { workspace = true, features = ["axum"] } tracing = { workspace = true } +tracing-subscriber = { workspace = true } futures = { workspace = true } governor = { workspace = true } async-stream = "0.3" diff --git a/crates/openre-api/src/lib.rs b/crates/openre-api/src/lib.rs index ed144db..30d9e62 100644 --- a/crates/openre-api/src/lib.rs +++ b/crates/openre-api/src/lib.rs @@ -10,6 +10,7 @@ pub mod state; pub mod validation; pub mod versioning; pub mod websocket; +pub mod workers; pub use auth::*; pub use error::*; @@ -21,3 +22,4 @@ pub use state::*; pub use validation::*; pub use versioning::*; pub use websocket::*; +pub use workers::*; diff --git a/crates/openre-api/src/main.rs b/crates/openre-api/src/main.rs new file mode 100644 index 0000000..ccb5708 --- /dev/null +++ b/crates/openre-api/src/main.rs @@ -0,0 +1,38 @@ +//! API server binary for open-re + +use openre_api::{AppState, http::start_server}; +use std::sync::Arc; +use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + // Initialize tracing + tracing_subscriber::registry() + .with(tracing_subscriber::EnvFilter::new( + std::env::var("RUST_LOG").unwrap_or_else(|_| "info".into()), + )) + .with(tracing_subscriber::fmt::layer()) + .init(); + + // Load configuration (uses Figment: config.toml, env vars, etc.) + let config = openre_config::Config::load()?; + + // Create application state + let state = Arc::new(AppState::new(config).await?); + + // Get server address from environment or use default + let host = std::env::var("HOST").unwrap_or_else(|_| "0.0.0.0".into()); + let port = std::env::var("PORT").unwrap_or_else(|_| "8080".into()); + let addr = format!("{}:{}", host, port); + + // Run database migrations + tracing::info!("Running database migrations..."); + state.global_store.run_migrations().await?; + tracing::info!("Database migrations completed"); + + // Start the server + tracing::info!("Starting server on {}", addr); + start_server(state, &addr).await?; + + Ok(()) +} diff --git a/crates/openre-api/src/state.rs b/crates/openre-api/src/state.rs index 9bc51e8..5c06012 100644 --- a/crates/openre-api/src/state.rs +++ b/crates/openre-api/src/state.rs @@ -12,7 +12,7 @@ use openre_security_ai::{ FindingProvider, ScanStorageFindingProvider, SecurityAnalyst, SecurityAnalystImpl, }; use openre_storage::{GlobalStore, ObjectStore}; -use openre_telemetry::{metrics::MetricsRegistry, TelemetryHandle}; +use openre_telemetry::{metrics::MetricsRegistry, TelemetryHandle, init_telemetry, TelemetryGuards}; use std::num::NonZeroU32; use std::sync::Arc; use std::time::Duration; @@ -22,14 +22,26 @@ use tokio::sync::RwLock; pub struct Telemetry { pub metrics: MetricsRegistry, pub _handle: TelemetryHandle, + pub _guards: TelemetryGuards, } impl Telemetry { - pub fn new(_config: &openre_config::TelemetryConfig) -> ApiResult { - Ok(Self { - metrics: MetricsRegistry::new(), - _handle: TelemetryHandle, - }) + pub async fn new(config: &openre_config::TelemetryConfig) -> ApiResult { + let full_config = openre_config::Config { + server: Default::default(), + database: Default::default(), + redis: Default::default(), + storage: Default::default(), + plugins: Default::default(), + ai: Default::default(), + queue: Default::default(), + telemetry: config.clone(), + security: Default::default(), + auth: Default::default(), + }; + let guards = init_telemetry(&full_config).await?; + + Ok(Self { metrics: MetricsRegistry::new(), _handle: TelemetryHandle, _guards: guards }) } } @@ -62,7 +74,7 @@ impl AppState { /// Create new application state pub async fn new(config: Config) -> ApiResult { // Initialize telemetry - let telemetry = Arc::new(Telemetry::new(&config.telemetry)?); + let telemetry = Arc::new(Telemetry::new(&config.telemetry).await?); // Initialize stores let global_store = Arc::new(GlobalStore::new(&config.database).await?); diff --git a/crates/openre-api/src/workers.rs b/crates/openre-api/src/workers.rs new file mode 100644 index 0000000..9868f46 --- /dev/null +++ b/crates/openre-api/src/workers.rs @@ -0,0 +1,103 @@ +//! Job handlers for open-re workers + +use crate::AppState; +use openre_core::ids::FileId; +use openre_core::traits::JobType; +use openre_queue::{BoxedJobHandler, Job, JobHandler}; +use std::sync::Arc; +use tokio::io::AsyncReadExt; +use tracing::{error, info, warn}; + +/// Analysis job handler +pub struct AnalysisJobHandler { + state: Arc, +} + +impl AnalysisJobHandler { + pub fn new(state: Arc) -> Self { + Self { state } + } +} + +#[async_trait::async_trait] +impl JobHandler for AnalysisJobHandler { + fn job_type(&self) -> JobType { + JobType::Analysis + } + + async fn handle(&self, job: Job) -> openre_core::error::OpenreResult { + info!(job_id = %job.id, "Starting analysis job"); + + // Extract payload + let file_id_str = job.payload.get("file_id").and_then(|v| v.as_str()); + let stages = job.payload.get("stages").and_then(|v| v.as_array()); + let _config = job.payload.get("config"); + + let file_id_str = match file_id_str { + Some(id) => id, + None => { + error!(job_id = %job.id, "No file_id in job payload"); + return Err(openre_core::Error::InvalidInput("Missing file_id".into())); + } + }; + + // Parse file ID + let file_id = file_id_str.parse::().map_err(|_| { + openre_core::Error::InvalidInput(format!("Invalid file_id: {}", file_id_str)) + })?; + + // Validate stages - at least one stage must be specified + let stage_count = stages.map(|s| s.len()).unwrap_or(0); + if stage_count == 0 { + warn!(job_id = %job.id, "No analysis stages specified, using default stages"); + } + + // Verify file exists in object storage (streaming check, no full load) + // We just open the stream and read a small amount to verify accessibility + let mut file_stream = self.state.object_store.get_object(file_id).await?; + let mut verify_buffer = [0u8; 1024]; + let bytes_read = tokio::io::AsyncReadExt::read(&mut file_stream, &mut verify_buffer).await?; + if bytes_read == 0 { + error!(job_id = %job.id, file_id = %file_id_str, "File is empty or inaccessible"); + return Err(openre_core::Error::InvalidInput("File is empty".into())); + } + + // Get file size from object store metadata (avoid loading entire file) + let file_size = self.state.object_store.get_size(file_id).await.unwrap_or(0); + + // TODO: Implement actual analysis pipeline + // This would include: + // 1. Binary format identification (ELF, PE, Mach-O, WASM) + // 2. Architecture detection + // 3. Function discovery and CFG construction + // 4. Data flow analysis + // 5. Type recovery + // 6. Decompilation + // 7. AI enrichment (if enabled) + // 8. Export results + + let start_time = std::time::Instant::now(); + + // Placeholder for actual analysis - in production this runs the full pipeline + // For now, we return a structured result indicating the job was received + // and what stages would be run + let result = serde_json::json!({ + "file_id": file_id_str, + "status": "completed", + "stages_requested": stage_count, + "stages_completed": 0, // Will be updated by actual pipeline + "functions_found": 0, + "analysis_duration_ms": start_time.elapsed().as_millis() as u64, + "file_size_bytes": file_size, + "note": "Analysis pipeline not yet implemented - this is a scaffold" + }); + + info!(job_id = %job.id, file_size = file_size, "Analysis job scaffold completed"); + Ok(result) + } +} + +/// Get all job handlers for the worker +pub fn get_job_handlers(state: Arc) -> Vec { + vec![Arc::new(AnalysisJobHandler::new(state))] +} diff --git a/crates/openre-cli/Cargo.toml b/crates/openre-cli/Cargo.toml index 7d0de6c..4f345da 100644 --- a/crates/openre-cli/Cargo.toml +++ b/crates/openre-cli/Cargo.toml @@ -32,6 +32,7 @@ serde_json = { workspace = true } uuid = { workspace = true } chrono = { workspace = true } reqwest = { workspace = true } +tracing = { workspace = true } indicatif = "0.17" colored = "2.0" tabled = { version = "0.16", features = ["derive"] } diff --git a/crates/openre-cli/src/commands/file.rs b/crates/openre-cli/src/commands/file.rs index ad7786d..9fe400f 100644 --- a/crates/openre-cli/src/commands/file.rs +++ b/crates/openre-cli/src/commands/file.rs @@ -138,7 +138,7 @@ impl FileCommands { if !response.status().is_success() { let error = response.text().await?; - return Err(CliError::ApiError(error)); + return Err(error.into()); } let file_response: FileResponse = response.json().await?; diff --git a/crates/openre-cli/src/commands/mod.rs b/crates/openre-cli/src/commands/mod.rs index 26dbc2c..fd07f17 100644 --- a/crates/openre-cli/src/commands/mod.rs +++ b/crates/openre-cli/src/commands/mod.rs @@ -13,3 +13,4 @@ pub mod project; pub mod report; pub mod scan; pub mod server; +pub mod worker; diff --git a/crates/openre-cli/src/commands/worker.rs b/crates/openre-cli/src/commands/worker.rs new file mode 100644 index 0000000..b052342 --- /dev/null +++ b/crates/openre-cli/src/commands/worker.rs @@ -0,0 +1,94 @@ +//! Worker commands + +use crate::{CliError, Context}; +use clap::{Parser, Subcommand}; +use colored::Colorize; +use openre_api::{AppState, get_job_handlers}; +use openre_config::Config; +use openre_queue::{WorkerPool, WorkerMetrics as QueueWorkerMetrics}; +use openre_telemetry::metrics::{MetricsRegistry, WorkerMetrics as TelemetryWorkerMetrics}; +use std::sync::Arc; +use tracing::{error, info}; + +#[derive(Subcommand)] +pub enum WorkerCommands { + /// Start the worker + Start { + /// Number of concurrent jobs + #[arg(short, long, default_value = "4")] + concurrency: usize, + + /// Queue priorities to process (comma-separated) + #[arg(short, long, default_value = "high,default,low")] + priorities: String, + + /// Enable AI capabilities + #[arg(long)] + ai_enabled: bool, + }, +} + +impl WorkerCommands { + pub async fn execute(self, mut ctx: Context) -> Result<(), CliError> { + match self { + WorkerCommands::Start { + concurrency, + priorities, + ai_enabled, + } => { + println!("{} Starting worker with concurrency: {}", "✓".green(), concurrency); + println!(" Priorities: {}", priorities); + println!(" AI enabled: {}", ai_enabled); + + // Load configuration (uses Figment: config.toml, env vars, etc.) + let config = Config::load()?; + + // Create application state (reusing API state creation, which initializes telemetry) + let state = Arc::new(AppState::new(config.clone()).await?); + + // Get job handlers + let handlers = get_job_handlers(state.clone()); + + // Create worker config from app config + let worker_config = openre_config::WorkerConfig { + min_workers: 1, + max_workers: concurrency, + max_concurrent_jobs: concurrency, + max_memory_mb: 4096, + heartbeat_interval_secs: 10, + graceful_shutdown_timeout_secs: 60, + target_queue_depth_per_worker: 10, + }; + + // Create worker metrics (using the telemetry from AppState which has Prometheus exporter running) + let worker_metrics = Arc::new(TelemetryWorkerMetrics::new(&state.telemetry.metrics)); + + // Create worker pool + let mut worker_pool = WorkerPool::new( + state.queue_manager.clone(), + worker_config, + config.queue.clone(), + worker_metrics, + ); + + // Start the worker pool + info!("Starting worker pool with {} workers", concurrency); + worker_pool.start(handlers).await.map_err(|e| CliError::CoreError(e))?; + + // Start scheduler + info!("Starting scheduler"); + state.scheduler.start().await; + + // Wait for shutdown signal + tokio::signal::ctrl_c().await?; + info!("Shutdown signal received, stopping workers..."); + + // Graceful shutdown + worker_pool.stop().await.map_err(|e| CliError::CoreError(e))?; + + println!("{} Worker stopped gracefully", "✓".green()); + } + } + Ok(()) + } +} diff --git a/crates/openre-cli/src/context.rs b/crates/openre-cli/src/context.rs index 515e67a..c1e850d 100644 --- a/crates/openre-cli/src/context.rs +++ b/crates/openre-cli/src/context.rs @@ -53,7 +53,7 @@ impl Context { if !response.status().is_success() { let error = response.text().await?; - return Err(CliError::ApiError(error)); + return Err(error.into()); } Ok(response) @@ -78,7 +78,7 @@ impl Context { if !response.status().is_success() { let error = response.text().await?; - return Err(CliError::ApiError(error)); + return Err(error.into()); } Ok(response) @@ -103,7 +103,7 @@ impl Context { if !response.status().is_success() { let error = response.text().await?; - return Err(CliError::ApiError(error)); + return Err(error.into()); } Ok(response) @@ -123,7 +123,7 @@ impl Context { if !response.status().is_success() { let error = response.text().await?; - return Err(CliError::ApiError(error)); + return Err(error.into()); } Ok(response) diff --git a/crates/openre-cli/src/error.rs b/crates/openre-cli/src/error.rs index 365ca6a..9b751bd 100644 --- a/crates/openre-cli/src/error.rs +++ b/crates/openre-cli/src/error.rs @@ -1,5 +1,7 @@ //! CLI error types +use openre_api::ApiError; +use openre_core::Error as CoreError; use thiserror::Error; /// CLI error @@ -12,7 +14,10 @@ pub enum CliError { FileNotFound(String), #[error("API error: {0}")] - ApiError(String), + ApiError(#[from] ApiError), + + #[error("API error: {0}")] + ApiErrorString(String), #[error("Not authenticated")] NotAuthenticated, @@ -38,8 +43,17 @@ pub enum CliError { #[error("TOML parse error: {0}")] TomlParseError(#[from] toml::de::Error), + #[error("Core error: {0}")] + CoreError(#[from] CoreError), + #[error("URL encoding error: {0}")] UrlEncodingError(String), } +impl From for CliError { + fn from(s: String) -> Self { + CliError::ApiErrorString(s) + } +} + pub type CliResult = Result; diff --git a/crates/openre-cli/src/main.rs b/crates/openre-cli/src/main.rs index 0e95dc5..4065a02 100644 --- a/crates/openre-cli/src/main.rs +++ b/crates/openre-cli/src/main.rs @@ -12,7 +12,7 @@ use commands::{ ai::AiCommands, analyst::AnalystCommands, auth::AuthCommands, config::ConfigCommands, file::FileCommands, finding::FindingCommands, function::FunctionCommands, plugin::PluginCommands, project::ProjectCommands, report::ReportCommands, scan::ScanCommands, - server::ServerCommands, + server::ServerCommands, worker::WorkerCommands, // analysis::AnalysisCommands, // Temporarily disabled due to compilation errors }; pub use config::CliConfig; @@ -111,6 +111,10 @@ enum Commands { /// Report generation #[command(subcommand)] Report(ReportCommands), + + /// Worker management + #[command(subcommand)] + Worker(WorkerCommands), } #[tokio::main] @@ -156,5 +160,6 @@ async fn main() -> Result<(), CliError> { Commands::Scan(cmd) => cmd.execute(ctx).await, Commands::Finding(cmd) => cmd.execute(ctx).await, Commands::Report(cmd) => cmd.execute(ctx).await, + Commands::Worker(cmd) => cmd.execute(ctx).await, } } diff --git a/crates/openre-storage/src/object.rs b/crates/openre-storage/src/object.rs index 6351281..a50591d 100644 --- a/crates/openre-storage/src/object.rs +++ b/crates/openre-storage/src/object.rs @@ -114,6 +114,14 @@ impl ObjectStore { Ok(buffer) } + /// Get file size in bytes + pub async fn get_size(&self, file_id: FileId) -> Result { + let path = self.object_path(file_id); + let file_path = self.local_base.join(&path); + let metadata = tokio::fs::metadata(&file_path).await?; + Ok(metadata.len()) + } + /// Generate object path for a file ID fn object_path(&self, file_id: FileId) -> String { let uuid_str = file_id.to_string(); diff --git a/docker-compose.yml b/docker-compose.yml index f383341..993016e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -92,6 +92,12 @@ services: - api_logs:/app/logs networks: - openre-network + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8080/health"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 40s deploy: resources: limits: @@ -128,13 +134,18 @@ services: - worker_logs:/app/logs networks: - openre-network + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:9090/metrics"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 40s deploy: resources: limits: memory: 4G reservations: memory: 1G - replicas: 2 # AI worker (with GPU support if available) worker-ai: @@ -169,6 +180,12 @@ services: - ./models:/app/models:ro networks: - openre-network + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:9090/metrics"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 40s deploy: resources: limits: @@ -196,6 +213,12 @@ services: - api networks: - openre-network + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost/health"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 10s # Prometheus for metrics prometheus: @@ -214,6 +237,12 @@ services: - "9090:9090" networks: - openre-network + healthcheck: + test: ["CMD", "wget", "--spider", "-q", "http://localhost:9090/-/healthy"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 10s # Grafana for visualization grafana: @@ -233,6 +262,12 @@ services: - prometheus networks: - openre-network + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:3000/api/health"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 10s volumes: postgres_data: diff --git a/docker/grafana-dashboards.yml b/docker/grafana-dashboards.yml new file mode 100644 index 0000000..dab28a8 --- /dev/null +++ b/docker/grafana-dashboards.yml @@ -0,0 +1,12 @@ +apiVersion: 1 + +providers: + - name: 'OpenRE Dashboards' + orgId: 1 + folder: 'OpenRE' + type: file + disableDeletion: false + updateIntervalSeconds: 10 + allowUiUpdates: true + options: + path: /var/lib/grafana/dashboards diff --git a/docker/grafana-datasources.yml b/docker/grafana-datasources.yml new file mode 100644 index 0000000..bb009bb --- /dev/null +++ b/docker/grafana-datasources.yml @@ -0,0 +1,9 @@ +apiVersion: 1 + +datasources: + - name: Prometheus + type: prometheus + access: proxy + url: http://prometheus:9090 + isDefault: true + editable: false diff --git a/docker/init-db.sql b/docker/init-db.sql new file mode 100644 index 0000000..ce8acab --- /dev/null +++ b/docker/init-db.sql @@ -0,0 +1,18 @@ +-- Initial database setup for open-re +-- This script runs when the PostgreSQL container starts for the first time + +-- Enable required extensions +CREATE EXTENSION IF NOT EXISTS "uuid-ossp"; +CREATE EXTENSION IF NOT EXISTS "pgcrypto"; +CREATE EXTENSION IF NOT EXISTS "pg_trgm"; + +-- Create the openre user if not exists (handled by POSTGRES_USER env var) +-- GRANT ALL PRIVILEGES ON DATABASE openre TO openre; + +-- Set default privileges +ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT ALL ON TABLES TO openre; +ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT ALL ON SEQUENCES TO openre; +ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT ALL ON FUNCTIONS TO openre; + +-- The actual schema migrations are run by the API server on startup +-- via the MigrationManager in openre-storage diff --git a/docker/prometheus.yml b/docker/prometheus.yml new file mode 100644 index 0000000..a9801c0 --- /dev/null +++ b/docker/prometheus.yml @@ -0,0 +1,42 @@ +global: + scrape_interval: 15s + evaluation_interval: 15s + +scrape_configs: + - job_name: 'prometheus' + static_configs: + - targets: ['localhost:9090'] + + - job_name: 'openre-api' + static_configs: + - targets: ['api:8080'] + metrics_path: '/metrics' + + - job_name: 'openre-worker' + static_configs: + - targets: ['worker:9090'] + metrics_path: '/metrics' + + - job_name: 'openre-worker-ai' + static_configs: + - targets: ['worker-ai:9090'] + metrics_path: '/metrics' + + - job_name: 'postgres' + static_configs: + - targets: ['postgres:5432'] + metrics_path: '/metrics' + + - job_name: 'redis' + static_configs: + - targets: ['redis:6379'] + metrics_path: '/metrics' + + - job_name: 'minio' + static_configs: + - targets: ['minio:9000'] + metrics_path: '/minio/v2/metrics/cluster' + + - job_name: 'node-exporter' + static_configs: + - targets: ['host.docker.internal:9100']