feat: Kafka Integration

.circleci/continue_config.yml

@@ -43,6 +43,21 @@ aliases:
     image: kong/httpbin
     name: httpbin_integration
 
+  - &IMAGE_DOCKER_KAFKA
+    image: confluentinc/cp-kafka:7.7.1
+    name: kafka_integration
+    environment:
+      - KAFKA_BROKER_ID=111
+      - KAFKA_CREATE_TOPICS=test-lowlevel:1:1,test-highlevel:1:1
+      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka_integration:9092
+      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT
+      - KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT
+      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
+      - KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1
+      - KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1
+      - KAFKA_AUTO_CREATE_TOPICS_ENABLE=true
+
   - &IMAGE_DOCKER_MEMCHACED
     image: memcached:1.5-alpine
     name: memcached_integration
@@ -100,6 +115,14 @@ aliases:
     image: gcr.io/cloud-spanner-emulator/emulator:1.5.25
     name: googlespanner_integration
 
+  - &IMAGE_DOCKER_ZOOKEEPER
+    image: confluentinc/cp-zookeeper:7.7.1
+    name: zookeeper
+    environment:
+      - ZOOKEEPER_CLIENT_PORT=2181
+      - ZOOKEEPER_TICK_TIME=2000
+
+
   - &STEP_ATTACH_WORKSPACE
     attach_workspace:
       at: ~/datadog
@@ -537,6 +560,22 @@ commands:
                 -e ES_JAVA_OPTS="-Xms1g -Xmx1g" \
                 -e discovery.type=single-node \
                 --name elasticsearch7_integration elasticsearch:7.17.0
+              retry_docker run --detach --rm --net net \
+                -e ZOOKEEPER_CLIENT_PORT=2181 \
+                -e ZOOKEEPER_TICK_TIME=2000 \
+                --name zookeeper confluentinc/cp-zookeeper:7.7.1
+              retry_docker run --detach --rm --net net \
+                -e KAFKA_BROKER_ID=111 \
+                -e KAFKA_CREATE_TOPICS=test-lowlevel:1:1,test-highlevel:1:1 \
+                -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
+                -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka_integration:9092 \
+                -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT \
+                -e KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT \
+                -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
+                -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
+                -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
+                -e KAFKA_AUTO_CREATE_TOPICS_ENABLE=true \
+                --name kafka_integration confluentinc/cp-kafka:7.7.1
               retry_docker run --detach --rm --net net \
                 --name redis_integration datadog/dd-trace-ci:php-redis-5.0
               retry_docker run --detach --rm --net net \
@@ -581,6 +620,8 @@ commands:
                 docker_image: elasticsearch2_integration
             - docker_logs:
                 docker_image: elasticsearch7_integration
+            - docker_logs:
+                docker_image: kafka_integration
             - docker_logs:
                 docker_image: redis_integration
             - docker_logs:
@@ -595,6 +636,8 @@ commands:
                 docker_image: sqlsrv_integration
             - docker_logs:
                 docker_image: googlespanner_integration
+            - docker_logs:
+                docker_image: zookeeper
       - run:
           name: Setup docker image << parameters.docker_image >>
           command: |
@@ -732,6 +775,7 @@ executors:
       - <<: *IMAGE_DOCKER_ELASTICSEARCH7
       - <<: *IMAGE_DOCKER_HTTPBIN
       - <<: *IMAGE_DOCKER_REDIS
+      - <<: *IMAGE_DOCKER_KAFKA
       - <<: *IMAGE_DOCKER_MEMCHACED
       - <<: *IMAGE_DOCKER_MYSQL
       - <<: *IMAGE_DOCKER_RABBITMQ
@@ -740,6 +784,7 @@ executors:
       - <<: *IMAGE_DOCKER_DD_TEST_AGENT
       - <<: *IMAGE_DOCKER_SQLSRV
       - <<: *IMAGE_DOCKER_GOOGLESPANNER
+      - <<: *IMAGE_DOCKER_ZOOKEEPER
   with_redis:
     environment:
       DDAGENT_HOSTNAME: 127.0.0.1

