背景需求
最近,SCOW 的一个项目需求是在前端实现一个 Shell 功能,以便通过 Shell 命令与 Kubernetes(K8s)中运行的容器进行交互式通信。
遇到的挑战
在尝试实现这一功能时,我遇到了一些技术挑战:
- Node.js实现参考资料稀缺:如果使用Golang作为后端,那么通过 kubernetes/client-go 库来实现相对较为简单,并且网络上有众多相关的教程。但是,转向 Node.js 实现时,虽然也存在一个 kubernetes/client-node 库,相关的实现示例却寥寥无几。
- Next.js 中的 WebSocket 实现不多:在 Next.js 中实现 WebSocket 而不依赖于如 Express 这样的框架的示例也不多见。
如果读者也遇到了这些小困难,那么欢迎您接着往下阅读。
实现思路
架构设计
- 前端实现:使用 xterm.js 在 Next.js 前端部分实现终端模拟效果,并通过 WebSocket 与后端建立通信。
- 后端服务:在 Next.js 后端启动 WebSocket 服务,监听特定 API 端点,并使用 K8s 的 Node 客户端库进入运行中的容器执行命令。
暂时无法在飞书文档外展示此内容
前端实现细节
首先,我们开发了一个 React 组件,用于处理对不同 Kubernetes 容器的连接请求。组件的属性包括 user
(用户信息),cluster
(集群标识),以及jobId
(作业 ID),这些属性帮助定位并连接到具体的容器。其中 jobId
是表示当前容器的一个唯一标识。
// JobShell.tsx
// 定义一个样式组件用于终端容器
const TerminalContainer = styled.div`
background-color: black; // 背景颜色设置为黑色
flex: 1; // 弹性布局,允许组件自动填充空间
width: 100%; // 宽度设置为100%
`;
// 定义组件接收的props类型
interface Props {
user: ClientUserInfo; // 用户信息
cluster: string; // 集群标识
jobId: string; // 作业ID
}
// JobShell组件定义
export const JobShell: React.FC<Props> = ({ user, cluster, jobId }) => {
const { publicConfig: { BASE_PATH } } = usePublicConfig();
// 用于引用终端容器元素
const container = useRef<HTMLDivElement>(null);
// 用于标记终端是否已初始化
const terminalInitialized = useRef<boolean>(false);
// useEffect钩子来初始化终端
useEffect(() => {
// 当容器存在且终端未初始化时
if (container.current && !terminalInitialized.current) {
// 创建一个新的Terminal实例
const term = new Terminal({
cursorBlink: true, // 光标闪烁
});
// 加载并应用自适应大小的插件
const fitAddon = new FitAddon();
term.loadAddon(fitAddon);
// 在容器中打开终端
term.open(container.current);
terminalInitialized.current = true;
// 设置连接的有效载荷数据
const payload = {
cluster,
jobId,
};
// 输出连接信息
term.write(
`*** Connecting to cluster ${payload.cluster} as ${user.identityId} \r\n`,
);
// 根据协议选择ws或wss并创建WebSocket连接
const socket = new WebSocket(
(location.protocol === "http:" ? "ws" : "wss") + "://" + location.host +
join(BASE_PATH, "/api/jobShell") + "?" + new URLSearchParams(payload).toString(),
);
// 设置WebSocket消息事件处理
socket.onmessage = (e) => {
const message = JSON.parse(e.data) as ShellOutputData;
switch (message.$case) {
case "data":
const data = Buffer.from(message.data.data);
term.write(data);
break;
case "exit":
term.write(`Process exited with code ${message.exit.code} and signal ${message.exit.signal}.`);
break;
}
};
// WebSocket连接开启时的逻辑
socket.onopen = () => {
term.clear(); // 清空终端
// 发送数据的辅助函数
const send = (data: ShellInputData) => {
socket.send(JSON.stringify(data));
};
// 创建并使用ResizeObserver来处理终端大小调整
const resizeObserver = new ResizeObserver(debounce(() => {
fitAddon.fit();
send({ $case: "resize", resize: { cols: term.cols, rows: term.rows } });
}));
// 观察终端容器的大小变化
resizeObserver.observe(container.current!);
// 终端数据输入事件处理
term.onData((data) => {
send({ $case: "data", data: { data } });
});
};
// 组件卸载时的清理逻辑
return () => {
if (socket) socket.close();
if (term) term.dispose();
terminalInitialized.current = false;
};
}
}, [container.current]); // 依赖项仅包括容器的当前引用
// 渲染终端容器
return (
<TerminalContainer ref={container} />
);
};
在上述代码中,我们通过 WebSocket 连接到后端,并随着终端窗口的大小变化发送更新。这样的设计确保了前端与 Kubernetes 容器间的实时、动态交互。
注意:对于Xterm.js的具体用法和细节,建议查阅其官方文档,本文不再深入探讨。
这里特别提一下关于窗口大小调整的问题。在前端实现中,我们使用了以下代码实现。我们监听了 terminal 容器大小的变化,如果大小发生变化则调整 xterm 的大小并向后端发送调整窗口大小的消息。
const resizeObserver = new ResizeObserver(debounce(() => {
fitAddon.fit();
send({ $case: "resize", resize: { cols: term.cols, rows: term.rows } });
}));
resizeObserver.observe(container.current!);
后端实现
启动 Websocket 服务
后端的实现稍微复杂些。我们采用了一种较为间接的方法来启动WebSocket服务,不依赖于传统的Web服务器。具体实现如下:
- 配置 Next.js 以启动 WebSocket: 我们在
next.config.mjs
文件中添加特定的逻辑,用于在 Next.js 启动时,通过调用一个设置 API 来初始化 WebSocket 服务。
export default async () => {
// ...其他配置
if (!building) {
// HACK setup ws proxy
setTimeout(() => {
const url = `http://localhost:${process . env . PORT || 3000}${join(BASE_PATH, "/api/setup")}`;
console . log("Calling setup url to initialize proxy and shell server", url);
fetch(url) . then(async (res) => {
console . log("Call completed. Response: ", await res . text());
}) . catch((e) => {
console . error("Error when calling proxy url to initialize ws proxy server", e);
});
});
}
// ...其他配置
return nextConfig;
};
- 处理WebSocket设置请求: 在
pages/api/setup.ts
中,我们检查是否已经设置WebSocket,如果未设置,则进行配置。
// pages/api/setup.ts
let setup = false;
export default async (req: NextApiRequest, res: any) => {
if (setup) {
res . send("Already setup");
return;
}
setupJobShellServer(res);
setup = true;
res . send("Setup complete");
};
在 setup 中调用 setupJobShellServer 来启动 websocket 服务,同样的方法,此处还能完成其他相关的服务监听。
- 启动 WebSocket 服务: 在
setupJobShellServer
函数中,我们配置 WebSocket 服务器,以处理特定的 API 路由请求。
// setupJobShellServer
const wss = new WebSocketServer({ noServer: true });
type AliveCheckedWebSocket = WebSocket & { isAlive: boolean };
export const setupJobShellServer = (req: NextApiRequest) => {
(req.socket as any).server.on("upgrade", async (req: IncomingMessage,
socket: any, head: any) => {
const url = normalizePathnameWithQuery(req.url!);
if (!url.startsWith(join(BASE_PATH, "/api/jobShell"))) {
return;
}
wss.handleUpgrade(req, socket, head, (ws) => {
// 动态地为 WebSocket 实例添加 isAlive 属性
const extendedWs = ws as AliveCheckedWebSocket;
extendedWs.isAlive = true;
wss.emit("connection", extendedWs, req);
});
});
};
通过这种方法,我们可以动态地响应 WebSocket 连接请求,并将其与 Next.js 的其他 HTTP 功能无缝整合。这种方法虽然略显复杂,但避免了引入额外的依赖或服务器。
Websocket 监听具体实现
接下来就是对 websocket 的 connection 事件的具体实现如下,先给出完整代码,再分段解释
// 定义ShellQuery类型,包括集群信息和可选的终端尺寸
export type ShellQuery = {
cluster: string;
cols?: string;
rows?: string;
}
// 定义Shell输入数据的类型,可以是调整大小、发送数据或断开连接
export type ShellInputData =
| { $case: "resize", resize: { cols: number; rows: number } }
| { $case: "data", data: { data: string } }
| { $case: "disconnect" };
// 定义Shell输出数据的类型,可以是数据或退出信息
export type ShellOutputData =
| { $case: "data", data: { data: string } }
| { $case: "exit", exit: { code?: number; signal?: string } };
// WebSocket服务端逻辑,处理连接事件
wss.on("connection", async (ws: AliveCheckedWebSocket, req) => {
// 从请求中获取用户Token
const token = getUserToken(req);
// 检查Token有效性
if (!token) {
console.log("[shell] token is not valid");
ws.close(0, "token is not valid");
return;
}
// 验证Token,获取用户信息
const userInfo = await validateToken(token);
if (!userInfo) {
console.log("[shell] userInfo is not valid");
ws.close(0, "userInfo is not valid");
return;
}
// 日志函数,便于跟踪用户操作
const log = (message: string, ...optionalParams: any[]) => console.log(
`[io] [${userInfo.identityId}] ${message}`, optionalParams);
log("Connection request received.");
// 解析请求中的查询参数
const fullUrl = "http://example.com" + req.url;
const query = new URL(fullUrl).searchParams;
const clusterId = query.get("cluster");
const jobId = query.get("jobId");
// 检查jobId是否传递
if (!jobId) {
log("[params] param-jobId not passed");
ws.close(0, "param-jobId not passed");
return;
}
// 检查集群ID是否传递且该集群是否已配置
if (!clusterId || !clusters[clusterId]) {
log("[params] param-clusterId not passed or unknown");
ws.close(0, "param-clusterId not passed or unknown");
return;
}
// 检查集群配置中是否有kubeconfig路径
if (!clusters[clusterId].k8s?.kubeconfig.path) {
log("[config] The current cluster does not have kubeconfig configured.");
ws.close(0, "The current cluster does not have kubeconfig configured.");
return;
}
// 获取运行作业的详细信息
const client = getAdapterClient(clusterId);
const runningJobsInfo = await asyncClientCall(client.job, "getJobs", {
fields: ["job_id"],
filter: {
users: [userInfo.identityId], accounts: [],
states: ["RUNNING"],
},
}).then((resp) => resp.jobs);
const currentJobInfo = runningJobsInfo.find((jobInfo) => String(jobInfo.jobId) === jobId);
if (!currentJobInfo) {
log(`[shell] Get running job node info failed, can't find job ${jobId}`);
ws.close(0, `Get running job node info failed, can't find job ${jobId}`);
return;
}
// 心跳机制,用于检查WebSocket连接是否存活
ws.isAlive = true;
ws.on("pong", () => {
heartbeat.call(ws as AliveCheckedWebSocket);
});
// 向客户端发送ping消息
ws.ping();
// 发送数据到WebSocket客户端的函数
const send = (data: ShellOutputData) => {
ws.send(JSON.stringify(data));
};
// 创建stdin, stdout, stderr流,用于与Kubernetes容器通信
const stdinStream = new PassThrough();
const stdoutStream = new PassThrough();
const stderrStream = new PassThrough();
// 处理从Kubernetes容器接收到的stdout和stderr数据
stdoutStream.on("data", (data) => {
send({ $case: "data", data: { data: data.toString() } });
});
stderrStream.on("data", (data) => {
send({ $case: "data", data: { data: data.toString() } });
});
// WebSocket错误处理
ws.on("error", async (err) => {
log("Error occurred from client. Disconnect.", err);
stdinStream.end(); // 结束stdin流
});
// 获取正在运行的容器信息
const { namespace, pod } = await asyncClientCall(client.app, "getRunningContainerJobInfo", {
jobId: currentJobInfo.jobId,
});
if (!namespace || !pod) {
log("[shell] Namespace or pod not obtained, please check the adapter version");
ws.close(0, "Namespace or pod not obtained, please check the adapter version");
return;
}
try {
// 配置和连接到Kubernetes集群
const kc = new k8sClient.KubeConfig();
kc.loadFromFile(join("/etc/scow", clusters[clusterId].k8s?.kubeconfig.path || "/kube/config"));
const k8sWs = await new k8sClient.Exec(kc)
.exec(namespace, pod, "", ["/bin/sh"], stdoutStream, stderrStream, stdinStream, true);
log("Connected to shell");
// 处理来自WebSocket客户端的消息
ws.on("message", (data) => {
const message = JSON.parse(data.toString());
switch (message.$case) {
case "data":
stdinStream.write(message.data.data);
break;
case "resize":
stdinStream.write(`stty cols ${message.resize.cols} rows ${message.resize.rows}\n`);
break;
case "disconnect":
stdinStream.end(); // 断开连接时结束stdin流
break;
}
});
// 处理WebSocket关闭事件
ws.on("close", () => {
stdinStream.end(); // 关闭stdin流
stdoutStream.end(); // 关闭stdout流
stderrStream.end(); // 关闭stderr流
k8sWs.close(); // 关闭Kubernetes WebSocket连接
});
} catch (error) {
console.error("Error executing command in Kubernetes", error);
ws.close();
}
});
-
权限校验和参数验证
在WebSocket连接建立时,首先进行用户的权限校验和请求参数的验证,以确保请求的合法性和正确性:
const token = getUserToken(req);
if (!token) {
console.log("[shell] token is not valid");
ws.close(0, "token is not valid");
return;
}
const userInfo = await validateToken(token);
if (!userInfo) {
console.log("[shell] userInfo is not valid");
ws.close(0, "userInfo is not valid");
return;
}
const log = (message: string, ...optionalParams: any[]) => console.log(
`[io] [${userInfo.identityId}] ${message}`, optionalParams);
log("Connection request received.");
const fullUrl = "http://example.com" + req.url;
const query = new URL(fullUrl).searchParams;
const clusterId = query.get("cluster");
const jobId = query.get("jobId");
if (!jobId) {
log("[params] param-jobId not passed");
ws.close(0, "param-jobId not passed");
return;
}
if (!clusterId || !clusters[clusterId]) {
log("[params] param-clusterId not passed or unknown");
ws.close(0, "param-clusterId not passed or unknown");
return;
}
-
获取容器信息
根据 jobId
和 clusterId
从后端获取容器的namespace
和podname
。此部分的代码应根据项目具体要求实现,关键在于如何有效地将前端请求转发到 Kubernetes 容器中进行处理。
-
创建数据流并处理输入输出
创建stdin
, stdout
, stderr
流,并监控这些流,将 Kubernetes 容器的输出实时发送回 WebSocket 客户端:
// 首先创建三个流
// 创建stdin, stdout, stderr流,用于与Kubernetes容器通信
const stdinStream = new PassThrough();
const stdoutStream = new PassThrough();
const stderrStream = new PassThrough();
// 监听 out、err 流的输出,将其送回到前端
// 将Kubernetes stdout和stderr的输出发送回WebSocket客户端
const send = (data: ShellOutputData) => {
ws.send(JSON.stringify(data));
};
stdoutStream.on("data", (data) => {
send({ $case: "data", data: { data: data.toString() } });
});
stderrStream.on("data", (data) => {
send({ $case: "data", data: { data: data.toString() } });
});
这种方式确保了与Kubernetes容器的实时交互和数据交换,为用户提供了一种高效、直接的方式来监控和操作正在运行的容器。
-
建立与Kubernetes容器的连接
在与Kubernetes容器建立连接时,我们需要确保具备适当的权限以执行必要的命令。以下代码详细说明了如何加载配置文件,建立连接,并处理来自WebSocket的消息。
为了能够在 Kubernetes 集群中通过 kubectl 进入到所有命名空间的容器中执行命令(例如 /bin/sh),需要提供一份 kubeconfig 配置文件。该配置文件的 current context 中的用户需要使用 ClusterRole 创建并具备一定的权限,这些权限包括对 pods/exec 的 create 操作,以及对 pods 的 get 和 list 操作。在我们代码中通过 loadFromFile
函数加载配置文件,通过 exec
方法执行 /bin/sh
命令启动 shell
环境,然后将之前创建的三个流传入,就能完成数据交互。
import * as k8sClient from "@kubernetes/client-node";
try {
// 配置和连接到Kubernetes集群
const kc = new k8sClient.KubeConfig();
kc.loadFromFile(join("/etc/scow", clusters[clusterId].k8s?.kubeconfig.path || "/kube/config"));
const k8sWs = await new k8sClient.Exec(kc)
.exec(namespace, pod, "", ["/bin/sh"], stdoutStream, stderrStream, stdinStream, true);
log("Connected to shell");
// 处理来自WebSocket客户端的消息
ws.on("message", (data) => {
const message = JSON.parse(data.toString());
switch (message.$case) {
case "data":
stdinStream.write(message.data.data);
break;
case "resize":
stdinStream.write(`stty cols ${message.resize.cols} rows ${message.resize.rows}\n`);
break;
case "disconnect":
stdinStream.end(); // 断开连接时结束stdin流
break;
}
});
// 处理WebSocket关闭事件
ws.on("close", () => {
stdinStream.end(); // 关闭stdin流
stdoutStream.end(); // 关闭stdout流
stderrStream.end(); // 关闭stderr流
k8sWs.close(); // 关闭Kubernetes WebSocket连接
});
} catch (error) {
console.error("Error executing command in Kubernetes", error);
ws.close();
}
关键点:
- 权限需求: 用户需要拥有对
pods/exec
的创建权限以及对pods
的获取和列表权限。 - 错误处理: 任何连接或执行错误都会被捕捉,并通过 WebSocket 通知客户端。
- 消息处理: 根据客户端请求动态处理输入大小调整、数据传输和断开连接操作。
- 资源管理: 确保在 WebSocket 关闭时,释放所有相关资源以避免内存泄漏。
待优化
目前,我们通过接收前端发来的 Resize
消息并向 Kubernetes 容器发送 stty
命令来调整窗口大小。这种方法虽然简单直接,但可能不是最优雅的解决方案。我还没有发现 Kubernetes 客户端库中内置的相关功能。如果您了解更好的方法,欢迎分享给我。
总结
本文详细介绍了如何在 Next.js 环境下,通过 WebSocket 连接实现与 Kubernetes 容器的交互式通信。首先,我们探讨了前端如何通过 React 组件建立 WebSocket 连接并处理终端输入输出。随后,我们详细阐述了后端如何在不依赖额外Web服务器的情况下启动 WebSocket 服务,并处理权限验证及参数校验,确保安全性。特别地,我们通过 Kubernetes 客户端与容器建立连接,并执行命令。尽管当前的窗口大小调整实现较为基础,我们仍在寻找更优雅的解决方案。希望这些内容能帮助到面临类似技术挑战的开发者,并期待社区的更多反馈与建议。
原文链接:https://juejin.cn/post/7357554457914818595 作者:_夜游神