Refactoring procmond to Use collector-core#

The procmond daemon now supports a dual-mode architecture with an actor-based monitor collector. It can operate in two modes:

• Actor Mode (with Broker): When the DAEMONEYE_BROKER_SOCKET environment variable is set, procmond runs as an actor, coordinating with the daemoneye-agent broker via message passing and an event bus connector. This enables coordinated startup, dynamic backpressure handling, crash-recoverable event delivery, and now also supports registration, heartbeat, and RPC-based control via the new RegistrationManager and RpcServiceHandler components.
• Standalone Mode: If no broker socket is configured, procmond falls back to the collector-core event source model, running independently as before.

The actor pattern introduces a message-passing architecture for the monitor collector, enabling coordinated state transitions (such as health checks, configuration updates, and graceful shutdown) and dynamic runtime behaviors (like interval adjustment on backpressure). With the addition of the registration and RPC subsystems, procmond can now register with the agent, periodically publish heartbeats, handle deregistration on shutdown, and respond to remote procedure calls for health checks, configuration updates, and shutdown requests.

Configuration can now be hot-reloaded at collection cycle boundaries, and event delivery is made crash-recoverable via a write-ahead log (WAL) when using the event bus connector. This architecture improves modularity, reliability, and integration with the broader DaemonEye ecosystem, while maintaining compatibility with the previous collector-core event source model.

Linting and Code Quality#

• Edition: Rust 2024 (MSRV: 1.91+)
• Linting: cargo clippy -- -D warnings (zero warnings)
• Format args: Use {variable} inlined syntax in format!/anyhow! macros (clippy::uninlined_format_args)
• If-else ordering: Clippy prefers == checks first in if-else (clippy::unnecessary_negation)
• Safety: unsafe_code = "forbid" at workspace level
• Formatting: rustfmt with 119 char line length
• Rustdoc: Escape brackets in paths like /proc/\[pid\]/stat to avoid broken link warnings

See below for details on the new registration and RPC service integration.

New CLI Options for Process Collection#

Process collection parameters are configured via explicit CLI options, which map directly to the ProcessSourceConfig struct. In addition, the DAEMONEYE_BROKER_SOCKET environment variable determines whether procmond operates in actor mode (with broker coordination) or standalone mode.

• --database <path>: Path to the process database (default: /var/lib/daemoneye/processes.db)
• --log-level <level>: Logging verbosity (default: info)
• --interval <seconds>: Collection interval in seconds (default: 30, min: 5, max: 3600)
• --max-processes <number>: Maximum processes to collect per cycle (0 for unlimited, default: 0)
• --enhanced-metadata: Enable enhanced metadata collection (boolean flag)
• --compute-hashes: Enable executable hashing (boolean flag)
• DAEMONEYE_BROKER_SOCKET: If set, enables actor mode and connects to the specified broker socket for coordinated operation.

Example usage:

DAEMONEYE_BROKER_SOCKET=/run/daemoneye/broker.sock procmond --database /tmp/proc.db --interval 60 --max-processes 1000 --enhanced-metadata --compute-hashes --log-level debug

In actor mode, some configuration (such as collection interval and lifecycle thresholds) can be hot-reloaded at runtime via coordinated messages. Other parameters still require a restart to take effect.

Removal of IPC Socket Configuration#

The previous IPC socket configuration and dedicated IPC server for process data have been removed. IPC is now handled internally by collector-core's CollectorIpcServer, which manages communication with external components (such as daemoneye-agent) using protobuf protocol and CRC32 framing. This architectural change centralizes configuration and lifecycle management, and removes the need for a separate IPC server or user-facing IPC socket options collector.rs.

Integration of ProcessEventSource into the Collector Lifecycle#

procmond now supports two integration modes:

• Actor-Based Monitor Collector (Actor Mode): When running with a broker, the monitor collector operates as an actor, processing messages sequentially via a bounded channel. The actor supports coordinated state management through typed messages, including health checks, configuration updates, graceful shutdown, begin monitoring (startup coordination), and dynamic interval adjustment in response to backpressure. The actor's state transitions are explicit and observable, and configuration can be hot-reloaded at collection cycle boundaries.

  In actor mode, the monitor collector integrates with an EventBusConnector for crash-recoverable event delivery, using a write-ahead log (WAL) to buffer events until they are acknowledged by the broker. Backpressure signals from the event bus dynamically adjust the collection interval (e.g., slowing down by 1.5x when under load). Startup is coordinated with the agent: the collector waits for a BeginMonitoring command before starting collection, ensuring all components are ready.

  Registration and Heartbeat:

  • The RegistrationManager handles the full registration lifecycle with the daemoneye-agent, including initial registration, periodic heartbeat publishing (with health and metrics), and graceful deregistration on shutdown. Registration is retried with backoff on failure, and the heartbeat interval can be assigned by the agent.
  • Heartbeats are published in a background task and include health status, buffer usage, and other metrics. Deregistration is performed on shutdown if the collector is registered.

  RPC Service:

  • The RpcServiceHandler subscribes to the control topic for the collector and handles incoming RPC requests from the agent. Supported operations include HealthCheck, UpdateConfig, and GracefulShutdown. Each request is dispatched to the actor as needed, and responses are prepared for publishing (full integration pending generic message support in the event bus connector).
  • The RPC service tracks statistics for requests, errors, and timeouts, and ensures that requests are handled within their deadlines.

