feat: sysdig integration

Author: minsiyang

src/common/config.ts

@@ -45,6 +45,12 @@ config.EXCLUDED_NAMESPACES = loadExcludedNamespaces();
 config.WORKERS_COUNT = Number(config.WORKERS_COUNT) || 10;
 config.SKOPEO_COMPRESSION_LEVEL = Number(config.SKOPEO_COMPRESSION_LEVEL) || 6;
 
+// return Sysdig endpoint information
+if (config.SYSDIG_ENDPOINT && config.SYSDIG_TOKEN) {
+  config.SYSDIG_ENDPOINT = config.SYSDIG_ENDPOINT.trim();
+  config.SYSDIG_TOKEN = config.SYSDIG_TOKEN.trim();
+}
+
 /**
  * Important: we delete the following env vars because we don't want to proxy requests to the Kubernetes API server.
  * The Kubernetes client library would honor the NO/HTTP/HTTPS_PROXY env vars.

src/data-scraper/index.ts

@@ -0,0 +1,75 @@
+import { logger } from '../common/logger';
+import { config } from '../common/config';
+import { sendRuntimeData } from '../transmitter';
+import { constructRuntimeData } from '../transmitter/payload';
+import { retryRequest } from '../transmitter';
+import { IRuntimeImagesResponse } from '../transmitter/types';
+import { NeedleOptions } from 'needle';
+import { Agent as HttpsAgent } from 'https';
+
+function isSuccessStatusCode(statusCode: number | undefined): boolean {
+  return statusCode !== undefined && statusCode >= 200 && statusCode < 300;
+}
+
+export async function scrapeData(): Promise<void> {
+  const base: string = config.SYSDIG_ENDPOINT;
+  const header: string = `Bearer ${config.SYSDIG_TOKEN}`;
+
+  const url: string = base + '/v1/runtimeimages';
+  // limit: min 1, max 500, default 250
+  const limit: number = 10;
+  const reqOptions: NeedleOptions = {
+    agent: new HttpsAgent({
+      keepAlive: true,
+      // We agreed with Sysdig to skip TLS certificates validation for HTTPS connection.
+      rejectUnauthorized: false,
+    }),
+    headers: {
+      authorization: header,
+    },
+  };
+
+  let cursor: string = '';
+  while (true) {
+    try {
+      logger.info({ cursor }, 'attempting to get runtime images');
+
+      const { response, attempt } = await retryRequest(
+        'get',
+        `${url}?limit=${limit}&cursor=${cursor}`,
+        {},
+        reqOptions,
+      );
+      if (!isSuccessStatusCode(response.statusCode)) {
+        throw new Error(`${response.statusCode} ${response.statusMessage}`);
+      }
+
+      logger.info(
+        {
+          attempt,
+          cursor,
+        },
+        'runtime images received successfully',
+      );
+
+      const responseBody: IRuntimeImagesResponse = response.body;
+      const runtimeDataPayload = constructRuntimeData(responseBody.data);
+      logger.info({}, 'sending runtime data upstream');
+      await sendRuntimeData(runtimeDataPayload);
+
+      cursor = responseBody.page.next || '';
+      if (!cursor) {
+        break;
+      }
+    } catch (error) {
+      logger.error(
+        {
+          error,
+          cursor,
+        },
+        'could not get runtime images',
+      );
+      break;
+    }
+  }
+}

src/index.ts

@@ -10,6 +10,7 @@ import { beginWatchingWorkloads } from './supervisor/watchers';
 import { loadAndSendWorkloadEventsPolicy } from './common/policy';
 import { sendClusterMetadata } from './transmitter';
 import { setSnykMonitorAgentId } from './supervisor/agent';
+import { scrapeData } from './data-scraper';
 
 process.on('uncaughtException', (err) => {
   if (state.shutdownInProgress) {
@@ -68,4 +69,20 @@ setImmediate(async function setUpAndMonitor(): Promise<void> {
   await sendClusterMetadata();
   await loadAndSendWorkloadEventsPolicy();
   await monitor();
+
+  const interval: number = 4 * 60 * 60 * 1000; // 4 hours in milliseconds
+  if (config.SYSDIG_ENDPOINT && config.SYSDIG_TOKEN) {
+    setInterval(async () => {
+      try {
+        await scrapeData();
+      } catch (error) {
+        logger.error(
+          { error },
+          'an error occurred while scraping runtime data',
+        );
+      }
+    }, interval).unref();
+  } else {
+    logger.info({}, 'Sysdig integration not detected');
+  }
 });

src/transmitter/index.ts

@@ -1,4 +1,3 @@
-import { parse } from 'url';
 import { queue } from 'async';
 import needle from 'needle';
 import sleep from 'sleep-promise';
@@ -17,6 +16,7 @@ import {
   IDependencyGraphPayload,
   IWorkloadEventsPolicyPayload,
   IClusterMetadataPayload,
+  IRuntimeDataPayload,
 } from './types';
 import { getProxyAgent } from './proxy';
 
@@ -27,24 +27,29 @@ interface KubernetesUpstreamRequest {
     | IDependencyGraphPayload
     | ScanResultsPayload
     | IWorkloadMetadataPayload
-    | IDeleteWorkloadPayload;
+    | IDeleteWorkloadPayload
+    | IClusterMetadataPayload
+    | IRuntimeDataPayload;
 }
 
 const upstreamUrl =
   config.INTEGRATION_API || config.DEFAULT_KUBERNETES_UPSTREAM_URL;
 
-let agent = new HttpAgent({
+let httpAgent = new HttpAgent({
   keepAlive: true,
 });
 
-if (parse(upstreamUrl).protocol?.startsWith('https')) {
-  agent = new HttpsAgent({
-    keepAlive: true,
-  });
+let httpsAgent = new HttpsAgent({
+  keepAlive: true,
+});
+
+function getAgent(u: string): HttpAgent {
+  const url = new URL(u);
+  return url.protocol === 'https:' ? httpsAgent : httpAgent;
 }
 
 // Async queue wraps around the call to retryRequest in order to limit
-// the number of requests in flight to Homebase at any one time.
+// the number of requests in flight to kubernetes upstream at any one time.
 const reqQueue = queue(async function (req: KubernetesUpstreamRequest) {
   return await retryRequest(req.method, req.url, req.payload);
 }, config.REQUEST_QUEUE_LENGTH);
@@ -60,7 +65,7 @@ export async function sendDepGraph(
       const request: KubernetesUpstreamRequest = {
         method: 'post',
         url: `${upstreamUrl}/api/v1/dependency-graph`,
-        payload: payload,
+        payload,
       };
 
       const { response, attempt } = await reqQueue.pushAsync(request);
@@ -91,7 +96,7 @@ export async function sendScanResults(
       const request: KubernetesUpstreamRequest = {
         method: 'post',
         url: `${upstreamUrl}/api/v1/scan-results`,
-        payload: payload,
+        payload,
       };
 
       const { response, attempt } = await reqQueue.pushAsync(request);
@@ -127,7 +132,7 @@ export async function sendWorkloadMetadata(
     const request: KubernetesUpstreamRequest = {
       method: 'post',
       url: `${upstreamUrl}/api/v1/workload`,
-      payload: payload,
+      payload,
     };
 
     const { response, attempt } = await reqQueue.pushAsync(request);
@@ -198,7 +203,7 @@ export async function deleteWorkload(
     const request: KubernetesUpstreamRequest = {
       method: 'delete',
       url: `${upstreamUrl}/api/v1/workload`,
-      payload: payload,
+      payload,
     };
 
     const { response, attempt } = await reqQueue.pushAsync(request);
@@ -229,10 +234,11 @@ function isSuccessStatusCode(statusCode: number | undefined): boolean {
   return statusCode !== undefined && statusCode > 100 && statusCode < 400;
 }
 
-async function retryRequest(
+export async function retryRequest(
   verb: NeedleHttpVerbs,
   url: string,
   payload: object,
+  reqOptions: NeedleOptions = {},
 ): Promise<IResponseWithAttempts> {
   const retry = {
     attempts: 3,
@@ -241,7 +247,8 @@ async function retryRequest(
   const options: NeedleOptions = {
     json: true,
     compressed: true,
-    agent,
+    agent: getAgent(url),
+    ...reqOptions,
   };
 
   if (config.HTTP_PROXY || config.HTTPS_PROXY) {
@@ -317,11 +324,13 @@ export async function sendClusterMetadata(): Promise<void> {
       'attempting to send cluster metadata',
     );
 
-    const { response, attempt } = await retryRequest(
-      'post',
-      `${upstreamUrl}/api/v1/cluster`,
+    const request: KubernetesUpstreamRequest = {
+      method: 'post',
+      url: `${upstreamUrl}/api/v1/cluster`,
       payload,
-    );
+    };
+
+    const { response, attempt } = await reqQueue.pushAsync(request);
     if (!isSuccessStatusCode(response.statusCode)) {
       throw new Error(`${response.statusCode} ${response.statusMessage}`);
     }
@@ -347,3 +356,46 @@ export async function sendClusterMetadata(): Promise<void> {
     );
   }
 }
+
+export async function sendRuntimeData(
+  payload: IRuntimeDataPayload,
+): Promise<void> {
+  const logContext = {
+    userLocator: payload.target.userLocator,
+    cluster: payload.target.cluster,
+    agentId: payload.target.agentId,
+    identity: payload.identity,
+  };
+
+  try {
+    logger.info(logContext, 'attempting to send runtime data');
+
+    const request: KubernetesUpstreamRequest = {
+      method: 'post',
+      url: `${upstreamUrl}/api/v1/runtime-results`,
+      payload,
+    };
+
+    const { response, attempt } = await reqQueue.pushAsync(request);
+
+    if (!isSuccessStatusCode(response.statusCode)) {
+      throw new Error(`${response.statusCode} ${response.statusMessage}`);
+    }
+
+    logger.info(
+      {
+        attempt,
+        ...logContext,
+      },
+      'runtime data sent upstream successfully',
+    );
+  } catch (error) {
+    logger.error(
+      {
+        error,
+        ...logContext,
+      },
+      'could not send runtime data',
+    );
+  }
+}

src/transmitter/payload.ts

@@ -13,8 +13,13 @@ import {
   IDependencyGraphPayload,
   IWorkloadEventsPolicyPayload,
   Telemetry,
+  IRuntimeDataPayload,
+  IRuntimeDataFact,
+  IRuntimeImage,
 } from './types';
 import { state } from '../state';
+import { isExcludedNamespace } from '../supervisor/watchers/internal-namespaces';
+import { logger } from '../common/logger';
 
 export function constructDepGraph(
   scannedImages: IScanResult[],
@@ -134,3 +139,49 @@ export function constructWorkloadEventsPolicy(
     agentId: config.AGENT_ID,
   };
 }
+
+const workloadKindMap = {
+  deployment: 'Deployment',
+  replicaset: 'ReplicaSet',
+  statefulset: 'StatefulSet',
+  daemonset: 'DaemonSet',
+  job: 'Job',
+  cronjob: 'CronJob',
+  replicationcontroller: 'ReplicationController',
+  deploymentconfig: 'DeploymentConfig',
+  pod: 'Pod',
+};
+export function constructRuntimeData(
+  runtimeResults: IRuntimeImage[],
+): IRuntimeDataPayload {
+  const filteredRuntimeResults = runtimeResults.reduce((acc, runtimeResult) => {
+    if (!isExcludedNamespace(runtimeResult.namespace)) {
+      const mappedWorkloadKind =
+        workloadKindMap[runtimeResult.workloadKind.toLowerCase()];
+      if (mappedWorkloadKind) {
+        runtimeResult.workloadKind = mappedWorkloadKind;
+        acc.push(runtimeResult);
+      } else {
+        logger.error({ runtimeResult }, 'invalid Sysdig workload kind');
+      }
+    }
+    return acc;
+  }, [] as IRuntimeImage[]);
+
+  const dataFact: IRuntimeDataFact = {
+    type: 'loadedPackages',
+    data: filteredRuntimeResults,
+  };
+
+  return {
+    identity: {
+      type: 'sysdig',
+    },
+    target: {
+      agentId: config.AGENT_ID,
+      userLocator: config.INTEGRATION_ID,
+      cluster: currentClusterName,
+    },
+    facts: [dataFact],
+  };
+}

src/transmitter/proxy.ts

@@ -1,4 +1,3 @@
-import { URL } from 'url';
 import { httpOverHttp, httpsOverHttp } from 'tunnel';
 import { Agent, globalAgent as HttpAgent } from 'http';
 import { globalAgent as HttpsAgent } from 'https';