andwati/dokploy
1
0
Fork 0
mirror of https://github.com/andwati/dokploy.git synced 2026-11-10 22:15:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(deployments): enhance deployment worker and queue handling for cloud environments

Browse Source 
- Refactored the deployment worker to create a no-op worker when Redis is disabled (e.g., IS_CLOUD), preventing BullMQ connection errors.
- Updated queue initialization to use a no-op queue in cloud environments, ensuring compatibility and stability.
- Improved error handling and logging for job processing in the deployment worker.
This commit is contained in:
Mauricio Siu
2026-02-10 03:11:33 -06:00
parent 17da1d5b3c
commit 744ebab15a
2 changed files with 102 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/dokploy/server/queues/deployments-queue.ts
+75 -61
View File
@@ -2,6 +2,7 @@ import {
deployApplication, deployApplication,
deployCompose, deployCompose,
deployPreviewApplication, deployPreviewApplication,
IS_CLOUD,
rebuildApplication, rebuildApplication,
rebuildCompose, rebuildCompose,
rebuildPreviewApplication, rebuildPreviewApplication,
@@ -13,70 +14,83 @@ import { type Job, Worker } from "bullmq";
import type { DeploymentJob } from "./queue-types"; import type { DeploymentJob } from "./queue-types";
import { redisConfig } from "./redis-connection"; import { redisConfig } from "./redis-connection";
export const deploymentWorker = new Worker( const createDeploymentWorker = () =>
"deployments", new Worker(
async (job: Job<DeploymentJob>) => { "deployments",
try { async (job: Job<DeploymentJob>) => {
if (job.data.applicationType === "application") { try {
await updateApplicationStatus(job.data.applicationId, "running"); if (job.data.applicationType === "application") {
await updateApplicationStatus(job.data.applicationId, "running");
if (job.data.type === "redeploy") { if (job.data.type === "redeploy") {
await rebuildApplication({ await rebuildApplication({
applicationId: job.data.applicationId, applicationId: job.data.applicationId,
titleLog: job.data.titleLog, titleLog: job.data.titleLog,
descriptionLog: job.data.descriptionLog, descriptionLog: job.data.descriptionLog,
});
} else if (job.data.type === "deploy") {
await deployApplication({
applicationId: job.data.applicationId,
titleLog: job.data.titleLog,
descriptionLog: job.data.descriptionLog,
});
}
} else if (job.data.applicationType === "compose") {
await updateCompose(job.data.composeId, {
composeStatus: "running",
}); });
} else if (job.data.type === "deploy") { if (job.data.type === "deploy") {
await deployApplication({ await deployCompose({
applicationId: job.data.applicationId, composeId: job.data.composeId,
titleLog: job.data.titleLog, titleLog: job.data.titleLog,
descriptionLog: job.data.descriptionLog, descriptionLog: job.data.descriptionLog,
});
} else if (job.data.type === "redeploy") {
await rebuildCompose({
composeId: job.data.composeId,
titleLog: job.data.titleLog,
descriptionLog: job.data.descriptionLog,
});
}
} else if (job.data.applicationType === "application-preview") {
await updatePreviewDeployment(job.data.previewDeploymentId, {
previewStatus: "running",
}); });
}
} else if (job.data.applicationType === "compose") {
await updateCompose(job.data.composeId, {
composeStatus: "running",
});
if (job.data.type === "deploy") {
await deployCompose({
composeId: job.data.composeId,
titleLog: job.data.titleLog,
descriptionLog: job.data.descriptionLog,
});
} else if (job.data.type === "redeploy") {
await rebuildCompose({
composeId: job.data.composeId,
titleLog: job.data.titleLog,
descriptionLog: job.data.descriptionLog,
});
}
} else if (job.data.applicationType === "application-preview") {
await updatePreviewDeployment(job.data.previewDeploymentId, {
previewStatus: "running",
});
if (job.data.type === "redeploy") { if (job.data.type === "redeploy") {
await rebuildPreviewApplication({ await rebuildPreviewApplication({
applicationId: job.data.applicationId, applicationId: job.data.applicationId,
titleLog: job.data.titleLog, titleLog: job.data.titleLog,
descriptionLog: job.data.descriptionLog, descriptionLog: job.data.descriptionLog,
previewDeploymentId: job.data.previewDeploymentId, previewDeploymentId: job.data.previewDeploymentId,
}); });
} else if (job.data.type === "deploy") { } else if (job.data.type === "deploy") {
await deployPreviewApplication({ await deployPreviewApplication({
applicationId: job.data.applicationId, applicationId: job.data.applicationId,
titleLog: job.data.titleLog, titleLog: job.data.titleLog,
descriptionLog: job.data.descriptionLog, descriptionLog: job.data.descriptionLog,
previewDeploymentId: job.data.previewDeploymentId, previewDeploymentId: job.data.previewDeploymentId,
}); });
}
} }
} catch (error) {
console.log("Error", error);
} }
} catch (error) { },
console.log("Error", error); {
} autorun: false,
}, connection: redisConfig,
{ },
autorun: false, );
connection: redisConfig,
}, /** No-op worker when Redis is disabled (e.g. IS_CLOUD). Avoids BullMQ connection errors. */
); const noopWorker = {
run: () => Promise.resolve(),
close: () => Promise.resolve(),
cancelJob: () => Promise.resolve(),
cancelAllJobs: () => Promise.resolve(),
};
export const deploymentWorker = !IS_CLOUD
? createDeploymentWorker()
: (noopWorker as unknown as Worker<DeploymentJob>);
apps/dokploy/server/queues/queueSetup.ts
+27 -14
View File
@@ -1,15 +1,26 @@
import { IS_CLOUD } from "@dokploy/server";
import { import {
execAsync, execAsync,
execAsyncRemote, execAsyncRemote,
} from "@dokploy/server/utils/process/execAsync"; } from "@dokploy/server/utils/process/execAsync";
import type { Job } from "bullmq";
import { Queue } from "bullmq"; import { Queue } from "bullmq";
import { deploymentWorker } from "./deployments-queue"; import { deploymentWorker } from "./deployments-queue";
import { redisConfig } from "./redis-connection"; import { redisConfig } from "./redis-connection";
const myQueue = new Queue("deployments", { /** No-op queue when Redis is disabled (e.g. IS_CLOUD). Avoids BullMQ connection errors. */
connection: redisConfig, const createNoopQueue = () => ({
getJobs: () => Promise.resolve([] as Job[]),
add: () =>
Promise.resolve({ id: "noop", remove: () => Promise.resolve() } as Job),
close: () => Promise.resolve(),
on: () => {},
}); });
const myQueue = !IS_CLOUD
? new Queue("deployments", { connection: redisConfig })
: (createNoopQueue() as unknown as Queue);
export const getJobsByApplicationId = async (applicationId: string) => { export const getJobsByApplicationId = async (applicationId: string) => {
const jobs = await myQueue.getJobs(); const jobs = await myQueue.getJobs();
return jobs.filter((job) => job?.data?.applicationId === applicationId); return jobs.filter((job) => job?.data?.applicationId === applicationId);
@@ -20,19 +31,21 @@ export const getJobsByComposeId = async (composeId: string) => {
return jobs.filter((job) => job?.data?.composeId === composeId); return jobs.filter((job) => job?.data?.composeId === composeId);
}; };
process.on("SIGTERM", () => { if (!IS_CLOUD) {
myQueue.close(); process.on("SIGTERM", () => {
process.exit(0); myQueue.close();
}); process.exit(0);
});
myQueue.on("error", (error) => { myQueue.on("error", (error) => {
if ((error as any).code === "ECONNREFUSED") { if ((error as any).code === "ECONNREFUSED") {
console.error( console.error(
"Make sure you have installed Redis and it is running.", "Make sure you have installed Redis and it is running.",
error, error,
); );
} }
}); });
}
export const cleanQueuesByApplication = async (applicationId: string) => { export const cleanQueuesByApplication = async (applicationId: string) => {
const jobs = await myQueue.getJobs(["waiting", "delayed"]); const jobs = await myQueue.getJobs(["waiting", "delayed"]);