Properly handle spans of image downloading (#1176)

.github/workflows/api-test.yml

@@ -37,3 +37,4 @@ jobs:
         run: bun test
         env:
           PGHOST: localhost
+          IMAGES_PATH: ./images

api/src/auth.ts

@@ -73,7 +73,7 @@ export const auth = new Elysia({ name: "auth" })
 	.macro({
 		permissions(perms: string[]) {
 			return {
-				beforeHandle: ({ jwt, status }) => {
+				beforeHandle: function permissionCheck({ jwt, status }) {
 					for (const perm of perms) {
 						if (!jwt!.permissions.includes(perm)) {
 							return status(403, {

api/src/controllers/seed/images.ts

@@ -1,4 +1,6 @@
 import path from "node:path";
+import { getCurrentSpan, record, setAttributes } from "@elysiajs/opentelemetry";
+import { SpanStatusCode } from "@opentelemetry/api";
 import { encode } from "blurhash";
 import { and, eq, is, lt, type SQL, sql } from "drizzle-orm";
 import { PgColumn, type PgTable } from "drizzle-orm/pg-core";
@@ -78,7 +80,34 @@ export const enqueueOptImage = async (
 };
 
 export const processImages = async () => {
-	async function processOne() {
+	return record("download images", async () => {
+		let running = false;
+		async function processAll() {
+			if (running) return;
+			running = true;
+
+			let found = true;
+			while (found) {
+				found = await processOne();
+			}
+			running = false;
+		}
+
+		const client = (await db.$client.connect()) as PoolClient;
+		client.on("notification", (evt) => {
+			if (evt.channel !== "kyoo_image") return;
+			processAll();
+		});
+		await client.query("listen kyoo_image");
+
+		// start processing old tasks
+		await processAll();
+		return () => client.release(true);
+	});
+};
+
+async function processOne() {
+	return record("download", async () => {
 		return await db.transaction(async (tx) => {
 			const [item] = await tx
 				.select()
@@ -91,6 +120,7 @@ export const processImages = async () => {
 			if (!item) return false;
 
 			const img = item.message as ImageTask;
+			setAttributes({ "item.url": img.url });
 			try {
 				const blurhash = await downloadImage(img.id, img.url);
 				const ret: Image = { id: img.id, source: img.url, blurhash };
@@ -105,6 +135,11 @@ export const processImages = async () => {
 
 				await tx.delete(mqueue).where(eq(mqueue.id, item.id));
 			} catch (err: any) {
+				const span = getCurrentSpan();
+				if (span) {
+					span.recordException(err);
+					span.setStatus({ code: SpanStatusCode.ERROR });
+				}
 				console.error("Failed to download image", img.url, err.message);
 				// don't use the transaction here, it can be aborted.
 				await db
@@ -114,31 +149,8 @@ export const processImages = async () => {
 			}
 			return true;
 		});
-	}
-
-	let running = false;
-	async function processAll() {
-		if (running) return;
-		running = true;
-
-		let found = true;
-		while (found) {
-			found = await processOne();
-		}
-		running = false;
-	}
-
-	const client = (await db.$client.connect()) as PoolClient;
-	client.on("notification", (evt) => {
-		if (evt.channel !== "kyoo_image") return;
-		processAll();
 	});
-	await client.query("listen kyoo_image");
+}
-
-	// start processing old tasks
-	await processAll();
-	return () => client.release(true);
-};
 
 async function downloadImage(id: string, url: string): Promise<string> {
 	const low = await getFile(path.join(imageDir, `${id}.low.jpg`))

chart/values.yaml

@@ -429,6 +429,8 @@ traefikproxy:
     extraArgs:
       - '--entryPoints.web.address=:80/tcp'
       - '--entryPoints.websecure.address=:443/tcp'
+      - '--entryPoints.web.forwardedHeaders.insecure=true'
+      - '--entryPoints.websecure.forwardedHeaders.insecure=true'
       - '--api.dashboard=true'
       - '--api.insecure=true'
       - '--log.level=INFO'

scanner/scanner/requests.py

@@ -101,6 +101,7 @@ class RequestProcessor:
 		finally:
 			self._processing = False
 
+	@tracer.start_as_current_span("process video")
 	async def process_request(self):
 		cur = await self._database.fetchrow(
 			"""
@@ -128,43 +129,44 @@ class RequestProcessor:
 			return False
 		request = Request.model_validate(cur)
 
-		with tracer.start_as_current_span(f"process {request.title}") as span:
+		span = trace.get_current_span()
-			logger.info(f"Starting to process {request.title}")
+		span.update_name(f"process {request.title}")
-			try:
+		logger.info(f"Starting to process {request.title}")
-				show = await self._run_request(request)
+		try:
-				finished = await self._database.fetchrow(
+			show = await self._run_request(request)
-					"""
+			finished = await self._database.fetchrow(
-					delete from scanner.requests
+				"""
-					where pk = $1
+				delete from scanner.requests
-					returning
+				where pk = $1
-						videos
+				returning
-					""",
+					videos
-					request.pk,
+				""",
+				request.pk,
+			)
+			if finished and finished["videos"] != request.videos:
+				videos = TypeAdapter(list[Request.Video]).validate_python(
+					finished["videos"]
 				)
-				if finished and finished["videos"] != request.videos:
+				await self._client.link_videos(
-					videos = TypeAdapter(list[Request.Video]).validate_python(
+					"movie" if request.kind == "movie" else "serie",
-						finished["videos"]
+					show.slug,
-					)
+					videos,
-					await self._client.link_videos(
-						"movie" if request.kind == "movie" else "serie",
-						show.slug,
-						videos,
-					)
-			except Exception as e:
-				span.set_status(trace.Status(trace.StatusCode.ERROR))
-				span.record_exception(e)
-				logger.error("Couldn't process request", exc_info=e)
-				cur = await self._database.execute(
-					"""
-					update
-						scanner.requests
-					set
-						status = 'failed'
-					where
-						pk = $1
-					""",
-					request.pk,
 				)
+		except Exception as e:
+			span.set_status(trace.Status(trace.StatusCode.ERROR))
+			span.record_exception(e)
+			logger.error("Couldn't process request", exc_info=e)
+			cur = await self._database.execute(
+				"""
+				update
+					scanner.requests
+				set
+					status = 'failed'
+				where
+					pk = $1
+				""",
+				request.pk,
+			)
 		return True
 
 	async def _run_request(self, request: Request) -> Resource: