[WIP] feat: SocketTracer — PLT-patch Netty native transport for socket I/O events

build-logic/conventions/src/main/kotlin/com/datadoghq/native/gtest/GtestPlugin.kt

@@ -103,7 +103,14 @@ class GtestPlugin : Plugin<Project> {
                 "/usr/local/include/gtest"
             )
         }
-        return locations.any { File(it).exists() }
+        if (locations.any { File(it).exists() }) return true
+
+        // Also check GTEST_HOME environment variable
+        val gtestHomeEnv = System.getenv("GTEST_HOME")
+        if (!gtestHomeEnv.isNullOrBlank()) {
+            return File("$gtestHomeEnv/include/gtest").exists()
+        }
+        return false
     }
 
     private fun createBuildNativeLibsTask(project: Project, extension: GtestExtension, hasGtest: Boolean) {
@@ -226,7 +233,15 @@ class GtestPlugin : Plugin<Project> {
                 }
                 listOf(File("$gtestPath/include"))
             }
-            Platform.LINUX -> emptyList() // System includes
+            Platform.LINUX -> {
+                // Support GTEST_HOME env var for custom installations
+                val gtestHomeEnv = System.getenv("GTEST_HOME")
+                if (!gtestHomeEnv.isNullOrBlank()) {
+                    listOf(File("$gtestHomeEnv/include"))
+                } else {
+                    emptyList() // System includes
+                }
+            }
         }
     }
 }

build-logic/conventions/src/main/kotlin/com/datadoghq/native/gtest/GtestTaskBuilder.kt

@@ -125,7 +125,13 @@ class GtestTaskBuilder(
                     val gtestPath = gtestHomePath()
                     libPath("$gtestPath/lib")
                 }
-                Platform.LINUX -> { /* System paths */ }
+                Platform.LINUX -> {
+                    // Support GTEST_HOME env var for custom installations
+                    val gtestHomeEnv = System.getenv("GTEST_HOME")
+                    if (!gtestHomeEnv.isNullOrBlank()) {
+                        libPath("$gtestHomeEnv/lib")
+                    }
+                }
             }
 
             // Add gtest libraries

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

