Multi-Collector Coordination System#

┌─────────────────────────────────────────────────────────────────┐
│ Multi-Collector Coordination │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Task │ │ Capability │ │ Result │ │
│ │ Distributor │ │ Router │ │ Aggregator │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ └──────────────────┼──────────────────┘ │
│ │ │
│ ┌────────┴────────┐ │
│ │ Load Balancer │ │
│ │ with Failover │ │
│ └─────────────────┘ │
│ │ │
├────────────────────────────┼───────────────────────────────────┤
│ │ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │procmond-1│ │procmond-2│ │netmond-1 │ │fsmond-1 │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘

use collector_core::{TaskDistributor, TaskPriority, DistributionTask};
use daemoneye_eventbus::DaemoneyeBroker;
use std::sync::Arc;

let broker = Arc::new(DaemoneyeBroker::new("/tmp/eventbus.sock").await?);
let distributor = TaskDistributor::new(broker);

// Create and distribute a task
let task = distributor.create_task_from_event(
    &event,
    TaskPriority::High,
    Duration::from_secs(30)
)?;

distributor.distribute_task(task).await?;

use collector_core::{CapabilityRouter, CollectorCapability, SourceCaps};

let router = CapabilityRouter::new(Duration::from_secs(30));

// Register a collector
let capability = CollectorCapability {
    collector_id: "procmond-1".to_string(),
    capabilities: SourceCaps::PROCESS | SourceCaps::REALTIME,
    // ... other fields
};
router.register_collector(capability).await?;

// Route a task
let decision = router.route_task(SourceCaps::PROCESS).await?;
println!("Selected collector: {}", decision.collector_id);

use collector_core::{ResultAggregator, AggregationConfig, CollectorResult};

let config = AggregationConfig::default();
let aggregator = ResultAggregator::new(config);

// Start aggregation
let correlation_id = "task-123".to_string();
aggregator.start_aggregation(correlation_id.clone(), Some(3)).await?;

// Add results as they arrive
let result = CollectorResult { /* ... */ };
if let Some(completed) = aggregator.add_result(&correlation_id, result).await? {
    println!("Aggregation complete with {} results", completed.results.len());
}

use collector_core::{LoadBalancer, LoadBalancerConfig, LoadBalancingStrategy};

let config = LoadBalancerConfig {
    strategy: LoadBalancingStrategy::LeastConnections,
    failover_threshold: 3,
    ..Default::default()
};
let balancer = LoadBalancer::new(config);

// Register collectors
balancer.register_collector("collector-1".to_string(), 1.0).await?;

// Select a collector for task
let selected = balancer.select_collector(
    SourceCaps::PROCESS,
    &available_collectors
).await?;

// Record failures and trigger failover if needed
if balancer.record_failure("collector-1").await? {
    let event = balancer.trigger_failover(
        "collector-1",
        &available_collectors,
        SourceCaps::PROCESS
    ).await?;
}

use collector_core::*;
use daemoneye_eventbus::DaemoneyeBroker;
use std::sync::Arc;

async fn coordinate_collectors() -> Result<()> {
    // 1. Initialize components
    let broker = Arc::new(DaemoneyeBroker::new("/tmp/eventbus.sock").await?);
    let distributor = TaskDistributor::new(Arc::clone(&broker));
    let router = CapabilityRouter::new(Duration::from_secs(30));
    let aggregator = ResultAggregator::new(AggregationConfig::default());
    let balancer = LoadBalancer::new(LoadBalancerConfig::default());

    // 2. Register collectors
    let collector = CollectorCapability {
        collector_id: "procmond-1".to_string(),
        capabilities: SourceCaps::PROCESS,
        // ... other fields
    };
    router.register_collector(collector).await?;
    balancer
        .register_collector("procmond-1".to_string(), 1.0)
        .await?;

    // 3. Create and route task
    let event = CollectionEvent::Process(/* ... */);
    let task =
        distributor.create_task_from_event(&event, TaskPriority::High, Duration::from_secs(30))?;

    let routing_decision = router.route_task(SourceCaps::PROCESS).await?;

    // 4. Distribute task with load balancing
    let selected = balancer
        .select_collector(SourceCaps::PROCESS, &available_collectors)
        .await?;

    distributor.distribute_task(task).await?;

    // 5. Aggregate results
    let correlation_id = task.correlation_id.clone();
    aggregator
        .start_aggregation(correlation_id.clone(), Some(1))
        .await?;

    // Results arrive asynchronously...
    let result = CollectorResult { /* ... */ };
    if let Some(completed) = aggregator.add_result(&correlation_id, result).await? {
        println!("Task completed with {} results", completed.results.len());
    }

    Ok(())
}

cargo test --test multi_collector_coordination