feat: Periodicaly check if domain name changed and close connections to old database.

wip: periodic check for domain change

wip: periodic checks

wip: close sockets on instance closed

wip: readme for below

Author: hessjcg

.github/workflows/tests.yml

@@ -154,7 +154,7 @@ jobs:
             MYSQL_DB:${{ vars.GOOGLE_CLOUD_PROJECT }}/MYSQL_DB
             POSTGRES_CONNECTION_NAME:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_CONNECTION_NAME
             POSTGRES_USER:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_USER
-            POSTGRES_USER_IAM_NODE:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_USER_IAM_NODE
+            POSTGRES_IAM_USER:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_USER_IAM_NODE
             POSTGRES_PASS:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_PASS
             POSTGRES_DB:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_DB
             POSTGRES_CAS_CONNECTION_NAME:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_CAS_CONNECTION_NAME
@@ -183,7 +183,7 @@ jobs:
           MYSQL_DB: "${{ steps.secrets.outputs.MYSQL_DB }}"
           POSTGRES_CONNECTION_NAME: "${{ steps.secrets.outputs.POSTGRES_CONNECTION_NAME }}"
           POSTGRES_USER: "${{ steps.secrets.outputs.POSTGRES_USER }}"
-          POSTGRES_USER_IAM_NODE: "${{ steps.secrets.outputs.POSTGRES_USER_IAM_NODE }}"
+          POSTGRES_IAM_USER: "${{ steps.secrets.outputs.POSTGRES_IAM_USER }}"
           POSTGRES_PASS: "${{ steps.secrets.outputs.POSTGRES_PASS }}"
           POSTGRES_DB: "${{ steps.secrets.outputs.POSTGRES_DB }}"
           POSTGRES_CAS_CONNECTION_NAME: "${{ steps.secrets.outputs.POSTGRES_CAS_CONNECTION_NAME }}"
@@ -279,7 +279,7 @@ jobs:
             MYSQL_DB:${{ vars.GOOGLE_CLOUD_PROJECT }}/MYSQL_DB
             POSTGRES_CONNECTION_NAME:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_CONNECTION_NAME
             POSTGRES_USER:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_USER
-            POSTGRES_USER_IAM_NODE:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_USER_IAM_NODE
+            POSTGRES_IAM_USER:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_USER_IAM_NODE
             POSTGRES_PASS:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_PASS
             POSTGRES_DB:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_DB
             POSTGRES_CAS_CONNECTION_NAME:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_CAS_CONNECTION_NAME
@@ -302,7 +302,7 @@ jobs:
           MYSQL_DB: "${{ steps.secrets.outputs.MYSQL_DB }}"
           POSTGRES_CONNECTION_NAME: "${{ steps.secrets.outputs.POSTGRES_CONNECTION_NAME }}"
           POSTGRES_USER: "${{ steps.secrets.outputs.POSTGRES_USER }}"
-          POSTGRES_IAM_USER: "${{ steps.secrets.outputs.POSTGRES_USER_IAM_NODE }}"
+          POSTGRES_IAM_USER: "${{ steps.secrets.outputs.POSTGRES_IAM_USER }}"
           POSTGRES_PASS: "${{ steps.secrets.outputs.POSTGRES_PASS }}"
           POSTGRES_DB: "${{ steps.secrets.outputs.POSTGRES_DB }}"
           POSTGRES_CAS_CONNECTION_NAME: "${{ steps.secrets.outputs.POSTGRES_CAS_CONNECTION_NAME }}"

README.md

@@ -444,6 +444,88 @@ variables. Here is a quick reference to supported values and their effect:
 - `GOOGLE_CLOUD_QUOTA_PROJECT`: Used to set a custom quota project to Cloud SQL
   APIs when defined.
 
