Handle GCP Cloud Redis resource exhaustion.

perfkitbenchmarker/providers/gcp/gcp_cloud_redis.py

@@ -186,7 +186,14 @@ def _Create(self):
     if self.clustered:
       cmd = self._GetClusterCreateCommand()
     cmd.flags['region'] = self.redis_region
-    cmd.Issue(timeout=COMMAND_TIMEOUT)
+    _, stderr, retcode = cmd.Issue(
+        timeout=COMMAND_TIMEOUT, raise_on_failure=False
+    )
+    if retcode:
+      util.CheckGcloudResponseKnownFailures(stderr, retcode)
+      raise errors.VmUtil.IssueCommandError(
+          f'Required gcloud command failed: {cmd}\n{stderr}'
+      )
 
   def _IsReady(self):
     """Returns whether cluster is ready."""

perfkitbenchmarker/providers/gcp/util.py

@@ -467,6 +467,9 @@ def _AddCommonFlags(self, common_resource: resource.BaseResource | None):
     # Not actually regex so escape as best practice
     re.compile(re.escape('ZONE_RESOURCE_POOL_EXHAUSTED')),
     re.compile(re.escape('ERROR_STOCKOUT')),
+    re.compile(
+        r'Not enough zonal resources are available to fulfill the request'
+    ),
 )
 _NOT_ENOUGH_RESOURCES_MESSAGE = 'Creation failed due to not enough resources: '
 

tests/providers/gcp/gcp_cloud_redis_test.py

@@ -11,14 +11,15 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-"""Tests for perfkitbenchmarker.providers.gcp.gcp_cloud_redis."""
+
 
 import inspect
 import unittest
 
 from absl import flags
 from absl.testing import flagsaver
 import mock
+from perfkitbenchmarker import errors
 from perfkitbenchmarker import managed_memory_store
 from perfkitbenchmarker.providers.gcp import gcp_cloud_redis
 from perfkitbenchmarker.providers.gcp import util
@@ -36,6 +37,7 @@ def setUp(self):
     ):
       FLAGS.project = 'project'
       FLAGS.zone = ['us-central1-a']
+      FLAGS.cloud_redis_region = 'us-central1'
       mock_spec = mock.Mock()
       self.redis = gcp_cloud_redis.CloudRedis(mock_spec)
 
@@ -47,6 +49,20 @@ def testCreate(self):
       gcloud.assert_called_once()
       self.assertTrue(self.redis._Exists())
 
+  def testCreateResourceExhausted(self):
+    stderr = (
+        'ERROR: (gcloud.redis.instances.create) '
+        'Not enough zonal resources are available to fulfill the request. '
+        'Try again later.'
+    )
+    with mock.patch.object(
+        util.GcloudCommand, 'Issue', return_value=('', stderr, 1)
+    ):
+      with self.assertRaises(
+          errors.Benchmarks.InsufficientCapacityCloudFailure
+      ):
+        self.redis._Create()
+
   def testDelete(self):
     with mock.patch.object(
         util.GcloudCommand, 'Issue', return_value=('{}', '', 0)