Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ec7b5ebf94 | ||
|
|
ae4afabfea | ||
|
|
114fb22eb8 | ||
|
|
9671218936 | ||
|
|
e1ff84830f |
No files matched your search
@@ -30,9 +30,8 @@ const nextConfig: NextConfig = {
|
||||
serverExternalPackages: ["dockerode", "isomorphic-dompurify", "jsdom"],
|
||||
experimental: {
|
||||
optimizePackageImports: ["@mantine/core", "@mantine/hooks", "@tabler/icons-react"],
|
||||
turbopackFileSystemCacheForDev: true,
|
||||
preloadEntriesOnStart: false,
|
||||
webpackMemoryOptimizations: true,
|
||||
turbopackFileSystemCacheForDev: false,
|
||||
webpackMemoryOptimizations: false,
|
||||
},
|
||||
transpilePackages: ["@homarr/ui", "@homarr/notifications", "@homarr/modals", "@homarr/spotlight", "@homarr/widgets"],
|
||||
images: {
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
const bytesToMiB = (bytes: number) => Number((bytes / (1024 * 1024)).toFixed(2));
|
||||
|
||||
const getActiveHandlesCount = () => {
|
||||
const processWithHandles = process as NodeJS.Process & {
|
||||
_getActiveHandles?: () => unknown[];
|
||||
};
|
||||
return processWithHandles._getActiveHandles?.().length ?? null;
|
||||
};
|
||||
|
||||
const getActiveRequestsCount = () => {
|
||||
const processWithRequests = process as NodeJS.Process & {
|
||||
_getActiveRequests?: () => unknown[];
|
||||
};
|
||||
return processWithRequests._getActiveRequests?.().length ?? null;
|
||||
};
|
||||
|
||||
export const getTasksRuntimeDiagnostics = () => {
|
||||
const memory = process.memoryUsage();
|
||||
const resources = process.resourceUsage();
|
||||
|
||||
return {
|
||||
rssMiB: bytesToMiB(memory.rss),
|
||||
heapUsedMiB: bytesToMiB(memory.heapUsed),
|
||||
heapTotalMiB: bytesToMiB(memory.heapTotal),
|
||||
externalMiB: bytesToMiB(memory.external),
|
||||
arrayBuffersMiB: bytesToMiB(memory.arrayBuffers),
|
||||
uptimeSeconds: Number(process.uptime().toFixed(0)),
|
||||
userCpuMicros: resources.userCPUTime,
|
||||
systemCpuMicros: resources.systemCPUTime,
|
||||
maxRssKiB: resources.maxRSS,
|
||||
activeHandles: getActiveHandlesCount(),
|
||||
activeRequests: getActiveRequestsCount(),
|
||||
};
|
||||
};
|
||||
@@ -2,11 +2,41 @@
|
||||
import "./overrides";
|
||||
|
||||
import { jobGroup } from "@homarr/cron-jobs";
|
||||
import { createLogger } from "@homarr/core/infrastructure/logs";
|
||||
|
||||
import { getTasksRuntimeDiagnostics } from "./diagnostics";
|
||||
import { onStartAsync } from "./on-start";
|
||||
|
||||
const logger = createLogger({ module: "tasksMain" });
|
||||
const RUNTIME_DIAGNOSTICS_INTERVAL_MS = 60_000;
|
||||
|
||||
const scheduleRuntimeDiagnostics = () => {
|
||||
setInterval(() => {
|
||||
logger.info("Tasks runtime diagnostics", {
|
||||
diagnostics: getTasksRuntimeDiagnostics(),
|
||||
});
|
||||
}, RUNTIME_DIAGNOSTICS_INTERVAL_MS).unref();
|
||||
};
|
||||
|
||||
void (async () => {
|
||||
logger.info("Tasks runtime bootstrap started", {
|
||||
diagnostics: getTasksRuntimeDiagnostics(),
|
||||
});
|
||||
|
||||
await onStartAsync();
|
||||
logger.info("Tasks on-start hooks completed", {
|
||||
diagnostics: getTasksRuntimeDiagnostics(),
|
||||
});
|
||||
|
||||
await jobGroup.initializeAsync();
|
||||
logger.info("Cron job group initialized in tasks runtime", {
|
||||
diagnostics: getTasksRuntimeDiagnostics(),
|
||||
});
|
||||
|
||||
await jobGroup.startAllAsync();
|
||||
logger.info("Cron job group started in tasks runtime", {
|
||||
diagnostics: getTasksRuntimeDiagnostics(),
|
||||
jobs: jobGroup.getKeys(),
|
||||
});
|
||||
scheduleRuntimeDiagnostics();
|
||||
})();
|
||||
@@ -1,5 +1,8 @@
|
||||
import { setGlobalDispatcher } from "undici";
|
||||
|
||||
import { UndiciHttpAgent } from "@homarr/core/infrastructure/http";
|
||||
import { createLogger } from "@homarr/core/infrastructure/logs";
|
||||
|
||||
const logger = createLogger({ module: "tasksOverrides" });
|
||||
setGlobalDispatcher(new UndiciHttpAgent());
|
||||
logger.info("Configured global undici dispatcher for tasks runtime");
|
||||
@@ -9,11 +9,18 @@ import {
|
||||
getAllTrustedCertificatesAsync,
|
||||
getTrustedCertificateHostnamesAsync,
|
||||
} from "@homarr/core/infrastructure/certificates";
|
||||
import { UndiciHttpAgent } from "@homarr/core/infrastructure/http";
|
||||
|
||||
import type { TrustedCertificateHostname } from "../certificates/hostnames";
|
||||
import { UndiciHttpAgent } from "./http-agent";
|
||||
import { withTimeoutAsync } from "./timeout";
|
||||
|
||||
const CERTIFICATE_DISPATCHER_TTL_MS = 5 * 60 * 1000;
|
||||
let trustedCertificateDispatcher: {
|
||||
dispatcher: UndiciHttpAgent;
|
||||
expiresAt: number;
|
||||
} | null = null;
|
||||
let creatingTrustedCertificateDispatcherPromise: Promise<UndiciHttpAgent> | null = null;
|
||||
|
||||
export const createCustomCheckServerIdentity = (
|
||||
trustedHostnames: TrustedCertificateHostname[],
|
||||
): typeof checkServerIdentity => {
|
||||
@@ -41,6 +48,36 @@ export const createCertificateAgentAsync = async (override?: {
|
||||
});
|
||||
};
|
||||
|
||||
const getSharedCertificateAgentAsync = async () => {
|
||||
const now = Date.now();
|
||||
if (trustedCertificateDispatcher && trustedCertificateDispatcher.expiresAt > now) {
|
||||
return trustedCertificateDispatcher.dispatcher;
|
||||
}
|
||||
if (creatingTrustedCertificateDispatcherPromise) {
|
||||
return await creatingTrustedCertificateDispatcherPromise;
|
||||
}
|
||||
|
||||
creatingTrustedCertificateDispatcherPromise = createCertificateAgentAsync(undefined)
|
||||
.then((dispatcher) => {
|
||||
const previousDispatcher = trustedCertificateDispatcher?.dispatcher;
|
||||
trustedCertificateDispatcher = {
|
||||
dispatcher,
|
||||
expiresAt: Date.now() + CERTIFICATE_DISPATCHER_TTL_MS,
|
||||
};
|
||||
if (previousDispatcher) {
|
||||
void previousDispatcher.close().catch(() => {
|
||||
previousDispatcher.destroy();
|
||||
});
|
||||
}
|
||||
return dispatcher;
|
||||
})
|
||||
.finally(() => {
|
||||
creatingTrustedCertificateDispatcherPromise = null;
|
||||
});
|
||||
|
||||
return await creatingTrustedCertificateDispatcherPromise;
|
||||
};
|
||||
|
||||
export const createHttpsAgentAsync = async (override?: Pick<AgentOptions, "ca" | "checkServerIdentity">) => {
|
||||
return new HttpsAgent({
|
||||
ca: await getAllTrustedCertificatesAsync(),
|
||||
@@ -63,7 +100,7 @@ export const fetchWithTrustedCertificatesAsync = async (
|
||||
url: RequestInfo,
|
||||
options?: RequestInit & { timeout?: number },
|
||||
): Promise<Response> => {
|
||||
const agent = await createCertificateAgentAsync(undefined);
|
||||
const agent = await getSharedCertificateAgentAsync();
|
||||
if (options?.timeout) {
|
||||
return await withTimeoutAsync(
|
||||
async (signal) =>
|
||||
|
||||
@@ -7,6 +7,7 @@ import { db } from "@homarr/db";
|
||||
|
||||
import type { Logger } from "./logger";
|
||||
import type { ValidateCron } from "./validation";
|
||||
import { createProcessDiagnosticsDelta, createProcessDiagnosticsSnapshot } from "./diagnostics";
|
||||
|
||||
export interface CreateCronJobCreatorOptions<TAllowedNames extends string> {
|
||||
beforeCallback?: (name: TAllowedNames) => MaybePromise<void>;
|
||||
@@ -33,19 +34,25 @@ const createCallback = <TAllowedNames extends string, TName extends TAllowedName
|
||||
const expectedMaximumDurationInMillis = options.expectedMaximumDurationInMillis ?? 2500;
|
||||
return (callback: () => MaybePromise<void>) => {
|
||||
const catchingCallbackAsync = async () => {
|
||||
const beforeExecutionSnapshot = createProcessDiagnosticsSnapshot();
|
||||
try {
|
||||
creatorOptions.logger.logDebug("The callback of cron job started", {
|
||||
name,
|
||||
diagnostics: beforeExecutionSnapshot,
|
||||
});
|
||||
const stopwatch = new Stopwatch();
|
||||
await creatorOptions.beforeCallback?.(name);
|
||||
const beforeCallbackTook = stopwatch.getElapsedInHumanWords();
|
||||
await callback();
|
||||
const afterExecutionSnapshot = createProcessDiagnosticsSnapshot();
|
||||
const callbackTook = stopwatch.getElapsedInHumanWords();
|
||||
creatorOptions.logger.logDebug("The callback of cron job succeeded", {
|
||||
name,
|
||||
beforeCallbackTook,
|
||||
callbackTook,
|
||||
diagnosticsBefore: beforeExecutionSnapshot,
|
||||
diagnosticsAfter: afterExecutionSnapshot,
|
||||
diagnosticsDelta: createProcessDiagnosticsDelta(beforeExecutionSnapshot, afterExecutionSnapshot),
|
||||
});
|
||||
|
||||
const durationInMillis = stopwatch.getElapsedInMilliseconds();
|
||||
@@ -58,11 +65,15 @@ const createCallback = <TAllowedNames extends string, TName extends TAllowedName
|
||||
}
|
||||
await creatorOptions.onCallbackSuccess?.(name);
|
||||
} catch (error) {
|
||||
const failedExecutionSnapshot = createProcessDiagnosticsSnapshot();
|
||||
creatorOptions.logger.logError(
|
||||
new ErrorWithMetadata(
|
||||
"The callback of cron job failed",
|
||||
{
|
||||
name,
|
||||
diagnosticsBefore: beforeExecutionSnapshot,
|
||||
diagnosticsAfter: failedExecutionSnapshot,
|
||||
diagnosticsDelta: createProcessDiagnosticsDelta(beforeExecutionSnapshot, failedExecutionSnapshot),
|
||||
},
|
||||
{ cause: error },
|
||||
),
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
const bytesToMiB = (bytes: number) => Number((bytes / (1024 * 1024)).toFixed(2));
|
||||
|
||||
const getActiveHandlesCount = () => {
|
||||
const processWithHandles = process as NodeJS.Process & {
|
||||
_getActiveHandles?: () => unknown[];
|
||||
};
|
||||
return processWithHandles._getActiveHandles?.().length ?? null;
|
||||
};
|
||||
|
||||
const getActiveRequestsCount = () => {
|
||||
const processWithRequests = process as NodeJS.Process & {
|
||||
_getActiveRequests?: () => unknown[];
|
||||
};
|
||||
return processWithRequests._getActiveRequests?.().length ?? null;
|
||||
};
|
||||
|
||||
export interface ProcessDiagnosticsSnapshot {
|
||||
rssMiB: number;
|
||||
heapUsedMiB: number;
|
||||
heapTotalMiB: number;
|
||||
externalMiB: number;
|
||||
arrayBuffersMiB: number;
|
||||
uptimeSeconds: number;
|
||||
userCpuMicros: number;
|
||||
systemCpuMicros: number;
|
||||
maxRssKiB: number;
|
||||
activeHandles: number | null;
|
||||
activeRequests: number | null;
|
||||
}
|
||||
|
||||
export const createProcessDiagnosticsSnapshot = (): ProcessDiagnosticsSnapshot => {
|
||||
const memory = process.memoryUsage();
|
||||
const resources = process.resourceUsage();
|
||||
|
||||
return {
|
||||
rssMiB: bytesToMiB(memory.rss),
|
||||
heapUsedMiB: bytesToMiB(memory.heapUsed),
|
||||
heapTotalMiB: bytesToMiB(memory.heapTotal),
|
||||
externalMiB: bytesToMiB(memory.external),
|
||||
arrayBuffersMiB: bytesToMiB(memory.arrayBuffers),
|
||||
uptimeSeconds: Number(process.uptime().toFixed(0)),
|
||||
userCpuMicros: resources.userCPUTime,
|
||||
systemCpuMicros: resources.systemCPUTime,
|
||||
maxRssKiB: resources.maxRSS,
|
||||
activeHandles: getActiveHandlesCount(),
|
||||
activeRequests: getActiveRequestsCount(),
|
||||
};
|
||||
};
|
||||
|
||||
export const createProcessDiagnosticsDelta = (
|
||||
before: ProcessDiagnosticsSnapshot,
|
||||
after: ProcessDiagnosticsSnapshot,
|
||||
) => {
|
||||
return {
|
||||
rssMiB: Number((after.rssMiB - before.rssMiB).toFixed(2)),
|
||||
heapUsedMiB: Number((after.heapUsedMiB - before.heapUsedMiB).toFixed(2)),
|
||||
heapTotalMiB: Number((after.heapTotalMiB - before.heapTotalMiB).toFixed(2)),
|
||||
externalMiB: Number((after.externalMiB - before.externalMiB).toFixed(2)),
|
||||
arrayBuffersMiB: Number((after.arrayBuffersMiB - before.arrayBuffersMiB).toFixed(2)),
|
||||
uptimeSeconds: Number((after.uptimeSeconds - before.uptimeSeconds).toFixed(0)),
|
||||
userCpuMicros: after.userCpuMicros - before.userCpuMicros,
|
||||
systemCpuMicros: after.systemCpuMicros - before.systemCpuMicros,
|
||||
maxRssKiB: after.maxRssKiB - before.maxRssKiB,
|
||||
activeHandles:
|
||||
before.activeHandles === null || after.activeHandles === null ? null : after.activeHandles - before.activeHandles,
|
||||
activeRequests:
|
||||
before.activeRequests === null || after.activeRequests === null
|
||||
? null
|
||||
: after.activeRequests - before.activeRequests,
|
||||
};
|
||||
};
|
||||
@@ -4,6 +4,7 @@ import { objectEntries, objectKeys } from "@homarr/common";
|
||||
import { db } from "@homarr/db";
|
||||
|
||||
import type { JobCallback } from "./creator";
|
||||
import { createProcessDiagnosticsSnapshot } from "./diagnostics";
|
||||
import type { Logger } from "./logger";
|
||||
import { jobRegistry } from "./registry";
|
||||
|
||||
@@ -38,6 +39,10 @@ export const createJobGroupCreator = <TAllowedNames extends string = string>(
|
||||
|
||||
return {
|
||||
initializeAsync: async () => {
|
||||
options.logger.logInfo("Initializing cron job group", {
|
||||
registeredJobs: [...jobRegistry.keys()],
|
||||
diagnostics: createProcessDiagnosticsSnapshot(),
|
||||
});
|
||||
const configurations = await db.query.cronJobConfigurations.findMany();
|
||||
for (const job of jobRegistry.values()) {
|
||||
const configuration = configurations.find(({ name }) => name === job.name);
|
||||
@@ -51,7 +56,15 @@ export const createJobGroupCreator = <TAllowedNames extends string = string>(
|
||||
const scheduledTask = await job.createTaskAsync();
|
||||
|
||||
tasks.set(job.name, scheduledTask);
|
||||
options.logger.logDebug("Created scheduled task for cron job", {
|
||||
name: job.name,
|
||||
diagnostics: createProcessDiagnosticsSnapshot(),
|
||||
});
|
||||
}
|
||||
options.logger.logInfo("Cron job group initialized", {
|
||||
taskCount: tasks.size,
|
||||
diagnostics: createProcessDiagnosticsSnapshot(),
|
||||
});
|
||||
},
|
||||
startAsync: async (name: keyof TJobs) => {
|
||||
const job = jobRegistry.get(name as string);
|
||||
@@ -65,6 +78,10 @@ export const createJobGroupCreator = <TAllowedNames extends string = string>(
|
||||
tasks.get(name as string)?.resume();
|
||||
},
|
||||
startAllAsync: async () => {
|
||||
options.logger.logInfo("Starting all cron jobs", {
|
||||
taskCount: tasks.size,
|
||||
diagnostics: createProcessDiagnosticsSnapshot(),
|
||||
});
|
||||
for (const job of jobRegistry.values()) {
|
||||
if (!tasks.has(job.name)) {
|
||||
continue;
|
||||
@@ -76,6 +93,10 @@ export const createJobGroupCreator = <TAllowedNames extends string = string>(
|
||||
await job.onStartAsync();
|
||||
tasks.get(job.name)?.resume();
|
||||
}
|
||||
options.logger.logInfo("Started all cron jobs", {
|
||||
taskCount: tasks.size,
|
||||
diagnostics: createProcessDiagnosticsSnapshot(),
|
||||
});
|
||||
},
|
||||
runManuallyAsync: async (name: keyof TJobs) => {
|
||||
const job = jobRegistry.get(name as string);
|
||||
|
||||
@@ -9,6 +9,7 @@ import { ErrorWithMetadata } from "@homarr/core/infrastructure/logs/error";
|
||||
import { createGetSetChannel } from "@homarr/redis";
|
||||
|
||||
const logger = createLogger({ module: "imageProxy" });
|
||||
const IMAGE_PROXY_ENTRY_TTL_SECONDS = 60 * 60 * 24 * 7;
|
||||
|
||||
const createHmacChannel = (hmac: `${string}.${string}`) => createGetSetChannel<string>(`image-proxy:hmac:${hmac}`);
|
||||
const createUrlByIdChannel = (id: string) =>
|
||||
@@ -31,12 +32,17 @@ export class ImageProxy {
|
||||
public async createImageAsync(url: string, headers?: Record<string, string>): Promise<string> {
|
||||
const existingId = await this.getExistingIdAsync(url, headers);
|
||||
if (existingId) {
|
||||
logger.debug("Image already exists in the proxy", {
|
||||
id: existingId,
|
||||
url: this.redactUrl(url),
|
||||
headers: this.redactHeaders(headers ?? null),
|
||||
});
|
||||
return this.createImageUrl(existingId);
|
||||
const existingImageData = await createUrlByIdChannel(existingId).getAsync();
|
||||
if (!existingImageData) {
|
||||
await this.removeHashForImageAsync(url, headers);
|
||||
} else {
|
||||
logger.debug("Image already exists in the proxy", {
|
||||
id: existingId,
|
||||
url: this.redactUrl(url),
|
||||
headers: this.redactHeaders(headers ?? null),
|
||||
});
|
||||
return this.createImageUrl(existingId);
|
||||
}
|
||||
}
|
||||
|
||||
const id = createId();
|
||||
@@ -116,8 +122,10 @@ export class ImageProxy {
|
||||
await urlHeaderChannel.setAsync({
|
||||
url: encryptSecret(url),
|
||||
headers: encryptSecret(JSON.stringify(headers ?? null)),
|
||||
}, {
|
||||
ttlSeconds: IMAGE_PROXY_ENTRY_TTL_SECONDS,
|
||||
});
|
||||
await hashChannel.setAsync(id);
|
||||
await hashChannel.setAsync(id, { ttlSeconds: IMAGE_PROXY_ENTRY_TTL_SECONDS });
|
||||
|
||||
logger.debug("Stored image in the proxy", {
|
||||
id,
|
||||
@@ -126,6 +134,13 @@ export class ImageProxy {
|
||||
});
|
||||
}
|
||||
|
||||
private async removeHashForImageAsync(url: string, headers: Record<string, string> | undefined): Promise<void> {
|
||||
const urlHash = this.hashSecret(url);
|
||||
const headerHash = this.hashSecret(JSON.stringify(headers ?? null));
|
||||
|
||||
await createHmacChannel(`${urlHash}.${headerHash}`).removeAsync();
|
||||
}
|
||||
|
||||
private hashSecret(value: string): string {
|
||||
return createHmac("sha256", ImageProxy.getHmacKey()).update(value).digest("hex");
|
||||
}
|
||||
|
||||
@@ -113,8 +113,14 @@ export const createGetSetChannel = <TData>(name: string) => {
|
||||
* Set data in the channel
|
||||
* @param data data to be stored in the channel
|
||||
*/
|
||||
setAsync: async (data: TData) => {
|
||||
await getSetClient.set(name, superjson.stringify(data));
|
||||
setAsync: async (data: TData, options?: { ttlSeconds?: number }) => {
|
||||
const ttlSeconds = options?.ttlSeconds;
|
||||
const serializedData = superjson.stringify(data);
|
||||
if (ttlSeconds && ttlSeconds > 0) {
|
||||
await getSetClient.set(name, serializedData, "EX", Math.ceil(ttlSeconds));
|
||||
return;
|
||||
}
|
||||
await getSetClient.set(name, serializedData);
|
||||
},
|
||||
/**
|
||||
* Remove data from the channel
|
||||
|
||||
Reference in new issue
Block a user