grpc
This commit is contained in:
Alex Holliday committed
1 parent
27c44a0fbd
commit
c7689cc64a
2 files changed
+187
-354
No files matched your search
@@ -0,0 +1,187 @@
|
||||
import { IStatusProvider } from "@/service/infrastructure/network/IStatusProvider.js";
|
||||
import { GrpcStatusPayload, MonitorStatusResponse } from "@/types/network.js";
|
||||
import { Monitor, MonitorType } from "@/types/monitor.js";
|
||||
import { AppError } from "@/utils/AppError.js";
|
||||
import { NETWORK_ERROR, timeRequest } from "@/service/infrastructure/network/utils.js";
|
||||
import { ILogger } from "@/utils/logger.js";
|
||||
import * as grpc from "@grpc/grpc-js";
|
||||
import * as protoLoader from "@grpc/proto-loader";
|
||||
|
||||
import path from "path";
|
||||
import { fileURLToPath } from "url";
|
||||
import { Network } from "inspector/promises";
|
||||
|
||||
type GrpcType = typeof grpc;
|
||||
type ProtoLoaderType = typeof protoLoader;
|
||||
|
||||
const SERVICE_NAME = "GrpcProvider";
|
||||
|
||||
export interface GrpcError extends Error {
|
||||
code: number;
|
||||
details?: string;
|
||||
}
|
||||
|
||||
export class GrpcProvider implements IStatusProvider<GrpcStatusPayload> {
|
||||
readonly type = "grpc";
|
||||
constructor(
|
||||
private logger: ILogger,
|
||||
private grpc: GrpcType,
|
||||
private protoLoader: ProtoLoaderType
|
||||
) {}
|
||||
|
||||
supports(type: MonitorType): boolean {
|
||||
return type === "grpc";
|
||||
}
|
||||
|
||||
private isGrpcError(error: unknown): error is GrpcError {
|
||||
return error instanceof Error && "code" in error && typeof (error as Record<string, unknown>).code === "number";
|
||||
}
|
||||
|
||||
private getGrpcStatusName(code: number): string {
|
||||
const statusNames: Record<number, string> = {
|
||||
0: "OK",
|
||||
1: "CANCELLED",
|
||||
2: "UNKNOWN",
|
||||
3: "INVALID_ARGUMENT",
|
||||
4: "DEADLINE_EXCEEDED",
|
||||
5: "NOT_FOUND",
|
||||
6: "ALREADY_EXISTS",
|
||||
7: "PERMISSION_DENIED",
|
||||
8: "RESOURCE_EXHAUSTED",
|
||||
9: "FAILED_PRECONDITION",
|
||||
10: "ABORTED",
|
||||
11: "OUT_OF_RANGE",
|
||||
12: "UNIMPLEMENTED",
|
||||
13: "INTERNAL",
|
||||
14: "UNAVAILABLE",
|
||||
15: "DATA_LOSS",
|
||||
16: "UNAUTHENTICATED",
|
||||
};
|
||||
return statusNames[code] || "UNKNOWN";
|
||||
}
|
||||
|
||||
async handle(monitor: Monitor): Promise<MonitorStatusResponse<GrpcStatusPayload>> {
|
||||
const { id, teamId, type, url, port, ignoreTlsErrors } = monitor;
|
||||
const grpcServiceName = monitor.grpcServiceName || "";
|
||||
const host = url?.replace(/^https?:\/\//, "").split(/[/?#:]/)[0];
|
||||
|
||||
try {
|
||||
if (!host || !port) {
|
||||
throw new Error("Valid host and port are required");
|
||||
}
|
||||
const target = `${host}:${port}`;
|
||||
const currentFilePath = fileURLToPath(import.meta.url);
|
||||
const protoPath = path.join(path.dirname(currentFilePath), "protos", "health.proto");
|
||||
const packageDefinition = this.protoLoader.loadSync(protoPath, {
|
||||
keepCase: true,
|
||||
longs: String,
|
||||
enums: String,
|
||||
defaults: true,
|
||||
oneofs: true,
|
||||
});
|
||||
const grpcObject = this.grpc.loadPackageDefinition(packageDefinition) as any;
|
||||
const HealthClient = grpcObject.grpc.health.v1.Health;
|
||||
let credentials;
|
||||
if (ignoreTlsErrors) {
|
||||
credentials = this.grpc.credentials.createSsl(null, null, null, {
|
||||
checkServerIdentity: () => undefined,
|
||||
});
|
||||
} else {
|
||||
credentials = this.grpc.credentials.createInsecure();
|
||||
}
|
||||
const client = new HealthClient(target, credentials);
|
||||
const deadline = new Date(Date.now() + 10000); // 10s timeout
|
||||
const { response, responseTime, error } = await timeRequest<GrpcStatusPayload>(() => {
|
||||
return new Promise<GrpcStatusPayload>((resolve, reject) => {
|
||||
client.Check({ service: grpcServiceName }, { deadline }, (err: unknown, response: unknown) => {
|
||||
client.close();
|
||||
|
||||
if (err) {
|
||||
const grpcErr = err as { code?: number; details?: string; message?: string };
|
||||
const payload: GrpcStatusPayload = {
|
||||
grpcStatusCode: grpcErr.code ?? -1,
|
||||
grpcStatusName: this.getGrpcStatusName(grpcErr.code ?? -1),
|
||||
serviceName: grpcServiceName,
|
||||
servingStatus: "UNKNOWN",
|
||||
};
|
||||
const grpcError = new AppError({
|
||||
message: grpcErr.details || grpcErr.message || "gRPC error",
|
||||
service: SERVICE_NAME,
|
||||
method: "handle",
|
||||
}) as AppError & { grpcPayload?: GrpcStatusPayload; grpcCode?: number };
|
||||
grpcError.grpcPayload = payload;
|
||||
grpcError.grpcCode = grpcErr.code;
|
||||
reject(grpcError);
|
||||
return;
|
||||
}
|
||||
|
||||
const resp = response as { status?: string } | undefined;
|
||||
const servingStatus = resp?.status ?? "UNKNOWN";
|
||||
resolve({
|
||||
grpcStatusCode: 0,
|
||||
grpcStatusName: "OK",
|
||||
serviceName: grpcServiceName,
|
||||
servingStatus,
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
if (error) {
|
||||
let grpcCode = -1;
|
||||
let message = error instanceof Error ? error.message : "gRPC health check failed";
|
||||
|
||||
if (this.isGrpcError(error)) {
|
||||
grpcCode = error.code;
|
||||
message = error.details || error.message;
|
||||
}
|
||||
|
||||
return {
|
||||
monitorId: id,
|
||||
teamId: teamId,
|
||||
type: type,
|
||||
status: false,
|
||||
code: 500,
|
||||
message: message,
|
||||
responseTime,
|
||||
payload: {
|
||||
grpcStatusCode: grpcCode,
|
||||
grpcStatusName: this.getGrpcStatusName(grpcCode),
|
||||
serviceName: grpcServiceName,
|
||||
servingStatus: "UNKNOWN",
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
if (!response) {
|
||||
throw new Error("No response received from gRPC health check");
|
||||
}
|
||||
|
||||
const isServing = response.servingStatus === "SERVING";
|
||||
|
||||
return {
|
||||
monitorId: id,
|
||||
teamId: teamId,
|
||||
type: type,
|
||||
status: isServing,
|
||||
code: isServing ? 200 : NETWORK_ERROR,
|
||||
message: isServing ? "gRPC health check successful" : "gRPC health check failed",
|
||||
responseTime,
|
||||
payload: {
|
||||
grpcStatusCode: isServing ? 200 : NETWORK_ERROR,
|
||||
grpcStatusName: response.grpcStatusName,
|
||||
serviceName: response.serviceName,
|
||||
servingStatus: response.servingStatus || "UNKNOWN",
|
||||
},
|
||||
};
|
||||
} catch (err: unknown) {
|
||||
const originalMessage = err instanceof Error ? err.message : String(err);
|
||||
throw new AppError({
|
||||
message: originalMessage || "Error performing gRPC health check",
|
||||
service: SERVICE_NAME,
|
||||
method: "handle",
|
||||
details: { url: monitor.url, port: monitor.port, grpcServiceName: monitor.grpcServiceName },
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -277,360 +277,6 @@ class NetworkService implements INetworkService {
|
||||
}
|
||||
}
|
||||
|
||||
private async requestHttp<T = unknown>(monitor: Monitor): Promise<MonitorStatusResponse<T>> {
|
||||
const { url, secret, ignoreTlsErrors, useAdvancedMatching, jsonPath, matchMethod, expectedValue } = monitor;
|
||||
const httpResponse = this.buildStatusResponse<T>({
|
||||
monitor,
|
||||
overrides: {
|
||||
status: false,
|
||||
code: this.NETWORK_ERROR,
|
||||
message: "Request not executed",
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
if (!url) {
|
||||
throw new Error("Monitor URL is required");
|
||||
}
|
||||
const config: Record<string, unknown> = {
|
||||
headers: secret ? { Authorization: `Bearer ${secret}` } : undefined,
|
||||
};
|
||||
|
||||
if (ignoreTlsErrors) {
|
||||
config.agent = {
|
||||
https: new this.https.Agent({
|
||||
rejectUnauthorized: false,
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
const response = await this.got(url, config);
|
||||
|
||||
let payload: T | string | undefined;
|
||||
const contentType = response.headers["content-type"];
|
||||
|
||||
if (contentType && contentType.includes("application/json")) {
|
||||
try {
|
||||
payload = JSON.parse(response.body);
|
||||
} catch {
|
||||
payload = response.body;
|
||||
}
|
||||
} else {
|
||||
payload = response.body;
|
||||
}
|
||||
|
||||
Object.assign(httpResponse, {
|
||||
code: response.statusCode,
|
||||
status: response.ok,
|
||||
message: response.statusMessage ?? "",
|
||||
responseTime: response.timings.phases.total || 0,
|
||||
payload,
|
||||
timings: response.timings,
|
||||
});
|
||||
|
||||
if (!useAdvancedMatching) {
|
||||
return httpResponse;
|
||||
}
|
||||
|
||||
if (expectedValue && !jsonPath) {
|
||||
let ok = false;
|
||||
if (matchMethod === "equal") ok = payload === expectedValue;
|
||||
if (matchMethod === "include" && typeof payload === "string") ok = payload.includes(expectedValue);
|
||||
if (matchMethod === "regex" && typeof payload === "string") ok = new RegExp(expectedValue).test(payload);
|
||||
|
||||
if (ok === true) {
|
||||
return httpResponse;
|
||||
} else {
|
||||
httpResponse.code = 500;
|
||||
httpResponse.status = false;
|
||||
httpResponse.message = "Expected value did not match";
|
||||
return httpResponse;
|
||||
}
|
||||
}
|
||||
|
||||
if (jsonPath) {
|
||||
const contentType = response.headers["content-type"];
|
||||
const isJson = contentType?.includes("application/json");
|
||||
if (!isJson) {
|
||||
httpResponse.status = false;
|
||||
httpResponse.message = "Response is not JSON";
|
||||
return httpResponse;
|
||||
}
|
||||
try {
|
||||
const extracted = this.jmespath.search(payload, jsonPath);
|
||||
if (expectedValue) {
|
||||
let ok = false;
|
||||
if (matchMethod === "equal") ok = extracted === expectedValue;
|
||||
if (matchMethod === "include" && typeof extracted === "string") ok = extracted.includes(expectedValue);
|
||||
if (matchMethod === "regex" && typeof extracted === "string") ok = new RegExp(expectedValue).test(extracted);
|
||||
|
||||
if (ok) {
|
||||
httpResponse.extracted = extracted;
|
||||
return httpResponse;
|
||||
} else {
|
||||
httpResponse.status = false;
|
||||
httpResponse.code = 500;
|
||||
httpResponse.message = "Expected value did not match";
|
||||
httpResponse.extracted = extracted;
|
||||
return httpResponse;
|
||||
}
|
||||
} else {
|
||||
const isFalsey = extracted === false || extracted === "false" || extracted === undefined || extracted === null;
|
||||
if (!isFalsey) {
|
||||
httpResponse.extracted = extracted;
|
||||
return httpResponse;
|
||||
} else {
|
||||
httpResponse.status = false;
|
||||
httpResponse.code = 500;
|
||||
httpResponse.message = "Expected value did not match";
|
||||
httpResponse.extracted = extracted;
|
||||
return httpResponse;
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
httpResponse.status = false;
|
||||
httpResponse.message = "Error evaluating JSON path";
|
||||
return httpResponse;
|
||||
}
|
||||
}
|
||||
return httpResponse;
|
||||
} catch (err: unknown) {
|
||||
if (err instanceof HTTPError || err instanceof RequestError) {
|
||||
httpResponse.code = err.response?.statusCode || this.NETWORK_ERROR;
|
||||
httpResponse.status = false;
|
||||
httpResponse.message = String(err.response?.statusCode) || err.message;
|
||||
httpResponse.responseTime = err.timings?.phases?.total || 0;
|
||||
httpResponse.payload = null;
|
||||
httpResponse.timings = err.timings || undefined;
|
||||
return httpResponse;
|
||||
}
|
||||
const originalMessage = err instanceof Error ? err.message : String(err);
|
||||
throw new AppError({
|
||||
message: originalMessage || "Error performing HTTP request",
|
||||
service: this.SERVICE_NAME,
|
||||
method: "requestHttp",
|
||||
details: { url: monitor.url },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async requestPageSpeed(monitor: Monitor): Promise<MonitorStatusResponse<PageSpeedStatusPayload>> {
|
||||
try {
|
||||
const url = monitor.url;
|
||||
if (!url) {
|
||||
throw new Error("Monitor URL is required");
|
||||
}
|
||||
let pageSpeedUrl = `https://pagespeedonline.googleapis.com/pagespeedonline/v5/runPagespeed?url=${url}&category=seo&category=accessibility&category=best-practices&category=performance`;
|
||||
const dbSettings = await this.settingsService.getDBSettings();
|
||||
if (dbSettings?.pagespeedApiKey) {
|
||||
pageSpeedUrl += `&key=${dbSettings.pagespeedApiKey}`;
|
||||
} else {
|
||||
this.logger.warn({
|
||||
message: "PageSpeed API key not found, job not executed",
|
||||
service: this.SERVICE_NAME,
|
||||
method: "requestPagespeed",
|
||||
details: { url },
|
||||
});
|
||||
}
|
||||
return await this.requestHttp<PageSpeedStatusPayload>({
|
||||
...monitor,
|
||||
url: pageSpeedUrl,
|
||||
});
|
||||
} catch (err: unknown) {
|
||||
const originalMessage = err instanceof Error ? err.message : String(err);
|
||||
throw new AppError({
|
||||
message: originalMessage || "Error performing PageSpeed request",
|
||||
service: this.SERVICE_NAME,
|
||||
method: "requestPageSpeed",
|
||||
details: { url: monitor.url },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async requestHardware(monitor: Monitor): Promise<MonitorStatusResponse<HardwareStatusPayload>> {
|
||||
try {
|
||||
return await this.requestHttp<HardwareStatusPayload>(monitor);
|
||||
} catch (err: unknown) {
|
||||
const originalMessage = err instanceof Error ? err.message : String(err);
|
||||
throw new AppError({
|
||||
message: originalMessage || "Error performing hardware request",
|
||||
service: this.SERVICE_NAME,
|
||||
method: "requestHardware",
|
||||
details: { url: monitor.url },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async requestDocker(monitor: Monitor): Promise<MonitorStatusResponse<DockerStatusPayload>> {
|
||||
try {
|
||||
if (!monitor.url) {
|
||||
throw new Error("Monitor URL is required");
|
||||
}
|
||||
|
||||
const docker = new this.Docker({
|
||||
socketPath: "/var/run/docker.sock",
|
||||
});
|
||||
|
||||
const dockerResponse = this.buildStatusResponse<DockerStatusPayload>({
|
||||
monitor,
|
||||
overrides: {
|
||||
status: false,
|
||||
code: this.NETWORK_ERROR,
|
||||
message: "No response",
|
||||
},
|
||||
});
|
||||
|
||||
const containers = await docker.listContainers({ all: true });
|
||||
|
||||
// Normalize input: strip leading slashes and convert to lowercase for comparison
|
||||
const normalizedInput = monitor.url.replace(/^\/+/, "").toLowerCase();
|
||||
|
||||
// Priority-based matching to avoid ambiguity:
|
||||
// 1. Exact full ID match (64-char)
|
||||
const exactIdMatch = containers.find((c) => c.Id.toLowerCase() === normalizedInput);
|
||||
|
||||
// 2. Exact container name match (case-insensitive)
|
||||
const exactNameMatch = containers.find((c) =>
|
||||
c.Names.some((name: string) => {
|
||||
const cleanName = name.replace(/^\/+/, "").toLowerCase();
|
||||
return cleanName === normalizedInput;
|
||||
})
|
||||
);
|
||||
|
||||
// 3. Partial ID match (fallback for backwards compatibility)
|
||||
const partialIdMatch = containers.find((c) => c.Id.toLowerCase().startsWith(normalizedInput));
|
||||
|
||||
// Select container based on priority
|
||||
const targetContainer = exactIdMatch || exactNameMatch || partialIdMatch;
|
||||
|
||||
// Return negative response if no container
|
||||
if (!targetContainer) {
|
||||
this.logger.warn({
|
||||
message: `No container found for "${monitor.url}".`,
|
||||
service: this.SERVICE_NAME,
|
||||
method: "requestDocker",
|
||||
details: { url: monitor.url },
|
||||
});
|
||||
|
||||
dockerResponse.code = 404;
|
||||
dockerResponse.status = false;
|
||||
dockerResponse.message = "Docker container not found";
|
||||
return dockerResponse;
|
||||
}
|
||||
|
||||
// Return negative response if ambiguous matches exist
|
||||
const matchTypes = [];
|
||||
if (exactIdMatch) matchTypes.push("exact ID");
|
||||
if (exactNameMatch) matchTypes.push("exact name");
|
||||
if (partialIdMatch && !exactIdMatch) matchTypes.push("partial ID");
|
||||
|
||||
if (matchTypes.length > 1) {
|
||||
this.logger.warn({
|
||||
message: `Ambiguous container match for "${monitor.url}". Matched by: ${matchTypes.join(", ")}. Using ${exactIdMatch ? "exact ID" : exactNameMatch ? "exact name" : "partial ID"} match.`,
|
||||
service: this.SERVICE_NAME,
|
||||
method: "requestDocker",
|
||||
details: { url: monitor.url },
|
||||
});
|
||||
dockerResponse.status = false;
|
||||
dockerResponse.message = `Ambiguous container match for "${monitor.url}". Matched by: ${matchTypes.join(", ")}. Using ${exactIdMatch ? "exact ID" : exactNameMatch ? "exact name" : "partial ID"} match.`;
|
||||
return dockerResponse;
|
||||
}
|
||||
|
||||
const container = docker.getContainer(targetContainer.Id);
|
||||
const { response, responseTime, error }: { response?: { State: { Status: string } } | null; responseTime: number; error: unknown } =
|
||||
await this.timeRequest(() => container.inspect());
|
||||
|
||||
dockerResponse.responseTime = responseTime;
|
||||
dockerResponse.status = response?.State?.Status === "running" ? true : false;
|
||||
dockerResponse.code = 200;
|
||||
dockerResponse.message = "Docker container status fetched successfully";
|
||||
|
||||
if (error) {
|
||||
dockerResponse.status = false;
|
||||
dockerResponse.code =
|
||||
(error && typeof error === "object" && "statusCode" in error ? (error as { statusCode?: number }).statusCode : undefined) ||
|
||||
this.NETWORK_ERROR;
|
||||
dockerResponse.message =
|
||||
(error && typeof error === "object" && "reason" in error ? (error as { reason?: string }).reason : undefined) ||
|
||||
"Failed to fetch Docker container information";
|
||||
return dockerResponse;
|
||||
}
|
||||
|
||||
return dockerResponse;
|
||||
} catch (err: unknown) {
|
||||
const originalMessage = err instanceof Error ? err.message : String(err);
|
||||
throw new AppError({
|
||||
message: originalMessage || "Error performing Docker request",
|
||||
service: this.SERVICE_NAME,
|
||||
method: "requestDocker",
|
||||
details: { url: monitor.url },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async requestPort(monitor: Monitor): Promise<MonitorStatusResponse<PortStatusPayload>> {
|
||||
try {
|
||||
const { url, port } = monitor;
|
||||
if (!port) {
|
||||
throw new Error("Port is required for port monitor");
|
||||
}
|
||||
|
||||
const { response, responseTime, error } = await this.timeRequest(async () => {
|
||||
return new Promise((resolve, reject) => {
|
||||
const socket = this.net.createConnection(
|
||||
{
|
||||
host: url,
|
||||
port,
|
||||
},
|
||||
() => {
|
||||
socket.end();
|
||||
socket.destroy();
|
||||
resolve({ success: true });
|
||||
}
|
||||
);
|
||||
|
||||
socket.setTimeout(5000);
|
||||
socket.on("timeout", () => {
|
||||
socket.destroy();
|
||||
reject(new Error("Connection timeout"));
|
||||
});
|
||||
|
||||
socket.on("error", (err: unknown) => {
|
||||
socket.destroy();
|
||||
reject(err);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
const portResponse = this.buildStatusResponse<PortStatusPayload>({
|
||||
monitor,
|
||||
overrides: {
|
||||
code: 200,
|
||||
status: (response as { success?: boolean })?.success ?? false,
|
||||
message: "Port check successful",
|
||||
responseTime,
|
||||
},
|
||||
});
|
||||
|
||||
if (error) {
|
||||
portResponse.code = this.NETWORK_ERROR;
|
||||
portResponse.status = false;
|
||||
portResponse.message = "Port check failed";
|
||||
return portResponse;
|
||||
}
|
||||
|
||||
return portResponse;
|
||||
} catch (err: unknown) {
|
||||
const originalMessage = err instanceof Error ? err.message : String(err);
|
||||
throw new AppError({
|
||||
message: originalMessage || "Error performing port check",
|
||||
service: this.SERVICE_NAME,
|
||||
method: "requestPort",
|
||||
details: { url: monitor.url, port: monitor.port },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async requestGrpc(monitor: Monitor): Promise<MonitorStatusResponse<GrpcStatusPayload>> {
|
||||
try {
|
||||
const { url, port, ignoreTlsErrors } = monitor;
|
||||
|
||||
Reference in new issue
Block a user