1import * as _ from 'lodash-es'
2import { CoreV1Api } from '@kubernetes/client-node'
3import { getKubeApis } from '../getApisF.mts'
4import { getAppPods } from '../podsF/k8sPodsF.mts'
5import { resourcesActionLite } from '../resourcesActionLiteF.mts'
6import { resourceForJobType } from '../jobTypeF.mts'
7import { resourcesAsPods } from '../getSinglePodF.mts'
8import { doKubeExecCapture } from '../kubeExecF.mts'
9import { pollUntil } from '../throwTimedOutAI.mts'
10import { getPodStatusFromPod } from '../getPodStatusAI.mts'
11import { mbit, gbit } from '../../../sharedF/strF/humanBytesF.mts'
12import { assertDefined } from '../../../sharedF/assertsF/assertDefinedF.mts'
13import { throwIf } from '../../../sharedF/throwErrorF/throwIfF.mts'
14import { throwDebugH } from '../../../serverF/throwErrorF/throwDebugHF.mts'
15
16type CliActionProps = { cluster_name: string, args: string[], kubeConfigPath?: string }
17type CliAction = { (props: CliActionProps): Promise<void>, cliDescript: string }
18
19const iperfImage = 'networkstatic/iperf3:latest'
20const iperfPort = 5201
21const iperfDurationSec = 30
22const podReadyTimeoutMs = 120_000
23const imageFailReasons = ['ErrImagePull', 'ImagePullBackOff']
24
25const parseIperfGbps = (output: string): number => {
26  const match = output.match(/([\d.]+)\s+(Gbits|Mbits)\/sec.*sender/)
27  assertDefined(match, {output})
28  const valStr = match[1], unit = match[2]
29  const bitsPerSec = parseFloat(valStr) * (unit === 'Mbits' ? mbit : gbit)
30  return bitsPerSec / gbit
31}
32
33export const netbench: CliAction = async ({cluster_name}) => {
34  const runId = Date.now().toString(36) + crypto.randomUUID().slice(0, 4)
35
36  const {kubeConfig} = await getKubeApis({cluster_name})
37  const api = kubeConfig.makeApiClient(CoreV1Api)
38  const {items: nodes} = await api.listNode()
39  const nodeNames = _.chain(nodes)
40    .filter(node => node.status?.conditions?.some(c => c.type === 'Ready' && c.status === 'True') === true)
41    .map(node => node.metadata?.name)
42    .compact()
43    .value()
44  throwIf(() => nodeNames.length < 2, {cluster_name})
45
46  const mkNetbenchJob = async ({name, node, taskCmd}: {name: string, node: string, taskCmd: string}) =>
47    resourceForJobType({
48      jobType: {kind: 'Job', backoffLimit: 0, nodeSelector: {'kubernetes.io/hostname': node}},
49      name, cluster_name, image: iperfImage, taskCmd, envLocal: {}, git_sha: runId,
50    })
51
52  const waitForPodRunning = async ({name}: {name: string}) => {
53    await pollUntil({
54      predicate: async () => {
55        const pods = await getAppPods({cluster_name, name})
56        const status = getPodStatusFromPod(pods[0])
57        if (status?.waitingReason && imageFailReasons.includes(status.waitingReason)) throwDebugH({name, waitingReason: status.waitingReason})
58        return !status?.running
59      },
60      timeoutLabel: `netbench pod ${name}`,
61      timeoutMs: podReadyTimeoutMs,
62      isQuiet: true,
63    })
64    const pods = await getAppPods({cluster_name, name})
65    const pod = pods[0]
66    assertDefined(pod)
67    const podName = pod.metadata?.name
68    assertDefined(podName)
69    const podIp = pod.status?.podIP
70    assertDefined(podIp)
71    return {podName, podIp}
72  }
73
74  const cleanupPods = async ({name}: {name: string}) => {
75    const pods = await getAppPods({cluster_name, name})
76    if (!pods.length) return
77    await resourcesActionLite({resources: resourcesAsPods(pods as object[]), action: 'delete', cluster_name, isQuiet: true})
78  }
79
80  const measureDirection = async ({source, target}: {source: string, target: string}): Promise<number> => {
81    const serverName = `netbench-${runId}-${target}-srv`
82    const clientName = `netbench-${runId}-${source}-cli`
83    const serverJob = await mkNetbenchJob({name: serverName, node: target, taskCmd: `while true; do iperf3 -s -p ${iperfPort}; done`})
84    const clientJob = await mkNetbenchJob({name: clientName, node: source, taskCmd: 'sleep 3600'})
85    try {
86      await resourcesActionLite({resources: [serverJob, clientJob], action: 'apply', cluster_name, isQuiet: true})
87      const {podIp: serverIp} = await waitForPodRunning({name: serverName})
88      const {podName: clientPodName} = await waitForPodRunning({name: clientName})
89      const output = await doKubeExecCapture({
90        podName: clientPodName, containerName: clientName, isQuiet: true,
91        cmdA: ['iperf3', '-c', serverIp, '-p', String(iperfPort), '-t', String(iperfDurationSec), '-f', 'g', '--connect-timeout', '5000'],
92      })
93      return parseIperfGbps(output)
94    } finally {
95      await resourcesActionLite({resources: [serverJob, clientJob], action: 'delete', cluster_name, isQuiet: true})
96      await cleanupPods({name: serverName})
97      await cleanupPods({name: clientName})
98    }
99  }
100
101  const uniqPairs: [string, string][] = []
102  for (let i = 0; i < nodeNames.length; i++) {
103    for (let j = i + 1; j < nodeNames.length; j++) {
104      uniqPairs.push([nodeNames[i], nodeNames[j]])
105    }
106  }
107
108  console.log(`netbench: ${uniqPairs.length} pairs across ${nodeNames.length} nodes (${iperfDurationSec}s each)`)
109  const failures: string[] = []
110  for (const [a, b] of uniqPairs) {
111    const results: string[] = []
112    for (const [source, target] of [[a, b], [b, a]] as [string, string][]) {
113      try {
114        const gbps = await measureDirection({source, target})
115        results.push(`${source} -> ${target} ${gbps.toFixed(1)} Gbps`)
116      } catch (err) {
117        const msg = err instanceof Error ? err.message : String(err)
118        results.push(`${source} -> ${target} FAILED: ${msg}`)
119        failures.push(`${source} -> ${target}: ${msg}`)
120      }
121    }
122    console.log(results.join(', '))
123  }
124  if (failures.length) throwDebugH({failures})
125}
126netbench.cliDescript = 'measure pod-network bandwidth between all node pairs'
127