fix: Address code review comments on PSC DNS error handling

- Implement transient vs non-transient error detection in MonitoredCache.
- Detect immutable instance DNS names using the corrected regex.
- Avoid checking error message strings by using custom exception type (DnsLoopError).

Address review comments:
- kgala2: 'Should we have a try catch block of this sort to have a graceful shut in case of transient failures?'
- hessjcg: 'Use an error code. Don'\''t check error message strings.'
- hessjcg: 'The correct logic for detecting an immutable instance dns name: IF name matches /\.sql(-\w+)?\.goog$/ AND name does not match /\.global\.sql(-\w+)?\.goog$/'

Author: hessjcg

build.sh

@@ -144,17 +144,44 @@ function write_e2e_env(){
   done
   # Aliases for python e2e tests
   echo "export POSTGRES_CUSTOMER_CAS_PASS_VALID_DOMAIN_NAME=\"\$POSTGRES_CUSTOMER_CAS_DOMAIN_NAME\""
+  echo "export MYSQL_USER_IAM='$(iam_user_mysql)'"
+  echo "export POSTGRES_USER_IAM='$(iam_user_pg)'"
   echo "export POSTGRES_IAM_USER=\"\$POSTGRES_USER_IAM\""
   echo "export MYSQL_IAM_USER=\"\$MYSQL_USER_IAM\""
   } > "$1"
 
 }
 
-## with_venv - runs a command with the venv activated
 function with_venv() {
   "$@"
 }
 
