feat(kafka): add MessageBroker/Kafka/Cluster/{id}/Produce|Consume/{topic} metrics - #4044
feat(kafka): add MessageBroker/Kafka/Cluster/{id}/Produce|Consume/{topic} metrics#4044shashank-reddy-nr wants to merge 21 commits into
Conversation
|
|
|
|
|
@shashank-reddy-nr do you mind sharing the context of this PR. I see it's adding a bunch of attributes and metrics but what is driving this? |
|
Hi @bizob2828, I made code changes to add a couple of new Kafka metrics to establish a relationship between our self-hosted Kafka and the APM. Here is the work request I raised for the APM team to review the changes and design, which has the complete context and explains why this is needed. |
|
Please update the PR description here with details on why this PR is needed and how this PR solves those needs. |
There was a problem hiding this comment.
Why are the changes in this file not part of the exported class? Are the two caches really necessary? What is the expected size of these caches in a typical, or extreme, deployment scenario?
There was a problem hiding this comment.
Hi @jsumners-nr, I'm new to this repo and leaned heavily on AI assistance to implement this, so I apologize upfront for anything that doesn't follow conventions here.
The two caches serve different lifecycles:
cluster-id-cache.jsis a module-level map (brokers string → cluster UUID). It's intentionally process-global because the cluster UUID is the same for every producer/consumer connecting to the same Kafka cluster - it makes no sense to refetch it per client instance._clusterIdByInstancedoesn't exist here - the only per-instance storage is via the existing kafkaCtx symbol already set on each client/consumer.
In a typical deployment you'd have 1–3 distinct Kafka clusters, so the cache holds 1–3 entries. In an extreme multi-tenant scenario (many distinct broker strings) it could grow unbounded - that's a fair concern. Happy to add a size cap or TTL if you'd prefer.
Open to moving the fetch logic inside the class if that's the convention - I'll follow your guidance. Can you please guide me?
There was a problem hiding this comment.
There will only be one instance of ConstructorSubscriber for the duration of an application's lifecycle. It can manage any required caches.
If it is possible for the cache to grow infinitely, then a size limited LRU cache should be used. We do not currently include such cache functionality. So if this is really a necessary feature, we will need to implement lib/lru-map-cache.js.
There was a problem hiding this comment.
Addressed in the latest commit (3b6e628): the two maps are now #clusterIdCache and #clusterIdInFlight private fields on ConstructorSubscriber itself.
On LRU vs hard cap: I opted for a simpler MAX_CACHE_SIZE = 128 constant rather than a full LRU implementation. The rationale is that each cache entry represents one distinct broker list (i.e. one Kafka cluster). In practice, a single Node.js process connects to a small number of clusters — typically 1–3, rarely more than 10. 128 is already generous for any realistic deployment. If the cap is ever hit, new broker-sets simply do not get cached (we silently skip the cache.set) but still fetch the cluster ID on first call and return it — so the metric still fires, we just do not avoid the re-fetch for that broker-set.
If you would prefer a true LRU, I am happy to implement lib/lru-map-cache.js — just let me know and I will add it as part of this PR.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #4044 +/- ##
==========================================
- Coverage 97.52% 97.45% -0.07%
==========================================
Files 522 526 +4
Lines 62930 63194 +264
Branches 1 1
==========================================
+ Hits 61374 61588 +214
- Misses 1556 1606 +50
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
5190150 to
3a6e6ae
Compare
jsumners-nr
left a comment
jsumners-nr
left a comment
There was a problem hiding this comment.
A few more comments subsequent to the revised changes.
Is there a spec PR number to review that defines what these changes are trying to accomplish?
There was a problem hiding this comment.
There will only be one instance of ConstructorSubscriber for the duration of an application's lifecycle. It can manage any required caches.
If it is possible for the cache to grow infinitely, then a size limited LRU cache should be used. We do not currently include such cache functionality. So if this is really a necessary feature, we will need to implement lib/lru-map-cache.js.
| * @param {boolean} batch Whether this is a sendBatch call. | ||
| * @param {object} data The send/sendBatch arguments object. | ||
| */ | ||
| function recordClusterProduceMetrics(metrics, clusterId, batch, data) { |
There was a problem hiding this comment.
Why isn't this function defined a file within the utils/ directory?
There was a problem hiding this comment.
Moved to lib/subscribers/kafkajs/utils/cluster-id-cache.js. It exports two pure functions (getClusterIdFromCache, fetchAndCacheClusterId) that accept the cache maps as parameters so they remain independently testable.
| const rawBrokers = args[0]?.brokers | ||
| client[kafkaCtx] = { brokers: typeof rawBrokers === 'function' ? [] : (rawBrokers ?? ['none']) } | ||
|
|
||
| _fetchAndCacheClusterId(client, client[kafkaCtx].brokers).then((id) => { |
There was a problem hiding this comment.
Why does this function have a leading _? Typically, in JavaScript, the convention is that things with a leading _ are to indicate they are private to the containing entity. This convention is really not necessary any longer, and it's not clear to me why it would have ever been needed for this function.
Also, why is it promise based? The end function is a synchronous function. By using a promise here, we are introducing activity happening outside of the context in which end is invoked.
There was a problem hiding this comment.
Both addressed in the latest commit:
- The leading
_is gone. - The function is now in
lib/subscribers/kafkajs/utils/cluster-id-cache.js.
On the async shape: AdminClient.connect() and AdminClient.describeCluster() are inherently async network operations, so there is no synchronous alternative. We fire-and-forget from end() and set client[kafkaCtx].clusterId when the promise resolves. The first few sends before the cluster ID arrives will not record the cluster metric, but that is intentional best-effort behaviour — the metric fires for every send after the ID is populated.
| const { arguments: args, self: client } = data | ||
| client[kafkaCtx] = { brokers: args[0].brokers ?? ['none'] } | ||
| const rawBrokers = args[0]?.brokers | ||
| client[kafkaCtx] = { brokers: typeof rawBrokers === 'function' ? [] : (rawBrokers ?? ['none']) } |
There was a problem hiding this comment.
What is this change doing? I don't see rawBrokers documented anywhere on https://kafka.js.org/docs/configuration as being an option for the client constructor.
There was a problem hiding this comment.
Removed. The line now reads:
const brokers = args[0]?.brokers
client[kafkaCtx] = { brokers: typeof brokers === 'function' ? [] : (brokers ?? ['none']) }This handles both array brokers and the function-based brokers form that kafkajs supports (if brokers is a function we store [] and skip the admin fetch, since we cannot sort/key by a function).
| // Capture clientId at consumer creation so it is available on every | ||
| // transaction without waiting for the async REQUEST event. | ||
| consumer[kafkaCtx] = { | ||
| ...client[kafkaCtx], | ||
| clientId: args[0]?.clientId ?? client[kafkaCtx]?.clientId ?? null | ||
| } |
There was a problem hiding this comment.
Please remove the copy operation. It's cheaper and cleaner to:
consumer[kafkaCtx] = client[kafkaCtx]
consumer[kafkaCtx].clientId = one ? or : the_otherThere was a problem hiding this comment.
Fixed — shared reference assignment now:
consumer[kafkaCtx] = client[kafkaCtx]
consumer[kafkaCtx].clientId = args[0]?.clientId ?? client[kafkaCtx]?.clientId ?? nullThis also means clusterId populated asynchronously on client[kafkaCtx] is visible on consumer[kafkaCtx] immediately.
| // REQUEST event may refine clientId from the live connection payload | ||
| if (data?.payload?.clientId) { | ||
| consumer[kafkaCtx].clientId = data.payload.clientId | ||
| } |
There was a problem hiding this comment.
This pattern is what the nullish coalescing assignment operator is for:
consumer[kafkaCtx] ??= data?.payload?.clientIdThere was a problem hiding this comment.
Fixed — updated to consumer[kafkaCtx].clientId ??= data?.payload?.clientId.
| if (instance[kafkaCtx].clusterId) { | ||
| const [data] = args | ||
| recordClusterProduceMetrics(self.agent.metrics, instance[kafkaCtx].clusterId, batch, data) | ||
| } |
There was a problem hiding this comment.
We are not in a transaction here. What is the purpose of recording metrics in this case?
There was a problem hiding this comment.
These use self.agent.metrics — the agent-level metric aggregator — not the per-transaction tx.metrics. The distinction matters: Kafka producers frequently run outside of any active web/background transaction (fire-and-forget background producers, CLI scripts, etc.). Using the global aggregator means the cluster-level produce count is always captured regardless of transaction context.
The consume side already does the same thing for the same reason: agentMetrics.getOrCreateMetric(...) in record-data-metrics.js:50. This PR follows the same pattern for the produce side.
@jsumners-nr I just raised a PR for agent kafka spec update and I will get it approved from APM agent board and let you know to review this PR post approval. For reference here is the #817. |
…record-data-metrics Cover the new recordClusterProduceMetrics utility (send and sendBatch paths) and the new agentMetrics/clusterId branch added to recordDataMetrics. Resolves the codecov/patch failure on PR newrelic#4044. Assisted-by: Claude Sonnet 4.6
Records per-cluster Kafka metrics (MessageBroker/Kafka/Cluster/{clusterId}/Topic/{topic}/Produce
and MessageBroker/Kafka/Cluster/{clusterId}/Topic/{topic}/Consume) to let customers track
throughput broken out by Kafka cluster, not just by topic. The cluster ID is fetched
once per unique broker set via a background AdminClient call; the hot produce/consume path
has no additional overhead. The feature is always-on and best-effort — it does not inject
anything into Kafka wire headers and does not add span or custom attributes.
Assisted-by: Claude Sonnet 4.6 <noreply@anthropic.com>
- Revert kafkajs_instrumentation flag to false (separate discussion needed) - Move recordClusterProduceMetrics to utils/record-cluster-produce-metrics.js - Rename _fetchAndCacheClusterId to fetchAndCacheClusterId (not private) - Rename rawBrokers local var to brokers to avoid confusion with kafkajs config - Consumer context: use direct reference instead of spread; set clientId separately - REQUEST event handler: use ??= for clientId update - Batch consume metrics: use incrementCallCount(length) instead of a loop Assisted-by: Claude Sonnet 4.6 <noreply@anthropic.com>
The Map instances (_clusterIdCache, _clusterIdInFlight) now live as private fields on ConstructorSubscriber, which is a process-lifetime singleton. cluster-id-cache.js functions accept the maps as parameters (dependency injection) so they remain independently testable. Adds a hard cap of 128 entries to the cache to prevent any theoretical unbounded growth; in practice the cache holds one entry per distinct Kafka cluster the application connects to (bounded by infrastructure config, not runtime data). Removes the getClusterIdFromCache fallback from record-data-metrics.js — the shared-reference approach (consumer[kafkaCtx] = client[kafkaCtx]) means clusterId is visible to all consumers once the async fetch resolves. Also fixes the cluster-id-cache tests which were importing the old _fetchAndCacheClusterId name and relying on module-level map state; tests now inject fresh maps per test. Assisted-by: Claude Sonnet 4.6
…t errors - Add missing @PARAM description for `brokers` in getClusterIdFromCache and fetchAndCacheClusterId (jsdoc/require-param-description) - Wrap ternary expressions in arrow functions with parens (no-confusing-arrow) - Put each argument on its own line in multi-line fetchAndCacheClusterId calls in tests (function-call-argument-newline) Assisted-by: Claude Sonnet 4.6
…ructorSubscriber Verifies that end() correctly extracts brokers and propagates cluster IDs across all kafkajs auth configurations: - SASL/PLAIN - SASL/SCRAM-SHA-256 and SCRAM-SHA-512 - SASL/OAUTHBEARER (with async token provider callback) - mTLS/SSL (object and boolean forms) - Function-based broker resolvers (cluster ID fetch is skipped gracefully) Also asserts that admin() is called with no arguments, confirming auth is inherited via kafkajs' internal CREATE_CLUSTER closure and not passed by our instrumentation code. Assisted-by: Claude Sonnet 4.6
Two bugs caused cluster IDs to be served stale forever for long-lived clients:
1. cluster-id-cache.js stored entries as plain strings; there was no expiry
check, so entries were never refreshed after the first successful fetch.
Fix: entries are now stored as { clusterId, fetchedAt } objects; both
getClusterIdFromCache and fetchAndCacheClusterId check TTL_MS (30 min).
2. client-constructor.js hot paths (eachBatch, send, sendBatch no-tx) used
`instance[kafkaCtx]?.clusterId ?? getClusterIdFromCache(...)` — the ??
short-circuit caused getClusterIdFromCache (the TTL-aware path) to be
skipped entirely once kafkaCtx.clusterId was set. Long-lived instances
never triggered re-fetches after TTL expiry.
Fix: call getClusterIdFromCache first; trigger fetchAndCacheClusterId when
it returns undefined (TTL expired or absent); fall back to the stale
kafkaCtx.clusterId during the re-fetch.
fetchAndCacheClusterId requires the kafkajs Client instance, which was not
previously available on the hot path. Fix: store a _kafkaClient reference
on kafkaCtx at constructor time so producer/consumer wrappers can call it.
Assisted-by: Claude Sonnet 4.6
Assisted-by: Claude Sonnet 4.6
When the 5-second timeout fires before admin.connect() resolves, Promise.race rejects and we enter the catch block. The original code called admin.disconnect() in catch, but connect() was still pending in the background. If it resolved after our disconnect(), the admin was left in a connected state with no corresponding disconnect(), leaking a TCP socket indefinitely. Fix: capture connectPromise before the race. In the catch block, save the admin reference, null out admin, then chain a .then() on connectPromise so that if connect() resolves after the timeout, we disconnect the now-live socket. The existing `await pendingAdmin.disconnect()` handles the case where connect() already resolved before we reach catch. KafkaJS handles double-disconnect gracefully, so both paths are safe. Assisted-by: Claude Sonnet 4.6
- Reduce cognitive complexity of nrWrappedMethod (19→5) by extracting #insertProduceDTHeaders and #refreshAndRecordProduceMetrics helpers; satisfies sonarjs/cognitive-complexity ≤ 15 rule. - Fix cluster-metric versioned tests broken by the cache-first lookup introduced in 0219d36: tests were injecting a fake clusterId via kafkaCtx.clusterId, but getClusterIdFromCache now takes precedence and returns the real cluster ID set at constructor time. Poll for the real cluster ID in beforeEach and use it in both the produce and consume cluster-metric assertions. Assisted-by: Claude Sonnet 4.6
Add kafka_cluster_metrics feature flag (default: false). All fetchAndCacheClusterId calls and cluster metric recording paths are now gated so the feature is opt-in. Assisted-by: Claude Sonnet 4.6
Assisted-by: Claude Sonnet 4.6
…d tests Both tests created an agent with only kafkajs_instrumentation: true, leaving kafka_cluster_metrics as false (the default). The ConstructorSubscriber guards fetchAndCacheClusterId behind kafka_cluster_metrics, so admin() was never called in unit tests and the cluster ID was never resolved in versioned tests, causing every cluster-metric assertion to fail. Assisted-by: Claude Sonnet 4.6
…record-data-metrics Cover the new recordClusterProduceMetrics utility (send and sendBatch paths) and the new agentMetrics/clusterId branch added to recordDataMetrics. Resolves the codecov/patch failure on PR newrelic#4044. Assisted-by: Claude Sonnet 4.6
Assisted-by: Claude Sonnet 4.6
…uster-metrics branches
Adds six unit tests for the previously uncovered lines in
client-constructor.js:
- producer.send() with kafka_cluster_metrics flag off → no metric
- producer.send() with cached cluster ID → records Produce metric
- producer.sendBatch() with cached cluster ID → records per-topic Produce metrics
- producer.send() with in-flight fetch + kafkaCtx.clusterId fallback → records via fallback
- producer.send() with no cache and no _kafkaClient → no metric
- consumer.run({ eachBatch }) with active transaction + cached cluster ID → records Consume metric per message
Introduces makeMetrics() (Map-backed stub), makeSubscriberFull()
(subscriber with full tracer/metrics mocks), and setupWithCache()
helpers scoped to the new test group.
Assisted-by: Claude Sonnet 4.6
Rename destructured kafka_cluster_metrics parameter to kafkaClusterMetrics (camelCase), use block-body arrow function for logger.child(), and change non-interpolated template literal to single-quoted string. Assisted-by: Claude Sonnet 4.6
… connection Capture the internal Cluster reference kafkajs itself creates at producer()/consumer() time (via two new subscribers hooking the internal factory modules) and read cluster.brokerPool.metadata.clusterId directly. This is a synchronous, in-memory, zero-network-call read, replacing the previous approach of opening a separate AdminClient connection with its own TTL/eviction/in-flight-dedup cache to amortize that cost. No cache needed since the read itself is now free. Assisted-by: Claude Sonnet 5
…args recordClusterProduceMetrics ran before kafkajs's own argument validation and indexed into messages arrays unconditionally, so a malformed send()/sendBatch() call (missing messages) threw synchronously out of the instrumented call site instead of failing the way kafkajs itself would. Assisted-by: Claude Sonnet 5
faf8b9a to
d1adeed
Compare
MessageBroker/Kafka/Cluster/{clusterId}/[Produce|Consume]/{topic} now
matches the existing Nodes/{broker}/[Produce|Consume]/{topic} convention
(verb before topic, not after) — per agent-specs newrelic#817 review.
No aggregation rework needed: metrics.getOrCreateMetric(...).incrementCallCount()
already merges into the MetricAggregator's in-memory Metrics table and
only flushes once per 60s harvest cycle, exactly like the existing Nodes
metric. The wire-level data was already aggregated per reporting
interval, not per message.
Assisted-by: Claude Sonnet 5
shashank-reddy-nr
changed the title
…ned test
kafkajs's `cluster.connect()` only opens the connection to the seed
broker — it never fetches cluster metadata (and thus never populates
clusterId). Metadata is only fetched lazily as a side effect of the
first produce/consume call. The versioned test's beforeEach polled
`readClusterId(producer)` right after `connect()`, so the poll always
timed out and `ctx.nr.clusterId` was always `null`, causing both
cluster-metric tests to assert against a bogus expected metric name
("...Cluster/null/...").
Explicitly call `.metadata({ topics: [] })` on the producer's captured
Cluster reference in beforeEach so it's warm before any assertions
read it, matching what a real send/consume call would trigger anyway.
Assisted-by: Claude Sonnet 5
…n/gh-docker-logs) versioned (22.x, 5) and versioned (24.x, 5) failed in the "Set up job" step downloading a third-party action archive from codeload.github.com, before any repo code ran. versioned (26.x, 5) — same test file, same fix — passed cleanly in the same run, confirming this is unrelated CI infra flakiness. Assisted-by: Claude Sonnet 5
| reverse_naming_rules: false, | ||
| unresolved_promise_cleanup: true, | ||
| kafkajs_instrumentation: false, | ||
| kafka_cluster_metrics: false, |
There was a problem hiding this comment.
For future eyes. The spec states be gated behind an opt-in configuration kafka.metrics.cluster.metrics.enabled (disabled by default). This is a feature flag, not configuration. it should live in lib/config/default.js
5 participants
Why
Customers running self-hosted Kafka can't see producer → topic → consumer topology in New Relic - there's no relationship linking a producer APM service, a Kafka topic, and a consumer service.
The blocker is that today's instrumentation keys off broker address (
MessageBroker/Kafka/Nodes/{broker_address}/...). That's unique by luck in managed Kafka (MSK, Confluent Cloud), but self-hosted clusters routinely reuse identical broker addresses across VPCs (e.g.kafka-broker-1:9092in both prod and staging), so address can't reliably join a service to a specific cluster's topic.The fix is Kafka's native cluster UUID (KIP-78): stable, globally unique (even for self-hosted), and already present in the broker metadata response. Emitting it in the metric name gives the APM metric and the existing Kafka Topic entity (
kafka.cluster.idtag, from OTel infra / nri-kafka) a shared key, so a relationship rule can join producer → topic → consumer across services.Purely additive — no existing metrics, distributed-tracing behavior, or instrumentation is changed or removed.
Summary
Adds two new agent-level timeslice metrics for Kafka produce/consume:
MessageBroker/Kafka/Cluster/{clusterId}/Produce/{topic}MessageBroker/Kafka/Cluster/{clusterId}/Consume/{topic}These complement the existing
Message/Kafka/Topic/Named/{topic}/Received/*metrics by adding the cluster UUID dimension, enabling cluster-level traffic segmentation in NRDB.How it works
Cluster ID capture (optimized) — kafkajs builds a brand-new internal
Clusterobject on every single.producer()/.consumer()call — it isn't reachable from the top-levelKafkaclient. Two new subscribers (producer-cluster-capture.js,consumer-cluster-capture.js) hook kafkajs's internal factory modules (src/producer/index.js,src/consumer/index.js) via this repo's existing declarative instrumentation config, capturing thatClusterreference the moment kafkajs itself creates it and stashing it on the returned producer/consumer instance via a newkafkaClusterSymbol.This replaces the original approach of resolving the cluster ID via a separate KafkaJS admin client (
describeCluster()) with its own TTL cache and in-flight-dedup map. The new read (utils/read-cluster-id.js→cluster.brokerPool.metadata.clusterId) is synchronous, in-memory, and best-effort — no extra connection, no cache, no TTL needed, since kafkajs keeps this field current on its own as part of normal operation. This is the same free-read approach used by upstream OpenTelemetry's kafkajs instrumentation. On a producer's very first send (before kafkajs has fetched any metadata), the cluster id may simply be unavailable yet — expected, and it resolves on the next send.Metric recording - Uses
agentMetrics.getOrCreateMetric(...).incrementCallCount()so metrics aggregate at agent level, not per-transaction. Produce metrics fire both inside and outside active transactions. Consume metrics fire fromrecord-data-metrics.jsand from theeachBatchwrapper.Metric name format + why no aggregation rework was needed — per agent-specs #817 review, the metric name now matches the
Nodesshape (verb before topic, not after):Cluster/{id}/Produce/{topic}, notCluster/{id}/Topic/{topic}/Produce. Separately, the spec no longer requires per-message emission — these metrics exist to power entity relationships, not to report exact throughput, so counts just need to be aggregated over the same interval asNodes. No further code change was needed for that:getOrCreateMetric(...).incrementCallCount()already merges into theMetricAggregator's in-memory table and only flushes once per 60s harvest — the same mechanism the existingNodesmetric already uses. Calling it once per message only costs a cheap in-memory counter bump; the data New Relic receives was already one aggregated count per harvest interval, not one per message.How to Test
Unit tests
Manual end-to-end
NEW_RELIC_FEATURE_FLAG_KAFKAJS_INSTRUMENTATION=true NEW_RELIC_FEATURE_FLAG_KAFKA_CLUSTER_METRICS=trueYou should see
MessageBroker/Kafka/Cluster/{uuid}/Produce/{topic}and…/Consume/{topic}entries carrying the real Kafka cluster UUID.Validated end-to-end against a self-hosted Kafka cluster (SASL/PLAIN, two-broker setup). All three topics' Produce metrics appeared with the correct cluster UUID (pre-rename; format updated since — see note above).
Bug fixes
recordClusterProduceMetricsran before kafkajs's own argument validation andindexed into
data.messages/topicMessage.messagesunconditionally. Amalformed
send()/sendBatch()call (missingmessages) threw synchronouslyout of the instrumented call site instead of failing the way kafkajs itself
would (via its own promise rejection). Added guards so a missing/malformed
messagesarray is skipped rather than throwing; added corresponding tests.Related Issues
Jira: NR-572216