feat: Kafka Integration

[PROFeNoM]

Description

This integration's particularity lies in an internal function call substitution from ProducerTopic::produce to ProducerTopic::producev.

public produce ( integer $partition , integer $msgflags [, string $payload = NULL [, string $key = NULL [, string $opaque = NULL ]]] ) : void
public producev ( integer $partition , integer $msgflags , string $payload [, string $key = NULL [, array $headers = NULL [, integer $timestamp_ms = NULL [, string $opaque = NULL ]]]] ) : void

This substitution is a trick to work around the missing $headers parameters when using ProducerTopic::produce. Remember that this parameter injects distributed tracing headers.

Moreover, the consumer span is started a posteriori - i.e., the start time of the consuming call is saved, and either a root span or child span is created depending on the retrieved headers from the message during the post hook. This is because not all consuming operation will receive a message with headers, and therefore, not all consuming operation ought to be a distributed span of its own.

Reviewer checklist

• Test coverage seems ok.
• Appropriate labels assigned.

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 96.58120% with 4 lines in your changes missing coverage. Please review.

Project coverage is 72.92%. Comparing base (dc9911b) to head (891fa03).
Report is 3 commits behind head on master.

Files with missing lines	Patch %	Lines
...rc/DDTrace/Integrations/Kafka/KafkaIntegration.php	96.58%	4 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #3004      +/-   ##
============================================
+ Coverage     72.74%   72.92%   +0.18%     
- Complexity     2750     2781      +31     
============================================
  Files           138      139       +1     
  Lines         15060    15177     +117     
  Branches       1026     1026              
============================================
+ Hits          10955    11068     +113     
- Misses         3546     3550       +4     
  Partials        559      559              

Flag	Coverage Δ
appsec-extension	67.95% <ø> (ø)
tracer-php	74.80% <96.58%> (+0.23%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
src/ddtrace_php_api.stubs.php	0.00% <ø> (ø)
...rc/DDTrace/Integrations/Kafka/KafkaIntegration.php	96.58% <96.58%> (ø)

---

Continue to review full report in Codecov by Sentry.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update dc9911b...891fa03. Read the comment docs.

[pr-commenter]

Benchmarks [ tracer ]

Benchmark execution time: 2025-01-03 13:55:28

Comparing candidate commit 891fa03 in PR branch alex/feat/kafka with baseline commit dc9911b in branch master.

Found 2 performance improvements and 1 performance regressions! Performance is the same for 175 metrics, 0 unstable metrics.

scenario:EmptyFileBench/benchEmptyFileBaseline

• 🟩 execution_time [-254.311µs; -86.609µs] or [-8.327%; -2.836%]

scenario:SamplingRuleMatchingBench/benchRegexMatching1-opcache

• 🟥 execution_time [+230.988ns; +569.012ns] or [+3.232%; +7.963%]

scenario:TraceFlushBench/benchFlushTrace

• 🟩 execution_time [-1000.000ns; -1000.000ns] or [-50.000%; -50.000%]

[pablomartinezbernardo]

q: Should we make this depend on DD_TRACE_KAFKA_DISTRIBUTED_TRACING as well? There may be scenarios where injecting headers could mess something up on the other end. You could argue they can just disable the integration, but they may be interested in keeping it for some reason. Not a strong requirement at all, just to make things a little bit more configurable. Do other distributed tracing integrations (even in other tracers) allow for that?

[PROFeNoM]

Do other distributed tracing integrations allow for that?

Two other integrations make use of a similar env var: Laravel Queue & Symfony Messenger. In both of them, this env var is only checked on the consuming end. However, your message reminded me that we should indeed check that distributed tracing is enabled on the producing end, and made the following commit.

Do other traces allow for that?

AFAIK, no. I'm not sure that other tracers leverage span links in their instrumentations.

[bwoebi]

Nicely done, thanks Alexandre :-)