feat: Experimental feature - Duplicate workflow job event to extra queue

examples/default/main.tf

@@ -13,7 +13,9 @@ resource "random_id" "random" {
 ################################################################################
 
 module "runners" {
-  source                          = "../../"
+  source = "../../"
+  # enable this flag to publish webhook events to secondary queue
+  # webhook_events_secondary_queue  = true
   create_service_linked_role_spot = true
   aws_region                      = local.aws_region
   vpc_id                          = module.vpc.vpc_id

main.tf

@@ -50,6 +50,12 @@ resource "aws_sqs_queue_policy" "build_queue_policy" {
   policy    = data.aws_iam_policy_document.deny_unsecure_transport.json
 }
 
+resource "aws_sqs_queue_policy" "monitored_build_events_policy" {
+  count     = var.webhook_events_secondary_queue ? 1 : 0
+  queue_url = aws_sqs_queue.monitored_build_events[0].id
+  policy    = data.aws_iam_policy_document.deny_unsecure_transport.json
+}
+
 resource "aws_sqs_queue" "queued_builds" {
   name                        = "${var.prefix}-queued-builds${var.fifo_build_queue ? ".fifo" : ""}"
   delay_seconds               = var.delay_webhook_event
@@ -66,6 +72,19 @@ resource "aws_sqs_queue" "queued_builds" {
   tags = var.tags
 }
 
+resource "aws_sqs_queue" "monitored_build_events" {
+  count                       = var.webhook_events_secondary_queue ? 1 : 0
+  name                        = "${var.prefix}-monitored-build-events"
+  delay_seconds               = 0
+  visibility_timeout_seconds  = var.runners_scale_up_lambda_timeout
+  message_retention_seconds   = var.job_queue_retention_in_seconds
+  fifo_queue                  = false
+  receive_wait_time_seconds   = 0
+  content_based_deduplication = false
+  redrive_policy              = null
+  tags                        = var.tags
+}
+
 
 resource "aws_sqs_queue_policy" "build_queue_dlq_policy" {
   count     = var.redrive_build_queue.enabled ? 1 : 0
@@ -92,14 +111,15 @@ module "ssm" {
 module "webhook" {
   source = "./modules/webhook"
 
-  aws_region  = var.aws_region
-  prefix      = var.prefix
-  tags        = local.tags
-  kms_key_arn = var.kms_key_arn
-
-  sqs_build_queue               = aws_sqs_queue.queued_builds
-  sqs_build_queue_fifo          = var.fifo_build_queue
-  github_app_webhook_secret_arn = module.ssm.parameters.github_app_webhook_secret.arn
+  aws_region                     = var.aws_region
+  prefix                         = var.prefix
+  tags                           = local.tags
+  kms_key_arn                    = var.kms_key_arn
+  webhook_events_secondary_queue = var.webhook_events_secondary_queue
+  sqs_build_queue                = aws_sqs_queue.queued_builds
+  sqs_build_queue_fifo           = var.fifo_build_queue
+  sqs_monitored_build_events     = length(aws_sqs_queue.monitored_build_events) > 0 ? aws_sqs_queue.monitored_build_events[0] : null
+  github_app_webhook_secret_arn  = module.ssm.parameters.github_app_webhook_secret.arn
 
   lambda_s3_bucket                 = var.lambda_s3_bucket
   webhook_lambda_s3_key            = var.webhook_lambda_s3_key

modules/runners/variables.tf

@@ -560,4 +560,4 @@ variable "lambda_architecture" {
     condition     = contains(["arm64", "x86_64"], var.lambda_architecture)
     error_message = "`lambda_architecture` value is not valid, valid values are: `arm64` and `x86_64`."
   }
-}
+}

modules/webhook/lambdas/webhook/src/sqs/index.test.ts

@@ -1,6 +1,6 @@
 import { SQS } from 'aws-sdk';
 
-import { ActionRequestMessage, sendActionRequest } from '.';
+import { ActionRequestMessage, GithubWorkflowEvent, sendActionRequest, sendMonitorGHWorkflowEvent } from '.';
 
 const mockSQS = {
   sendMessage: jest.fn(() => {
@@ -64,3 +64,27 @@ describe('Test sending message to SQS.', () => {
     expect(result).resolves;
   });
 });
+
+describe('Test sending message to monitoring SQS.', () => {
+  const message: GithubWorkflowEvent = {
+    eventType: 'type',
+    id: 0,
+    jobEvent: {},
+  };
+  const sqsMessage: SQS.Types.SendMessageRequest = {
+    QueueUrl: 'https://sqs.eu-west-1.amazonaws.com/123456789/monitored-build-events',
+    MessageBody: JSON.stringify(message),
+  };
+
+  it('based on defaults', async () => {
+    // Arrange
+    process.env.SQS_MONITORED_BUILD_EVENTS = sqsMessage.QueueUrl;
+
+    // Act
+    const result = await sendMonitorGHWorkflowEvent(message);
+
+    // Assert
+    expect(mockSQS.sendMessage).toBeCalledWith(sqsMessage);
+    expect(result).resolves;
+  });
+});

modules/webhook/lambdas/webhook/src/sqs/index.ts

@@ -9,6 +9,11 @@ export interface ActionRequestMessage {
   repositoryOwner: string;
   installationId: number;
 }
+export interface GithubWorkflowEvent {
+  id: number;
+  eventType: string;
+  jobEvent: any;
+}
 
 export const sendActionRequest = async (message: ActionRequestMessage): Promise<void> => {
   const sqs = new SQS({ region: process.env.AWS_REGION });
@@ -28,3 +33,15 @@ export const sendActionRequest = async (message: ActionRequestMessage): Promise<
 
   await sqs.sendMessage(sqsMessage).promise();
 };
+
+export const sendMonitorGHWorkflowEvent = async (message: GithubWorkflowEvent): Promise<void> => {
+  const sqs = new SQS({ region: process.env.AWS_REGION });
+
+  const sqsMessage: SQS.Types.SendMessageRequest = {
+    QueueUrl: String(process.env.SQS_MONITORED_BUILD_EVENTS),
+    MessageBody: JSON.stringify(message),
+  };
+
+  logger.debug(`sending message to monitoring SQS: ${JSON.stringify(sqsMessage)}`, LogFields.print());
+  await sqs.sendMessage(sqsMessage).promise();
+};

modules/webhook/lambdas/webhook/src/webhook/handler.test.ts

@@ -4,7 +4,7 @@ import nock from 'nock';
 
 import checkrun_event from '../../test/resources/github_check_run_event.json';
 import workflowjob_event from '../../test/resources/github_workflowjob_event.json';
-import { sendActionRequest } from '../sqs';
+import { sendActionRequest, sendMonitorGHWorkflowEvent } from '../sqs';
 import { getParameterValue } from '../ssm';
 import { handle } from './handler';
 
@@ -299,4 +299,29 @@ describe('handler', () => {
       expect(sendActionRequest).toBeCalled();
     });
   });
+
+  describe('Test for monitoring github action events: ', () => {
+    beforeEach(() => {
+      process.env.SQS_MONITORED_BUILD_EVENTS = 'https://sqs.eu-west-1.amazonaws.com/123456789/monitored-build-events';
+    });
+    it('sends monitoring events', async () => {
+      const event = JSON.stringify(workflowjob_event);
+      const resp = await handle(
+        { 'X-Hub-Signature': await webhooks.sign(event), 'X-GitHub-Event': 'workflow_job' },
+        event,
+      );
+      expect(resp.statusCode).toBe(201);
+      expect(sendMonitorGHWorkflowEvent).toBeCalled();
+    });
+    it('Doesnt sends monitoring events', async () => {
+      process.env.SQS_MONITORED_BUILD_EVENTS = '';
+      const event = JSON.stringify(workflowjob_event);
+      const resp = await handle(
+        { 'X-Hub-Signature': await webhooks.sign(event), 'X-GitHub-Event': 'workflow_job' },
+        event,
+      );
+      expect(resp.statusCode).toBe(201);
+      expect(sendMonitorGHWorkflowEvent).not.toBeCalled();
+    });
+  });
 });

modules/webhook/lambdas/webhook/src/webhook/handler.ts

@@ -3,7 +3,7 @@ import { CheckRunEvent, WorkflowJobEvent } from '@octokit/webhooks-types';
 import { IncomingHttpHeaders } from 'http';
 
 import { Response } from '../lambda';
-import { sendActionRequest } from '../sqs';
+import { sendActionRequest, sendMonitorGHWorkflowEvent } from '../sqs';
 import { getParameterValue } from '../ssm';
 import { LogFields, logger as rootLogger } from './logger';
 
@@ -73,10 +73,26 @@ export async function handle(headers: IncomingHttpHeaders, body: string): Promis
   } else if (githubEvent == 'check_run') {
     response = await handleCheckRun(payload as CheckRunEvent, githubEvent);
   }
