Struct ComputeController

pub struct ComputeController<T: ComputeControllerTimestamp> {

    instances: BTreeMap<ComputeInstanceId, InstanceState<T>>,
    instance_workload_classes: Arc<Mutex<BTreeMap<ComputeInstanceId, Option<String>>>>,
    build_info: &'static BuildInfo,
    storage_collections: Arc<dyn StorageCollections<Timestamp = T> + Send + Sync>,
    initialized: bool,
    read_only: bool,
    config: ComputeParameters,
    peek_stash_persist_location: PersistLocation,
    stashed_response: Option<ComputeControllerResponse<T>>,
    metrics: ComputeControllerMetrics,
    now: NowFn,
    wallclock_lag: WallclockLagFn<T>,
    dyncfg: Arc<ConfigSet>,
    response_rx: UnboundedReceiver<ComputeControllerResponse<T>>,
    response_tx: UnboundedSender<ComputeControllerResponse<T>>,
    introspection_rx: Option<UnboundedReceiver<(IntrospectionType, Vec<(Row, Diff)>)>>,
    introspection_tx: UnboundedSender<(IntrospectionType, Vec<(Row, Diff)>)>,
    maintenance_ticker: Interval,
    maintenance_scheduled: bool,
}

A controller for the compute layer.

Fields

instances: BTreeMap<ComputeInstanceId, InstanceState<T>>

instance_workload_classes: Arc<Mutex<BTreeMap<ComputeInstanceId, Option<String>>>>

A map from an instance ID to an arbitrary string that describes the class of the workload that compute instance is running (e.g., production or staging).

build_info: &'static BuildInfo

storage_collections: Arc<dyn StorageCollections<Timestamp = T> + Send + Sync>

A handle providing access to storage collections.

initialized: bool

Set to true once initialization_complete has been called.

read_only: bool

Whether or not this controller is in read-only mode.

When in read-only mode, neither this controller nor the instances controlled by it are allowed to affect changes to external systems (largely persist).

config: ComputeParameters

Compute configuration to apply to new instances.

peek_stash_persist_location: PersistLocation

The persist location where we can stash large peek results.

stashed_response: Option<ComputeControllerResponse<T>>

A controller response to be returned on the next call to ComputeController::process.

metrics: ComputeControllerMetrics

The compute controller metrics.

now: NowFn

A function that produces the current wallclock time.

wallclock_lag: WallclockLagFn<T>

A function that computes the lag between the given time and wallclock time.

dyncfg: Arc<ConfigSet>

Dynamic system configuration.

Updated through ComputeController::update_configuration calls and shared with all subcomponents of the compute controller.

response_rx: UnboundedReceiver<ComputeControllerResponse<T>>

Receiver for responses produced by Instances.

response_tx: UnboundedSender<ComputeControllerResponse<T>>

Response sender that’s passed to new Instances.

introspection_rx: Option<UnboundedReceiver<(IntrospectionType, Vec<(Row, Diff)>)>>

Receiver for introspection updates produced by Instances.

When ComputeController::start_introspection_sink is first called, this receiver is passed to the introspection sink task.

introspection_tx: UnboundedSender<(IntrospectionType, Vec<(Row, Diff)>)>

Introspection updates sender that’s passed to new Instances.

maintenance_ticker: Interval

Ticker for scheduling periodic maintenance work.

maintenance_scheduled: bool

Whether maintenance work was scheduled.

Implementations

impl<T: ComputeControllerTimestamp> ComputeController<T>

