Update worker tracing

Author: juffalow

src/services/aws.ts

@@ -27,6 +27,17 @@ export default () => {
       }));
     },
 
+    get sqsConsumer(): SQSClient {
+      return new SQSClient({
+        apiVersion: '2012-11-05',
+        region: config.services.queue.region,
+        credentials: {
+          accessKeyId: config.services.queue.accessKeyId,
+          secretAccessKey: config.services.queue.secretAccessKey,
+        }
+      });
+    },
+
     get mc(): MediaConvertClient {
       return services.Trace.captureAWSv3Client(new MediaConvertClient({
         apiVersion: '2017-08-29',

src/services/trace/AWSXRay.ts

@@ -6,8 +6,7 @@ class AWSXRay implements Services.Trace {
   constructor (plugins: string) {
     this.setPlugins(plugins);
 
-    XRay.captureHTTPsGlobal(http, true);
-    XRay.captureHTTPsGlobal(https, true);
+    XRay.middleware.disableCentralizedSampling();
   }
 
   public openSegment(defaultName: string) {
@@ -18,8 +17,8 @@ class AWSXRay implements Services.Trace {
     return XRay.express.closeSegment();
   }
 
-  public createSegment(name: string) {
-    return new XRay.Segment(name);
+  public createSegment(name: string, rootId?: string | null, parentId?: string | null) {
+    return new XRay.Segment(name, rootId, parentId);
   }
 
   public setSegment(segment: unknown): void {
@@ -29,7 +28,7 @@ class AWSXRay implements Services.Trace {
   public getTraceId(): string {
     const namespace = XRay.getNamespace();
 
-    if (namespace.active === null) {
+    if (namespace.active === null || 'segment' in namespace.active === false) {
       return undefined;
     }
 
@@ -46,6 +45,11 @@ class AWSXRay implements Services.Trace {
     return XRay.captureAWSv3Client(client as T & { middlewareStack: { remove: any, use: any }, config: any });
   }
 
+  public captureHTTPRequests(): void {
+    XRay.captureHTTPsGlobal(http, true);
+    XRay.captureHTTPsGlobal(https, true);
+  }
+
   public setDaemonAddress(address: string) {
     XRay.setDaemonAddress(address);
   }
@@ -63,6 +67,10 @@ class AWSXRay implements Services.Trace {
 
     XRay.config(xrayPlugins); 
   }
+
+  public processTraceData(data: string): { [key: string]: string } {
+    return XRay.utils.processTraceData(data);
+  }
 }
 
 export default AWSXRay;

src/services/trace/CLSHooked.ts

@@ -44,9 +44,17 @@ class CLSHooked implements Services.Trace {
     return client;
   }
 
+  public captureHTTPRequests(): void {
+    return;
+  }
+
   public getNamespace() {
     return cls.getNamespace(NAMESPACE) || cls.createNamespace(NAMESPACE);
   }
+
+  public processTraceData(): { [key: string]: string } {
+    return {};
+  }
 }
 
 export default CLSHooked;

src/services/worker/Consumer.ts

@@ -56,7 +56,7 @@ declare namespace Services {
 
     closeSegment(): unknown;
 
-    createSegment(name: string): unknown;
+    createSegment(name: string, rootId?: string | null, parentId?: string | null): unknown;
 
     setSegment(segment: unknown): void;
 
@@ -65,5 +65,9 @@ declare namespace Services {
     getNamespace(): unknown;
 
     captureAWSv3Client<T>(client: T): T;
+
+    captureHTTPRequests(): void;
+
+    processTraceData(data: string): { [key: string]: string };
   }
 }

src/types/services.d.ts

@@ -1,7 +1,6 @@
 import services from './services';
 import config from './config';
 import logger from './logger';
-import Consumer from './services/worker/Consumer';
 
 async function start() {
   const namespace = config.services.trace.daemonAddressNamespace;
@@ -14,8 +13,12 @@ async function start() {
 
   logger.info('Upload worker is running');
 
-  const consumer = Consumer.get();
-  consumer.start();
+  const consumer = (await import('./worker/Consumer')).default;
+  const ns = services.Trace.getNamespace();
+
+  (ns as any).run(() => {
+    consumer.start();
+  });
 }
 
 start();

src/worker.ts

@@ -0,0 +1,63 @@
+import { Consumer } from 'sqs-consumer';
+import services from '../services';
+import config from '../config';
+import logger from '../logger';
+import video from '../services/job/video';
+import repositories from '../repositories';
+
+const consumer = Consumer.create({
+  queueUrl: config.services.queue.url,
+  attributeNames: ['AWSTraceHeader', 'X-Request-ID'] as any,
+  handleMessage: async (message) => {
+    const traceHeader = message.Attributes.AWSTraceHeader;
+    const traceData = services.Trace.processTraceData(traceHeader);
+    const segment: any = services.Trace.createSegment('upload-service-worker', traceData.root, traceData.parent);
+    services.Trace.setSegment(segment);
+
+    logger.debug('Received message from queue!', { ...message });
+
+    const body = JSON.parse(message.Body);
+
+    if ('fileId' in body) {
+      try {
+        const file = await repositories.File.get(body.fileId, null);
+        const values = await services.Job.process(body.mimetype, file) as Record<string, string | number | boolean>;
+        await repositories.File.update({ ...values, cacheable: !body.mimetype.startsWith('video') }, { id: file.id });
+      } catch (err) {
+        logger.error('Could not process file!', { error: err.message, stack: err.stack });
+      }
+    }
+
+    if ('detail' in body && 'jobId' in body.detail) {
+      if (body.detail.status === 'COMPLETE') {
+        const updatedFile = await video.AWSVideo.complete(body) as Record<string, string | number | boolean>;
+        await repositories.File.update({ ...updatedFile, cacheable: true }, { id: updatedFile['id'] as string });
+      }
+
+      if (body.detail.status === 'ERROR') {
+        logger.error('Media convert job ended with error!', body);
+      }
+    }
+
+    segment && segment.close();
+  },
+  sqs: services.AWS.sqsConsumer,
+});
+
+consumer.on('message_processed', (message) => {
+  logger.debug('Message successfully processed and removed from the queue!', { message });
+});
+
+consumer.on('error', (err) => {
+  logger.error('Unknown error occured!', { message: err.message, stack: err.stack });
+});
+
+consumer.on('processing_error', (err) => {
+  logger.error('Error while processing message from queue!', { message: err.message, stack: err.stack });
+});
+
+consumer.on('timeout_error', (err) => {
+  logger.error('Timeout error!', { message: err.message, stack: err.stack });
+});
+
+export default consumer;