Merge branch 'main' into ntran/compact_os_sql_tests

influxdb_iox/tests/common/server_fixture.rs

@@ -286,7 +286,11 @@ impl ServerFixture {
             {
                 return status;
             }
-            assert!(t_0.elapsed() < Duration::from_secs(10));
+            assert!(
+                t_0.elapsed() < Duration::from_secs(10),
+                "Server and all databases are not initialized. Status:\n\n {:#?}",
+                status
+            );
             tokio::time::sleep(Duration::from_millis(100)).await;
         }
     }

influxdb_iox/tests/end_to_end_cases/database_migration.rs

@@ -18,7 +18,8 @@ use crate::{
     common::server_fixture::{ServerFixture, ServerType},
     end_to_end_cases::scenario::{
         collect_query, create_readable_database, data_dir, db_data_dir, rand_name,
-        wait_for_exact_chunk_states, wait_for_operations_to_complete,
+        wait_for_database_initialized, wait_for_exact_chunk_states,
+        wait_for_operations_to_complete,
     },
 };
 
@@ -230,7 +231,7 @@ async fn migrate_table_files_from_one_server_to_another() {
     wait_for_operations_to_complete(&fixture, &db_name, Duration::from_secs(5)).await;
 
     // Wait for all databases to complete re-initialization here
-    fixture.wait_server_initialized().await;
+    wait_for_database_initialized(&fixture, &db_name, Duration::from_secs(5)).await;
 
     // Now the data shoudl be available for the_table
     let query_results = flight_client

influxdb_iox/tests/end_to_end_cases/scenario.rs

@@ -585,7 +585,41 @@ pub async fn wait_for_operations_to_complete(
                 db_name, wait_time, operations
             );
         }
-        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+    }
+}
+
+pub async fn wait_for_database_initialized(
+    fixture: &ServerFixture,
+    db_name: &str,
+    wait_time: std::time::Duration,
+) {
+    use generated_types::influxdata::iox::management::v1::database_status::DatabaseState;
+
+    let t_start = std::time::Instant::now();
+    let mut management_client = fixture.management_client();
+
+    loop {
+        let status = management_client.get_server_status().await.unwrap();
+        if status
+            .database_statuses
+            .iter()
+            .filter(|status| status.db_name == db_name)
+            .all(|status| {
+                DatabaseState::from_i32(status.state).unwrap() == DatabaseState::Initialized
+            })
+        {
+            println!("Database {} is initialized", db_name);
+            return;
+        }
+
+        if t_start.elapsed() >= wait_time {
+            panic!(
+                "Database {} was not initialized in {:?}:\n\n{:#?}",
+                db_name, wait_time, status
+            );
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
     }
 }