pub fn new( build_info: &'static BuildInfo, storage_collections: Arc<dyn StorageCollections<Timestamp = T> + Send + Sync>, read_only: bool, metrics_registry: &MetricsRegistry, peek_stash_persist_location: PersistLocation, controller_metrics: ControllerMetrics, now: NowFn, wallclock_lag: WallclockLagFn<T>, ) -> Self

Construct a new ComputeController.

pub fn start_introspection_sink( &mut self, storage_controller: &dyn StorageController<Timestamp = T>, )

Start sinking the compute controller’s introspection data into storage.

This method should be called once the introspection collections have been registered with the storage controller. It will panic if invoked earlier than that.

pub fn instance_exists(&self, id: ComputeInstanceId) -> bool

TODO(database-issues#7533): Add documentation.

fn instance( &self, id: ComputeInstanceId, ) -> Result<&InstanceState<T>, InstanceMissing>

Return a reference to the indicated compute instance.

fn instance_mut( &mut self, id: ComputeInstanceId, ) -> Result<&mut InstanceState<T>, InstanceMissing>

Return a mutable reference to the indicated compute instance.

pub fn collection_ids( &self, instance_id: ComputeInstanceId, ) -> Result<impl Iterator<Item = GlobalId> + '_, InstanceMissing>

List the IDs of all collections in the identified compute instance.

pub fn collection_frontiers( &self, collection_id: GlobalId, instance_id: Option<ComputeInstanceId>, ) -> Result<CollectionFrontiers<T>, CollectionLookupError>

Return the frontiers of the indicated collection.

If an instance_id is provided, the collection is assumed to be installed on that instance. Otherwise all available instances are searched.

pub fn collection_reverse_dependencies( &self, instance_id: ComputeInstanceId, id: GlobalId, ) -> Result<impl Iterator<Item = GlobalId> + '_, InstanceMissing>

List compute collections that depend on the given collection.

pub async fn collection_hydrated( &self, instance_id: ComputeInstanceId, collection_id: GlobalId, ) -> Result<bool, Error>

Returns true iff the given collection has been hydrated.

For this check, zero-replica clusters are always considered hydrated. Their collections would never normally be considered hydrated but it’s clearly intentional that they have no replicas.

pub fn collections_hydrated_for_replicas( &self, instance_id: ComputeInstanceId, replicas: Vec<ReplicaId>, exclude_collections: BTreeSet<GlobalId>, ) -> Result<Receiver<bool>, Error>

Returns true if all non-transient, non-excluded collections are hydrated on any of the provided replicas.

For this check, zero-replica clusters are always considered hydrated. Their collections would never normally be considered hydrated but it’s clearly intentional that they have no replicas.

pub async fn dump(&self) -> Result<Value, Error>

Returns the state of the ComputeController formatted as JSON.

The returned value is not guaranteed to be stable and may change at any point in time.

impl<T> ComputeController<T>

where T: ComputeControllerTimestamp,

pub fn create_instance( &mut self, id: ComputeInstanceId, arranged_logs: BTreeMap<LogVariant, GlobalId>, workload_class: Option<String>, ) -> Result<(), InstanceExists>

Create a compute instance.

pub fn update_instance_workload_class( &mut self, id: ComputeInstanceId, workload_class: Option<String>, ) -> Result<(), InstanceMissing>

Updates a compute instance’s workload class.

pub fn drop_instance(&mut self, id: ComputeInstanceId)

Remove a compute instance.

Panics

Panics if the identified instance still has active replicas.

pub fn dyncfg(&self) -> &Arc<ConfigSet>

Returns the compute controller’s config set.

pub fn update_configuration(&mut self, config_params: ComputeParameters)

Update compute configuration.

pub fn initialization_complete(&mut self)

Mark the end of any initialization commands.

The implementor may wait for this method to be called before implementing prior commands, and so it is important for a user to invoke this method as soon as it is comfortable. This method can be invoked immediately, at the potential expense of performance.

pub async fn ready(&mut self)

Wait until the controller is ready to do some processing.

This method may block for an arbitrarily long time.

When the method returns, the caller should call ComputeController::process.

This method is cancellation safe.

pub fn add_replica_to_instance( &mut self, instance_id: ComputeInstanceId, replica_id: ReplicaId, location: ClusterReplicaLocation, config: ComputeReplicaConfig, ) -> Result<(), ReplicaCreationError>

Adds replicas of an instance.

pub fn drop_replica( &mut self, instance_id: ComputeInstanceId, replica_id: ReplicaId, ) -> Result<(), ReplicaDropError>

Removes a replica from an instance, including its service in the orchestrator.

pub fn create_dataflow( &mut self, instance_id: ComputeInstanceId, dataflow: DataflowDescription<Plan<T>, (), T>, subscribe_target_replica: Option<ReplicaId>, ) -> Result<(), DataflowCreationError>

Creates the described dataflow and initializes state for its output.

If a subscribe_target_replica is given, any subscribes exported by the dataflow are configured to target that replica, i.e., only subscribe responses sent by that replica are considered.

pub fn drop_collections( &mut self, instance_id: ComputeInstanceId, collection_ids: Vec<GlobalId>, ) -> Result<(), CollectionUpdateError>

Drop the read capability for the given collections and allow their resources to be reclaimed.

pub fn peek( &self, instance_id: ComputeInstanceId, peek_target: PeekTarget, literal_constraints: Option<Vec<Row>>, uuid: Uuid, timestamp: T, result_desc: RelationDesc, finishing: RowSetFinishing, map_filter_project: SafeMfpPlan, target_replica: Option<ReplicaId>, peek_response_tx: Sender<PeekResponse>, ) -> Result<(), PeekError>

Initiate a peek request for the contents of the given collection at timestamp.

pub fn cancel_peek( &self, instance_id: ComputeInstanceId, uuid: Uuid, reason: PeekResponse, ) -> Result<(), InstanceMissing>

Cancel an existing peek request.

Canceling a peek is best effort. The caller may see any of the following after canceling a peek request:

• A PeekResponse::Rows indicating that the cancellation request did not take effect in time and the query succeeded.
• A PeekResponse::Canceled affirming that the peek was canceled.
• No PeekResponse at all.

pub fn set_read_policy( &self, instance_id: ComputeInstanceId, policies: Vec<(GlobalId, ReadPolicy<T>)>, ) -> Result<(), ReadPolicyError>

Assign a read policy to specific identifiers.

The policies are assigned in the order presented, and repeated identifiers should conclude with the last policy. Changing a policy will immediately downgrade the read capability if appropriate, but it will not “recover” the read capability if the prior capability is already ahead of it.

Identifiers not present in policies retain their existing read policies.

It is an error to attempt to set a read policy for a collection that is not readable in the context of compute. At this time, only indexes are readable compute collections.

pub fn acquire_read_hold( &self, instance_id: ComputeInstanceId, collection_id: GlobalId, ) -> Result<ReadHold<T>, CollectionUpdateError>

Acquires a ReadHold for the identified compute collection.

fn determine_time_dependence( &self, instance_id: ComputeInstanceId, dataflow: &DataflowDescription<Plan<T>, (), T>, ) -> Result<Option<TimeDependence>, TimeDependenceError>

Determine the time dependence for a dataflow.

pub fn process(&mut self) -> Option<ComputeControllerResponse<T>>

Processes the work queued by ComputeController::ready.

fn maintain(&mut self)