shutdown sidecar handler

shutdown-sidecar-handler.ts

import { retry } from '@at/infrastructure-runtime-common/decorators'
import { isRejected } from '@at/infrastructure-runtime-common/utils/promises'
import type { Kubernetes } from '@at/kubernetes'
import { PatchType } from '@at/kubernetes/client'
import type { Models } from '@at/kubernetes/models'
import { KubernetesEventType } from '@at/kubernetes/watchClient'
import { Logger } from '@at/platform-logger'
import { EventHandler } from 'lib/handlers/eventHandler'
import type { ObserverHandlerOptions } from 'types/dynamic/server'

/**
 * These are the sidecar names we should attempt to shut down
 * Any others will be ignored
 */
const knownSidecars = [
  'istio-proxy',
  'cloudsql-proxy',
  'cloudsql-mysql-proxy',
  'cloudsql-postgres-proxy',
  'cloudsql-sqlserver-proxy'
]

export class TerminateSidecarsInJobsHandler extends EventHandler<Models.Core.Pod> {
  private static readonly TARGET_LABEL = 'kubitzer.io/terminate-sidecars'

  private readonly logger = new Logger(this.constructor.name)

  constructor(private readonly kubernetes: Kubernetes) {
    super({
      apiVersion: 'v1',
      kind: 'Pod',
      interestedTypes: [
        KubernetesEventType.ADDED,
        KubernetesEventType.MODIFIED
      ],
      labelSelector: {
        matchLabels: { [TerminateSidecarsInJobsHandler.TARGET_LABEL]: 'true' }
      },
      fieldSelector: {
        matchExpressions: [
          {
            operator: 'Equals',
            key: 'status.phase',
            value: 'Running'
          }
        ]
      },
      startFrom: { position: 'beginning' }
    })
  }

  public override created(
    this: TerminateSidecarsInJobsHandler,
    options: ObserverHandlerOptions<Models.Core.Pod>
  ): Promise<void> {
    return this.handle(options)
  }
  public override updated(
    this: TerminateSidecarsInJobsHandler,
    options: ObserverHandlerOptions<Models.Core.Pod>
  ): Promise<void> {
    return this.handle(options)
  }
  public override deleted(
    this: TerminateSidecarsInJobsHandler,
    options: ObserverHandlerOptions<Models.Core.Pod>
  ): Promise<void> {
    return this.handle(options)
  }

  protected override shouldHandle(model: Models.Core.Pod): boolean {
    const containerStatuses = model.status?.containerStatuses
    const nonSidecarContainers =
      containerStatuses?.filter(
        (status) => !knownSidecars.includes(status.name)
      ) ?? []
    if (nonSidecarContainers.length === 0) {
      return false
    }
    const allNonSidecarsAreTerminated =
      nonSidecarContainers.filter((item) => item.state?.terminated).length ===
      nonSidecarContainers.length

    const isRestartPolicyNever = model.spec.restartPolicy === 'Never'
    return allNonSidecarsAreTerminated && isRestartPolicyNever
  }

  private async handle(
    this: TerminateSidecarsInJobsHandler,
    options: ObserverHandlerOptions<Models.Core.Pod>
  ): Promise<void> {
    const { resource, contextualLogger } = options

    const containerStatuses = resource.status?.containerStatuses
    const { name, namespace } = resource.metadata
    if (!containerStatuses) {
      return
    }
    const sidecarContainers = containerStatuses.filter(
      (status) => knownSidecars.includes(status.name) && status.state?.running
    )

    try {
      contextualLogger.info(
        'setting the termination label to already-handled to prevent future events from firing'
      )

      // Set the label to something other than true, so we dont handle it again
      await this.kubernetes.patch<Models.Core.Pod>('v1', 'Pod', {
        name,
        namespace,
        patchType: PatchType.MergePatch,
        patch: {
          metadata: {
            labels: {
              [TerminateSidecarsInJobsHandler.TARGET_LABEL]: 'already-handled'
            },
            annotations: {
              'events.kubitzer.io/prevent-processing': 'true'
            }
          }
        }
      })
    } catch (err) {
      if (err.code === 404) {
        contextualLogger.info(
          'Went to patch the container with the terminating status, but it appears the container has already been removed from the cluster'
        )
        return
      }
      throw err
    }

    contextualLogger.info('terminating sidecars', {
      containers: sidecarContainers.map((item) => item.name)
    })

    const promises = sidecarContainers.map((container) => {
      return this.shutdownContainer({
        namespace,
        name,
        containerName: container.name
      })
    })

    const results = await Promise.allSettled(promises)
    const failed = results.filter(isRejected)
    if (failed.length > 0) {
      contextualLogger.error(
        'some termination commands failed, pod may need manual cleanup',
        {
          reasons: failed.map((item) => item.reason?.message)
        }
      )
    }
  }

  @retry({ maxAttempts: 3, backoffDurationMs: 5_000 })
  private async shutdownContainer(
    this: TerminateSidecarsInJobsHandler,
    options: {
      namespace: string
      name: string
      containerName: string
    }
  ): Promise<void> {
    try {
      await this.kubernetes.exec({
        namespace: options.namespace,
        name: options.name,
        container: options.containerName,
        command: ['kill', '1']
      })
    } catch (ex) {
      if (ex.code === 404) {
        this.logger.info(
          'Went to shut down the container, but it appears the container has already been removed from the cluster',
          { containerName: options.containerName }
        )
        return
      }
      throw ex
    }
  }
}