diff --git a/errors.py b/errors.py
index 27e7505..6dcd83e 100644
--- a/errors.py
+++ b/errors.py
@@ -143,6 +143,10 @@
     """Path does not exist."""
 
 
+class CheckGCEZonesQuotaError(CreateError):
+    """There is no zone have enough quota."""
+
+
 class UnsupportedInstanceImageType(CreateError):
     """Unsupported create action for given instance/image type."""
 
diff --git a/internal/lib/android_compute_client.py b/internal/lib/android_compute_client.py
index 1bc69bc..71baab1 100755
--- a/internal/lib/android_compute_client.py
+++ b/internal/lib/android_compute_client.py
@@ -82,6 +82,21 @@
         self._instance_name_pattern = acloud_config.instance_name_pattern
         self._AddPerInstanceSshkey()
 
+    # TODO(147047953): New args to contorl zone metrics check.
+    def _VerifyZoneByQuota(self):
+        """Verify the zone must have enough quota to create instance.
+
+        Returns:
+            Boolean, True if zone have enough quota to create instance.
+
+        Raises:
+            errors.CheckGCEZonesQuotaError: the zone doesn't have enough quota.
+        """
+        if self.EnoughMetricsInZone(self._zone):
+            return True
+        raise errors.CheckGCEZonesQuotaError(
+            "There is no enough quota in zone: %s" % self._zone)
+
     def _AddPerInstanceSshkey(self):
         """Add per-instance ssh key.
 
@@ -396,21 +411,3 @@
         """
         return super(AndroidComputeClient, self).GetSerialPortOutput(
             instance, zone or self._zone, port)
-
-    def GetInstanceNamesByIPs(self, ips, zone=None):
-        """Get Instance names by IPs.
-
-        This function will go through all instances, which
-        could be slow if there are too many instances.  However, currently
-        GCE doesn't support search for instance by IP.
-
-        Args:
-            ips: A set of IPs.
-            zone: String, representing zone name, e.g. "us-central1-f"
-
-        Returns:
-            A dictionary where key is ip and value is instance name or None
-            if instance is not found for the given IP.
-        """
-        return super(AndroidComputeClient, self).GetInstanceNamesByIPs(
-            ips, zone or self._zone)
diff --git a/internal/lib/cvd_compute_client_multi_stage.py b/internal/lib/cvd_compute_client_multi_stage.py
index 2c1821c..05bcef9 100644
--- a/internal/lib/cvd_compute_client_multi_stage.py
+++ b/internal/lib/cvd_compute_client_multi_stage.py
@@ -188,6 +188,7 @@
         if avd_spec and avd_spec.instance_name_to_reuse:
             self._ip = self._ReusingGceInstance(avd_spec)
         else:
+            self._VerifyZoneByQuota()
             self._ip = self._CreateGceInstance(instance, image_name, image_project,
                                                extra_scopes, boot_disk_size_gb,
                                                avd_spec)
diff --git a/internal/lib/cvd_compute_client_multi_stage_test.py b/internal/lib/cvd_compute_client_multi_stage_test.py
index c0fed24..2ee543a 100644
--- a/internal/lib/cvd_compute_client_multi_stage_test.py
+++ b/internal/lib/cvd_compute_client_multi_stage_test.py
@@ -83,6 +83,8 @@
         """Set up the test."""
         super(CvdComputeClientTest, self).setUp()
         self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "InitResourceHandle")
+        self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "_VerifyZoneByQuota",
+                   return_value=True)
         self.Patch(android_build_client.AndroidBuildClient, "InitResourceHandle")
         self.Patch(android_build_client.AndroidBuildClient, "DownloadArtifact")
         self.Patch(list_instances, "GetInstancesFromInstanceNames", return_value=mock.MagicMock())
diff --git a/internal/lib/gcompute_client.py b/internal/lib/gcompute_client.py
index 3c0b295..2a1d414 100755
--- a/internal/lib/gcompute_client.py
+++ b/internal/lib/gcompute_client.py
@@ -51,6 +51,14 @@
 _ITEMS = "items"
 _METADATA = "metadata"
 _ZONE_RE = re.compile(r"^zones/(?P<zone>.+)")
+# Quota metrics
+_METRIC_CPUS = "CPUS"
+_METRIC_DISKS_GB = "DISKS_TOTAL_GB"
+_METRICS = [_METRIC_CPUS, _METRIC_DISKS_GB]
+_USAGE = "usage"
+_LIMIT = "limit"
+# The minimum requirement to create an instance.
+_REQUIRE_METRICS = {_METRIC_CPUS: 8, _METRIC_DISKS_GB: 1000}
 
 BASE_DISK_ARGS = {
     "type": "PERSISTENT",
@@ -196,11 +204,75 @@
         """Get project information.
 
         Returns:
-          A project resource in json.
+            A project resource in json.
         """
         api = self.service.projects().get(project=self._project)
         return self.Execute(api)
 
+    def GetRegionInfo(self):
+        """Get region information that includes all quotas limit.
+
+        The region info example:
+        {"items":
+            [{"status": "UP",
+              "name": "asia-east1",
+              "quotas":
+                [{"usage": 92, "metric": "CPUS", "limit": 100},
+                 {"usage": 640, "metric": "DISKS_TOTAL_GB", "limit": 10240},
+              ...]]}
+        }
+
+        Returns:
+            A region resource in json.
+        """
+        api = self.service.regions().list(project=self._project)
+        return self.Execute(api)
+
+    @staticmethod
+    def GetMetricQuota(regions_info, zone, metric):
+        """Get CPU quota limit in specific zone and project.
+
+        Args:
+            regions_info: Dict, regions resource in json.
+            zone: String, name of zone.
+            metric: String, name of metric, e.g. "CPUS".
+
+        Returns:
+            A dict of quota information. Such as
+            {"usage": 100, "metric": "CPUS", "limit": 200}
+        """
+        for region_info in regions_info["items"]:
+            if region_info["name"] in zone:
+                for quota in region_info["quotas"]:
+                    if quota["metric"] == metric:
+                        return quota
+        logger.info("Can't get %s quota info from zone(%s)", metric, zone)
+        return None
+
+    def EnoughMetricsInZone(self, zone):
+        """Check the zone have enough metrics to create instance.
+
+        The metrics include CPUS and DISKS.
+
+        Args:
+            zone: String, name of zone.
+
+        Returns:
+            Boolean. True if zone have enough quota.
+        """
+        regions_info = self.GetRegionInfo()
+        for metric in _METRICS:
+            quota = self.GetMetricQuota(regions_info, zone, metric)
+            if not quota:
+                logger.debug(
+                    "Can't query the metric(%s) in zone(%s)", metric, zone)
+                return False
+            if quota[_LIMIT] - quota[_USAGE] < _REQUIRE_METRICS[metric]:
+                logger.debug(
+                    "The metric(%s) is over limit in zone(%s)", metric, zone)
+                return False
+        return True
+
     def GetDisk(self, disk_name, zone):
         """Get disk information.