Added audit log clean up job and modified the db pool settings a bit (#946)

Author: Uros Marolt

backend/src/bin/jobs/cleanUp.ts

@@ -3,6 +3,7 @@ import IncomingWebhookRepository from '../../database/repositories/incomingWebho
 import IntegrationRunRepository from '../../database/repositories/integrationRunRepository'
 import SequelizeRepository from '../../database/repositories/sequelizeRepository'
 import { CrowdJob } from '../../types/jobTypes'
+import AuditLogRepository from '../../database/repositories/auditLogRepository'
 
 const MAX_MONTHS_TO_KEEP = 3
 
@@ -38,6 +39,13 @@ export const cleanUpOldWebhooks = async () => {
   await repo.cleanUpOldWebhooks(MAX_MONTHS_TO_KEEP)
 }
 
+export const cleanUpOldAuditLogs = async () => {
+  const dbOptions = await SequelizeRepository.getDefaultIRepositoryOptions()
+
+  log.info(`Cleaning up audit logs that are older than 1 month!`)
+  await AuditLogRepository.cleanUpOldAuditLogs(1, dbOptions)
+}
+
 const job: CrowdJob = {
   name: 'Clean up old data',
   // run once every week on Sunday at 1AM

backend/src/database/models/index.ts

@@ -67,9 +67,9 @@ function models() {
         write: { host: DB_CONFIG.writeHost },
       },
       pool: {
-        max: SERVICE === configTypes.ServiceType.API ? 100 : 10,
+        max: SERVICE === configTypes.ServiceType.API ? 20 : 10,
         min: 0,
-        acquire: 30000,
+        acquire: 50000,
         idle: 10000,
       },
       logging: DB_CONFIG.logging

backend/src/database/repositories/auditLogRepository.ts

@@ -1,4 +1,4 @@
-import Sequelize from 'sequelize'
+import Sequelize, { QueryTypes } from 'sequelize'
 import SequelizeRepository from './sequelizeRepository'
 import SequelizeFilterUtils from '../utils/sequelizeFilterUtils'
 import { IRepositoryOptions } from './IRepositoryOptions'
@@ -54,6 +54,22 @@ export default class AuditLogRepository {
     return log
   }
 
+  static async cleanUpOldAuditLogs(
+    maxMonthsToKeep: number,
+    options: IRepositoryOptions,
+  ): Promise<void> {
+    const seq = SequelizeRepository.getSequelize(options)
+
+    await seq.query(
+      `
+      delete from "auditLogs" where timestamp < now() - interval '${maxMonthsToKeep} months'
+      `,
+      {
+        type: QueryTypes.DELETE,
+      },
+    )
+  }
+
   static async findAndCountAll(
     { filter, limit = 0, offset = 0, orderBy = '' },
     options: IRepositoryOptions,

services/apps/data_sink_worker/src/main.ts

@@ -7,20 +7,28 @@ import { initializeSentimentAnalysis } from '@crowd/sentiment'
 
 const log = getServiceLogger()
 
+const MAX_CONCURRENT_PROCESSING = 2
+
 setImmediate(async () => {
   log.info('Starting data sink worker...')
 
   const sqsClient = getSqsClient(SQS_CONFIG())
 
-  const dbConnection = getDbConnection(DB_CONFIG())
+  const dbConnection = getDbConnection(DB_CONFIG(), MAX_CONCURRENT_PROCESSING)
 
   if (SENTIMENT_CONFIG()) {
     initializeSentimentAnalysis(SENTIMENT_CONFIG())
   }
 
   const nodejsWorkerEmitter = new NodejsWorkerEmitter(sqsClient, log)
 
-  const queue = new WorkerQueueReceiver(sqsClient, dbConnection, nodejsWorkerEmitter, log)
+  const queue = new WorkerQueueReceiver(
+    sqsClient,
+    dbConnection,
+    nodejsWorkerEmitter,
+    log,
+    MAX_CONCURRENT_PROCESSING,
+  )
 
   try {
     await nodejsWorkerEmitter.init()

services/apps/data_sink_worker/src/queue/index.ts

@@ -19,8 +19,9 @@ export class WorkerQueueReceiver extends SqsQueueReceiver {
     private readonly dbConn: DbConnection,
     private readonly nodejsWorkerEmitter: NodejsWorkerEmitter,
     parentLog: Logger,
+    maxConcurrentProcessing: number,
   ) {
-    super(client, DATA_SINK_WORKER_QUEUE_SETTINGS, 2, parentLog)
+    super(client, DATA_SINK_WORKER_QUEUE_SETTINGS, maxConcurrentProcessing, parentLog)
   }
 
   override async processMessage(message: IQueueMessage): Promise<void> {

services/apps/integration_data_worker/src/main.ts

@@ -7,12 +7,14 @@ import { WorkerQueueReceiver } from './queue'
 
 const log = getServiceLogger()
 
+const MAX_CONCURRENT_PROCESSING = 2
+
 setImmediate(async () => {
   log.info('Starting integration data worker...')
 
   const sqsClient = getSqsClient(SQS_CONFIG())
 
-  const dbConnection = getDbConnection(DB_CONFIG())
+  const dbConnection = getDbConnection(DB_CONFIG(), MAX_CONCURRENT_PROCESSING)
   const redisClient = await getRedisClient(REDIS_CONFIG(), true)
 
   const streamWorkerEmitter = new IntegrationStreamWorkerEmitter(sqsClient, log)
@@ -25,6 +27,7 @@ setImmediate(async () => {
     streamWorkerEmitter,
     dataSinkWorkerEmitter,
     log,
+    MAX_CONCURRENT_PROCESSING,
   )
 
   try {

services/apps/integration_data_worker/src/queue/index.ts

@@ -23,8 +23,9 @@ export class WorkerQueueReceiver extends SqsQueueReceiver {
     private readonly streamWorkerEmitter: IntegrationStreamWorkerEmitter,
     private readonly dataSinkWorkerEmitter: DataSinkWorkerEmitter,
     parentLog: Logger,
+    maxConcurrentProcessing: number,
   ) {
-    super(client, INTEGRATION_DATA_WORKER_QUEUE_SETTINGS, 2, parentLog)
+    super(client, INTEGRATION_DATA_WORKER_QUEUE_SETTINGS, maxConcurrentProcessing, parentLog)
   }
 
   override async processMessage(message: IQueueMessage): Promise<void> {

services/apps/integration_run_worker/src/main.ts

@@ -11,12 +11,14 @@ import { ApiPubSubEmitter, getRedisClient } from '@crowd/redis'
 
 const log = getServiceLogger()
 
+const MAX_CONCURRENT_PROCESSING = 2
+
 setImmediate(async () => {
   log.info('Starting integration run worker...')
 
   const sqsClient = getSqsClient(SQS_CONFIG())
 
-  const dbConnection = getDbConnection(DB_CONFIG())
+  const dbConnection = getDbConnection(DB_CONFIG(), MAX_CONCURRENT_PROCESSING)
   const redisClient = await getRedisClient(REDIS_CONFIG(), true)
 
   const runWorkerEmitter = new IntegrationRunWorkerEmitter(sqsClient, log)
@@ -32,6 +34,7 @@ setImmediate(async () => {
     runWorkerEmitter,
     apiPubSubEmitter,
     log,
+    MAX_CONCURRENT_PROCESSING,
   )
 
   try {

services/apps/integration_run_worker/src/queue/index.ts

@@ -28,8 +28,9 @@ export class WorkerQueueReceiver extends SqsQueueReceiver {
     private readonly runWorkerEmitter: IntegrationRunWorkerEmitter,
     private readonly apiPubSubEmitter: ApiPubSubEmitter,
     parentLog: Logger,
+    maxConcurrentProcessing: number,
   ) {
-    super(client, INTEGRATION_RUN_WORKER_QUEUE_SETTINGS, 2, parentLog)
+    super(client, INTEGRATION_RUN_WORKER_QUEUE_SETTINGS, maxConcurrentProcessing, parentLog)
   }
 
   override async processMessage(message: IQueueMessage): Promise<void> {

services/apps/integration_stream_worker/src/main.ts

@@ -12,12 +12,14 @@ import { WorkerQueueReceiver } from './queue'
 
 const log = getServiceLogger()
 
+const MAX_CONCURRENT_PROCESSING = 2
+
 setImmediate(async () => {
   log.info('Starting integration stream worker...')
 
   const sqsClient = getSqsClient(SQS_CONFIG())
 
-  const dbConnection = getDbConnection(DB_CONFIG())
+  const dbConnection = getDbConnection(DB_CONFIG(), MAX_CONCURRENT_PROCESSING)
   const redisClient = await getRedisClient(REDIS_CONFIG(), true)
 
   const runWorkerEmiiter = new IntegrationRunWorkerEmitter(sqsClient, log)
@@ -32,6 +34,7 @@ setImmediate(async () => {
     dataWorkerEmitter,
     streamWorkerEmitter,
     log,
+    MAX_CONCURRENT_PROCESSING,
   )
 
   try {