Update uploads.js

This commit is contained in:
Luke Berndt
2024-08-21 15:02:41 -04:00
parent fc77bd2934
commit 7352c8af82

View File

@@ -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_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_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'; 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) { exports.upload = async function (req, res, next) {
const tracer = trace.getTracer("upload-service")
const parentSpan = trace.getActiveSpan(); 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 { try {
span.setAttribute('call.shortName', req.params.shortName.toLowerCase()); span.setAttribute('call.shortName', req.params.shortName.toLowerCase());
span.setAttribute('call.talkgroup_num', req.body.talkgroup_num); span.setAttribute('call.talkgroup_num', req.body.talkgroup_num);
@@ -95,89 +97,88 @@ exports.upload = async function (req, res, next) {
console.log("Finished parsing request data"); console.log("Finished parsing request data");
// Validate that the system exists and the API key is correct // Validate that the system exists and the API key is correct
await trace.getTracer('upload-service').startActiveSpan('validate_system', { parent: span }, async (validateSpan) => { const validateSpan = tracer.startSpan("validate_system");
try {
item = await System.findOne({ 'shortName': shortName }, ["key", "ignoreUnknownTalkgroup"]);
validateSpan.setAttribute('system.exists', !!item);
if (!item) { try {
validateSpan.setStatus({ item = await System.findOne({ 'shortName': shortName }, ["key", "ignoreUnknownTalkgroup"]);
code: opentelemetry.SpanStatusCode.ERROR, validateSpan.setAttribute('system.exists', !!item);
message: "ShortName does not exist",
});
console.info("[" + req.params.shortName + "] Error /:shortName/upload ShortName does not exist");
res.status(500); if (!item) {
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({ validateSpan.setStatus({
code: opentelemetry.SpanStatusCode.ERROR, 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; 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"); console.log("Finished validating system API key");
// Blocking sensitive talkgroups // 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"); console.log("Finished checking sensitive talkgroups");
if (item.ignoreUnknownTalkgroup == true) { if (item.ignoreUnknownTalkgroup == true) {
await trace.getTracer('upload-service').startActiveSpan('check_talkgroup_exists', { parent: span }, async (talkgroupSpan) => { const talkgroupSpan = tracer.startSpan("check_talkgroup_exists");
talkgroupExists = await Talkgroup.exists({
talkgroupExists = await Talkgroup.exists({ 'shortName': shortName,
'shortName': shortName, 'num': talkgroupNum
'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();
}); });
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"); console.log("Finished checking talkgroup exists");
@@ -187,116 +188,114 @@ exports.upload = async function (req, res, next) {
// Prepare call object // Prepare call object
let call; 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 local_path = "/" + shortName + "/" + time.getFullYear() + "/" + (time.getMonth() + 1) + "/" + time.getDate() + "/";
var object_key = "media/" + shortName + "-" + talkgroupNum + "-" + startTime + path.extname(req.file.originalname); var object_key = "media/" + shortName + "-" + talkgroupNum + "-" + startTime + path.extname(req.file.originalname);
var endpoint = s3_endpoint; var endpoint = s3_endpoint;
var bucket = s3_bucket; var bucket = s3_bucket;
var url = s3_public_url + "/" + object_key; var url = s3_public_url + "/" + object_key;
call = new Call({ call = new Call({
shortName: shortName, shortName: shortName,
talkgroupNum: talkgroupNum, talkgroupNum: talkgroupNum,
objectKey: object_key, objectKey: object_key,
endpoint: endpoint, endpoint: endpoint,
bucket: bucket, bucket: bucket,
time: time, time: time,
name: talkgroupNum + "-" + startTime + ".m4a", name: talkgroupNum + "-" + startTime + ".m4a",
freq: freq, freq: freq,
errorCount: errorCount, errorCount: errorCount,
spikeCount: spikeCount, spikeCount: spikeCount,
url: url, url: url,
emergency: emergency, emergency: emergency,
path: local_path, path: local_path,
srcList: srcList, srcList: srcList,
len: -1 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();
}); });
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"); console.log("Finished preparing call object");
let fileContent; let fileContent;
// Upload file to S3 // 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 { const command = new PutObjectCommand({
fileContent = fs.readFileSync(req.file.path); Bucket: s3_bucket,
//s3Src = fs.createReadStream(req.file.path); Key: object_key,
Body: fileContent,
ACL: 'public-read'
});
await client.send(command);
const command = new PutObjectCommand({ } catch (err) {
Bucket: s3_bucket, uploadSpan.recordException(err);
Key: object_key, uploadSpan.setStatus({
Body: fileContent, code: opentelemetry.SpanStatusCode.ERROR,
ACL: 'public-read' 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"); console.log("Finished uploading to S3");
// Save call to database // Save call to database
await trace.getTracer('upload-service').startActiveSpan('save_call',{ parent: span }, async (saveSpan) => { const saveSpan = tracer.startSpan('save_call');
try {
await call.save(); try {
sysStats.addCall(call.toObject()); await call.save();
// we only want to notify clients if the clip is longer than 1 second. sysStats.addCall(call.toObject());
if (call.len >= 1) { // we only want to notify clients if the clip is longer than 1 second.
req.call = call.toObject(); if (call.len >= 1) {
next(); 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();
} }
}); } 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"); console.log("Finished saving call to database");
// Clean up temporary file // Clean up temporary file
await trace.getTracer('upload-service').startActiveSpan('cleanup_temp_file',{ parent: span }, async (cleanupSpan) => { const cleanupSpan = tracer.startSpan('cleanup_temp_file');
try {
fs.unlinkSync(req.file.path); try {
cleanupSpan.setAttribute('file.deleted', true); fs.unlinkSync(req.file.path);
} catch (err) { cleanupSpan.setAttribute('file.deleted', true);
cleanupSpan.recordException(err); } catch (err) {
console.warn("There was an Error deleting: " + req.file.path); cleanupSpan.recordException(err);
} finally { console.warn("There was an Error deleting: " + req.file.path);
cleanupSpan.end(); } finally {
} cleanupSpan.end();
}); }
} catch (error) { } catch (error) {
span.recordException(error); span.recordException(error);