Jsonata: Support tracing (#68)

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

Author: pierDipi

transform-jsonata/README.md

@@ -20,6 +20,14 @@ pnpm dev
 # or pnpm dev-kubevirt to inject a different example transformation
 ```
 
+### Tracing
+
+```shell
+docker run --rm -d -p 9411:9411 openzipkin/zipkin
+
+pnpm dev-zipkin
+```
+
 ## Building
 
 Assuming current working directory is `transform-jsonata`

transform-jsonata/jsonata.js

@@ -5,6 +5,32 @@ const fs = require('node:fs');
 const fsPromises = require('node:fs').promises;
 const {buffer} = require('node:stream/consumers');
 
+const {
+    trace,
+    context,
+    SpanStatusCode,
+    propagation,
+    DiagConsoleLogger,
+    DiagLogLevel,
+    diag
+} = require('@opentelemetry/api');
+const {NodeTracerProvider} = require('@opentelemetry/sdk-trace-node');
+const {Resource} = require('@opentelemetry/resources');
+const {
+    ATTR_SERVICE_NAME,
+    ATTR_SERVER_ADDRESS,
+    ATTR_SERVER_PORT,
+    ATTR_URL_SCHEME
+} = require('@opentelemetry/semantic-conventions');
+const {
+    BatchSpanProcessor,
+    ParentBasedSampler,
+    TraceIdRatioBasedSampler, ConsoleSpanExporter
+} = require('@opentelemetry/sdk-trace-base');
+const {ZipkinExporter} = require('@opentelemetry/exporter-zipkin');
+const {W3CTraceContextPropagator, CompositePropagator} = require("@opentelemetry/core");
+const {B3InjectEncoding, B3Propagator} = require("@opentelemetry/propagator-b3");
+
 const port = process.env.PORT = process.env.PORT || 8080;
 const k_sink = process.env.K_SINK || undefined;
 const jsonata_transform_file_name = process.env.JSONATA_TRANSFORM_FILE_NAME || undefined;
@@ -14,6 +40,8 @@ const jsonata_response_transform_file_name = process.env.JSONATA_RESPONSE_TRANSF
 
 const jsonata_discard_response_body = process.env.JSONATA_DISCARD_RESPONSE_BODY === "true" || false;
 
+const jsonata_config_tracing = JSON.parse(process.env.JSONATA_CONFIG_TRACING || '{}')
+
 const oidc_token_file = process.env.OIDC_TOKEN_FILE || undefined
 if (oidc_token_file && !fs.existsSync(oidc_token_file)) {
     console.info(`${oidc_token_file} file doesn't exist, token will not be forwarded to K_SINK endpoint (if specified)`);
@@ -57,7 +85,98 @@ function logDebug(...inputs) {
     }
 }
 
+const w3cPropagator = new W3CTraceContextPropagator();
+const propagator = new CompositePropagator({
+    propagators: [
+        w3cPropagator,
+        new B3Propagator({
+            injectEncoding: B3InjectEncoding.MULTI_HEADER
+        }),
+        new B3Propagator({
+            injectEncoding: B3InjectEncoding.SINGLE_HEADER,
+        })
+    ],
+})
+
+if (process.env.NODE_ENV === "development") {
+    // Enable OpenTelemetry debug logging
+    diag.setLogger(new DiagConsoleLogger(), DiagLogLevel.DEBUG);
+}
+
+logDebug(jsonata_config_tracing)
+
+let exporter = undefined
+if ("zipkin-endpoint" in jsonata_config_tracing &&
+    jsonata_config_tracing['zipkin-endpoint'] !== "" &&
+    'backend' in jsonata_config_tracing &&
+    jsonata_config_tracing.backend === 'zipkin') {
+
+    console.info("Using zipkin tracing exporter")
+    exporter = new ZipkinExporter({
+        url: jsonata_config_tracing['zipkin-endpoint'],
+        serviceName: 'transform-jsonata',
+    })
+} else {
+    console.info("Using console tracing exporter")
+    exporter = new ConsoleSpanExporter()
+}
+
+const batchSpanProcessor = new BatchSpanProcessor(exporter, {
+    maxQueueSize: 1000, // Maximum queue size. After this, spans are dropped
+    maxExportBatchSize: 100, // Maximum batch size of every export
+    scheduledDelayMillis: 5000, // Delay interval between two consecutive exports
+    exportTimeoutMillis: 30000, // How long the export can run before it is cancelled
+});
+
+let sampleRate = undefined
+if ('sample-rate' in jsonata_config_tracing) {
+    sampleRate = Number.parseFloat(jsonata_config_tracing['sample-rate']);
+    console.info(`Tracing sample rate is ${sampleRate}`)
+}
+
+const traceProvider = new NodeTracerProvider({
+    resource: new Resource({
+        [ATTR_SERVICE_NAME]: 'transform-jsonata',
+    }),
+    spanProcessors: [batchSpanProcessor],
+    sampler: new ParentBasedSampler({
+        root: new TraceIdRatioBasedSampler(sampleRate ?? 0),
+    }),
+});
+
+traceProvider.register({propagator: propagator});
+
+// Get a tracer
+const tracer = trace.getTracer('tracer');
+
 const app = express()