• ProcessEventSource (Standalone Mode): If no broker is configured, procmond uses the legacy ProcessEventSource, which implements the collector-core EventSource trait and streams process events directly to the collector runtime.

The actor pattern enables robust, coordinated, and dynamically adjustable process monitoring, while maintaining compatibility with the collector-core event source model for standalone operation.

See the following sections for example integration and lifecycle diagrams including registration and RPC flows.

Example Integration#

Standalone mode example (no broker):

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

let db_manager = Arc::new(Mutex::new(DatabaseManager::new("/var/lib/daemoneye/processes.db")?));
let config = ProcessSourceConfig {
    collection_interval: Duration::from_secs(60),
    collect_enhanced_metadata: false,
    max_processes_per_cycle: 1000,
    compute_executable_hashes: true,
};
let process_source = ProcessEventSource::with_config(db_manager, config);

let mut collector = Collector::new(CollectorConfig::default());
collector.register(Box::new(process_source))?;
collector.run().await?;

Actor mode example (with broker):

use procmond::monitor_collector::{ProcmondMonitorCollector, ProcmondMonitorConfig};
use procmond::event_bus_connector::EventBusConnector;
use daemoneye_lib::storage::DatabaseManager;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{Mutex, mpsc};

let db_manager = Arc::new(Mutex::new(DatabaseManager::new("/var/lib/daemoneye/processes.db")?));
let monitor_config = ProcmondMonitorConfig::default();
let (actor_handle, message_receiver) = ProcmondMonitorCollector::create_channel();
let mut collector = ProcmondMonitorCollector::new(db_manager, monitor_config, message_receiver)?;

let mut event_bus_connector = EventBusConnector::new("/var/lib/daemoneye/wal").await?;
event_bus_connector.connect().await?;
collector.set_event_bus_connector(event_bus_connector);

let (event_tx, mut event_rx) = mpsc::channel(1000);
tokio::spawn(async move {
    collector.run(event_tx).await.unwrap();
});

// Send BeginMonitoring to start collection
actor_handle.begin_monitoring()?;

Lifecycle Diagram#

In actor mode, startup is coordinated via messages and the collector waits for a BeginMonitoring command from the agent before starting collection. Backpressure signals from the EventBusConnector can dynamically adjust the collection interval. The RegistrationManager manages registration, heartbeat, and deregistration with the agent, while the RpcServiceHandler processes control requests from the agent and interacts with the actor.

Testing Guidance#

Testing for the updated procmond daemon should cover both standalone and actor modes:

• Unit and Integration Tests for Actor-Based Collector: Test creation, message-passing (health check, config update, graceful shutdown), state transitions, and event delivery in actor mode. Use Tokio async tests and simulate broker signals and backpressure.
• Dual-Mode Startup Tests: Validate correct startup behavior with and without the DAEMONEYE_BROKER_SOCKET environment variable. Ensure that actor mode and standalone mode both function as expected.
• Configuration Hot-Reload Tests: In actor mode, test that configuration updates (such as collection interval) are applied at cycle boundaries and reflected in health checks.
• Backpressure and WAL Tests: Simulate backpressure signals and verify that the collection interval is dynamically adjusted and that events are buffered and replayed from the WAL as needed.
• CLI Tests: Validate argument parsing and option handling for all CLI parameters, including the effect of the broker environment variable.

Example actor-based test:

#[tokio::test]
async fn test_actor_health_check() {
    let db_manager = create_test_database().await;
    let config = ProcmondMonitorConfig::default();
    let (collector, handle) = create_collector_with_channel(db_manager, config).unwrap();
    // Spawn the actor run loop in a background task
    let (event_tx, _event_rx) = mpsc::channel(100);
    tokio::spawn(async move {
        collector.run(event_tx).await.unwrap();
    });
    // Send health check message
    let health = handle.health_check().await.unwrap();
    assert_eq!(health.state, CollectorState::WaitingForAgent);
}

See monitor_collector.rs for more actor-based test examples.

Configuration Changes#

Configuration is now streamlined and supports runtime updates in actor mode. The database path and process collection parameters are set via CLI options as before. When running in actor mode (with DAEMONEYE_BROKER_SOCKET set), certain configuration parameters—such as collection interval and lifecycle thresholds—can be hot-reloaded at runtime via coordinated messages to the actor. Other parameters (such as excluded PIDs or enabling event-driven mode) still require a restart.

The collector is configured via CollectorConfig and, in actor mode, integrates with an EventBusConnector for reliable event delivery. The environment variable DAEMONEYE_BROKER_SOCKET determines whether procmond operates in coordinated actor mode or standalone mode.

Registration and Heartbeat:

• Registration with the agent is managed by the RegistrationManager, which handles retries, agent-assigned heartbeat intervals, and deregistration on shutdown.
• Heartbeat intervals and assigned topics may be provided by the agent at registration time and are used for periodic health reporting.

RPC Control:

• The RpcServiceHandler enables remote health checks, configuration updates, and shutdown via RPC requests from the agent. Supported configuration updates are applied at runtime if possible; others require a restart.

Update deployment scripts and configuration files to set the appropriate environment variables for your environment, and ensure that agent-side configuration is compatible with the registration and RPC control flows.