+## Using DNS domain names to identify instances
+
+The connector can be configured to use DNS to look up an instance. This would
+allow you to configure your application to connect to a database instance, and
+centrally configure which instance in your DNS zone.
+
+### Configure your DNS Records
+
+Add a DNS TXT record for the Cloud SQL instance to a **private** DNS server
+or a private Google Cloud DNS Zone used by your application.
+
+**Note:** You are strongly discouraged from adding DNS records for your
+Cloud SQL instances to a public DNS server. This would allow anyone on the
+internet to discover the Cloud SQL instance name.
+
+For example: suppose you wanted to use the domain name
+`prod-db.mycompany.example.com` to connect to your database instance
+`my-project:region:my-instance`. You would create the following DNS record:
+
+- Record type: `TXT`
+- Name: `prod-db.mycompany.example.com` – This is the domain name used by the application
+- Value: `my-project:region:my-instance` – This is the instance name
+
+### Configure the connector
+
+Configure the connector as described above, replacing the connector ID with
+the DNS name.
+
+Adapting the MySQL + database/sql example above:
+
+```js
+import mysql from 'mysql2/promise';
+import {Connector} from '@google-cloud/cloud-sql-connector';
+
+const connector = new Connector();
+const clientOpts = await connector.getOptions({
+  domainName: 'prod-db.mycompany.example.com',
+  ipType: 'PUBLIC',
+});
+
+const pool = await mysql.createPool({
+  ...clientOpts,
+  user: 'my-user',
+  password: 'my-password',
+  database: 'db-name',
+});
+const conn = await pool.getConnection();
+const [result] = await conn.query(`SELECT NOW();`);
+console.table(result); // prints returned time value from server
+
+await pool.end();
+connector.close();
+```
+
+## Automatic failover using DNS domain names
+
+For example: suppose application is configured to connect using the
+domain name `prod-db.mycompany.example.com`. Initially the private DNS
+zone has a TXT record with the value `my-project:region:my-instance`. The
+application establishes connections to the `my-project:region:my-instance`
+Cloud SQL instance. Configure the connector using the `domainName` option:
+
+Then, to reconfigure the application to use a different database
+instance, change the value of the `prod-db.mycompany.example.com` DNS record
+from `my-project:region:my-instance` to `my-project:other-region:my-instance-2`
+
+The connector inside the application detects the change to this
+DNS record. Now, when the application connects to its database using the
+domain name `prod-db.mycompany.example.com`, it will connect to the
+`my-project:other-region:my-instance-2` Cloud SQL instance.
+
+The connector will automatically close all existing connections to
+`my-project:region:my-instance`. This will force the connection pools to
+establish new connections. Also, it may cause database queries in progress
+to fail.
+
+The connector will poll for changes to the DNS name every 30 seconds by default.
+You may configure the frequency of the connections using the Connector's
+`failoverPeriod` option. When this is set to 0, the connector will disable
+polling and only check if the DNS record changed when it is creating a new 
+connection.
+
 ## Support policy
 
 ### Major version lifecycle

src/cloud-sql-instance.ts

@@ -14,13 +14,26 @@
 
 import {IpAddressTypes, selectIpAddress} from './ip-addresses';
 import {InstanceConnectionInfo} from './instance-connection-info';
-import {resolveInstanceName} from './parse-instance-connection-name';
+import {
+  isSameInstance,
+  resolveInstanceName,
+} from './parse-instance-connection-name';
 import {InstanceMetadata} from './sqladmin-fetcher';
 import {generateKeys} from './crypto';
 import {RSAKeys} from './rsa-keys';
 import {SslCert} from './ssl-cert';
 import {getRefreshInterval, isExpirationTimeValid} from './time';
 import {AuthTypes} from './auth-types';
