test(pubub): add cleanup for topics (#3363)

Cleanup stale topics created by integration tests. Pubsub topics don't have a create time field, so instead create a label that records the creation time. There are restrictions on the characters used in labels including no uppercase letters. The wkt::Timestamp serialization uses uppercase letters, so instead this keeps it simple for the tests and stores the time in seconds.

Author: suzmue

Cargo.lock

@@ -30,6 +30,7 @@ run-showcase-tests = []
 [dependencies]
 anyhow.workspace      = true
 bytes.workspace       = true
+chrono                = { workspace = true, features = ["now"] }
 crc32c.workspace      = true
 futures.workspace     = true
 auth.workspace        = true

src/integration-tests/Cargo.toml

@@ -12,11 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::{Error, Result};
+
+use gax::paginator::ItemPaginator as _;
 use pubsub::{client::TopicAdmin, model::Topic};
 use rand::{Rng, distr::Alphanumeric};
 
-use crate::Result;
-
 pub async fn basic_topic() -> Result<()> {
     // Enable a basic subscriber. Useful to troubleshoot problems and visually
     // verify tracing is doing something.
@@ -43,7 +44,7 @@ pub async fn basic_topic() -> Result<()> {
     tracing::info!("success with get_topic={get_topic:?}");
     assert_eq!(get_topic.name, topic.name);
 
-    cleanup_test_topic(client, topic.name).await?;
+    cleanup_test_topic(&client, topic.name).await?;
 
     Ok(())
 }
@@ -61,28 +62,76 @@ fn random_topic_name(project: String) -> String {
 }
 
 pub async fn create_test_topic() -> Result<(TopicAdmin, Topic)> {
-    let project = crate::project_id()?;
+    let project_id = crate::project_id()?;
     let client = pubsub::client::TopicAdmin::builder()
         .with_tracing()
         .build()
         .await?;
-    let topic_name = random_topic_name(project);
+
+    cleanup_stale_topics(&client, &project_id).await?;
+
+    let topic_name = random_topic_name(project_id);
+    let now = chrono::Utc::now().timestamp().to_string();
 
     tracing::info!("testing create_topic()");
     let topic = client
         .create_topic()
         .set_name(topic_name)
-        .set_labels([("integration-test", "true")])
+        .set_labels([("integration-test", "true"), ("create-time", &now)])
         .send()
         .await?;
     tracing::info!("success on create_topic: {topic:?}");
 
     Ok((client, topic))
 }
 
-pub async fn cleanup_test_topic(client: TopicAdmin, topic_name: String) -> Result<()> {
+pub async fn cleanup_test_topic(client: &TopicAdmin, topic_name: String) -> Result<()> {
     tracing::info!("testing delete_topic()");
     client.delete_topic().set_topic(topic_name).send().await?;
     tracing::info!("success on delete_topic");
     Ok(())
 }
+
+pub async fn cleanup_stale_topics(client: &TopicAdmin, project_id: &str) -> Result<()> {
+    let stale_deadline = chrono::Utc::now() - chrono::Duration::hours(48);
+
+    let mut topics = client
+        .list_topics()
+        .set_project(format!("projects/{project_id}"))
+        .by_item();
+
+    let mut pending = Vec::new();
+    let mut names = Vec::new();
+    while let Some(topic) = topics.next().await {
+        let topic = topic?;
+        if topic
+            .labels
+            .get("integration-test")
+            .is_some_and(|v| v == "true")
+            && topic
+                .labels
+                .get("create-time")
+                .and_then(|v| v.parse::<i64>().ok())
+                .and_then(|s| chrono::DateTime::from_timestamp(s, 0))
+                .is_some_and(|create_time| create_time < stale_deadline)
+        {
+            let client = client.clone();
+            let name = topic.name.clone();
+            pending.push(tokio::spawn(async move {
+                cleanup_test_topic(&client, name).await
+            }));
+            names.push(topic.name);
+        }
+    }
+
+    let r: std::result::Result<Vec<_>, _> = futures::future::join_all(pending)
+        .await
+        .into_iter()
+        .collect();
+    r.map_err(Error::from)?
+        .into_iter()
+        .zip(names)
+        .for_each(|(r, name)| tracing::info!("deleting topic {name} resulted in {r:?}"));
+
+    Ok(())
+}