.gitlab/ci-images.yml

@@ -14,7 +14,7 @@ CentOS:
     - when: manual
   needs: []
   tags: ["arch:amd64"]
-  timeout: 1h
+  timeout: 4h
   image: 486234852809.dkr.ecr.us-east-1.amazonaws.com/docker:24.0.4-gbi-focal
   parallel:
     matrix:
@@ -41,7 +41,7 @@ Alpine:
     - when: manual
   needs: []
   tags: ["arch:amd64"]
-  timeout: 1h
+  timeout: 4h
   image: 486234852809.dkr.ecr.us-east-1.amazonaws.com/docker:24.0.4-gbi-focal
   parallel:
     matrix:
@@ -68,7 +68,7 @@ Bookworm:
     - when: manual
   needs: []
   tags: ["arch:amd64"]
-  timeout: 1h
+  timeout: 4h
   image: 486234852809.dkr.ecr.us-east-1.amazonaws.com/docker:24.0.4-gbi-focal
   parallel:
     matrix:
@@ -97,7 +97,7 @@ Buster:
     - when: manual
   needs: []
   tags: ["arch:amd64"]
-  timeout: 1h
+  timeout: 4h
   image: 486234852809.dkr.ecr.us-east-1.amazonaws.com/docker:24.0.4-gbi-focal
   parallel:
     matrix:

Makefile

@@ -547,6 +547,7 @@ TELEMETRY_ENABLED=0
 TEST_INTEGRATIONS_70 := \
 	test_integrations_deferred_loading \
 	test_integrations_curl \
+	test_integrations_kafka \
 	test_integrations_memcache \
 	test_integrations_memcached \
 	test_integrations_mongodb1 \
@@ -588,6 +589,7 @@ TEST_INTEGRATIONS_71 := \
 	test_integrations_amqp2 \
 	test_integrations_amqp35 \
 	test_integrations_curl \
+	test_integrations_kafka \
 	test_integrations_memcache \
 	test_integrations_memcached \
 	test_integrations_mongodb1 \
@@ -639,6 +641,7 @@ TEST_INTEGRATIONS_72 := \
 	test_integrations_amqp2 \
 	test_integrations_amqp35 \
 	test_integrations_curl \
+	test_integrations_kafka \
 	test_integrations_memcache \
 	test_integrations_memcached \
 	test_integrations_mongodb1 \
@@ -697,6 +700,7 @@ TEST_INTEGRATIONS_73 :=\
 	test_integrations_amqp2 \
 	test_integrations_amqp35 \
 	test_integrations_curl \
+	test_integrations_kafka \
 	test_integrations_memcache \
 	test_integrations_memcached \
 	test_integrations_mongodb1 \
@@ -754,6 +758,7 @@ TEST_INTEGRATIONS_74 := \
 	test_integrations_amqp2 \
 	test_integrations_amqp35 \
 	test_integrations_curl \
+	test_integrations_kafka \
 	test_integrations_memcache \
 	test_integrations_memcached \
 	test_integrations_mongodb1 \
@@ -819,6 +824,7 @@ TEST_INTEGRATIONS_80 := \
 	test_integrations_amqp2 \
 	test_integrations_amqp35 \
 	test_integrations_curl \
+	test_integrations_kafka \
 	test_integrations_laminaslog2 \
 	test_integrations_memcache \
 	test_integrations_memcached \
@@ -871,6 +877,7 @@ TEST_INTEGRATIONS_81 := \
 	test_integrations_amqp35 \
 	test_integrations_curl \
 	test_integrations_deferred_loading \
+	test_integrations_kafka \
 	test_integrations_laminaslog2 \
 	test_integrations_memcache \
 	test_integrations_memcached \
@@ -925,6 +932,7 @@ TEST_INTEGRATIONS_82 := \
 	test_integrations_amqp35 \
 	test_integrations_curl \
 	test_integrations_deferred_loading \
+	test_integrations_kafka \
 	test_integrations_laminaslog2 \
 	test_integrations_memcache \
 	test_integrations_memcached \