-
+  await monitorWorkflowEvents(githubEvent, payload);
   return response;
 }
 
+async function monitorWorkflowEvents(githubEvent: string, payload: any) {
+  const webhook_events_secondary_queue = process.env.SQS_MONITORED_BUILD_EVENTS || 'false';
+  logger.debug('Webhook events secondary queue name: ', webhook_events_secondary_queue);
+  if (webhook_events_secondary_queue != 'false') {
+    if (githubEvent == 'workflow_job') {
+      let workflowEventPayload = payload as WorkflowJobEvent;
+      logger.debug('Sending Webhook events to the secondary queue: ', webhook_events_secondary_queue);
+      await sendMonitorGHWorkflowEvent({
+        id: workflowEventPayload.workflow_job.id,
+        eventType: githubEvent,
+        jobEvent: workflowEventPayload,
+      });
+    }
+  }
+}
+
 function readEnvironmentVariables() {
   const environment = process.env.ENVIRONMENT;
   const enableWorkflowLabelCheckEnv = process.env.ENABLE_WORKFLOW_JOB_LABELS_CHECK || 'false';

modules/webhook/variables.tf

@@ -37,7 +37,13 @@ variable "sqs_build_queue" {
     arn = string
   })
 }
-
+variable "sqs_monitored_build_events" {
+  description = "SQS queue to monitor github events."
+  type = object({
+    id  = string
+    arn = string
+  })
+}
 variable "lambda_zip" {
   description = "File location of the lambda zip file."
   type        = string
@@ -177,4 +183,10 @@ variable "lambda_architecture" {
     condition     = contains(["arm64", "x86_64"], var.lambda_architecture)
     error_message = "`lambda_architecture` value is not valid, valid values are: `arm64` and `x86_64`."
   }
-}
+}
+
+variable "webhook_events_secondary_queue" {
+  description = "Enabling this feature will create a secondory sqs queue to wich a copy of the event will be delivered."
+  type        = bool
+  default     = false
+}

