1import { Log } from '@kubernetes/client-node'
2import { nodeStream } from '../../../serverF/libStreamF.mts'
3import { getKubeApis } from '../getApisF.mts'
4import type { WithClusterName } from '../ctxF/klusterCtxF.mts'
5import { chalkCyan, chalkGreen, chalkMagenta, chalkYellow } from '../../../serverF/libChalkF.mts'
6import { getOfflineReason, getKubeTimeoutSec } from '../ctxF/klusterReachabilityAI.mts'
7import { throPtErr } from '../../../sharedF/throwErrorF/sharedErrF.mts'
8
9const podColors = [chalkCyan, chalkGreen, chalkMagenta, chalkYellow]
10
11type Ret1PodLogsProps = WithClusterName & {
12  podName: string
13  containerName: string
14  tailLines?: number
15  follow?: boolean
16  pretty?: boolean
17  timestamps?: boolean
18  streamStdout?: boolean
19}
20
21export const ret1PodLogs = async (props: Ret1PodLogsProps) => {
22  const {podName, containerName, cluster_name, tailLines, follow = false, pretty = false, timestamps = false, streamStdout = false} = props
23
24  const offlineReason = getOfflineReason(cluster_name)
25  if (offlineReason) {
26    throPtErr('cluster offline', {cluster_name, offlineReason, what: 'retPodLogs'})
27  }
28
29  const {kubeConfig} = await getKubeApis({cluster_name})
30
31  const log = new Log(kubeConfig)
32  const stream = new nodeStream.PassThrough()
33  if (streamStdout) {
34    stream.pipe(process.stdout)
35  }
36
37  let retStr = ''
38  stream.on('data', (chunk) => {
39    retStr += chunk
40  })
41
42  // Set up end listener BEFORE calling log.log() to avoid race condition
43  const endPromise = new Promise<string>((resolve) => {
44    stream.on('end', () => resolve(retStr))
45  })
46
47  const namespace = 'default'
48
49  if (follow) {
50    await log.log(namespace, podName, containerName, stream, {follow, tailLines, pretty, timestamps})
51    return await endPromise
52  }
53
54  const timeoutMs = getKubeTimeoutSec(cluster_name) * 1000
55  return await new Promise<string>((resolve, reject) => {
56    const timer = setTimeout(() => {
57      stream.destroy()
58      reject(new Error(`k8s log timeout (${timeoutMs}ms) cluster=${cluster_name}`))
59    }, timeoutMs)
60
61    endPromise.then((str) => { clearTimeout(timer); resolve(str) })
62
63    log.log(namespace, podName, containerName, stream, {follow, tailLines, pretty, timestamps})
64      .catch((err) => { clearTimeout(timer); reject(err) })
65  })
66}
67
68type PodInfo = {podName: string, containerName: string}
69type RetMultiPodLogsProps = WithClusterName & {
70  pods: PodInfo[]
71  follow?: boolean
72  tailLines?: number
73}
74
75export const retMultiPodLogs = async (props: RetMultiPodLogsProps) => {
76  const {pods, cluster_name, tailLines, follow = false} = props
77  const {kubeConfig} = await getKubeApis({cluster_name})
78  const log = new Log(kubeConfig)
79  const namespace = 'default'
80
81  const streamPromises = pods.map(async ({podName, containerName}, idx) => {
82    const colorFn = podColors[idx % podColors.length]
83    const podShort = podName.slice(-8)
84    const prefix = colorFn(`[${podShort}]`)
85    
86    const stream = new nodeStream.PassThrough()
87    let buffer = ''
88    stream.on('data', (chunk: Buffer) => {
89      buffer += chunk.toString()
90      const lines = buffer.split('\n')
91      buffer = lines.pop() || ''
92      for (const line of lines) {
93        process.stdout.write(`${prefix} ${line}\n`)
94      }
95    })
96    stream.on('end', () => {
97      if (buffer) process.stdout.write(`${prefix} ${buffer}\n`)
98    })
99
100    // Set up end listener BEFORE calling log.log() to avoid race condition
101    const endPromise = new Promise<void>((resolve) => stream.on('end', resolve))
102    await log.log(namespace, podName, containerName, stream, {follow, tailLines, pretty: true, timestamps: false})
103    return endPromise
104  })
105
106  await Promise.all(streamPromises)
107}
108