feat: implement metrics and monitoring

Author: stylebending

Cargo.lock

@@ -1,9 +1,9 @@
 [workspace]
 resolver = "2"
-members = ["actions", "bus", "engine", "engine_core", "rules"]
+members = ["actions", "bus", "engine", "engine_core", "metrics", "rules"]
 
 [workspace.package]
-version = "0.1.3"
+version = "0.1.4"
 edition = "2024"
 authors = ["Your Name"]
 license = "MIT"

Cargo.toml

@@ -226,6 +226,53 @@ sc stop WinEventEngine
 engine.exe --uninstall
 ```
 
+## Metrics and Monitoring
+
+The engine includes built-in metrics collection with a local HTTP endpoint for monitoring.
+
+### Available Metrics
+
+The following metrics are collected automatically:
+
+- **events_total** - Total events processed by source and type
+- **events_dropped_total** - Events dropped due to full buffer
+- **events_processing_duration_seconds** - Event processing latency
+- **rules_evaluated_total** - Rule evaluations by rule name
+- **rules_matched_total** - Successful rule matches
+- **rules_match_duration_seconds** - Rule matching latency
+- **actions_executed_total** - Action executions by name and status
+- **actions_execution_duration_seconds** - Action execution latency
+- **plugins_events_generated_total** - Events generated per plugin
+- **plugins_errors_total** - Plugin errors
+- **config_reload_total** - Configuration reloads
+- **engine_uptime_seconds** - Engine uptime
+
+### Retention
+
+Metrics are retained with a sliding window:
+- **Regular metrics**: 1 hour
+- **Error metrics**: 24 hours
+
+### Accessing Metrics
+
+The metrics endpoint runs on `127.0.0.1:9090` (localhost only):
+
+```bash
+# Prometheus format
+curl http://127.0.0.1:9090/metrics
+
+# JSON snapshot
+curl http://127.0.0.1:9090/api/snapshot
+
+# Health check
+curl http://127.0.0.1:9090/health
+
+# Web UI
+curl http://127.0.0.1:9090/
+```
+
+**Note**: The metrics endpoint is only accessible from localhost for security.
+
 ## Architecture
 
 ```
@@ -255,6 +302,9 @@ engine.exe --uninstall
 │  ├── PowerShell Script                                  │
 │  ├── Log Message                                        │
 │  └── HTTP Request (extensible)                          │
