-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor: Move Data Layer provisioning from Runner to Coordinator (#805)
This PR updates Coordinator to handle Data Layer provisioning, removing the implicit step from Runner. Provisioning itself is still completed within Runner, but Coordinator will trigger and monitor it. Functionally, provisioning is the same, but there are some subtle differences around how it is handled: - Provisioning will now happen as soon as the Indexer is registered, rather than when the first matched block is executed. - Block Streams/Executors will not be started until provisioning is successful, neither will start when either pending or failed. A `provisioned_state` enum has been added to the Indexer state within Redis. This is used to persist what stage of provisioning the Data Layer is at, as well as ensuring we only provision once. ## Concerns with current implementation Overall, I'm not happy with the implementation here, but feel it is the best we can do given the current structure of Coordinator. As to not block this feature I decided to go forward with this approach, and will create a ticket to refactor/update later. Provisioning is triggered within the "new" handler, and then polled within the "existing" handler, which seems a little awkward. The separated handling is necessary since no operation within the control loop (i.e. `Synchroniser`) should block, as that would block synchronisation for all other Indexers. So we need to trigger the provisioning initially, and then poll the completion each subsequent control loop. I feel we have outgrown the current control loop, and am planning to refactor later. Rather than have a single control loop for _all_ Indexers, I'm thinking we can have dedicated loops for each of them. We could spawn a new task for each Indexer, which then manages its own lifecycle. Then each Indexer is free to wait for as long as it wants, without impacting other Indexers. This would allow us to handle the blocking provisioning step much more elegantly.
- Loading branch information
1 parent
5462ec5
commit 973efd6
Showing
19 changed files
with
461 additions
and
422 deletions.
There are no files selected for viewing
File renamed without changes.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,96 @@ | ||
| #![cfg_attr(test, allow(dead_code))] | ||
|
|
||
| pub use runner::data_layer::ProvisioningStatus; | ||
|
|
||
| use anyhow::Context; | ||
| use runner::data_layer::data_layer_client::DataLayerClient; | ||
| use runner::data_layer::{CheckProvisioningTaskStatusRequest, ProvisionRequest}; | ||
| use tonic::transport::channel::Channel; | ||
| use tonic::{Request, Status}; | ||
|
|
||
| use crate::indexer_config::IndexerConfig; | ||
|
|
||
| #[cfg(not(test))] | ||
| pub use DataLayerHandlerImpl as DataLayerHandler; | ||
| #[cfg(test)] | ||
| pub use MockDataLayerHandlerImpl as DataLayerHandler; | ||
|
|
||
| pub struct DataLayerHandlerImpl { | ||
| client: DataLayerClient<Channel>, | ||
| } | ||
|
|
||
| #[cfg_attr(test, mockall::automock)] | ||
| impl DataLayerHandlerImpl { | ||
| pub fn from_env() -> anyhow::Result<Self> { | ||
| let runner_url = std::env::var("RUNNER_URL").context("RUNNER_URL is not set")?; | ||
| Self::connect(&runner_url) | ||
| } | ||
|
|
||
| pub fn connect(runner_url: &str) -> anyhow::Result<Self> { | ||
| let channel = Channel::from_shared(runner_url.to_string()) | ||
| .context("Runner URL is invalid")? | ||
| .connect_lazy(); | ||
| let client = DataLayerClient::new(channel); | ||
|
|
||
| Ok(Self { client }) | ||
| } | ||
|
|
||
| pub async fn start_provisioning_task( | ||
| &self, | ||
| indexer_config: &IndexerConfig, | ||
| ) -> anyhow::Result<ProvisioningStatus> { | ||
| let request = ProvisionRequest { | ||
| account_id: indexer_config.account_id.to_string(), | ||
| function_name: indexer_config.function_name.clone(), | ||
| schema: indexer_config.schema.clone(), | ||
| }; | ||
|
|
||
| let response = self | ||
| .client | ||
| .clone() | ||
| .start_provisioning_task(Request::new(request)) | ||
| .await; | ||
|
|
||
| if let Err(error) = response { | ||
| if error.code() == tonic::Code::AlreadyExists { | ||
| return Ok(ProvisioningStatus::Pending); | ||
| } | ||
|
|
||
| return Err(error.into()); | ||
| } | ||
|
|
||
| let status = match response.unwrap().into_inner().status { | ||
| 1 => ProvisioningStatus::Pending, | ||
| 2 => ProvisioningStatus::Complete, | ||
| 3 => ProvisioningStatus::Failed, | ||
| _ => ProvisioningStatus::Unspecified, | ||
| }; | ||
|
|
||
| Ok(status) | ||
| } | ||
|
|
||
| pub async fn check_provisioning_task_status( | ||
| &self, | ||
| indexer_config: &IndexerConfig, | ||
| ) -> anyhow::Result<ProvisioningStatus> { | ||
| let request = CheckProvisioningTaskStatusRequest { | ||
| account_id: indexer_config.account_id.to_string(), | ||
| function_name: indexer_config.function_name.clone(), | ||
| }; | ||
|
|
||
| let response = self | ||
| .client | ||
| .clone() | ||
| .check_provisioning_task_status(Request::new(request)) | ||
| .await?; | ||
|
|
||
| let status = match response.into_inner().status { | ||
| 1 => ProvisioningStatus::Pending, | ||
| 2 => ProvisioningStatus::Complete, | ||
| 3 => ProvisioningStatus::Failed, | ||
| _ => ProvisioningStatus::Unspecified, | ||
| }; | ||
|
|
||
| Ok(status) | ||
| } | ||
| } |
File renamed without changes.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,3 @@ | ||
| pub mod block_streams; | ||
| pub mod data_layer; | ||
| pub mod executors; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.