Merge pull request #779 from snyk/feat/send_cluster_metadata_on_startup

[RUN-1560] Send cluster metadata on start up

Author: KatieArmstr

src/index.ts

@@ -8,6 +8,7 @@ import { logger } from './common/logger';
 import { currentClusterName } from './supervisor/cluster';
 import { beginWatchingWorkloads } from './supervisor/watchers';
 import { loadAndSendWorkloadEventsPolicy } from './common/policy';
+import { sendClusterMetadata } from './transmitter';
 
 process.on('uncaughtException', (err) => {
   if (state.shutdownInProgress) {
@@ -62,6 +63,7 @@ cleanUpTempStorage();
 
 // Allow running in an async context
 setImmediate(async function setUpAndMonitor(): Promise<void> {
+  await sendClusterMetadata();
   await loadAndSendWorkloadEventsPolicy();
   await monitor();
 });

src/transmitter/index.ts

@@ -15,7 +15,8 @@ import {
   IRequestError,
   ScanResultsPayload,
   IDependencyGraphPayload,
-  WorkloadEventsPolicyPayload,
+  IWorkloadEventsPolicyPayload,
+  IClusterMetadataPayload,
 } from './types';
 import { getProxyAgent } from './proxy';
 
@@ -85,8 +86,7 @@ export async function sendScanResults(
 ): Promise<boolean> {
   for (const payload of payloads) {
     // Intentionally removing scan results as they would be too big to log
-    // eslint-disable-next-line @typescript-eslint/no-unused-vars
-    const { scanResults, ...payloadWithoutScanResults } = payload;
+    const payloadWithoutScanResults = { ...payload, scanResults: undefined };
     try {
       const request: HomebaseRequest = {
         method: 'post',
@@ -148,7 +148,7 @@ export async function sendWorkloadMetadata(
 }
 
 export async function sendWorkloadEventsPolicy(
-  payload: WorkloadEventsPolicyPayload,
+  payload: IWorkloadEventsPolicyPayload,
 ): Promise<void> {
   try {
     logger.info(
@@ -296,3 +296,53 @@ function shouldRetryRequest(
 
   return false;
 }
+
+export async function sendClusterMetadata(): Promise<void> {
+  const payload: IClusterMetadataPayload = {
+    userLocator: config.INTEGRATION_ID,
+    cluster: config.CLUSTER_NAME,
+    agentId: config.AGENT_ID,
+    version: config.MONITOR_VERSION,
+    namespace: config.NAMESPACE,
+  };
+
+  try {
+    logger.info(
+      {
+        userLocator: payload.userLocator,
+        cluster: payload.cluster,
+        agentId: payload.agentId,
+      },
+      'attempting to send cluster metadata',
+    );
+
+    const { response, attempt } = await retryRequest(
+      'post',
+      `${upstreamUrl}/api/v1/cluster`,
+      payload,
+    );
+    if (!isSuccessStatusCode(response.statusCode)) {
+      throw new Error(`${response.statusCode} ${response.statusMessage}`);
+    }
+
+    logger.info(
+      {
+        userLocator: payload.userLocator,
+        cluster: payload.cluster,
+        agentId: payload.agentId,
+        attempt,
+      },
+      'cluster metadata sent upstream successfully',
+    );
+  } catch (error) {
+    logger.error(
+      {
+        error,
+        userLocator: payload.userLocator,
+        cluster: payload.cluster,
+        agentId: payload.agentId,
+      },
+      'could not send cluster metadata',
+    );
+  }
+}

src/transmitter/payload.ts

@@ -12,7 +12,7 @@ import {
   IKubernetesMonitorMetadata,
   ScanResultsPayload,
   IDependencyGraphPayload,
-  WorkloadEventsPolicyPayload,
+  IWorkloadEventsPolicyPayload,
 } from './types';
 
 export function constructDepGraph(
@@ -135,7 +135,7 @@ export function constructDeleteWorkload(
 
 export function constructWorkloadEventsPolicy(
   policy: string,
-): WorkloadEventsPolicyPayload {
+): IWorkloadEventsPolicyPayload {
   return {
     policy,
     userLocator: config.INTEGRATION_ID,

src/transmitter/types.ts

@@ -63,13 +63,21 @@ export interface IDeleteWorkloadPayload {
   agentId: string;
 }
 
-export interface WorkloadEventsPolicyPayload {
+export interface IWorkloadEventsPolicyPayload {
   userLocator: string;
   cluster: string;
   agentId: string;
   policy: string;
 }
 
+export interface IClusterMetadataPayload {
+  userLocator: string;
+  cluster: string;
+  agentId: string;
+  version: string;
+  namespace?: string;
+}
+
 export interface IWorkload {
   type: string;
   name: string;

test/system/kind.spec.ts

@@ -97,11 +97,11 @@ test('Kubernetes-Monitor with KinD', async (jestDoneCallback) => {
     .times(1)
     .reply(
       200,
-      (uri, requestBody: transmitterTypes.WorkloadEventsPolicyPayload) => {
+      (uri, requestBody: transmitterTypes.IWorkloadEventsPolicyPayload) => {
         try {
           expect(
             requestBody,
-          ).toEqual<transmitterTypes.WorkloadEventsPolicyPayload>({
+          ).toEqual<transmitterTypes.IWorkloadEventsPolicyPayload>({
             agentId: expect.any(String),
             cluster: expect.any(String),
             userLocator: expect.any(String),
@@ -113,6 +113,28 @@ test('Kubernetes-Monitor with KinD', async (jestDoneCallback) => {
       },
     );
 
+  nock('https://kubernetes-upstream.snyk.io')
+    .post('/api/v1/cluster')
+    .times(1)
+    .reply(
+      200,
+      (uri, requestBody: transmitterTypes.IClusterMetadataPayload) => {
+        try {
+          expect(requestBody).toEqual<
+            Partial<transmitterTypes.IClusterMetadataPayload>
+          >({
+            agentId: expect.any(String),
+            cluster: expect.any(String),
+            userLocator: expect.any(String),
+            // also should have version here but due to test limitation it is undefined
+            // as it is injected as an environment variable via the Helm chart
+          });
+        } catch (error) {
+          jestDoneCallback(error);
+        }
+      },
+    );
+
   // Setup nocks
   nock(/https\:\/\/127\.0\.0\.1\:\d+/, { allowUnmocked: true })
     .get('/api/v1/namespaces')