+import {CloudSQLConnectorError} from './errors';
+
+// Private types that describe exactly the methods
+// needed from tls.Socket to be able to close
+// sockets when the DNS Name changes.
+type EventFn = () => void;
+type DestroyableSocket = {
+  destroy: (error?: Error) => void;
+  once: (name: string, handler: EventFn) => void;
+};
 
 interface Fetcher {
   getInstanceMetadata({
@@ -42,6 +55,7 @@ interface CloudSQLInstanceOptions {
   ipType: IpAddressTypes;
   limitRateInterval?: number;
   sqlAdminFetcher: Fetcher;
+  failoverPeriod?: number;
 }
 
 interface RefreshResult {
@@ -74,9 +88,13 @@ export class CloudSQLInstance {
   // The ongoing refresh promise is referenced by the `next` property
   private next?: Promise<RefreshResult>;
   private scheduledRefreshID?: ReturnType<typeof setTimeout> | null = undefined;
+  private checkDomainID?: ReturnType<typeof setInterval> | null = undefined;
   /* eslint-disable-next-line @typescript-eslint/no-explicit-any */
   private throttle?: any;
   private closed = false;
+  private failoverPeriod: number;
+  private sockets = new Set<DestroyableSocket>();
+
   public readonly instanceInfo: InstanceConnectionInfo;
   public ephemeralCert?: SslCert;
   public host?: string;
@@ -98,6 +116,7 @@ export class CloudSQLInstance {
     this.ipType = options.ipType || IpAddressTypes.PUBLIC;
     this.limitRateInterval = options.limitRateInterval || 30 * 1000; // 30 seconds
     this.sqlAdminFetcher = options.sqlAdminFetcher;
+    this.failoverPeriod = options.failoverPeriod || 30 * 1000; // 30 seconds
   }
 
   // p-throttle library has to be initialized in an async scope in order to
@@ -153,6 +172,19 @@ export class CloudSQLInstance {
       return Promise.reject('closed');
     }
 
+    // Lazy instantiation of the checkDomain interval on the first refresh
+    // This avoids issues with test cases that instantiate a CloudSqlInstance.
+    // If failoverPeriod is 0 (or negative) don't check for DNS updates.
+    if (
+      this?.instanceInfo?.domainName &&
+      !this.checkDomainID &&
+      this.failoverPeriod > 0
+    ) {
+      this.checkDomainID = setInterval(() => {
+        this.checkDomainChanged();
+      }, this.failoverPeriod);
+    }
+
     const currentRefreshId = this.scheduledRefreshID;
 
     // Since forceRefresh might be invoked during an ongoing refresh
@@ -312,9 +344,48 @@ export class CloudSQLInstance {
   close(): void {
     this.closed = true;
     this.cancelRefresh();
+    if (this.checkDomainID) {
+      clearInterval(this.checkDomainID);
+      this.checkDomainID = null;
+    }
+    for (const socket of this.sockets) {
+      socket.destroy(
+        new CloudSQLConnectorError({
+          code: 'ERRCLOSED',
+          message: 'The connector was closed.',
+        })
+      );
+    }
   }
 
   isClosed(): boolean {
     return this.closed;
   }
+  async checkDomainChanged() {
+    if (!this.instanceInfo.domainName) {
+      return;
+    }
+
+    const newInfo = await resolveInstanceName(
+      undefined,
+      this.instanceInfo.domainName
+    );
+    if (!isSameInstance(this.instanceInfo, newInfo)) {
+      // Domain name changed. Close and remove, then create a new map entry.
+      this.close();
+    }
+  }
+  addSocket(socket: DestroyableSocket) {
+    if (!this.instanceInfo.domainName) {
+      // This was not connected by domain name. Ignore all sockets.
+      return;
+    }
+
+    // Add the socket to the list
+    this.sockets.add(socket);
+    // When the socket is closed, remove it.
+    socket.once('closed', () => {
+      this.sockets.delete(socket);
+    });
+  }
 }

src/connector.ts

@@ -44,6 +44,7 @@ export declare interface ConnectionOptions {
   ipType?: IpAddressTypes;
   instanceConnectionName: string;
   domainName?: string;
+  failoverPeriod?: number;
   limitRateInterval?: number;
 }
 
@@ -129,6 +130,7 @@ class CloudSQLInstanceMap extends Map<string, CacheEntry> {
     const entry = this.get(key);
     if (entry) {
       if (entry.isResolved()) {
+        await entry.instance?.checkDomainChanged();
         if (!entry.instance?.isClosed()) {
           // The instance is open and the domain has not changed.
           // use the cached instance.
@@ -154,6 +156,7 @@ class CloudSQLInstanceMap extends Map<string, CacheEntry> {
       ipType: opts.ipType || IpAddressTypes.PUBLIC,
       limitRateInterval: opts.limitRateInterval || 30 * 1000, // 30 sec
       sqlAdminFetcher: this.sqlAdminFetcher,
+      failoverPeriod: opts.failoverPeriod,
     });
     this.set(key, new CacheEntry(promise));
 
@@ -257,6 +260,9 @@ export class Connector {
           tlsSocket.once('secureConnect', async () => {
             cloudSqlInstance.setEstablishedConnection();
           });
+
+          cloudSqlInstance.addSocket(tlsSocket);
+
           return tlsSocket;
         }
 

system-test/pg-connect.cjs

@@ -20,13 +20,13 @@ const {Client} = pg;
 t.test('open connection and retrieves standard pg tables', async t => {
   const connector = new Connector();
   const clientOpts = await connector.getOptions({
-    instanceConnectionName: String(process.env.POSTGRES_CONNECTION_NAME),
+    instanceConnectionName: process.env.POSTGRES_CONNECTION_NAME,
   });
   const client = new Client({
     ...clientOpts,
-    user: String(process.env.POSTGRES_USER),
-    password: String(process.env.POSTGRES_PASS),
-    database: String(process.env.POSTGRES_DB),
+    user: process.env.POSTGRES_USER,
+    password: process.env.POSTGRES_PASS,
+    database: process.env.POSTGRES_DB,
   });
   t.after(async () => {
     try {
@@ -48,14 +48,14 @@ t.test('open connection and retrieves standard pg tables', async t => {
 t.test('open IAM connection and retrieves standard pg tables', async t => {
   const connector = new Connector();
   const clientOpts = await connector.getOptions({
-    instanceConnectionName: String(process.env.POSTGRES_CONNECTION_NAME),
-    ipType: "PUBLIC",
-    authType: "IAM",
+    instanceConnectionName: process.env.POSTGRES_CONNECTION_NAME,
+    ipType: 'PUBLIC',
+    authType: 'IAM',
   });
   const client = new Client({
     ...clientOpts,
-    user: String(process.env.POSTGRES_USER_IAM_NODE),
-    database: String(process.env.POSTGRES_DB),
+    user: process.env.POSTGRES_IAM_USER,
+    database: process.env.POSTGRES_DB,
   });
   t.after(async () => {
     try {
@@ -82,9 +82,9 @@ t.test(
     });
     const client = new Client({
       ...clientOpts,
-      user: String(process.env.POSTGRES_USER),
-      password: String(process.env.POSTGRES_CAS_PASS),
-      database: String(process.env.POSTGRES_DB),
+      user: process.env.POSTGRES_USER,
+      password: process.env.POSTGRES_CAS_PASS,
+      database: process.env.POSTGRES_DB,
     });
     t.after(async () => {
       try {

system-test/pg-connect.mjs

@@ -50,12 +50,12 @@ t.test('open IAM connection and retrieves standard pg tables', async t => {
   const connector = new Connector();
   const clientOpts = await connector.getOptions({
     instanceConnectionName: String(process.env.POSTGRES_CONNECTION_NAME),
-    ipType: "PUBLIC",
-    authType: "IAM",
+    ipType: 'PUBLIC',
+    authType: 'IAM',
   });
   const client = new Client({
     ...clientOpts,
-    user: String(process.env.POSTGRES_USER_IAM_NODE),
+    user: String(process.env.POSTGRES_IAM_USER),
     database: String(process.env.POSTGRES_DB),
   });
   t.after(async () => {

system-test/pg-connect.ts

@@ -58,7 +58,7 @@ t.test('open IAM connection and retrieves standard pg tables', async t => {
   });
   const client = new Client({
     ...clientOpts,
-    user: String(process.env.POSTGRES_USER_IAM_NODE),
+    user: String(process.env.POSTGRES_IAM_USER),
     database: String(process.env.POSTGRES_DB),
   });
   t.after(async () => {