How to use applicationinsights when processing EventHubs messages?

[vladiibine]

I have a stack which works like this:

• Python function app 1 sends messages to EventHubs
• Node.js function app 2 gets triggered for batches messages from EventHubs

I want to achieve distributed log correlation via the operation_id field.

What seems to work: The Python function app, I assume, correctly sends tracing data to EventHubs. I assume tracing data is attached correctly because I can see a link between the Python function app and the EventHubs instance in the Azure portal, in the Application Insights Application Map. Moreover, if I inspect the context received by the Node.js function app, I can tracing information in this field: context.bindingData.propertiesArray[<index>]["Diagnostic-Id"] . For each message in the payload, there will be a corresponding Diagnostic-Id here. Diagnostic-ID's look like this: "00-60de00e3807c0132a958f6cddd015503-ce57ff6768dcc932-01". The value 60de00e3807c0132a958f6cddd015503 is the value of the operation_Id field for the log entry emitted by the Python function app.

What I want to do: When processing each of those messages in the batch, if an error is encountered, I want to log a message which will have the same operation_Id as the log messages emitted by the Python function app which pushed that particular message to EventHubs.

What I tried:

// eventhub-nodef1/index.js
// This is the EventHubs trigger handler
const instrumentationKey ='aaaaaaaa-aaaa-4444-aaaa-aaaaaaaaaaaa';

const appInsights = require('applicationinsights');
appInsights
    .setup(instrumentationKey)
    .setDistributedTracingMode(appInsights.DistributedTracingModes.AI_AND_W3C)
    .setAutoCollectPerformance(false);

appInsights.defaultClient.config.samplingPercentage = 100; 

appInsights.start();

const {addEventHubDistributedTracingContext} = require('../poc-common/Common')

const FUNC_NAME = 'poc9-node-trigger';

const theEventhubsTriggerFunc = async function (context, eventHubMessages) {
    let ai = appInsights;
    // I don't expect to be able to correlate the messages logged here
    context.log(`${FUNC_NAME} called for message array ${JSON.stringify(eventHubMessages)}`);
    context.log(`${FUNC_NAME} called with context ${JSON.stringify(context)}`)

    eventHubMessages.forEach(
        (message, index) =>
            addEventHubDistributedTracingContext(context, appInsights, index,
                () => {
                    // but I DO EXPECT to be able to correlate these logged messages
                    context.log(`${FUNC_NAME} processing ${JSON.stringify(message)}`);
                    if (message && message.fail) {
                        context.log(`${FUNC_NAME}: failing because the message was: ${message}`)
                        throw new Error(`${FUNC_NAME} Life's too short to always succeed, ${message.name}`)
                    }
                }
            )
    )
};
module.exports = theEventhubsTriggerFunc

And this is the code for addEventHubDistributedTracingContext

function addEventHubDistributedTracingContext(context, appInsights, msgIndex, callback){
    const diagnosticId = context.bindingData.propertiesArray[msgIndex]['Diagnostic-Id'];
    const diagnosticIdComponents = diagnosticId.split('-');
    const longId = diagnosticIdComponents[1];
    const shortId = diagnosticIdComponents[2];
    const tinyId = diagnosticIdComponents[3];

    // the context for an EventHubs trigger doesn't look like the one for an HTTP trigger
    // I tried to create a fake context, with data from the Diagnostic-ID field, filling in for tracing data that an HTTP trigger would have
    // ...so tried like this and it didn't seem to work
    const spanContextAdapter = {
        traceId: longId,
        spanId: shortId,
        tracestate: tinyId,
    }
    // and also like this and it also didn't seem to work
    const traceparentContextAdapter = {
        traceContext: {
            traceparent: diagnosticId,
            tracestate: shortId,
        }
    }

    // const adapterContext = spanContextAdapter;
    const adapterContext = traceparentContextAdapter;
    const correlationContext = appInsights.startOperation(adapterContext, 'request-string-dunno-hope-it-worx');

    return appInsights.wrapWithCorrelationContext(
        callback, 
        correlationContext
    )();
}

Here's my host.json

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[1.*, 2.0.0)"
  }
}

Any my function.json

{
  "bindings": [
    {
      "type": "eventHubTrigger",
      "name": "eventHubMessages",
      "direction": "in",
      "eventHubName": "poc9-eventhub",
      "connection": "some_connection_name",
      "cardinality": "many",
      "consumerGroup": "$Default"
    }
  ]
}

What happens is that when logging, I just get new tracing information, with a new operation_Id.

So any ideas of what I'm missing? (FYI, javascript is not my main language, please be kind :P )

Thanks a lot :) ^^

[hectorhdzg]

@vladiibine are you using Open Census SDK in Python side?, if you do correlation would be handled automatically in Azure Functions and you can just add code to handle it we have in readme, I'm not seeing the code where Application Insights Node.js SDK is actually sending the telemetry in the code you shared, can you check SDK version property in your query?, just want to be sure the telemetry you are comparing is coming from this SDK and not .NET SDK, I would expect some appInsights track calls here, I believe context.log will end as a trace in .NET SDK