Struct ControllerService 

pub struct ControllerService {
    saved_workers: Arc<Mutex<Vec<WorkerSender<Result<Instruction, Status>>>>>,
    cli_sender: Arc<Mutex<Option<Sender<Result<ReplyBatch, Status>>>>>,
    measurement: Arc<RwLock<Option<MeasurementState>>>,
    unique_id: Arc<Mutex<u32>>,
    worker_config: Option<HashMap<String, u32>>,
    max_rate: Option<u32>,
    allowed_origins: Option<Vec<AllowedOrigin>>,
    access: Access,
}

The main orchestrator service struct.

Fields

saved_workers: Arc<Mutex<Vec<WorkerSender<Result<Instruction, Status>>>>>

List of connected workers

cli_sender: Arc<Mutex<Option<Sender<Result<ReplyBatch, Status>>>>>

Sender to the CLI for streaming results

measurement: Arc<RwLock<Option<MeasurementState>>>

All per-measurement state. None when idle.

unique_id: Arc<Mutex<u32>>

Last used unique worker ID

worker_config: Option<HashMap<String, u32>>

Optional static mapping of hostnames to worker IDs

max_rate: Option<u32>

Maximum probing rate (probes per second, per worker) allowed for measurements (None = unlimited)

allowed_origins: Option<Vec<AllowedOrigin>>

Optional allow-list of origins CLIs may use (None = all origins allowed)

access: Access

Which clients this instance serves

Implementations

impl ControllerService

fn validate_rate(&self, probing_rate: u32) -> Result<(), Status>

Validate a requested probing rate against the orchestrator’s configured maximum (--max_rate).

Errors

Returns an error naming the requested and maximum rate.

fn validate_origins(&self, m_def: &ScheduleMeasurement) -> Result<(), Status>

Validate the origins of a measurement definition against the orchestrator’s origin allow-list (--origins). Without an allow-list, every origin is allowed.

An origin is allowed when a rule matches its source address and permits its protocol.

Errors

Returns an error naming the refused origin and listing the available origins.

fn classify_workers( &self, m_def: &ScheduleMeasurement, ) -> Result<ClassifiedWorkers, Status>

Classify connected workers as probing, listening, or idle based on the measurement configuration. Validates that at least one worker can participate and that all configured worker IDs correspond to connected workers.

Returns the worker senders plus the participating and probing worker ID lists.

fn try_rejoin( &self, worker_id: u32, hostname: &str, tx: &Sender<Result<Instruction, Status>>, status: &Arc<Mutex<WorkerStatus>>, )

Re-admit a reconnecting worker into the active measurement, if it was participating.

Sends the Start instruction for the worker, and restores it for the task distributor.

Rejoin is refused if the measurement is finalizing.

fn init_measurement( &self, m_def: &ScheduleMeasurement, m_id: u32, participating_ids: &[u32], probing_ids: &[u32], is_gated_broadcast: bool, ) -> Result<(), Status>

Initialize the shared measurement state. Errors if a measurement is already active.

fn spawn_discovery_sweeper(&self)

Spawn the discovery-timeout sweeper for a live measurement.

Every second, expired pending discovery targets are dropped as unresponsive. The sweeper exits when the measurement ends.

fn setup_traceroute(&self, trace_options: TraceOptions, origin_id: u32)

Initialize traceroute configuration within the measurement state and spawn the timeout handler thread that monitors active trace sessions.

impl ControllerService

fn check_worker_access(&self, call: &str) -> Result<(), Status>

Refuse Worker calls on a CLI-only Orc service

fn check_cli_access(&self, call: &str) -> Result<(), Status>

Refuse CLI calls on a Worker-only Orc service

fn get_unique_id(&self) -> u32

Gets a unique worker ID for a new connecting worker. Increments the unique ID counter after returning the ID (for the next worker).

fn get_worker_id(&self, hostname: &str) -> Result<(u32, bool), Status>

Gets a worker ID for a connecting worker based on its hostname. If the hostname already exists, it returns the existing worker ID. If the hostname does not exist, it checks for a statically configured ID or generates a new unique ID.

Arguments

• hostname - the hostname of the worker

Returns

A tuple containing: the worker ID and a boolean indicating if this is a reconnection of a closed worker.

Errors

Returns an error if the hostname already exists and is used by a connected worker.

fn next_m_id(&self) -> u32

Get a random measurement ID (u16)

Trait Implementations

impl Clone for ControllerService

fn clone(&self) -> ControllerService

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Controller for ControllerService

Implementation of the Controller trait for the ControllerService Handles communication with the workers and the CLI

fn measurement_finished<'life0, 'async_trait>( &'life0 self, request: Request<Finished>, ) -> Pin<Box<dyn Future<Output = Result<Response<Ack>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Called by the worker when it has finished its current measurement. When all connected workers have finished this measurement, it will notify the CLI that the measurement is finished.

Arguments

• request - a Finished message containing the measurement ID of the measurement that has finished

Errors

Returns an error if the measurement ID is unknown.

fn worker_connect<'life0, 'async_trait>( &'life0 self, request: Request<Worker>, ) -> Pin<Box<dyn Future<Output = Result<Response<Self::WorkerConnectStream>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handles a worker connecting to this orchestrator formally. Ensures the hostname is unique and returns a unique worker ID Returns the receiver side of a stream to which the orchestrator will send tasks

Arguments

• request - a Metadata message containing the hostname of the worker

fn do_measurement<'life0, 'async_trait>( &'life0 self, request: Request<ScheduleMeasurement>, ) -> Pin<Box<dyn Future<Output = Result<Response<Self::DoMeasurementStream>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handles a measurement request from the CLI.

Classifies workers, initializes measurement state, sends Start instructions to all participating workers, optionally sets up traceroute, and launches the task distributor. Returns a stream of results to the CLI.

Errors

Returns an error if there is already an active measurement, if there are no connected workers, if the configuration references unknown worker IDs, if the probing rate exceeds the configured --max_rate, or if an origin is not allowed by this orchestrator (--configs).

fn live_measurement<'life0, 'async_trait>( &'life0 self, request: Request<Streaming<CliMessage>>, ) -> Pin<Box<dyn Future<Output = Result<Response<Self::LiveMeasurementStream>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handles a live (feed-based) measurement request from the CLI.

The first message on the stream must be the measurement definition; subsequent messages carry targets to probe.

Errors

Returns an error if the first message is not a measurement definition, if the measurement type is not catchment, if the probing rate exceeds the configured --max_rate, if an origin is not allowed by this orchestrator, if there is already an active measurement, or if no workers can participate.

fn list_workers<'life0, 'async_trait>( &'life0 self, _request: Request<Empty>, ) -> Pin<Box<dyn Future<Output = Result<Response<Status>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handle the list_clients command from the CLI.

Returns the connected clients.

fn send_result<'life0, 'async_trait>( &'life0 self, request: Request<ReplyBatch>, ) -> Pin<Box<dyn Future<Output = Result<Response<Ack>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Receive a batch of results from a worker and put it in the stream towards the CLI.

Arguments

• request - a ReplyBatch containing results from a worker

Errors

Returns an error if the CLI has disconnected.

type WorkerConnectStream = WorkerReceiver<Result<Instruction, Status>>

Server streaming response type for the WorkerConnect method.

type DoMeasurementStream = CLIReceiver<Result<ReplyBatch, Status>>

Server streaming response type for the DoMeasurement method.

type LiveMeasurementStream = CLIReceiver<Result<ReplyBatch, Status>>

Server streaming response type for the LiveMeasurement method.

impl Debug for ControllerService

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.