feat: kafkajs instrumentation #2089
Conversation
Codecov ReportAll modified and coverable lines are covered by tests ✅
Additional details and impacted files@@ Coverage Diff @@
## main #2089 +/- ##
==========================================
- Coverage 90.97% 90.47% -0.51%
==========================================
Files 146 149 +3
Lines 7492 7592 +100
Branches 1502 1591 +89
==========================================
+ Hits 6816 6869 +53
- Misses 676 723 +47 |
|
Thank you @seemk |
| | `consumerHook` | `KafkaConsumerCustomAttributeFunction` | Function called before a consumer message is processed. Allows for adding custom attributes to the span. | | ||
| | `moduleVersionAttributeName` | `string` | When set, the kafkajs module version will be added to span attributes under the key of `moduleVersionAttributeName`. | | ||
|
|
||
| ## Useful links |
There was a problem hiding this comment.
There is an initiative to document the status of semantic conventions in all instrumentations #2025
Could you add a section here listing the attributes used by this instrumentation please? You can check the format here https://github.com/open-telemetry/opentelemetry-js-contrib/pull/2088/files#diff-b6fe7e13efb8fc9ee719f8a1650dc2a05540770ec5dfa3679ef2e4ccba8bd509
There was a problem hiding this comment.
Added the attributes, however I noticed that with the current schema, messaging.destination.kind (previously messaging.destination_kind) is removed completely. So I wonder whether we should drop this attribute from the instrumentation to avoid breaking changes in the future.
There was a problem hiding this comment.
Removed the attribute completely
| } | ||
| ); | ||
| const batchMessagePromise: Promise<void> = original!.apply( | ||
| this, |
There was a problem hiding this comment.
inside a fat arrow function this does not refer to the object the original method belongs to. So I wonder if the call is correct.
There was a problem hiding this comment.
Isn't this inherited from the parent scope (in this case the this argument of function eachBatch(this: unknown, ...args))?
There was a problem hiding this comment.
you're right as per MDN docs at the end of section https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Functions/Arrow_functions#cannot_be_used_as_methods says
For similar reasons, the call(), apply(), and bind() methods are not useful when called on arrow functions, because arrow functions establish
thisbased on the scope the arrow function is defined within, and thethisvalue does not change based on how the function is invoked.
|
looks very good now :) please fill the table in the readme as requested in https://github.com/open-telemetry/opentelemetry-js-contrib/pull/2089/files#r1562439567 and you have my 👍 |
|
|
||
| The library contains the following changes compared to the original: | ||
| * The codebase was converted to TypeScript. | ||
| * bufferTextMapGetter compares propagation headers in a case insensitive manner. |
There was a problem hiding this comment.
can also mention removing of the moduleVersionAttributeName option from config, and change the hook format for producerHook and consumerHook
|
|
||
| const module = new InstrumentationNodeModuleDefinition( | ||
| 'kafkajs', | ||
| ['*'], |
There was a problem hiding this comment.
Since the README states that the supported version is <3.0.0, I think it makes sense to also align here, since we can not attempt patching v3 automatically one day which might be breaking for us
| ['*'], | |
| ['< 3'], |
| 'kafkajs', | ||
| ['*'], | ||
| (moduleExports: typeof kafkaJs) => { | ||
| this._diag.debug('Applying patch for kafkajs'); |
There was a problem hiding this comment.
As #2107 merged, can you please also remove the diag print here? thanks
There was a problem hiding this comment.
Removed them, thanks!
| const unpatch = (moduleExports: typeof kafkaJs) => { | ||
| this._diag.debug('Removing patch for kafkajs'); | ||
| if (isWrapped(moduleExports?.Kafka?.prototype.producer)) { | ||
| this._unwrap(moduleExports.Kafka.prototype, 'producer'); | ||
| } | ||
| if (isWrapped(moduleExports?.Kafka?.prototype.consumer)) { | ||
| this._unwrap(moduleExports.Kafka.prototype, 'consumer'); | ||
| } | ||
| }; |
There was a problem hiding this comment.
nit: I know it's not you who wrote it, but it's an optional opportunity to improve.
The unpatch function is defined as a const variable, whereas the patch is defined inline. should these be aligned?
Defining the unpatch this way is not something common in repo, but there are other common patterns to define the unpatch as a class method
There was a problem hiding this comment.
unpatch is used at the beginning of patch function, think making unpatch a member function would help?
|
|
||
| protected init() { | ||
| const unpatch = (moduleExports: typeof kafkaJs) => { | ||
| this._diag.debug('Removing patch for kafkajs'); |
There was a problem hiding this comment.
#2107 removed these prints across the repo, so need to remove it here as well:
| this._diag.debug('Removing patch for kafkajs'); |
| "devDependencies": { | ||
| "@opentelemetry/api": "^1.3.0", | ||
| "@opentelemetry/contrib-test-utils": "^0.38.0", | ||
| "@opentelemetry/sdk-trace-base": "^1.8.0", | ||
| "@types/mocha": "7.0.2", | ||
| "@types/node": "18.6.5", | ||
| "@types/sinon": "^10.0.11", | ||
| "kafkajs": "^2.2.4", | ||
| "mocha": "7.2.0", | ||
| "nyc": "15.1.0", | ||
| "rimraf": "5.0.5", | ||
| "sinon": "15.2.0", | ||
| "ts-mocha": "10.0.0", | ||
| "typescript": "4.4.4" | ||
| }, | ||
| "dependencies": { | ||
| "@opentelemetry/instrumentation": "^0.50.0", | ||
| "@opentelemetry/semantic-conventions": "^1.23.0" | ||
| }, |
There was a problem hiding this comment.
reminder to update the opentelemetry to stable ^1.24.0 and experimental ^0.51.0 before merging
|
Another thought, should this new instrumentation be added to "@opentelemetry/auto-instrumentations-node" as part of this PR, or should it be done separately in a followup PR? |
Added myself, any ideas for the other one? |
Will do it in a followup |
Which problem is this PR solving?
Prerequisite for #1845
Short description of the changes
Adds the kafkajs instrumentation created by Aspecto which does not seem to be maintained any longer.
The codebase has been converted to TypeScript and differs from the original how the propagation headers are read (in this case they are read case insensitively, this fixes propagation issues across language SDKs). Added a NOTICE file attributing the original work and describing the changes.