switch off otel plugin

Author: izadoesdev

apps/basket/package.json

@@ -18,11 +18,12 @@
     "@databuddy/shared": "workspace:*",
     "@databuddy/validation": "workspace:*",
     "@elysiajs/cors": "^1.3.3",
-    "@elysiajs/opentelemetry": "^1.4.6",
     "@elysiajs/server-timing": "^1.3.0",
     "@maxmind/geoip2-node": "^6.1.0",
+    "@opentelemetry/api": "^1.9.0",
     "@opentelemetry/exporter-trace-otlp-proto": "^0.208.0",
     "@opentelemetry/resources": "^2.2.0",
+    "@opentelemetry/sdk-node": "^0.208.0",
     "@opentelemetry/sdk-trace-node": "^2.2.0",
     "@opentelemetry/semantic-conventions": "^1.38.0",
     "@types/ua-parser-js": "^0.7.39",

apps/basket/src/hooks/auth.ts

@@ -7,8 +7,8 @@
 
 import { and, db, eq, member, websites } from "@databuddy/db";
 import { cacheable } from "@databuddy/redis";
-import { record, setAttributes } from "@elysiajs/opentelemetry";
 import { logger } from "../lib/logger";
+import { record, setAttributes } from "../lib/tracing";
 
 type Website = typeof websites.$inferSelect;
 
@@ -130,10 +130,7 @@ export function isValidOrigin(
 		return true;
 	}
 	if (!allowedDomain?.trim()) {
-		logger.warn(
-			{ originHeader },
-			"[isValidOrigin] No allowed domain provided"
-		);
+		logger.warn({ originHeader }, "[isValidOrigin] No allowed domain provided");
 		return false;
 	}
 	try {
@@ -339,7 +336,10 @@ const getWebsiteByIdWithOwnerCached = cacheable(
 			const ownerId = await _resolveOwnerId(website);
 			return { ...website, ownerId };
 		} catch (error) {
-			logger.error({ error, websiteId: id }, "Failed to get website by ID from cache");
+			logger.error(
+				{ error, websiteId: id },
+				"Failed to get website by ID from cache"
+			);
 			return null;
 		}
 	},
@@ -351,9 +351,7 @@ const getWebsiteByIdWithOwnerCached = cacheable(
 	}
 );
 
