Add OpenTelemetry support

.dockerignore

@@ -17,3 +17,4 @@
 !tests
 !utils
 !wflib
+!tracing.js

Dockerfile

@@ -3,4 +3,4 @@ FROM node:12-alpine
 #ENV PATH $PATH:/node_modules/.bin
 
 COPY . /hyperflow
-RUN npm install -g /hyperflow 
+RUN npm install -g /hyperflow

examples/RemoteJobs/functions.js

@@ -1,3 +1,4 @@
+const tracer = process.env.HF_VAR_ENABLE_TRACING === "1" ? require("../../tracing.js")("hyperflow-service") : undefined;
 var spawn = require('child_process').spawn;
 var log4js = require('log4js');
 var createJobMessage = require('../../common/jobMessage.js').createJobMessage;
@@ -6,64 +7,89 @@ var createJobMessage = require('../../common/jobMessage.js').createJobMessage;
 // Spawns a job "node handler.js" and waits for the notification of its
 // completion using the Redis job status notification mechanism
 async function submitRemoteJob(ins, outs, context, cb) {
-  let fname='wftrace-' + context.hfId + '-' + context.appId + '.log';
+
+  let jobMessage = JSON.stringify(createJobMessage(ins, outs, context));
+
+  let fname = 'wftrace-' + context.hfId + '-' + context.appId + '.log';
+
   log4js.configure({
-    appenders: { hftrace: { type: 'file', filename: fname } },
-    categories: { default: { appenders: ['hftrace'], level: 'error' } }
+    appenders: {hftrace: {type: 'file', filename: fname}},
+    categories: {default: {appenders: ['hftrace'], level: 'error'}}
   });
 
   var logger = log4js.getLogger();
 
   logger.level = 'debug';
   console.log("Spawning process...");
 
-  //console.log(ins.map(i => i));
-
-  var input_dir = context.executor.input_dir,
+  function runJob(trace_id = undefined, parent_id = undefined, span = undefined) {
+    var input_dir = context.executor.input_dir,
       work_dir = context.executor.work_dir,
       output_dir = context.executor.output_dir;
-
-  let jobMessage = JSON.stringify(createJobMessage(ins, outs, context));
-
-  var cmd;
-
-  // if 'container' is present, run through Docker, mounting all directories if necessary
-  if (context.container) {
-    cmd = 'docker run ';
-    if (input_dir) cmd += ' -v ' + input_dir + ':/input_dir ';
-    if (work_dir) cmd += ' -v ' + work_dir + ':/work_dir ';
-    if (output_dir) cmd += ' -v ' + output_dir + ':/output_dir ';
-    cmd += container + ' node';
-  } else cmd = 'node'
 
-  try {
-    if (work_dir) { process.chdir(work_dir); }
-  } catch (error) {
-    throw error;
+    var cmd;
+
+    // if 'container' is present, run through Docker, mounting all directories if necessary
+    if (context.container) {
+      cmd = 'docker run ';
+      if (input_dir) cmd += ' -v ' + input_dir + ':/input_dir ';
+      if (work_dir) cmd += ' -v ' + work_dir + ':/work_dir ';
+      if (output_dir) cmd += ' -v ' + output_dir + ':/output_dir ';
+      cmd += container + ' node';
+    } else cmd = 'node'
+
+    try {
+      if (work_dir) {
+        process.chdir(work_dir);
+      }
+    } catch (error) {
+      throw error;
+    }
+
+    // "submit" job (start the handler process)
+    var proc = spawn(cmd, ['../../../hyperflow-job-executor/jobexec.js', context.taskId, context.redis_url],
+        {shell: true, env: {...process.env, HF_VAR_OT_PARENT_ID: parent_id, HF_VAR_OT_TRACE_ID: trace_id,
+            HF_VAR_ENABLE_TRACING: process.env.HF_VAR_ENABLE_TRACING}});
+
+    if (span !== undefined && span.isRecording()) {
+      span.setAttributes({
+        'hfId': context.hfId, 'appId': context.appId, 'input_dir': input_dir,
+        'work_dir': work_dir, 'output_dir': output_dir
+      })
+    }
+
+    proc.stderr.on('data', function (data) {
+      logger.debug(data.toString());
+      console.log(data.toString());
+    });
+
+    proc.stdout.on('data', function (data) {
+      logger.debug(data.toString());
+      console.log(data.toString());
+    });
+
+    proc.on('exit', function (code) {
+      logger.debug('Process exited with code', code);
+    });
   }
 
-  // "submit" job (start the handler process)
-  var proc = spawn(cmd, ['../../../hyperflow-job-executor/jobexec.js', context.taskId, context.redis_url], {shell: true});
-
-  proc.stderr.on('data', function(data) {
-    logger.debug(data.toString());
-    console.log(data.toString());
-  });
-
-  proc.stdout.on('data', function(data) {
-    logger.debug(data.toString());
-    console.log(data.toString());
-  });
-
-  proc.on('exit', function(code) {
-    logger.debug('Process exited with code', code);
-  });
+  if (process.env.HF_VAR_ENABLE_TRACING === "1") {
+    tracer.startActiveSpan('submitRemoteJob', span => {
+      //console.log(ins.map(i => i));
+      var trace_id = span.spanContext().traceId
+      var parent_id = span.spanContext().spanId
+      runJob(trace_id, parent_id, span)
+      span.end();
+    });
+  } else {
+    runJob()
+  }
 
   // send message to the job (command to be executed)
   try {
     await context.sendMsgToJob(jobMessage, context.taskId);
     logger.info('[' + context.taskId + '] job message sent');
-  } catch(err) {
+  } catch (err) {
     console.error(err);
     throw err;
   }
@@ -74,10 +100,10 @@ async function submitRemoteJob(ins, outs, context, cb) {
     logger.info('[' + context.taskId + '] job result received:', jobResult);
     console.log('Received job result:', jobResult);
     cb(null, outs);
-  } catch(err) {
+  } catch (err) {
     console.error(err);
     throw err;
   }
 }
 
-exports.submitRemoteJob = submitRemoteJob;
+exports.submitRemoteJob = submitRemoteJob;

examples/RemoteJobs/handler.js

@@ -1,4 +1,5 @@
 // Executor of 'jobs' using the Redis task status notification mechanism
+process.env.HF_VAR_ENABLE_TRACING  === "1"  && require("../../tracing.js")("hyperflow-service");
 const redis = require('redis');
 const { spawn } = require('child_process');
 

functions/k8sCommand.js

@@ -4,7 +4,7 @@ const k8s = require('@kubernetes/client-node');
 const yaml = require('js-yaml');
 var fs = require('fs');
 
-async function k8sCommand(ins, outs, context, cb) {
+async function k8sCommand(ins, outs, context, cb, parentId, traceId) {
 
   let handlerStart = Date.now();
   console.log("[DEBUG] K8sInvoke called.");
@@ -29,7 +29,7 @@ async function k8sCommand(ins, outs, context, cb) {
 
   const k8sApi = kc.makeApiClient(k8s.BatchV1Api);
   try {
-    var command = 'hflow-job-execute ' + context.taskId + ' ' + context.redis_url;
+    var command = 'hflow-job-execute ' + context.taskId + ' ' + context.redis_url + ' ' + parentId + ' ' + traceId;
     var containerName = process.env.HF_VAR_WORKER_CONTAINER;
     var volumePath = '/work_dir';
     var jobName = Math.random().toString(36).substring(7) + '-' + context.name.replace(/_/g, '-') + "-" + context.procId;

functions/kubernetes/k8sCommand.js

@@ -1,5 +1,6 @@
 // Runs a job as a Pod (Kubernetes Job) in a Kubernetes cluster
 
+const tracer = process.env.HF_VAR_ENABLE_TRACING  === "1" ? require("../../tracing.js")("hyperflow-kubernetes") : undefined;
 const k8s = require('@kubernetes/client-node');
 var BufferManager = require('./buffer_manager.js').BufferManager;
 var RestartCounter = require('./restart_counter.js').RestartCounter;
@@ -125,18 +126,39 @@ async function k8sCommandGroup(bufferItems) {
   kubeconfig.loadFromDefault(); // loadFromString(JSON.stringify(kconfig))
 
   let jobExitCodes = [];
-  try {
+
+  const executeJobs = async (span) => {
     if (getExecutorType(context) === "WORKER_POOL") {
       await amqpEnqueueJobs(jobArr, taskIdArr, contextArr, customParams)
     } else {
-      await submitK8sJob(kubeconfig, jobArr, taskIdArr, contextArr, customParams)
+      var traceId = span && span.spanContext().traceId
+      var parentId = span && span.spanContext().spanId
+      await submitK8sJob(kubeconfig, jobArr, taskIdArr, contextArr, customParams, parentId, traceId)
     }
     jobExitCodes = await synchronizeJobs(jobArr, taskIdArr, contextArr, customParams, restartFn);
-  } catch (err) {
-    console.log("Error when submitting job:", err);
-    throw err;
   }
 
+  if (process.env.HF_VAR_ENABLE_TRACING  === "1" && tracer){
+    tracer.startActiveSpan('k8sCommands', async span => {
+      try {
+        await executeJobs();
+      } catch (err) {
+        console.log("Error when submitting job:", err);
+        throw err;
+      } finally {
+        span.end();
+      }
+    });
+  } else {
+    try {
+      await executeJobs();
+    } catch (err) {
+        console.log("Error when submitting job:", err);
+        throw err;
+    }
+  }
+
+
   let endTime = Date.now();
   console.log("Ending k8sCommandGroup function, time:", endTime, "exit codes:", jobExitCodes);
 

functions/kubernetes/k8sJobSubmit.js

@@ -51,7 +51,7 @@ function createK8sJobMessage(job, taskId, context) {
 //
 // Returns:
 // - jobYaml: string with job YAML to create the k8s job
-var createK8sJobYaml = (job, taskIds, context, jobYamlTemplate, customParams) => {
+var createK8sJobYaml = (job, taskIds, context, jobYamlTemplate, customParams, parentId, traceId) => {
   let quotedTaskIds = taskIds.map(x => '"' + x + '"');
   var command = 'hflow-job-execute ' + context.redis_url + ' -a -- ' + quotedTaskIds.join(' ');
   var containerName = job.image || process.env.HF_VAR_WORKER_CONTAINER;
@@ -81,7 +81,9 @@ var createK8sJobYaml = (job, taskIds, context, jobYamlTemplate, customParams) =>
     restartPolicy: restartPolicy, backoffLimit: backoffLimit,
     experimentId: context.hfId + ":" + context.appId,
     workflowName: context.wfname, taskName: job.name,
-    appId: context.appId, workingDirPath: workingDirPath
+    appId: context.appId, workingDirPath: workingDirPath,
+    enableTracing: process.env.HF_VAR_ENABLE_TRACING, optParentId: parentId,
+    optTraceId: traceId
   }
 
   // Add/override custom parameters for the job
@@ -105,7 +107,7 @@ var createK8sJobYaml = (job, taskIds, context, jobYamlTemplate, customParams) =>
 //
 //
 // Returns: job exit code
-var submitK8sJob = async(kubeconfig, jobArr, taskIdArr, contextArr, customParams) => {
+var submitK8sJob = async(kubeconfig, jobArr, taskIdArr, contextArr, customParams, parentId, traceId) => {
 
   // Load definition of the the worker job pod
   // File 'job-template.yaml' should be provided externally during deployment
@@ -116,7 +118,7 @@ var submitK8sJob = async(kubeconfig, jobArr, taskIdArr, contextArr, customParams
   let context = contextArr[0]; 
 
   // CAUTION: When creating job YAML first job details (requests, container) are used.
-  var jobYaml = createK8sJobYaml(jobArr[0], taskIdArr, contextArr[0], jobYamlTemplate, customParams);
+  var jobYaml = createK8sJobYaml(jobArr[0], taskIdArr, contextArr[0], jobYamlTemplate, customParams, parentId, traceId);
   let jobMessages = [];
   for (var i=0; i<jobArr.length; i++) {
     let job = jobArr[i];