背景需求

最近，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 作者：_夜游神

如何在 Next.js 中实现与 Kubernetes 容器的 Shell 交互

背景需求

遇到的挑战

实现思路

架构设计

前端实现细节

后端实现

启动 Websocket 服务

Websocket 监听具体实现

权限校验和参数验证

获取容器信息

创建数据流并处理输入输出

建立与Kubernetes容器的连接

关键点：

待优化

总结

发表回复

如何在 Next.js 中实现与 Kubernetes 容器的 Shell 交互

背景需求

遇到的挑战

实现思路

架构设计

前端实现细节

后端实现

启动 Websocket 服务

Websocket 监听具体实现

权限校验和参数验证

获取容器信息

创建数据流并处理输入输出

建立与Kubernetes容器的连接

关键点：

待优化

总结

相关推荐

发表回复