Struct Instance

pub(crate) struct Instance<T> {
    pub workload_class: Option<String>,
    replicas: BTreeMap<ReplicaId, Replica<T>>,
    active_ingestions: BTreeMap<GlobalId, ActiveIngestion>,
    ingestion_exports: BTreeMap<GlobalId, GlobalId>,
    active_exports: BTreeMap<GlobalId, ActiveExport>,
    history: CommandHistory<T>,
    epoch: ClusterStartupEpoch,
    metrics: InstanceMetrics,
    now: NowFn,
    response_tx: UnboundedSender<(Option<ReplicaId>, StorageResponse<T>)>,
}

A controller for a storage instance.

Encapsulates communication with replicas in this instance, and their rehydration.

Note that storage objects (sources and sinks) don’t currently support replication (database-issues#5051). An instance can have muliple replicas connected, but only if it has no storage objects installed. Attempting to install storage objects on multi-replica instances, or attempting to add more than one replica to instances that have storage objects installed, is illegal and will lead to panics.

Fields

workload_class: Option<String>

The workload class of this instance.

This is currently only used to annotate metrics.

replicas: BTreeMap<ReplicaId, Replica<T>>

The replicas connected to this storage instance.

active_ingestions: BTreeMap<GlobalId, ActiveIngestion>

The ingestions currently running on this instance.

While this is derivable from history on demand, keeping a denormalized list of running ingestions is quite a bit more convenient in the implementation of StorageController::active_ingestions.

ingestion_exports: BTreeMap<GlobalId, GlobalId>

A map from ingestion export ID to the ingestion that is producing it.

active_exports: BTreeMap<GlobalId, ActiveExport>

The exports currently running on this instance.

While this is derivable from history on demand, keeping a denormalized list of running exports is quite a bit more convenient for the controller.

history: CommandHistory<T>

The command history, used to replay past commands when introducing new replicas or reconnecting to existing replicas.

epoch: ClusterStartupEpoch

The current cluster startup epoch.

The replica value of the epoch is increased every time a replica is (re)connected, allowing the distinction of different replica incarnations.

metrics: InstanceMetrics

Metrics tracked for this storage instance.

now: NowFn

A function that returns the current time.

response_tx: UnboundedSender<(Option<ReplicaId>, StorageResponse<T>)>

A sender for responses from replicas.

Responses are tagged with the ReplicaId of the replica that sent the response. Responses that don’t originate from a replica (e.g. a “paused” status update, when no replicas are connected) are tagged with None.

Implementations

impl<T> Instance<T>

where T: Timestamp + Lattice + TotalOrder, StorageGrpcClient: StorageClient<T>,

pub fn new( envd_epoch: NonZeroI64, metrics: InstanceMetrics, now: NowFn, instance_response_tx: UnboundedSender<(Option<ReplicaId>, StorageResponse<T>)>, enable_snapshot_frontier: ConfigValHandle<bool>, ) -> Self

Creates a new Instance.

pub fn replica_ids(&self) -> impl Iterator<Item = ReplicaId> + '_

Returns the IDs of all replicas connected to this storage instance.

pub fn add_replica(&mut self, id: ReplicaId, config: ReplicaConfig)

Adds a new replica to this storage instance.

pub fn replay_commands(&mut self, replica_id: ReplicaId)

Replays commands to the specified replica.

pub fn drop_replica(&mut self, id: ReplicaId)

Removes the identified replica from this storage instance.

pub fn rehydrate_failed_replicas(&mut self)

Rehydrates any failed replicas of this storage instance.

pub fn active_ingestions(&self) -> impl Iterator<Item = &GlobalId>

Returns the ingestions running on this instance.

pub fn active_exports(&self) -> impl Iterator<Item = &GlobalId>

Returns the exports running on this instance.

fn update_paused_statuses(&mut self)

Sets the status to paused for all sources/sinks in the history.

pub fn send(&mut self, command: StorageCommand<T>)

Sends a command to this storage instance.

fn absorb_ingestions(&mut self, ingestions: Vec<RunIngestionCommand>)

Updates internal state based on incoming ingestion commands.

This does not send commands to replicas, we only record the ingestion in state and potentially update scheduling decisions.

fn absorb_exports(&mut self, exports: Vec<RunSinkCommand<T>>)

Updates internal state based on incoming export commands.

This does not send commands to replicas, we only record the export in state and potentially update scheduling decisions.

fn update_scheduling(&mut self, send_commands: bool)

Update scheduling decisions, that is what replicas should be running a given object, if needed.

An important property of this scheduling algorithm is that we never change the scheduling decision for single-replica objects unless we have to, that is unless the replica that they are running on goes away. We do this, so that we don’t send a mix of “run”/“allow compaction”/“run” messages to replicas, which wouldn’t deal well with this. When we do have to make a scheduling decision we schedule a single-replica ingestion on the first replica, according to the sort order of ReplicaId. We do this latter so that the scheduling decision is stable across restarts of environmentd/the controller.

For multi-replica objects (e.g. Kafka ingestions), each active object is scheduled on all replicas.

If send_commands is true, will send commands for newly-scheduled single-replica objects.

pub fn get_ingestion_description( &self, id: &GlobalId, ) -> Option<IngestionDescription<CollectionMetadata>>

Returns the ingestion description for the given ingestion ID, if it exists.

This function searches through the command history to find the most recent RunIngestionCommand for the specified ingestion ID and returns its description. Returns None if no ingestion with the given ID is found.

pub fn get_export_description( &self, id: &GlobalId, ) -> Option<StorageSinkDesc<CollectionMetadata, T>>

Returns the export description for the given export ID, if it exists.

This function searches through the command history to find the most recent RunSinkCommand for the specified export ID and returns its description. Returns None if no ingestion with the given ID is found.

fn absorb_compactions(&mut self, cmds: Vec<(GlobalId, Antichain<T>)>)

Updates internal state based on incoming compaction commands.

fn active_replicas( &mut self, id: &GlobalId, ) -> Box<dyn Iterator<Item = &mut Replica<T>> + '_>

Returns the replicas that are actively running the given object (ingestion or export).

fn is_active_replica(&self, id: &GlobalId, replica_id: &ReplicaId) -> bool

Returns whether the given replica is actively running the given object (ingestion or export).

pub(crate) fn refresh_state_metrics(&self)

Refresh the controller state metrics for this instance.

We could also do state metric updates directly in response to state changes, but that would mean littering the code with metric update calls. Encapsulating state metric maintenance in a single method is less noisy.

This method is invoked by Controller::maintain, which we expect to be called once per second during normal operation.

pub fn get_active_replicas_for_object( &self, id: &GlobalId, ) -> BTreeSet<ReplicaId>

Returns the set of replica IDs that are actively running the given object (ingestion, ingestion export (aka. subsource), or export).

Trait Implementations

impl<T: Debug> Debug for Instance<T>

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

Formats the value using the given formatter.