[WIP] Explore tracking Netty / Native socket events

ddprof-lib/src/main/cpp/arguments.cpp

@@ -362,6 +362,25 @@ Error Arguments::parse(const char *args) {
         _remote_symbolication = true;
       }
 
+      CASE("nativesock")
+      if (value != NULL) {
+        switch (value[0]) {
+        case 'n': // no
+        case 'f': // false
+        case '0': // 0
+          _native_sockets = false;
+          break;
+        case 'y': // yes
+        case 't': // true
+        case '1': // 1
+        default:
+          _native_sockets = true;
+        }
+      } else {
+        // No value means enable
+        _native_sockets = true;
+      }
+
       CASE("wallsampler")
       if (value != NULL) {
           switch (value[0]) {

ddprof-lib/src/main/cpp/arguments.h

@@ -189,6 +189,7 @@ class Arguments {
   bool _lightweight;
   bool _enable_method_cleanup;
   bool _remote_symbolication;  // Enable remote symbolication for native frames
+  bool _native_sockets;        // Enable PLT patching of Netty native socket functions
 
   Arguments(bool persistent = false)
       : _buf(NULL),
@@ -223,7 +224,8 @@ class Arguments {
         _wallclock_sampler(ASGCT),
         _lightweight(false),
         _enable_method_cleanup(true),
-        _remote_symbolication(false) {}
+        _remote_symbolication(false),
+        _native_sockets(false) {}
 
   ~Arguments();
 

ddprof-lib/src/main/cpp/codeCache.cpp

@@ -336,6 +336,24 @@ void CodeCache::addImport(void **entry, const char *name) {
       case 'r':
           if (strcmp(name, "realloc") == 0) {
               saveImport(im_realloc, entry);
+          } else if (strcmp(name, "recv") == 0) {
+              saveImport(im_recv, entry);
+          } else if (strcmp(name, "recvfrom") == 0) {
+              saveImport(im_recvfrom, entry);
+          } else if (strcmp(name, "readv") == 0) {
+              saveImport(im_readv, entry);
+          }
+          break;
+      case 's':
+          if (strcmp(name, "send") == 0) {
+              saveImport(im_send, entry);
+          } else if (strcmp(name, "sendto") == 0) {
+              saveImport(im_sendto, entry);
+          }
+          break;
+      case 'w':
+          if (strcmp(name, "writev") == 0) {
+              saveImport(im_writev, entry);
           }
           break;
     }

ddprof-lib/src/main/cpp/codeCache.h

@@ -31,6 +31,13 @@ enum ImportId {
   im_calloc,
   im_realloc,
   im_free,
+  // Socket I/O — intercepted in Netty native transport libraries
+  im_recv,
+  im_send,
+  im_recvfrom,
+  im_sendto,
+  im_readv,
+  im_writev,
   NUM_IMPORTS
 };
 

ddprof-lib/src/main/cpp/event.h

@@ -173,4 +173,22 @@ typedef struct QueueTimeEvent {
   u32 _queueLength;
 } QueueTimeEvent;
 
+// Operation types for NativeSocketEvent
+enum SocketOp {
+  SOCKET_OP_RECV = 0,
+  SOCKET_OP_SEND = 1,
+  SOCKET_OP_RECVFROM = 2,
+  SOCKET_OP_SENDTO = 3,
+  SOCKET_OP_READV = 4,
+  SOCKET_OP_WRITEV = 5,
+};
+
+typedef struct NativeSocketEvent {
+  u64 _start;
+  u64 _end;
+  int _fd;
+  u64 _bytes;
+  u32 _operation;
+} NativeSocketEvent;
+
 #endif // _EVENT_H

ddprof-lib/src/main/cpp/flightRecorder.cpp

@@ -1556,6 +1556,20 @@ void Recording::recordQueueTime(Buffer *buf, int tid, QueueTimeEvent *event) {
   flushIfNeeded(buf);
 }
 
+void Recording::recordNativeSocketEvent(Buffer *buf, int tid, NativeSocketEvent *event) {
+  int start = buf->skip(1);
+  buf->putVar64(T_NATIVE_SOCKET_EVENT);
+  buf->putVar64(event->_start);
+  buf->putVar64(event->_end - event->_start);
+  buf->putVar64(tid);
+  buf->putVar64(static_cast<u32>(event->_fd));
+  buf->putVar64(event->_bytes);
+  buf->putVar64(event->_operation);
+  writeContext(buf, Contexts::get());
+  writeEventSizePrefix(buf, start);
+  flushIfNeeded(buf);
+}
+
 void Recording::recordAllocation(RecordingBuffer *buf, int tid,
                                  u64 call_trace_id, AllocEvent *event) {
   int start = buf->skip(1);
@@ -1759,6 +1773,18 @@ void FlightRecorder::recordQueueTime(int lock_index, int tid,
   }
 }
 
+void FlightRecorder::recordNativeSocketEvent(int lock_index, int tid,
+                                             NativeSocketEvent *event) {
+  OptionalSharedLockGuard locker(&_rec_lock);
+  if (locker.ownsLock()) {
+    Recording* rec = _rec;
+    if (rec != nullptr) {
+      Buffer *buf = rec->buffer(lock_index);
+      rec->recordNativeSocketEvent(buf, tid, event);
+    }
+  }
+}
+
 void FlightRecorder::recordDatadogSetting(int lock_index, int length,
                                           const char *name, const char *value,
                                           const char *unit) {

ddprof-lib/src/main/cpp/flightRecorder.h

@@ -278,6 +278,7 @@ class Recording {
   void recordWallClockEpoch(Buffer *buf, WallClockEpochEvent *event);
   void recordTraceRoot(Buffer *buf, int tid, TraceRootEvent *event);
   void recordQueueTime(Buffer *buf, int tid, QueueTimeEvent *event);
+  void recordNativeSocketEvent(Buffer *buf, int tid, NativeSocketEvent *event);
   void recordAllocation(RecordingBuffer *buf, int tid, u64 call_trace_id,
                         AllocEvent *event);
   void recordHeapLiveObject(Buffer *buf, int tid, u64 call_trace_id,
@@ -344,6 +345,7 @@ class FlightRecorder {
   void wallClockEpoch(int lock_index, WallClockEpochEvent *event);
   void recordTraceRoot(int lock_index, int tid, TraceRootEvent *event);
   void recordQueueTime(int lock_index, int tid, QueueTimeEvent *event);
+  void recordNativeSocketEvent(int lock_index, int tid, NativeSocketEvent *event);
 
   bool active() const { return _rec != NULL; }
 

ddprof-lib/src/main/cpp/javaApi.cpp

@@ -21,6 +21,7 @@
 #include "counters.h"
 #include "common.h"
 #include "engine.h"
+#include "event.h"
 #include "incbin.h"
 #include "os.h"
 #include "otel_process_ctx.h"
@@ -283,6 +284,22 @@ Java_com_datadoghq_profiler_JavaProfiler_recordSettingEvent0(
       tid, length, name_str.c_str(), value_str.c_str(), unit_str.c_str());
 }
 
+extern "C" DLLEXPORT void JNICALL
+Java_com_datadoghq_profiler_JavaProfiler_recordNativeSocketEvent0(
+    JNIEnv *env, jclass unused, jlong startTime, jlong endTime,
+    jint fd, jlong bytes, jint operation) {
+  if (fd < 0 || bytes < 0 || operation < SOCKET_OP_RECV || operation > SOCKET_OP_WRITEV) {
+    return;
+  }
+  NativeSocketEvent event;
+  event._start = startTime;
+  event._end = endTime;
+  event._fd = fd;
+  event._bytes = (u64)bytes;
+  event._operation = (u32)operation;
+  Profiler::instance()->recordNativeSocketEvent(&event);
+}
+
 static int dictionarizeClassName(JNIEnv* env, jstring className) {
     JniString str(env, className);
     return Profiler::instance()->lookupClass(str.c_str(), str.length());

ddprof-lib/src/main/cpp/jfrMetadata.cpp

@@ -286,6 +286,19 @@ void JfrMetadata::initialize(
               << field("name", T_STRING, "Name")
               << field("count", T_LONG, "Count"))
 
+          << (type("datadog.NativeSocketEvent", T_NATIVE_SOCKET_EVENT,
+                   "Native Socket Event")
+                  << category("Datadog")
+                  << field("startTime", T_LONG, "Start Time", F_TIME_TICKS)
+                  << field("duration", T_LONG, "Duration", F_DURATION_TICKS)
+                  << field("eventThread", T_THREAD, "Event Thread", F_CPOOL)
+                  << field("fd", T_INT, "File Descriptor")
+                  << field("bytesTransferred", T_LONG, "Bytes Transferred")
+                  << field("operation", T_INT, "Operation")
+                  << field("spanId", T_LONG, "Span ID")
+                  << field("localRootSpanId", T_LONG, "Local Root Span ID") ||
+              contextAttributes)
+
           << (type("jdk.OSInformation", T_OS_INFORMATION, "OS Information")
               << category("Operating System")
               << field("startTime", T_LONG, "Start Time", F_TIME_TICKS)

ddprof-lib/src/main/cpp/jfrMetadata.h

@@ -78,6 +78,7 @@ enum JfrType {
   T_DATADOG_CLASSREF_CACHE = 124,
   T_DATADOG_COUNTER = 125,
   T_UNWIND_FAILURE = 126,
+  T_NATIVE_SOCKET_EVENT = 127,
   T_ANNOTATION = 200,
   T_LABEL = 201,
   T_CATEGORY = 202,

ddprof-lib/src/main/cpp/libraries.cpp

@@ -3,6 +3,7 @@
 #include "libraries.h"
 #include "libraryPatcher.h"
 #include "log.h"
+#include "socketPatcher.h"
 #include "symbols.h"
 #include "symbols_linux.h"
 #include "vmEntry.h"
@@ -38,6 +39,7 @@ void Libraries::mangle(const char *name, char *buf, size_t size) {
 void Libraries::updateSymbols(bool kernel_symbols) {
   Symbols::parseLibraries(&_native_libs, kernel_symbols);
   LibraryPatcher::patch_libraries();
+  SocketPatcher::patch_libraries();
 }
 
 // Platform-specific implementation of updateBuildIds() is in libraries_linux.cpp (Linux)

ddprof-lib/src/main/cpp/profiler.cpp

@@ -18,6 +18,7 @@
 #include "j9WallClock.h"
 #include "libraryPatcher.h"
 #include "objectSampler.h"
+#include "socketPatcher.h"
 #include "os.h"
 #include "perfEvents.h"
 #include "safeAccess.h"
@@ -924,6 +925,21 @@ void Profiler::recordQueueTime(int tid, QueueTimeEvent *event) {
   _locks[lock_index].unlock();
 }
 
+void Profiler::recordNativeSocketEvent(NativeSocketEvent *event) {
+  int tid = ProfiledThread::currentTid();
+  if (tid < 0) {
+    return;
+  }
+  u32 lock_index = getLockIndex(tid);
+  if (!_locks[lock_index].tryLock() &&
+      !_locks[lock_index = (lock_index + 1) % CONCURRENCY_LEVEL].tryLock() &&
+      !_locks[lock_index = (lock_index + 2) % CONCURRENCY_LEVEL].tryLock()) {
+    return;
+  }
+  _jfr.recordNativeSocketEvent(lock_index, tid, event);
+  _locks[lock_index].unlock();
+}
+
 void Profiler::recordExternalSample(u64 weight, int tid, int num_frames,
                                     ASGCT_CallFrame *frames, bool truncated,
                                     jint event_type, Event *event) {
@@ -1282,6 +1298,7 @@ Error Profiler::start(Arguments &args, bool reset) {
 
   _omit_stacktraces = args._lightweight;
   _remote_symbolication = args._remote_symbolication;
+  _native_sockets = args._native_sockets;
   _event_mask =
       ((args._event != NULL && strcmp(args._event, EVENT_NOOP) != 0) ? EM_CPU
                                                                      : 0) |
@@ -1503,7 +1520,12 @@ Error Profiler::stop() {
 
   // Unpatch libraries AFTER JFR serialization completes
   // Remote symbolication RemoteFrameInfo structs contain pointers to build-ID strings
-  // owned by library metadata, so we must keep library patches active until after serialization
+  // owned by library metadata, so we must keep library patches active until after serialization.
+  // Note: between unlockAll() above and unpatch_libraries() below, PLT hooks may still
+  // fire from concurrent Netty I/O threads. This is safe because
+  // FlightRecorder::recordNativeSocketEvent checks _rec != nullptr (set to null by _jfr.stop()).
+  _native_sockets = false;  // Prevent dlopen hooks from re-patching after unpatch
+  SocketPatcher::unpatch_libraries();
   LibraryPatcher::unpatch_libraries();
 
   _state = IDLE;

ddprof-lib/src/main/cpp/profiler.h

@@ -155,6 +155,7 @@ class alignas(alignof(SpinLock)) Profiler {
   u32 _num_context_attributes;
   bool _omit_stacktraces;
   bool _remote_symbolication;  // Enable remote symbolication for native frames
+  bool _native_sockets;        // Enable PLT patching of Netty native socket functions
 
   // dlopen() hook support
   void **_dlopen_entry;
@@ -209,7 +210,7 @@ class alignas(alignof(SpinLock)) Profiler {
         _num_context_attributes(0), _class_map(1), _string_label_map(2),
         _context_value_map(3), _cpu_engine(), _alloc_engine(), _event_mask(0),
         _stop_time(), _total_samples(0), _failures(), _cstack(CSTACK_NO),
-        _omit_stacktraces(false) {
+        _omit_stacktraces(false), _native_sockets(false) {
 
     for (int i = 0; i < CONCURRENCY_LEVEL; i++) {
       _calltrace_buffer[i] = NULL;
@@ -375,6 +376,8 @@ class alignas(alignof(SpinLock)) Profiler {
   void recordWallClockEpoch(int tid, WallClockEpochEvent *event);
   void recordTraceRoot(int tid, TraceRootEvent *event);
   void recordQueueTime(int tid, QueueTimeEvent *event);
+  void recordNativeSocketEvent(NativeSocketEvent *event);
+  bool nativeSockets() const { return _native_sockets; }
   void writeLog(LogLevel level, const char *message);
   void writeLog(LogLevel level, const char *message, size_t len);
   void writeDatadogProfilerSetting(int tid, int length, const char *name,

ddprof-lib/src/main/cpp/socketPatcher.h

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2026, Datadog, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef _SOCKETPATCHER_H
+#define _SOCKETPATCHER_H
+
+#include "codeCache.h"
+#include "spinLock.h"
+
+#ifdef __linux__
+
+static const int MAX_SOCKET_PATCHED_LIBS = 16;
+
+// Stores the set of PLT entries patched for a single library.
+// Each entry holds the PLT location and the original function pointer
+// for one of the six intercepted socket functions.
+typedef struct _socketPatchSet {
+  CodeCache* _lib;
+  void** _locations[6];  // recv, send, recvfrom, sendto, readv, writev
+  void*  _originals[6];
+} SocketPatchSet;
+
+// Patches PLT entries for socket I/O functions (recv, send, recvfrom, sendto,
+// readv, writev) in Netty native transport libraries. Hook functions record
+// NativeSocketEvent entries to JFR via the profiler's recording pipeline.
+//
+// Only libraries whose basename contains "libnetty_transport_native" are
+// patched. The profiler's own library is never patched — hook functions
+// resolve to real libc symbols through the profiler's unpatched PLT.
+class SocketPatcher {
+private:
+  static SpinLock       _lock;
+  static SocketPatchSet _patched_libs[MAX_SOCKET_PATCHED_LIBS];
+  static int            _size;
+
+  static void patch_library_unlocked(CodeCache* lib);
+
+public:
+  static bool isNettyNativeLibrary(const char* name);
+  static void patch_libraries();
+  static void unpatch_libraries();
+};
+
+#else
+
+class SocketPatcher {
+public:
+  static void patch_libraries() { }
+  static void unpatch_libraries() { }
+};
+
+#endif
+
+#endif // _SOCKETPATCHER_H