feat: Add PSC DNS and Global Write Endpoint support to Python Connector (#1446)

The Private Service Connect (PSC) DNS Name Resolution & Fallback feature enables the Python connector to securely connect to Cloud SQL database instances using Private Service Connect (PSC) DNS, custom domain names (CNAMEs), and Global Write Endpoints.

The connector can now dial PSC instances using:

- Its regional DNS hostname (e.g., 0123456789ab.fedcba9876543.us-central1.sql-psc.goog)
- A custom domain CNAME pointing to the instance PSC DNS hostname
- A Global Write Endpoint DNS hostname (e.g., 0123456789ab.fedcba9876543.global.sql-psc.goog)
- The connectors will dynamically resolve this hostname to its real Cloud SQL instance connection name via the SQL Admin API.

For Global Write Endpoints (which use the "global" region in their DNS name), the resolver bypasses direct API resolution (as "global" is not a valid API region) and forces fallback to CNAME resolution. This CNAME should point to the active regional instance DNS hostname, enabling automatic failover.

See GoogleCloudPlatform/cloud-sql-go-connector#1106

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_monitored_cache.py

@@ -238,3 +238,60 @@ async def test_MonitoredCache_purge_closed_sockets(
     # call _purge_closed_sockets and verify socket is removed
     monitored_cache._purge_closed_sockets()
     assert len(monitored_cache.sockets) == 0
+
+
+async def test_MonitoredCache_OptimizationForImmutableNames(
+    fake_client: CloudSQLClient,
+) -> None:
+    """
+    Test that MonitoredCache disables DNS polling for immutable instance DNS names
+    but enables it for custom DNS names and mutable global instance DNS names.
+    """
+    # 1. Custom DNS name -> should enable ticker
+    conn_name_custom = ConnectionName(
+        "test-project", "test-region", "test-instance", "db.example.com"
+    )
+    cache_custom = LazyRefreshCache(
+        conn_name_custom,
+        client=fake_client,
+        keys=asyncio.create_task(generate_keys()),
+        enable_iam_auth=False,
+    )
+    monitored_cache_custom = MonitoredCache(cache_custom, 30, DnsResolver())
+    assert monitored_cache_custom.domain_name_ticker is not None
+    await monitored_cache_custom.close()
+
+    # 2. Immutable instance DNS name -> should NOT enable ticker
+    conn_name_immutable = ConnectionName(
+        "test-project",
+        "test-region",
+        "test-instance",
+        "0123456789ab.fedcba9876543.us-central1.sql-psc.goog",
+    )
+    cache_immutable = LazyRefreshCache(
+        conn_name_immutable,
+        client=fake_client,
+        keys=asyncio.create_task(generate_keys()),
+        enable_iam_auth=False,
+    )
+    monitored_cache_immutable = MonitoredCache(cache_immutable, 30, DnsResolver())
+    assert monitored_cache_immutable.domain_name_ticker is None
+    await monitored_cache_immutable.close()
+
+    # 3. Mutable global instance DNS name -> should enable ticker
+    conn_name_global = ConnectionName(
+        "test-project",
+        "test-region",
+        "test-instance",
+        "0123456789ab.fedcba9876543.global.sql-psc.goog",
+    )
+    cache_global = LazyRefreshCache(
+        conn_name_global,
+        client=fake_client,
+        keys=asyncio.create_task(generate_keys()),
+        enable_iam_auth=False,
+    )
+    monitored_cache_global = MonitoredCache(cache_global, 30, DnsResolver())
+    assert monitored_cache_global.domain_name_ticker is not None
+    await monitored_cache_global.close()
+

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"