1import { Exec } from "@kubernetes/client-node"
2import { nodeStream } from "../../serverF/libStreamF.mts";
3import { PtErr } from "../../sharedF/throwErrorF/sharedErrF.mts";
4import { getKubeApis } from './getApisF.mts'
5import { getKlusterCtx } from "./ctxF/klusterCtxF.mts";
6import { maskQs } from '../../serverF/dbF/maskQsF.mts'
7
8type DoKubeExecProps = { cmdA: string[], podName: string, containerName: string, isQuiet?: boolean }
9
10// single source of truth for the kubernetes Exec status callback. uses Promise reject (not
11// throPtErr) so failures in the websocket callback become catchable Promise rejections — the
12// callback is invoked by @kubernetes/client-node's websocket onmessage, OUTSIDE the Promise
13// executor, so a synchronous throw would be an unhandled rejection that crashes the process.
14const execWithStatus = (kubeConfig: any, {podName, containerName, cmdA, stream, onResult}: {
15  podName: string, containerName: string, cmdA: string[], stream: any, onResult?: () => any
16}): Promise<any> =>
17  new Promise((resolv, rej) => {
18    const exec = new Exec(kubeConfig)
19    exec.exec('default', podName, containerName, cmdA, stream, stream, null, false, (status: any) => {
20      if (status.status == 'Success') {
21        resolv(onResult?.())
22        return
23      }
24      rej(new PtErr('!kubeExec', {cmdA, podName, containerName}))
25    })
26  })
27
28export const doKubeExec = async ({cmdA, podName, containerName, isQuiet}: DoKubeExecProps) => {
29  const {cluster_name} = getKlusterCtx()
30  console.log('doKubeExec', cluster_name, 'exec -it', podName, cmdA.map(maskQs).join(' '))
31  const {kubeConfig} = await getKubeApis()
32  const stream = new nodeStream.PassThrough()
33  if (!isQuiet) {
34    stream.pipe(process.stdout)
35  }
36  return execWithStatus(kubeConfig, {podName, containerName, cmdA, stream})
37}
38
39export const doKubeExecCapture = async ({cmdA, podName, containerName, isQuiet}: DoKubeExecProps) => {
40  const {cluster_name} = getKlusterCtx()
41  if (!isQuiet) console.log('doKubeExecCapture', cluster_name, 'exec', podName, cmdA.map(maskQs).join(' '))
42  const {kubeConfig} = await getKubeApis()
43  const chunks: Buffer[] = []
44  const stream = new nodeStream.PassThrough()
45  stream.on('data', (chunk: Buffer) => chunks.push(chunk))
46  return execWithStatus(kubeConfig, {podName, containerName, cmdA, stream, onResult: () => Buffer.concat(chunks).toString('utf-8').trim()})
47}
48