diff --git a/apps/dokploy/__test__/log-management/log-provider-service.test.ts b/apps/dokploy/__test__/log-management/log-provider-service.test.ts index 1a00756c7..83b465218 100644 --- a/apps/dokploy/__test__/log-management/log-provider-service.test.ts +++ b/apps/dokploy/__test__/log-management/log-provider-service.test.ts @@ -262,6 +262,41 @@ describe("updateLogProvider — merges partial update with existing row before v ); }); + it("clears an optional credential when it is explicitly set to null", async () => { + mocks.findFirst.mockResolvedValue({ + logProviderId: "lp-1", + name: "loki-prod", + providerType: "loki", + endpoint: "https://loki.example.com", + apiKey: "old-token", + apiSecret: null, + extraConfig: { username: "user" }, + }); + mocks.updateSet.mockReturnValue([{ logProviderId: "lp-1" }]); + + await updateLogProvider("lp-1", { apiKey: null }); + + expect(mocks.updateSet).toHaveBeenCalledWith( + expect.objectContaining({ apiKey: null }), + ); + }); + + it("rejects clearing the elasticsearch password while a username stays saved", async () => { + mocks.findFirst.mockResolvedValue({ + logProviderId: "lp-1", + name: "es-prod", + providerType: "elasticsearch", + endpoint: "https://es.example.com:9200", + apiKey: "old-password", + apiSecret: null, + extraConfig: { username: "elastic" }, + }); + + await expect(updateLogProvider("lp-1", { apiKey: null })).rejects.toThrow( + /Password/, + ); + }); + it("rejects a providerType change that doesn't satisfy the new type's required fields", async () => { mocks.findFirst.mockResolvedValue({ logProviderId: "lp-1", diff --git a/apps/dokploy/__test__/log-management/remove-log-provider.test.ts b/apps/dokploy/__test__/log-management/remove-log-provider.test.ts index 3ab878424..54c528590 100644 --- a/apps/dokploy/__test__/log-management/remove-log-provider.test.ts +++ b/apps/dokploy/__test__/log-management/remove-log-provider.test.ts @@ -55,8 +55,8 @@ describe("removeLogProvider", () => { expect(render(mocks.updates[1]?.values.logProviderIds)).toBe( 'array_remove("webServerSettings"."logProviderIds", $1)', ); - expect(render(mocks.updates[1]?.values.logManagementOrganizationId)).toBe( - 'CASE WHEN cardinality(array_remove("webServerSettings"."logProviderIds", $1)) = 0 THEN NULL ELSE "webServerSettings"."logManagementOrganizationId" END', + expect(mocks.updates[1]?.values).not.toHaveProperty( + "logManagementOrganizationId", ); }); diff --git a/apps/dokploy/__test__/log-management/vector-agent.test.ts b/apps/dokploy/__test__/log-management/vector-agent.test.ts index 5c9a981b5..2a4a29ae0 100644 --- a/apps/dokploy/__test__/log-management/vector-agent.test.ts +++ b/apps/dokploy/__test__/log-management/vector-agent.test.ts @@ -2,7 +2,10 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const mocks = vi.hoisted(() => ({ claimWebServerLogManagement: vi.fn(), + releaseWebServerLogManagement: vi.fn(), + getWebServerSettings: vi.fn(), updateServerLogProviders: vi.fn(), + getAccessibleServerIds: vi.fn(), removeVectorAgent: vi.fn(), setupVectorAgent: vi.fn(), assertServerBelongsToOrg: vi.fn(), @@ -14,13 +17,18 @@ const loadAgent = async (isCloud: boolean) => { vi.doMock("@dokploy/server/constants", () => ({ IS_CLOUD: isCloud })); vi.doMock("@dokploy/server/services/server", () => ({ updateServerLogProviders: mocks.updateServerLogProviders, + getAccessibleServerIds: mocks.getAccessibleServerIds, })); vi.doMock("@dokploy/server/services/web-server-settings", () => ({ claimWebServerLogManagement: mocks.claimWebServerLogManagement, + releaseWebServerLogManagement: mocks.releaseWebServerLogManagement, + getWebServerSettings: mocks.getWebServerSettings, })); vi.doMock("@dokploy/server/setup/vector-setup", () => ({ removeVectorAgent: mocks.removeVectorAgent, setupVectorAgent: mocks.setupVectorAgent, + withVectorTargetLock: (_: string | undefined, fn: () => Promise) => + fn(), })); vi.doMock("@dokploy/server/services/log-management/service", () => ({ assertServerBelongsToOrg: mocks.assertServerBelongsToOrg, @@ -29,27 +37,62 @@ const loadAgent = async (isCloud: boolean) => { return await import("@dokploy/server/services/log-management/vector-agent"); }; +const sessionA = { userId: "user-a", activeOrganizationId: "org-a" }; +const sessionB = { userId: "user-b", activeOrganizationId: "org-b" }; + describe("local Vector agent guards", () => { beforeEach(() => { for (const mock of Object.values(mocks)) mock.mockReset(); + mocks.getAccessibleServerIds.mockResolvedValue(new Set(["server-1"])); }); it("does not remove the local agent when another organization owns it", async () => { - mocks.claimWebServerLogManagement.mockResolvedValue(null); + mocks.getWebServerSettings.mockResolvedValue({ + logManagementOrganizationId: "org-a", + }); const { removeLogManagement } = await loadAgent(false); - await expect(removeLogManagement("org-b", undefined)).rejects.toMatchObject( - { code: "CONFLICT" }, - ); - expect(mocks.claimWebServerLogManagement).toHaveBeenCalledWith("org-b", []); + await expect( + removeLogManagement(sessionB, undefined), + ).rejects.toMatchObject({ code: "CONFLICT" }); expect(mocks.removeVectorAgent).not.toHaveBeenCalled(); + expect(mocks.releaseWebServerLogManagement).not.toHaveBeenCalled(); }); - it("removes the local agent once the claim is released", async () => { - mocks.claimWebServerLogManagement.mockResolvedValue({ id: "settings" }); + it("removes the local agent and only then releases the claim", async () => { + mocks.getWebServerSettings.mockResolvedValue({ + logManagementOrganizationId: "org-a", + }); const { removeLogManagement } = await loadAgent(false); - await removeLogManagement("org-a", undefined); + await removeLogManagement(sessionA, undefined); + expect(mocks.removeVectorAgent).toHaveBeenCalledWith(); + expect(mocks.releaseWebServerLogManagement).toHaveBeenCalledWith("org-a"); + expect(mocks.removeVectorAgent.mock.invocationCallOrder[0]).toBeLessThan( + mocks.releaseWebServerLogManagement.mock.invocationCallOrder[0] ?? 0, + ); + }); + + it("keeps the claim when removing the local agent fails", async () => { + mocks.getWebServerSettings.mockResolvedValue({ + logManagementOrganizationId: "org-a", + }); + mocks.removeVectorAgent.mockRejectedValue(new Error("docker down")); + const { removeLogManagement } = await loadAgent(false); + + await expect(removeLogManagement(sessionA, undefined)).rejects.toThrow( + "docker down", + ); + expect(mocks.releaseWebServerLogManagement).not.toHaveBeenCalled(); + }); + + it("removes an unclaimed local agent", async () => { + mocks.getWebServerSettings.mockResolvedValue({ + logManagementOrganizationId: null, + }); + const { removeLogManagement } = await loadAgent(false); + + await removeLogManagement(sessionA, undefined); expect(mocks.removeVectorAgent).toHaveBeenCalledWith(); }); @@ -57,7 +100,7 @@ describe("local Vector agent guards", () => { const { deployLogManagement } = await loadAgent(true); await expect( - deployLogManagement("org-a", undefined, ["lp-1"]), + deployLogManagement(sessionA, undefined, ["lp-1"]), ).rejects.toMatchObject({ code: "BAD_REQUEST" }); expect(mocks.claimWebServerLogManagement).not.toHaveBeenCalled(); expect(mocks.setupVectorAgent).not.toHaveBeenCalled(); @@ -66,17 +109,17 @@ describe("local Vector agent guards", () => { it("rejects removing the local agent in cloud", async () => { const { removeLogManagement } = await loadAgent(true); - await expect(removeLogManagement("org-a", undefined)).rejects.toMatchObject( - { code: "BAD_REQUEST" }, - ); - expect(mocks.claimWebServerLogManagement).not.toHaveBeenCalled(); + await expect( + removeLogManagement(sessionA, undefined), + ).rejects.toMatchObject({ code: "BAD_REQUEST" }); + expect(mocks.getWebServerSettings).not.toHaveBeenCalled(); expect(mocks.removeVectorAgent).not.toHaveBeenCalled(); }); it("still manages remote servers in cloud", async () => { const { removeLogManagement } = await loadAgent(true); - await removeLogManagement("org-a", "server-1"); + await removeLogManagement(sessionA, "server-1"); expect(mocks.assertServerBelongsToOrg).toHaveBeenCalledWith( "server-1", "org-a", @@ -85,3 +128,139 @@ describe("local Vector agent guards", () => { expect(mocks.updateServerLogProviders).toHaveBeenCalledWith("server-1", []); }); }); + +describe("remote server access", () => { + beforeEach(() => { + for (const mock of Object.values(mocks)) mock.mockReset(); + mocks.getAccessibleServerIds.mockResolvedValue(new Set(["server-1"])); + }); + + it("rejects deploying on a server the member cannot access", async () => { + const { deployLogManagement } = await loadAgent(false); + + await expect( + deployLogManagement(sessionA, "server-2", ["lp-1"]), + ).rejects.toMatchObject({ code: "UNAUTHORIZED" }); + expect(mocks.getAccessibleServerIds).toHaveBeenCalledWith(sessionA); + expect(mocks.updateServerLogProviders).not.toHaveBeenCalled(); + expect(mocks.setupVectorAgent).not.toHaveBeenCalled(); + }); + + it("rejects removing from a server the member cannot access", async () => { + const { removeLogManagement } = await loadAgent(false); + + await expect( + removeLogManagement(sessionA, "server-2"), + ).rejects.toMatchObject({ code: "UNAUTHORIZED" }); + expect(mocks.removeVectorAgent).not.toHaveBeenCalled(); + }); + + it("deploys on an accessible server", async () => { + const { deployLogManagement } = await loadAgent(false); + + await deployLogManagement(sessionA, "server-1", ["lp-1"]); + expect(mocks.updateServerLogProviders).toHaveBeenCalledWith("server-1", [ + "lp-1", + ]); + expect(mocks.setupVectorAgent).toHaveBeenCalledWith("org-a", "server-1", [ + "lp-1", + ]); + }); + + it("keeps the saved providers when the server deploy fails", async () => { + mocks.setupVectorAgent.mockRejectedValue(new Error("validation failed")); + const { deployLogManagement } = await loadAgent(false); + + await expect( + deployLogManagement(sessionA, "server-1", ["lp-2"]), + ).rejects.toThrow("validation failed"); + expect(mocks.updateServerLogProviders).not.toHaveBeenCalled(); + }); + + it("removes the agent on server deletion even with no providers left", async () => { + const { removeServerLogManagement } = await loadAgent(false); + + await removeServerLogManagement({ + serverId: "server-1", + sshKeyId: "key-1", + logProviderIds: [], + } as never); + expect(mocks.removeVectorAgent).toHaveBeenCalledWith("server-1"); + }); +}); + +describe("local agent deploy", () => { + beforeEach(() => { + for (const mock of Object.values(mocks)) mock.mockReset(); + mocks.claimWebServerLogManagement.mockResolvedValue({}); + }); + + it("saves the new providers only after the deploy succeeds", async () => { + mocks.getWebServerSettings.mockResolvedValue({ + logManagementOrganizationId: "org-a", + logProviderIds: ["lp-1"], + }); + const { deployLogManagement } = await loadAgent(false); + + await deployLogManagement(sessionA, undefined, ["lp-2"]); + expect(mocks.claimWebServerLogManagement).toHaveBeenNthCalledWith( + 1, + "org-a", + ["lp-1"], + ); + expect(mocks.claimWebServerLogManagement).toHaveBeenNthCalledWith( + 2, + "org-a", + ["lp-2"], + ); + expect(mocks.setupVectorAgent.mock.invocationCallOrder[0]).toBeLessThan( + mocks.claimWebServerLogManagement.mock.invocationCallOrder[1] ?? 0, + ); + }); + + it("keeps the previous providers when a redeploy fails", async () => { + mocks.getWebServerSettings.mockResolvedValue({ + logManagementOrganizationId: "org-a", + logProviderIds: ["lp-1"], + }); + mocks.setupVectorAgent.mockRejectedValue(new Error("validation failed")); + const { deployLogManagement } = await loadAgent(false); + + await expect( + deployLogManagement(sessionA, undefined, ["lp-2"]), + ).rejects.toThrow("validation failed"); + expect(mocks.claimWebServerLogManagement).toHaveBeenCalledTimes(1); + expect(mocks.claimWebServerLogManagement).toHaveBeenCalledWith("org-a", [ + "lp-1", + ]); + expect(mocks.releaseWebServerLogManagement).not.toHaveBeenCalled(); + }); + + it("releases the claim when the first local deploy fails", async () => { + mocks.getWebServerSettings.mockResolvedValue({ + logManagementOrganizationId: null, + logProviderIds: [], + }); + mocks.setupVectorAgent.mockRejectedValue(new Error("validation failed")); + const { deployLogManagement } = await loadAgent(false); + + await expect( + deployLogManagement(sessionA, undefined, ["lp-1"]), + ).rejects.toThrow("validation failed"); + expect(mocks.releaseWebServerLogManagement).toHaveBeenCalledWith("org-a"); + }); + + it("rejects deploying when another organization owns the local agent", async () => { + mocks.getWebServerSettings.mockResolvedValue({ + logManagementOrganizationId: "org-a", + logProviderIds: ["lp-1"], + }); + mocks.claimWebServerLogManagement.mockResolvedValue(null); + const { deployLogManagement } = await loadAgent(false); + + await expect( + deployLogManagement(sessionB, undefined, ["lp-9"]), + ).rejects.toMatchObject({ code: "CONFLICT" }); + expect(mocks.setupVectorAgent).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/dokploy/__test__/log-management/vector-status.test.ts b/apps/dokploy/__test__/log-management/vector-status.test.ts new file mode 100644 index 000000000..0fb46ed8b --- /dev/null +++ b/apps/dokploy/__test__/log-management/vector-status.test.ts @@ -0,0 +1,120 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + listTasks: vi.fn(), + findServersByOrganizationForLogManagement: vi.fn(), + getAccessibleServerIds: vi.fn(), + getWebServerSettings: vi.fn(), +})); + +vi.mock("@dokploy/server/db", () => ({ db: {} })); +vi.mock("@dokploy/server/constants", () => ({ + IS_CLOUD: false, + paths: () => ({ VECTOR_PATH: "/etc/dokploy/vector" }), +})); +vi.mock("@dokploy/server/services/server", () => ({ + findServerById: vi.fn(), + findServersByOrganizationForLogManagement: + mocks.findServersByOrganizationForLogManagement, + getAccessibleServerIds: mocks.getAccessibleServerIds, +})); +vi.mock("@dokploy/server/services/web-server-settings", () => ({ + getWebServerSettings: mocks.getWebServerSettings, +})); +vi.mock("@dokploy/server/utils/servers/remote-docker", () => ({ + getRemoteDocker: async () => ({ listTasks: mocks.listTasks }), +})); + +const { getLogManagementServerStatus, withVectorTargetLock } = await import( + "@dokploy/server/setup/vector-setup" +); + +const session = { userId: "user-1", activeOrganizationId: "org-1" }; + +describe("getLogManagementServerStatus", () => { + beforeEach(() => { + for (const mock of Object.values(mocks)) mock.mockReset(); + mocks.findServersByOrganizationForLogManagement.mockResolvedValue([]); + mocks.getAccessibleServerIds.mockResolvedValue(new Set()); + mocks.getWebServerSettings.mockResolvedValue({ + logManagementOrganizationId: "org-1", + logProviderIds: ["lp-1"], + }); + }); + + it("reports running only when a task of the service is running", async () => { + mocks.listTasks.mockResolvedValue([{ Status: { State: "running" } }]); + + const [local] = await getLogManagementServerStatus(session); + expect(local?.status).toBe("running"); + expect(mocks.listTasks).toHaveBeenCalledWith({ + filters: JSON.stringify({ + service: ["dokploy-vector"], + "desired-state": ["running"], + }), + }); + }); + + it("reports not-running when the service has no running task", async () => { + mocks.listTasks.mockResolvedValue([{ Status: { State: "pending" } }]); + + const [local] = await getLogManagementServerStatus(session); + expect(local?.status).toBe("not-running"); + }); + + it("reports stopped when the service does not exist", async () => { + mocks.listTasks.mockRejectedValue( + Object.assign(new Error("not found"), { statusCode: 404 }), + ); + + const [local] = await getLogManagementServerStatus(session); + expect(local?.status).toBe("stopped"); + }); + + it("reports unknown on any other docker error", async () => { + mocks.listTasks.mockRejectedValue(new Error("connect ECONNREFUSED")); + + const [local] = await getLogManagementServerStatus(session); + expect(local?.status).toBe("unknown"); + }); + + it("hides servers the member cannot access", async () => { + mocks.listTasks.mockResolvedValue([]); + mocks.findServersByOrganizationForLogManagement.mockResolvedValue([ + { serverId: "server-1", name: "one", ipAddress: "1.1.1.1" }, + { serverId: "server-2", name: "two", ipAddress: "2.2.2.2" }, + ]); + mocks.getAccessibleServerIds.mockResolvedValue(new Set(["server-2"])); + + const targets = await getLogManagementServerStatus(session); + expect(targets.map((t) => t.serverId)).toEqual([null, "server-2"]); + }); +}); + +describe("withVectorTargetLock", () => { + it("runs operations on the same target one after another", async () => { + const events: string[] = []; + let releaseFirst = () => {}; + const first = withVectorTargetLock("server-1", async () => { + events.push("first:start"); + await new Promise((resolve) => { + releaseFirst = resolve; + }); + events.push("first:end"); + throw new Error("first failed"); + }); + const second = withVectorTargetLock("server-1", async () => { + events.push("second"); + }); + const other = withVectorTargetLock("server-2", async () => { + events.push("other"); + }); + + await other; + expect(events).toEqual(["first:start", "other"]); + releaseFirst(); + await expect(first).rejects.toThrow("first failed"); + await second; + expect(events).toEqual(["first:start", "other", "first:end", "second"]); + }); +}); diff --git a/apps/dokploy/components/dashboard/settings/log-management/handle-log-provider.tsx b/apps/dokploy/components/dashboard/settings/log-management/handle-log-provider.tsx index 18a98cc44..3d824f08c 100644 --- a/apps/dokploy/components/dashboard/settings/log-management/handle-log-provider.tsx +++ b/apps/dokploy/components/dashboard/settings/log-management/handle-log-provider.tsx @@ -50,6 +50,7 @@ export const HandleLogProvider = ({ logProviderId }: Props) => { const [initialFieldValues, setInitialFieldValues] = useState< Record >({}); + const [clearedKeys, setClearedKeys] = useState>({}); const { data: availableTypes } = api.logProvider.availableTypes.useQuery(); const { data: provider } = api.logProvider.one.useQuery( @@ -96,12 +97,14 @@ export const HandleLogProvider = ({ logProviderId }: Props) => { >; setFieldValues(extraConfig); setInitialFieldValues(extraConfig); + setClearedKeys({}); } else if (isOpen) { setName(""); setProviderType(""); setEnabled(true); setFieldValues({}); setInitialFieldValues({}); + setClearedKeys({}); } }, [provider, isOpen]); @@ -114,6 +117,8 @@ export const HandleLogProvider = ({ logProviderId }: Props) => { if (TOP_LEVEL_KEYS.has(field.key)) { if (value) { payload[field.key] = value; + } else if (clearedKeys[field.key]) { + payload[field.key] = null; } } else { hasExtraConfigFields = true; @@ -129,7 +134,9 @@ export const HandleLogProvider = ({ logProviderId }: Props) => { const topLevelKeys = (selectedType?.credentialFields ?? []) .filter((field) => TOP_LEVEL_KEYS.has(field.key)) .map((field) => field.key); - const touchedTopLevelKeys = topLevelKeys.filter((key) => !!fieldValues[key]); + const touchedTopLevelKeys = topLevelKeys.filter( + (key) => !!fieldValues[key] || clearedKeys[key], + ); const extraConfigKeys = (selectedType?.credentialFields ?? []) .filter((field) => !TOP_LEVEL_KEYS.has(field.key)) .map((field) => field.key); @@ -190,6 +197,9 @@ export const HandleLogProvider = ({ logProviderId }: Props) => { } as any) .then(() => { utils.logProvider.all.invalidate(); + if (logProviderId) { + utils.logProvider.one.invalidate({ logProviderId }); + } const agents = logProviderId && agentsUsingProvider( @@ -307,24 +317,49 @@ export const HandleLogProvider = ({ logProviderId }: Props) => { {field.helpText} )} - - setFieldValues((prev) => ({ - ...prev, - [field.key]: e.target.value, - })) - } - /> +
+ + setFieldValues((prev) => ({ + ...prev, + [field.key]: e.target.value, + })) + } + /> + {logProviderId && + TOP_LEVEL_KEYS.has(field.key) && + !field.required && ( + + )} +
))} diff --git a/apps/dokploy/components/dashboard/settings/log-management/show-log-management-servers.tsx b/apps/dokploy/components/dashboard/settings/log-management/show-log-management-servers.tsx index 166c25fd0..98209463b 100644 --- a/apps/dokploy/components/dashboard/settings/log-management/show-log-management-servers.tsx +++ b/apps/dokploy/components/dashboard/settings/log-management/show-log-management-servers.tsx @@ -17,12 +17,15 @@ import { Checkbox } from "@/components/ui/checkbox"; import { Label } from "@/components/ui/label"; import { api } from "@/utils/api"; -type Status = "running" | "stopped" | "unknown"; +type Status = "running" | "not-running" | "stopped" | "unknown"; const statusBadge = (status: Status) => { if (status === "running") { return Running; } + if (status === "not-running") { + return Not running; + } if (status === "stopped") { return Not deployed; } @@ -117,6 +120,9 @@ export const ShowLogManagementServers = () => { }; const hasProviders = (providers?.length ?? 0) > 0; + const visibleTargets = hasProviders + ? targets + : targets?.filter((target) => target.status !== "stopped"); return (
@@ -130,7 +136,9 @@ export const ShowLogManagementServers = () => { Pick which providers each server ships to, then deploy its Vector agent. Changing a provider only takes effect on a server after you - redeploy it. + redeploy it. Services deployed before Log Management need a + redeploy to get the labels it uses; until then their logs are not + shipped from the Dokploy server. @@ -139,7 +147,7 @@ export const ShowLogManagementServers = () => { Checking servers...
- ) : !hasProviders ? ( + ) : !hasProviders && !visibleTargets?.length ? (
@@ -148,11 +156,11 @@ export const ShowLogManagementServers = () => {
) : (
- {targets?.map((target) => { + {visibleTargets?.map((target) => { const key = keyOf(target.serverId); const ids = idsFor(target.serverId, target.logProviderIds); const isBusy = pendingTarget === key; - const isDeployed = target.status === "running"; + const isDeployed = target.status !== "stopped"; const hasEnabledSelected = ids.some( (id) => providers?.find((p) => p.logProviderId === id)?.enabled, @@ -196,9 +204,11 @@ export const ShowLogManagementServers = () => {
- - Log providers - + {hasProviders && ( + + Log providers + + )}
{providers?.map((provider) => { const ProviderIcon = @@ -274,27 +284,29 @@ export const ShowLogManagementServers = () => { )} -
- {ids.length > 0 && !hasEnabledSelected && ( - - Every selected provider is disabled — nothing would - ship. - - )} - -
+ {hasProviders && ( +
+ {ids.length > 0 && !hasEnabledSelected && ( + + Every selected provider is disabled — nothing + would ship. + + )} + +
+ )}
); })} - {targets?.length === 0 && ( + {visibleTargets?.length === 0 && (
diff --git a/apps/dokploy/server/api/routers/log-provider.ts b/apps/dokploy/server/api/routers/log-provider.ts index 234647958..a532a1f11 100644 --- a/apps/dokploy/server/api/routers/log-provider.ts +++ b/apps/dokploy/server/api/routers/log-provider.ts @@ -111,7 +111,7 @@ export const logProviderRouter = createTRPCRouter({ }); }), serverStatus: withPermission("logProvider", "read").query(({ ctx }) => - getLogManagementServerStatus(ctx.session.activeOrganizationId), + getLogManagementServerStatus(ctx.session), ), deployOnServer: withPermission("logProvider", "create") .input( @@ -122,7 +122,7 @@ export const logProviderRouter = createTRPCRouter({ ) .mutation(async ({ ctx, input }) => { await deployLogManagement( - ctx.session.activeOrganizationId, + ctx.session, input.serverId ?? undefined, input.logProviderIds, ); @@ -137,10 +137,7 @@ export const logProviderRouter = createTRPCRouter({ removeOnServer: withPermission("logProvider", "create") .input(z.object({ serverId: z.string().nullable().optional() })) .mutation(async ({ ctx, input }) => { - await removeLogManagement( - ctx.session.activeOrganizationId, - input.serverId ?? undefined, - ); + await removeLogManagement(ctx.session, input.serverId ?? undefined); await audit(ctx, { action: "delete", resourceType: "server", diff --git a/packages/server/src/services/log-management/service.ts b/packages/server/src/services/log-management/service.ts index 85ff6085a..cd32e112e 100644 --- a/packages/server/src/services/log-management/service.ts +++ b/packages/server/src/services/log-management/service.ts @@ -158,12 +158,11 @@ export const removeLogProvider = async (logProviderId: string) => { }) .where(arrayContains(server.logProviderIds, [logProviderId])); - const remainingLocalIds = sql`array_remove(${webServerSettings.logProviderIds}, ${logProviderId})`; + // Keeps logManagementOrganizationId: the local agent keeps running until it is explicitly removed. await tx .update(webServerSettings) .set({ - logProviderIds: remainingLocalIds, - logManagementOrganizationId: sql`CASE WHEN cardinality(${remainingLocalIds}) = 0 THEN NULL ELSE ${webServerSettings.logManagementOrganizationId} END`, + logProviderIds: sql`array_remove(${webServerSettings.logProviderIds}, ${logProviderId})`, }) .where(arrayContains(webServerSettings.logProviderIds, [logProviderId])); diff --git a/packages/server/src/services/log-management/vector-agent.ts b/packages/server/src/services/log-management/vector-agent.ts index 57700884c..4fa79bec1 100644 --- a/packages/server/src/services/log-management/vector-agent.ts +++ b/packages/server/src/services/log-management/vector-agent.ts @@ -1,12 +1,18 @@ import { IS_CLOUD } from "@dokploy/server/constants"; import { + getAccessibleServerIds, type Server, updateServerLogProviders, } from "@dokploy/server/services/server"; -import { claimWebServerLogManagement } from "@dokploy/server/services/web-server-settings"; +import { + claimWebServerLogManagement, + getWebServerSettings, + releaseWebServerLogManagement, +} from "@dokploy/server/services/web-server-settings"; import { removeVectorAgent, setupVectorAgent, + withVectorTargetLock, } from "@dokploy/server/setup/vector-setup"; import { TRPCError } from "@trpc/server"; import { @@ -24,21 +30,45 @@ const assertLocalAgentAllowed = (serverId: string | undefined) => { } }; +const assertServerAccessible = async ( + serverId: string, + session: { userId: string; activeOrganizationId: string }, +) => { + await assertServerBelongsToOrg(serverId, session.activeOrganizationId); + const accessibleIds = await getAccessibleServerIds(session); + if (!accessibleIds.has(serverId)) { + throw new TRPCError({ + code: "UNAUTHORIZED", + message: "You are not authorized to access this server", + }); + } +}; + export const deployLogManagement = async ( - organizationId: string, + session: { userId: string; activeOrganizationId: string }, serverId: string | undefined, logProviderIds: string[], ) => { + const organizationId = session.activeOrganizationId; assertLocalAgentAllowed(serverId); - await assertServerBelongsToOrg(serverId, organizationId); + if (serverId) { + await assertServerAccessible(serverId, session); + } await assertLogProvidersBelongToOrg(logProviderIds, organizationId); - if (serverId) { - await updateServerLogProviders(serverId, logProviderIds); - } else { + await withVectorTargetLock(serverId, async () => { + if (serverId) { + await setupVectorAgent(organizationId, serverId, logProviderIds); + await updateServerLogProviders(serverId, logProviderIds); + return; + } + + const settings = await getWebServerSettings(); + const previousOwner = settings?.logManagementOrganizationId ?? null; + const previousIds = settings?.logProviderIds ?? []; const claimed = await claimWebServerLogManagement( organizationId, - logProviderIds, + previousOwner ? previousIds : [], ); if (!claimed) { throw new TRPCError({ @@ -47,38 +77,56 @@ export const deployLogManagement = async ( "The local Vector agent is already claimed by another organization", }); } - } - await setupVectorAgent(organizationId, serverId, logProviderIds); + try { + await setupVectorAgent(organizationId, undefined, logProviderIds); + } catch (error) { + if (!previousOwner) { + await releaseWebServerLogManagement(organizationId); + } + throw error; + } + await claimWebServerLogManagement(organizationId, logProviderIds); + }); }; export const removeLogManagement = async ( - organizationId: string, + session: { userId: string; activeOrganizationId: string }, serverId: string | undefined, ) => { + const organizationId = session.activeOrganizationId; assertLocalAgentAllowed(serverId); if (serverId) { - await assertServerBelongsToOrg(serverId, organizationId); - await removeVectorAgent(serverId); - await updateServerLogProviders(serverId, []); + await assertServerAccessible(serverId, session); + await withVectorTargetLock(serverId, async () => { + await removeVectorAgent(serverId); + await updateServerLogProviders(serverId, []); + }); return; } - const released = await claimWebServerLogManagement(organizationId, []); - if (!released) { - throw new TRPCError({ - code: "CONFLICT", - message: "The local Vector agent is claimed by another organization", - }); - } - await removeVectorAgent(); + await withVectorTargetLock(undefined, async () => { + const settings = await getWebServerSettings(); + const owner = settings?.logManagementOrganizationId; + if (owner && owner !== organizationId) { + throw new TRPCError({ + code: "CONFLICT", + message: "The local Vector agent is claimed by another organization", + }); + } + await removeVectorAgent(); + await releaseWebServerLogManagement(organizationId); + }); }; +// Runs even with no providers assigned: deleting the last provider leaves the agent deployed. export const removeServerLogManagement = async (server: Server) => { - if (!server.logProviderIds?.length) return; + if (!server.sshKeyId) return; try { - await removeVectorAgent(server.serverId); + await withVectorTargetLock(server.serverId, () => + removeVectorAgent(server.serverId), + ); } catch (error) { console.error( `[Vector] Failed to remove agent for server ${server.serverId} before deletion:`, diff --git a/packages/server/src/services/organization.ts b/packages/server/src/services/organization.ts index 19e9b87c7..480455ecd 100644 --- a/packages/server/src/services/organization.ts +++ b/packages/server/src/services/organization.ts @@ -1,19 +1,26 @@ import { IS_CLOUD } from "@dokploy/server/constants"; import { db } from "@dokploy/server/db"; import { organization } from "@dokploy/server/db/schema"; -import { findServersWithLogManagementEnabled } from "@dokploy/server/services/server"; +import { findServersByOrganizationForLogManagement } from "@dokploy/server/services/server"; import { - claimWebServerLogManagement, getWebServerSettings, + releaseWebServerLogManagement, } from "@dokploy/server/services/web-server-settings"; -import { removeVectorAgent } from "@dokploy/server/setup/vector-setup"; +import { + removeVectorAgent, + withVectorTargetLock, +} from "@dokploy/server/setup/vector-setup"; import { eq } from "drizzle-orm"; const removeVectorAgents = async (organizationId: string) => { - const servers = await findServersWithLogManagementEnabled(organizationId); + // Every server, not only those with providers: deleting the last provider leaves the agent deployed. + const servers = + await findServersByOrganizationForLogManagement(organizationId); await Promise.all( servers.map((server) => - removeVectorAgent(server.serverId).catch((error) => { + withVectorTargetLock(server.serverId, () => + removeVectorAgent(server.serverId), + ).catch((error) => { console.error( `[Vector] Failed to remove the agent on server ${server.serverId} before deleting organization ${organizationId}:`, error, @@ -23,15 +30,17 @@ const removeVectorAgents = async (organizationId: string) => { ); if (IS_CLOUD) return; - const settings = await getWebServerSettings(); - if (settings?.logManagementOrganizationId !== organizationId) return; - await removeVectorAgent().catch((error) => { - console.error( - `[Vector] Failed to remove the local agent before deleting organization ${organizationId}:`, - error, - ); + await withVectorTargetLock(undefined, async () => { + const settings = await getWebServerSettings(); + if (settings?.logManagementOrganizationId !== organizationId) return; + await removeVectorAgent().catch((error) => { + console.error( + `[Vector] Failed to remove the local agent before deleting organization ${organizationId}:`, + error, + ); + }); + await releaseWebServerLogManagement(organizationId); }); - await claimWebServerLogManagement(organizationId, []); }; export const deleteOrganization = async (organizationId: string) => { diff --git a/packages/server/src/services/server.ts b/packages/server/src/services/server.ts index 443f8a01c..6065464ae 100644 --- a/packages/server/src/services/server.ts +++ b/packages/server/src/services/server.ts @@ -7,7 +7,7 @@ import { } from "@dokploy/server/db/schema"; import { hasValidLicense } from "@dokploy/server/services/proprietary/license-key"; import { TRPCError } from "@trpc/server"; -import { and, desc, eq, isNotNull, sql } from "drizzle-orm"; +import { and, desc, eq, isNotNull } from "drizzle-orm"; import type { z } from "zod"; export type Server = typeof server.$inferSelect; @@ -87,18 +87,6 @@ export const findServersByUserId = async (userId: string) => { return servers; }; -export const findServersWithLogManagementEnabled = async ( - organizationId: string, -) => { - return await db.query.server.findMany({ - where: and( - eq(server.organizationId, organizationId), - sql`array_length(${server.logProviderIds}, 1) > 0`, - ), - columns: { serverId: true }, - }); -}; - export const findServersByOrganizationForLogManagement = async ( organizationId: string, ) => { diff --git a/packages/server/src/services/web-server-settings.ts b/packages/server/src/services/web-server-settings.ts index 7a2f83cdd..ec0e2ba83 100644 --- a/packages/server/src/services/web-server-settings.ts +++ b/packages/server/src/services/web-server-settings.ts @@ -56,8 +56,7 @@ export const claimWebServerLogManagement = async ( .update(webServerSettings) .set({ logProviderIds, - logManagementOrganizationId: - logProviderIds.length > 0 ? organizationId : null, + logManagementOrganizationId: organizationId, updatedAt: new Date(), }) .where( @@ -73,3 +72,14 @@ export const claimWebServerLogManagement = async ( return updated ?? null; }; + +export const releaseWebServerLogManagement = async (organizationId: string) => { + await db + .update(webServerSettings) + .set({ + logProviderIds: [], + logManagementOrganizationId: null, + updatedAt: new Date(), + }) + .where(eq(webServerSettings.logManagementOrganizationId, organizationId)); +}; diff --git a/packages/server/src/setup/vector-setup.ts b/packages/server/src/setup/vector-setup.ts index 7fa5e784a..cfe57582e 100644 --- a/packages/server/src/setup/vector-setup.ts +++ b/packages/server/src/setup/vector-setup.ts @@ -7,6 +7,7 @@ import { import { findServerById, findServersByOrganizationForLogManagement, + getAccessibleServerIds, } from "@dokploy/server/services/server"; import { getWebServerSettings } from "@dokploy/server/services/web-server-settings"; import { encodeBase64 } from "@dokploy/server/utils/docker/utils"; @@ -315,6 +316,26 @@ const deployVectorService = async (serverId?: string) => { } }; +const vectorTargetLocks = new Map>(); + +// In-process only: Dokploy runs a single dashboard process per host. +export const withVectorTargetLock = async ( + serverId: string | undefined, + fn: () => Promise, +): Promise => { + const key = serverId ?? "local"; + const previous = vectorTargetLocks.get(key) ?? Promise.resolve(); + const current = previous.catch(() => {}).then(fn); + vectorTargetLocks.set(key, current); + try { + return await current; + } finally { + if (vectorTargetLocks.get(key) === current) { + vectorTargetLocks.delete(key); + } + } +}; + export const setupVectorAgent = async ( organizationId: string, serverId?: string, @@ -348,39 +369,56 @@ export const removeVectorAgent = async (serverId?: string) => { } }; -const isVectorRunning = async (serverId?: string): Promise => { +type VectorStatus = "running" | "not-running" | "stopped" | "unknown"; + +const inspectVectorStatus = async ( + serverId?: string, +): Promise => { const docker = await getRemoteDocker(serverId); try { - await docker.getService(VECTOR_SERVICE_NAME).inspect(); - return true; - } catch { - return false; + const tasks = await docker.listTasks({ + filters: JSON.stringify({ + service: [VECTOR_SERVICE_NAME], + "desired-state": ["running"], + }), + }); + return tasks.some((task) => task.Status?.State === "running") + ? "running" + : "not-running"; + } catch (error: any) { + if (error?.statusCode === 404) { + return "stopped"; + } + throw error; } }; const VECTOR_CHECK_TIMEOUT_MS = 5000; -const vectorStatus = async ( - serverId: string | null, -): Promise<"running" | "stopped" | "unknown"> => { +const vectorStatus = async (serverId: string | null): Promise => { try { - const running = await Promise.race([ - isVectorRunning(serverId ?? undefined), + return await Promise.race([ + inspectVectorStatus(serverId ?? undefined), new Promise((_, reject) => setTimeout(() => reject(new Error("timeout")), VECTOR_CHECK_TIMEOUT_MS), ), ]); - return running ? "running" : "stopped"; } catch { return "unknown"; } }; -export const getLogManagementServerStatus = async (organizationId: string) => { - const [servers, settings] = await Promise.all([ +export const getLogManagementServerStatus = async (session: { + userId: string; + activeOrganizationId: string; +}) => { + const organizationId = session.activeOrganizationId; + const [allServers, settings, accessibleIds] = await Promise.all([ findServersByOrganizationForLogManagement(organizationId), IS_CLOUD ? Promise.resolve(null) : getWebServerSettings(), + getAccessibleServerIds(session), ]); + const servers = allServers.filter((s) => accessibleIds.has(s.serverId)); const webOwnsThisOrg = !!settings &&