Extract Kafka cluster ID as tag for DSM#6044
Merged
Merged
Conversation
hokitam
commented
Oct 12, 2023
| producerProps.put(ProducerConfig.METADATA_MAX_AGE_CONFIG, 1000) | ||
| def producerFactory = new DefaultKafkaProducerFactory<String, String>(producerProps) | ||
| def kafkaTemplate = new KafkaTemplate<String, String>(producerFactory) | ||
| kafkaTemplate.flush() |
Contributor
Author
There was a problem hiding this comment.
We are setting the METADATA_MAX_AGE_CONFIG to 1s, such that Metadata updates once every second. We call kafkaTemplate.flush() to ensure that the producer is actually created under the hood (otherwise it wouldn't start updating the Metadata. Sleeping for 1.5s allow us to ensure that the metadata we need in the instrumentation would be updated within the 1s timeframe..
Contributor
There was a problem hiding this comment.
This test may be flaky, depending on how the metadata update is done. Maybe we can check for the data we need in a while loop, until we get the data or some reasonable timeout is reached?
Contributor
Author
There was a problem hiding this comment.
good idea, updated!
BenchmarksStartupParameters
See matching parameters
SummaryFound 0 performance improvements and 0 performance regressions! Performance is the same for 46 metrics, 8 unstable metrics. Startup time reports for petclinicgantt
title petclinic - global startup overhead: candidate=1.26.0-SNAPSHOT~9ff3556f43, baseline=1.26.0-SNAPSHOT~9d5c7ea524
dateFormat X
axisFormat %s
section tracing
Agent [baseline] (1.043 s) : 0, 1042895
Total [baseline] (9.324 s) : 0, 9323566
Agent [candidate] (1.044 s) : 0, 1043792
Total [candidate] (9.379 s) : 0, 9379017
section appsec
Agent [baseline] (1.129 s) : 0, 1129465
Total [baseline] (9.382 s) : 0, 9382162
Agent [candidate] (1.132 s) : 0, 1131921
Total [candidate] (9.393 s) : 0, 9393368
section iast
Agent [baseline] (1.17 s) : 0, 1169614
Total [baseline] (9.67 s) : 0, 9669786
Agent [candidate] (1.164 s) : 0, 1163540
Total [candidate] (9.585 s) : 0, 9585415
section profiling
Agent [baseline] (1.235 s) : 0, 1234892
Total [baseline] (9.624 s) : 0, 9624188
Agent [candidate] (1.234 s) : 0, 1234069
Total [candidate] (9.62 s) : 0, 9620486
gantt
title petclinic - break down per module: candidate=1.26.0-SNAPSHOT~9ff3556f43, baseline=1.26.0-SNAPSHOT~9d5c7ea524
dateFormat X
axisFormat %s
section tracing
BytebuddyAgent [baseline] (645.987 ms) : 0, 645987
BytebuddyAgent [candidate] (645.54 ms) : 0, 645540
GlobalTracer [baseline] (306.211 ms) : 0, 306211
GlobalTracer [candidate] (308.145 ms) : 0, 308145
AppSec [baseline] (48.455 ms) : 0, 48455
AppSec [candidate] (48.116 ms) : 0, 48116
Remote Config [baseline] (676.881 µs) : 0, 677
Remote Config [candidate] (665.906 µs) : 0, 666
Telemetry [baseline] (7.278 ms) : 0, 7278
Telemetry [candidate] (7.127 ms) : 0, 7127
section appsec
BytebuddyAgent [baseline] (644.67 ms) : 0, 644670
BytebuddyAgent [candidate] (645.766 ms) : 0, 645766
GlobalTracer [baseline] (305.81 ms) : 0, 305810
GlobalTracer [candidate] (307.684 ms) : 0, 307684
AppSec [baseline] (136.546 ms) : 0, 136546
AppSec [candidate] (136.778 ms) : 0, 136778
Remote Config [baseline] (640.592 µs) : 0, 641
Remote Config [candidate] (642.972 µs) : 0, 643
Telemetry [baseline] (7.44 ms) : 0, 7440
Telemetry [candidate] (6.82 ms) : 0, 6820
section iast
BytebuddyAgent [baseline] (773.547 ms) : 0, 773547
BytebuddyAgent [candidate] (768.15 ms) : 0, 768150
GlobalTracer [baseline] (287.294 ms) : 0, 287294
GlobalTracer [candidate] (285.515 ms) : 0, 285515
AppSec [baseline] (46.687 ms) : 0, 46687
AppSec [candidate] (46.585 ms) : 0, 46585
IAST [baseline] (19.455 ms) : 0, 19455
IAST [candidate] (20.187 ms) : 0, 20187
Remote Config [baseline] (601.429 µs) : 0, 601
Remote Config [candidate] (592.491 µs) : 0, 592
Telemetry [baseline] (7.307 ms) : 0, 7307
Telemetry [candidate] (8.079 ms) : 0, 8079
section profiling
BytebuddyAgent [baseline] (656.77 ms) : 0, 656770
BytebuddyAgent [candidate] (657.04 ms) : 0, 657040
GlobalTracer [baseline] (378.046 ms) : 0, 378046
GlobalTracer [candidate] (377.315 ms) : 0, 377315
AppSec [baseline] (48.602 ms) : 0, 48602
AppSec [candidate] (49.059 ms) : 0, 49059
Remote Config [baseline] (691.928 µs) : 0, 692
Remote Config [candidate] (689.338 µs) : 0, 689
Telemetry [baseline] (7.435 ms) : 0, 7435
Telemetry [candidate] (7.379 ms) : 0, 7379
ProfilingAgent [baseline] (89.087 ms) : 0, 89087
ProfilingAgent [candidate] (88.459 ms) : 0, 88459
Profiling [baseline] (89.112 ms) : 0, 89112
Profiling [candidate] (88.486 ms) : 0, 88486
Startup time reports for insecure-bankgantt
title insecure-bank - global startup overhead: candidate=1.26.0-SNAPSHOT~9ff3556f43, baseline=1.26.0-SNAPSHOT~9d5c7ea524
dateFormat X
axisFormat %s
section tracing
Agent [baseline] (1.048 s) : 0, 1048458
Total [baseline] (8.688 s) : 0, 8687760
Agent [candidate] (1.051 s) : 0, 1051089
Total [candidate] (8.726 s) : 0, 8725539
section iast
Agent [baseline] (1.171 s) : 0, 1171246
Total [baseline] (9.306 s) : 0, 9306433
Agent [candidate] (1.163 s) : 0, 1162786
Total [candidate] (9.244 s) : 0, 9243737
section iast_TELEMETRY_OFF
Agent [baseline] (1.151 s) : 0, 1150882
Total [baseline] (9.23 s) : 0, 9229692
Agent [candidate] (1.155 s) : 0, 1154564
Total [candidate] (9.241 s) : 0, 9241046
gantt
title insecure-bank - break down per module: candidate=1.26.0-SNAPSHOT~9ff3556f43, baseline=1.26.0-SNAPSHOT~9d5c7ea524
dateFormat X
axisFormat %s
section tracing
BytebuddyAgent [baseline] (649.873 ms) : 0, 649873
BytebuddyAgent [candidate] (650.633 ms) : 0, 650633
GlobalTracer [baseline] (307.791 ms) : 0, 307791
GlobalTracer [candidate] (309.395 ms) : 0, 309395
AppSec [baseline] (48.389 ms) : 0, 48389
AppSec [candidate] (48.76 ms) : 0, 48760
Remote Config [baseline] (679.85 µs) : 0, 680
Remote Config [candidate] (669.89 µs) : 0, 670
Telemetry [baseline] (7.182 ms) : 0, 7182
Telemetry [candidate] (7.184 ms) : 0, 7184
section iast
BytebuddyAgent [baseline] (773.806 ms) : 0, 773806
BytebuddyAgent [candidate] (767.874 ms) : 0, 767874
GlobalTracer [baseline] (287.618 ms) : 0, 287618
GlobalTracer [candidate] (285.831 ms) : 0, 285831
AppSec [baseline] (46.715 ms) : 0, 46715
AppSec [candidate] (46.55 ms) : 0, 46550
IAST [baseline] (21.086 ms) : 0, 21086
IAST [candidate] (20.978 ms) : 0, 20978
Remote Config [baseline] (607.268 µs) : 0, 607
Remote Config [candidate] (605.967 µs) : 0, 606
Telemetry [baseline] (6.679 ms) : 0, 6679
Telemetry [candidate] (6.599 ms) : 0, 6599
section iast_TELEMETRY_OFF
BytebuddyAgent [baseline] (758.286 ms) : 0, 758286
BytebuddyAgent [candidate] (759.844 ms) : 0, 759844
GlobalTracer [baseline] (285.004 ms) : 0, 285004
GlobalTracer [candidate] (286.385 ms) : 0, 286385
AppSec [baseline] (47.935 ms) : 0, 47935
AppSec [candidate] (47.801 ms) : 0, 47801
IAST [baseline] (16.832 ms) : 0, 16832
IAST [candidate] (16.989 ms) : 0, 16989
Remote Config [baseline] (590.279 µs) : 0, 590
Remote Config [candidate] (596.845 µs) : 0, 597
Telemetry [baseline] (7.947 ms) : 0, 7947
Telemetry [candidate] (8.738 ms) : 0, 8738
LoadParameters
See matching parameters
SummaryFound 0 performance improvements and 0 performance regressions! Performance is the same for 9 metrics, 13 unstable metrics. Request duration reports for petclinicgantt
title petclinic - request duration [CI 0.99] : candidate=1.26.0-SNAPSHOT~9ff3556f43, baseline=1.26.0-SNAPSHOT~9d5c7ea524
dateFormat X
axisFormat %s
section baseline
no_agent (1.344 ms) : 1325, 1363
. : milestone, 1344,
appsec (1.776 ms) : 1751, 1801
. : milestone, 1776,
iast (1.53 ms) : 1506, 1554
. : milestone, 1530,
profiling (1.549 ms) : 1522, 1575
. : milestone, 1549,
tracing (1.481 ms) : 1456, 1507
. : milestone, 1481,
section candidate
no_agent (1.346 ms) : 1327, 1365
. : milestone, 1346,
appsec (1.724 ms) : 1698, 1749
. : milestone, 1724,
iast (1.541 ms) : 1517, 1566
. : milestone, 1541,
profiling (1.574 ms) : 1548, 1601
. : milestone, 1574,
tracing (1.511 ms) : 1486, 1536
. : milestone, 1511,
Request duration reports for insecure-bankgantt
title insecure-bank - request duration [CI 0.99] : candidate=1.26.0-SNAPSHOT~9ff3556f43, baseline=1.26.0-SNAPSHOT~9d5c7ea524
dateFormat X
axisFormat %s
section baseline
no_agent (365.994 µs) : 346, 386
. : milestone, 366,
iast (486.675 µs) : 466, 507
. : milestone, 487,
iast_FULL (533.205 µs) : 513, 554
. : milestone, 533,
iast_INACTIVE (441.426 µs) : 421, 462
. : milestone, 441,
iast_TELEMETRY_OFF (481.713 µs) : 461, 503
. : milestone, 482,
tracing (444.854 µs) : 424, 466
. : milestone, 445,
section candidate
no_agent (359.981 µs) : 341, 379
. : milestone, 360,
iast (474.844 µs) : 454, 496
. : milestone, 475,
iast_FULL (533.107 µs) : 513, 553
. : milestone, 533,
iast_INACTIVE (446.558 µs) : 425, 468
. : milestone, 447,
iast_TELEMETRY_OFF (465.261 µs) : 445, 486
. : milestone, 465,
tracing (444.988 µs) : 424, 466
. : milestone, 445,
|
kr-igor
approved these changes
Oct 13, 2023
| producerProps.put(ProducerConfig.METADATA_MAX_AGE_CONFIG, 1000) | ||
| def producerFactory = new DefaultKafkaProducerFactory<String, String>(producerProps) | ||
| def kafkaTemplate = new KafkaTemplate<String, String>(producerFactory) | ||
| kafkaTemplate.flush() |
Contributor
There was a problem hiding this comment.
This test may be flaky, depending on how the metadata update is done. Maybe we can check for the data we need in a while loop, until we get the data or some reasonable timeout is reached?
| if (iterator != null) { | ||
| String group = InstrumentationContext.get(ConsumerRecords.class, String.class).get(records); | ||
| iterator = new TracingIterator(iterator, KAFKA_CONSUME, CONSUMER_DECORATE, group); | ||
| String group = null; |
Contributor
There was a problem hiding this comment.
nit: it seems like this code is repeated several times, maybe we can extract it to a static method? Something like props = KafkaConsumerInfo.get(record)
Contributor
Author
There was a problem hiding this comment.
updated by passing ContextStore as recommended by the APM Java team
dougqh
reviewed
Oct 24, 2023
| return clientMetadata; | ||
| } | ||
|
|
||
| public static class Builder { |
Contributor
There was a problem hiding this comment.
Builders are rather wasteful in terms of overhead.
For just two parameters, a simple static factory method would be sufficient.
Contributor
Author
There was a problem hiding this comment.
removed the Builder
dougqh
reviewed
Oct 24, 2023
|
|
||
| @Override | ||
| public int hashCode() { | ||
| return Objects.hash(consumerGroup, clientMetadata); |
Contributor
There was a problem hiding this comment.
Objects.hash performs poorly. If this is on the hot path, you should should implement hashCode in a more direct fashion.
Contributor
Author
There was a problem hiding this comment.
we only use KafkaConsumerInfo as a value to the context store, not as a key, so it shouldn't be on the hot path, but updated the implementation to match what Pair does just in case.
dougqh
reviewed
Oct 24, 2023
| @Override | ||
| public Map<String, String> contextStore() { | ||
| return singletonMap("org.apache.kafka.clients.consumer.ConsumerRecords", "java.lang.String"); | ||
| Map<String, String> contextStores = new HashMap<>(); |
Contributor
There was a problem hiding this comment.
I suppose this is only called once, but size the collection?
dougqh
reviewed
Oct 24, 2023
| if (iterable != null) { | ||
| String group = InstrumentationContext.get(ConsumerRecords.class, String.class).get(records); | ||
| iterable = new TracingList(iterable, KAFKA_CONSUME, CONSUMER_DECORATE, group); | ||
| Pair<String, String> data = |
Contributor
There was a problem hiding this comment.
This looks like extra overhead that is always on -- even when DSM is disabled.
That's a violation of our requirement that DSM imposes no overhead when not enabled.
I'd like to see this benchmarked to determine the cost.
Contributor
Author
There was a problem hiding this comment.
I benchmarked this yesterday using JMH. For the produce call I simply called kafkaProducer.send in the benchmark method, whereas to benchmark the consume call I pre-sent 100k messages in setup and then called kafkaConsume.poll in the benchmark method. Here is the results:
dougqh
reviewed
Oct 24, 2023
| import org.apache.kafka.clients.consumer.ConsumerRecords; | ||
|
|
||
| public class KafkaConsumerInstrumentationHelper { | ||
| public static Pair<String, String> extractGroupAndClusterId( |
Contributor
There was a problem hiding this comment.
Allocating a Pair object all the time is worrisome from a performance perspective.
Contributor
Author
There was a problem hiding this comment.
refactored to not use a Pair anymore. Though we technically still have a Pair-like structure in KafkaConsumerInfo
dougqh
reviewed
Oct 24, 2023
| producer.flush() | ||
| clusterId = producer.metadata.cluster.clusterResource().clusterId() | ||
| while (clusterId == null || clusterId.isEmpty()) { | ||
| Thread.sleep(1500) |
Contributor
There was a problem hiding this comment.
I'm not keen on adding sleep-s into the test. Is there some better way to coordinate the test verification?
Contributor
Author
There was a problem hiding this comment.
We can also just remove the Thread.sleep and it would still work, because we're just waiting for the data to come back from Embedded kafka. Otherwise, we might also be able to add resource listeners by directly accessing it from the metadata (https://github.com/apache/kafka/blob/0.11.0/clients/src/main/java/org/apache/kafka/clients/Metadata.java#L99) But then we'll have to run the test inside the listener, which I'm not quite sure if it's feasible.
lmk what your thoughts are!
dougqh
approved these changes
Oct 24, 2023
dougqh
left a comment
dougqh
left a comment
Contributor
There was a problem hiding this comment.
I'm concerned by the overhead being added into the instrumentation.
I'm particularly concerned that it looks like this overhead is being added all the time not just when DSM is enabled.
…, and remove Builder class from helperClassNames in KafkaConsumerInfoInstrumentation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What Does This Do
Added new logic to Kafka instrumentation to collect Kafka cluster ID, and propagate that ID in the Data Streams Monitoring (DSM) payloads. This cluster ID will allow Data Streams Monitoring to perform more fine-tuned monitoring for Kafka workloads that happen across different clusters.
Motivation
The Data Streams Monitoring metrics that track end-to-end latency and Kafka lag today do not keep track of the clusters that the workloads are communicating with. As such, if two workload in separate clusters have similar metadata (for example, similarly named services talking to similarly named Kafka topics that are situated in different regions), DSM is unable to automatically differentiate between them today. Adding the cluster ID as a tag to the DSM metrics will allow DSM to monitor workload across different Kafka clusters separately.
Additional Notes
Jira ticket: [PROJ-IDENT]