DaemonEye EventBus Message Schemas#

message EventBusMessage {
  MessageMetadata metadata = 1;
  CorrelationMetadata correlation = 2;
  oneof payload {
    CollectionEventPayload collection_event = 3;
    RpcRequestPayload rpc_request = 4;
    RpcResponsePayload rpc_response = 5;
    ControlMessagePayload control_message = 6;
    HeartbeatPayload heartbeat = 7;
    AlertPayload alert = 8;
  }
}

message CorrelationMetadata {
  string correlation_id = 1; // Primary correlation ID
  optional string parent_correlation_id = 2; // Hierarchical relationships
  string root_correlation_id = 3; // Workflow root
  optional string trace_id = 4; // Distributed tracing
  optional string span_id = 5; // Tracing span
  uint64 sequence_number = 6; // Event ordering
  optional string workflow_stage = 7; // Pipeline stage
  map<string, string> correlation_tags = 8; // Custom grouping
}

message ProcessEventData {
  ProcessRecord process = 1; // Core process information
  ProcessEventSubtype subtype = 2; // Event type (created, updated, terminated)
  optional ProcessTreeContext tree_context = 3; // Parent/child relationships
  optional SecurityContext security_context = 4; // User, group, privileges
  optional ProcessPerformanceMetrics performance = 5; // CPU, memory, I/O
}

message NetworkEventData {
  NetworkRecord network = 1; // Connection information
  NetworkEventSubtype subtype = 2; // Connection state changes
  optional NetworkConnectionMetadata connection_metadata = 3; // Duration, flags
  optional TrafficAnalysisResults traffic_analysis = 4; // Threat detection
}

message FilesystemEventData {
  FilesystemRecord filesystem = 1; // File operation details
  FilesystemEventSubtype subtype = 2; // Operation type
  optional FileMetadata file_metadata = 3; // File type, attributes
  optional AccessPatternAnalysis access_analysis = 4; // Anomaly detection
}

message PerformanceEventData {
  PerformanceRecord performance = 1; // Metric information
  PerformanceEventSubtype subtype = 2; // Threshold, anomaly events
  optional ThresholdInfo threshold_info = 3; // Threshold violations
  optional TrendAnalysis trend_analysis = 4; // Trend detection
}

message RpcRequestPayload {
  RpcRequest request = 1; // Core request information
  RpcRoutingInfo routing = 2; // Load balancing, circuit breaker
  optional RpcAuthInfo auth_info = 3; // Authentication/authorization
}

message RpcResponsePayload {
  RpcResponse response = 1; // Core response information
  RpcRoutingInfo routing = 2; // Response routing
  optional RpcPerformanceMetrics performance_metrics = 3; // Timing metrics
}

message MessageVersion {
  uint32 major = 1; // Breaking changes
  uint32 minor = 2; // Backward compatible additions
  uint32 patch = 3; // Bug fixes
  string protocol_version = 4; // Human-readable version
}

message VersionNegotiationRequest {
  repeated MessageVersion supported_versions = 1; // Client capabilities
  MessageVersion preferred_version = 2; // Client preference
  repeated string client_capabilities = 3; // Feature requirements
}

message VersionNegotiationResponse {
  MessageVersion negotiated_version = 1; // Agreed version
  repeated string server_capabilities = 2; // Available features
  repeated string compatibility_warnings = 3; // Deprecation notices
  bool negotiation_success = 4; // Success status
}

enum DeliveryGuarantee {
  AT_MOST_ONCE = 0; // Fire and forget
  AT_LEAST_ONCE = 1; // With retries
  EXACTLY_ONCE = 2; // With deduplication
}

message RetentionPolicy {
  uint64 retention_duration_ms = 1; // Time-based retention
  optional uint32 max_retained_messages = 2; // Count-based retention
  optional uint64 max_retained_size_bytes = 3; // Size-based retention
}

message RpcAuthInfo {
  optional string client_cert_fingerprint = 1; // mTLS authentication
  optional string auth_token = 2; // Bearer token
  repeated string permissions = 3; // Access control
  string auth_method = 4; // Authentication method
}

use daemoneye_eventbus::{EventBus, CollectionEvent, CorrelationMetadata};

// Create correlation metadata
let correlation = CorrelationMetadata::new()
    .with_stage("process_analysis".to_string())
    .with_tag("collector".to_string(), "procmond".to_string());

// Publish event with correlation
let event = CollectionEvent::Process(process_event);
event_bus.publish_with_correlation(event, correlation).await?;

use daemoneye_eventbus::{CollectorRpcClient, CollectorOperation, RpcRequest};

// Create RPC client
let mut rpc_client = CollectorRpcClient::new("control.collector.procmond").await?;

// Make lifecycle request
let request = RpcRequest::lifecycle(
    "agent-id".to_string(),
    "control.collector.procmond".to_string(),
    CollectorOperation::Start,
    lifecycle_request,
    Duration::from_secs(30)
);

let response = rpc_client.call(request, Duration::from_secs(30)).await?;

use daemoneye_eventbus::{VersionNegotiator, VersionNegotiationRequest};

// Negotiate version with server
let negotiator = VersionNegotiator::new();
let request = VersionNegotiationRequest::for_current_version("client-id".to_string());
let response = negotiator.negotiate_version(&request)?;

// Check negotiated capabilities
if response.server_capabilities.contains(&"advanced_routing".to_string()) {
    // Use advanced routing features
}