'test_instance_create_w_two_clusters' systest flakes with 503

[tseaver]

From this failed systest build:

_____________________ test_instance_create_w_two_clusters ______________________

args = (parent: "projects/precise-truck-742/instances/dif-1634667466967"
table_id: "test-get-cluster-states"
table {
}
,)
kwargs = {'metadata': [('x-goog-request-params', 'parent=projects/precise-truck-742/instances/dif-1634667466967'), ('x-goog-api-client', 'gl-python/3.8.12 grpc/1.41.0 gax/2.1.1')]}

 @functools.wraps(callable_)
 def error_remapped_callable(*args, **kwargs):
 try:
> return callable_(*args, **kwargs)

.nox/system-3-8/lib/python3.8/site-packages/google/api_core/grpc_helpers.py:66:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <grpc._channel._UnaryUnaryMultiCallable object at 0x7f4d5eecd100>
request = parent: "projects/precise-truck-742/instances/dif-1634667466967"
table_id: "test-get-cluster-states"
table {
}

timeout = None
metadata = [('x-goog-request-params', 'parent=projects/precise-truck-742/instances/dif-1634667466967'), ('x-goog-api-client', 'gl-python/3.8.12 grpc/1.41.0 gax/2.1.1')]
credentials = None, wait_for_ready = None, compression = None

 def __call__(self,
 request,
 timeout=None,
 metadata=None,
 credentials=None,
 wait_for_ready=None,
 compression=None):
 state, call, = self._blocking(request, timeout, metadata, credentials,
 wait_for_ready, compression)
> return _end_unary_response_blocking(state, call, False, None)

.nox/system-3-8/lib/python3.8/site-packages/grpc/_channel.py:946:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

state = <grpc._channel._RPCState object at 0x7f4d5ee8b610>
call = <grpc._cython.cygrpc.SegregatedCall object at 0x7f4d5ed214c0>
with_call = False, deadline = None

 def _end_unary_response_blocking(state, call, with_call, deadline):
 if state.code is grpc.StatusCode.OK:
 if with_call:
 rendezvous = _MultiThreadedRendezvous(state, call, None, deadline)
 return state.response, rendezvous
 else:
 return state.response
 else:
> raise _InactiveRpcError(state)
E grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
E 	status = StatusCode.UNAVAILABLE
E 	details = "The service is currently unavailable."
E 	debug_error_string = "{"created":"@1634667658.122902172","description":"Error received from peer ipv4:74.125.197.95:443","file":"src/core/lib/surface/call.cc","file_line":1069,"grpc_message":"The service is currently unavailable.","grpc_status":14}"
E >

.nox/system-3-8/lib/python3.8/site-packages/grpc/_channel.py:849: _InactiveRpcError

The above exception was the direct cause of the following exception:

admin_client = <google.cloud.bigtable.client.Client object at 0x7f4d5ef80e50>
unique_suffix = '-1634667466967'
admin_instance_populated = <google.cloud.bigtable.instance.Instance object at 0x7f4d5eef10a0>
admin_cluster = <google.cloud.bigtable.cluster.Cluster object at 0x7f4d5eef1160>
location_id = 'us-central1-c'
instance_labels = {'python-system': '2021-10-19t18-17-46'}
instances_to_delete = [<google.cloud.bigtable.instance.Instance object at 0x7f4d5ee5cc70>]
skip_on_emulator = None

 def test_instance_create_w_two_clusters(
 admin_client,
 unique_suffix,
 admin_instance_populated,
 admin_cluster,
 location_id,
 instance_labels,
 instances_to_delete,
 skip_on_emulator,
 ):
 alt_instance_id = f"dif{unique_suffix}"
 instance = admin_client.instance(
 alt_instance_id,
 instance_type=enums.Instance.Type.PRODUCTION,
 labels=instance_labels,
 )

 serve_nodes = 1

 alt_cluster_id_1 = f"{alt_instance_id}-c1"
 cluster_1 = instance.cluster(
 alt_cluster_id_1,
 location_id=location_id,
 serve_nodes=serve_nodes,
 default_storage_type=enums.StorageType.HDD,
 )

 alt_cluster_id_2 = f"{alt_instance_id}-c2"
 location_id_2 = "us-central1-f"
 cluster_2 = instance.cluster(
 alt_cluster_id_2,
 location_id=location_id_2,
 serve_nodes=serve_nodes,
 default_storage_type=enums.StorageType.HDD,
 )
 operation = instance.create(clusters=[cluster_1, cluster_2])
 instances_to_delete.append(instance)
 operation.result(timeout=120) # Ensure the operation completes.

 # Create a new instance instance and make sure it is the same.
 instance_alt = admin_client.instance(alt_instance_id)
 instance_alt.reload()

 assert instance == instance_alt
 assert instance.display_name == instance_alt.display_name
 assert instance.type_ == instance_alt.type_

 clusters, failed_locations = instance_alt.list_clusters()
 assert failed_locations == []

 alt_cluster_1, alt_cluster_2 = sorted(clusters, key=lambda x: x.name)

 assert cluster_1.location_id == alt_cluster_1.location_id
 assert alt_cluster_1.state == enums.Cluster.State.READY
 assert cluster_1.serve_nodes == alt_cluster_1.serve_nodes
 assert cluster_1.default_storage_type == alt_cluster_1.default_storage_type
 assert cluster_2.location_id == alt_cluster_2.location_id
 assert alt_cluster_2.state == enums.Cluster.State.READY
 assert cluster_2.serve_nodes == alt_cluster_2.serve_nodes
 assert cluster_2.default_storage_type == alt_cluster_2.default_storage_type

 # Test list clusters in project via 'client.list_clusters'
 clusters, failed_locations = admin_client.list_clusters()
 assert not failed_locations
 found = set([cluster.name for cluster in clusters])
 expected = {alt_cluster_1.name, alt_cluster_2.name, admin_cluster.name}
 assert expected.issubset(found)

 temp_table_id = "test-get-cluster-states"
 temp_table = instance.table(temp_table_id)
> _helpers.retry_grpc_unavailable(temp_table.create)()

tests/system/test_instance_admin.py:272:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.nox/system-3-8/lib/python3.8/site-packages/test_utils/retry.py:100: in wrapped_function
 return to_wrap(*args, **kwargs)
google/cloud/bigtable/table.py:402: in create
 table_client.create_table(
google/cloud/bigtable_admin_v2/services/bigtable_table_admin/client.py:543: in create_table
 response = rpc(request, retry=retry, timeout=timeout, metadata=metadata,)
.nox/system-3-8/lib/python3.8/site-packages/google/api_core/gapic_v1/method.py:142: in __call__
 return wrapped_func(*args, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

args = (parent: "projects/precise-truck-742/instances/dif-1634667466967"
table_id: "test-get-cluster-states"
table {
}
,)
kwargs = {'metadata': [('x-goog-request-params', 'parent=projects/precise-truck-742/instances/dif-1634667466967'), ('x-goog-api-client', 'gl-python/3.8.12 grpc/1.41.0 gax/2.1.1')]}

 @functools.wraps(callable_)
 def error_remapped_callable(*args, **kwargs):
 try:
 return callable_(*args, **kwargs)
 except grpc.RpcError as exc:
> raise exceptions.from_grpc_error(exc) from exc
E google.api_core.exceptions.ServiceUnavailable: 503 The service is currently unavailable.

.nox/system-3-8/lib/python3.8/site-packages/google/api_core/grpc_helpers.py:68: ServiceUnavailable