feat: removed audit log queue and switched to resource clean up queue

This commit is contained in:
=
2024-05-16 15:46:19 +05:30
parent 08e7815ec1
commit 3ed5dd6109
4 changed files with 77 additions and 32 deletions
@@ -3,7 +3,6 @@ import { RawAxiosRequestHeaders } from "axios";
import { SecretKeyEncoding } from "@app/db/schemas"; import { SecretKeyEncoding } from "@app/db/schemas";
import { request } from "@app/lib/config/request"; import { request } from "@app/lib/config/request";
import { infisicalSymmetricDecrypt } from "@app/lib/crypto/encryption"; import { infisicalSymmetricDecrypt } from "@app/lib/crypto/encryption";
import { logger } from "@app/lib/logger";
import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue"; import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
import { TProjectDALFactory } from "@app/services/project/project-dal"; import { TProjectDALFactory } from "@app/services/project/project-dal";
@@ -113,35 +112,7 @@ export const auditLogQueueServiceFactory = ({
); );
}); });
queueService.start(QueueName.AuditLogPrune, async () => {
logger.info(`${QueueName.AuditLogPrune}: queue task started`);
await auditLogDAL.pruneAuditLog();
logger.info(`${QueueName.AuditLogPrune}: queue task completed`);
});
// we do a repeat cron job in utc timezone at 12 Midnight each day
const startAuditLogPruneJob = async () => {
// clear previous job
await queueService.stopRepeatableJob(
QueueName.AuditLogPrune,
QueueJobs.AuditLogPrune,
{ pattern: "0 0 * * *", utc: true },
QueueName.AuditLogPrune // just a job id
);
await queueService.queue(QueueName.AuditLogPrune, QueueJobs.AuditLogPrune, undefined, {
delay: 5000,
jobId: QueueName.AuditLogPrune,
repeat: { pattern: "0 0 * * *", utc: true }
});
};
queueService.listen(QueueName.AuditLogPrune, "failed", (err) => {
logger.error(err?.failedReason, `${QueueName.AuditLogPrune}: log pruning failed`);
});
return { return {
pushToLog, pushToLog
startAuditLogPruneJob
}; };
}; };
+11 -1
View File
@@ -12,7 +12,9 @@ export enum QueueName {
SecretRotation = "secret-rotation", SecretRotation = "secret-rotation",
SecretReminder = "secret-reminder", SecretReminder = "secret-reminder",
AuditLog = "audit-log", AuditLog = "audit-log",
// TODO(akhilmhdh): This will get removed later. For now this is kept to stop the repeatable queue
AuditLogPrune = "audit-log-prune", AuditLogPrune = "audit-log-prune",
DailyResourceCleanUp = "daily-resource-cleanup",
TelemetryInstanceStats = "telemtry-self-hosted-stats", TelemetryInstanceStats = "telemtry-self-hosted-stats",
IntegrationSync = "sync-integrations", IntegrationSync = "sync-integrations",
SecretWebhook = "secret-webhook", SecretWebhook = "secret-webhook",
@@ -26,7 +28,9 @@ export enum QueueJobs {
SecretReminder = "secret-reminder-job", SecretReminder = "secret-reminder-job",
SecretRotation = "secret-rotation-job", SecretRotation = "secret-rotation-job",
AuditLog = "audit-log-job", AuditLog = "audit-log-job",
// TODO(akhilmhdh): This will get removed later. For now this is kept to stop the repeatable queue
AuditLogPrune = "audit-log-prune-job", AuditLogPrune = "audit-log-prune-job",
DailyResourceCleanUp = "daily-resource-cleanup-job",
SecWebhook = "secret-webhook-trigger", SecWebhook = "secret-webhook-trigger",
TelemetryInstanceStats = "telemetry-self-hosted-stats", TelemetryInstanceStats = "telemetry-self-hosted-stats",
IntegrationSync = "secret-integration-pull", IntegrationSync = "secret-integration-pull",
@@ -55,6 +59,10 @@ export type TQueueJobTypes = {
name: QueueJobs.AuditLog; name: QueueJobs.AuditLog;
payload: TCreateAuditLogDTO; payload: TCreateAuditLogDTO;
}; };
[QueueName.DailyResourceCleanUp]: {
name: QueueJobs.DailyResourceCleanUp;
payload: undefined;
};
[QueueName.AuditLogPrune]: { [QueueName.AuditLogPrune]: {
name: QueueJobs.AuditLogPrune; name: QueueJobs.AuditLogPrune;
payload: undefined; payload: undefined;
@@ -172,7 +180,9 @@ export const queueServiceFactory = (redisUrl: string) => {
jobId?: string jobId?: string
) => { ) => {
const q = queueContainer[name]; const q = queueContainer[name];
return q.removeRepeatable(job, repeatOpt, jobId); if (q) {
return q.removeRepeatable(job, repeatOpt, jobId);
}
}; };
const stopRepeatableJobByJobId = async <T extends QueueName>(name: T, jobId: string) => { const stopRepeatableJobByJobId = async <T extends QueueName>(name: T, jobId: string) => {
+7 -1
View File
@@ -113,6 +113,7 @@ import { projectMembershipServiceFactory } from "@app/services/project-membershi
import { projectUserMembershipRoleDALFactory } from "@app/services/project-membership/project-user-membership-role-dal"; import { projectUserMembershipRoleDALFactory } from "@app/services/project-membership/project-user-membership-role-dal";
import { projectRoleDALFactory } from "@app/services/project-role/project-role-dal"; import { projectRoleDALFactory } from "@app/services/project-role/project-role-dal";
import { projectRoleServiceFactory } from "@app/services/project-role/project-role-service"; import { projectRoleServiceFactory } from "@app/services/project-role/project-role-service";
import { dailyResourceCleanUpQueueServiceFactory } from "@app/services/resource-cleanup/resource-cleanup-queue";
import { secretDALFactory } from "@app/services/secret/secret-dal"; import { secretDALFactory } from "@app/services/secret/secret-dal";
import { secretQueueFactory } from "@app/services/secret/secret-queue"; import { secretQueueFactory } from "@app/services/secret/secret-queue";
import { secretServiceFactory } from "@app/services/secret/secret-service"; import { secretServiceFactory } from "@app/services/secret/secret-service";
@@ -757,14 +758,19 @@ export const registerRoutes = async (
folderDAL, folderDAL,
licenseService licenseService
}); });
const dailyResourceCleanUp = dailyResourceCleanUpQueueServiceFactory({
auditLogDAL,
queueService,
identityAccessTokenDAL
});
await superAdminService.initServerCfg(); await superAdminService.initServerCfg();
// //
// setup the communication with license key server // setup the communication with license key server
await licenseService.init(); await licenseService.init();
await auditLogQueue.startAuditLogPruneJob();
await telemetryQueue.startTelemetryCheck(); await telemetryQueue.startTelemetryCheck();
await dailyResourceCleanUp.startCleanUp();
// inject all services // inject all services
server.decorate<FastifyZodProvider["services"]>("services", { server.decorate<FastifyZodProvider["services"]>("services", {
@@ -0,0 +1,58 @@
import { TAuditLogDALFactory } from "@app/ee/services/audit-log/audit-log-dal";
import { logger } from "@app/lib/logger";
import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
import { TIdentityAccessTokenDALFactory } from "../identity-access-token/identity-access-token-dal";
type TDailyResourceCleanUpQueueServiceFactoryDep = {
auditLogDAL: Pick<TAuditLogDALFactory, "pruneAuditLog">;
identityAccessTokenDAL: Pick<TIdentityAccessTokenDALFactory, "removeExpiredTokens">;
queueService: TQueueServiceFactory;
};
export type TDailyResourceCleanUpQueueServiceFactory = ReturnType<typeof dailyResourceCleanUpQueueServiceFactory>;
export const dailyResourceCleanUpQueueServiceFactory = ({
auditLogDAL,
queueService,
identityAccessTokenDAL
}: TDailyResourceCleanUpQueueServiceFactoryDep) => {
queueService.start(QueueName.DailyResourceCleanUp, async () => {
logger.info(`${QueueName.DailyResourceCleanUp}: queue task started`);
await auditLogDAL.pruneAuditLog();
await identityAccessTokenDAL.removeExpiredTokens();
logger.info(`${QueueName.DailyResourceCleanUp}: queue task completed`);
});
// we do a repeat cron job in utc timezone at 12 Midnight each day
const startCleanUp = async () => {
// TODO(akhilmhdh): remove later
await queueService.stopRepeatableJob(
QueueName.AuditLogPrune,
QueueJobs.AuditLogPrune,
{ pattern: "0 0 * * *", utc: true },
QueueName.AuditLogPrune // just a job id
);
// clear previous job
await queueService.stopRepeatableJob(
QueueName.DailyResourceCleanUp,
QueueJobs.DailyResourceCleanUp,
{ pattern: "0 0 * * *", utc: true },
QueueName.DailyResourceCleanUp // just a job id
);
await queueService.queue(QueueName.DailyResourceCleanUp, QueueJobs.DailyResourceCleanUp, undefined, {
delay: 5000,
jobId: QueueName.DailyResourceCleanUp,
repeat: { pattern: "0 0 * * *", utc: true }
});
};
queueService.listen(QueueName.DailyResourceCleanUp, "failed", (_, err) => {
logger.error(err, `${QueueName.DailyResourceCleanUp}: resource cleanup failed`);
});
return {
startCleanUp
};
};