unity-builder/src/model/cloud-runner/providers/k8s/kubernetes-task-runner.ts

import { CoreV1Api, KubeConfig } from '@kubernetes/client-node';
import CloudRunnerLogger from '../../services/core/cloud-runner-logger';
import waitUntil from 'async-wait-until';
import { CloudRunnerSystem } from '../../services/core/cloud-runner-system';
import CloudRunner from '../../cloud-runner';
import KubernetesPods from './kubernetes-pods';
import { FollowLogStreamService } from '../../services/core/follow-log-stream-service';
import { RemoteClientLogger } from '../../remote-client/remote-client-logger';

class KubernetesTaskRunner {
  static readonly maxRetry: number = 3;
  static async runTask(
    kubeConfig: KubeConfig,
    kubeClient: CoreV1Api,
    jobName: string,
    podName: string,
    containerName: string,
    namespace: string,
  ) {
    let output = '';
    let shouldReadLogs = true;
    let shouldCleanup = true;
    let retriesAfterFinish = 0;
    // eslint-disable-next-line no-constant-condition
    while (true) {
      await new Promise((resolve) => setTimeout(resolve, 3000));
      CloudRunnerLogger.log(
        `Streaming logs from pod: ${podName} container: ${containerName} namespace: ${namespace} ${CloudRunner.buildParameters.kubeVolumeSize}/${CloudRunner.buildParameters.containerCpu}/${CloudRunner.buildParameters.containerMemory}`,
      );
      let extraFlags = ``;
      extraFlags += (await KubernetesPods.IsPodRunning(podName, namespace, kubeClient))
        ? ` -f -c ${containerName}`
        : ` --previous`;

      const callback = (outputChunk: string) => {
        output += outputChunk;

        // split output chunk and handle per line
        for (const chunk of outputChunk.split(`\n`)) {
          // check if log start included in logs if so log a message
          if (chunk.includes(`Collected Logs`)) {
            CloudRunnerLogger.log(`Log Start found in logs`);
          }
          if (chunk.includes(`LOGHASH:`)) {
            RemoteClientLogger.HandleLogHash(chunk);
            CloudRunnerLogger.log(`Loghash found`);
          }
          if (chunk.includes(`LOGS:`)) {
            CloudRunnerLogger.log(`LOGS: found`);

            // remove "LOGS: " and decode base64 remaining
            const unpacked = Buffer.from(chunk.split(`LOGS: `)[1], 'base64').toString('ascii');
            const result = RemoteClientLogger.HandleLogFull(unpacked);
            CloudRunnerLogger.log(`Logs found HandleLogChunkLineResult:${result}`);
            if (result) {
              FollowLogStreamService.DidReceiveEndOfTransmission = true;
            }

            return;
          }
          ({ shouldReadLogs, shouldCleanup, output } = FollowLogStreamService.handleIteration(
            chunk,
            shouldReadLogs,
            shouldCleanup,
            output,
          ));
        }
      };
      try {
        await CloudRunnerSystem.Run(`kubectl logs ${podName}${extraFlags}`, false, true, callback);
      } catch (error: any) {
        await new Promise((resolve) => setTimeout(resolve, 3000));
        const continueStreaming = await KubernetesPods.IsPodRunning(podName, namespace, kubeClient);
        CloudRunnerLogger.log(`K8s logging error ${error} ${continueStreaming}`);
        if (continueStreaming) {
          continue;
        }
        if (retriesAfterFinish < KubernetesTaskRunner.maxRetry) {
          retriesAfterFinish++;

          continue;
        }
        throw error;
      }
      if (FollowLogStreamService.DidReceiveEndOfTransmission) {
        CloudRunnerLogger.log('end of log stream');
        break;
      }
    }

    return output;
  }

  static async watchUntilPodRunning(kubeClient: CoreV1Api, podName: string, namespace: string) {
    let waitComplete: boolean = false;
    let message = ``;
    CloudRunnerLogger.log(`Watching ${podName} ${namespace}`);
    await waitUntil(
      async () => {
        const status = await kubeClient.readNamespacedPodStatus(podName, namespace);
        const phase = status?.body.status?.phase;
        waitComplete = phase !== 'Pending';
        message = `Phase:${status.body.status?.phase} \n Reason:${
          status.body.status?.conditions?.[0].reason || ''
        } \n Message:${status.body.status?.conditions?.[0].message || ''}`;

        // CloudRunnerLogger.log(
        //   JSON.stringify(
        //     (await kubeClient.listNamespacedEvent(namespace)).body.items
        //       .map((x) => {
        //         return {
        //           message: x.message || ``,
        //           name: x.metadata.name || ``,
        //           reason: x.reason || ``,
        //         };
        //       })
        //       .filter((x) => x.name.includes(podName)),
        //     undefined,
        //     4,
        //   ),
        // );
        if (waitComplete || phase !== 'Pending') return true;

        return false;
      },
      {
        timeout: 2000000,
        intervalBetweenAttempts: 15000,
      },
    );
    if (!waitComplete) {
      CloudRunnerLogger.log(message);
    }

    return waitComplete;
  }
}

export default KubernetesTaskRunner;