Logging: unflake systests (#5698)

* Harden 'test_list_metrics' against concurrent CI runs.

Closes #5543.

* Harden 'test_list_sinks' against concurrent CI runs.

Closes #5692.

* Harden 'test_log_handler_async' against concurrent CI runs.

Use unique ID for handler, and therefore the underlying back-end log.

Closes #5632.
Closes #5556.

Author: tseaver

tests/system/test_system.py

@@ -231,9 +231,10 @@ def test_log_struct(self):
     def test_log_handler_async(self):
         LOG_MESSAGE = 'It was the worst of times'
 
-        handler = CloudLoggingHandler(Config.CLIENT)
+        handler_name = 'gcp-async' + unique_resource_id('-')
+        handler = CloudLoggingHandler(Config.CLIENT, name=handler_name)
         # only create the logger to delete, hidden otherwise
-        logger = Config.CLIENT.logger(handler.name)
+        logger = Config.CLIENT.logger(handler_name)
         self.to_delete.append(logger)
 
         cloud_logger = logging.getLogger(handler.name)
@@ -334,14 +335,14 @@ def test_list_metrics(self):
             METRIC_NAME, DEFAULT_FILTER, DEFAULT_DESCRIPTION)
         self.assertFalse(metric.exists())
         before_metrics = list(Config.CLIENT.list_metrics())
-        before_names = set(metric.name for metric in before_metrics)
+        before_names = set(before.name for before in before_metrics)
+        self.failIf(metric.name in before_names)
         metric.create()
         self.to_delete.append(metric)
         self.assertTrue(metric.exists())
         after_metrics = list(Config.CLIENT.list_metrics())
-        after_names = set(metric.name for metric in after_metrics)
-        self.assertEqual(after_names - before_names,
-                         set([METRIC_NAME]))
+        after_names = set(after.name for after in after_metrics)
+        self.assertTrue(metric.name in after_names)
 
     def test_reload_metric(self):
         METRIC_NAME = 'test-reload-metric%s' % (_RESOURCE_ID,)
@@ -472,14 +473,14 @@ def test_list_sinks(self):
         sink = Config.CLIENT.sink(SINK_NAME, DEFAULT_FILTER, uri)
         self.assertFalse(sink.exists())
         before_sinks = list(Config.CLIENT.list_sinks())
-        before_names = set(sink.name for sink in before_sinks)
+        before_names = set(before.name for before in before_sinks)
+        self.failIf(sink.name in before_names)
         sink.create()
         self.to_delete.append(sink)
         self.assertTrue(sink.exists())
         after_sinks = list(Config.CLIENT.list_sinks())
-        after_names = set(sink.name for sink in after_sinks)
-        self.assertEqual(after_names - before_names,
-                         set([SINK_NAME]))
+        after_names = set(after.name for after in after_sinks)
+        self.assertTrue(sink.name in after_names)
 
     def test_reload_sink(self):
         SINK_NAME = 'test-reload-sink%s' % (_RESOURCE_ID,)