@@ -302,16 +302,30 @@ void CodeCache::saveImport(ImportId id, void** entry) {
 
 void CodeCache::addImport(void **entry, const char *name) {
     switch (name[0]) {
+      case 'a':
+          if (strcmp(name, "accept") == 0) {
+              saveImport(im_accept, entry);
+          } else if (strcmp(name, "accept4") == 0) {
+              saveImport(im_accept4, entry);
+          }
+          break;
       case 'c':
           if (strcmp(name, "calloc") == 0) {
               saveImport(im_calloc, entry);
+          } else if (strcmp(name, "connect") == 0) {
+              saveImport(im_connect, entry);
           }
           break;
       case 'd':
           if (strcmp(name, "dlopen") == 0) {
               saveImport(im_dlopen, entry);
           }
           break;
+      case 'e':
+          if (strcmp(name, "epoll_wait") == 0) {
+              saveImport(im_epoll_wait, entry);
+          }
+          break;
       case 'f':
           if (strcmp(name, "free") == 0) {
               saveImport(im_free, entry);
@@ -334,8 +348,30 @@ void CodeCache::addImport(void **entry, const char *name) {
           }
           break;
       case 'r':
-          if (strcmp(name, "realloc") == 0) {
+          if (strcmp(name, "read") == 0) {
+              saveImport(im_read, entry);
+          } else if (strcmp(name, "readv") == 0) {
+              saveImport(im_readv, entry);
+          } else if (strcmp(name, "realloc") == 0) {
               saveImport(im_realloc, entry);
+          } else if (strcmp(name, "recv") == 0) {
+              saveImport(im_recv, entry);
+          } else if (strcmp(name, "recvmsg") == 0) {
+              saveImport(im_recvmsg, entry);
+          }
+          break;
+      case 's':
+          if (strcmp(name, "send") == 0) {
+              saveImport(im_send, entry);
+          } else if (strcmp(name, "sendmsg") == 0) {
+              saveImport(im_sendmsg, entry);
+          }
+          break;
+      case 'w':
+          if (strcmp(name, "write") == 0) {
+              saveImport(im_write, entry);
+          } else if (strcmp(name, "writev") == 0) {
+              saveImport(im_writev, entry);
           }
           break;
     }

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

@@ -31,6 +31,18 @@ enum ImportId {
   im_calloc,
   im_realloc,
   im_free,
+  im_read,
+  im_write,
+  im_readv,
+  im_writev,
+  im_recv,
+  im_send,
+  im_recvmsg,
+  im_sendmsg,
+  im_connect,
+  im_accept,
+  im_accept4,
+  im_epoll_wait,
   NUM_IMPORTS
 };
 

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

@@ -63,6 +63,7 @@
   X(SKIPPED_WALLCLOCK_UNWINDS, "skipped_wallclock_unwinds")                    \
   X(UNWINDING_TIME_ASYNC, "unwinding_ticks_async")                             \
   X(UNWINDING_TIME_JVMTI, "unwinding_ticks_jvmti")                             \
+  X(UNWINDING_TIME_SOCKETIO, "unwinding_ticks_socketio")                       \
   X(CALLTRACE_STORAGE_DROPPED, "calltrace_storage_dropped_traces")             \
   X(LINE_NUMBER_TABLES, "line_number_tables")                                  \
   X(REMOTE_SYMBOLICATION_FRAMES, "remote_symbolication_frames")                \

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

@@ -173,4 +173,12 @@ typedef struct QueueTimeEvent {
   u32 _queueLength;
 } QueueTimeEvent;
 
+/** Represents a single socket I/O operation intercepted via native PLT patching. */
+typedef struct SocketIOEvent {
+  u64 _start;                // start time in TSC ticks
+  u64 _end;                  // end time in TSC ticks
+  const char* _operation;    // operation name (pointer to a static string literal)
+  long _bytes;               // bytes transferred (negative when not applicable)
+} SocketIOEvent;
+
 #endif // _EVENT_H

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

@@ -375,19 +375,11 @@ MethodInfo *Lookup::resolveMethod(ASGCT_CallFrame &frame) {
       // Note: This is called during JFR serialization with lockAll() held (see Profiler::dump),
       // so the library array is stable - no concurrent dlopen_hook calls can modify it.
       CodeCache* lib = Libraries::instance()->getLibraryByIndex(lib_index);
-      if (lib != nullptr && lib->hasBuildId() && Profiler::instance()->isRemoteSymbolication()) {
+      if (lib != nullptr) {
         TEST_LOG("Found library: %s, build_id=%s", lib->name(), lib->buildId());
-        // Remote symbolication: defer to backend
+        // Create temporary RemoteFrameInfo for fillRemoteFrameInfo
         RemoteFrameInfo rfi(lib->buildId(), pc_offset, lib_index);
         fillRemoteFrameInfo(mi, &rfi);
-      } else if (lib != nullptr) {
-        // Locally unsymbolized: render as [libname+0xoffset]
-        char name_buf[256];
-        const char* s = lib->name();
-        const char* basename = strrchr(s, '/');
-        if (basename) basename++; else basename = s;
-        snprintf(name_buf, sizeof(name_buf), "[%s+0x%" PRIxPTR "]", basename, pc_offset);
-        fillNativeMethodInfo(mi, name_buf, nullptr);
       } else {
         TEST_LOG("WARNING: Library lookup failed for index %u", lib_index);
         fillNativeMethodInfo(mi, "unknown_library", nullptr);
@@ -1556,6 +1548,20 @@ void Recording::recordQueueTime(Buffer *buf, int tid, QueueTimeEvent *event) {
   flushIfNeeded(buf);
 }
 
+void Recording::recordSocketIO(Buffer *buf, int tid, u64 call_trace_id, SocketIOEvent *event) {
+  int start = buf->skip(1);
+  buf->putVar64(T_SOCKET_IO);
+  buf->putVar64(event->_start);
+  buf->putVar64(event->_end - event->_start);
+  buf->putVar64(tid);
+  buf->putVar64(call_trace_id);
+  buf->putUtf8(event->_operation);
+  buf->putVar64(event->_bytes);
+  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 +1765,7 @@ void FlightRecorder::recordQueueTime(int lock_index, int tid,
   }
 }
 
+
 void FlightRecorder::recordDatadogSetting(int lock_index, int length,
                                           const char *name, const char *value,
                                           const char *unit) {
@@ -1812,6 +1819,9 @@ void FlightRecorder::recordEvent(int lock_index, int tid, u64 call_trace_id,
         case BCI_PARK:
           rec->recordThreadPark(buf, tid, call_trace_id, (LockEvent *)event);
           break;
+        case BCI_SOCKET_IO:
+          rec->recordSocketIO(buf, tid, call_trace_id, (SocketIOEvent *)event);
+          break;
         }
         rec->flushIfNeeded(buf);
         rec->addThread(lock_index, tid);

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 recordSocketIO(Buffer *buf, int tid, u64 call_trace_id, SocketIOEvent *event);
   void recordAllocation(RecordingBuffer *buf, int tid, u64 call_trace_id,
                         AllocEvent *event);
   void recordHeapLiveObject(Buffer *buf, int tid, u64 call_trace_id,

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

@@ -286,6 +286,18 @@ void JfrMetadata::initialize(
               << field("name", T_STRING, "Name")
               << field("count", T_LONG, "Count"))
 
+          << (type("datadog.SocketIO", T_SOCKET_IO, "Socket I/O")
+                  << 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("stackTrace", T_STACK_TRACE, "Stack Trace", F_CPOOL)
+                  << field("operation", T_STRING, "Operation")
+                  << field("bytesTransferred", T_LONG, "Bytes Transferred")
+                  << 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_SOCKET_IO = 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 "socketTracer.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();
+  SocketTracer::patchLibraries();
 }
 
 // Platform-specific implementation of updateBuildIds() is in libraries_linux.cpp (Linux)