From bbbd4e18195402afa2ee34a306d4cabc4efffafe Mon Sep 17 00:00:00 2001 From: Luke Berndt Date: Sun, 18 Aug 2024 21:09:03 -0400 Subject: [PATCH 1/8] better tracing --- backend/controllers/uploads.js | 471 ++++++++++++++++++--------------- 1 file changed, 257 insertions(+), 214 deletions(-) diff --git a/backend/controllers/uploads.js b/backend/controllers/uploads.js index cdd5e1a..c4793d5 100644 --- a/backend/controllers/uploads.js +++ b/backend/controllers/uploads.js @@ -5,7 +5,8 @@ var sysStats = require("../sys_stats"); var System = require("../models/system"); var Talkgroup = require("../models/talkgroup"); var { callModel: Call } = require("../models/call"); - +const opentelemetry = require('@opentelemetry/api'); +const { trace } = opentelemetry; const { PutObjectCommand, S3Client } = require("@aws-sdk/client-s3"); const { fromIni } = require("@aws-sdk/credential-providers"); @@ -35,232 +36,274 @@ const client = new S3Client({ }); exports.upload = async function (req, res, next) { - - process.nextTick(async () => { - if (!req.file || ((path.extname(req.file.originalname) != '.m4a') && (path.extname(req.file.originalname) != '.mp3'))) { - console.warn("[" + req.params.shortName + "] Error file name is wrong or file does not exist"); - res.status(500); - res.send("Error, invalid filename:\n"); - return; - } - var shortName = req.params.shortName.toLowerCase(); - var apiKey = req.body.api_key; - var talkgroupNum = parseInt(req.body.talkgroup_num); - var freq = parseFloat(req.body.freq); - var time = new Date(parseInt(req.body.start_time) * 1000); - var stopTime = new Date(); - if (req.body.stop_time) { - var stopTime = new Date(parseInt(req.body.stop_time) * 1000); - } - var startTime = req.body.start_time; - var emergency = parseInt(req.body.emergency); - - let errorCount = parseInt(req.body.error_count); - let spikeCount = parseInt(req.body.spike_count); - - if (Number.isNaN(errorCount)) { - errorCount = 0; - } - if (Number.isNaN(spikeCount)) { - spikeCount = 0; - } - + const parentSpan = trace.getActiveSpan(); + await trace.getTracer('upload-service').startActiveSpan('upload_handler', async (span) => { try { - var srcList = JSON.parse(req.body.source_list); - } catch (err) { - var srcList = []; - console.warn("[" + req.params.shortName + "] Error /:shortName/upload Parsing Source/Freq List - Error: " + err); - res.status(500); - res.send("Error parsing sourcelist " + err); - return; - } + 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); - let item = null; - const span = opentelemetry.trace.getActiveSpan(); - span.setAttribute('call.shortName', shortName); - span.setAttribute('call.talkgroupNum', talkgroupNum); - span.setAttribute('call.startTime', startTime); - - - - // Validate that the system exists and the API key is correct - - try { - item = await System.findOne({ 'shortName': shortName }, ["key", "ignoreUnknownTalkgroup"]); - } catch (err) { - span.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "Error /:shortName/upload - Error: " + err, - }); - console.warn("[" + req.params.shortName + "] Error /:shortName/upload - Error: " + err); - res.status(500); - res.send("Error, invalid shortName:\n" + err); - return; - } - - if (!item) { - span.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "Error /:shortName/upload API Key Mismatch - Provided key: " + apiKey, - }); - console.info("[" + req.params.shortName + "] Error /:shortName/upload ShortName does not exist"); - res.status(500); - res.send("ShortName does not exist: " + shortName + "\n"); - return; - } - if (apiKey != item.key) { - span.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "Error /:shortName/upload API Key Mismatch - Provided key: " + apiKey, - }); - console.warn("[" + req.params.shortName + "] Error /:shortName/upload API Key Mismatch - Provided key: " + apiKey); - res.status(500); - res.send("API Keys do not match!\n"); - return; - } - - // Blocking sensitive talkgroups - if ((shortName == "hennearmer") && ((talkgroupNum == 3421) || (talkgroupNum == 3423))) { - span.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "Sensitive Talkgroup", - }); - res.status(200).end(); - return; - } - - if (item.ignoreUnknownTalkgroup == true) { - talkgroupExists = await Talkgroup.exists({ - 'shortName': shortName, - 'num': talkgroupNum - }); - - if (!talkgroupExists) { - try { - fs.unlinkSync(req.file.path) - //file removed - } catch (err) { - console.log("[" + call.shortName + "] error deleting: " + req.file.path); - } - - res.status(500); - res.send("Talkgroup does not exist, skipping.\n"); + if (!req.file || ((path.extname(req.file.originalname) != '.m4a') && (path.extname(req.file.originalname) != '.mp3'))) { + console.warn("[" + req.params.shortName + "] Error file name is wrong or file does not exist"); span.setStatus({ code: opentelemetry.SpanStatusCode.ERROR, - message: "Ignore Unknown Talkgroup Set - Talkgroup does not exist: " + talkgroupNum + " ShortName: " + shortName, + message: "Invalid file or filename", }); - + res.status(500); + res.send("Error, invalid filename:\n"); return; } - } - // Add in an API Key check + + // Extract and parse request data + await trace.getTracer('upload-service').startActiveSpan('parse_request_data', async (parseSpan) => { + + var shortName = req.params.shortName.toLowerCase(); + var apiKey = req.body.api_key; + var talkgroupNum = parseInt(req.body.talkgroup_num); + var freq = parseFloat(req.body.freq); + var time = new Date(parseInt(req.body.start_time) * 1000); + var stopTime = new Date(); + if (req.body.stop_time) { + var stopTime = new Date(parseInt(req.body.stop_time) * 1000); + } + var startTime = req.body.start_time; + var emergency = parseInt(req.body.emergency); + + let errorCount = parseInt(req.body.error_count); + let spikeCount = parseInt(req.body.spike_count); + + if (Number.isNaN(errorCount)) { + errorCount = 0; + } + if (Number.isNaN(spikeCount)) { + spikeCount = 0; + } + + try { + var srcList = JSON.parse(req.body.source_list); + } catch (err) { + var srcList = []; + console.warn("[" + req.params.shortName + "] Error /:shortName/upload Parsing Source/Freq List - Error: " + err); + res.status(500); + res.send("Error parsing sourcelist " + err); + return; + } + parseSpan.end(); + }); + + let item = null; + + + // Validate that the system exists and the API key is correct + + await trace.getTracer('upload-service').startActiveSpan('validate_system', async (validateSpan) => { + + try { + item = await System.findOne({ 'shortName': shortName }, ["key", "ignoreUnknownTalkgroup"]); + validateSpan.setAttribute('system.exists', !!item); + + + if (!item) { + validateSpan.setStatus({ + code: opentelemetry.SpanStatusCode.ERROR, + message: "ShortName does not exist", + }); + + console.info("[" + req.params.shortName + "] Error /:shortName/upload ShortName does not exist"); + res.status(500); + res.send("ShortName does not exist: " + shortName + "\n"); + return; + } + if (apiKey != item.key) { + validateSpan.setStatus({ + code: opentelemetry.SpanStatusCode.ERROR, + message: "API Key Mismatch", + }); + console.warn("[" + req.params.shortName + "] Error /:shortName/upload API Key Mismatch - Provided key: " + apiKey); + res.status(500); + res.send("API Keys do not match!\n"); + return; + } + } catch (err) { + validateSpan.recordException(err); + validateSpan.setStatus({ + code: opentelemetry.SpanStatusCode.ERROR, + message: "Error validating system: " + err.message, + }); + res.status(500).send("Error, invalid shortName:\n" + err); + return; + } finally { + validateSpan.end(); + } + }); + + // Blocking sensitive talkgroups + await trace.getTracer('upload-service').startActiveSpan('check_sensitive_talkgroups', async (sensitiveSpan) => { + + if ((shortName == "hennearmer") && ((talkgroupNum == 3421) || (talkgroupNum == 3423))) { + sensitiveSpan.setStatus({ + code: opentelemetry.SpanStatusCode.ERROR, + message: "Sensitive Talkgroup", + }); + res.status(200).end(); + return; + } + sensitiveSpan.end(); + }); + + if (item.ignoreUnknownTalkgroup == true) { + await trace.getTracer('upload-service').startActiveSpan('check_talkgroup_exists', async (talkgroupSpan) => { + + talkgroupExists = await Talkgroup.exists({ + 'shortName': shortName, + 'num': talkgroupNum + }); + + if (!talkgroupExists) { + try { + fs.unlinkSync(req.file.path) + //file removed + } catch (err) { + console.log("[" + call.shortName + "] error deleting: " + req.file.path); + } + + res.status(500); + res.send("Talkgroup does not exist, skipping.\n"); - res.status(200).end(); - - - var local_path = "/" + shortName + "/" + time.getFullYear() + "/" + (time.getMonth() + 1) + "/" + time.getDate() + "/"; - var object_key = "media/" + shortName + "-" + talkgroupNum + "-" + startTime + path.extname(req.file.originalname); - - var endpoint = s3_endpoint; - var bucket = s3_bucket; - var url = s3_endpoint + "/" + s3_bucket + "/" + object_key; - - var call = new Call({ - shortName: shortName, - talkgroupNum: talkgroupNum, - objectKey: object_key, - endpoint: endpoint, - bucket: bucket, - time: time, - name: talkgroupNum + "-" + startTime + ".m4a", - freq: freq, - errorCount: errorCount, - spikeCount: spikeCount, - url: url, - emergency: emergency, - path: local_path, - srcList: srcList, - len: -1 - }); - - - try { - if (req.body.call_length) { - call.len = parseFloat(req.body.call_length); - } else { - call.len = (stopTime - time) / 1000; + return; + } + talkgroupSpan.end(); + }); } - } catch (err) { - console.error(err); - } - let fileContent; - try { - fileContent = fs.readFileSync(req.file.path); - //s3Src = fs.createReadStream(req.file.path); - } catch (err) { + // Add in an API Key check + + + + res.status(200).end(); + + // Prepare call object + let call; + await trace.getTracer('upload-service').startActiveSpan('prepare_call_object', async (prepareSpan) => { + + var local_path = "/" + shortName + "/" + time.getFullYear() + "/" + (time.getMonth() + 1) + "/" + time.getDate() + "/"; + var object_key = "media/" + shortName + "-" + talkgroupNum + "-" + startTime + path.extname(req.file.originalname); + + var endpoint = s3_endpoint; + var bucket = s3_bucket; + var url = s3_endpoint + "/" + s3_bucket + "/" + object_key; + + call = new Call({ + shortName: shortName, + talkgroupNum: talkgroupNum, + objectKey: object_key, + endpoint: endpoint, + bucket: bucket, + time: time, + name: talkgroupNum + "-" + startTime + ".m4a", + freq: freq, + errorCount: errorCount, + spikeCount: spikeCount, + url: url, + emergency: emergency, + path: local_path, + srcList: srcList, + len: -1 + }); + + + try { + if (req.body.call_length) { + call.len = parseFloat(req.body.call_length); + } else { + call.len = (stopTime - time) / 1000; + } + } catch (err) { + console.error(err); + } + prepareSpan.setAttribute('call.object_key', object_key); + prepareSpan.setAttribute('call.url', url); + prepareSpan.end(); + }); + + let fileContent; + // Upload file to S3 + await trace.getTracer('upload-service').startActiveSpan('upload_to_s3', async (uploadSpan) => { + + try { + fileContent = fs.readFileSync(req.file.path); + //s3Src = fs.createReadStream(req.file.path); + + const command = new PutObjectCommand({ + Bucket: s3_bucket, + Key: object_key, + Body: fileContent, + ACL: 'public-read' + }); + + + await client.send(command); + + + } 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(); + } + }); + + // Save call to database + await trace.getTracer('upload-service').startActiveSpan('save_call', async (saveSpan) => { + try { + await call.save(); + sysStats.addCall(call.toObject()); + // we only want to notify clients if the clip is longer than 1 second. + if (call.len >= 1) { + req.call = call.toObject(); + next(); + } + + fs.unlinkSync(req.file.path); + + + } 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(); + } + }); + // Clean up temporary file + await trace.getTracer('upload-service').startActiveSpan('cleanup_temp_file', 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(); + } + }); + + } catch (error) { + span.recordException(error); span.setStatus({ code: opentelemetry.SpanStatusCode.ERROR, - message: "Unable to open file: " + req.file.path + " Error: " + err, + message: error.message, }); - console.error("[" + call.shortName + "] Unable to open file: " + req.file.path + " Error: " + err); - return; + } finally { + span.end(); } - const command = new PutObjectCommand({ - Bucket: s3_bucket, - Key: object_key, - Body: fileContent, - ACL: 'public-read' - }); - - try { - await client.send(command); - - } catch (err) { - span.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "Upload Error: " + err + " Filename: " + call.name + " content-length: " + req.headers['content-length'] + " Key: " + object_key + " Bucket: " + s3_bucket, - }); - console.warn("[" + call.shortName + "] Upload Error: " + err + " Filename: " + call.name + " content-length: " + req.headers['content-length'] + " Key: " + object_key + " Bucket: " + s3_bucket); - /*res.status(500); - sysStats.addError(call.toObject()); - res.contentType('json'); - res.status(500); - res.send(JSON.stringify({ - success: false, - error: "File Upload" - }));*/ - } - - try { - await call.save(); - sysStats.addCall(call.toObject()); - // we only want to notify clients if the clip is longer than 1 second. - if (call.len >= 1) { - req.call = call.toObject(); - next(); - } - } catch (err) { - console.warn("[" + call.shortName + "] Error saving call: " + err); - span.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "Error saving call: " + err, - }); - /*res.status(500); - sysStats.addError(call.toObject()); - res.contentType('json'); - res.status(500); - res.send(JSON.stringify({ - success: false, - error: "File Upload" - }));*/ - } - - try { - fs.unlinkSync(req.file.path); - } catch (err) { - console.warn("There was an Error uploading an deleting: " + req.file.path); - }; }); -} +}; From bf7e09d4e9bd361bbd588c71b681421587b3430b Mon Sep 17 00:00:00 2001 From: Luke Berndt Date: Tue, 20 Aug 2024 11:29:48 -0400 Subject: [PATCH 2/8] Update uploads.js --- backend/controllers/uploads.js | 1 - 1 file changed, 1 deletion(-) diff --git a/backend/controllers/uploads.js b/backend/controllers/uploads.js index 14c1b95..acd9e1c 100644 --- a/backend/controllers/uploads.js +++ b/backend/controllers/uploads.js @@ -5,7 +5,6 @@ var sysStats = require("../sys_stats"); var System = require("../models/system"); var Talkgroup = require("../models/talkgroup"); var { callModel: Call } = require("../models/call"); -const opentelemetry = require('@opentelemetry/api'); const { trace } = opentelemetry; const { PutObjectCommand, S3Client } = require("@aws-sdk/client-s3"); From 2e4205197316851f99b7487c533f68b70a266729 Mon Sep 17 00:00:00 2001 From: Luke Berndt Date: Tue, 20 Aug 2024 11:36:35 -0400 Subject: [PATCH 3/8] Update uploads.js --- backend/controllers/uploads.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/controllers/uploads.js b/backend/controllers/uploads.js index acd9e1c..30c2c53 100644 --- a/backend/controllers/uploads.js +++ b/backend/controllers/uploads.js @@ -54,7 +54,7 @@ exports.upload = async function (req, res, next) { } // Extract and parse request data - await trace.getTracer('upload-service').startActiveSpan('parse_request_data', async (parseSpan) => { + var shortName = req.params.shortName.toLowerCase(); var apiKey = req.body.api_key; @@ -88,7 +88,7 @@ exports.upload = async function (req, res, next) { return; } parseSpan.end(); - }); + let item = null; From 3ea2613ba9dd1309fea15af75c43f602423b0b1c Mon Sep 17 00:00:00 2001 From: Luke Berndt Date: Tue, 20 Aug 2024 11:40:57 -0400 Subject: [PATCH 4/8] Update uploads.js --- backend/controllers/uploads.js | 58 +++++++++++++++++----------------- 1 file changed, 29 insertions(+), 29 deletions(-) diff --git a/backend/controllers/uploads.js b/backend/controllers/uploads.js index 30c2c53..3c2c83d 100644 --- a/backend/controllers/uploads.js +++ b/backend/controllers/uploads.js @@ -56,38 +56,38 @@ exports.upload = async function (req, res, next) { // Extract and parse request data - var shortName = req.params.shortName.toLowerCase(); - var apiKey = req.body.api_key; - var talkgroupNum = parseInt(req.body.talkgroup_num); - var freq = parseFloat(req.body.freq); - var time = new Date(parseInt(req.body.start_time) * 1000); - var stopTime = new Date(); - if (req.body.stop_time) { - var stopTime = new Date(parseInt(req.body.stop_time) * 1000); - } - var startTime = req.body.start_time; - var emergency = parseInt(req.body.emergency); + var shortName = req.params.shortName.toLowerCase(); + var apiKey = req.body.api_key; + var talkgroupNum = parseInt(req.body.talkgroup_num); + var freq = parseFloat(req.body.freq); + var time = new Date(parseInt(req.body.start_time) * 1000); + var stopTime = new Date(); + if (req.body.stop_time) { + var stopTime = new Date(parseInt(req.body.stop_time) * 1000); + } + var startTime = req.body.start_time; + var emergency = parseInt(req.body.emergency); - let errorCount = parseInt(req.body.error_count); - let spikeCount = parseInt(req.body.spike_count); + let errorCount = parseInt(req.body.error_count); + let spikeCount = parseInt(req.body.spike_count); - if (Number.isNaN(errorCount)) { - errorCount = 0; - } - if (Number.isNaN(spikeCount)) { - spikeCount = 0; - } + if (Number.isNaN(errorCount)) { + errorCount = 0; + } + if (Number.isNaN(spikeCount)) { + spikeCount = 0; + } - try { - var srcList = JSON.parse(req.body.source_list); - } catch (err) { - var srcList = []; - console.warn("[" + req.params.shortName + "] Error /:shortName/upload Parsing Source/Freq List - Error: " + err); - res.status(500); - res.send("Error parsing sourcelist " + err); - return; - } - parseSpan.end(); + try { + var srcList = JSON.parse(req.body.source_list); + } catch (err) { + var srcList = []; + console.warn("[" + req.params.shortName + "] Error /:shortName/upload Parsing Source/Freq List - Error: " + err); + res.status(500); + res.send("Error parsing sourcelist " + err); + return; + } + parseSpan.end(); let item = null; From fc77bd2934d7a8147580e3eaab89db969557256b Mon Sep 17 00:00:00 2001 From: Luke Berndt Date: Wed, 21 Aug 2024 10:41:41 -0400 Subject: [PATCH 5/8] upgrading telem --- backend/agents/otel-tracing.js | 24 +++++++++++++++++++++++ backend/controllers/uploads.js | 35 +++++++++++++++++++++------------- backend/index.js | 2 +- backend/package-lock.json | 22 ++++++++++----------- backend/package.json | 2 +- docker-compose.yml | 4 ++++ 6 files changed, 63 insertions(+), 26 deletions(-) create mode 100644 backend/agents/otel-tracing.js diff --git a/backend/agents/otel-tracing.js b/backend/agents/otel-tracing.js new file mode 100644 index 0000000..86a9f89 --- /dev/null +++ b/backend/agents/otel-tracing.js @@ -0,0 +1,24 @@ +'use strict'; + +const opentelemetry = require('@opentelemetry/sdk-node'); +const { TraceIdRatioBasedSampler } = require('@opentelemetry/sdk-trace-node'); +const { OTLPTraceExporter } = require('@opentelemetry/exporter-trace-otlp-http'); +const { getNodeAutoInstrumentations } = require('@opentelemetry/auto-instrumentations-node'); + +const samplePercentage = 0.05; + +const sdk = new opentelemetry.NodeSDK({ + traceExporter: new OTLPTraceExporter(), + sampler: new TraceIdRatioBasedSampler(samplePercentage), + instrumentations: [ + getNodeAutoInstrumentations({ + // we recommend disabling fs autoinstrumentation since it can be noisy + // and expensive during startup + '@opentelemetry/instrumentation-fs': { + enabled: false, + }, + }), + ], +}); + +sdk.start(); \ No newline at end of file diff --git a/backend/controllers/uploads.js b/backend/controllers/uploads.js index 3c2c83d..23c7282 100644 --- a/backend/controllers/uploads.js +++ b/backend/controllers/uploads.js @@ -36,7 +36,7 @@ const client = new S3Client({ exports.upload = async function (req, res, next) { const parentSpan = trace.getActiveSpan(); - await trace.getTracer('upload-service').startActiveSpan('upload_handler', async (span) => { + await trace.getTracer('upload-service').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); @@ -93,9 +93,9 @@ exports.upload = async function (req, res, next) { let item = null; + console.log("Finished parsing request data"); // Validate that the system exists and the API key is correct - - await trace.getTracer('upload-service').startActiveSpan('validate_system', async (validateSpan) => { + await trace.getTracer('upload-service').startActiveSpan('validate_system', { parent: span }, async (validateSpan) => { try { item = await System.findOne({ 'shortName': shortName }, ["key", "ignoreUnknownTalkgroup"]); @@ -136,22 +136,25 @@ exports.upload = async function (req, res, next) { } }); + console.log("Finished validating system API key"); // Blocking sensitive talkgroups - await trace.getTracer('upload-service').startActiveSpan('check_sensitive_talkgroups', async (sensitiveSpan) => { + await trace.getTracer('upload-service').startActiveSpan('check_sensitive_talkgroups', { parent: span }, async (sensitiveSpan) => { if ((shortName == "hennearmer") && ((talkgroupNum == 3421) || (talkgroupNum == 3423))) { sensitiveSpan.setStatus({ code: opentelemetry.SpanStatusCode.ERROR, message: "Sensitive Talkgroup", }); - res.status(200).end(); + //res.status(200).end(); return; } sensitiveSpan.end(); }); + console.log("Finished checking sensitive talkgroups"); + if (item.ignoreUnknownTalkgroup == true) { - await trace.getTracer('upload-service').startActiveSpan('check_talkgroup_exists', async (talkgroupSpan) => { + await trace.getTracer('upload-service').startActiveSpan('check_talkgroup_exists', { parent: span }, async (talkgroupSpan) => { talkgroupExists = await Talkgroup.exists({ 'shortName': shortName, @@ -176,15 +179,15 @@ exports.upload = async function (req, res, next) { talkgroupSpan.end(); }); } - // Add in an API Key check + + console.log("Finished checking talkgroup exists"); - - res.status(200).end(); + // Prepare call object let call; - await trace.getTracer('upload-service').startActiveSpan('prepare_call_object', async (prepareSpan) => { + await trace.getTracer('upload-service').startActiveSpan('prepare_call_object', { parent: span }, async (prepareSpan) => { var local_path = "/" + shortName + "/" + time.getFullYear() + "/" + (time.getMonth() + 1) + "/" + time.getDate() + "/"; var object_key = "media/" + shortName + "-" + talkgroupNum + "-" + startTime + path.extname(req.file.originalname); @@ -226,9 +229,10 @@ exports.upload = async function (req, res, next) { prepareSpan.end(); }); + console.log("Finished preparing call object"); let fileContent; // Upload file to S3 - await trace.getTracer('upload-service').startActiveSpan('upload_to_s3', async (uploadSpan) => { + await trace.getTracer('upload-service').startActiveSpan('upload_to_s3',{ parent: span }, async (uploadSpan) => { try { fileContent = fs.readFileSync(req.file.path); @@ -257,8 +261,9 @@ exports.upload = async function (req, res, next) { } }); + console.log("Finished uploading to S3"); // Save call to database - await trace.getTracer('upload-service').startActiveSpan('save_call', async (saveSpan) => { + await trace.getTracer('upload-service').startActiveSpan('save_call',{ parent: span }, async (saveSpan) => { try { await call.save(); sysStats.addCall(call.toObject()); @@ -278,8 +283,10 @@ exports.upload = async function (req, res, next) { saveSpan.end(); } }); + + console.log("Finished saving call to database"); // Clean up temporary file - await trace.getTracer('upload-service').startActiveSpan('cleanup_temp_file', async (cleanupSpan) => { + await trace.getTracer('upload-service').startActiveSpan('cleanup_temp_file',{ parent: span }, async (cleanupSpan) => { try { fs.unlinkSync(req.file.path); cleanupSpan.setAttribute('file.deleted', true); @@ -300,5 +307,7 @@ exports.upload = async function (req, res, next) { } finally { span.end(); } + + console.log("Finished processing call upload"); }); }; diff --git a/backend/index.js b/backend/index.js index 54e481e..98ce947 100644 --- a/backend/index.js +++ b/backend/index.js @@ -1,4 +1,4 @@ -const tracing = require("./agents/honeycomb-tracing"); +const tracing = require("./agents/otel-tracing"); var express = require("express"); var configureExpress = require("./config/express"); diff --git a/backend/package-lock.json b/backend/package-lock.json index 2eb420f..df81416 100644 --- a/backend/package-lock.json +++ b/backend/package-lock.json @@ -14,7 +14,7 @@ "@aws-sdk/credential-providers": "^3.309.0", "@honeycombio/opentelemetry-node": "^0.6.1", "@opentelemetry/api": "^1.7.0", - "@opentelemetry/auto-instrumentations-node": "^0.41.0", + "@opentelemetry/auto-instrumentations-node": "^0.41.1", "@smithy/node-http-handler": "^3.1.4", "adm-zip": "^0.5.10", "aws-sdk": "^2.1353", @@ -1580,14 +1580,14 @@ } }, "node_modules/@opentelemetry/auto-instrumentations-node": { - "version": "0.41.0", - "resolved": "https://registry.npmjs.org/@opentelemetry/auto-instrumentations-node/-/auto-instrumentations-node-0.41.0.tgz", - "integrity": "sha512-CmVVzphp1eb6Aenrr4B6TDsCNuDvut8NQbosTSdJFfnEX4+3szVQuaJquaicGI5tLvVxrgVbZAv52C7bJgMHkg==", + "version": "0.41.1", + "resolved": "https://registry.npmjs.org/@opentelemetry/auto-instrumentations-node/-/auto-instrumentations-node-0.41.1.tgz", + "integrity": "sha512-gQG0mHlPVBQveuemqNYkrL4IB3T6v2e5sMiDI6FQzFWeLzzFrC04R4fY6AE1YmkhOyteCDpHL/U6CULP2mkt8w==", "dependencies": { "@opentelemetry/instrumentation": "^0.48.0", "@opentelemetry/instrumentation-amqplib": "^0.34.0", "@opentelemetry/instrumentation-aws-lambda": "^0.38.0", - "@opentelemetry/instrumentation-aws-sdk": "^0.38.0", + "@opentelemetry/instrumentation-aws-sdk": "^0.38.1", "@opentelemetry/instrumentation-bunyan": "^0.35.0", "@opentelemetry/instrumentation-cassandra-driver": "^0.35.0", "@opentelemetry/instrumentation-connect": "^0.33.0", @@ -2153,9 +2153,9 @@ } }, "node_modules/@opentelemetry/instrumentation-aws-sdk": { - "version": "0.38.0", - "resolved": "https://registry.npmjs.org/@opentelemetry/instrumentation-aws-sdk/-/instrumentation-aws-sdk-0.38.0.tgz", - "integrity": "sha512-q1R+evR1j8sqd5LG+I8fKqP5TAPJEEOmJTvtEDYCVCdtzukI0ABYN8SHEIgDgYZmGBDM0yvC6jH0GmoEwvYuMw==", + "version": "0.38.1", + "resolved": "https://registry.npmjs.org/@opentelemetry/instrumentation-aws-sdk/-/instrumentation-aws-sdk-0.38.1.tgz", + "integrity": "sha512-/Tupb4UfVVkmcopq2H2nr2D/pHLF0riVw2biQQUJ/jGIkfrOgUMMKbShi2RQE4Zy8NFv3xaDn4/pNxzodLBy3w==", "dependencies": { "@opentelemetry/core": "^1.8.0", "@opentelemetry/instrumentation": "^0.48.0", @@ -2810,9 +2810,9 @@ } }, "node_modules/@opentelemetry/propagation-utils": { - "version": "0.30.6", - "resolved": "https://registry.npmjs.org/@opentelemetry/propagation-utils/-/propagation-utils-0.30.6.tgz", - "integrity": "sha512-qvnYee5I/xrBY5XClUlOASIOdoTbnFGNI6+ViKqdoErF2xKmhysXcmxlJNzQFNDK0muTfjvJMZcFyEielARk7g==", + "version": "0.30.10", + "resolved": "https://registry.npmjs.org/@opentelemetry/propagation-utils/-/propagation-utils-0.30.10.tgz", + "integrity": "sha512-hhTW8pFp9PSyosYzzuUL9rdm7HF97w3OCyElufFHyUnYnKkCBbu8ne2LyF/KSdI/xZ81ubxWZs78hX4S7pLq5g==", "engines": { "node": ">=14" }, diff --git a/backend/package.json b/backend/package.json index 07cd8be..3ef6bf1 100644 --- a/backend/package.json +++ b/backend/package.json @@ -8,7 +8,7 @@ "@aws-sdk/credential-providers": "^3.309.0", "@honeycombio/opentelemetry-node": "^0.6.1", "@opentelemetry/api": "^1.7.0", - "@opentelemetry/auto-instrumentations-node": "^0.41.0", + "@opentelemetry/auto-instrumentations-node": "^0.41.1", "@smithy/node-http-handler": "^3.1.4", "adm-zip": "^0.5.10", "aws-sdk": "^2.1353", diff --git a/docker-compose.yml b/docker-compose.yml index 479cc1e..275b978 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -63,6 +63,10 @@ services: image: "backend:${TAG}-${STAGE}" pid: "host" environment: + - OTEL_SERVICE_NAME="test" + - OTEL_EXPORTER_OTLP_PROTOCOL="http/protobuf" + - OTEL_EXPORTER_OTLP_ENDPOINT="https://api.honeycomb.io:443" + - OTEL_EXPORTER_OTLP_HEADERS="x-honeycomb-team=${HONEYCOMB_API_KEY}" - MONGO_NODE_DRIVER_HOST=mongo - MAILJET_KEY - MAILJET_SECRET From 7352c8af82f42aa937a0efc715d6fea14f7655c8 Mon Sep 17 00:00:00 2001 From: Luke Berndt Date: Wed, 21 Aug 2024 15:02:41 -0400 Subject: [PATCH 6/8] Update uploads.js --- backend/controllers/uploads.js | 319 ++++++++++++++++----------------- 1 file changed, 159 insertions(+), 160 deletions(-) diff --git a/backend/controllers/uploads.js b/backend/controllers/uploads.js index 23c7282..854db1c 100644 --- a/backend/controllers/uploads.js +++ b/backend/controllers/uploads.js @@ -17,6 +17,7 @@ const agent = new https.Agent({ }); + var s3_endpoint = process.env['S3_ENDPOINT'] != null ? process.env['S3_ENDPOINT'] : 'https://s3.us-west-1.wasabisys.com'; var s3_region = process.env['S3_REGION'] != null ? process.env['S3_REGION'] : 'us-west-1'; var s3_bucket = process.env['S3_BUCKET'] != null ? process.env['S3_BUCKET'] : 'openmhz-west'; @@ -35,8 +36,9 @@ const client = new S3Client({ }); exports.upload = async function (req, res, next) { + const tracer = trace.getTracer("upload-service") const parentSpan = trace.getActiveSpan(); - await trace.getTracer('upload-service').startActiveSpan('upload_handler', { parent: parentSpan }, async (span) => { + 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); @@ -95,208 +97,205 @@ exports.upload = async function (req, res, next) { console.log("Finished parsing request data"); // Validate that the system exists and the API key is correct - await trace.getTracer('upload-service').startActiveSpan('validate_system', { parent: span }, async (validateSpan) => { - - try { - item = await System.findOne({ 'shortName': shortName }, ["key", "ignoreUnknownTalkgroup"]); - validateSpan.setAttribute('system.exists', !!item); + const validateSpan = tracer.startSpan("validate_system"); - if (!item) { - validateSpan.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "ShortName does not exist", - }); + try { + item = await System.findOne({ 'shortName': shortName }, ["key", "ignoreUnknownTalkgroup"]); + validateSpan.setAttribute('system.exists', !!item); - console.info("[" + req.params.shortName + "] Error /:shortName/upload ShortName does not exist"); - res.status(500); - res.send("ShortName does not exist: " + shortName + "\n"); - return; - } - if (apiKey != item.key) { - validateSpan.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "API Key Mismatch", - }); - console.warn("[" + req.params.shortName + "] Error /:shortName/upload API Key Mismatch - Provided key: " + apiKey); - res.status(500); - res.send("API Keys do not match!\n"); - return; - } - } catch (err) { - validateSpan.recordException(err); + + if (!item) { validateSpan.setStatus({ code: opentelemetry.SpanStatusCode.ERROR, - message: "Error validating system: " + err.message, + message: "ShortName does not exist", }); - res.status(500).send("Error, invalid shortName:\n" + err); + + console.info("[" + req.params.shortName + "] Error /:shortName/upload ShortName does not exist"); + res.status(500); + res.send("ShortName does not exist: " + shortName + "\n"); return; - } finally { - validateSpan.end(); } - }); + if (apiKey != item.key) { + validateSpan.setStatus({ + code: opentelemetry.SpanStatusCode.ERROR, + message: "API Key Mismatch", + }); + console.warn("[" + req.params.shortName + "] Error /:shortName/upload API Key Mismatch - Provided key: " + apiKey); + res.status(500); + res.send("API Keys do not match!\n"); + return; + } + } catch (err) { + validateSpan.recordException(err); + validateSpan.setStatus({ + code: opentelemetry.SpanStatusCode.ERROR, + message: "Error validating system: " + err.message, + }); + res.status(500).send("Error, invalid shortName:\n" + err); + return; + } finally { + validateSpan.end(); + } + console.log("Finished validating system API key"); // Blocking sensitive talkgroups - await trace.getTracer('upload-service').startActiveSpan('check_sensitive_talkgroups', { parent: span }, async (sensitiveSpan) => { + const sensitiveSpan = tracer.startSpan("check_sensitive_talkgroups"); + if ((shortName == "hennearmer") && ((talkgroupNum == 3421) || (talkgroupNum == 3423))) { + sensitiveSpan.setStatus({ + code: opentelemetry.SpanStatusCode.ERROR, + message: "Sensitive Talkgroup", + }); + //res.status(200).end(); + return; + } + sensitiveSpan.end(); - if ((shortName == "hennearmer") && ((talkgroupNum == 3421) || (talkgroupNum == 3423))) { - sensitiveSpan.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "Sensitive Talkgroup", - }); - //res.status(200).end(); - return; - } - sensitiveSpan.end(); - }); console.log("Finished checking sensitive talkgroups"); if (item.ignoreUnknownTalkgroup == true) { - await trace.getTracer('upload-service').startActiveSpan('check_talkgroup_exists', { parent: span }, async (talkgroupSpan) => { - - talkgroupExists = await Talkgroup.exists({ - 'shortName': shortName, - 'num': talkgroupNum - }); - - if (!talkgroupExists) { - try { - fs.unlinkSync(req.file.path) - //file removed - } catch (err) { - console.log("[" + call.shortName + "] error deleting: " + req.file.path); - } - - res.status(500); - res.send("Talkgroup does not exist, skipping.\n"); - - - - return; - } - talkgroupSpan.end(); + const talkgroupSpan = tracer.startSpan("check_talkgroup_exists"); + talkgroupExists = await Talkgroup.exists({ + 'shortName': shortName, + 'num': talkgroupNum }); + + if (!talkgroupExists) { + try { + fs.unlinkSync(req.file.path) + //file removed + } catch (err) { + console.log("[" + call.shortName + "] error deleting: " + req.file.path); + } + + res.status(500); + res.send("Talkgroup does not exist, skipping.\n"); + + + + return; + } + talkgroupSpan.end(); + } console.log("Finished checking talkgroup exists"); - + // Prepare call object let call; - await trace.getTracer('upload-service').startActiveSpan('prepare_call_object', { parent: span }, async (prepareSpan) => { + const prepareSpan = tracer.startSpan('prepare_call_object'); - var local_path = "/" + shortName + "/" + time.getFullYear() + "/" + (time.getMonth() + 1) + "/" + time.getDate() + "/"; - var object_key = "media/" + shortName + "-" + talkgroupNum + "-" + startTime + path.extname(req.file.originalname); + var local_path = "/" + shortName + "/" + time.getFullYear() + "/" + (time.getMonth() + 1) + "/" + time.getDate() + "/"; + var object_key = "media/" + shortName + "-" + talkgroupNum + "-" + startTime + path.extname(req.file.originalname); - var endpoint = s3_endpoint; - var bucket = s3_bucket; - var url = s3_public_url + "/" + object_key; + var endpoint = s3_endpoint; + var bucket = s3_bucket; + var url = s3_public_url + "/" + object_key; - call = new Call({ - shortName: shortName, - talkgroupNum: talkgroupNum, - objectKey: object_key, - endpoint: endpoint, - bucket: bucket, - time: time, - name: talkgroupNum + "-" + startTime + ".m4a", - freq: freq, - errorCount: errorCount, - spikeCount: spikeCount, - url: url, - emergency: emergency, - path: local_path, - srcList: srcList, - len: -1 - }); - - - try { - if (req.body.call_length) { - call.len = parseFloat(req.body.call_length); - } else { - call.len = (stopTime - time) / 1000; - } - } catch (err) { - console.error(err); - } - prepareSpan.setAttribute('call.object_key', object_key); - prepareSpan.setAttribute('call.url', url); - prepareSpan.end(); + call = new Call({ + shortName: shortName, + talkgroupNum: talkgroupNum, + objectKey: object_key, + endpoint: endpoint, + bucket: bucket, + time: time, + name: talkgroupNum + "-" + startTime + ".m4a", + freq: freq, + errorCount: errorCount, + spikeCount: spikeCount, + url: url, + emergency: emergency, + path: local_path, + srcList: srcList, + len: -1 }); + + try { + if (req.body.call_length) { + call.len = parseFloat(req.body.call_length); + } else { + call.len = (stopTime - time) / 1000; + } + } catch (err) { + console.error(err); + } + prepareSpan.setAttribute('call.object_key', object_key); + prepareSpan.setAttribute('call.url', url); + prepareSpan.end(); + + console.log("Finished preparing call object"); let fileContent; // Upload file to S3 - await trace.getTracer('upload-service').startActiveSpan('upload_to_s3',{ parent: span }, async (uploadSpan) => { + const uploadSpan = tracer.startSpan('upload_to_s3'); + try { + fileContent = fs.readFileSync(req.file.path); + //s3Src = fs.createReadStream(req.file.path); - try { - fileContent = fs.readFileSync(req.file.path); - //s3Src = fs.createReadStream(req.file.path); + const command = new PutObjectCommand({ + Bucket: s3_bucket, + Key: object_key, + Body: fileContent, + ACL: 'public-read' + }); + await client.send(command); - const command = new PutObjectCommand({ - Bucket: s3_bucket, - Key: object_key, - Body: fileContent, - ACL: 'public-read' - }); + } 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(); + } - await client.send(command); - - - } 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(); - } - }); - console.log("Finished uploading to S3"); // Save call to database - await trace.getTracer('upload-service').startActiveSpan('save_call',{ parent: span }, async (saveSpan) => { - try { - await call.save(); - sysStats.addCall(call.toObject()); - // we only want to notify clients if the clip is longer than 1 second. - if (call.len >= 1) { - req.call = call.toObject(); - next(); - } - } 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(); + const saveSpan = tracer.startSpan('save_call'); + + try { + await call.save(); + sysStats.addCall(call.toObject()); + // we only want to notify clients if the clip is longer than 1 second. + if (call.len >= 1) { + req.call = call.toObject(); + next(); } - }); + } 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(); + } + console.log("Finished saving call to database"); // Clean up temporary file - await trace.getTracer('upload-service').startActiveSpan('cleanup_temp_file',{ parent: span }, 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(); - } - }); + const cleanupSpan = tracer.startSpan('cleanup_temp_file'); + + 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(); + } + } catch (error) { span.recordException(error); From 0e6e69c56e2d25a402baf32fc77e4cac59c855c4 Mon Sep 17 00:00:00 2001 From: Luke Berndt Date: Wed, 21 Aug 2024 15:11:23 -0400 Subject: [PATCH 7/8] Update uploads.js --- backend/controllers/uploads.js | 4 ---- 1 file changed, 4 deletions(-) diff --git a/backend/controllers/uploads.js b/backend/controllers/uploads.js index 854db1c..1209895 100644 --- a/backend/controllers/uploads.js +++ b/backend/controllers/uploads.js @@ -180,12 +180,8 @@ exports.upload = async function (req, res, next) { talkgroupSpan.end(); } - console.log("Finished checking talkgroup exists"); - - - // Prepare call object let call; const prepareSpan = tracer.startSpan('prepare_call_object'); From 00dbc4725ef8762d8cbdab5730606325e4d403a0 Mon Sep 17 00:00:00 2001 From: Luke Berndt Date: Wed, 21 Aug 2024 21:41:57 -0400 Subject: [PATCH 8/8] its working --- backend/controllers/uploads.js | 34 +++++++--------------------------- 1 file changed, 7 insertions(+), 27 deletions(-) diff --git a/backend/controllers/uploads.js b/backend/controllers/uploads.js index 1209895..a3bcb3f 100644 --- a/backend/controllers/uploads.js +++ b/backend/controllers/uploads.js @@ -56,7 +56,7 @@ exports.upload = async function (req, res, next) { } // Extract and parse request data - + const parseSpan = tracer.startSpan('parse_request_data'); var shortName = req.params.shortName.toLowerCase(); var apiKey = req.body.api_key; @@ -91,15 +91,11 @@ exports.upload = async function (req, res, next) { } parseSpan.end(); - let item = null; - - console.log("Finished parsing request data"); // Validate that the system exists and the API key is correct const validateSpan = tracer.startSpan("validate_system"); - try { item = await System.findOne({ 'shortName': shortName }, ["key", "ignoreUnknownTalkgroup"]); validateSpan.setAttribute('system.exists', !!item); @@ -111,7 +107,7 @@ exports.upload = async function (req, res, next) { message: "ShortName does not exist", }); - console.info("[" + req.params.shortName + "] Error /:shortName/upload ShortName does not exist"); + console.warn("[" + req.params.shortName + "] Error /:shortName/upload ShortName does not exist"); res.status(500); res.send("ShortName does not exist: " + shortName + "\n"); return; @@ -138,8 +134,6 @@ exports.upload = async function (req, res, next) { validateSpan.end(); } - - console.log("Finished validating system API key"); // Blocking sensitive talkgroups const sensitiveSpan = tracer.startSpan("check_sensitive_talkgroups"); if ((shortName == "hennearmer") && ((talkgroupNum == 3421) || (talkgroupNum == 3423))) { @@ -147,14 +141,11 @@ exports.upload = async function (req, res, next) { code: opentelemetry.SpanStatusCode.ERROR, message: "Sensitive Talkgroup", }); - //res.status(200).end(); + res.status(200).end(); return; } sensitiveSpan.end(); - - console.log("Finished checking sensitive talkgroups"); - if (item.ignoreUnknownTalkgroup == true) { const talkgroupSpan = tracer.startSpan("check_talkgroup_exists"); talkgroupExists = await Talkgroup.exists({ @@ -167,20 +158,16 @@ exports.upload = async function (req, res, next) { fs.unlinkSync(req.file.path) //file removed } catch (err) { - console.log("[" + call.shortName + "] error deleting: " + req.file.path); + console.error("[" + call.shortName + "] error deleting: " + req.file.path); } - res.status(500); res.send("Talkgroup does not exist, skipping.\n"); - - - return; } talkgroupSpan.end(); - } - console.log("Finished checking talkgroup exists"); + + res.status(200).end(); // Prepare call object let call; @@ -225,8 +212,6 @@ exports.upload = async function (req, res, next) { prepareSpan.setAttribute('call.url', url); prepareSpan.end(); - - console.log("Finished preparing call object"); let fileContent; // Upload file to S3 const uploadSpan = tracer.startSpan('upload_to_s3'); @@ -253,8 +238,6 @@ exports.upload = async function (req, res, next) { uploadSpan.end(); } - - console.log("Finished uploading to S3"); // Save call to database const saveSpan = tracer.startSpan('save_call'); @@ -277,8 +260,6 @@ exports.upload = async function (req, res, next) { saveSpan.end(); } - - console.log("Finished saving call to database"); // Clean up temporary file const cleanupSpan = tracer.startSpan('cleanup_temp_file'); @@ -294,6 +275,7 @@ exports.upload = async function (req, res, next) { } catch (error) { + console.error("Error processing call upload: " + error); span.recordException(error); span.setStatus({ code: opentelemetry.SpanStatusCode.ERROR, @@ -302,7 +284,5 @@ exports.upload = async function (req, res, next) { } finally { span.end(); } - - console.log("Finished processing call upload"); }); };