-export function getWebsiteByIdV2(
-	id: string
-): Promise<WebsiteWithOwner | null> {
+export function getWebsiteByIdV2(id: string): Promise<WebsiteWithOwner | null> {
 	return record("getWebsiteByIdV2", async () => {
 		setAttributes({
 			"website.id": id,

apps/basket/src/index.ts

@@ -1,14 +1,18 @@
 import "./polyfills/compression";
 
-import { opentelemetry } from "@elysiajs/opentelemetry";
-import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto";
-import { BatchSpanProcessor } from "@opentelemetry/sdk-trace-node";
 import { Elysia } from "elysia";
 import { logger } from "./lib/logger";
 import { getProducerStats } from "./lib/producer";
+import {
+	endRequestSpan,
+	initTracing,
+	startRequestSpan,
+} from "./lib/tracing";
 import basketRouter from "./routes/basket";
 import emailRouter from "./routes/email";
 
+initTracing();
+
 function getKafkaHealth() {
 	const stats = getProducerStats();
 
@@ -37,27 +41,9 @@ function getKafkaHealth() {
 }
 
 const app = new Elysia()
-	.use(
-		opentelemetry({
-			serviceName: "basket",
-			spanProcessors: [
-				new BatchSpanProcessor(
-					new OTLPTraceExporter({
-						url: "https://api.axiom.co/v1/traces",
-						headers: {
-							Authorization: `Bearer ${process.env.AXIOM_TOKEN}`,
-							"X-Axiom-Dataset": process.env.AXIOM_DATASET ?? "basket",
-						},
-					})
-				),
-			],
-		})
-	)
-	.onError(function handleError({ error, code }) {
-		if (code === "NOT_FOUND") {
-			return new Response(null, { status: 404 });
-		}
-		logger.error({ error }, "Error in basket service");
+	.state("tracing", {
+		span: null as ReturnType<typeof startRequestSpan> | null,
+		startTime: 0,
 	})
 	.onBeforeHandle(function handleCors({ request, set }) {
 		const origin = request.headers.get("origin");
@@ -71,6 +57,34 @@ const app = new Elysia()
 			set.headers["Access-Control-Allow-Credentials"] = "true";
 		}
 	})
+	.onBeforeHandle(function startTrace({ request, path, store }) {
+		const method = request.method;
+		const startTime = Date.now();
+		const span = startRequestSpan(method, request.url, path);
+
+		// Store span and start time in Elysia store
+		store.tracing = {
+			span,
+			startTime,
+		};
+	})
+	.onAfterHandle(function endTrace({ response, store }) {
+		if (store.tracing?.span && store.tracing.startTime) {
+			const statusCode = response instanceof Response ? response.status : 200;
+			endRequestSpan(store.tracing.span, statusCode, store.tracing.startTime);
+		}
+	})
+	.onError(function handleError({ error, code, store }) {
+		if (store.tracing?.span && store.tracing.startTime) {
+			const statusCode = code === "NOT_FOUND" ? 404 : 500;
+			endRequestSpan(store.tracing.span, statusCode, store.tracing.startTime);
+		}
+
+		if (code === "NOT_FOUND") {
+			return new Response(null, { status: 404 });
+		}
+		logger.error({ error }, "Error in basket service");
+	})
 	.options("*", () => new Response(null, { status: 204 }))
 	.use(basketRouter)
 	.use(emailRouter)

apps/basket/src/lib/event-service.ts

@@ -6,7 +6,6 @@ import type {
 	ErrorEvent,
 	WebVitalsEvent,
 } from "@databuddy/db";
-import { record, setAttributes } from "@elysiajs/opentelemetry";
 import { getGeo } from "../utils/ip-geo";
 import { parseUserAgent } from "../utils/user-agent";
 import {
@@ -18,6 +17,7 @@ import {
 import { logger } from "./logger";
 import { sendEvent, sendEventBatch } from "./producer";
 import { checkDuplicate, getDailySalt, saltAnonymousId } from "./security";
+import { record, setAttributes } from "./tracing";
 
 /**
  * Insert an error event into the database
@@ -70,7 +70,8 @@ export function insertError(
 				VALIDATION_LIMITS.SHORT_STRING_MAX_LENGTH
 			),
 			session_id: validateSessionId(payload.sessionId),
-			timestamp: typeof payload.timestamp === "number" ? payload.timestamp : now,
+			timestamp:
+				typeof payload.timestamp === "number" ? payload.timestamp : now,
 			path: sanitizeString(payload.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
 			message: sanitizeString(
 				payload.message,
@@ -108,10 +109,7 @@ export function insertError(
 		try {
 			sendEvent("analytics-errors", errorEvent);
 		} catch (error) {
-			logger.error(
-				{ error, eventId },
-				"Failed to queue error event"
-			);
+			logger.error({ error, eventId }, "Failed to queue error event");
 		}
 	});
 }
@@ -175,10 +173,7 @@ export async function insertWebVitals(
 	try {
 		sendEvent("analytics-web-vitals", webVitalsEvent);
 	} catch (error) {
-		logger.error(
-			{ error, eventId },
-			"Failed to queue web vitals event"
-		);
+		logger.error({ error, eventId }, "Failed to queue web vitals event");
 		// Don't throw - event is buffered or sent async
 	}
 }
@@ -229,10 +224,7 @@ export async function insertCustomEvent(
 	try {
 		sendEvent("analytics-custom-events", customEvent);
 	} catch (error) {
-		logger.error(
-			{ error, eventId },
-			"Failed to queue custom event"
-		);
+		logger.error({ error, eventId }, "Failed to queue custom event");
 		// Don't throw - event is buffered or sent async
 	}
 }
@@ -281,10 +273,7 @@ export async function insertOutgoingLink(
 	try {
 		sendEvent("analytics-outgoing-links", outgoingLinkEvent);
 	} catch (error) {
-		logger.error(
-			{ error, eventId },
-			"Failed to queue outgoing link event"
-		);
+		logger.error({ error, eventId }, "Failed to queue outgoing link event");
 	}
 }
 
@@ -370,7 +359,10 @@ export function insertTrackEvent(
 			),
 			url: sanitizeString(trackData.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
 			path: sanitizeString(trackData.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
-			title: sanitizeString(trackData.title, VALIDATION_LIMITS.STRING_MAX_LENGTH),
+			title: sanitizeString(
+				trackData.title,
+				VALIDATION_LIMITS.STRING_MAX_LENGTH
+			),
 
 			ip: anonymizedIP || "",
 			user_agent: "",
@@ -412,7 +404,9 @@ export function insertTrackEvent(
 			connection_time: validatePerformanceMetric(trackData.connection_time),
 			render_time: validatePerformanceMetric(trackData.render_time),
 			redirect_time: validatePerformanceMetric(trackData.redirect_time),
-			domain_lookup_time: validatePerformanceMetric(trackData.domain_lookup_time),
+			domain_lookup_time: validatePerformanceMetric(
+				trackData.domain_lookup_time
+			),
 
 			properties: trackData.properties
 				? JSON.stringify(trackData.properties)
@@ -431,10 +425,7 @@ export function insertTrackEvent(
 		try {
 			sendEvent("analytics-events", trackEvent);
 		} catch (error) {
-			logger.error(
-				{ error, eventId },
-				"Failed to queue track event"
-			);
+			logger.error({ error, eventId }, "Failed to queue track event");
 		}
 	});
 }
@@ -485,9 +476,7 @@ export function insertErrorsBatch(events: ErrorEvent[]): Promise<void> {
 	});
 }
 
-export function insertWebVitalsBatch(
-	events: WebVitalsEvent[]
-): Promise<void> {
+export function insertWebVitalsBatch(events: WebVitalsEvent[]): Promise<void> {
 	return record("insertWebVitalsBatch", async () => {
 		if (events.length === 0) {
 			return;
@@ -509,9 +498,7 @@ export function insertWebVitalsBatch(
 	});
 }
 
-export function insertCustomEventsBatch(
-	events: CustomEvent[]
-): Promise<void> {
+export function insertCustomEventsBatch(events: CustomEvent[]): Promise<void> {
 	return record("insertCustomEventsBatch", async () => {
 		if (events.length === 0) {
 			return;

apps/basket/src/lib/producer.ts

@@ -1,9 +1,9 @@
 import type { ClickHouseClient } from "@clickhouse/client";
 import { clickHouse, TABLE_NAMES } from "@databuddy/db";
-import { record, setAttributes } from "@elysiajs/opentelemetry";
 import { Semaphore } from "async-mutex";
 import { CompressionTypes, Kafka, type Producer } from "kafkajs";
 import { logger } from "./logger";
+import { record, setAttributes } from "./tracing";
 
 type BufferedEvent = {
 	table: string;
@@ -176,7 +176,10 @@ export class EventProducer {
 			this.lastRetry = Date.now();
 			this.stats.errors += 1;
 			this.stats.lastErrorTime = Date.now();
-			logger.error({ error }, "Redpanda connection failed, using ClickHouse fallback");
+			logger.error(
+				{ error },
+				"Redpanda connection failed, using ClickHouse fallback"
+			);
 			if (this.dependencies.onError) {
 				this.dependencies.onError(new Error(String(error)));
 			}
@@ -247,7 +250,7 @@ export class EventProducer {
 								logger.error(
 									{ error },
 									`Dropped event (retries: ${retries}, age: ${age}ms)`,
-									{ table, eventId: (event as { event_id?: string }).event_id },
+									{ table, eventId: (event as { event_id?: string }).event_id }
 								);
 							}
 						}
@@ -259,7 +262,10 @@ export class EventProducer {
 
 			const failures = results.filter((r) => r.status === "rejected");
 			if (failures.length > 0) {
-				logger.error({ failures: failures.length }, "Table flush operations failed");
+				logger.error(
+					{ failures: failures.length },
+					"Table flush operations failed"
+				);
 			}
 		} catch (error) {
 			this.stats.errors += 1;
@@ -300,7 +306,10 @@ export class EventProducer {
 
 		if (this.buffer.length >= this.config.bufferHardMax) {
 			this.stats.dropped += 1;
-			logger.error({ bufferLength: this.buffer.length }, "Buffer overflow, dropping event");
+			logger.error(
+				{ bufferLength: this.buffer.length },
+				"Buffer overflow, dropping event"
+			);
 			return;
 		}
 
@@ -363,7 +372,10 @@ export class EventProducer {
 						return;
 					} catch (error) {
 						this.stats.failed += 1;
-						logger.error({ error }, "Redpanda send failed, buffering to ClickHouse");
+						logger.error(
+							{ error },
+							"Redpanda send failed, buffering to ClickHouse"
+						);
 						this.failed = true;
 						setAttributes({
 							"kafka.send_failed": true,
@@ -457,7 +469,10 @@ export class EventProducer {
 						return;
 					} catch (error) {
 						this.stats.failed += events.length;
-						logger.error({ error }, "Redpanda batch failed, buffering to ClickHouse");
+						logger.error(
+							{ error },
+							"Redpanda batch failed, buffering to ClickHouse"
+						);
 						this.failed = true;
 						setAttributes({
 							"kafka.send_failed": true,
@@ -506,11 +521,7 @@ export class EventProducer {
 		await this.flush();
 
 		let finalFlushAttempts = 0;
-		while (
-			this.buffer.length > 0 &&
-			finalFlushAttempts < 3 &&
-			!this.flushing
-		) {
+		while (this.buffer.length > 0 && finalFlushAttempts < 3 && !this.flushing) {
 			finalFlushAttempts += 1;
 			await this.flush();
 			await new Promise((r) => setTimeout(r, 1000));
@@ -602,11 +613,15 @@ export const disconnectProducer = async (): Promise<void> => {
 export const getProducerStats = () => getDefaultProducer().getStats();
 
 process.on("SIGTERM", async () => {
-	await disconnectProducer().catch((error) => logger.error({ error }, "SIGTERM error"));
+	await disconnectProducer().catch((error) =>
+		logger.error({ error }, "SIGTERM error")
+	);
 	process.exit(0);
 });
 
 process.on("SIGINT", async () => {
-	await disconnectProducer().catch((error) => logger.error({ error }, "SIGINT error"));
+	await disconnectProducer().catch((error) =>
+		logger.error({ error }, "SIGINT error")
+	);
 	process.exit(0);
 });