transcribe: create otel child spans for each stt result that is returned in a long-running transcribe (#368)

lib/tasks/transcribe.js

@@ -14,6 +14,8 @@ const {
 } = require('../utils/constants');
 const { normalizeJambones } = require('@jambonz/verb-specifications');
 
+const STT_LISTEN_SPAN_NAME = 'stt-listen';
+
 class TaskTranscribe extends Task {
   constructor(logger, opts, parentTask) {
     super(logger, opts);
@@ -50,6 +52,8 @@ class TaskTranscribe extends Task {
 
     /* buffer for soniox transcripts */
     this._sonioxTranscripts = [];
+
+    this.childSpan = [null, null];
   }
 
   get name() { return TaskName.Transcribe; }
@@ -237,6 +241,10 @@ class TaskTranscribe extends Task {
       .catch((err) => this.logger.info(err, 'Error setting channel variables'));
 
     await this._transcribe(ep);
+
+    /* start child span for this channel */
+    const {span, ctx} = this.startChildSpan(`${STT_LISTEN_SPAN_NAME}:${channel}`);
+    this.childSpan[channel - 1] = {span, ctx};
   }
 
   async _transcribe(ep) {
@@ -286,6 +294,12 @@ class TaskTranscribe extends Task {
       }
     }
 
+    /* we've got a transcript, so end the otel child span for this channel */
+    if (this.childSpan[channel - 1] && this.childSpan[channel - 1].span) {
+      this.childSpan[channel - 1].span.setAttributes({'stt.resolve': 'transcript', 'stt.result': JSON.stringify(evt)});
+      this.childSpan[channel - 1].span.end();
+    }
+
     if (this.transcriptionHook) {
       const b3 = this.getTracingPropagation();
       const httpHeaders = b3 && {b3};
@@ -316,16 +330,38 @@ class TaskTranscribe extends Task {
       this._clearTimer();
       this.notifyTaskDone();
     }
+    else {
+      /* start another child span for this channel */
+      const {span, ctx} = this.startChildSpan(`${STT_LISTEN_SPAN_NAME}:${channel}`);
+      this.childSpan[channel - 1] = {span, ctx};
+    }
   }
 
   _onNoAudio(cs, ep, channel) {
     this.logger.debug(`TaskTranscribe:_onNoAudio restarting transcription on channel ${channel}`);
+    if (this.childSpan[channel - 1] && this.childSpan[channel - 1].span) {
+      this.childSpan[channel - 1].span.setAttributes({'stt.resolve': 'timeout'});
+      this.childSpan[channel - 1].span.end();
+    }
     this._transcribe(ep);
+
+    /* start new child span for this channel */
+    const {span, ctx} = this.startChildSpan(`${STT_LISTEN_SPAN_NAME}:${channel}`);
+    this.childSpan[channel - 1] = {span, ctx};
   }
 
   _onMaxDurationExceeded(cs, ep, channel) {
     this.logger.debug(`TaskTranscribe:_onMaxDurationExceeded restarting transcription on channel ${channel}`);
+    if (this.childSpan[channel - 1] && this.childSpan[channel - 1].span) {
+      this.childSpan[channel - 1].span.setAttributes({'stt.resolve': 'max duration exceeded'});
+      this.childSpan[channel - 1].span.end();
+    }
+
     this._transcribe(ep);
+
+    /* start new child span for this channel */
+    const {span, ctx} = this.startChildSpan(`${STT_LISTEN_SPAN_NAME}:${channel}`);
+    this.childSpan[channel - 1] = {span, ctx};
   }
 
   _clearTimer() {
@@ -338,7 +374,7 @@ class TaskTranscribe extends Task {
     this.logger.debug('TaskTranscribe:_onDeepgramConnect');
   }
 
-  _onDeepGramConnectFailure(cs, _ep, _channel, evt) {
+  _onDeepGramConnectFailure(cs, _ep, channel, evt) {
     const {reason} = evt;
     const {writeAlerts, AlertType} = cs.srf.locals;
     this.logger.info({evt}, 'TaskTranscribe:_onDeepgramConnectFailure');
@@ -349,6 +385,11 @@ class TaskTranscribe extends Task {
       vendor: 'deepgram',
     }).catch((err) => this.logger.info({err}, 'Error generating alert for deepgram connection failure'));
     this.notifyError(`Failed connecting to speech vendor deepgram: ${reason}`);
+
+    if (this.childSpan[channel - 1] && this.childSpan[channel - 1].span) {
+      this.childSpan[channel - 1].span.setAttributes({'stt.resolve': 'connection failure'});
+      this.childSpan[channel - 1].span.end();
+    }
     this.notifyTaskDone();
   }
 
@@ -356,7 +397,7 @@ class TaskTranscribe extends Task {
     this.logger.debug('TaskTranscribe:_onIbmConnect');
   }
 
-  _onIbmConnectFailure(cs, _ep, _channel, evt) {
+  _onIbmConnectFailure(cs, _ep, channel, evt) {
     const {reason} = evt;
     const {writeAlerts, AlertType} = cs.srf.locals;
     this.logger.info({evt}, 'TaskTranscribe:_onIbmConnectFailure');
@@ -367,6 +408,11 @@ class TaskTranscribe extends Task {
       vendor: 'ibm',
     }).catch((err) => this.logger.info({err}, 'Error generating alert for IBM connection failure'));
     this.notifyError(`Failed connecting to speech vendor IBM: ${reason}`);
+
+    if (this.childSpan[channel - 1] && this.childSpan[channel - 1].span) {
+      this.childSpan[channel - 1].span.setAttributes({'stt.resolve': 'connection failure'});
+      this.childSpan[channel - 1].span.end();
+    }
     this.notifyTaskDone();
   }
   _onIbmError(cs, _ep, _channel, evt) {