[kafkajs]: Instrument Producer

[bizob2828]

Description

We should add instrumentation to support the producing messages based on the docs

Producer

Sending a message with the producer MUST be wrapped in a trace.

Trace duration MUST include time of the following:

• Serialize the message
• Send the message
  Trace duration MAY include time of the following:
• Sort the message

In async "sending the message" means adding it to the queue to be sent out later.
In sync "sending the message" means actually sending the message to the consumer.

Trace Name

The trace name MUST be in the following format:
MessageBroker/Kafka/Topic/Produce/Named/{topic_name}

Serialization Metrics

Because serializing and deserializing is such a common thing to do with Kafka, many
implementations include this as an option directly on the producer/consumer or as a
specialized Serializingproducer Deserializingconsumer. It very useful to capture the
time it takes to serialize the producer's message as there may be significant slow downs
due to this type of operation. A message consists of both a key and a value which are
serialized independently. If serialization monitoring is possible, the following metrics
MUST be captured:

• MessageBroker/Kafka/Topic/Named/{topic_name}/Serialization/Value
• MessageBroker/Kafka/Topic/Named/{topic_name}/Serialization/Key

Additional Context

• The serialization metrics may not be possible with kafkajs.
• We should be able to use the message-shim recordProduce method with the appropriate spec

[workato-integration]

https://new-relic.atlassian.net/browse/NR-274405

[jsumners-nr]

The library does not provide hook points for serialization:

1. Serialization of keys is either a simple Buffer.from on a string, or requires the user to encode the key themselves and provide it as a Buffer instance. See the conversation in Add support for keys other than strings tulios/kafkajs#1544
2. Serialization of values is handled by various methods on https://github.com/tulios/kafkajs/blob/master/src/protocol/encoder.js and happens through an invocation of a broker send method (https://github.com/tulios/kafkajs/blob/55b0b416308b9e597a5a6b97b0a6fd6b846255dc/src/broker/index.js#L241). As an example, the stub test we have started with follows the path broker > protocol/v7 > protocol/v6 > protocol/v5 > protocol/v3 > https://github.com/tulios/kafkajs/blob/55b0b416308b9e597a5a6b97b0a6fd6b846255dc/src/protocol/requests/produce/v3/request.js#L44-L54.

Given the way the library abstracts the sending of data, we are likely to instrument the sendMessages (https://github.com/tulios/kafkajs/blob/55b0b416308b9e597a5a6b97b0a6fd6b846255dc/src/producer/sendMessages.js) function or one level up, the sendBatch method (https://github.com/tulios/kafkajs/blob/55b0b416308b9e597a5a6b97b0a6fd6b846255dc/src/producer/messageProducer.js#L54).

[bizob2828]

@jsumners-nr I think that's fine. The span will account for all the internal processing. I'm not sure if our message shim will capture these by default

MessageBroker/Kafka/Topic/Named/{topic_name}/Serialization/Value
MessageBroker/Kafka/Topic/Named/{topic_name}/Serialization/Key

, they should but hard to tell

[jsumners-nr]

I am saying that we cannot generate those metrics. Which is okay by our spec. It says that we must record them if we are able to.