Merge pull request #996 from snyk/fix/cronjob

[RUN-1518] fix: support both v1 and v1beta1 CronJobs

Author: KatieArmstr

src/supervisor/types.ts

@@ -2,6 +2,7 @@ import { IncomingMessage } from 'http';
 import {
   AppsV1Api,
   BatchV1Api,
+  BatchV1beta1Api,
   CoreV1Api,
   CustomObjectsApi,
   KubeConfig,
@@ -16,7 +17,10 @@ export enum WorkloadKind {
   StatefulSet = 'StatefulSet',
   DaemonSet = 'DaemonSet',
   Job = 'Job',
+  /** Available since Kubernetes 1.20. */
   CronJob = 'CronJob',
+  /** @deprecated Will be removed in Kubernetes 1.25. */
+  CronJobV1Beta1 = 'CronJobV1Beta1',
   ReplicationController = 'ReplicationController',
   Pod = 'Pod',
   DeploymentConfig = 'DeploymentConfig',
@@ -40,20 +44,23 @@ export interface IK8sClients {
   readonly appsClient: AppsV1Api;
   readonly coreClient: CoreV1Api;
   readonly batchClient: BatchV1Api;
+  readonly batchUnstableClient: BatchV1beta1Api;
   readonly customObjectsClient: CustomObjectsApi;
 }
 
 export class K8sClients implements IK8sClients {
   public readonly appsClient: AppsV1Api;
   public readonly coreClient: CoreV1Api;
   public readonly batchClient: BatchV1Api;
+  public readonly batchUnstableClient: BatchV1beta1Api;
   /** This client is used to access Custom Resources in the cluster, e.g. DeploymentConfig on OpenShift. */
   public readonly customObjectsClient: CustomObjectsApi;
 
   constructor(config: KubeConfig) {
     this.appsClient = config.makeApiClient(AppsV1Api);
     this.coreClient = config.makeApiClient(CoreV1Api);
     this.batchClient = config.makeApiClient(BatchV1Api);
+    this.batchUnstableClient = config.makeApiClient(BatchV1beta1Api);
     this.customObjectsClient = config.makeApiClient(CustomObjectsApi);
   }
 }

src/supervisor/watchers/handlers/cron-job.ts

@@ -1,4 +1,9 @@
-import { V1CronJob, V1CronJobList } from '@kubernetes/client-node';
+import {
+  V1CronJob,
+  V1CronJobList,
+  V1beta1CronJob,
+  V1beta1CronJobList,
+} from '@kubernetes/client-node';
 import { deleteWorkload, trimWorkload } from './workload';
 import { WorkloadKind } from '../../types';
 import { FALSY_WORKLOAD_NAME_MARKER } from './types';
@@ -27,7 +32,27 @@ export async function paginatedCronJobList(namespace: string): Promise<{
   );
 }
 
-export async function cronJobWatchHandler(cronJob: V1CronJob): Promise<void> {
+export async function paginatedCronJobV1Beta1List(namespace: string): Promise<{
+  response: IncomingMessage;
+  body: V1beta1CronJobList;
+}> {
+  const cronJobList = new V1beta1CronJobList();
+  cronJobList.apiVersion = 'batch/v1beta1';
+  cronJobList.kind = 'CronJobList';
+  cronJobList.items = new Array<V1beta1CronJob>();
+
+  return await paginatedList(
+    namespace,
+    cronJobList,
+    k8sApi.batchUnstableClient.listNamespacedCronJob.bind(
+      k8sApi.batchUnstableClient,
+    ),
+  );
+}
+
+export async function cronJobWatchHandler(
+  cronJob: V1CronJob | V1beta1CronJob,
+): Promise<void> {
   cronJob = trimWorkload(cronJob);
 
   if (

src/supervisor/watchers/handlers/index.ts

@@ -5,12 +5,18 @@ import {
   ERROR,
   UPDATE,
   KubernetesObject,
+  BatchV1beta1Api,
+  BatchV1Api,
 } from '@kubernetes/client-node';
 
 import { logger } from '../../../common/logger';
 import { WorkloadKind } from '../../types';
 import { podWatchHandler, podDeletedHandler, paginatedPodList } from './pod';
-import { cronJobWatchHandler, paginatedCronJobList } from './cron-job';
+import {
+  cronJobWatchHandler,
+  paginatedCronJobList,
+  paginatedCronJobV1Beta1List,
+} from './cron-job';
 import { daemonSetWatchHandler, paginatedDaemonSetList } from './daemon-set';
 import { deploymentWatchHandler, paginatedDeploymentList } from './deployment';
 import { jobWatchHandler, paginatedJobList } from './job';
@@ -75,6 +81,13 @@ const workloadWatchMetadata: Readonly<IWorkloadWatchMetadata> = {
     },
     listFactory: (namespace) => () => paginatedCronJobList(namespace),
   },
+  [WorkloadKind.CronJobV1Beta1]: {
+    endpoint: '/apis/batch/v1beta1/watch/namespaces/{namespace}/cronjobs',
+    handlers: {
+      [DELETE]: cronJobWatchHandler,
+    },
+    listFactory: (namespace) => () => paginatedCronJobV1Beta1List(namespace),
+  },
   [WorkloadKind.Job]: {
     endpoint: '/apis/batch/v1/watch/namespaces/{namespace}/jobs',
     handlers: {
@@ -125,10 +138,75 @@ async function isSupportedWorkload(
   namespace: string,
   workloadKind: WorkloadKind,
 ): Promise<boolean> {
-  if (workloadKind !== WorkloadKind.DeploymentConfig) {
-    return true;
+  switch (workloadKind) {
+    case WorkloadKind.DeploymentConfig:
+      return await isDeploymentConfigSupported(namespace);
+    case WorkloadKind.CronJobV1Beta1:
+      return await isCronJobVersionSupported(
+        workloadKind,
+        namespace,
+        k8sApi.batchUnstableClient,
+      );
+    case WorkloadKind.CronJob:
+      return await isCronJobVersionSupported(
+        workloadKind,
+        namespace,
+        k8sApi.batchClient,
+      );
+    default:
+      return true;
   }
+}
 
+async function isCronJobVersionSupported(
+  workloadKind: WorkloadKind,
+  namespace: string,
+  client: BatchV1Api | BatchV1beta1Api,
+): Promise<boolean> {
+  try {
+    const pretty = undefined;
+    const allowWatchBookmarks = undefined;
+    const continueToken = undefined;
+    const fieldSelector = undefined;
+    const labelSelector = undefined;
+    const limit = 1; // Try to grab only a single object
+    const resourceVersion = undefined; // List anything in the cluster
+    const resourceVersionMatch = undefined;
+    const timeoutSeconds = 10; // Don't block the snyk-monitor indefinitely
+    const attemptedApiCall =
+      await kubernetesApiWrappers.retryKubernetesApiRequest(() =>
+        client.listNamespacedCronJob(
+          namespace,
+          pretty,
+          allowWatchBookmarks,
+          continueToken,
+          fieldSelector,
+          labelSelector,
+          limit,
+          resourceVersion,
+          resourceVersionMatch,
+          timeoutSeconds,
+        ),
+      );
+    return (
+      attemptedApiCall !== undefined &&
+      attemptedApiCall.response !== undefined &&
+      attemptedApiCall.response.statusCode !== undefined &&
+      attemptedApiCall.response.statusCode >= 200 &&
+      attemptedApiCall.response.statusCode < 300
+    );
+  } catch (error) {
+    logger.debug(
+      { error, workloadKind: workloadKind },
+      'Failed on Kubernetes API call to list CronJob or v1beta1 CronJob',
+    );
+    return false;
+  }
+}
+
+async function isDeploymentConfigSupported(
+  namespace: string,
+): Promise<boolean> {
   try {
     const pretty = undefined;
     const continueToken = undefined;
@@ -162,7 +240,7 @@ async function isSupportedWorkload(
     );
   } catch (error) {
     logger.debug(
-      { error, workloadKind },
+      { error, workloadKind: WorkloadKind.DeploymentConfig },
       'Failed on Kubernetes API call to list DeploymentConfig',
     );
     return false;

src/supervisor/workload-reader.ts

@@ -208,9 +208,19 @@ const jobReader: IWorkloadReaderFunc = async (workloadName, namespace) => {
 
 // cronJobReader can read v1 and v1beta1 CronJobs
 const cronJobReader: IWorkloadReaderFunc = async (workloadName, namespace) => {
-  const cronJobResult = await kubernetesApiWrappers.retryKubernetesApiRequest(
-    () => k8sApi.batchClient.readNamespacedCronJob(workloadName, namespace),
-  );
+  const cronJobResult = await kubernetesApiWrappers
+    .retryKubernetesApiRequest(() =>
+      k8sApi.batchClient.readNamespacedCronJob(workloadName, namespace),
+    )
+    // In case the V1 client fails, try using the V1Beta1 client.
+    .catch(() =>
+      kubernetesApiWrappers.retryKubernetesApiRequest(() =>
+        k8sApi.batchUnstableClient.readNamespacedCronJob(
+          workloadName,
+          namespace,
+        ),
+      ),
+    );
   const cronJob = trimWorkload(cronJobResult.body);
 
   if (
@@ -280,13 +290,17 @@ function logIncompleteWorkload(workloadName: string, namespace: string): void {
 // Here we are using the "kind" property of a k8s object as a key to map it to a reader.
 // This gives us a quick look up table where we can abstract away the internal implementation of reading a resource
 // and just grab a generic handler/reader that does that for us (based on the "kind").
-const workloadReader = {
+const workloadReader: Record<string, IWorkloadReaderFunc> = {
   [WorkloadKind.Deployment]: deploymentReader,
   [WorkloadKind.ReplicaSet]: replicaSetReader,
   [WorkloadKind.StatefulSet]: statefulSetReader,
   [WorkloadKind.DaemonSet]: daemonSetReader,
   [WorkloadKind.Job]: jobReader,
   [WorkloadKind.CronJob]: cronJobReader,
+  // ------------
+  // Note: WorkloadKind.CronJobV1Beta1 is intentionally not listed here.
+  // The WorkloadKind.CronJob reader can handle both v1 and v1beta1 API versions.
+  // ------------
   [WorkloadKind.ReplicationController]: replicationControllerReader,
   [WorkloadKind.DeploymentConfig]: deploymentConfigReader,
 };

test/helpers/kubernetes-upstream.ts

@@ -13,7 +13,7 @@ import { config } from '../../src/common/config';
 
 const UPSTREAM_POLLING_CONFIGURATION = {
   WAIT_BETWEEN_REQUESTS_MS: 5000,
-  MAXIMUM_REQUESTS: 120,
+  MAXIMUM_REQUESTS: 180,
 };
 
 export async function getUpstreamResponseBody(
@@ -73,9 +73,11 @@ export async function validateUpstreamStoredMetadata(
   remainingChecks: number = UPSTREAM_POLLING_CONFIGURATION.MAXIMUM_REQUESTS,
 ): Promise<boolean> {
   while (remainingChecks > 0) {
-    console.log(
-      `Pinging upstream for existing metadata (${remainingChecks} checks remaining)...`,
-    );
+    if (remainingChecks % 10 === 0) {
+      console.log(
+        `Pinging upstream for existing metadata (${remainingChecks} checks remaining)...`,
+      );
+    }
     const responseBody = await getUpstreamResponseBody(relativeUrl);
     const workloadInfo: IWorkloadMetadata | undefined =
       responseBody.workloadInfo;

test/integration/kubernetes.spec.ts

@@ -14,6 +14,7 @@ import {
 } from '../helpers/kubernetes-upstream';
 import * as kubectl from '../helpers/kubectl';
 import { execWrapper as exec } from '../helpers/exec';
+import { IWorkloadLocator } from '../../src/transmitter/types';
 
 let integrationId: string;
 let namespace: string;
@@ -37,6 +38,19 @@ test('deploy snyk-monitor', async () => {
   integrationId = await setup.deployMonitor();
 });
 
+const cronJobValidator = (workloads: IWorkloadLocator[]) =>
+  workloads.find(
+    (workload) =>
+      workload.name === 'cron-job' && workload.type === WorkloadKind.CronJob,
+  ) !== undefined;
+
+const cronJobV1Beta1Validator = (workloads: IWorkloadLocator[]) =>
+  workloads.find(
+    (workload) =>
+      workload.name === 'cron-job-v1beta1' &&
+      workload.type === WorkloadKind.CronJob,
+  ) !== undefined;
+
 // Next we apply some sample workloads
 test('deploy sample workloads', async () => {
   const servicesNamespace = 'services';
@@ -50,8 +64,14 @@ test('deploy sample workloads', async () => {
     kubectl.applyK8sYaml('./test/fixtures/centos-deployment.yaml'),
     kubectl.applyK8sYaml('./test/fixtures/scratch-deployment.yaml'),
     kubectl.applyK8sYaml('./test/fixtures/consul-deployment.yaml'),
-    kubectl.applyK8sYaml('./test/fixtures/cronjob.yaml'),
-    kubectl.applyK8sYaml('./test/fixtures/cronjob-v1beta1.yaml'),
+    kubectl.applyK8sYaml('./test/fixtures/cronjob.yaml').catch((error) => {
+      console.log('CronJob is possibly unsupported', error);
+    }),
+    kubectl
+      .applyK8sYaml('./test/fixtures/cronjob-v1beta1.yaml')
+      .catch((error) => {
+        console.log('CronJobV1Beta1 is possibly unsupported', error);
+      }),
     kubectl.createPodFromImage(
       'alpine-from-sha',
       someImageWithSha,
@@ -152,16 +172,8 @@ test('snyk-monitor sends data to kubernetes-upstream', async () => {
           workload.name === 'consul' &&
           workload.type === WorkloadKind.Deployment,
       ) !== undefined &&
-      workloads.find(
-        (workload) =>
-          workload.name === 'cron-job' &&
-          workload.type === WorkloadKind.CronJob,
-      ) !== undefined &&
-      workloads.find(
-        (workload) =>
-          workload.name === 'cron-job-v1beta1' &&
-          workload.type === WorkloadKind.CronJob,
-      ) !== undefined
+      // only one of the cronjob versions needs to be valid
+      (cronJobValidator(workloads) || cronJobV1Beta1Validator(workloads))
     );
   };