使用Lambda Layer连接AWS Keyspaces时出现间歇性连接错误

错误现象 #

• 常规请求中约10-20%出现连接失败：

All host(s) tried for query failed. First host tried, 3.248.244.53:9142: Host considered as DOWN. See innerErrors.

• 并发调用（5个并发）时，偶尔出现Socket关闭错误（约50次调用出现1次）：

"All host(s) tried for query failed. First host tried,
3.248.244.5:9142: DriverError: Socket was closed at Connection.clearAndInvokePending
(/opt/node_modules/cassandra-driver/lib/connection.js:265:15) at
Connection.close
(/opt/node_modules/cassandra-driver/lib/connection.js:618:8) at
TLSSocket.
(/opt/node_modules/cassandra-driver/lib/connection.js:93:10) at
TLSSocket.emit (node:events:525:35)\n at node:net:313:12\n at
TCP.done (node:_tls_wrap:587:7) { info: 'Cassandra Driver Error',
isSocketError: true, coordinator: '3.248.244.5:9142'}"

当前代码实现 #

import cassandra from 'cassandra-driver';
import fs from 'fs';

export default class AmazonKeyspace {

  tpmsClient = null;

  constructor () {
    let auth = new cassandra.auth.PlainTextAuthProvider('cass-user-at-xxxxxxxxxx', 'zzzzzzzzz');
    let sslOptions1 = {
      ca: [ fs.readFileSync('/opt/utils/AmazonRootCA1.pem', 'utf-8')],
      host: 'cassandra.eu-west-1.amazonaws.com',
      rejectUnauthorized: true
    };
    this.tpmsClient = new cassandra.Client({
      contactPoints: ['cassandra.eu-west-1.amazonaws.com'],
      localDataCenter: 'eu-west-1',
      authProvider: auth,
      sslOptions: sslOptions1,
      keyspace: 'tpms',
      protocolOptions: { port: 9142 }
    });
  }

  getOrganisation = async (orgKey) => {
    const SQL = 'select * FROM organisation where organisation_id=?;';

    return new Promise((resolve, reject) => {
      this.tpmsClient.execute(SQL, [orgKey], {prepare: true}, (err, result) => {
        if (!err?.message) resolve(result.rows);
        else reject(err.message);
      });
    });
  };
}

排查背景 #

已严格遵循AWS官方文档配置，熟悉Cassandra集群管理，AWS控制台无相关错误，怀疑是超时或contact points配置问题。

1. 扩展Contact Points配置 #

AWS Keyspaces的单一域名会解析到多个IP节点，仅配置单个域名可能导致部分IP不可用时请求失败。通过动态解析域名获取多个IP作为contact points，提升连接容错性：

import cassandra from 'cassandra-driver';
import fs from 'fs';
const dns = require('dns').promises;

async function getContactPoints() {
  const records = await dns.resolve4('cassandra.eu-west-1.amazonaws.com');
  return records;
}

export default class AmazonKeyspace {
  static async createInstance() {
    const contactPoints = await getContactPoints();
    let auth = new cassandra.auth.PlainTextAuthProvider('cass-user-at-xxxxxxxxxx', 'zzzzzzzzz');
    let sslOptions1 = {
      ca: [ fs.readFileSync('/opt/utils/AmazonRootCA1.pem', 'utf-8')],
      host: 'cassandra.eu-west-1.amazonaws.com',
      rejectUnauthorized: true
    };
    const client = new cassandra.Client({
      contactPoints: contactPoints,
      localDataCenter: 'eu-west-1',
      authProvider: auth,
      sslOptions: sslOptions1,
      keyspace: 'tpms',
      protocolOptions: { port: 9142 }
    });
    await client.connect();
    const instance = new AmazonKeyspace();
    instance.tpmsClient = client;
    return instance;
  }

  getOrganisation = async (orgKey) => {
    const SQL = 'select * FROM organisation where organisation_id=?;';
    const result = await this.tpmsClient.execute(SQL, [orgKey], {prepare: true});
    return result.rows;
  };
}

2. 调整超时与重试策略 #

Lambda短生命周期环境下，需优化驱动的超时和重试配置，适配Keyspaces的服务特性：

// 在Client配置中新增以下参数
policies: {
  retry: new cassandra.policies.retry.RetryPolicy({
    maxRetries: 3,
    retryOnTimeout: true
  }),
  reconnection: new cassandra.policies.reconnection.ConstantReconnectionPolicy(1000)
},
socketOptions: {
  connectTimeout: 10000,
  readTimeout: 15000
}

3. 全局复用客户端连接 #

Lambda函数执行时，避免每次请求创建新客户端实例，将客户端初始化放在全局作用域，复用已建立的连接：

import cassandra from 'cassandra-driver';
import fs from 'fs';
const dns = require('dns').promises;

let tpmsClient = null;

async function initClient() {
  if (!tpmsClient) {
    const contactPoints = await dns.resolve4('cassandra.eu-west-1.amazonaws.com');
    let auth = new cassandra.auth.PlainTextAuthProvider('cass-user-at-xxxxxxxxxx', 'zzzzzzzzz');
    let sslOptions1 = {
      ca: [ fs.readFileSync('/opt/utils/AmazonRootCA1.pem', 'utf-8')],
      host: 'cassandra.eu-west-1.amazonaws.com',
      rejectUnauthorized: true
    };
    tpmsClient = new cassandra.Client({
      contactPoints: contactPoints,
      localDataCenter: 'eu-west-1',
      authProvider: auth,
      sslOptions: sslOptions1,
      keyspace: 'tpms',
      protocolOptions: { port: 9142 },
      policies: {
        retry: new cassandra.policies.retry.RetryPolicy({ maxRetries: 3, retryOnTimeout: true }),
        reconnection: new cassandra.policies.reconnection.ConstantReconnectionPolicy(1000)
      },
      socketOptions: { connectTimeout: 10000, readTimeout: 15000 }
    });
    await tpmsClient.connect();
  }
  return tpmsClient;
}

export default class AmazonKeyspace {
  async getOrganisation(orgKey) {
    const client = await initClient();
    const SQL = 'select * FROM organisation where organisation_id=?;';
    const result = await client.execute(SQL, [orgKey], { prepare: true });
    return result.rows;
  }
}

4. 配置心跳抑制Socket闲置关闭 #

Keyspaces会关闭闲置超时的连接，调整驱动心跳参数保持连接活性：

// 在Client配置中新增心跳参数
heartbeatInterval: 30000,
idleTimeout: 60000

内容的提问来源于stack exchange，提问作者Andrew Haynes