GitHub/trunk-server
1
0
Fork 0
mirror of https://github.com/openmhz/trunk-server.git synced 2026-11-11 00:07:30 -05:00
Code Issues Packages Projects Releases Wiki Activity

Reduce backend per-upload CPU to relieve upload timeouts

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Luke Berndt
2026-11-09 18:06:48 -05:00
parent c0492ebf53
commit 6825aaf9a4
6 changed files with 220 additions and 290 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
backend/agents/otel-tracing.js
View File

@@ -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();

12
backend/controllers/calls.js
View File

@@ -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);
}

381
backend/controllers/uploads.js
View File

@@ -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,
});
});
};
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}`);
}
}
};

14
backend/index.js
View File

@@ -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;
}

17
nginx-proxy/prod/vhosts/site.template
View File

@@ -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;

17
nginx-proxy/test/vhosts/site.template
View File

@@ -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;