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 072078f..a3bcb3f 100644 --- a/backend/controllers/uploads.js +++ b/backend/controllers/uploads.js @@ -5,7 +5,7 @@ var sysStats = require("../sys_stats"); var System = require("../models/system"); var Talkgroup = require("../models/talkgroup"); var { callModel: Call } = require("../models/call"); - +const { trace } = opentelemetry; const { PutObjectCommand, S3Client } = require("@aws-sdk/client-s3"); const { fromIni } = require("@aws-sdk/credential-providers"); @@ -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,232 +36,253 @@ 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 tracer = trace.getTracer("upload-service") + const parentSpan = trace.getActiveSpan(); + await tracer.startActiveSpan('upload_handler', { parent: parentSpan }, 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 + const parseSpan = tracer.startSpan('parse_request_data'); - - 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_public_url + "/" + 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; + 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); } - } catch (err) { - console.error(err); - } - let fileContent; - try { - fileContent = fs.readFileSync(req.file.path); - //s3Src = fs.createReadStream(req.file.path); - } catch (err) { - span.setStatus({ - code: opentelemetry.SpanStatusCode.ERROR, - message: "Unable to open file: " + req.file.path + " Error: " + err, - }); - console.error("[" + call.shortName + "] Unable to open file: " + req.file.path + " Error: " + err); - return; - } - const command = new PutObjectCommand({ - Bucket: s3_bucket, - Key: object_key, - Body: fileContent, - ACL: 'public-read' - }); + var startTime = req.body.start_time; + var emergency = parseInt(req.body.emergency); - try { - await client.send(command); + let errorCount = parseInt(req.body.error_count); + let spikeCount = parseInt(req.body.spike_count); - } 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(); + if (Number.isNaN(errorCount)) { + errorCount = 0; } - } catch (err) { - console.warn("[" + call.shortName + "] Error saving call: " + err); + 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 + const validateSpan = tracer.startSpan("validate_system"); + + 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.warn("[" + 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 + 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 (item.ignoreUnknownTalkgroup == true) { + 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.error("[" + call.shortName + "] error deleting: " + req.file.path); + } + res.status(500); + res.send("Talkgroup does not exist, skipping.\n"); + return; + } + talkgroupSpan.end(); + } + + res.status(200).end(); + + // Prepare call object + let call; + 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 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(); + + let fileContent; + // Upload file to S3 + const uploadSpan = tracer.startSpan('upload_to_s3'); + 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 + 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(); + } + + // Clean up temporary file + 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) { + console.error("Error processing call upload: " + error); + span.recordException(error); span.setStatus({ code: opentelemetry.SpanStatusCode.ERROR, - message: "Error saving call: " + err, + message: error.message, }); - /*res.status(500); - sysStats.addError(call.toObject()); - res.contentType('json'); - res.status(500); - res.send(JSON.stringify({ - success: false, - error: "File Upload" - }));*/ + } finally { + span.end(); } - - try { - fs.unlinkSync(req.file.path); - } catch (err) { - console.warn("There was an Error uploading an deleting: " + req.file.path); - }; }); -} +}; 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