@@ -987,6 +995,7 @@ TEST_INTEGRATIONS_83 := \
 	test_integrations_amqp35 \
 	test_integrations_curl \
 	test_integrations_deferred_loading \
+	test_integrations_kafka \
 	test_integrations_laminaslog2 \
 	test_integrations_memcache \
 	test_integrations_memcached \
@@ -1230,6 +1239,8 @@ test_integrations_guzzle6: global_test_run_dependencies  tests/Integrations/Guzz
 	$(call run_tests_debug,tests/Integrations/Guzzle/V6)
 test_integrations_guzzle7: global_test_run_dependencies tests/Integrations/Guzzle/V7/composer.lock-php$(PHP_MAJOR_MINOR)
 	$(call run_tests_debug,tests/Integrations/Guzzle/V7)
+test_integrations_kafka: global_test_run_dependencies
+	$(call run_tests_debug,tests/Integrations/Kafka)
 test_integrations_laminaslog2: global_test_run_dependencies tests/Integrations/Logs/LaminasLogV2/composer.lock-php$(PHP_MAJOR_MINOR)
 	$(call run_tests_debug,tests/Integrations/Logs/LaminasLogV2)
 test_integrations_memcached: global_test_run_dependencies

config.m4

@@ -184,6 +184,7 @@ if test "$PHP_DDTRACE" != "no"; then
     ext/handlers_api.c \
     ext/handlers_exception.c \
     ext/handlers_internal.c \
+    ext/handlers_kafka.c \
     ext/handlers_pcntl.c \
     ext/handlers_signal.c \
     ext/integrations/exec_integration.c \

config.w32

