Develop #3
Develop #3
Conversation
| pub async fn get_pg_connection_pool(pg_url: &str, num_attempts: u32) -> Result<PgPool, anyhow::Error> { | ||
| info!("Trying to establish a PostgreSQL connection pool"); | ||
|
|
||
| let mut attempts = 0; | ||
| let mut err: Option<anyhow::Error> = None; | ||
|
|
||
| while attempts < num_attempts { | ||
| info!("Attempt to connect to PostgreSQL {} of {}", attempts + 1, num_attempts); | ||
| match PgPoolOptions::new() | ||
| .max_connections(10) | ||
| .connect(&pg_url) | ||
| .await | ||
| { | ||
| Ok(pg_con_pool) => { | ||
| info!("PostgreSQL connection successfull"); | ||
| return Ok(pg_con_pool) | ||
| }, | ||
| Err(e) => { | ||
| error!("Failed to connect to PostgreSQL. Attempt {} of {}: {}", attempts + 1, num_attempts, e); | ||
| err = Some(anyhow!(e)); | ||
| } | ||
| } | ||
| attempts += 1; | ||
| tokio::time::sleep(std::time::Duration::from_secs(5)).await; //@todo: move param somewhere else? | ||
| } | ||
| Err(err.unwrap_or_else(|| anyhow!("Failed to connect to PostgreSQL"))) | ||
| } | ||
|
|
||
|
|
||
| pub async fn check_blaze_connection(blaze_base_url: &str, num_attempts: u32) -> Result<bool, anyhow::Error> { | ||
| info!("Attempting to connect to Blaze"); | ||
|
|
||
| let mut attempts = 0; | ||
| let mut err: Option<anyhow::Error> = None; | ||
| let client = Client::new(); | ||
|
|
||
| while attempts < num_attempts { | ||
| info!("Attempt to connect to Blaze {} of {}", attempts + 1, num_attempts); | ||
| match client.get(format!("{}/health", blaze_base_url)).send().await { | ||
| Ok(_) => { | ||
| info!("Blaze connection successfull"); | ||
| return Ok(true) | ||
| }, | ||
| Err(e) => { | ||
| error!("Failed to connect to Blaze. Attempt {} of {}: {}", attempts + 1, num_attempts, e); | ||
| err = Some(anyhow!(e)); | ||
| } | ||
| } | ||
| attempts += 1; | ||
| tokio::time::sleep(std::time::Duration::from_secs(5)).await; //@todo: move param somewhere else? | ||
| } | ||
| Err(err.unwrap_or_else(|| anyhow!("Failed to connect to Blaze"))) | ||
|
|
||
| } |
There was a problem hiding this comment.
You might want to consider using tryhard as we ended up doing in focus as well for the retries.
There was a problem hiding this comment.
For focus the goal was to implement exponential backoff, right. That makes sense to me. In this case here a scheduled job is supposed to run that may take up considerable resources from the Bridgehead servers. For the time being I'd prefer to have a very easy to understand window of time when sync_blaze_2_pg is supposed to run.
| use tracing::info; | ||
|
|
||
| pub async fn wait_for_signal() { | ||
| use tokio::signal::unix::{signal,SignalKind}; |
There was a problem hiding this comment.
This will only compile on unix now because of the import. I usually just use tokio::signal::ctrl_c which sadly does not work in docker for some reason so I set STOPSIGNAL sigint in the Dockerfile.
| if let Some(next_url) = next_url { | ||
| url = next_url; | ||
| } else { | ||
| break; | ||
| } |
There was a problem hiding this comment.
Instead of a loop with this I would use a while let loop. Although you might need to fight a bit with the compiler maybe its not so simple.
There was a problem hiding this comment.
I have to admit I still don't understand why you use % operator here but I trust you that it works
if update_counter > 0 && update_counter % batch_size == 0 {
update_helper(pg_con_pool, &update_batch, &table_name).await?;
update_batch.clear();
} else if insert_counter > 0 && insert_counter % batch_size == 0 {
insert_helper(pg_con_pool, &insert_batch, &table_name).await?;
insert_batch.clear();
}
src/db_utils.rs
Outdated
| info!("Creating PostgreSQL tables"); | ||
|
|
||
| let create_tables_queries = vec
No description provided.