+│                                                         │
+│  Metrics Collector (1h sliding window)                  │
+│  └── HTTP Endpoint @ 127.0.0.1:9090                     │
 └─────────────────────────────────────────────────────────┘
 ```
 
@@ -291,6 +341,11 @@ win_event_engine/
 │   ├── src/
 │   │   └── lib.rs
 │   └── Cargo.toml
+├── metrics/            # Metrics collection and HTTP endpoint
+│   ├── src/
+│   │   ├── lib.rs      # Metrics collector
+│   │   └── server.rs   # HTTP server
+│   └── Cargo.toml
 ├── config.toml.example # Example configuration
 ├── rules.toml.example  # Example rules
 ├── AGENTS.md          # Developer guidelines
@@ -445,8 +500,8 @@ This project is licensed under the MIT License - see the [LICENSE](LICENSE) file
 - [x] Rule engine
 - [x] Configuration hot-reloading
 - [x] Windows service wrapper
+- [x] Metrics and monitoring
 - [ ] Web dashboard
-- [ ] Metrics and monitoring
 - [ ] Plugin system for custom actions
 
 ## Support

README.md

@@ -27,6 +27,7 @@ engine_core = { path = "../engine_core" }
 bus = { path = "../bus" }
 rules = { path = "../rules" }
 actions = { path = "../actions" }
+metrics = { path = "../metrics" }
 
 [dev-dependencies]
 tempfile = "3"

engine/Cargo.toml

@@ -7,11 +7,16 @@ use actions::{Action, ActionExecutor, ExecuteAction, LogAction, LogLevel, PowerS
 use bus::create_event_bus;
 use engine_core::event::EventKind;
 use engine_core::plugin::EventSourcePlugin;
+use metrics::{
+    record_action_execution, record_config_reload, record_event,
+    record_event_processing_duration, record_rule_match, record_rule_match_duration,
+    record_rule_evaluation, MetricsCollector,
+};
 use rules::{EventKindMatcher, FilePatternMatcher, Rule, RuleMatcher, WindowMatcher, WindowEventType};
 use std::path::PathBuf;
 use std::sync::Arc;
 use tokio::sync::mpsc;
-use tokio::time::{Duration, timeout};
+use tokio::time::{Duration, timeout, Instant};
 use tracing::{error, info, warn};
 
 pub struct Engine {
@@ -23,10 +28,13 @@ pub struct Engine {
     event_sender: Option<mpsc::Sender<engine_core::event::Event>>,
     shutdown_flag: Arc<std::sync::atomic::AtomicBool>,
     config_reload_rx: Option<mpsc::Receiver<()>>,
+    metrics: Arc<MetricsCollector>,
 }
 
 impl Engine {
     pub fn new(config: Config, config_path: Option<PathBuf>) -> Self {
+        let metrics = Arc::new(MetricsCollector::new());
+        
         Self {
             config,
             config_path,
@@ -36,8 +44,14 @@ impl Engine {
             event_sender: None,
             shutdown_flag: Arc::new(std::sync::atomic::AtomicBool::new(false)),
             config_reload_rx: None,
+            metrics,
         }
     }
+    
+    /// Get a reference to the metrics collector
+    pub fn metrics(&self) -> Arc<MetricsCollector> {
+        self.metrics.clone()
+    }
 
     pub fn take_config_reload_rx(&mut self) -> Option<mpsc::Receiver<()>> {
         self.config_reload_rx.take()
@@ -62,28 +76,66 @@ impl Engine {
         // Start event processing loop
         let rules = self.rules.clone();
         let action_executor = self.action_executor.clone();
+        let metrics = self.metrics.clone();
 
         tokio::spawn(async move {
             info!("Event processing loop started");
 
             while let Some(event) = receiver.recv().await {
+                let start_time = Instant::now();
+                let event_source = event.source.clone();
+                let event_type = format!("{:?}", event.kind);
+                
+                // Record event received
+                record_event(&metrics, &event_source, &event_type);
+                
                 tracing::debug!("Processing event: {:?} from {}", event.kind, event.source);
 
                 for (idx, rule) in rules.iter().enumerate() {
                     if !rule.enabled {
                         continue;
                     }
 
-                    if rule.matches(&event) {
+                    // Record rule evaluation
+                    record_rule_evaluation(&metrics, &rule.name);
+                    
+                    let match_start = Instant::now();
+                    let matched = rule.matches(&event);
+                    record_rule_match_duration(&metrics, &rule.name, match_start.elapsed());
+
+                    if matched {
+                        // Record successful rule match
+                        record_rule_match(&metrics, &rule.name);
                         info!("Rule '{}' matched event from {}", rule.name, event.source);
 
                         let action_name = format!("rule_{}_action", idx);
+                        let action_start = Instant::now();
+                        
                         match action_executor.execute(&action_name, &event) {
-                            Ok(result) => info!("Action executed successfully: {:?}", result),
-                            Err(e) => error!("Action execution failed: {}", e),
+                            Ok(result) => {
+                                record_action_execution(
+                                    &metrics, 
+                                    &action_name, 
+                                    true, 
+                                    action_start.elapsed()
+                                );
+                                info!("Action executed successfully: {:?}", result);
+                            }
+                            Err(e) => {
+                                record_action_execution(
+                                    &metrics, 
+                                    &action_name, 
+                                    false, 
+                                    action_start.elapsed()
+                                );
+                                error!("Action execution failed: {}", e);
+                            }
                         }
                     }
                 }
+                
+                // Record total event processing duration
+                record_event_processing_duration(&metrics, start_time.elapsed());
             }
 
             info!("Event processing loop stopped");
@@ -485,6 +537,7 @@ public class MediaKeys {
                 "New configuration validation failed: {}, keeping current config",
                 e
             );
+            record_config_reload(&self.metrics, false);
             return Err(EngineError::Config(e.to_string()));
         }
 
@@ -509,30 +562,66 @@ public class MediaKeys {
         if let Some(_sender) = &self.event_sender {
             let rules = self.rules.clone();
             let action_executor = self.action_executor.clone();
+            let metrics = self.metrics.clone();
             let mut receiver = bus::create_event_bus(self.config.engine.event_buffer_size).1;
 
             tokio::spawn(async move {
                 while let Some(event) = receiver.recv().await {
+                    let start_time = Instant::now();
+                    let event_source = event.source.clone();
+                    let event_type = format!("{:?}", event.kind);
+                    
+                    record_event(&metrics, &event_source, &event_type);
+                    
                     tracing::debug!("Processing event: {:?} from {}", event.kind, event.source);
 
                     for (idx, rule) in rules.iter().enumerate() {
                         if !rule.enabled {
                             continue;
                         }
 
-                        if rule.matches(&event) {
+                        record_rule_evaluation(&metrics, &rule.name);
+                        
+                        let match_start = Instant::now();
+                        let matched = rule.matches(&event);
+                        record_rule_match_duration(&metrics, &rule.name, match_start.elapsed());
+
+                        if matched {
+                            record_rule_match(&metrics, &rule.name);
                             info!("Rule '{}' matched event from {}", rule.name, event.source);
 
                             let action_name = format!("rule_{}_action", idx);
+                            let action_start = Instant::now();
+                            
                             match action_executor.execute(&action_name, &event) {
-                                Ok(result) => info!("Action executed successfully: {:?}", result),
-                                Err(e) => error!("Action execution failed: {}", e),
+                                Ok(result) => {
+                                    record_action_execution(
+                                        &metrics, 
+                                        &action_name, 
+                                        true, 
+                                        action_start.elapsed()
+                                    );
+                                    info!("Action executed successfully: {:?}", result);
+                                }
+                                Err(e) => {
+                                    record_action_execution(
+                                        &metrics, 
+                                        &action_name, 
+                                        false, 
+                                        action_start.elapsed()
+                                    );
+                                    error!("Action execution failed: {}", e);
+                                }
                             }
                         }
                     }
+                    
+                    record_event_processing_duration(&metrics, start_time.elapsed());
                 }
             });
         }
+        
+        record_config_reload(&self.metrics, true);
 
         let status = self.get_status();
         info!(

engine/src/engine.rs

@@ -7,7 +7,9 @@ mod service;
 mod integration_tests;
 
 use clap::Parser;
+use metrics::server::MetricsServer;
 use std::path::PathBuf;
+use std::sync::Arc;
 use tracing::{Level, error, info};
 use tracing_subscriber;
 
@@ -216,6 +218,17 @@ async fn main() {
         std::process::exit(1);
     }
 
+    // Start metrics server and cleanup task
+    let metrics = engine_instance.metrics();
+    metrics.start_cleanup_task().await;
+    let metrics_server = MetricsServer::new(metrics, 9090);
+    tokio::spawn(async move {
+        if let Err(e) = metrics_server.start().await {
+            error!("Metrics server error: {}", e);
+        }
+    });
+    info!("Metrics server available at http://127.0.0.1:9090");
+
     let status = engine_instance.get_status();
     info!(
         "Engine running with {} plugins and {} rules",
@@ -290,6 +303,7 @@ async fn main() {
     }
 
     // Shutdown
+    engine_for_shutdown.metrics().stop_cleanup_task().await;
     engine_for_shutdown.shutdown().await;
     info!("Engine stopped");
 }

engine/src/main.rs

@@ -0,0 +1,15 @@
+[package]
+name = "metrics"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+serde = { version = "1", features = ["derive"] }
+tokio = { version = "1", features = ["full"] }
+tracing = "0.1"
+chrono = { version = "0.4", features = ["serde"] }
+dashmap = "5"
+axum = "0.7"
+
+[dev-dependencies]
+tokio-test = "0.4"