🚧 back: WIP Initial diagnostic endpoint (#762)
This commit is contained in:
@@ -16,6 +16,14 @@
|
||||
"logger": {
|
||||
"level": "LOG_LEVEL"
|
||||
},
|
||||
"diagnostics": {
|
||||
"skipKeys": {
|
||||
"__name": "DIAG_SKIP_KEYS",
|
||||
"__format": "json"
|
||||
},
|
||||
"probeSecret": "DIAG_PROBE_SECRET",
|
||||
"secret": "DIAG_SECRET"
|
||||
},
|
||||
"webserver": {
|
||||
"host": "TWAKE_DRIVE_HOST",
|
||||
"logger": {
|
||||
|
||||
@@ -35,6 +35,12 @@
|
||||
"logger": {
|
||||
"level": "debug"
|
||||
},
|
||||
"diagnostics": {
|
||||
"skipKeys": [],
|
||||
"probeSecret": "",
|
||||
"secret": "",
|
||||
"secretChallengeRefreshS": 15
|
||||
},
|
||||
"tracker": {
|
||||
"type": "segment",
|
||||
"segment": {
|
||||
@@ -226,6 +232,7 @@
|
||||
},
|
||||
"services": [
|
||||
"auth",
|
||||
"diagnostics",
|
||||
"push",
|
||||
"storage",
|
||||
"webserver",
|
||||
|
||||
@@ -0,0 +1,218 @@
|
||||
import assert from "node:assert";
|
||||
import config from "../../../config";
|
||||
|
||||
/**
|
||||
* Values that can match a set of diagnostic providers.
|
||||
*
|
||||
* `startup`, `ready` and `live` are meant to match the meanings of the corresponding
|
||||
* kubernetes probes:
|
||||
* https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/
|
||||
*/
|
||||
export type TDiagnosticTag =
|
||||
| "*" // Special value that includes everything, and is always included
|
||||
| "startup" // Tests that are absolutely required (and light) to even begin the other tests
|
||||
| "ready" // Tests required before traffic can be sent our way
|
||||
| "live" // Tests required to prevent from being restarted
|
||||
| "stats"; // Expensive diagnostics that should not often be ran
|
||||
|
||||
/** Detail requested from platform service self-diagnostics */
|
||||
export enum TServiceDiagnosticDepth {
|
||||
/** Minimal cost information that tests functioning service */
|
||||
critical = 0,
|
||||
/** Statistics that have a little impact enough for regular tracking into a time series */
|
||||
tracked_statistics = 1,
|
||||
/** Statistics possibly expensive and large to calculate, for occasional debug operations */
|
||||
deep_statistics = 2,
|
||||
}
|
||||
|
||||
interface IDiagnosticsConfig {
|
||||
// Diagnostic keys that should be considered ok without evaluation
|
||||
skipKeys?: string[];
|
||||
// This secret must be provided to the diagnostic endpoints as a query param
|
||||
// therefor it's likely to leak, through logs etc, for ex. and should not
|
||||
// relied on for security because disabling diagnostics. At worst this
|
||||
// provides access to the DB statistics.
|
||||
probeSecret?: string;
|
||||
// This secret is required to sign more dangerous diagnostic endpoints, such
|
||||
// the heap snapshot. It should never be sent over the network.
|
||||
secret?: string;
|
||||
// Maximum time to keep the same challenge token for diagnostic endpoints, in seconds.
|
||||
// Must be large enough to have a reasonable chance of running the token request
|
||||
// then the action on the same backend instance.
|
||||
secretChallengeRefreshS: number;
|
||||
}
|
||||
|
||||
export const getConfig = (): IDiagnosticsConfig => {
|
||||
let configSection = config.get("diagnostics") as IDiagnosticsConfig;
|
||||
if (typeof configSection.skipKeys === "string")
|
||||
configSection = {
|
||||
...configSection,
|
||||
skipKeys: (configSection.skipKeys as string)
|
||||
.trim()
|
||||
.split(/[,\s]+/g)
|
||||
.filter(x => !!x),
|
||||
};
|
||||
if (typeof configSection.secretChallengeRefreshS === "string")
|
||||
configSection = {
|
||||
...configSection,
|
||||
secretChallengeRefreshS: parseInt(configSection.secretChallengeRefreshS, 10),
|
||||
};
|
||||
return configSection;
|
||||
};
|
||||
|
||||
/** Code-wide unique key for each provider */
|
||||
export type TDiagnosticKey = string;
|
||||
|
||||
/** Each provider should return an object of this format. The key of the provider defines the schema. */
|
||||
export type TDiagnosticResult = { ok: boolean; warn?: string } & { [key: string]: unknown };
|
||||
|
||||
/** Implemented by objects that want to provide data to the diagnostic check */
|
||||
export interface IDiagnosticProvider {
|
||||
/** Code-wide unique key underwhich the result of `get` will be included */
|
||||
key: TDiagnosticKey;
|
||||
|
||||
/** This result is present in any included request tag */
|
||||
tags: TDiagnosticTag[] | "*";
|
||||
|
||||
/**
|
||||
* If set, this provider will be polled at that interval.
|
||||
* If `undefined`, this provider will be ran at each request.
|
||||
*/
|
||||
pollPeriodMs?: number;
|
||||
|
||||
/**
|
||||
* Returns an object as presented to a diagnostic requester.
|
||||
* Warning: this could be public and readable to the internet.
|
||||
* @param completeButSlow If `true`, perform additional operations for a
|
||||
* more informative
|
||||
*/
|
||||
get(): Promise<TDiagnosticResult>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Platform services that can provide generic diagnostic implementations may use this interface.
|
||||
*
|
||||
* Matching from {@link TDiagnosticTag} to {@link TServiceDiagnosticDepth} is expected to
|
||||
* be done by intermediary providers. This is because not all services are equally critical,
|
||||
* or have the same tolerable down times.
|
||||
*/
|
||||
export interface IServiceDiagnosticProvider {
|
||||
/** The return format is specific to each service, but should include a `{ok: boolean}` field. */
|
||||
getDiagnostics(depth: TServiceDiagnosticDepth): Promise<TDiagnosticResult>;
|
||||
}
|
||||
|
||||
const isProviderIncludedInTag = (
|
||||
tag: TDiagnosticTag,
|
||||
provider: IDiagnosticProvider,
|
||||
config: IDiagnosticsConfig,
|
||||
) =>
|
||||
(provider.tags === "*" || provider.tags.indexOf(tag) >= 0 || provider.tags.indexOf("*") >= 0) &&
|
||||
(!config.skipKeys?.length || !config.skipKeys.includes(provider.key));
|
||||
|
||||
// registered providers with `pollPeriodMs === undefined`
|
||||
const immediateDiagnosticProviders: IDiagnosticProvider[] = [];
|
||||
// registered providers with `pollPeriodMs !== undefined`
|
||||
const periodicDiagnosticProviders: IDiagnosticProvider[] = [];
|
||||
|
||||
const now = () => Math.round(process.uptime() * 1000);
|
||||
|
||||
const isKeyAlreadyRegistered = (key: TDiagnosticKey) =>
|
||||
immediateDiagnosticProviders.some(provider => key == provider.key) ||
|
||||
periodicDiagnosticProviders.some(provider => key == provider.key);
|
||||
|
||||
// stores results of all the `pollPeriodMs` truthy providers
|
||||
const latestPeriodicDiagnostics: { [key: TDiagnosticKey]: object } = {};
|
||||
const recordDiagnostic = (startMs: number, key: TDiagnosticKey, data?: object, error?: object) =>
|
||||
(latestPeriodicDiagnostics[key] = {
|
||||
durationMs: Math.round(now() - startMs),
|
||||
...(error ? { ok: false, error } : { ...data }),
|
||||
});
|
||||
|
||||
const runProvider = async provider => {
|
||||
const startMs = now();
|
||||
try {
|
||||
const result = await provider.get();
|
||||
if (!result.ok || result.warn)
|
||||
logger.error(
|
||||
{ provider: provider.key, result },
|
||||
"Got diagnostic provider result with ok=false",
|
||||
);
|
||||
else if (result.warn)
|
||||
logger.warn(
|
||||
{ provider: provider.key, result },
|
||||
"Got diagnostic provider result with ok=true but a warning",
|
||||
);
|
||||
return recordDiagnostic(startMs, provider.key, result);
|
||||
} catch (error) {
|
||||
return recordDiagnostic(startMs, provider.key, undefined, error);
|
||||
}
|
||||
};
|
||||
|
||||
const pendingTimeouts: number[] = []; // Pending return values from `setTimeout` calls
|
||||
const forgetPendingTimeout = (timeoutId: number) => {
|
||||
const index = pendingTimeouts.indexOf(timeoutId);
|
||||
assert(index >= 0);
|
||||
pendingTimeouts.splice(index, 1);
|
||||
};
|
||||
|
||||
let hasShutdown = false;
|
||||
const ensureHasntShutdown = () => {
|
||||
if (hasShutdown) throw new Error("Diagnostics service already shutdown");
|
||||
};
|
||||
|
||||
export default {
|
||||
/** Add a provider to be included in diagnostics output */
|
||||
registerProviders(...providers: IDiagnosticProvider[]) {
|
||||
ensureHasntShutdown();
|
||||
providers.forEach(provider => {
|
||||
if (isKeyAlreadyRegistered(provider.key)) throw new Error("Provider with duplicate key");
|
||||
if (provider.pollPeriodMs) {
|
||||
periodicDiagnosticProviders.push(provider);
|
||||
} else {
|
||||
immediateDiagnosticProviders.push(provider);
|
||||
return;
|
||||
}
|
||||
let triggerUpdate: () => void = () => undefined; // The empty function is for the linter. I love you linter <3
|
||||
const updateProvider = (timeoutId: number) => async () => {
|
||||
forgetPendingTimeout(timeoutId);
|
||||
await runProvider(provider);
|
||||
triggerUpdate();
|
||||
};
|
||||
triggerUpdate = () => pendingTimeouts.push(setTimeout(updateProvider, provider.pollPeriodMs));
|
||||
triggerUpdate();
|
||||
});
|
||||
},
|
||||
|
||||
/** Cancel all pending diagnostic updates */
|
||||
shutdown() {
|
||||
ensureHasntShutdown();
|
||||
pendingTimeouts.forEach(timeout => clearTimeout(timeout));
|
||||
hasShutdown = true;
|
||||
},
|
||||
|
||||
/** Return the values of all providers which include the provided tag */
|
||||
async get(
|
||||
tag: TDiagnosticTag,
|
||||
): Promise<{ ok: boolean } | { [key: TDiagnosticKey]: TDiagnosticResult }> {
|
||||
const config = getConfig();
|
||||
const result = { ok: true };
|
||||
let atLeastOneCheck = false;
|
||||
periodicDiagnosticProviders.forEach(provider => {
|
||||
if (!isProviderIncludedInTag(tag, provider, config)) return;
|
||||
atLeastOneCheck = true;
|
||||
result[provider.key] = latestPeriodicDiagnostics[provider.key];
|
||||
if (!result[provider.key].ok) result.ok = false;
|
||||
});
|
||||
await Promise.all(
|
||||
immediateDiagnosticProviders.map(async provider => {
|
||||
if (!isProviderIncludedInTag(tag, provider, config)) return;
|
||||
atLeastOneCheck = true;
|
||||
const providerResult = await runProvider(provider);
|
||||
if (!providerResult.ok) result.ok = false;
|
||||
return (result[provider.key] = providerResult);
|
||||
}),
|
||||
);
|
||||
if (!atLeastOneCheck) result.ok = false;
|
||||
return result;
|
||||
},
|
||||
};
|
||||
@@ -1,10 +1,11 @@
|
||||
import { TdriveServiceProvider } from "../../framework";
|
||||
import type { IServiceDiagnosticProvider } from "../../framework/api/diagnostics";
|
||||
import { Connector } from "./services/orm/connectors";
|
||||
import Manager from "./services/orm/manager";
|
||||
import Repository from "./services/orm/repository/repository";
|
||||
import { EntityTarget } from "./services/orm/types";
|
||||
|
||||
export interface DatabaseServiceAPI extends TdriveServiceProvider {
|
||||
export interface DatabaseServiceAPI extends TdriveServiceProvider, IServiceDiagnosticProvider {
|
||||
/**
|
||||
* Get the database connector
|
||||
*/
|
||||
|
||||
@@ -7,6 +7,10 @@ import { MongoConnectionOptions } from "./orm/connectors/mongodb/mongodb";
|
||||
import { EntityTarget } from "./orm/types";
|
||||
import { RepositoryManager } from "./orm/repository/manager";
|
||||
import { PostgresConnectionOptions } from "./orm/connectors/postgres/postgres";
|
||||
import type {
|
||||
TDiagnosticResult,
|
||||
TServiceDiagnosticDepth,
|
||||
} from "../../../framework/api/diagnostics";
|
||||
|
||||
export default class DatabaseService implements DatabaseServiceAPI {
|
||||
version = "1";
|
||||
@@ -38,6 +42,15 @@ export default class DatabaseService implements DatabaseServiceAPI {
|
||||
}
|
||||
}
|
||||
|
||||
async getDiagnostics(depth: TServiceDiagnosticDepth): Promise<TDiagnosticResult> {
|
||||
const connector = this.getConnector();
|
||||
const result = await connector.getDiagnostics(depth);
|
||||
return {
|
||||
type: connector.getType(),
|
||||
...result,
|
||||
};
|
||||
}
|
||||
|
||||
getManager(): Manager<unknown> {
|
||||
return new Manager<unknown>(this.connector);
|
||||
}
|
||||
|
||||
+5
@@ -4,12 +4,17 @@ import { ConnectionOptions, DatabaseType } from "../..";
|
||||
import { FindOptions } from "../repository/repository";
|
||||
import { ColumnDefinition, EntityDefinition } from "../types";
|
||||
import { ListResult, Paginable, Pagination } from "../../../../../framework/api/crud-service";
|
||||
import type {
|
||||
TDiagnosticResult,
|
||||
TServiceDiagnosticDepth,
|
||||
} from "../../../../../framework/api/diagnostics";
|
||||
|
||||
export abstract class AbstractConnector<T extends ConnectionOptions> implements Connector {
|
||||
constructor(protected type: DatabaseType, protected options: T, protected secret: string) {}
|
||||
|
||||
abstract connect(): Promise<this>;
|
||||
abstract disconnect(): Promise<this>;
|
||||
abstract getDiagnostics(depth: TServiceDiagnosticDepth): Promise<TDiagnosticResult>;
|
||||
|
||||
abstract drop(): Promise<this>;
|
||||
|
||||
|
||||
+2
-1
@@ -5,6 +5,7 @@ import { ColumnDefinition, EntityDefinition } from "../types";
|
||||
import { FindOptions } from "../repository/repository";
|
||||
import { ListResult, Paginable, Pagination } from "../../../../../framework/api/crud-service";
|
||||
import { PostgresConnectionOptions } from "./postgres/postgres";
|
||||
import type { IServiceDiagnosticProvider } from "../../../../../framework/api/diagnostics";
|
||||
|
||||
export * from "./mongodb/mongodb";
|
||||
|
||||
@@ -14,7 +15,7 @@ export type UpsertOptions = {
|
||||
|
||||
export type RemoveOptions = any;
|
||||
|
||||
export interface Connector extends Initializable {
|
||||
export interface Connector extends Initializable, IServiceDiagnosticProvider {
|
||||
/**
|
||||
* Connect to the database
|
||||
*/
|
||||
|
||||
+22
@@ -8,6 +8,10 @@ import { AbstractConnector } from "../abstract-connector";
|
||||
import { buildSelectQuery } from "./query-builder";
|
||||
import { transformValueFromDbString, transformValueToDbString } from "./typeTransforms";
|
||||
import { logger } from "../../../../../../framework";
|
||||
import {
|
||||
TDiagnosticResult,
|
||||
TServiceDiagnosticDepth,
|
||||
} from "../../../../../../framework/api/diagnostics";
|
||||
|
||||
export interface MongoConnectionOptions {
|
||||
// TODO: More options
|
||||
@@ -42,6 +46,24 @@ export class MongoConnector extends AbstractConnector<MongoConnectionOptions> {
|
||||
return this;
|
||||
}
|
||||
|
||||
private async ping(): Promise<boolean> {
|
||||
const wasConnected = !!this.client;
|
||||
await (await this.getDatabase()).admin().ping();
|
||||
return !wasConnected;
|
||||
}
|
||||
|
||||
async getDiagnostics(depth: TServiceDiagnosticDepth): Promise<TDiagnosticResult> {
|
||||
switch (depth) {
|
||||
case TServiceDiagnosticDepth.critical:
|
||||
return { ok: true, didConnect: await this.ping() };
|
||||
case TServiceDiagnosticDepth.deep_statistics:
|
||||
case TServiceDiagnosticDepth.tracked_statistics:
|
||||
return { ok: true, warn: "unsupported_depth" };
|
||||
|
||||
default:
|
||||
throw new Error(`Unexpected TServiceDiagnosticDepth: ${JSON.stringify(depth)}`);
|
||||
}
|
||||
}
|
||||
getClient(): mongo.MongoClient {
|
||||
return this.client;
|
||||
}
|
||||
|
||||
+28
-1
@@ -8,6 +8,10 @@ import { getEntityDefinition } from "../../../orm/utils";
|
||||
import { UpsertOptions } from "src/core/platform/services/database/services/orm/connectors";
|
||||
import { PostgresDataTransformer, TypeMappings } from "./postgres-data-transform";
|
||||
import { PostgresQueryBuilder, Query } from "./postgres-query-builder";
|
||||
import {
|
||||
TDiagnosticResult,
|
||||
TServiceDiagnosticDepth,
|
||||
} from "../../../../../../framework/api/diagnostics";
|
||||
|
||||
export interface PostgresConnectionOptions {
|
||||
database: string;
|
||||
@@ -25,6 +29,7 @@ export interface PostgresConnectionOptions {
|
||||
export class PostgresConnector extends AbstractConnector<PostgresConnectionOptions> {
|
||||
private logger = getLogger("PostgresConnector");
|
||||
private client: Client = new Client(this.options);
|
||||
private connected = false;
|
||||
private dataTransformer = new PostgresDataTransformer({ secret: this.secret });
|
||||
private queryBuilder = new PostgresQueryBuilder(this.secret);
|
||||
|
||||
@@ -40,10 +45,12 @@ export class PostgresConnector extends AbstractConnector<PostgresConnectionOptio
|
||||
if (this.client) {
|
||||
this.logger.info("Connecting to DB");
|
||||
this.client.on("error", err => {
|
||||
this.logger.warn(err, "PostgreSQL connection error");
|
||||
this.logger.error(err, "PostgreSQL connection error");
|
||||
});
|
||||
this.client.on("end", () => (this.connected = false));
|
||||
|
||||
await this.client.connect();
|
||||
this.connected = true;
|
||||
this.logger.info("Connection pool created");
|
||||
await this.healthcheck();
|
||||
}
|
||||
@@ -56,6 +63,26 @@ export class PostgresConnector extends AbstractConnector<PostgresConnectionOptio
|
||||
return this;
|
||||
}
|
||||
|
||||
private async ping(): Promise<boolean> {
|
||||
const wasConnected = this.connected;
|
||||
if (wasConnected) await this.healthcheck();
|
||||
else await this.connect();
|
||||
return !wasConnected;
|
||||
}
|
||||
|
||||
async getDiagnostics(depth: TServiceDiagnosticDepth): Promise<TDiagnosticResult> {
|
||||
switch (depth) {
|
||||
case TServiceDiagnosticDepth.critical:
|
||||
return { ok: true, didConnect: await this.ping() };
|
||||
case TServiceDiagnosticDepth.stats_basic:
|
||||
case TServiceDiagnosticDepth.stats_track:
|
||||
case TServiceDiagnosticDepth.stats_deep:
|
||||
return { ok: true, warn: "unsupported_depth" };
|
||||
default:
|
||||
throw new Error(`Unexpected TServiceDiagnosticDepth: ${JSON.stringify(depth)}`);
|
||||
}
|
||||
}
|
||||
|
||||
async init(): Promise<this> {
|
||||
if (!this.client) {
|
||||
await this.connect();
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
import { TdriveService, Consumes, Prefix, ServiceName } from "../../framework";
|
||||
import web from "./web";
|
||||
import DiagnosticsServiceAPI from "./service-provider";
|
||||
import DiagnosticsServiceImpl from "./service";
|
||||
import WebServerAPI from "../webserver/provider";
|
||||
import registerBasicProviders from "./providers";
|
||||
|
||||
/**
|
||||
* The diagnostics service exposes endpoint that are of use for operational reasons.
|
||||
*
|
||||
*/
|
||||
@Prefix("/api/diagnostics")
|
||||
@Consumes(["webserver"])
|
||||
@ServiceName("diagnostics")
|
||||
export default class DiagnosticsService extends TdriveService<DiagnosticsServiceAPI> {
|
||||
name = "diagnostics";
|
||||
service: DiagnosticsServiceAPI;
|
||||
|
||||
api(): DiagnosticsServiceAPI {
|
||||
return this.service;
|
||||
}
|
||||
|
||||
public async doInit(): Promise<this> {
|
||||
this.service = new DiagnosticsServiceImpl();
|
||||
const fastify = this.context.getProvider<WebServerAPI>("webserver").getServer();
|
||||
registerBasicProviders();
|
||||
|
||||
fastify.register((instance, _opts, next) => {
|
||||
web(instance, { prefix: this.prefix });
|
||||
next();
|
||||
});
|
||||
|
||||
return this;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
import diagnostics, { TServiceDiagnosticDepth } from "../../../framework/api/diagnostics";
|
||||
import globalResolver from "../../../../../services/global-resolver";
|
||||
|
||||
export default () =>
|
||||
diagnostics.registerProviders({
|
||||
key: "db-ping",
|
||||
tags: ["startup", "live", "ready"],
|
||||
get: () => globalResolver.database.getDiagnostics(TServiceDiagnosticDepth.critical),
|
||||
});
|
||||
@@ -0,0 +1,9 @@
|
||||
import registerDBPingProvider from "./db-ping";
|
||||
import registerPlatformProvider from "./platform-started";
|
||||
import registerProcessProvider from "./process";
|
||||
|
||||
export default () => {
|
||||
registerDBPingProvider();
|
||||
registerPlatformProvider();
|
||||
registerProcessProvider();
|
||||
};
|
||||
+14
@@ -0,0 +1,14 @@
|
||||
import diagnostics from "../../../framework/api/diagnostics";
|
||||
import globalResolver from "../../../../../services/global-resolver";
|
||||
|
||||
export default () =>
|
||||
diagnostics.registerProviders({
|
||||
key: "platform",
|
||||
tags: "*",
|
||||
get: async () => {
|
||||
return {
|
||||
ok: globalResolver.isPlatformStarted(),
|
||||
uptimeMs: Math.floor((process.uptime() + Math.random()) * 1000),
|
||||
};
|
||||
},
|
||||
});
|
||||
@@ -0,0 +1,16 @@
|
||||
import diagnostics from "../../../framework/api/diagnostics";
|
||||
|
||||
export default () =>
|
||||
diagnostics.registerProviders({
|
||||
key: "process",
|
||||
tags: ["live", "ready"],
|
||||
async get() {
|
||||
return {
|
||||
ok: true,
|
||||
gc: !!global.gc,
|
||||
mem: process.memoryUsage(),
|
||||
pid: process.pid,
|
||||
res: process.resourceUsage(),
|
||||
};
|
||||
},
|
||||
});
|
||||
@@ -0,0 +1,5 @@
|
||||
import type { TdriveServiceProvider } from "../../framework";
|
||||
|
||||
type DiagnosticsServiceAPI = TdriveServiceProvider;
|
||||
|
||||
export default DiagnosticsServiceAPI;
|
||||
@@ -0,0 +1,5 @@
|
||||
import DiagnosticsServiceAPI from "./service-provider";
|
||||
|
||||
export default class DiagnosticsServiceImpl implements DiagnosticsServiceAPI {
|
||||
version: "1";
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
import { logger } from "../../framework";
|
||||
import { Readable, Transform } from "node:stream";
|
||||
import { Session } from "node:inspector";
|
||||
|
||||
/** Time the execution of the callback, and return its duration in ms with the result */
|
||||
export async function timeCallback<T>(
|
||||
cb: () => Promise<T>,
|
||||
): Promise<{ durationMs: number; result: T }> {
|
||||
const startMs = new Date().getTime();
|
||||
const result = await cb();
|
||||
return { durationMs: new Date().getTime() - startMs, result };
|
||||
}
|
||||
|
||||
/** Use `node:inspector` to create a snapshot of the heap, and
|
||||
* synchroneously (because of session.post) pipe it to a stream
|
||||
* readable from the first argument of the callback.
|
||||
*/
|
||||
export function getHeapSnapshotSync(cb: (readable: Readable) => void) {
|
||||
logger.info({ gcExposed: !!global.gc }, "Beginning heap snapshot");
|
||||
if (global.gc) global.gc();
|
||||
const session = new Session();
|
||||
session.connect();
|
||||
try {
|
||||
const transform = new Transform();
|
||||
try {
|
||||
let size = 0;
|
||||
session.on("HeapProfiler.addHeapSnapshotChunk", message => {
|
||||
size += message.params.chunk.length;
|
||||
transform.push(message.params.chunk);
|
||||
});
|
||||
cb(transform);
|
||||
session.post("HeapProfiler.takeHeapSnapshot", null);
|
||||
logger.info({ size }, "Heap snapshot sent");
|
||||
} finally {
|
||||
transform.end();
|
||||
}
|
||||
} finally {
|
||||
session.disconnect();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
import type { FastifyInstance, FastifyPluginCallback } from "fastify";
|
||||
import { getConfig as getDiagnosticsGetConfig } from "../../../framework/api/diagnostics";
|
||||
import { getHeapSnapshotSync } from "../utils";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
|
||||
const routes: FastifyPluginCallback = (fastify: FastifyInstance, _opts, next) => {
|
||||
const diagnosticsConfig = getDiagnosticsGetConfig();
|
||||
|
||||
if (diagnosticsConfig?.probeSecret?.length && diagnosticsConfig?.secret?.length) {
|
||||
const getRunningToken = (() => {
|
||||
const newToken = () => [randomUUID(), randomUUID()].join("*");
|
||||
let token = newToken();
|
||||
let updatedS = process.uptime();
|
||||
return () => {
|
||||
const nowS = process.uptime();
|
||||
if (nowS - updatedS > diagnosticsConfig.secretChallengeRefreshS) {
|
||||
token = newToken();
|
||||
updatedS = nowS;
|
||||
}
|
||||
return token;
|
||||
};
|
||||
})();
|
||||
const hashToken = (token: string = getRunningToken()) =>
|
||||
createHash("sha512").update(`+${token}+${diagnosticsConfig.secret}+`).digest("hex");
|
||||
|
||||
/*
|
||||
Example flow:
|
||||
Configuration:
|
||||
{ secret: 'secretValue', probeSecret: 'probeSecretValue' }
|
||||
GET current token using the probeSecret:
|
||||
$ curl "$SERVER/api/diagnostics/heap?secret=probeSecretValue"
|
||||
{"token":"d5888f91-d929-4775-b9dc-de6e2fb4d7cd*b2a57b32-7029-4957-82e9-c7c25794727f"}
|
||||
Hash `+${token}+${secret}+` in SHA-512:
|
||||
Shell example:
|
||||
$ echo -n '+d5888f91-d929-4775-b9dc-de6e2fb4d7cd*b2a57b32-7029-4957-82e9-c7c25794727f+secretValue+' | sha512sum | cut -f1 -d' '
|
||||
320070a508da218baf0f3363e837080fdb902bd0d986bd0f33b806b1230608c0868accdd10bd25261ca91b57c0459edf76218deb26571c72f6b93b077846abe3
|
||||
JS example:
|
||||
require('crypto').createHash('sha512').update(`+${token}+${secret}+`).digest('hex')
|
||||
POST to download snapshot:
|
||||
$ curl "$SERVER/api/diagnostics/heap"
|
||||
|
||||
Shell script example to download heap snapshot (requires curl, node and jq):
|
||||
set -euo pipefail
|
||||
SERVER="http://localhost:4000"
|
||||
DIAG_PROBE_SECRET="probeSecretValue"
|
||||
DIAG_SECRET="secretValue"
|
||||
|
||||
urlencode() { node -e 'console.log(encodeURIComponent(process.argv[1]))' "$@" ; }
|
||||
DIAG_PROBE_SECRET_URLENCODED="$(urlencode "$DIAG_PROBE_SECRET")"
|
||||
TOKEN="$(curl --fail-with-body "$SERVER/api/diagnostics/heap?secret=$DIAG_PROBE_SECRET_URLENCODED" | jq -r .token)"
|
||||
HASHED="$(echo -n "+$TOKEN+$DIAG_SECRET+" | sha512sum | cut -f1 -d' ')"
|
||||
curl --fail-with-body --remote-name --remote-header-name \
|
||||
--header "Content-Type: application/json" \
|
||||
--data '{"hash":"'"$HASHED"'"}' \
|
||||
"$SERVER/api/diagnostics/heap"
|
||||
# The file should be downloaded as `twake-drive-snap-$date.heapsnapshot`
|
||||
ls twake-drive-snap-*.heapsnapshot
|
||||
*/
|
||||
fastify.get("/heap", async (request, reply) => {
|
||||
if ((request.query as { secret: string }).secret !== diagnosticsConfig.probeSecret)
|
||||
return reply.status(403).send();
|
||||
return reply.send({ token: getRunningToken() });
|
||||
});
|
||||
|
||||
fastify.post("/heap", async (request, reply) => {
|
||||
const hashExpected = hashToken();
|
||||
const hashProvided = (request.body as { hash: string })?.hash;
|
||||
if (hashProvided !== hashExpected) return reply.status(403).send();
|
||||
const filenameTimestamp = new Date()
|
||||
.toISOString()
|
||||
.replace(/(\.\d\d\d)?Z$/, "")
|
||||
.replace(/\D/g, "-");
|
||||
reply.header(
|
||||
"Content-Disposition",
|
||||
`attachment; filename="twake-drive-snap-${filenameTimestamp}.heapsnapshot"`,
|
||||
);
|
||||
let replyResult;
|
||||
getHeapSnapshotSync(readable => (replyResult = reply.send(readable)));
|
||||
return replyResult;
|
||||
});
|
||||
}
|
||||
|
||||
next();
|
||||
};
|
||||
|
||||
export default routes;
|
||||
@@ -0,0 +1,11 @@
|
||||
import type { FastifyInstance, FastifyRegisterOptions } from "fastify";
|
||||
import probeRoutes from "./probe-routes";
|
||||
import heapRoutes from "./heap-routes";
|
||||
|
||||
export default (
|
||||
fastify: FastifyInstance,
|
||||
opts: FastifyRegisterOptions<{ prefix: string }>,
|
||||
): void => {
|
||||
fastify.register(probeRoutes, opts);
|
||||
fastify.register(heapRoutes, opts);
|
||||
};
|
||||
@@ -0,0 +1,25 @@
|
||||
import type { FastifyInstance, FastifyPluginCallback } from "fastify";
|
||||
import diagnostics, {
|
||||
getConfig as getDiagnosticsGetConfig,
|
||||
} from "../../../framework/api/diagnostics";
|
||||
|
||||
const routes: FastifyPluginCallback = (fastify: FastifyInstance, _opts, next) => {
|
||||
const diagnosticsConfig = getDiagnosticsGetConfig();
|
||||
if (diagnosticsConfig?.probeSecret?.length) {
|
||||
const tagParamName = "diagnosticTags";
|
||||
fastify.get(`/t/:${tagParamName}`, async (request, reply) => {
|
||||
const tag = request.params[tagParamName];
|
||||
if (
|
||||
tag === "*" ||
|
||||
(request.query as { secret: string }).secret !== diagnosticsConfig.probeSecret
|
||||
)
|
||||
return reply.status(403).send();
|
||||
const results = await diagnostics.get(tag);
|
||||
if (!results.ok) reply.status(503);
|
||||
return reply.send(results);
|
||||
});
|
||||
}
|
||||
next();
|
||||
};
|
||||
|
||||
export default routes;
|
||||
@@ -32,8 +32,8 @@ export class AVServiceImpl implements TdriveServiceProvider, Initializable {
|
||||
},
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error({ error: `${error}` }, "Error while initializing Antivirus Service");
|
||||
} catch (err) {
|
||||
logger.error({ err }, "Error while initializing Antivirus Service");
|
||||
throw AVException.initializationFailed("Failed to initialize Antivirus service");
|
||||
}
|
||||
return this;
|
||||
|
||||
@@ -15,7 +15,7 @@ import TrackerAPI from "../core/platform/services/tracker/provider";
|
||||
import WebServerAPI from "../core/platform/services/webserver/provider";
|
||||
|
||||
import assert from "assert";
|
||||
import { logger } from "../core/platform/framework";
|
||||
import { logger, TdriveServiceState } from "../core/platform/framework";
|
||||
import { ApplicationServiceImpl } from "./applications/services/applications";
|
||||
import { CompanyApplicationServiceImpl } from "./applications/services/company-applications";
|
||||
import { ApplicationHooksService } from "./applications/services/hooks";
|
||||
@@ -78,6 +78,7 @@ class GlobalResolver {
|
||||
public services: TdriveServices;
|
||||
public platformServices: PlatformServices;
|
||||
public database: DatabaseServiceAPI;
|
||||
private platform?: TdrivePlatform;
|
||||
|
||||
public fastify: FastifyInstance<Server, IncomingMessage, ServerResponse>;
|
||||
|
||||
@@ -87,6 +88,7 @@ class GlobalResolver {
|
||||
if (this.alreadyInitialized) {
|
||||
return;
|
||||
}
|
||||
this.platform = platform;
|
||||
this.database = platform.getProvider<DatabaseServiceAPI>("database");
|
||||
|
||||
this.platformServices = {
|
||||
@@ -152,6 +154,11 @@ class GlobalResolver {
|
||||
logger.info("Global resolver finished initializing services");
|
||||
this.alreadyInitialized = true;
|
||||
}
|
||||
|
||||
/** `true` if all components are in the started state. This means we should respond to http at least (for kubernetes Startup probe). */
|
||||
isPlatformStarted(): boolean {
|
||||
return this.alreadyInitialized && this.platform?.state.getValue() == TdriveServiceState.Started;
|
||||
}
|
||||
}
|
||||
|
||||
export default new GlobalResolver();
|
||||
|
||||
Reference in New Issue
Block a user