-
-
Notifications
You must be signed in to change notification settings - Fork 563
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(rust): wip add lessor processor to automatically revoke expired …
…tokens
- Loading branch information
1 parent
8b5013f
commit 1823bc3
Showing
6 changed files
with
131 additions
and
61 deletions.
There are no files selected for viewing
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
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 |
|---|---|---|
| @@ -1,5 +1,6 @@ | ||
| mod lease_token; | ||
| pub mod token_lessor_node_service; | ||
| mod token_lessor_worker; | ||
| mod token_lessor_processor; | ||
|
|
||
| pub use token_lessor_node_service::StartInfluxDBLeaseManagerRequest; |
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
69 changes: 69 additions & 0 deletions
69
implementations/rust/ockam/ockam_api/src/influxdb/token_lessor_processor.rs
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,69 @@ | ||
| use crate::influxdb::lease_token::LeaseToken; | ||
| use crate::influxdb::token_lessor_worker::InfluxDBTokenLessorState; | ||
| use ockam_core::{async_trait, Processor}; | ||
| use ockam_node::Context; | ||
| use std::sync::Arc; | ||
| use tokio::sync::RwLock; | ||
|
|
||
| pub(crate) struct InfluxDBTokenLessorProcessor { | ||
| state: Arc<RwLock<InfluxDBTokenLessorState>>, | ||
| } | ||
|
|
||
| impl InfluxDBTokenLessorProcessor { | ||
| pub(crate) fn new(state: Arc<RwLock<InfluxDBTokenLessorState>>) -> Self { | ||
| Self { state } | ||
| } | ||
|
|
||
| async fn list_tokens(&self) -> ockam_core::Result<Vec<LeaseToken>> { | ||
| debug!("Listing tokens"); | ||
| Ok(vec![]) | ||
| } | ||
|
|
||
| async fn revoke_outstanding_tokens(&self) -> ockam_core::Result<()> { | ||
| debug!("Revoking outstanding tokens"); | ||
| let mut state = self.state.write().await; | ||
| let expired_tokens = state | ||
| .active_tokens | ||
| .iter() | ||
| .filter(|token| token.is_expired().unwrap_or(false)) | ||
| .collect::<Vec<_>>(); | ||
| let mut to_remove = vec![]; | ||
| for token in expired_tokens { | ||
| to_remove.push(token.id.clone()); | ||
| } | ||
| state | ||
| .active_tokens | ||
| .retain(|token| !to_remove.contains(&token.id)); | ||
| Ok(()) | ||
| } | ||
| } | ||
|
|
||
| #[async_trait] | ||
| impl Processor for InfluxDBTokenLessorProcessor { | ||
| type Context = Context; | ||
|
|
||
| async fn initialize(&mut self, _context: &mut Self::Context) -> ockam_core::Result<()> { | ||
| loop { | ||
| if let Ok(tokens) = self.list_tokens().await { | ||
| let mut state = self.state.write().await; | ||
| state.active_tokens = tokens; | ||
| break; | ||
| } | ||
| tokio::time::sleep(std::time::Duration::from_secs(1)).await; | ||
| } | ||
| Ok(()) | ||
| } | ||
|
|
||
| async fn shutdown(&mut self, _context: &mut Self::Context) -> ockam_core::Result<()> { | ||
| debug!("Shutting down InfluxDBTokenLeaseManagerWorker"); | ||
| Ok(()) | ||
| } | ||
|
|
||
| async fn process(&mut self, _context: &mut Self::Context) -> ockam_core::Result<bool> { | ||
| if let Err(err) = self.revoke_outstanding_tokens().await { | ||
| error!("Failed to revoke outstanding tokens: {err}"); | ||
| } | ||
| tokio::time::sleep(std::time::Duration::from_secs(1)).await; | ||
| Ok(true) | ||
| } | ||
| } |
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.