mirror of
https://github.com/Dokploy/dokploy.git
synced 2026-10-19 23:21:36 +00:00
Merge pull request #5624 from Dokploy/feat/logs-and-metrics
feat(logs-and-metrics): ship container logs and metrics to external providers via Vector
This commit is contained in:
Yash Kumar
@@ -8,6 +8,7 @@ import {
|
||||
user,
|
||||
verification,
|
||||
} from "@dokploy/server/db/schema";
|
||||
import { removeVectorAgents } from "@dokploy/server/services/organization";
|
||||
import { TRPCError } from "@trpc/server";
|
||||
import { eq, inArray } from "drizzle-orm";
|
||||
import { nanoid } from "nanoid";
|
||||
@@ -128,6 +129,15 @@ export const consumeAccountDeletionCode = async (
|
||||
};
|
||||
|
||||
export const deleteUserAccountData = async (userId: string) => {
|
||||
// The owned organizations go away by FK cascade, which skips deleteOrganization's agent cleanup.
|
||||
const ownedOrganizations = await db.query.organization.findMany({
|
||||
where: eq(organization.ownerId, userId),
|
||||
columns: { id: true },
|
||||
});
|
||||
for (const org of ownedOrganizations) {
|
||||
await removeVectorAgents(org.id);
|
||||
}
|
||||
|
||||
return db.transaction(async (tx) => {
|
||||
const target = await tx.query.user.findFirst({
|
||||
where: eq(user.id, userId),
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
import { type LookupAddress, lookup } from "node:dns";
|
||||
import { BlockList, isIP, type LookupFunction } from "node:net";
|
||||
import { IS_CLOUD } from "@dokploy/server/constants";
|
||||
import { Agent } from "undici";
|
||||
|
||||
export const PROVIDER_REQUEST_TIMEOUT_MS = 15_000;
|
||||
|
||||
const metadataAddresses = new BlockList();
|
||||
metadataAddresses.addSubnet("169.254.0.0", 16, "ipv4");
|
||||
metadataAddresses.addSubnet("fe80::", 10, "ipv6");
|
||||
metadataAddresses.addAddress("fd00:ec2::254", "ipv6");
|
||||
|
||||
const privateAddresses = new BlockList();
|
||||
privateAddresses.addSubnet("0.0.0.0", 8, "ipv4");
|
||||
privateAddresses.addSubnet("127.0.0.0", 8, "ipv4");
|
||||
privateAddresses.addSubnet("10.0.0.0", 8, "ipv4");
|
||||
privateAddresses.addSubnet("100.64.0.0", 10, "ipv4");
|
||||
privateAddresses.addSubnet("172.16.0.0", 12, "ipv4");
|
||||
privateAddresses.addSubnet("192.168.0.0", 16, "ipv4");
|
||||
privateAddresses.addAddress("::", "ipv6");
|
||||
privateAddresses.addAddress("::1", "ipv6");
|
||||
privateAddresses.addSubnet("fc00::", 7, "ipv6");
|
||||
|
||||
const isBlockedAddress = (address: string): boolean => {
|
||||
const ip = address.replace(/^\[|\]$/g, "");
|
||||
const family = isIP(ip);
|
||||
if (family === 0) {
|
||||
return IS_CLOUD && ip.toLowerCase() === "localhost";
|
||||
}
|
||||
const type = family === 4 ? "ipv4" : "ipv6";
|
||||
return (
|
||||
metadataAddresses.check(ip, type) ||
|
||||
(IS_CLOUD && privateAddresses.check(ip, type))
|
||||
);
|
||||
};
|
||||
|
||||
const METADATA_HOSTNAMES = new Set([
|
||||
"metadata.google.internal",
|
||||
"metadata.goog",
|
||||
]);
|
||||
|
||||
class BlockedEndpointError extends Error {
|
||||
constructor() {
|
||||
super(
|
||||
"This endpoint resolves to a blocked address (cloud metadata or private network) and can't be used here.",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const assertNotBlockedEndpoint = (rawUrl: string): void => {
|
||||
let url: URL;
|
||||
try {
|
||||
url = new URL(rawUrl);
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
if (
|
||||
METADATA_HOSTNAMES.has(url.hostname.toLowerCase()) ||
|
||||
isBlockedAddress(url.hostname)
|
||||
) {
|
||||
throw new BlockedEndpointError();
|
||||
}
|
||||
};
|
||||
|
||||
// Checks the addresses the socket actually connects to; a separate pre-flight lookup could be answered differently (DNS rebinding).
|
||||
export const guardedLookup: LookupFunction = (hostname, options, callback) => {
|
||||
lookup(hostname, { ...options, all: true }, (error, addresses) => {
|
||||
if (error) {
|
||||
callback(error, "");
|
||||
return;
|
||||
}
|
||||
const list = addresses as LookupAddress[];
|
||||
const first = list[0];
|
||||
if (!first) {
|
||||
callback(new Error(`No addresses found for ${hostname}`), "");
|
||||
return;
|
||||
}
|
||||
if (list.some((entry) => isBlockedAddress(entry.address))) {
|
||||
callback(new BlockedEndpointError(), "");
|
||||
return;
|
||||
}
|
||||
if (options.all) {
|
||||
callback(null, list);
|
||||
} else {
|
||||
callback(null, first.address, first.family);
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
const guardedAgent = new Agent({ connect: { lookup: guardedLookup } });
|
||||
|
||||
export const providerFetch = async (url: string, init: RequestInit = {}) => {
|
||||
assertNotBlockedEndpoint(url);
|
||||
try {
|
||||
return await fetch(url, {
|
||||
...init,
|
||||
redirect: "error",
|
||||
signal: AbortSignal.timeout(PROVIDER_REQUEST_TIMEOUT_MS),
|
||||
// @ts-expect-error Node's fetch accepts an undici dispatcher, but lib.dom's RequestInit doesn't declare it.
|
||||
dispatcher: guardedAgent,
|
||||
});
|
||||
} catch (error) {
|
||||
if (error instanceof Error && error.cause instanceof BlockedEndpointError) {
|
||||
throw error.cause;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
@@ -0,0 +1,227 @@
|
||||
import {
|
||||
CloudWatchClient,
|
||||
PutMetricDataCommand,
|
||||
} from "@aws-sdk/client-cloudwatch";
|
||||
import {
|
||||
CloudWatchLogsClient,
|
||||
DescribeLogGroupsCommand,
|
||||
} from "@aws-sdk/client-cloudwatch-logs";
|
||||
import { PROVIDER_REQUEST_TIMEOUT_MS } from "../provider-fetch";
|
||||
import type {
|
||||
TelemetryProviderAdapter,
|
||||
TelemetryProviderRuntimeConfig,
|
||||
TelemetrySignal,
|
||||
VectorSinkConfig,
|
||||
VectorTransformConfig,
|
||||
} from "../types";
|
||||
import { DEFAULT_DISK_BUFFER, stringValue, vrlString } from "../types";
|
||||
|
||||
const DEFAULT_STREAM_TEMPLATE = "{{ container_name }}";
|
||||
|
||||
export const CLOUDWATCH_METRICS_ALLOWLIST = [
|
||||
"cpu_seconds_total",
|
||||
"memory_used_bytes",
|
||||
"memory_total_bytes",
|
||||
"filesystem_used_bytes",
|
||||
"filesystem_total_bytes",
|
||||
"network_receive_bytes_total",
|
||||
"network_transmit_bytes_total",
|
||||
"load1",
|
||||
"load5",
|
||||
"load15",
|
||||
"container_cpu_usage_seconds_total",
|
||||
"container_memory_usage_bytes",
|
||||
"container_network_receive_bytes_total",
|
||||
"container_network_transmit_bytes_total",
|
||||
];
|
||||
|
||||
const resolveMetricsNamespace = (config: TelemetryProviderRuntimeConfig) => {
|
||||
const namespace = stringValue(config.extraConfig?.metricsNamespace);
|
||||
if (namespace.startsWith("AWS/")) {
|
||||
throw new Error("The metrics namespace cannot start with AWS/");
|
||||
}
|
||||
return namespace;
|
||||
};
|
||||
|
||||
const credentials = (config: TelemetryProviderRuntimeConfig) => ({
|
||||
accessKeyId: config.apiKey ?? "",
|
||||
secretAccessKey: config.apiSecret ?? "",
|
||||
});
|
||||
|
||||
export const awsCloudwatchAdapter: TelemetryProviderAdapter = {
|
||||
type: "aws_cloudwatch",
|
||||
signals: ["logs", "metrics"],
|
||||
label: "AWS CloudWatch",
|
||||
docsUrl:
|
||||
"https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/WhatIsCloudWatchLogs.html",
|
||||
credentialFields: [
|
||||
{
|
||||
key: "apiKey",
|
||||
label: "Access Key ID",
|
||||
type: "text",
|
||||
required: true,
|
||||
},
|
||||
{
|
||||
key: "apiSecret",
|
||||
label: "Secret Access Key",
|
||||
type: "password",
|
||||
required: true,
|
||||
},
|
||||
{
|
||||
key: "region",
|
||||
label: "Region",
|
||||
type: "text",
|
||||
required: true,
|
||||
placeholder: "us-east-1",
|
||||
},
|
||||
{
|
||||
key: "logGroup",
|
||||
label: "Log Group Name",
|
||||
type: "text",
|
||||
required: true,
|
||||
signal: "logs",
|
||||
placeholder: "/dokploy/logs",
|
||||
helpText: "Created automatically on first write if it doesn't exist.",
|
||||
fullWidth: true,
|
||||
},
|
||||
{
|
||||
key: "logStream",
|
||||
label: "Log Stream Name Template",
|
||||
type: "text",
|
||||
required: false,
|
||||
signal: "logs",
|
||||
placeholder: DEFAULT_STREAM_TEMPLATE,
|
||||
helpText:
|
||||
"Vector event template. Defaults to one stream per container name if left blank.",
|
||||
fullWidth: true,
|
||||
},
|
||||
{
|
||||
key: "metricsNamespace",
|
||||
label: "Metrics namespace",
|
||||
type: "text",
|
||||
required: true,
|
||||
signal: "metrics",
|
||||
placeholder: "Dokploy",
|
||||
helpText:
|
||||
"E.g. Dokploy. Cannot start with AWS/. The IAM user needs cloudwatch:PutMetricData; testing the connection writes a real data point.",
|
||||
fullWidth: true,
|
||||
},
|
||||
],
|
||||
validateConfig(config: TelemetryProviderRuntimeConfig): void {
|
||||
if (config.signals.includes("metrics")) {
|
||||
resolveMetricsNamespace(config);
|
||||
}
|
||||
},
|
||||
toVectorTransform(
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
_transformId: string,
|
||||
scopeTransformId: string,
|
||||
signal: TelemetrySignal,
|
||||
): VectorTransformConfig | null {
|
||||
if (signal === "logs") return null;
|
||||
return {
|
||||
type: "remap",
|
||||
inputs: [scopeTransformId],
|
||||
drop_on_abort: true,
|
||||
source: [
|
||||
`if !includes(${JSON.stringify(CLOUDWATCH_METRICS_ALLOWLIST)}, .name) { abort }`,
|
||||
`.namespace = ${vrlString(resolveMetricsNamespace(config))}`,
|
||||
].join("\n"),
|
||||
};
|
||||
},
|
||||
toVectorSink(
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
_sinkId: string,
|
||||
inputId: string,
|
||||
signal: TelemetrySignal,
|
||||
): VectorSinkConfig {
|
||||
const region = stringValue(config.extraConfig?.region);
|
||||
if (signal === "metrics") {
|
||||
return {
|
||||
type: "aws_cloudwatch_metrics",
|
||||
inputs: [inputId],
|
||||
default_namespace: resolveMetricsNamespace(config),
|
||||
region,
|
||||
auth: {
|
||||
access_key_id: config.apiKey ?? "",
|
||||
secret_access_key: config.apiSecret ?? "",
|
||||
},
|
||||
buffer: DEFAULT_DISK_BUFFER,
|
||||
};
|
||||
}
|
||||
const logGroup = config.extraConfig?.logGroup;
|
||||
const logStream = config.extraConfig?.logStream;
|
||||
return {
|
||||
type: "aws_cloudwatch_logs",
|
||||
inputs: [inputId],
|
||||
group_name: typeof logGroup === "string" ? logGroup : "",
|
||||
stream_name:
|
||||
typeof logStream === "string" && logStream.length > 0
|
||||
? logStream
|
||||
: DEFAULT_STREAM_TEMPLATE,
|
||||
region,
|
||||
encoding: { codec: "json" },
|
||||
auth: {
|
||||
access_key_id: config.apiKey ?? "",
|
||||
secret_access_key: config.apiSecret ?? "",
|
||||
},
|
||||
buffer: DEFAULT_DISK_BUFFER,
|
||||
};
|
||||
},
|
||||
async testConnection(config: TelemetryProviderRuntimeConfig): Promise<void> {
|
||||
const region = stringValue(config.extraConfig?.region);
|
||||
if (!config.apiKey || !config.apiSecret || !region) {
|
||||
throw new Error("AWS access key, secret key and region are required");
|
||||
}
|
||||
const abortSignal = AbortSignal.timeout(PROVIDER_REQUEST_TIMEOUT_MS);
|
||||
if (config.signals.includes("logs")) {
|
||||
const logGroup = stringValue(config.extraConfig?.logGroup);
|
||||
if (!logGroup) {
|
||||
throw new Error("AWS log group is required");
|
||||
}
|
||||
const client = new CloudWatchLogsClient({
|
||||
region,
|
||||
credentials: credentials(config),
|
||||
});
|
||||
await client.send(
|
||||
new DescribeLogGroupsCommand({
|
||||
logGroupNamePrefix: logGroup,
|
||||
limit: 1,
|
||||
}),
|
||||
{ abortSignal },
|
||||
);
|
||||
}
|
||||
if (config.signals.includes("metrics")) {
|
||||
const namespace = resolveMetricsNamespace(config);
|
||||
if (!namespace) {
|
||||
throw new Error("AWS metrics namespace is required");
|
||||
}
|
||||
const client = new CloudWatchClient({
|
||||
region,
|
||||
credentials: credentials(config),
|
||||
});
|
||||
try {
|
||||
await client.send(
|
||||
new PutMetricDataCommand({
|
||||
Namespace: namespace,
|
||||
MetricData: [
|
||||
{
|
||||
MetricName: "dokploy_connection_test",
|
||||
Value: 1,
|
||||
Dimensions: [{ Name: "dokploy_test", Value: "true" }],
|
||||
},
|
||||
],
|
||||
}),
|
||||
{ abortSignal },
|
||||
);
|
||||
} catch (error) {
|
||||
if ((error as { name?: string })?.name === "AccessDenied") {
|
||||
throw new Error(
|
||||
"The IAM user needs cloudwatch:PutMetricData to ship metrics",
|
||||
);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,107 @@
|
||||
import { providerFetch } from "../provider-fetch";
|
||||
import type {
|
||||
TelemetryProviderAdapter,
|
||||
TelemetryProviderRuntimeConfig,
|
||||
TelemetrySignal,
|
||||
VectorSinkConfig,
|
||||
VectorTransformConfig,
|
||||
} from "../types";
|
||||
import { DEFAULT_DISK_BUFFER, normalizeEndpointUrl } from "../types";
|
||||
|
||||
const baseUri = (config: TelemetryProviderRuntimeConfig) =>
|
||||
normalizeEndpointUrl(config.endpoint ?? "").replace(/\/$/, "");
|
||||
|
||||
export const betterStackAdapter: TelemetryProviderAdapter = {
|
||||
type: "betterstack",
|
||||
signals: ["logs", "metrics"],
|
||||
label: "Better Stack",
|
||||
docsUrl: "https://betterstack.com/docs/logs/vector/",
|
||||
credentialFields: [
|
||||
{
|
||||
key: "apiKey",
|
||||
label: "Source Token",
|
||||
type: "password",
|
||||
required: true,
|
||||
helpText:
|
||||
"Bearer token from your Better Stack source. Testing the connection sends a real test log and/or metric to your account. Not every source accepts metrics; if yours does not, create a second provider for metrics with a source that does.",
|
||||
fullWidth: true,
|
||||
},
|
||||
{
|
||||
key: "endpoint",
|
||||
label: "Ingesting Host",
|
||||
type: "url",
|
||||
required: true,
|
||||
placeholder: "in.logs.betterstack.com",
|
||||
helpText:
|
||||
"Shown next to the Source Token when you create the source. Paste it as-is, no https:// needed.",
|
||||
},
|
||||
],
|
||||
toVectorTransform(
|
||||
_config: TelemetryProviderRuntimeConfig,
|
||||
_transformId: string,
|
||||
scopeTransformId: string,
|
||||
signal: TelemetrySignal,
|
||||
): VectorTransformConfig | null {
|
||||
if (signal === "metrics") return null;
|
||||
return {
|
||||
type: "remap",
|
||||
inputs: [scopeTransformId],
|
||||
source: ".dt = del(.timestamp)",
|
||||
};
|
||||
},
|
||||
toVectorSink(
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
_sinkId: string,
|
||||
inputId: string,
|
||||
signal: TelemetrySignal,
|
||||
): VectorSinkConfig {
|
||||
return {
|
||||
type: "http",
|
||||
inputs: [inputId],
|
||||
method: "post",
|
||||
uri: `${baseUri(config)}/${signal === "metrics" ? "metrics" : ""}`,
|
||||
encoding: { codec: "json" },
|
||||
compression: "gzip",
|
||||
auth: { strategy: "bearer", token: config.apiKey ?? "" },
|
||||
buffer: DEFAULT_DISK_BUFFER,
|
||||
};
|
||||
},
|
||||
async testConnection(config: TelemetryProviderRuntimeConfig): Promise<void> {
|
||||
if (!config.endpoint || !config.apiKey) {
|
||||
throw new Error(
|
||||
"Better Stack ingesting host and source token are required",
|
||||
);
|
||||
}
|
||||
const headers = {
|
||||
"Content-Type": "application/json",
|
||||
Authorization: `Bearer ${config.apiKey}`,
|
||||
};
|
||||
if (config.signals.includes("logs")) {
|
||||
const response = await providerFetch(`${baseUri(config)}/`, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify({
|
||||
message: "dokploy-connection-test",
|
||||
dokploy_test: true,
|
||||
}),
|
||||
});
|
||||
if (!response.ok) {
|
||||
throw new Error(
|
||||
`Better Stack test event failed with status ${response.status}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
if (config.signals.includes("metrics")) {
|
||||
const response = await providerFetch(`${baseUri(config)}/metrics`, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify([{ name: "dokploy_test", gauge: { value: 1 } }]),
|
||||
});
|
||||
if (!response.ok) {
|
||||
throw new Error(
|
||||
`Better Stack test metric failed with status ${response.status} — this source may not accept metrics`,
|
||||
);
|
||||
}
|
||||
}
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,97 @@
|
||||
import { providerFetch } from "../provider-fetch";
|
||||
import type {
|
||||
TelemetryProviderAdapter,
|
||||
TelemetryProviderRuntimeConfig,
|
||||
TelemetrySignal,
|
||||
VectorSinkConfig,
|
||||
VectorTransformConfig,
|
||||
} from "../types";
|
||||
import { DEFAULT_DISK_BUFFER } from "../types";
|
||||
|
||||
const DDTAGS_VRL = `
|
||||
tags = []
|
||||
if .dokploy_project != "" { tags = push(tags, "dokploy_project:" + replace(to_string!(.dokploy_project), ",", "_")) }
|
||||
if .dokploy_environment != "" { tags = push(tags, "dokploy_environment:" + replace(to_string!(.dokploy_environment), ",", "_")) }
|
||||
if .dokploy_application != "" { tags = push(tags, "dokploy_application:" + replace(to_string!(.dokploy_application), ",", "_")) }
|
||||
if .dokploy_organization != "" { tags = push(tags, "dokploy_organization:" + replace(to_string!(.dokploy_organization), ",", "_")) }
|
||||
if .dokploy_server != "" { tags = push(tags, "dokploy_server:" + replace(to_string!(.dokploy_server), ",", "_")) }
|
||||
.ddtags = join!(tags, ",")
|
||||
`.trim();
|
||||
|
||||
const resolveSite = (config: TelemetryProviderRuntimeConfig): string => {
|
||||
const raw = config.extraConfig?.site;
|
||||
if (typeof raw !== "string" || raw.length === 0) return "datadoghq.com";
|
||||
return raw
|
||||
.trim()
|
||||
.replace(/^https?:\/\//, "")
|
||||
.replace(/\/+$/, "");
|
||||
};
|
||||
|
||||
export const datadogAdapter: TelemetryProviderAdapter = {
|
||||
type: "datadog",
|
||||
signals: ["logs", "metrics"],
|
||||
label: "Datadog",
|
||||
docsUrl: "https://docs.datadoghq.com/logs/",
|
||||
credentialFields: [
|
||||
{
|
||||
key: "apiKey",
|
||||
label: "API Key",
|
||||
type: "password",
|
||||
required: true,
|
||||
fullWidth: true,
|
||||
},
|
||||
{
|
||||
key: "site",
|
||||
label: "Site",
|
||||
type: "text",
|
||||
required: false,
|
||||
placeholder: "datadoghq.com",
|
||||
helpText: "E.g. datadoghq.com (US1) or datadoghq.eu (EU).",
|
||||
fullWidth: true,
|
||||
},
|
||||
],
|
||||
toVectorTransform(
|
||||
_config: TelemetryProviderRuntimeConfig,
|
||||
_transformId: string,
|
||||
scopeTransformId: string,
|
||||
signal: TelemetrySignal,
|
||||
): VectorTransformConfig | null {
|
||||
if (signal === "metrics") return null;
|
||||
return {
|
||||
type: "remap",
|
||||
inputs: [scopeTransformId],
|
||||
source: DDTAGS_VRL,
|
||||
};
|
||||
},
|
||||
toVectorSink(
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
_sinkId: string,
|
||||
inputId: string,
|
||||
signal: TelemetrySignal,
|
||||
): VectorSinkConfig {
|
||||
return {
|
||||
type: signal === "metrics" ? "datadog_metrics" : "datadog_logs",
|
||||
inputs: [inputId],
|
||||
default_api_key: config.apiKey ?? "",
|
||||
site: resolveSite(config),
|
||||
buffer: DEFAULT_DISK_BUFFER,
|
||||
};
|
||||
},
|
||||
async testConnection(config: TelemetryProviderRuntimeConfig): Promise<void> {
|
||||
if (!config.apiKey) {
|
||||
throw new Error("Datadog API key is required");
|
||||
}
|
||||
const site = resolveSite(config);
|
||||
const response = await providerFetch(
|
||||
`https://api.${site}/api/v1/validate`,
|
||||
{
|
||||
headers: { "DD-API-KEY": config.apiKey },
|
||||
},
|
||||
);
|
||||
if (!response.ok) {
|
||||
throw new Error(
|
||||
`Datadog API key validation failed with status ${response.status}`,
|
||||
);
|
||||
}
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,185 @@
|
||||
import { providerFetch } from "../provider-fetch";
|
||||
import type {
|
||||
TelemetryProviderAdapter,
|
||||
TelemetryProviderRuntimeConfig,
|
||||
TelemetrySignal,
|
||||
VectorSinkConfig,
|
||||
VectorTransformConfig,
|
||||
} from "../types";
|
||||
import { DEFAULT_DISK_BUFFER, normalizeEndpointUrl } from "../types";
|
||||
|
||||
const LABEL_TO_JSON_VRL = `
|
||||
if exists(.label) { .label = encode_json(.label) }
|
||||
`.trim();
|
||||
|
||||
const DEFAULT_METRICS_INDEX = "dokploy-metrics-%Y.%m.%d";
|
||||
|
||||
type AuthMode =
|
||||
| { kind: "basic"; username: string; password: string }
|
||||
| { kind: "apiKey"; apiKey: string }
|
||||
| { kind: "none" };
|
||||
|
||||
const resolveAuthMode = (config: TelemetryProviderRuntimeConfig): AuthMode => {
|
||||
const username = config.extraConfig?.username;
|
||||
const hasUsername = typeof username === "string" && username.length > 0;
|
||||
if (hasUsername && !config.apiKey) {
|
||||
throw new Error(
|
||||
"A Password / API Key is required when Username is set (leave both blank for no auth).",
|
||||
);
|
||||
}
|
||||
if (hasUsername && config.apiKey) {
|
||||
return {
|
||||
kind: "basic",
|
||||
username: username as string,
|
||||
password: config.apiKey,
|
||||
};
|
||||
}
|
||||
if (config.apiKey) {
|
||||
return { kind: "apiKey", apiKey: config.apiKey };
|
||||
}
|
||||
return { kind: "none" };
|
||||
};
|
||||
|
||||
const buildAuthAndHeaders = (
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
): Record<string, unknown> => {
|
||||
const mode = resolveAuthMode(config);
|
||||
if (mode.kind === "basic") {
|
||||
return {
|
||||
auth: { strategy: "basic", user: mode.username, password: mode.password },
|
||||
};
|
||||
}
|
||||
if (mode.kind === "apiKey") {
|
||||
return {
|
||||
request: { headers: { Authorization: `ApiKey ${mode.apiKey}` } },
|
||||
};
|
||||
}
|
||||
return {};
|
||||
};
|
||||
|
||||
export const elasticsearchAdapter: TelemetryProviderAdapter = {
|
||||
type: "elasticsearch",
|
||||
signals: ["logs", "metrics"],
|
||||
label: "Elasticsearch / OpenSearch",
|
||||
docsUrl:
|
||||
"https://vector.dev/docs/reference/configuration/sinks/elasticsearch/",
|
||||
credentialFields: [
|
||||
{
|
||||
key: "endpoint",
|
||||
label: "Endpoint",
|
||||
type: "url",
|
||||
required: true,
|
||||
placeholder: "https://elasticsearch.example.com:9200",
|
||||
},
|
||||
{
|
||||
key: "username",
|
||||
label: "Username",
|
||||
type: "text",
|
||||
required: false,
|
||||
helpText: "Leave blank to use an API key instead of basic auth.",
|
||||
},
|
||||
{
|
||||
key: "apiKey",
|
||||
label: "Password / API Key",
|
||||
type: "password",
|
||||
required: false,
|
||||
helpText:
|
||||
"Password for Username, or an API key alone. Leave both blank for no auth.",
|
||||
fullWidth: true,
|
||||
},
|
||||
{
|
||||
key: "index",
|
||||
label: "Index",
|
||||
type: "text",
|
||||
required: false,
|
||||
signal: "logs",
|
||||
placeholder: "vector-%Y.%m.%d",
|
||||
helpText: "Defaults to vector-%Y.%m.%d, a daily index.",
|
||||
fullWidth: true,
|
||||
},
|
||||
{
|
||||
key: "metricsIndex",
|
||||
label: "Metrics index",
|
||||
type: "text",
|
||||
required: false,
|
||||
signal: "metrics",
|
||||
placeholder: DEFAULT_METRICS_INDEX,
|
||||
helpText: `Defaults to ${DEFAULT_METRICS_INDEX}, a daily index. Keep it separate from the logs index.`,
|
||||
fullWidth: true,
|
||||
},
|
||||
],
|
||||
toVectorTransform(
|
||||
_config: TelemetryProviderRuntimeConfig,
|
||||
_transformId: string,
|
||||
scopeTransformId: string,
|
||||
signal: TelemetrySignal,
|
||||
): VectorTransformConfig | null {
|
||||
if (signal === "metrics") return null;
|
||||
return {
|
||||
type: "remap",
|
||||
inputs: [scopeTransformId],
|
||||
source: LABEL_TO_JSON_VRL,
|
||||
};
|
||||
},
|
||||
toVectorSink(
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
_sinkId: string,
|
||||
inputId: string,
|
||||
signal: TelemetrySignal,
|
||||
): VectorSinkConfig {
|
||||
const common = {
|
||||
inputs: [inputId],
|
||||
endpoints: [normalizeEndpointUrl(config.endpoint ?? "")],
|
||||
api_version: "v8",
|
||||
...buildAuthAndHeaders(config),
|
||||
buffer: DEFAULT_DISK_BUFFER,
|
||||
};
|
||||
if (signal === "metrics") {
|
||||
const metricsIndex = config.extraConfig?.metricsIndex;
|
||||
return {
|
||||
type: "elasticsearch",
|
||||
...common,
|
||||
bulk: {
|
||||
index:
|
||||
typeof metricsIndex === "string" && metricsIndex.length > 0
|
||||
? metricsIndex
|
||||
: DEFAULT_METRICS_INDEX,
|
||||
},
|
||||
};
|
||||
}
|
||||
const index = config.extraConfig?.index;
|
||||
return {
|
||||
type: "elasticsearch",
|
||||
...common,
|
||||
...(typeof index === "string" && index.length > 0
|
||||
? { bulk: { index } }
|
||||
: {}),
|
||||
};
|
||||
},
|
||||
async testConnection(config: TelemetryProviderRuntimeConfig): Promise<void> {
|
||||
if (!config.endpoint) {
|
||||
throw new Error("Elasticsearch/OpenSearch endpoint is required");
|
||||
}
|
||||
const mode = resolveAuthMode(config);
|
||||
const headers: Record<string, string> = {};
|
||||
if (mode.kind === "basic") {
|
||||
headers.Authorization = `Basic ${Buffer.from(
|
||||
`${mode.username}:${mode.password}`,
|
||||
).toString("base64")}`;
|
||||
} else if (mode.kind === "apiKey") {
|
||||
headers.Authorization = `ApiKey ${mode.apiKey}`;
|
||||
}
|
||||
const response = await providerFetch(
|
||||
`${normalizeEndpointUrl(config.endpoint).replace(/\/$/, "")}/_cluster/health`,
|
||||
{ headers },
|
||||
);
|
||||
if (!response.ok) {
|
||||
throw new Error(
|
||||
`Elasticsearch/OpenSearch health check failed with status ${response.status}`,
|
||||
);
|
||||
}
|
||||
},
|
||||
validateConfig(config: TelemetryProviderRuntimeConfig): void {
|
||||
resolveAuthMode(config);
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,102 @@
|
||||
import { providerFetch } from "../provider-fetch";
|
||||
import type {
|
||||
TelemetryProviderAdapter,
|
||||
TelemetryProviderRuntimeConfig,
|
||||
VectorSinkConfig,
|
||||
} from "../types";
|
||||
import {
|
||||
DEFAULT_DISK_BUFFER,
|
||||
normalizeEndpointUrl,
|
||||
stringValue,
|
||||
} from "../types";
|
||||
|
||||
export const influxdbAdapter: TelemetryProviderAdapter = {
|
||||
type: "influxdb",
|
||||
signals: ["metrics"],
|
||||
label: "InfluxDB",
|
||||
docsUrl:
|
||||
"https://vector.dev/docs/reference/configuration/sinks/influxdb_metrics/",
|
||||
credentialFields: [
|
||||
{
|
||||
key: "endpoint",
|
||||
label: "URL",
|
||||
type: "url",
|
||||
required: true,
|
||||
placeholder: "http://influxdb.example.com:8086",
|
||||
helpText: "http://host:8086 or your InfluxDB Cloud URL.",
|
||||
fullWidth: true,
|
||||
},
|
||||
{
|
||||
key: "org",
|
||||
label: "Organization",
|
||||
type: "text",
|
||||
required: true,
|
||||
helpText: "InfluxDB 3 ignores it but the API requires it.",
|
||||
},
|
||||
{
|
||||
key: "bucket",
|
||||
label: "Bucket",
|
||||
type: "text",
|
||||
required: true,
|
||||
helpText: "Bucket (2.x) or database (3.x).",
|
||||
},
|
||||
{
|
||||
key: "apiKey",
|
||||
label: "API token",
|
||||
type: "password",
|
||||
required: true,
|
||||
helpText:
|
||||
"A token with write access to the bucket. Testing the connection writes a real data point.",
|
||||
fullWidth: true,
|
||||
},
|
||||
],
|
||||
toVectorSink(
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
_sinkId: string,
|
||||
inputId: string,
|
||||
): VectorSinkConfig {
|
||||
return {
|
||||
type: "influxdb_metrics",
|
||||
inputs: [inputId],
|
||||
endpoint: normalizeEndpointUrl(config.endpoint ?? ""),
|
||||
org: stringValue(config.extraConfig?.org),
|
||||
bucket: stringValue(config.extraConfig?.bucket),
|
||||
token: config.apiKey ?? "",
|
||||
default_namespace: "dokploy",
|
||||
version: "2",
|
||||
buffer: DEFAULT_DISK_BUFFER,
|
||||
};
|
||||
},
|
||||
async testConnection(config: TelemetryProviderRuntimeConfig): Promise<void> {
|
||||
const org = stringValue(config.extraConfig?.org);
|
||||
const bucket = stringValue(config.extraConfig?.bucket);
|
||||
if (!config.endpoint || !config.apiKey || !org || !bucket) {
|
||||
throw new Error(
|
||||
"InfluxDB URL, organization, bucket and token are required",
|
||||
);
|
||||
}
|
||||
const base = normalizeEndpointUrl(config.endpoint).replace(/\/$/, "");
|
||||
const query = new URLSearchParams({ org, bucket, precision: "s" });
|
||||
const response = await providerFetch(`${base}/api/v2/write?${query}`, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "text/plain; charset=utf-8",
|
||||
Authorization: `Token ${config.apiKey}`,
|
||||
},
|
||||
body: `dokploy_test,dokploy_test=true value=1 ${Math.floor(Date.now() / 1000)}`,
|
||||
});
|
||||
if (response.status === 401 || response.status === 403) {
|
||||
throw new Error(
|
||||
`InfluxDB rejected the token (status ${response.status})`,
|
||||
);
|
||||
}
|
||||
if (response.status === 404) {
|
||||
throw new Error("InfluxDB organization or bucket not found (status 404)");
|
||||
}
|
||||
if (!response.ok) {
|
||||
throw new Error(
|
||||
`InfluxDB write test failed with status ${response.status}`,
|
||||
);
|
||||
}
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,120 @@
|
||||
import { providerFetch } from "../provider-fetch";
|
||||
import type {
|
||||
TelemetryProviderAdapter,
|
||||
TelemetryProviderRuntimeConfig,
|
||||
VectorSinkConfig,
|
||||
} from "../types";
|
||||
import { DEFAULT_DISK_BUFFER, normalizeEndpointUrl } from "../types";
|
||||
|
||||
export const lokiAdapter: TelemetryProviderAdapter = {
|
||||
type: "loki",
|
||||
signals: ["logs"],
|
||||
label: "Grafana Loki",
|
||||
docsUrl: "https://grafana.com/docs/loki/latest/",
|
||||
credentialFields: [
|
||||
{
|
||||
key: "endpoint",
|
||||
label: "Endpoint",
|
||||
type: "url",
|
||||
required: true,
|
||||
placeholder: "https://loki.example.com",
|
||||
helpText:
|
||||
"Base URL of your Loki instance (push API). Loki only takes logs; for Grafana Cloud metrics, add a Prometheus remote write provider with the Prometheus Instance ID.",
|
||||
},
|
||||
{
|
||||
key: "username",
|
||||
label: "User / Instance ID",
|
||||
type: "text",
|
||||
required: false,
|
||||
helpText:
|
||||
"Leave blank for an unauthenticated Loki. For Grafana Cloud, this is your numeric Instance ID.",
|
||||
},
|
||||
{
|
||||
key: "apiKey",
|
||||
label: "Password / API Token",
|
||||
type: "password",
|
||||
required: false,
|
||||
helpText:
|
||||
"Leave blank for an unauthenticated Loki. For Grafana Cloud, use an API token here.",
|
||||
},
|
||||
{
|
||||
key: "tenantId",
|
||||
label: "Tenant ID",
|
||||
type: "text",
|
||||
required: false,
|
||||
helpText: "Only if your Loki runs in multi-tenant mode.",
|
||||
fullWidth: true,
|
||||
},
|
||||
],
|
||||
toVectorSink(
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
_sinkId: string,
|
||||
inputId: string,
|
||||
): VectorSinkConfig {
|
||||
const tenantId = config.extraConfig?.tenantId;
|
||||
const username = config.extraConfig?.username;
|
||||
return {
|
||||
type: "loki",
|
||||
inputs: [inputId],
|
||||
endpoint: normalizeEndpointUrl(config.endpoint ?? ""),
|
||||
encoding: { codec: "json" },
|
||||
labels: {
|
||||
dokploy_project: "{{ dokploy_project }}",
|
||||
dokploy_environment: "{{ dokploy_environment }}",
|
||||
dokploy_application: "{{ dokploy_application }}",
|
||||
dokploy_organization: "{{ dokploy_organization }}",
|
||||
dokploy_server: "{{ dokploy_server }}",
|
||||
},
|
||||
...(typeof tenantId === "string" && tenantId.length > 0
|
||||
? { tenant_id: tenantId }
|
||||
: {}),
|
||||
...(config.apiKey
|
||||
? {
|
||||
auth: {
|
||||
strategy: "basic",
|
||||
user: typeof username === "string" ? username : "",
|
||||
password: config.apiKey,
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
buffer: DEFAULT_DISK_BUFFER,
|
||||
};
|
||||
},
|
||||
async testConnection(config: TelemetryProviderRuntimeConfig): Promise<void> {
|
||||
if (!config.endpoint) {
|
||||
throw new Error("Loki endpoint is required");
|
||||
}
|
||||
const username = config.extraConfig?.username;
|
||||
const tenantId = config.extraConfig?.tenantId;
|
||||
const response = await providerFetch(
|
||||
`${normalizeEndpointUrl(config.endpoint).replace(/\/$/, "")}/loki/api/v1/push`,
|
||||
{
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
...(config.apiKey
|
||||
? {
|
||||
Authorization: `Basic ${Buffer.from(
|
||||
`${typeof username === "string" ? username : ""}:${config.apiKey}`,
|
||||
).toString("base64")}`,
|
||||
}
|
||||
: {}),
|
||||
...(typeof tenantId === "string" && tenantId.length > 0
|
||||
? { "X-Scope-OrgID": tenantId }
|
||||
: {}),
|
||||
},
|
||||
body: JSON.stringify({
|
||||
streams: [
|
||||
{
|
||||
stream: { dokploy_test: "true" },
|
||||
values: [[`${Date.now()}000000`, "dokploy-connection-test"]],
|
||||
},
|
||||
],
|
||||
}),
|
||||
},
|
||||
);
|
||||
if (!response.ok) {
|
||||
throw new Error(`Loki push test failed with status ${response.status}`);
|
||||
}
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,106 @@
|
||||
import { providerFetch } from "../provider-fetch";
|
||||
import type {
|
||||
TelemetryProviderAdapter,
|
||||
TelemetryProviderRuntimeConfig,
|
||||
VectorSinkConfig,
|
||||
} from "../types";
|
||||
import { DEFAULT_DISK_BUFFER } from "../types";
|
||||
|
||||
const resolveRegion = (config: TelemetryProviderRuntimeConfig): "us" | "eu" => {
|
||||
const raw = config.extraConfig?.region;
|
||||
const region = typeof raw === "string" ? raw.trim().toLowerCase() : "";
|
||||
if (region === "" || region === "us") return "us";
|
||||
if (region === "eu") return "eu";
|
||||
throw new Error('New Relic region must be "us" or "eu"');
|
||||
};
|
||||
|
||||
export const newRelicAdapter: TelemetryProviderAdapter = {
|
||||
type: "new_relic",
|
||||
signals: ["metrics"],
|
||||
label: "New Relic",
|
||||
docsUrl:
|
||||
"https://docs.newrelic.com/docs/data-apis/ingest-apis/metric-api/introduction-metric-api/",
|
||||
credentialFields: [
|
||||
{
|
||||
key: "apiKey",
|
||||
label: "License key",
|
||||
type: "password",
|
||||
required: true,
|
||||
helpText:
|
||||
"An INGEST - LICENSE key. Testing the connection sends a real data point.",
|
||||
fullWidth: true,
|
||||
},
|
||||
{
|
||||
key: "accountId",
|
||||
label: "Account ID",
|
||||
type: "text",
|
||||
required: true,
|
||||
placeholder: "1234567",
|
||||
helpText: "The numeric account ID.",
|
||||
},
|
||||
{
|
||||
key: "region",
|
||||
label: "Region",
|
||||
type: "text",
|
||||
required: false,
|
||||
placeholder: "us",
|
||||
helpText: "us (default) or eu.",
|
||||
},
|
||||
],
|
||||
validateConfig(config: TelemetryProviderRuntimeConfig): void {
|
||||
resolveRegion(config);
|
||||
},
|
||||
toVectorSink(
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
_sinkId: string,
|
||||
inputId: string,
|
||||
): VectorSinkConfig {
|
||||
const accountId = config.extraConfig?.accountId;
|
||||
return {
|
||||
type: "new_relic",
|
||||
inputs: [inputId],
|
||||
api: "metrics",
|
||||
license_key: config.apiKey ?? "",
|
||||
account_id: typeof accountId === "string" ? accountId.trim() : "",
|
||||
region: resolveRegion(config),
|
||||
buffer: DEFAULT_DISK_BUFFER,
|
||||
};
|
||||
},
|
||||
async testConnection(config: TelemetryProviderRuntimeConfig): Promise<void> {
|
||||
if (!config.apiKey) {
|
||||
throw new Error("New Relic license key is required");
|
||||
}
|
||||
const host =
|
||||
resolveRegion(config) === "eu"
|
||||
? "metric-api.eu.newrelic.com"
|
||||
: "metric-api.newrelic.com";
|
||||
const response = await providerFetch(`https://${host}/metric/v1`, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
"Api-Key": config.apiKey,
|
||||
},
|
||||
body: JSON.stringify([
|
||||
{
|
||||
metrics: [
|
||||
{
|
||||
name: "dokploy.test",
|
||||
type: "gauge",
|
||||
value: 1,
|
||||
timestamp: Math.floor(Date.now() / 1000),
|
||||
attributes: { dokploy_test: "true" },
|
||||
},
|
||||
],
|
||||
},
|
||||
]),
|
||||
});
|
||||
if (response.status === 403) {
|
||||
throw new Error("New Relic rejected the license key (status 403)");
|
||||
}
|
||||
if (!response.ok) {
|
||||
throw new Error(
|
||||
`New Relic metric test failed with status ${response.status}`,
|
||||
);
|
||||
}
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,131 @@
|
||||
import { providerFetch } from "../provider-fetch";
|
||||
import type {
|
||||
TelemetryProviderAdapter,
|
||||
TelemetryProviderRuntimeConfig,
|
||||
VectorSinkConfig,
|
||||
} from "../types";
|
||||
import { DEFAULT_DISK_BUFFER, stringValue } from "../types";
|
||||
|
||||
const assertHttpUrl = (endpoint: string | null) => {
|
||||
let url: URL;
|
||||
try {
|
||||
url = new URL(endpoint ?? "");
|
||||
} catch {
|
||||
throw new Error("Remote write URL must be a valid http(s) URL");
|
||||
}
|
||||
if (url.protocol !== "http:" && url.protocol !== "https:") {
|
||||
throw new Error("Remote write URL must be a valid http(s) URL");
|
||||
}
|
||||
return url;
|
||||
};
|
||||
|
||||
const buildAuth = (config: TelemetryProviderRuntimeConfig) => {
|
||||
const username = stringValue(config.extraConfig?.username);
|
||||
if (!config.apiKey) return {};
|
||||
return username
|
||||
? { auth: { strategy: "basic", user: username, password: config.apiKey } }
|
||||
: { auth: { strategy: "bearer", token: config.apiKey } };
|
||||
};
|
||||
|
||||
const buildAuthHeader = (
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
): Record<string, string> => {
|
||||
const username = stringValue(config.extraConfig?.username);
|
||||
if (!config.apiKey) return {};
|
||||
return username
|
||||
? {
|
||||
Authorization: `Basic ${Buffer.from(`${username}:${config.apiKey}`).toString("base64")}`,
|
||||
}
|
||||
: { Authorization: `Bearer ${config.apiKey}` };
|
||||
};
|
||||
|
||||
export const prometheusRemoteWriteAdapter: TelemetryProviderAdapter = {
|
||||
type: "prometheus_remote_write",
|
||||
signals: ["metrics"],
|
||||
label: "Prometheus remote write",
|
||||
docsUrl:
|
||||
"https://vector.dev/docs/reference/configuration/sinks/prometheus_remote_write/",
|
||||
credentialFields: [
|
||||
{
|
||||
key: "endpoint",
|
||||
label: "Remote write URL",
|
||||
type: "url",
|
||||
required: true,
|
||||
placeholder: "http://prometheus.example.com:9090/api/v1/write",
|
||||
helpText:
|
||||
"Prometheus: http://host:9090/api/v1/write (start it with --web.enable-remote-write-receiver). Grafana Cloud: the /api/prom/push URL from the Prometheus card. Mimir: http://host:9009/api/v1/push. VictoriaMetrics: http://host:8428/api/v1/write.",
|
||||
fullWidth: true,
|
||||
},
|
||||
{
|
||||
key: "username",
|
||||
label: "Username / Instance ID",
|
||||
type: "text",
|
||||
required: false,
|
||||
helpText:
|
||||
"Leave blank if the endpoint needs no auth. Grafana Cloud: the Prometheus Instance ID (not the Loki one).",
|
||||
},
|
||||
{
|
||||
key: "apiKey",
|
||||
label: "Password / Token",
|
||||
type: "password",
|
||||
required: false,
|
||||
helpText:
|
||||
"With a username it is sent as basic auth; alone, as a bearer token. Grafana Cloud: an Access Policy token with metrics:write.",
|
||||
},
|
||||
{
|
||||
key: "tenantId",
|
||||
label: "Tenant ID",
|
||||
type: "text",
|
||||
required: false,
|
||||
helpText: "Only for multi-tenant Mimir/Cortex (X-Scope-OrgID header).",
|
||||
fullWidth: true,
|
||||
},
|
||||
],
|
||||
validateConfig(config: TelemetryProviderRuntimeConfig): void {
|
||||
assertHttpUrl(config.endpoint);
|
||||
},
|
||||
toVectorSink(
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
_sinkId: string,
|
||||
inputId: string,
|
||||
): VectorSinkConfig {
|
||||
const tenantId = stringValue(config.extraConfig?.tenantId);
|
||||
return {
|
||||
type: "prometheus_remote_write",
|
||||
inputs: [inputId],
|
||||
endpoint: config.endpoint ?? "",
|
||||
// The sink healthcheck is an empty POST that Prometheus and Mimir answer with 400.
|
||||
healthcheck: { enabled: false },
|
||||
...(tenantId ? { tenant_id: tenantId } : {}),
|
||||
...buildAuth(config),
|
||||
buffer: DEFAULT_DISK_BUFFER,
|
||||
};
|
||||
},
|
||||
async testConnection(config: TelemetryProviderRuntimeConfig): Promise<void> {
|
||||
const url = assertHttpUrl(config.endpoint);
|
||||
const tenantId = stringValue(config.extraConfig?.tenantId);
|
||||
// Remote write needs snappy protobuf, so the empty test body gets a 400 once auth and path pass.
|
||||
const response = await providerFetch(url.toString(), {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/x-protobuf",
|
||||
"Content-Encoding": "snappy",
|
||||
"X-Prometheus-Remote-Write-Version": "0.1.0",
|
||||
...buildAuthHeader(config),
|
||||
...(tenantId ? { "X-Scope-OrgID": tenantId } : {}),
|
||||
},
|
||||
});
|
||||
if (response.ok || response.status === 400) return;
|
||||
if (response.status === 401 || response.status === 403) {
|
||||
throw new Error(
|
||||
`Remote write endpoint rejected the credentials (status ${response.status})`,
|
||||
);
|
||||
}
|
||||
if (response.status === 404 || response.status === 405) {
|
||||
throw new Error(
|
||||
`Remote write endpoint not found at this URL (status ${response.status}) — check the path`,
|
||||
);
|
||||
}
|
||||
throw new Error(`Remote write test failed with status ${response.status}`);
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,37 @@
|
||||
import type { TelemetryProviderAdapter, TelemetryProviderType } from "../types";
|
||||
import { awsCloudwatchAdapter } from "./aws-cloudwatch";
|
||||
import { betterStackAdapter } from "./betterstack";
|
||||
import { datadogAdapter } from "./datadog";
|
||||
import { elasticsearchAdapter } from "./elasticsearch";
|
||||
import { influxdbAdapter } from "./influxdb";
|
||||
import { lokiAdapter } from "./loki";
|
||||
import { newRelicAdapter } from "./new-relic";
|
||||
import { prometheusRemoteWriteAdapter } from "./prometheus-remote-write";
|
||||
import { splunkAdapter } from "./splunk";
|
||||
|
||||
export const telemetryProviderAdapters: Record<
|
||||
TelemetryProviderType,
|
||||
TelemetryProviderAdapter
|
||||
> = {
|
||||
loki: lokiAdapter,
|
||||
prometheus_remote_write: prometheusRemoteWriteAdapter,
|
||||
new_relic: newRelicAdapter,
|
||||
influxdb: influxdbAdapter,
|
||||
datadog: datadogAdapter,
|
||||
aws_cloudwatch: awsCloudwatchAdapter,
|
||||
splunk_hec: splunkAdapter,
|
||||
elasticsearch: elasticsearchAdapter,
|
||||
betterstack: betterStackAdapter,
|
||||
};
|
||||
|
||||
export function getTelemetryProviderAdapter(
|
||||
type: TelemetryProviderType,
|
||||
): TelemetryProviderAdapter {
|
||||
const adapter = telemetryProviderAdapters[type];
|
||||
if (!adapter) {
|
||||
throw new Error(
|
||||
`No TelemetryProviderAdapter registered for type "${type}"`,
|
||||
);
|
||||
}
|
||||
return adapter;
|
||||
}
|
||||
@@ -0,0 +1,154 @@
|
||||
import { providerFetch } from "../provider-fetch";
|
||||
import type {
|
||||
TelemetryProviderAdapter,
|
||||
TelemetryProviderRuntimeConfig,
|
||||
TelemetrySignal,
|
||||
VectorSinkConfig,
|
||||
} from "../types";
|
||||
import {
|
||||
DEFAULT_DISK_BUFFER,
|
||||
normalizeEndpointUrl,
|
||||
stringValue,
|
||||
} from "../types";
|
||||
|
||||
const baseUrl = (config: TelemetryProviderRuntimeConfig) =>
|
||||
normalizeEndpointUrl(config.endpoint ?? "").replace(/\/$/, "");
|
||||
|
||||
export const splunkAdapter: TelemetryProviderAdapter = {
|
||||
type: "splunk_hec",
|
||||
signals: ["logs", "metrics"],
|
||||
label: "Splunk HTTP Event Collector",
|
||||
docsUrl:
|
||||
"https://vector.dev/docs/reference/configuration/sinks/splunk_hec_logs/",
|
||||
credentialFields: [
|
||||
{
|
||||
key: "endpoint",
|
||||
label: "Endpoint",
|
||||
type: "url",
|
||||
required: true,
|
||||
placeholder: "https://splunk.example.com:8088",
|
||||
helpText: "Base URL of your Splunk instance, including the HEC port.",
|
||||
},
|
||||
{
|
||||
key: "apiKey",
|
||||
label: "HEC Token",
|
||||
type: "password",
|
||||
required: true,
|
||||
},
|
||||
{
|
||||
key: "index",
|
||||
label: "Index",
|
||||
type: "text",
|
||||
required: false,
|
||||
signal: "logs",
|
||||
},
|
||||
{
|
||||
key: "sourcetype",
|
||||
label: "Source Type",
|
||||
type: "text",
|
||||
required: false,
|
||||
signal: "logs",
|
||||
placeholder: "httpevent",
|
||||
},
|
||||
{
|
||||
key: "metricsIndex",
|
||||
label: "Metrics index",
|
||||
type: "text",
|
||||
required: true,
|
||||
signal: "metrics",
|
||||
placeholder: "dokploy_metrics",
|
||||
helpText:
|
||||
"An index of type Metrics the token can write to (Allowed Indexes). If the token can't, create a second Splunk provider for metrics with another token. Testing the connection writes a real data point.",
|
||||
fullWidth: true,
|
||||
},
|
||||
],
|
||||
toVectorSink(
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
_sinkId: string,
|
||||
inputId: string,
|
||||
signal: TelemetrySignal,
|
||||
): VectorSinkConfig {
|
||||
if (signal === "metrics") {
|
||||
return {
|
||||
type: "splunk_hec_metrics",
|
||||
inputs: [inputId],
|
||||
endpoint: normalizeEndpointUrl(config.endpoint ?? ""),
|
||||
default_token: config.apiKey ?? "",
|
||||
index: stringValue(config.extraConfig?.metricsIndex),
|
||||
source: "dokploy",
|
||||
sourcetype: "dokploy:metrics",
|
||||
buffer: DEFAULT_DISK_BUFFER,
|
||||
};
|
||||
}
|
||||
const index = config.extraConfig?.index;
|
||||
const sourcetype = config.extraConfig?.sourcetype;
|
||||
return {
|
||||
type: "splunk_hec_logs",
|
||||
inputs: [inputId],
|
||||
endpoint: normalizeEndpointUrl(config.endpoint ?? ""),
|
||||
default_token: config.apiKey ?? "",
|
||||
encoding: { codec: "json" },
|
||||
...(typeof index === "string" && index.length > 0 ? { index } : {}),
|
||||
...(typeof sourcetype === "string" && sourcetype.length > 0
|
||||
? { sourcetype }
|
||||
: {}),
|
||||
buffer: DEFAULT_DISK_BUFFER,
|
||||
};
|
||||
},
|
||||
async testConnection(config: TelemetryProviderRuntimeConfig): Promise<void> {
|
||||
if (!config.endpoint || !config.apiKey) {
|
||||
throw new Error("Splunk endpoint and HEC token are required");
|
||||
}
|
||||
const authorization = { Authorization: `Splunk ${config.apiKey}` };
|
||||
if (config.signals.includes("logs")) {
|
||||
const response = await providerFetch(
|
||||
`${baseUrl(config)}/services/collector/health`,
|
||||
{ headers: authorization },
|
||||
);
|
||||
if (!response.ok) {
|
||||
throw new Error(
|
||||
`Splunk HEC health check failed with status ${response.status}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
if (config.signals.includes("metrics")) {
|
||||
const metricsIndex = stringValue(config.extraConfig?.metricsIndex);
|
||||
if (!metricsIndex) {
|
||||
throw new Error("Splunk metrics index is required");
|
||||
}
|
||||
const response = await providerFetch(
|
||||
`${baseUrl(config)}/services/collector`,
|
||||
{
|
||||
method: "POST",
|
||||
headers: { ...authorization, "Content-Type": "application/json" },
|
||||
body: JSON.stringify({
|
||||
time: Math.floor(Date.now() / 1000),
|
||||
event: "metric",
|
||||
source: "dokploy",
|
||||
index: metricsIndex,
|
||||
fields: { metric_name: "dokploy.test", _value: 1 },
|
||||
}),
|
||||
},
|
||||
);
|
||||
if (response.status === 401 || response.status === 403) {
|
||||
throw new Error(
|
||||
`Splunk rejected the HEC token (status ${response.status})`,
|
||||
);
|
||||
}
|
||||
const body = (await response.json().catch(() => null)) as {
|
||||
code?: number;
|
||||
text?: string;
|
||||
} | null;
|
||||
if (body?.code === 7) {
|
||||
throw new Error(
|
||||
`The HEC token can't write to the metrics index "${metricsIndex}"`,
|
||||
);
|
||||
}
|
||||
if (!response.ok || (body && body.code !== 0)) {
|
||||
throw new Error(
|
||||
`Splunk metric test failed with status ${response.status}${body?.text ? `: ${body.text}` : ""}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,415 @@
|
||||
import { db } from "@dokploy/server/db";
|
||||
import {
|
||||
type apiCreateTelemetryProvider,
|
||||
server,
|
||||
telemetryProvider,
|
||||
webServerSettings,
|
||||
} from "@dokploy/server/db/schema";
|
||||
import { TRPCError } from "@trpc/server";
|
||||
import { and, arrayContains, eq, sql } from "drizzle-orm";
|
||||
import type { z } from "zod";
|
||||
import { getTelemetryProviderAdapter } from "./providers/registry";
|
||||
import {
|
||||
TELEMETRY_SIGNALS,
|
||||
type TelemetryProviderRuntimeConfig,
|
||||
type TelemetryProviderType,
|
||||
type TelemetrySignal,
|
||||
} from "./types";
|
||||
|
||||
export type TelemetryProvider = typeof telemetryProvider.$inferSelect;
|
||||
|
||||
const CREDENTIAL_COLUMNS = {
|
||||
endpoint: false,
|
||||
apiKey: false,
|
||||
apiSecret: false,
|
||||
} as const;
|
||||
|
||||
const sortSignals = (signals: TelemetrySignal[]) =>
|
||||
TELEMETRY_SIGNALS.filter((signal) => signals.includes(signal));
|
||||
|
||||
export const assertSignalsSupported = (
|
||||
providerType: TelemetryProviderType,
|
||||
signals: TelemetrySignal[],
|
||||
) => {
|
||||
const adapter = getTelemetryProviderAdapter(providerType);
|
||||
if (new Set(signals).size !== signals.length) {
|
||||
throw new TRPCError({
|
||||
code: "BAD_REQUEST",
|
||||
message: "A signal can only be selected once",
|
||||
});
|
||||
}
|
||||
const unsupported = signals.filter((s) => !adapter.signals.includes(s));
|
||||
if (unsupported.length > 0) {
|
||||
throw new TRPCError({
|
||||
code: "BAD_REQUEST",
|
||||
message: `${adapter.label} cannot send ${unsupported.join(", ")}`,
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
export const assertRequiredCredentialFields = (
|
||||
providerType: TelemetryProviderType,
|
||||
signals: TelemetrySignal[],
|
||||
values: {
|
||||
endpoint?: string | null;
|
||||
apiKey?: string | null;
|
||||
apiSecret?: string | null;
|
||||
extraConfig?: Record<string, unknown> | null;
|
||||
},
|
||||
) => {
|
||||
const adapter = getTelemetryProviderAdapter(providerType);
|
||||
const fieldValues: Record<string, unknown> = {
|
||||
...(values.extraConfig ?? {}),
|
||||
endpoint: values.endpoint,
|
||||
apiKey: values.apiKey,
|
||||
apiSecret: values.apiSecret,
|
||||
};
|
||||
const missing = adapter.credentialFields
|
||||
.filter((field) => !field.signal || signals.includes(field.signal))
|
||||
.filter((field) => field.required)
|
||||
.filter((field) => {
|
||||
const value = fieldValues[field.key];
|
||||
return value === undefined || value === null || value === "";
|
||||
})
|
||||
.map((field) => field.label);
|
||||
|
||||
if (missing.length > 0) {
|
||||
throw new TRPCError({
|
||||
code: "BAD_REQUEST",
|
||||
message: `Missing required field(s) for ${adapter.label}: ${missing.join(", ")}`,
|
||||
});
|
||||
}
|
||||
|
||||
if (adapter.validateConfig) {
|
||||
try {
|
||||
adapter.validateConfig({
|
||||
telemetryProviderId: "validate",
|
||||
name: "",
|
||||
signals,
|
||||
endpoint: values.endpoint ?? null,
|
||||
apiKey: values.apiKey ?? null,
|
||||
apiSecret: values.apiSecret ?? null,
|
||||
extraConfig: values.extraConfig ?? null,
|
||||
});
|
||||
} catch (error) {
|
||||
throw new TRPCError({
|
||||
code: "BAD_REQUEST",
|
||||
message:
|
||||
error instanceof Error ? error.message : "Invalid configuration",
|
||||
});
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
export const createTelemetryProvider = async (
|
||||
input: z.infer<typeof apiCreateTelemetryProvider>,
|
||||
organizationId: string,
|
||||
) => {
|
||||
assertSignalsSupported(input.providerType, input.signals);
|
||||
assertRequiredCredentialFields(input.providerType, input.signals, input);
|
||||
|
||||
const created = await db
|
||||
.insert(telemetryProvider)
|
||||
.values({
|
||||
...input,
|
||||
signals: sortSignals(input.signals),
|
||||
organizationId,
|
||||
})
|
||||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
|
||||
if (!created) {
|
||||
throw new TRPCError({
|
||||
code: "BAD_REQUEST",
|
||||
message: "Error creating provider",
|
||||
});
|
||||
}
|
||||
return created;
|
||||
};
|
||||
|
||||
// null is the Dokploy host.
|
||||
export type VectorTarget = string | null;
|
||||
|
||||
const findTargetsUsingProvider = async (
|
||||
telemetryProviderId: string,
|
||||
): Promise<VectorTarget[]> => {
|
||||
const [servers, local] = await Promise.all([
|
||||
db.query.server.findMany({
|
||||
where: arrayContains(server.telemetryProviderIds, [telemetryProviderId]),
|
||||
columns: { serverId: true },
|
||||
}),
|
||||
db.query.webServerSettings.findFirst({
|
||||
where: arrayContains(webServerSettings.telemetryProviderIds, [
|
||||
telemetryProviderId,
|
||||
]),
|
||||
columns: { id: true },
|
||||
}),
|
||||
]);
|
||||
return [...servers.map((s) => s.serverId), ...(local ? [null] : [])];
|
||||
};
|
||||
|
||||
const SHIPPING_FIELDS = [
|
||||
"providerType",
|
||||
"enabled",
|
||||
"endpoint",
|
||||
"apiKey",
|
||||
"apiSecret",
|
||||
"extraConfig",
|
||||
"signals",
|
||||
] as const;
|
||||
|
||||
export const updateTelemetryProvider = async (
|
||||
telemetryProviderId: string,
|
||||
data: Partial<z.infer<typeof apiCreateTelemetryProvider>>,
|
||||
) => {
|
||||
const existing =
|
||||
await findTelemetryProviderByIdWithCredentials(telemetryProviderId);
|
||||
const isChangingType =
|
||||
data.providerType !== undefined &&
|
||||
data.providerType !== existing.providerType;
|
||||
const dataToApply = isChangingType
|
||||
? {
|
||||
endpoint: null,
|
||||
apiKey: null,
|
||||
apiSecret: null,
|
||||
extraConfig: null,
|
||||
...data,
|
||||
}
|
||||
: data.extraConfig != null
|
||||
? {
|
||||
...data,
|
||||
extraConfig: { ...(existing.extraConfig ?? {}), ...data.extraConfig },
|
||||
}
|
||||
: data;
|
||||
const merged = { ...existing, ...dataToApply };
|
||||
const signals = merged.signals as TelemetrySignal[];
|
||||
assertSignalsSupported(merged.providerType, signals);
|
||||
assertRequiredCredentialFields(merged.providerType, signals, merged);
|
||||
if (dataToApply.signals) {
|
||||
dataToApply.signals = sortSignals(dataToApply.signals);
|
||||
}
|
||||
|
||||
const updated = await db
|
||||
.update(telemetryProvider)
|
||||
.set(dataToApply)
|
||||
.where(eq(telemetryProvider.telemetryProviderId, telemetryProviderId))
|
||||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
|
||||
if (!updated) {
|
||||
throw new TRPCError({
|
||||
code: "NOT_FOUND",
|
||||
message: "Provider not found",
|
||||
});
|
||||
}
|
||||
const shippingChanged = SHIPPING_FIELDS.some(
|
||||
(field) =>
|
||||
field in dataToApply &&
|
||||
JSON.stringify(dataToApply[field]) !== JSON.stringify(existing[field]),
|
||||
);
|
||||
return {
|
||||
provider: updated,
|
||||
targets: shippingChanged
|
||||
? await findTargetsUsingProvider(telemetryProviderId)
|
||||
: [],
|
||||
};
|
||||
};
|
||||
|
||||
export const removeTelemetryProvider = async (telemetryProviderId: string) => {
|
||||
return await db.transaction(async (tx) => {
|
||||
const removed = await tx
|
||||
.delete(telemetryProvider)
|
||||
.where(eq(telemetryProvider.telemetryProviderId, telemetryProviderId))
|
||||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
|
||||
if (!removed) {
|
||||
throw new TRPCError({
|
||||
code: "NOT_FOUND",
|
||||
message: "Provider not found",
|
||||
});
|
||||
}
|
||||
|
||||
const servers = await tx
|
||||
.update(server)
|
||||
.set({
|
||||
telemetryProviderIds: sql`array_remove(${server.telemetryProviderIds}, ${telemetryProviderId})`,
|
||||
})
|
||||
.where(arrayContains(server.telemetryProviderIds, [telemetryProviderId]))
|
||||
.returning({ serverId: server.serverId });
|
||||
|
||||
const local = await tx
|
||||
.update(webServerSettings)
|
||||
.set({
|
||||
telemetryProviderIds: sql`array_remove(${webServerSettings.telemetryProviderIds}, ${telemetryProviderId})`,
|
||||
vectorAgentOrganizationId: sql`CASE WHEN cardinality(array_remove(${webServerSettings.telemetryProviderIds}, ${telemetryProviderId})) = 0 THEN NULL ELSE ${webServerSettings.vectorAgentOrganizationId} END`,
|
||||
})
|
||||
.where(
|
||||
arrayContains(webServerSettings.telemetryProviderIds, [
|
||||
telemetryProviderId,
|
||||
]),
|
||||
)
|
||||
.returning({ id: webServerSettings.id });
|
||||
|
||||
const targets: VectorTarget[] = [
|
||||
...servers.map((s) => s.serverId),
|
||||
...(local.length > 0 ? [null] : []),
|
||||
];
|
||||
return { provider: removed, targets };
|
||||
});
|
||||
};
|
||||
|
||||
export const findTelemetryProviderForOrganization = async (
|
||||
telemetryProviderId: string,
|
||||
organizationId: string,
|
||||
) => {
|
||||
const provider = await findTelemetryProviderById(telemetryProviderId);
|
||||
if (provider.organizationId !== organizationId) {
|
||||
throw new TRPCError({
|
||||
code: "UNAUTHORIZED",
|
||||
message: "You are not allowed to access this provider",
|
||||
});
|
||||
}
|
||||
return provider;
|
||||
};
|
||||
|
||||
export const findTelemetryProviderById = async (
|
||||
telemetryProviderId: string,
|
||||
) => {
|
||||
const found = await db.query.telemetryProvider.findFirst({
|
||||
where: eq(telemetryProvider.telemetryProviderId, telemetryProviderId),
|
||||
columns: CREDENTIAL_COLUMNS,
|
||||
});
|
||||
if (!found) {
|
||||
throw new TRPCError({
|
||||
code: "NOT_FOUND",
|
||||
message: "Provider not found",
|
||||
});
|
||||
}
|
||||
return found;
|
||||
};
|
||||
|
||||
export const findTelemetryProviderByIdWithCredentials = async (
|
||||
telemetryProviderId: string,
|
||||
) => {
|
||||
const found = await db.query.telemetryProvider.findFirst({
|
||||
where: eq(telemetryProvider.telemetryProviderId, telemetryProviderId),
|
||||
});
|
||||
if (!found) {
|
||||
throw new TRPCError({
|
||||
code: "NOT_FOUND",
|
||||
message: "Provider not found",
|
||||
});
|
||||
}
|
||||
return found;
|
||||
};
|
||||
|
||||
export const findTelemetryProvidersByOrganization = async (
|
||||
organizationId: string,
|
||||
) => {
|
||||
return await db.query.telemetryProvider.findMany({
|
||||
where: eq(telemetryProvider.organizationId, organizationId),
|
||||
columns: CREDENTIAL_COLUMNS,
|
||||
});
|
||||
};
|
||||
|
||||
export const findEnabledTelemetryProvidersByOrganization = async (
|
||||
organizationId: string,
|
||||
) => {
|
||||
return await db.query.telemetryProvider.findMany({
|
||||
where: and(
|
||||
eq(telemetryProvider.organizationId, organizationId),
|
||||
eq(telemetryProvider.enabled, true),
|
||||
),
|
||||
});
|
||||
};
|
||||
|
||||
export const toRuntimeConfig = (
|
||||
provider: TelemetryProvider,
|
||||
): TelemetryProviderRuntimeConfig => ({
|
||||
telemetryProviderId: provider.telemetryProviderId,
|
||||
name: provider.name,
|
||||
signals: provider.signals as TelemetrySignal[],
|
||||
endpoint: provider.endpoint,
|
||||
apiKey: provider.apiKey,
|
||||
apiSecret: provider.apiSecret,
|
||||
extraConfig: provider.extraConfig,
|
||||
});
|
||||
|
||||
export const testTelemetryProviderConnection = async (
|
||||
params:
|
||||
| { telemetryProviderId: string; signals?: TelemetrySignal[] }
|
||||
| {
|
||||
providerType: TelemetryProviderType;
|
||||
config: TelemetryProviderRuntimeConfig;
|
||||
},
|
||||
) => {
|
||||
let providerType: TelemetryProviderType;
|
||||
let runtimeConfig: TelemetryProviderRuntimeConfig;
|
||||
|
||||
if ("telemetryProviderId" in params) {
|
||||
const provider = await findTelemetryProviderByIdWithCredentials(
|
||||
params.telemetryProviderId,
|
||||
);
|
||||
providerType = provider.providerType;
|
||||
runtimeConfig = {
|
||||
...toRuntimeConfig(provider),
|
||||
...(params.signals ? { signals: params.signals } : {}),
|
||||
};
|
||||
} else {
|
||||
providerType = params.providerType;
|
||||
runtimeConfig = params.config;
|
||||
}
|
||||
|
||||
const adapter = getTelemetryProviderAdapter(providerType);
|
||||
if (!adapter.testConnection) {
|
||||
return {
|
||||
success: true,
|
||||
warning: "This provider does not support connection testing",
|
||||
};
|
||||
}
|
||||
|
||||
try {
|
||||
await adapter.testConnection(runtimeConfig);
|
||||
return { success: true };
|
||||
} catch (error) {
|
||||
throw new TRPCError({
|
||||
code: "BAD_REQUEST",
|
||||
message:
|
||||
error instanceof Error ? error.message : "Connection test failed",
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
export const sanitizeTelemetryProvider = <
|
||||
T extends { endpoint?: unknown; apiKey?: unknown; apiSecret?: unknown },
|
||||
>(
|
||||
provider: T,
|
||||
) => {
|
||||
const { endpoint, apiKey, apiSecret, ...rest } = provider;
|
||||
return rest;
|
||||
};
|
||||
|
||||
export const assertProvidersBelongToOrg = async (
|
||||
telemetryProviderIds: string[],
|
||||
organizationId: string,
|
||||
) => {
|
||||
const providers = await findTelemetryProvidersByOrganization(organizationId);
|
||||
const byId = new Map(providers.map((p) => [p.telemetryProviderId, p]));
|
||||
for (const id of telemetryProviderIds) {
|
||||
const provider = byId.get(id);
|
||||
if (!provider) {
|
||||
throw new TRPCError({
|
||||
code: "UNAUTHORIZED",
|
||||
message: "One of the selected providers is not in this organization",
|
||||
});
|
||||
}
|
||||
}
|
||||
if (!telemetryProviderIds.some((id) => byId.get(id)?.enabled)) {
|
||||
throw new TRPCError({
|
||||
code: "BAD_REQUEST",
|
||||
message:
|
||||
"Select at least one enabled provider — a disabled one ships nothing.",
|
||||
});
|
||||
}
|
||||
};
|
||||
@@ -0,0 +1,85 @@
|
||||
import type { telemetryProviderType } from "@dokploy/server/db/schema/telemetry-provider";
|
||||
|
||||
export type TelemetrySignal = "logs" | "metrics";
|
||||
|
||||
export const TELEMETRY_SIGNALS: TelemetrySignal[] = ["logs", "metrics"];
|
||||
|
||||
export interface TelemetryProviderCredentialField {
|
||||
key: string;
|
||||
label: string;
|
||||
type: "text" | "password" | "url";
|
||||
required: boolean;
|
||||
signal?: TelemetrySignal;
|
||||
placeholder?: string;
|
||||
helpText?: string;
|
||||
fullWidth?: boolean;
|
||||
}
|
||||
|
||||
export interface VectorTransformConfig {
|
||||
type: string;
|
||||
inputs: string[];
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
export interface VectorSinkConfig {
|
||||
type: string;
|
||||
inputs: string[];
|
||||
buffer: {
|
||||
type: "disk" | "memory";
|
||||
max_size: number;
|
||||
when_full?: "block" | "drop_newest";
|
||||
};
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
export interface TelemetryProviderRuntimeConfig {
|
||||
telemetryProviderId: string;
|
||||
name: string;
|
||||
signals: TelemetrySignal[];
|
||||
endpoint: string | null;
|
||||
apiKey: string | null;
|
||||
apiSecret: string | null;
|
||||
extraConfig: Record<string, unknown> | null;
|
||||
}
|
||||
|
||||
export type TelemetryProviderType =
|
||||
(typeof telemetryProviderType.enumValues)[number];
|
||||
|
||||
export const DEFAULT_DISK_BUFFER = {
|
||||
type: "disk" as const,
|
||||
max_size: 268_435_488,
|
||||
when_full: "block" as const,
|
||||
};
|
||||
|
||||
export const normalizeEndpointUrl = (endpoint: string): string =>
|
||||
!endpoint || /^https?:\/\//.test(endpoint) ? endpoint : `https://${endpoint}`;
|
||||
|
||||
export const stringValue = (value: unknown): string =>
|
||||
typeof value === "string" ? value.trim() : "";
|
||||
|
||||
// VRL string literals do not accept JSON's \b, \f or \uXXXX escapes, so control characters are dropped first.
|
||||
export const vrlString = (value: string): string =>
|
||||
// biome-ignore lint/suspicious/noControlCharactersInRegex: that is the point
|
||||
JSON.stringify(value.replace(/[\u0000-\u001f\u007f]/g, ""));
|
||||
|
||||
export interface TelemetryProviderAdapter {
|
||||
type: TelemetryProviderType;
|
||||
signals: TelemetrySignal[];
|
||||
label: string;
|
||||
docsUrl?: string;
|
||||
credentialFields: TelemetryProviderCredentialField[];
|
||||
toVectorTransform?(
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
transformId: string,
|
||||
scopeTransformId: string,
|
||||
signal: TelemetrySignal,
|
||||
): VectorTransformConfig | null;
|
||||
toVectorSink(
|
||||
config: TelemetryProviderRuntimeConfig,
|
||||
sinkId: string,
|
||||
inputId: string,
|
||||
signal: TelemetrySignal,
|
||||
): VectorSinkConfig;
|
||||
testConnection?(config: TelemetryProviderRuntimeConfig): Promise<void>;
|
||||
validateConfig?(config: TelemetryProviderRuntimeConfig): void;
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
import { IS_CLOUD } from "@dokploy/server/constants";
|
||||
import {
|
||||
findServerById,
|
||||
getAccessibleServerIds,
|
||||
type Server,
|
||||
setServerProviderIds,
|
||||
} from "@dokploy/server/services/server";
|
||||
import { setWebServerProviderIds } from "@dokploy/server/services/web-server-settings";
|
||||
import {
|
||||
LOCAL_SERVER_NAME,
|
||||
reconcileVectorAgent,
|
||||
removeVectorAgent,
|
||||
withVectorTargetLock,
|
||||
} from "@dokploy/server/setup/vector-setup";
|
||||
import { TRPCError } from "@trpc/server";
|
||||
import { assertProvidersBelongToOrg, type VectorTarget } from "./service";
|
||||
|
||||
const assertLocalAgentAllowed = (serverId: string | undefined) => {
|
||||
if (!serverId && IS_CLOUD) {
|
||||
throw new TRPCError({
|
||||
code: "BAD_REQUEST",
|
||||
message:
|
||||
"The Vector agent on the Dokploy host is not available in Dokploy Cloud",
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
const assertServerAccessible = async (
|
||||
serverId: string,
|
||||
session: { userId: string; activeOrganizationId: string },
|
||||
) => {
|
||||
const server = await findServerById(serverId);
|
||||
const accessibleIds = await getAccessibleServerIds(session);
|
||||
if (
|
||||
server.organizationId !== session.activeOrganizationId ||
|
||||
!accessibleIds.has(serverId)
|
||||
) {
|
||||
throw new TRPCError({
|
||||
code: "UNAUTHORIZED",
|
||||
message: "You are not authorized to access this server",
|
||||
});
|
||||
}
|
||||
if (!server.sshKeyId) {
|
||||
throw new TRPCError({
|
||||
code: "BAD_REQUEST",
|
||||
message:
|
||||
"This server has no SSH key, so the Vector agent can't be managed on it",
|
||||
});
|
||||
}
|
||||
return server;
|
||||
};
|
||||
|
||||
export const applyVectorAgentSelection = async (
|
||||
session: { userId: string; activeOrganizationId: string },
|
||||
serverId: string | undefined,
|
||||
providerIds: string[],
|
||||
) => {
|
||||
const organizationId = session.activeOrganizationId;
|
||||
assertLocalAgentAllowed(serverId);
|
||||
const server = serverId
|
||||
? await assertServerAccessible(serverId, session)
|
||||
: undefined;
|
||||
if (providerIds.length > 0) {
|
||||
await assertProvidersBelongToOrg(providerIds, organizationId);
|
||||
}
|
||||
|
||||
await withVectorTargetLock(serverId, async () => {
|
||||
if (serverId) {
|
||||
await setServerProviderIds(serverId, providerIds);
|
||||
} else if (!(await setWebServerProviderIds(organizationId, providerIds))) {
|
||||
throw new TRPCError({
|
||||
code: "CONFLICT",
|
||||
message: "The local Vector agent is claimed by another organization",
|
||||
});
|
||||
}
|
||||
await reconcileVectorAgent(serverId);
|
||||
});
|
||||
return { serverName: server?.name ?? LOCAL_SERVER_NAME };
|
||||
};
|
||||
|
||||
// Re-applies the saved selection after a provider changed; a failure is reported, not thrown.
|
||||
export const reconcileVectorTargets = async (targets: VectorTarget[]) => {
|
||||
const failures: string[] = [];
|
||||
for (const target of targets) {
|
||||
if (target === null && IS_CLOUD) continue;
|
||||
try {
|
||||
await withVectorTargetLock(target ?? undefined, () =>
|
||||
reconcileVectorAgent(target ?? undefined),
|
||||
);
|
||||
} catch (error) {
|
||||
failures.push(
|
||||
`${target ?? LOCAL_SERVER_NAME}: ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
return failures.length > 0
|
||||
? `The Vector agent could not be re-applied on ${failures.join("; ")}`
|
||||
: undefined;
|
||||
};
|
||||
|
||||
// Even with an empty selection: Remove saves it before stopping the agent, so a failed Remove can leave one running.
|
||||
export const removeServerVectorAgent = async (server: Server) => {
|
||||
if (!server.sshKeyId) {
|
||||
return server.telemetryProviderIds.length > 0
|
||||
? "This server has no SSH key, so the Vector agent could not be removed from it"
|
||||
: undefined;
|
||||
}
|
||||
try {
|
||||
await withVectorTargetLock(server.serverId, () =>
|
||||
removeVectorAgent(server.serverId),
|
||||
);
|
||||
} catch (error) {
|
||||
console.error(
|
||||
`[Vector] Failed to remove agent for server ${server.serverId} before deletion:`,
|
||||
error,
|
||||
);
|
||||
return error instanceof Error ? error.message : String(error);
|
||||
}
|
||||
};
|
||||
@@ -0,0 +1,50 @@
|
||||
import { IS_CLOUD } from "@dokploy/server/constants";
|
||||
import { db } from "@dokploy/server/db";
|
||||
import { organization } from "@dokploy/server/db/schema";
|
||||
import { findServersByOrganizationForVectorAgent } from "@dokploy/server/services/server";
|
||||
import {
|
||||
getWebServerSettings,
|
||||
releaseWebServerAgent,
|
||||
} from "@dokploy/server/services/web-server-settings";
|
||||
import {
|
||||
removeVectorAgent,
|
||||
withVectorTargetLock,
|
||||
} from "@dokploy/server/setup/vector-setup";
|
||||
import { eq } from "drizzle-orm";
|
||||
|
||||
export const removeVectorAgents = async (organizationId: string) => {
|
||||
// Even with an empty selection: Remove saves it before stopping the agent, so a failed Remove can leave one running.
|
||||
const servers = await findServersByOrganizationForVectorAgent(organizationId);
|
||||
await Promise.all(
|
||||
servers.map((server) =>
|
||||
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,
|
||||
);
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
if (IS_CLOUD) return;
|
||||
await withVectorTargetLock(undefined, async () => {
|
||||
const settings = await getWebServerSettings();
|
||||
if (settings?.vectorAgentOrganizationId !== organizationId) return;
|
||||
await removeVectorAgent().catch((error) => {
|
||||
console.error(
|
||||
`[Vector] Failed to remove the local agent before deleting organization ${organizationId}:`,
|
||||
error,
|
||||
);
|
||||
});
|
||||
await releaseWebServerAgent(organizationId);
|
||||
});
|
||||
};
|
||||
|
||||
export const deleteOrganization = async (organizationId: string) => {
|
||||
await removeVectorAgents(organizationId);
|
||||
return await db
|
||||
.delete(organization)
|
||||
.where(eq(organization.id, organizationId));
|
||||
};
|
||||
@@ -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, eq } from "drizzle-orm";
|
||||
import { and, desc, eq, isNotNull } from "drizzle-orm";
|
||||
import type { z } from "zod";
|
||||
|
||||
export type Server = typeof server.$inferSelect;
|
||||
@@ -87,6 +87,34 @@ export const findServersByUserId = async (userId: string) => {
|
||||
return servers;
|
||||
};
|
||||
|
||||
export const findServersByOrganizationForVectorAgent = async (
|
||||
organizationId: string,
|
||||
) => {
|
||||
return await db.query.server.findMany({
|
||||
where: and(
|
||||
eq(server.organizationId, organizationId),
|
||||
isNotNull(server.sshKeyId),
|
||||
eq(server.serverType, "deploy"),
|
||||
),
|
||||
columns: {
|
||||
serverId: true,
|
||||
name: true,
|
||||
ipAddress: true,
|
||||
telemetryProviderIds: true,
|
||||
},
|
||||
orderBy: [desc(server.createdAt)],
|
||||
});
|
||||
};
|
||||
|
||||
export const setServerProviderIds = async (serverId: string, ids: string[]) => {
|
||||
const [updated] = await db
|
||||
.update(server)
|
||||
.set({ telemetryProviderIds: ids })
|
||||
.where(eq(server.serverId, serverId))
|
||||
.returning();
|
||||
return updated ?? null;
|
||||
};
|
||||
|
||||
export const deleteServer = async (serverId: string) => {
|
||||
const currentServer = await db
|
||||
.delete(server)
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { db } from "@dokploy/server/db";
|
||||
import { webServerSettings } from "@dokploy/server/db/schema";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { and, eq, isNull, or } from "drizzle-orm";
|
||||
|
||||
/**
|
||||
* Get the web server settings (singleton - only one row should exist)
|
||||
@@ -42,3 +42,44 @@ export const updateWebServerSettings = async (
|
||||
|
||||
return updated;
|
||||
};
|
||||
|
||||
export const setWebServerProviderIds = async (
|
||||
organizationId: string,
|
||||
ids: string[],
|
||||
) => {
|
||||
const current = await getWebServerSettings();
|
||||
if (!current) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const [updated] = await db
|
||||
.update(webServerSettings)
|
||||
.set({
|
||||
telemetryProviderIds: ids,
|
||||
vectorAgentOrganizationId: ids.length > 0 ? organizationId : null,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(webServerSettings.id, current.id),
|
||||
or(
|
||||
isNull(webServerSettings.vectorAgentOrganizationId),
|
||||
eq(webServerSettings.vectorAgentOrganizationId, organizationId),
|
||||
),
|
||||
),
|
||||
)
|
||||
.returning();
|
||||
|
||||
return updated ?? null;
|
||||
};
|
||||
|
||||
export const releaseWebServerAgent = async (organizationId: string) => {
|
||||
await db
|
||||
.update(webServerSettings)
|
||||
.set({
|
||||
telemetryProviderIds: [],
|
||||
vectorAgentOrganizationId: null,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(webServerSettings.vectorAgentOrganizationId, organizationId));
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user