@@ -35,6 +35,7 @@ if (PHP_DDTRACE != 'no') {
 	DDTRACE_EXT_SOURCES += " handlers_curl" + (version < 800 ? "_php7" : "") + ".c";
 	DDTRACE_EXT_SOURCES += " handlers_exception.c";
 	DDTRACE_EXT_SOURCES += " handlers_internal.c";
+	DDTRACE_EXT_SOURCES += " handlers_kafka.c";
 	DDTRACE_EXT_SOURCES += " handlers_pcntl.c";
 	DDTRACE_EXT_SOURCES += " ip_extraction.c";
 	DDTRACE_EXT_SOURCES += " standalone_limiter.c";

docker-compose.yml

@@ -30,6 +30,7 @@ x-aliases:
         - mysql_integration
         - memcached_integration
         - googlespanner_integration
+        - kafka_integration
       environment:
         - REDIS_HOSTNAME=redis_integration
         - DDAGENT_HOSTNAME=ddagent_integration
@@ -169,6 +170,36 @@ services:
         aliases:
           - spanner.local
 
+  kafka_integration:
+    image: confluentinc/cp-kafka:7.7.1
+    ports:
+      - "9092:9092"
+    networks:
+      default:
+        aliases:
+          - kafka_integration
+    environment:
+        - KAFKA_BROKER_ID=111
+        - KAFKA_CREATE_TOPICS=test-lowlevel:1:1,test-highlevel:1:1
+        - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+        - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka_integration:9092
+        - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT
+        - KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT
+        - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
+        - KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1
+        - KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1
+        - KAFKA_AUTO_CREATE_TOPICS_ENABLE=true
+    depends_on:
+      - zookeeper
+
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.7.1
+    ports:
+      - "2181:2181"
+    environment:
+      - ZOOKEEPER_CLIENT_PORT=2181
+      - ZOOKEEPER_TICK_TIME=2000
+
   mongodb_integration:
     image: "circleci/mongo:4.0"
     ports:

dockerfiles/ci/alpine/Dockerfile

@@ -28,6 +28,7 @@ RUN set -eux; \
         libffi-dev \
         libmemcached \
         libmemcached-dev \
+        librdkafka-dev \
         libsodium-dev \
         libxml2-dev \
         libzip-dev \

dockerfiles/ci/alpine_compile_extension/base.Dockerfile

@@ -19,6 +19,7 @@ RUN set -eux; \
         libedit-dev \
         libffi-dev \
         libmcrypt-dev \
+        librdkafka-dev \
         libsodium-dev \
         libxml2-dev \
         gnu-libiconv-dev \

dockerfiles/ci/buster/Dockerfile

@@ -27,6 +27,7 @@ ENV DEVLIBS \
     libpq-dev \
     libpng-dev \
     librabbitmq-dev \
+    librdkafka-dev \
     libsodium-dev \
     libsqlite3-dev \
     libssl-dev \

dockerfiles/ci/buster/build-extensions.sh

@@ -128,6 +128,7 @@ else
   yes 'no' | pecl install memcached; echo "extension=memcached.so" >> ${iniDir}/memcached.ini;
   yes '' | pecl install memcache$MEMCACHE_VERSION; echo "extension=memcache.so" >> ${iniDir}/memcache.ini;
   pecl install mongodb$MONGODB_VERSION; echo "extension=mongodb.so" >> ${iniDir}/mongodb.ini;
+  pecl install rdkafka; echo "extension=rdkafka.so" >> ${iniDir}/rdkafka.ini;
   pecl install sqlsrv$SQLSRV_VERSION;
   # Xdebug is disabled by default
   for VERSION in "${XDEBUG_VERSIONS[@]}"; do

dockerfiles/ci/centos/7/base.Dockerfile

@@ -232,6 +232,7 @@ RUN set -eux; \
     bzip2-devel \
     httpd-devel \
     libmemcached-devel \
+    librdkafka-devel \
     libsodium-devel \
     libsqlite3x-devel \
     libxml2-devel \

ext/compatibility.h

@@ -622,6 +622,16 @@ static zend_always_inline zend_result zend_call_function_with_return_value(zend_
 
 #define Z_PARAM_ZVAL_OR_NULL(dest) Z_PARAM_ZVAL_EX(dest, 1, 0)
 
+#ifndef Z_PARAM_STRING_OR_NULL
+#define Z_PARAM_STRING_OR_NULL(dest, dest_len) \
+    Z_PARAM_STRING_EX(dest, dest_len, 1, 0)
+#endif
+
+#ifndef Z_PARAM_STR_OR_NULL
+#define Z_PARAM_STR_OR_NULL(dest) \
+    Z_PARAM_STR_EX(dest, 1, 0)
+#endif
+
 #define ZEND_GUARD_PROPERTY_MASK 0xf
 
 // strip const

ext/configuration.h

@@ -137,6 +137,7 @@ enum ddtrace_sampling_rules_format {
     CONFIG(STRING, DD_TRACE_MEMORY_LIMIT, "")                                                                  \
     CONFIG(BOOL, DD_TRACE_REPORT_HOSTNAME, "false")                                                            \
     CONFIG(BOOL, DD_TRACE_FLUSH_COLLECT_CYCLES, "false")                                                       \
+    CONFIG(BOOL, DD_TRACE_KAFKA_DISTRIBUTED_TRACING, "true")                                                   \
     CONFIG(BOOL, DD_TRACE_LARAVEL_QUEUE_DISTRIBUTED_TRACING, "true")                                           \
     CONFIG(BOOL, DD_TRACE_SYMFONY_MESSENGER_DISTRIBUTED_TRACING, "true")                                       \
     CONFIG(BOOL, DD_TRACE_SYMFONY_MESSENGER_MIDDLEWARES, "false")                                              \

ext/handlers_internal.c

@@ -30,6 +30,7 @@ void ddtrace_free_unregistered_class(zend_class_entry *ce) {
 void ddtrace_curl_handlers_startup(void);
 void ddtrace_exception_handlers_startup(void);
 void ddtrace_pcntl_handlers_startup(void);
+void ddtrace_kafka_handlers_startup(void);
 #ifndef _WIN32
 void ddtrace_signal_block_handlers_startup(void);
 #endif
@@ -150,6 +151,7 @@ void ddtrace_internal_handlers_startup() {
     ddtrace_exception_handlers_startup();
 
     ddtrace_exec_handlers_startup();
+    ddtrace_kafka_handlers_startup();
 #ifndef _WIN32
     // Block remote-config signals of some functions
     ddtrace_signal_block_handlers_startup();