+function iam_user_pg() {
+  local email
+  local pguser
+
+  email="$(iam_user_email)"
+  pguser="${email%%.gserviceaccount.com}"
+  if [[ -n "$pguser" ]] ; then
+    echo "$pguser"
+  else
+    echo "$email"
+  fi
+}
+
+function iam_user_mysql() {
+  local email
+  local mysqluser
+
+  email=$(iam_user_email)
+  mysqluser="${email%%@*}"
+  echo "$mysqluser"
+}
+
+function iam_user_email() {
+  gcloud auth list --format json | jq -r '.[] | select (.status == "ACTIVE") | .account'
+}
+
 ## help - prints the help details
 ##
 function help() {

google/cloud/sql/connector/exceptions.py

@@ -73,6 +73,12 @@ class DnsResolutionError(Exception):
     """
 
 
+class DnsLoopError(DnsResolutionError):
+    """
+    Exception to be raised when a CNAME loop is detected during DNS resolution.
+    """
+
+
 class CacheClosedError(Exception):
     """
     Exception to be raised when a ConnectionInfoCache can not be accessed after

google/cloud/sql/connector/monitored_cache.py

@@ -19,13 +19,17 @@
 import ssl
 from typing import Any, Callable
 
+import aiohttp
+
 from google.cloud.sql.connector.connection_info import ConnectionInfo
 from google.cloud.sql.connector.connection_info import ConnectionInfoCache
 from google.cloud.sql.connector.exceptions import CacheClosedError
+from google.cloud.sql.connector.exceptions import DnsLoopError
 from google.cloud.sql.connector.instance import RefreshAheadCache
 from google.cloud.sql.connector.lazy import LazyRefreshCache
 from google.cloud.sql.connector.resolver import DefaultResolver
 from google.cloud.sql.connector.resolver import DnsResolver
+from google.cloud.sql.connector.resolver import is_instance_dns_name
 
 logger = logging.getLogger(name=__name__)
 
@@ -44,7 +48,16 @@ def __init__(
 
         # If domain name is configured for instance and failover period is set,
         # poll for DNS record changes.
-        if self.cache.conn_name.domain_name and failover_period > 0:
+        is_instance_dns = False
+        if self.cache.conn_name.domain_name:
+            is_instance_dns = is_instance_dns_name(
+                self.cache.conn_name.domain_name or ""
+            )
+        if (
+            self.cache.conn_name.domain_name
+            and failover_period > 0
+            and not is_instance_dns
+        ):
             self.domain_name_ticker = asyncio.create_task(
                 ticker(failover_period, self._check_domain_name)
             )
@@ -90,12 +103,28 @@ async def _check_domain_name(self) -> None:
                 await self.close()
 
         except Exception as e:  # noqa: BLE001
-            # Domain name checks should not be fatal, log error and continue.
-            logger.debug(
-                f"['{self.cache.conn_name}']: Unable to check domain name, "
-                f"domain name {self.cache.conn_name.domain_name} did not "
-                f"resolve: {e}"
+            if self._is_non_transient(e):
+                logger.debug(
+                    f"['{self.cache.conn_name}']: Domain name check failed with "
+                    f"non-transient error, closing all connections: {e}"
+                )
+                await self.close()
+            else:
+                # Domain name checks should not be fatal for transient errors, log error and continue.
+                logger.debug(
+                    f"['{self.cache.conn_name}']: Unable to check domain name, "
+                    f"domain name {self.cache.conn_name.domain_name} did not "
+                    f"resolve (transient error): {e}"
+                )
+
+    def _is_non_transient(self, e: Exception) -> bool:
+        return (
+            isinstance(e, (ValueError, DnsLoopError))
+            or (
+                isinstance(e, aiohttp.ClientResponseError)
+                and e.status in (403, 404)
             )
+        )
 
     async def connect_info(self) -> ConnectionInfo:
         if self.closed:

google/cloud/sql/connector/resolver.py

@@ -25,12 +25,25 @@
     _parse_connection_name_with_domain_name,
 )
 from google.cloud.sql.connector.connection_name import ConnectionName
+from google.cloud.sql.connector.exceptions import DnsLoopError
 from google.cloud.sql.connector.exceptions import DnsResolutionError
 
-PSC_DNS_PATTERN = re.compile(
-    r"^([a-f0-9]{12})\.([^.]+)\.([a-z0-9]+-[a-z0-9]+)\.(sql|sql-psa|sql-psc)\.goog\.?$"
+# ^([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])
+
+INSTANCE_DNS_NAME_PATTERN = re.compile(
+    r"^([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])\.([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])\.([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])\.(sql|sql-psa|sql-psc)\.goog\.?$"
 )
 
+GLOBAL_INSTANCE_DNS_PATTERN = re.compile(r"\.global\.sql(-\w+)?\.goog\.?$")
+
+
+def is_instance_dns_name(name: str) -> bool:
+    lower = name.lower()
+    return bool(
+        INSTANCE_DNS_NAME_PATTERN.search(lower)
+        and not GLOBAL_INSTANCE_DNS_PATTERN.search(lower)
+    )
+
 
 class DefaultResolver:
     """DefaultResolver simply validates and parses instance connection name."""
@@ -68,7 +81,7 @@ async def resolve(self, dns: str) -> ConnectionName:  # type: ignore
                 pass
 
             dns_normalized = current.rstrip(".")
-            match = PSC_DNS_PATTERN.match(dns_normalized.lower())
+            match = INSTANCE_DNS_NAME_PATTERN.match(dns_normalized.lower())
             if match:
                 region = match.group(3)
                 if region != "global":
@@ -102,7 +115,7 @@ async def resolve(self, dns: str) -> ConnectionName:  # type: ignore
 
             if cname_found:
                 if cname in visited:
-                    raise DnsResolutionError(f"CNAME loop detected for `{dns}`")
+                    raise DnsLoopError(f"CNAME loop detected for `{dns}`")
                 visited.add(cname)
                 current = cname
                 continue
@@ -130,7 +143,7 @@ async def resolve(self, dns: str) -> ConnectionName:  # type: ignore
                 else f"Unable to resolve TXT record for `{current}`"
             )
 
-        raise DnsResolutionError(
+        raise DnsLoopError(
             f"CNAME loop detected or max resolution depth reached for `{dns}`"
         )
 

tests/unit/test_resolver.py

@@ -194,6 +194,29 @@ async def test_DnsResolver_with_direct_psc_dns_name() -> None:
     )
 
 
+async def test_DnsResolver_with_direct_psc_dns_name_permissive() -> None:
+    """Test DnsResolver resolves direct PSC DNS name with permissive formatting."""
+    dns_name = "g123.p.uscentral.sql-psc.goog"
+    real_conn_name = ConnectionName(
+        "my-project", "uscentral", "my-instance", dns_name
+    )
+
+    mock_client = AsyncMock()
+    mock_client.resolve_connect_settings.return_value = {
+        "connectionName": "my-project:uscentral:my-instance"
+    }
+
+    resolver = DnsResolver(client=mock_client)
+
+    result = await resolver.resolve(dns_name)
+
+    assert result == real_conn_name
+    mock_client.resolve_connect_settings.assert_awaited_once_with(
+        dns_name + ".", "uscentral"
+    )
+
+
+
 async def test_DnsResolver_with_cname_resolving_to_psc_dns_name() -> None:
     """Test DnsResolver resolves CNAME to PSC DNS and returns proper connection name."""
     dns_name = "db.example.com"