From 6825aaf9a4bbd8151024c8f3426f047f378cd487 Mon Sep 17 00:00:00 2001 From: Luke Berndt Date: Mon, 9 Nov 2026 18:06:48 -0500 Subject: [PATCH] Reduce backend per-upload CPU to relieve upload timeouts The Node main thread was saturated at about 39 uploads/s, overflowing the listen queue so uploads timed out connecting from nginx. - Enable keepAlive on the S3 agent; every upload did a full TLS handshake - Remove OpenTelemetry tracing - Cache the system key lookup for 60s - Use lean queries when listing calls - Serialize each call once in notify_clients, not per recipient - Add an nginx upstream keepalive pool, and hold Node's keepAliveTimeout above it so nginx never reuses a connection Node has closed Co-Authored-By: Claude Opus 5.5 --- backend/agents/otel-tracing.js | 69 ----- backend/controllers/calls.js | 12 +- backend/controllers/uploads.js | 381 ++++++++++++-------------- backend/index.js | 14 +- nginx-proxy/prod/vhosts/site.template | 17 +- nginx-proxy/test/vhosts/site.template | 17 +- 6 files changed, 220 insertions(+), 290 deletions(-) delete mode 100644 backend/agents/otel-tracing.js diff --git a/backend/agents/otel-tracing.js b/backend/agents/otel-tracing.js deleted file mode 100644 index 1afb823..0000000 --- a/backend/agents/otel-tracing.js +++ /dev/null @@ -1,69 +0,0 @@ -'use strict'; - -const opentelemetry = require('@opentelemetry/sdk-node'); -const { MongoDBInstrumentation } = require('@opentelemetry/instrumentation-mongodb'); -const { TraceIdRatioBasedSampler } = require('@opentelemetry/sdk-trace-node'); -const { OTLPTraceExporter } = require('@opentelemetry/exporter-trace-otlp-http'); -const { getNodeAutoInstrumentations } = require('@opentelemetry/auto-instrumentations-node'); -const { diag, DiagConsoleLogger, DiagLogLevel } = require('@opentelemetry/api'); -const { - Detector, - DetectorSync, - IResource, - ResourceDetectionConfig, - envDetectorSync, - hostDetectorSync, - processDetectorSync, -} = require("@opentelemetry/resources") -// const { -// dockerCGroupV1Detector, -// } = require('@opentelemetry/resource-detector-docker'); - -// For troubleshooting, set the log level to DiagLogLevel.DEBUG -diag.setLogger(new DiagConsoleLogger(), DiagLogLevel.WARN); - -console.log("Starting OpenTelemetry Node SDK"); -console.log("OTEL_EXPORTER_OTLP_ENDPOINT: " + process.env.OTEL_EXPORTER_OTLP_ENDPOINT); -console.log("OTEL_EXPORTER_OTLP_HEADERS: " + process.env.OTEL_EXPORTER_OTLP_HEADERS); -console.log("OTEL_SERVICE_NAME: " + process.env.OTEL_SERVICE_NAME); - - -// from https://github.com/open-telemetry/opentelemetry-js/issues/4638 -// this is a workaround to wait for async attributes to be resolved -function awaitAttributes(detector){ - return { - async detect(config) { - const resource = detector.detect(config) - await resource.waitForAsyncAttributes() - - return resource - }, - } -} - -const samplePercentage = 0.05; - -const sdk = new opentelemetry.NodeSDK({ - traceExporter: new OTLPTraceExporter(), - sampler: new TraceIdRatioBasedSampler(samplePercentage), - instrumentations: [ - new MongoDBInstrumentation({ - // see under for available configuration - }), - getNodeAutoInstrumentations({ - // we recommend disabling fs autoinstrumentation since it can be noisy - // and expensive during startup - '@opentelemetry/instrumentation-fs': { - enabled: false, - }, - }), - ], - resourceDetectors: [ - awaitAttributes(envDetectorSync), - awaitAttributes(processDetectorSync), - awaitAttributes(hostDetectorSync), - ], - //resourceDetectors: [dockerCGroupV1Detector], -}); - -sdk.start(); \ No newline at end of file diff --git a/backend/controllers/calls.js b/backend/controllers/calls.js index a8fa53c..2cdd10b 100644 --- a/backend/controllers/calls.js +++ b/backend/controllers/calls.js @@ -2,7 +2,6 @@ const { ObjectId } = require('mongodb'); const Call = require("../models/call"); const Group = require("../models/group"); -const opentelemetry = require('@opentelemetry/api'); var defaultNumResults = 50; var channels = {}; @@ -17,11 +16,12 @@ const build_call_list = (items) => { url: item.url, filename: item.path + item.name, time: item.time, - srcList: item.srcList, - star: item.star, + // Lean results skip schema defaults, so older documents can lack these. + srcList: item.srcList ?? [], + star: item.star ?? 0, emergency: item.emergency, freq: item.freq, - patches: item.patches, + patches: item.patches ?? [], len: Math.round(item.len) }; calls.push(call); @@ -49,7 +49,7 @@ async function get_calls(query, numResults, middleDate, res) { const sort = { length: -1 }; try { - const items = await Call.find(query.filter, fields).sort(query.sort_order).limit(numResults); + const items = await Call.find(query.filter, fields).sort(query.sort_order).limit(numResults).lean(); const refined_items = build_call_list(items); calls.push(...refined_items); @@ -59,7 +59,7 @@ async function get_calls(query, numResults, middleDate, res) { $gt: middleDate }; - const items = await Call.find(query.filter, fields).sort(query.sort_order).limit(numResults); + const items = await Call.find(query.filter, fields).sort(query.sort_order).limit(numResults).lean(); const refined_items = build_call_list(items); calls.push(...refined_items); } diff --git a/backend/controllers/uploads.js b/backend/controllers/uploads.js index 919f343..d7f0a7a 100644 --- a/backend/controllers/uploads.js +++ b/backend/controllers/uploads.js @@ -1,18 +1,18 @@ const path = require('path'); const fs = require('fs'); -const opentelemetry = require('@opentelemetry/api'); const sysStats = require("../sys_stats"); const systemSchema = require("../models/systemSchema"); const talkgroupSchema = require("../models/talkgroupSchema"); const callSchema = require("../models/callSchema"); -const { trace, context } = opentelemetry; const mongoose = require("mongoose"); const { S3Client, PutObjectCommand } = require("@aws-sdk/client-s3"); const { fromIni } = require("@aws-sdk/credential-providers"); const { NodeHttpHandler } = require('@smithy/node-http-handler'); const https = require('https'); +// keepAlive matters: without it every upload does a full TCP + TLS handshake to S3. const agent = new https.Agent({ + keepAlive: true, maxSockets: 250, }); @@ -59,8 +59,27 @@ const client = new S3Client({ maxAttempts: 2, }); +// Every upload needs the system's key, so cache the lookup. A changed key or +// ignoreUnknownTalkgroup setting takes up to the TTL to apply. Only found systems +// are cached, so a newly created system works immediately. +const SYSTEM_CACHE_TTL_MS = 60 * 1000; +const systemCache = new Map(); + +async function getSystem(shortName) { + const cached = systemCache.get(shortName); + if (cached && cached.expires > Date.now()) { + return cached.system; + } + const system = await mongo_conn_fast.model("System").findOne({ shortName }, ["key", "ignoreUnknownTalkgroup"]).lean(); + if (system) { + systemCache.set(shortName, { system, expires: Date.now() + SYSTEM_CACHE_TTL_MS }); + } else { + systemCache.delete(shortName); + } + return system; +} + exports.upload = async function (req, res, next) { - const tracer = trace.getTracer("upload-service"); let start_time = Date.now(); let validateSystemTime = 0; let readFileTime = 0; @@ -68,217 +87,165 @@ exports.upload = async function (req, res, next) { let saveCallTime = 0; let cleanupTime = 0; let statsTime = 0; - return context.with(context.active(), async () => { - const parentSpan = trace.getActiveSpan(context.active()); - await tracer.startActiveSpan('upload_handler', { parent: parentSpan }, async (span) => { - try { - span.setAttribute('call.shortName', req.params.shortName.toLowerCase()); - span.setAttribute('call.talkgroup_num', req.body.talkgroup_num); - span.setAttribute('call.start_time', req.body.start_time); + try { + if (!req.file || !['.m4a', '.mp3'].includes(path.extname(req.file.originalname))) { + console.warn(`[${req.params.shortName}] Error file name is wrong or file does not exist`); + res.status(500).send("Error, invalid filename:\n"); + return; + } - if (!req.file || !['.m4a', '.mp3'].includes(path.extname(req.file.originalname))) { - console.warn(`[${req.params.shortName}] Error file name is wrong or file does not exist`); - span.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "Invalid file or filename", - }); - res.status(500).send("Error, invalid filename:\n"); - return; - } + const shortName = req.params.shortName.toLowerCase(); + const apiKey = req.body.api_key; + const talkgroupNum = parseInt(req.body.talkgroup_num); + const freq = parseFloat(req.body.freq); + const time = new Date(parseInt(req.body.start_time) * 1000); + let stopTime = new Date(); + if (req.body.stop_time) { + stopTime = new Date(parseInt(req.body.stop_time) * 1000); + } + const startTime = req.body.start_time; + const emergency = parseInt(req.body.emergency); - const shortName = req.params.shortName.toLowerCase(); - const apiKey = req.body.api_key; - const talkgroupNum = parseInt(req.body.talkgroup_num); - const freq = parseFloat(req.body.freq); - const time = new Date(parseInt(req.body.start_time) * 1000); - let stopTime = new Date(); - if (req.body.stop_time) { - stopTime = new Date(parseInt(req.body.stop_time) * 1000); - } - const startTime = req.body.start_time; - const emergency = parseInt(req.body.emergency); + let errorCount = parseInt(req.body.error_count) || 0; + let spikeCount = parseInt(req.body.spike_count) || 0; - let errorCount = parseInt(req.body.error_count) || 0; - let spikeCount = parseInt(req.body.spike_count) || 0; + let srcList = []; + try { + srcList = JSON.parse(req.body.source_list); + } catch (err) { + console.warn(`[${req.params.shortName}] Error /:shortName/upload Parsing Source/Freq List - Error: ${err}`); + res.status(500).send("Error parsing sourcelist " + err); + return; + } - let srcList = []; - try { - srcList = JSON.parse(req.body.source_list); - } catch (err) { - console.warn(`[${req.params.shortName}] Error /:shortName/upload Parsing Source/Freq List - Error: ${err}`); - res.status(500).send("Error parsing sourcelist " + err); - return; - } + patches = []; - patches = []; + let req_patches; + req_patches = req.body.patch_list; - let req_patches; - req_patches = req.body.patch_list; - - if(typeof req_patches != "undefined"){ - var split_patches = req_patches.replace("[","").replace("]","").split(","); - - for (var patch in split_patches){ - patches.push(split_patches[patch]); - } - } + if(typeof req_patches != "undefined"){ + var split_patches = req_patches.replace("[","").replace("]","").split(","); - let item = null; - - await tracer.startActiveSpan('validate_system', { parent: trace.getActiveSpan(context.active()) }, async (validateSpan) => { - try { - item = await mongo_conn_fast.model("System").findOne({ shortName }, ["key", "ignoreUnknownTalkgroup"]); - validateSpan.setAttribute('system.exists', !!item); - } catch (err) { - validateSpan.recordException(err); - validateSpan.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "Error validating system: " + err.message, - }); - } finally { - validateSpan.end(); - } - }); - - if (!item) { - console.warn(`[${req.params.shortName}] Error /:shortName/upload ShortName does not exist`); - res.status(500).send("ShortName does not exist: " + shortName + "\n"); - return; - } - if (apiKey !== item.key) { - console.warn(`[${req.params.shortName}] Error /:shortName/upload API Key Mismatch - Provided key: ${apiKey}`); - res.status(500).send("API Keys do not match!\n"); - return; - } - - if (shortName === "hennearmer" && [3421, 3423].includes(talkgroupNum)) { - res.status(200).end(); - return; - } - validateSystemTime = Date.now(); - if (item.ignoreUnknownTalkgroup) { - const talkgroupExists = await mongo_conn_fast.model("Talkgroup").exists({ - shortName, - num: talkgroupNum, - }); - - if (!talkgroupExists) { - try { - fs.unlinkSync(req.file.path); - } catch (err) { - console.error(`[${call.shortName}] error deleting: ${req.file.path}`); - } - res.status(500).send("Talkgroup does not exist, skipping.\n"); - return; - } - } - - res.status(200).end(); - - const local_path = `/${shortName}/${time.getFullYear()}/${time.getMonth() + 1}/${time.getDate()}/`; - const object_key = `media/${shortName}/${talkgroupNum}/${shortName}-${talkgroupNum}-${startTime}${path.extname(req.file.originalname)}`; - const url = `${s3_public_url}/${object_key}`; - - const call = new (mongo_conn_slow.model("Call"))({ - shortName, - talkgroupNum, - objectKey: object_key, - endpoint: s3_endpoint, - bucket: s3_bucket, - time, - name: `${talkgroupNum}-${startTime}.m4a`, - freq, - errorCount, - spikeCount, - url, - emergency, - path: local_path, - patches: patches, - srcList, - len: req.body.call_length ? parseFloat(req.body.call_length) : (stopTime - time) / 1000, - }); - - let fileContent; - - await tracer.startActiveSpan('upload_to_s3', { parent: trace.getActiveSpan(context.active()) }, async (uploadSpan) => { - try { - fileContent = fs.readFileSync(req.file.path); - readFileTime = Date.now(); - const command = new PutObjectCommand({ - Bucket: s3_bucket, - Key: object_key, - Body: fileContent, - ACL: 'public-read', - }); - var result = await client.send(command); - if (result && result.$metadata.httpStatusCode !== 200) { - console.error(`[${shortName}] Upload Error status code: ${result.$metadata.httpStatusCode}`); - console.error(result); - } - } catch (err) { - uploadSpan.recordException(err); - uploadSpan.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "Upload Error: " + err.message, - }); - console.warn(`[${call.shortName}] Upload Error: ${err}`); - } finally { - uploadSpan.end(); - } - }); - uploadFileTime = Date.now(); - await tracer.startActiveSpan('save_call', { parent: trace.getActiveSpan(context.active()) }, async (saveSpan) => { - try { - await call.save(); - } catch (err) { - saveSpan.recordException(err); - saveSpan.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "Error saving call: " + err.message, - }); - console.warn(`[${call.shortName}] Error saving call: ${err}`); - } finally { - saveSpan.end(); - } - }); - - saveCallTime = Date.now(); - sysStats.addCall(call.toObject()); - statsTime = Date.now(); - - if (call.len >= 1) { - req.call = call.toObject(); - next(); - } - - await tracer.startActiveSpan('cleanup_temp_file', { parent: trace.getActiveSpan(context.active()) }, async (cleanupSpan) => { - try { - fs.unlinkSync(req.file.path); - cleanupSpan.setAttribute('file.deleted', true); - } catch (err) { - cleanupSpan.recordException(err); - console.warn("There was an Error deleting: " + req.file.path); - } finally { - cleanupSpan.end(); - } - }); - cleanupTime = Date.now(); - //console.log(`[${call.shortName}] \t Verify System: ${validateSystemTime - start_time} \t Read file: ${readFileTime - validateSystemTime} \t Upload: ${ uploadFileTime - validateSystemTime} \t Save: ${saveCallTime - uploadFileTime} \tStats: ${statsTime - saveCallTime}\tCleanup: ${cleanupTime - statsTime} \t\t Total: ${cleanupTime - start_time}`); - } catch (error) { - console.error("Error processing call upload: " + error); - span.recordException(error); - span.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: error.message, - }); - } finally { - const totalTime = Date.now() - start_time; - if (totalTime > 10000) { - console.warn(`[${req.params.shortName}] Slow Upload - Size: ${call.len} Verify System: ${validateSystemTime - start_time} Read file: ${readFileTime - validateSystemTime} Upload: ${ uploadFileTime - validateSystemTime} Save: ${saveCallTime - uploadFileTime} Stats: ${statsTime - saveCallTime} Cleanup: ${cleanupTime - statsTime} Total: ${totalTime}`); - } - span.end(); + for (var patch in split_patches){ + patches.push(split_patches[patch]); } + } + let item = null; + + try { + item = await getSystem(shortName); + } catch (err) { + console.warn(`[${req.params.shortName}] Error validating system: ${err}`); + } + + if (!item) { + console.warn(`[${req.params.shortName}] Error /:shortName/upload ShortName does not exist`); + res.status(500).send("ShortName does not exist: " + shortName + "\n"); + return; + } + if (apiKey !== item.key) { + console.warn(`[${req.params.shortName}] Error /:shortName/upload API Key Mismatch - Provided key: ${apiKey}`); + res.status(500).send("API Keys do not match!\n"); + return; + } + + if (shortName === "hennearmer" && [3421, 3423].includes(talkgroupNum)) { + res.status(200).end(); + return; + } + validateSystemTime = Date.now(); + if (item.ignoreUnknownTalkgroup) { + const talkgroupExists = await mongo_conn_fast.model("Talkgroup").exists({ + shortName, + num: talkgroupNum, + }); + + if (!talkgroupExists) { + try { + fs.unlinkSync(req.file.path); + } catch (err) { + console.error(`[${call.shortName}] error deleting: ${req.file.path}`); + } + res.status(500).send("Talkgroup does not exist, skipping.\n"); + return; + } + } + + res.status(200).end(); + + const local_path = `/${shortName}/${time.getFullYear()}/${time.getMonth() + 1}/${time.getDate()}/`; + const object_key = `media/${shortName}/${talkgroupNum}/${shortName}-${talkgroupNum}-${startTime}${path.extname(req.file.originalname)}`; + const url = `${s3_public_url}/${object_key}`; + + const call = new (mongo_conn_slow.model("Call"))({ + shortName, + talkgroupNum, + objectKey: object_key, + endpoint: s3_endpoint, + bucket: s3_bucket, + time, + name: `${talkgroupNum}-${startTime}.m4a`, + freq, + errorCount, + spikeCount, + url, + emergency, + path: local_path, + patches: patches, + srcList, + len: req.body.call_length ? parseFloat(req.body.call_length) : (stopTime - time) / 1000, }); - }); -}; \ No newline at end of file + + let fileContent; + + try { + fileContent = fs.readFileSync(req.file.path); + readFileTime = Date.now(); + const command = new PutObjectCommand({ + Bucket: s3_bucket, + Key: object_key, + Body: fileContent, + ACL: 'public-read', + }); + var result = await client.send(command); + if (result && result.$metadata.httpStatusCode !== 200) { + console.error(`[${shortName}] Upload Error status code: ${result.$metadata.httpStatusCode}`); + console.error(result); + } + } catch (err) { + console.warn(`[${call.shortName}] Upload Error: ${err}`); + } + uploadFileTime = Date.now(); + + try { + await call.save(); + } catch (err) { + console.warn(`[${call.shortName}] Error saving call: ${err}`); + } + + saveCallTime = Date.now(); + sysStats.addCall(call.toObject()); + statsTime = Date.now(); + + if (call.len >= 1) { + req.call = call.toObject(); + next(); + } + + try { + fs.unlinkSync(req.file.path); + } catch (err) { + console.warn("There was an Error deleting: " + req.file.path); + } + cleanupTime = Date.now(); + //console.log(`[${call.shortName}] \t Verify System: ${validateSystemTime - start_time} \t Read file: ${readFileTime - validateSystemTime} \t Upload: ${ uploadFileTime - validateSystemTime} \t Save: ${saveCallTime - uploadFileTime} \tStats: ${statsTime - saveCallTime}\tCleanup: ${cleanupTime - statsTime} \t\t Total: ${cleanupTime - start_time}`); + } catch (error) { + console.error("Error processing call upload: " + error); + } finally { + const totalTime = Date.now() - start_time; + if (totalTime > 10000) { + console.warn(`[${req.params.shortName}] Slow Upload - Size: ${call.len} Verify System: ${validateSystemTime - start_time} Read file: ${readFileTime - validateSystemTime} Upload: ${ uploadFileTime - validateSystemTime} Save: ${saveCallTime - uploadFileTime} Stats: ${statsTime - saveCallTime} Cleanup: ${cleanupTime - statsTime} Total: ${totalTime}`); + } + } +}; diff --git a/backend/index.js b/backend/index.js index 36f63f2..bd08608 100644 --- a/backend/index.js +++ b/backend/index.js @@ -1,4 +1,3 @@ -const tracing = require("./agents/otel-tracing"); var express = require("express"); var configureExpress = require("./config/express"); @@ -25,6 +24,12 @@ var app = express(); const server = require('http').createServer(app); +// Must outlast nginx's upstream keepalive_timeout (60s), so nginx is always the +// side that closes an idle connection. Otherwise it can send a request on a +// connection Node has just closed, and nginx won't retry a POST. +server.keepAliveTimeout = 65000; +server.headersTimeout = 70000; + const io = require('socket.io')(server, { cors: { @@ -201,6 +206,7 @@ app.use(function (err, req, res, next) { function notify_clients(call) { call.type = "calls"; + const message = JSON.stringify(call); var sent = 0; for (var key in clients) { @@ -211,7 +217,7 @@ function notify_clients(call) { // if client if ((client.firehose == true) || (client.filterType == "all")) { sent++; - client.socket.emit("new message", JSON.stringify(call)); + client.socket.emit("new message", message); } else if (client.filterType == "unit") { var codeArray = client.filterCode.split(','); var success = false; @@ -219,7 +225,7 @@ function notify_clients(call) { for (var k = 0; k < call.srcList.length; k++) { if (codeArray[j] == call.srcList[k]) { sent++; - client.socket.emit("new message", JSON.stringify(call)); + client.socket.emit("new message", message); success = true; break; } @@ -232,7 +238,7 @@ function notify_clients(call) { var codeArray = client.talkgroupNums; for (var j = 0; j < codeArray.length; ++j) { if (codeArray[j] == call.talkgroupNum) { - client.socket.emit("new message", JSON.stringify(call)); + client.socket.emit("new message", message); sent++ break; } diff --git a/nginx-proxy/prod/vhosts/site.template b/nginx-proxy/prod/vhosts/site.template index 30b2384..b72c9c4 100644 --- a/nginx-proxy/prod/vhosts/site.template +++ b/nginx-proxy/prod/vhosts/site.template @@ -1,5 +1,18 @@ limit_req_zone $binary_remote_addr zone=perip:10m rate=1r/s; +# Only send "Connection: upgrade" for real WebSocket requests, so ordinary +# requests can reuse the upstream keepalive pool. +map $http_upgrade $backend_connection { + default upgrade; + '' ''; +} + +upstream backend_api { + server backend:3005; + keepalive 64; + keepalive_timeout 60s; +} + server { listen 80; server_name account.${DOMAIN_NAME} api.${DOMAIN_NAME} admin.${DOMAIN_NAME} www.${DOMAIN_NAME} ${DOMAIN_NAME}; @@ -87,10 +100,10 @@ server { location / { - proxy_pass http://backend:3005; + proxy_pass http://backend_api; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; - proxy_set_header Connection "upgrade"; + proxy_set_header Connection $backend_connection; proxy_set_header Host $host; proxy_cache_bypass $http_upgrade; include /etc/nginx/proxy.conf; diff --git a/nginx-proxy/test/vhosts/site.template b/nginx-proxy/test/vhosts/site.template index e4d96d7..ce08448 100644 --- a/nginx-proxy/test/vhosts/site.template +++ b/nginx-proxy/test/vhosts/site.template @@ -1,4 +1,17 @@ +# Only send "Connection: upgrade" for real WebSocket requests, so ordinary +# requests can reuse the upstream keepalive pool. +map $http_upgrade $backend_connection { + default upgrade; + '' ''; +} + +upstream backend_api { + server backend:3005; + keepalive 64; + keepalive_timeout 60s; +} + server { listen 80; @@ -89,11 +102,11 @@ server { } location / { - proxy_pass http://backend:3005; + proxy_pass http://backend_api; include /etc/nginx/proxy.conf; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; - proxy_set_header Connection "upgrade"; + proxy_set_header Connection $backend_connection; proxy_set_header Host $host; proxy_cache_bypass $http_upgrade; access_log /dev/stdout main;