Implementation:ArroyoSystems Arroyo Controller Server

Overview

ControllerServer is the central coordination service for the Arroyo streaming engine. It implements the ControllerGrpc trait to manage pipeline lifecycle, worker registration, heartbeats, checkpoint coordination, and metric collection via gRPC.

Description

The controller server is defined in lib.rs and serves as the main entry point for the arroyo-controller crate. It manages the full lifecycle of streaming jobs through a state machine pattern (defined in a separate states module). Key responsibilities include:

• Worker management: Registering workers, processing heartbeats, and handling worker completion notifications via RegisterWorkerReq, HeartbeatReq, and WorkerFinishedReq.
• Checkpoint coordination: Processing TaskCheckpointCompletedReq and TaskCheckpointEventReq messages from workers.
• Job state tracking: Maintaining JobConfig and JobStatus structs per pipeline, including stop mode, checkpoint interval, TTL, parallelism overrides, and restart configuration.
• Metrics collection: Integrating with JobMetrics for per-operator metrics and exposing Prometheus gauges for active pipelines and compaction tuples.
• Sink data routing: Forwarding output data subscriptions to consumers via GrpcOutputSubscription.

The server uses a database-backed approach (via cornucopia_async) for persistent job state and supports both ProcessScheduler and NodeScheduler backends.

Usage

The controller is started as a gRPC server listening on a configured address. It is typically launched as the main process in an Arroyo deployment and coordinates all other components.

Code Reference

Source Location

crates/arroyo-controller/src/lib.rs

Signature

pub struct JobConfig {
    id: Arc<String>,
    organization_id: String,
    pipeline_name: String,
    pipeline_id: i64,
    stop_mode: StopMode,
    checkpoint_interval: Duration,
    ttl: Option<Duration>,
    parallelism_overrides: HashMap<u32, usize>,
    restart_nonce: i32,
    restart_mode: RestartMode,
    ignore_state_before_epoch: Option<i32>,
}

pub struct JobStatus {
    id: Arc<String>,
    run_id: u64,
    state: String,
    // ...
}

Import

use arroyo_controller::{JobConfig, JobStatus};

I/O Contract

Inputs

Name	Type	Description
RegisterWorkerReq	gRPC message	Worker registration with machine ID, task slots, and address
HeartbeatReq	gRPC message	Worker heartbeat with current metric counters
TaskCheckpointCompletedReq	gRPC message	Notification that a task has completed its checkpoint
SinkDataReq	gRPC message	Sink data from workers for output subscriptions

Outputs

Name	Type	Description
HeartbeatResp	gRPC message	Acknowledgement of worker heartbeat
JobMetricsResp	gRPC message	Aggregated per-operator metrics for a job
OutputData	gRPC stream	Streaming output data for subscribed consumers

Usage Examples

// The controller is typically started via the arroyo_server_common framework
// and configured through environment variables or config files
let config = config();
let scheduler: Box<dyn Scheduler> = if config.use_kubernetes {
    Box::new(KubernetesScheduler::new().await)
} else {
    Box::new(ProcessScheduler::new())
};

Related Pages

• Implementation:ArroyoSystems_Arroyo_Job_Metrics - Per-job metric collection used by the controller
• Implementation:ArroyoSystems_Arroyo_Scheduler_Trait - Scheduler abstraction for worker lifecycle
• Implementation:ArroyoSystems_Arroyo_Kubernetes_Scheduler - Kubernetes-based scheduler implementation