oneacrefund/superset-websocket

Superset WebSocket服务器

概述

一个Node.js WebSocket服务器，用于向Superset Web应用前端发送异步事件数据。

要求

• Node.js 12+（未测试旧版本）
• Redis 5+

要使用此功能，Superset需配置启用全局异步查询并使用WebSocket作为传输方式（详见下文）。

架构

基于https://github.com/apache/superset/issues/9190%E5%AE%9A%E4%B9%89%E7%9A%84%E6%9E%B6%E6%9E%84%E5%AE%9E%E7%8E%B0%E3%80%82

流（Streams）

异步事件从https://github.com/preset-io/superset/blob/master/superset/async_events/async_query_manager.py%E6%8E%A8%E9%80%81%E5%88%B0Redis%E6%B5%81%E3%80%82%E7%89%B9%E5%AE%9A%E7%94%A8%E6%88%B7%E7%9A%84%E4%BA%8B%E4%BB%B6%E4%BC%9A%E5%8F%91%E5%B8%83%E5%88%B0%E4%B8%A4%E4%B8%AA%E6%B5%81%EF%BC%9A1%EF%BC%89%E5%8C%85%E5%90%AB%E6%89%80%E6%9C%89%E7%94%A8%E6%88%B7%E4%BA%8B%E4%BB%B6%E7%9A%84%E5%85%A8%E5%B1%80%E4%BA%8B%E4%BB%B6%E6%B5%81%EF%BC%9B2%EF%BC%89%E4%BB%85%E8%AF%A5%E7%94%A8%E6%88%B7%E7%9A%84%E9%A2%91%E9%81%93/%E4%BC%9A%E8%AF%9D%E7%89%B9%E5%AE%9A%E6%B5%81%E3%80%82%E8%BF%99%E7%A7%8D%E6%96%B9%E5%BC%8F%E5%9C%A8%E6%80%A7%E8%83%BD%EF%BC%88%E4%BB%8E%E5%8D%95%E4%B8%AA%E5%85%A8%E5%B1%80%E6%B5%81%E8%AF%BB%E5%8F%96%EF%BC%89%E5%92%8C%E5%AE%B9%E9%94%99%E6%80%A7%EF%BC%88%E6%96%AD%E5%BC%80%E7%9A%84%E8%BF%9E%E6%8E%A5%E5%8F%AF%E9%80%9A%E8%BF%87%E9%A2%91%E9%81%93%E7%89%B9%E5%AE%9A%E6%B5%81%E2%80%9C%E8%BF%BD%E8%B5%B6%E2%80%9D%EF%BC%89%E4%B9%8B%E9%97%B4%E5%8F%96%E5%BE%97%E5%B9%B3%E8%A1%A1%E3%80%82

注意：此处未使用Redis流的消费者组，因为每个组仅接收流数据的子集，而WebSocket客户端与每个应用实例有持久连接，需要访问流中的所有数据。WebSocket应用的水平扩展需要多个WebSocket服务器，每个都能完全访问Redis流数据。

连接

用户浏览器初始通过HTTP连接WebSocket服务器，请求中包含Flask应用设置的JWT认证Cookie。注意：由于基于Cookie的认证方式，WebSocket服务器必须与Web应用运行在同一主机上。 服务器使用共享密钥（配置：jwtSecret）验证JWT令牌，验证通过后将连接升级为WebSocket。JWT中包含用户会话的“频道”ID，作为向用户连接的socket发送事件的依据。

用户在单个频道（会话）ID下可能有多个WebSocket连接（例如多个浏览器标签）。在此情况下，特定频道的所有事件会发送到所有连接的socket，由消费者决定哪些事件与当前应用上下文相关。

重连

由于网络波动，用户的WebSocket连接可能断开。Superset前端代码跟踪最后接收的异步事件ID，并尝试使用初始HTTP请求中的last_id查询参数重连WebSocket服务器。如果连接包含有效的last_id，服务器会从用户特定的Redis流中读取可能已接收但发送失败的事件，并重新发送到新连接。之后全局事件流负责向连接的socket发送后续事件。

连接管理

服务器利用标准WebSocket的ping/pong功能判断活动连接是否存活。定期（配置：pingSocketsIntervalMs）向活动socket发送ping，收到pong响应时更新内部socket注册表的时间戳。若在超时时间（配置：socketResponseTimeoutMs）内未收到pong响应，将终止socket并从注册表中移除。

此外，内部频道注册表会定期清理（配置：gcChannelsIntervalMs）以删除过期引用，防止长期运行导致的内存过度消耗。

安装

安装依赖：

bash
npm ci

配置

WebSocket服务器配置

将config.example.json复制为config.json，并根据环境调整值。也支持通过环境变量配置，完整配置项可参考src/config.ts。

Superset配置

需在Superset Flask应用中配置（superset_config.py）以启用全局异步查询：

启用GLOBAL_ASYNC_QUERIES特性标志：

python
"GLOBAL_ASYNC_QUERIES": True

配置以下Superset值：

python
GLOBAL_ASYNC_QUERIES_TRANSPORT = "ws"
GLOBAL_ASYNC_QUERIES_WEBSOCKET_URL = "ws://<主机>:<端口>/"

注意：WebSocket服务器必须与Web应用运行在同一主机名（不同端口），以便Cookie在Flask应用和WebSocket服务器之间共享。此外，localhost和127.0.0.1不视为同一主机。例如，若浏览器访问localhost:<端口>的Superset，则WebSocket URL需配置为localhost:<端口>。

以下配置值在Flask应用配置和config.json中必须相同：

text
GLOBAL_ASYNC_QUERIES_CACHE_BACKEND
GLOBAL_ASYNC_QUERIES_REDIS_STREAM_PREFIX
GLOBAL_ASYNC_QUERIES_JWT_COOKIE_NAME
GLOBAL_ASYNC_QUERIES_JWT_SECRET

更多Superset异步查询配置信息：https://github.com/apache/superset/blob/master/CONTRIBUTING.md#async-chart-queries

StatsD监控

应用使用hot-shots库通过StatsD跟踪指标，如连接客户端数量、向客户端发送消息失败尝试次数等。

可通过配置文件中的statsd对象配置StatsD。默认配置如下：

json
{
  "statsd": {
    "host": "127.0.0.1",
    "port": 8125,
    "globalTags": []
  }
}

运行

本地开发服务器

bash
npm run dev-server

生产环境

bash
npm run build && npm start

健康检查

WebSocket服务器支持通过以下方式进行健康检查：

GET /health

或

HEAD /health

容器化

TODO：容器化WebSocket服务器