ProcessEventSource in procmond#

The ProcessEventSource component in procmond is responsible for enumerating system processes, collecting metadata, and emitting process events as part of the unified collector-core event collection framework. It replaces the legacy IPC server logic by integrating directly with the collector-core lifecycle, enabling asynchronous event streaming, capability negotiation, and topic-based event publishing.

Architecture and Integration#

ProcessEventSource wraps the existing ProcessMessageHandler logic and implements the EventSource trait from collector-core. This allows it to participate fully in the collector lifecycle, including initialization, event collection, health checking, and graceful shutdown. Events are published to topics (such as collector/process/events) and routed by a central message broker in daemoneye-agent to subscribers, supporting flexible pub/sub communication and multi-collector coordination. The component can also emit trigger events (e.g., CollectionEvent::TriggerRequest) to request on-demand analysis from other collectors such as binary_hasher (source).

Diagram: Integration Overview#

Configuration Parameters#

Configuration is managed via the ProcessSourceConfig struct, which exposes the following parameters (source):

• interval (collection_interval: Duration): The interval between process collection cycles. Default is 30 seconds. Configurable via CLI as --interval (minimum 5, maximum 3600 seconds).
• max_processes (max_processes_per_cycle: usize): The maximum number of processes to collect per cycle. 0 means unlimited. Configurable via CLI as --max_processes.
• enhanced_metadata (collect_enhanced_metadata: bool): Enables collection of detailed process metadata (such as start time, CPU usage, memory usage). Default is true. Configurable via CLI as --enhanced_metadata.
• compute_hashes (compute_executable_hashes: bool): Enables hashing of process binaries. Default is false. Configurable via CLI as --compute_hashes.

Example: Default Configuration#

let config = ProcessSourceConfig {
    collection_interval: Duration::from_secs(30),
    collect_enhanced_metadata: true,
    max_processes_per_cycle: 0,
    compute_executable_hashes: false,
};

CLI Options#

The procmond daemon exposes CLI options that map directly to ProcessEventSource configuration:

--interval <seconds> Collection interval (min: 5, max: 3600, default: 30)
--max_processes <count> Maximum processes per cycle (0 = unlimited, default: 0)
--enhanced_metadata Enable enhanced metadata collection (default: true)
--compute_hashes Enable executable hashing (default: false)

Example invocation:

procmond --interval 10 --max_processes 500 --enhanced_metadata --compute_hashes

Usage Examples#

Instantiating ProcessEventSource#

use procmond::event_source::{ProcessEventSource, ProcessSourceConfig};
use daemoneye_lib::storage::DatabaseManager;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;

// Create database manager
let db_manager = Arc::new(Mutex::new(
    DatabaseManager::new("/var/lib/daemoneye/processes.db")
        .expect("Failed to create database manager")
));

// Default configuration
let source = ProcessEventSource::new(db_manager.clone());

// Custom configuration
let config = ProcessSourceConfig {
    collection_interval: Duration::from_secs(60),
    collect_enhanced_metadata: false,
    max_processes_per_cycle: 1000,
    compute_executable_hashes: true,
};
let source = ProcessEventSource::with_config(db_manager, config);

// Register with collector-core
collector.register(Box::new(source))?;

(source)

Collector Lifecycle Integration#

ProcessEventSource is registered with the collector-core Collector instance. The collector manages the lifecycle, IPC, and event processing. When collector.run() is called, the event source participates in the unified lifecycle, including health checks and graceful shutdown (source).

Replacement of Legacy IPC Server Logic#

Previously, process enumeration and event emission were handled by a dedicated IPC server using ad-hoc logic. ProcessEventSource replaces this by implementing the EventSource trait, integrating directly with collector-core's IPC server infrastructure. Events are now published to topics and routed by a central message broker, supporting flexible pub/sub communication and multi-collector coordination. This removes the need for separate IPC server code and enables robust, topic-based event delivery (source).

Event Model#

Each process is converted to a ProcessEvent struct, which includes fields such as pid, ppid, name, executable_path, command_line, start_time (if enhanced metadata enabled), cpu_usage, memory_usage, executable_hash (if enabled), user_id, accessible, file_exists, and timestamp. Events are emitted as CollectionEvent::Process via the collector-core event bus (source).

Testing Strategies#

Testing for ProcessEventSource includes:

• Unit tests for creation, configuration, event conversion, health check, and capability flags. These use temporary databases and tokio async tests.
• Integration tests for lifecycle management and agent integration, including event publishing and topic routing.
• Performance tests for process enumeration speed and resource usage.
• Security tests for privilege handling and metadata sanitization.

Example unit test patterns:

#[tokio::test]
async fn test_process_event_source_creation() {
    let db_manager = create_test_database();
    let source = ProcessEventSource::new(db_manager);
    assert_eq!(source.name(), "process-monitor");
    assert!(source.capabilities().contains(SourceCaps::PROCESS));
}

(source)

Integration tests for topic-based event routing and multi-collector scenarios are recommended, including tests for wildcard subscriptions, backpressure, and graceful shutdown (source).

For further details, refer to the procmond source code and collector-core documentation.