modules/webhook/webhook.tf

@@ -22,6 +22,7 @@ resource "aws_lambda_function" "webhook" {
       RUNNER_LABELS                    = jsonencode(split(",", var.runner_labels))
       SQS_URL_WEBHOOK                  = var.sqs_build_queue.id
       SQS_IS_FIFO                      = var.sqs_build_queue_fifo
+      SQS_MONITORED_BUILD_EVENTS       = try(var.sqs_monitored_build_events, null) != null ? var.sqs_monitored_build_events.id : ""
     }
   }
 
@@ -78,6 +79,15 @@ resource "aws_iam_role_policy" "webhook_sqs" {
     sqs_resource_arn = var.sqs_build_queue.arn
   })
 }
+resource "aws_iam_role_policy" "webhook_sqs_1" {
+  count = var.webhook_events_secondary_queue ? 1 : 0
+  name  = "${var.prefix}-lambda-webhook-publish-sqs-policy-1"
+  role  = aws_iam_role.webhook_lambda.name
+
+  policy = templatefile("${path.module}/policies/lambda-publish-sqs-policy.json", {
+    sqs_resource_arn = var.sqs_monitored_build_events.arn
+  })
+}
 
 resource "aws_iam_role_policy" "webhook_ssm" {
   name = "${var.prefix}-lambda-webhook-publish-ssm-policy"

outputs.tf

@@ -39,7 +39,8 @@ output "ssm_parameters" {
 output "queues" {
   description = "SQS queues."
   value = {
-    build_queue_arn     = aws_sqs_queue.queued_builds.arn
-    build_queue_dlq_arn = var.redrive_build_queue.enabled ? aws_sqs_queue.queued_builds_dlq[0].arn : null
+    build_queue_arn         = aws_sqs_queue.queued_builds.arn
+    build_queue_dlq_arn     = var.redrive_build_queue.enabled ? aws_sqs_queue.queued_builds_dlq[0].arn : null
+    webhook_secondary_queue = try(aws_sqs_queue.monitored_build_events[0], null) != null ? aws_sqs_queue.monitored_build_events[0].arn : ""
   }
 }

variables.tf

@@ -656,4 +656,10 @@ variable "lambda_architecture" {
     condition     = contains(["arm64", "x86_64"], var.lambda_architecture)
     error_message = "`lambda_architecture` value is not valid, valid values are: `arm64` and `x86_64`."
   }
-}
+}
+
+variable "webhook_events_secondary_queue" {
+  description = "Enabling this feature will create a secondory sqs queue to wich a copy of the event will be delivered."
+  type        = bool
+  default     = false
+}