[batch] Make GCP disk attachment idempotent (hail-is#13955)

This is a fix for an error Ben found.

```
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/dist-packages/batch/worker/worker.py", line 1907, in run
    await self.setup_io()
  File "/usr/local/lib/python3.9/dist-packages/batch/worker/worker.py", line 1848, in setup_io
    await self.disk.create(labels=labels)
  File "/usr/local/lib/python3.9/dist-packages/batch/cloud/gcp/worker/disk.py", line 47, in create
    await self._attach()
  File "/usr/local/lib/python3.9/dist-packages/batch/cloud/gcp/worker/disk.py", line 112, in _attach
    self.last_response = await self.compute_client.attach_disk(
  File "/usr/local/lib/python3.9/dist-packages/hailtop/aiocloud/aiogoogle/client/compute_client.py", line 83, in attach_disk
    return await self._request_with_zonal_operations_response(self.post, path, params, **kwargs)
  File "/usr/local/lib/python3.9/dist-packages/hailtop/aiocloud/aiogoogle/client/compute_client.py", line 126, in _request_with_zonal_operations_response
    return await retry_transient_errors(request_and_wait)
  File "/usr/local/lib/python3.9/dist-packages/hailtop/utils/utils.py", line 763, in retry_transient_errors
    return await retry_transient_errors_with_debug_string('', 0, f, *args, **kwargs)
  File "/usr/local/lib/python3.9/dist-packages/hailtop/utils/utils.py", line 775, in retry_transient_errors_with_debug_string
    return await f(*args, **kwargs)
  File "/usr/local/lib/python3.9/dist-packages/hailtop/aiocloud/aiogoogle/client/compute_client.py", line 116, in request_and_wait
    raise GCPOperationError(result['httpErrorStatusCode'],
hailtop.aiocloud.aiogoogle.client.compute_client.GCPOperationError: GCPOperationError: 400:BAD REQUEST ['RESOURCE_IN_USE_BY_ANOTHER_RESOURCE'] ["The disk resource 'projects/hail-vdc/zones/us-central1-b/disks/batch-disk-82XXXXX' is already being used by 'projects/hail-vdc/zones/us-central1-b/instances/batch-worker-default-standard-yjXXXX'"]; {'kind': 'compute#operation', 'id': 'XXXXX', 'name': 'operation-XXXXX', 'zone': 'https://www.googleapis.com/compute/v1/projects/hail-vdc/zones/us-central1-b', 'clientOperationId': 'XXXX', 'operationType': 'attachDisk', 'targetLink': 'https://www.googleapis.com/compute/v1/projects/hail-vdc/zones/us-central1-b/instances/batch-worker-default-standard-yjupd', 'targetId': 'XXXX', 'status': 'DONE', 'user': '[email protected]', 'progress': 100, 'insertTime': '2023-10-30T20:38:40.145-07:00', 'startTime': '2023-10-30T20:38:41.871-07:00', 'endTime': '2023-10-30T20:38:42.367-07:00', 'error': {'errors': [{'code': 'RESOURCE_IN_USE_BY_ANOTHER_RESOURCE', 'message': "The disk resource 'projects/hail-vdc/zones/us-central1-b/disks/batch-disk-82XXXXX' is already being used by 'projects/hail-vdc/zones/us-central1-b/instances/batch-worker-default-standard-yjXXXX'"}]}, 'httpErrorStatusCode': 400, 'httpErrorMessage': 'BAD REQUEST', 'selfLink': 'https://www.googleapis.com/compute/v1/projects/hail-vdc/zones/us-central1-b/operations/operation-XXX'}
```

Author: jigold

batch/batch/cloud/gcp/worker/disk.py

@@ -101,6 +101,14 @@ async def _create(self, labels: Optional[Dict[str, str]] = None):
             self.last_response = await self.compute_client.create_disk(f'/zones/{self.zone}/disks', json=config)
             self._created = True
 
+    def _disk_is_already_attached_to_this_machine(self, e: aiogoogle.client.compute_client.GCPOperationError) -> bool:
+        if e.status == 400:
+            assert e.error_messages and e.error_codes
+            return all(self.instance_name in em for em in e.error_messages) and all(
+                em == 'RESOURCE_IN_USE_BY_ANOTHER_RESOURCE' for em in e.error_codes
+            )
+        return False
+
     async def _attach(self):
         async with LoggingTimer(f'attaching disk {self.name} to {self.instance_name}'):
             config = {
@@ -109,9 +117,14 @@ async def _attach(self):
                 'deviceName': self.name,
             }
 
-            self.last_response = await self.compute_client.attach_disk(
-                f'/zones/{self.zone}/instances/{self.instance_name}/attachDisk', json=config
-            )
+            try:
+                self.last_response = await self.compute_client.attach_disk(
+                    f'/zones/{self.zone}/instances/{self.instance_name}/attachDisk', json=config
+                )
+            except aiogoogle.client.compute_client.GCPOperationError as e:
+                if not self._disk_is_already_attached_to_this_machine(e):
+                    raise e
+
             self._attached = True
 
     async def _detach(self):