+app.use((req, res, next) => {
+    const parentContext = propagation.extract(context.active(), req.headers);
+
+    const span = tracer.startSpan(
+        'http_request',
+        {
+            attributes: {
+                'http.method': req.method,
+                'http.url': req.url,
+                'http.route': req.route?.path,
+            },
+        },
+        parentContext
+    );
+
+    // Store the span in the context
+    return context.with(trace.setSpan(context.active(), span), () => {
+        // Add a callback to end the span when the response is finished
+        res.on('finish', () => {
+            span.setAttributes({
+                'http.status_code': res.statusCode,
+            });
+            span.end();
+        });
+        next();
+    });
+});
 app.use(express.json());
 app.use(express.text());
 app.use(express.raw({type: '*/*'}));
@@ -75,7 +194,16 @@ app.use((req, res, next) => {
     next();
 });
 
+const headerSetter = {
+    set: (carrier, key, value) => {
+        carrier[key] = value;
+    }
+};
+
 app.post("/", async (req, res) => {
+    const processSpan = tracer.startSpan('process_request');
+    const processSpanContext = trace.setSpan(context.active(), processSpan)
+
     try {
         let input = null
         try {
@@ -119,12 +247,41 @@ app.post("/", async (req, res) => {
             }
         }
 
-        const response = await fetch(k_sink, {
-            method: "POST",
-            headers: k_sink_request_headers,
-            body: JSON.stringify(transformed),
-            redirect: 'error',
-        })
+        const k_sink_url = new URL(k_sink)
+        const kSinkSendSpan = tracer.startSpan('k_sink_send', {
+            attributes: {
+                [ATTR_URL_SCHEME]: k_sink_url.protocol.endsWith(':') ? k_sink_url.protocol.substring(0, k_sink_url.protocol.length-1) : k_sink_url.protocol,
+                [ATTR_SERVER_ADDRESS]: k_sink_url.hostname,
+                [ATTR_SERVER_PORT]: k_sink_url.port,
+            }
+        }, processSpanContext);
+
+        const response = await context.with(
+            trace.setSpan(context.active(), kSinkSendSpan),
+            async () => {
+                try {
+                    w3cPropagator.inject(context.active(), k_sink_request_headers, headerSetter)
+
+                    const result = await fetch(k_sink, {
+                        method: "POST",
+                        headers: k_sink_request_headers,
+                        body: JSON.stringify(transformed),
+                        redirect: 'error',
+                    })
+                    kSinkSendSpan.setAttributes({
+                        'http.status_code': result.status,
+                        'http.response_content_length': result.headers.get('content-length'),
+                    });
+                    return result;
+                } catch (error) {
+                    kSinkSendSpan.recordException(error);
+                    kSinkSendSpan.setStatus({code: SpanStatusCode.ERROR});
+                    throw error;
+                } finally {
+                    kSinkSendSpan.end();
+                }
+            }
+        );
 
         if (jsonata_discard_response_body) {
             logDebug(`Received response from K_SINK, discarding response body and responding with ${response.status}`)
@@ -192,11 +349,15 @@ app.post("/", async (req, res) => {
             .send(JSON.stringify(transformed_response))
 
     } catch (error) {
+        processSpan.recordException(error)
+        processSpan.setStatus({code: SpanStatusCode.ERROR});
         console.error(error);
         return res
             .header("Reason", error.toString())
             .status(500)
             .send()
+    } finally {
+        processSpan.end();
     }
 });
 
@@ -256,7 +417,11 @@ function shutDownNow() {
 
     server.close(() => {
         console.log('Closed out remaining connections');
-        process.exit(0);
+
+        console.log('Shutting down tracing...');
+        batchSpanProcessor.shutdown().then(() => {
+            process.exit(0);
+        });
     });
 
     setTimeout(() => {

transform-jsonata/package.json

@@ -3,12 +3,21 @@
   "author": "Knative authors",
   "scripts": {
     "dev": "NODE_ENV=development OIDC_TOKEN_FILE=./examples/example-token.txt JSONATA_TRANSFORM_FILE_NAME=./examples/jsonata_transform_identity.jsonata nodemon ./jsonata.js",
-    "dev-kubevirt": "NODE_ENV=development OIDC_TOKEN_FILE=./examples/example-token.txt JSONATA_TRANSFORM_FILE_NAME=./examples/ce_apiserversource_kubevirt.jsonata nodemon ./jsonata.js"
+    "dev-kubevirt": "NODE_ENV=development OIDC_TOKEN_FILE=./examples/example-token.txt JSONATA_TRANSFORM_FILE_NAME=./examples/ce_apiserversource_kubevirt.jsonata nodemon ./jsonata.js",
+    "dev-zipkin": "NODE_ENV=development JSONATA_CONFIG_TRACING='{\"backend\":\"zipkin\", \"zipkin-endpoint\": \"http://localhost:9411/api/v2/spans\", \"sample-rate\":\"1\"}' OIDC_TOKEN_FILE=./examples/example-token.txt JSONATA_TRANSFORM_FILE_NAME=./examples/jsonata_transform_identity.jsonata nodemon ./jsonata.js"
   },
   "dependencies": {
     "jsonata": "^2.0.6",
     "cloudevents": "^8.0.2",
-    "express": "^4.21.2"
+    "express": "^4.21.2",
+    "@opentelemetry/api": "1.9.0",
+    "@opentelemetry/core": "^1.30.1",
+    "@opentelemetry/sdk-trace-node": "^1.30.1",
+    "@opentelemetry/resources": "^1.30.1",
+    "@opentelemetry/semantic-conventions": "^1.30.0",
+    "@opentelemetry/sdk-trace-base": "^1.30.1",
+    "@opentelemetry/exporter-zipkin": "^1.30.1",
+    "@opentelemetry/propagator-b3": "^1.30.1"
   },
   "devDependencies": {
     "nodemon": "^3.1.9",