diff --git a/ddprof-lib/src/main/cpp/arguments.cpp b/ddprof-lib/src/main/cpp/arguments.cpp index 4436e80a8..b3353e100 100644 --- a/ddprof-lib/src/main/cpp/arguments.cpp +++ b/ddprof-lib/src/main/cpp/arguments.cpp @@ -382,6 +382,18 @@ Error Arguments::parse(const char *args) { msg = "nativemem must be >= 0"; } + CASE("natsock") + if (value != NULL) { + _nativesocket_interval = parseUnits(value, NANOS); + if (_nativesocket_interval < 0) { + msg = "natsock interval must be >= 0"; + } else { + _nativesocket = true; + } + } else { + _nativesocket = true; + } + DEFAULT() if (_unknown_arg == NULL) _unknown_arg = arg; diff --git a/ddprof-lib/src/main/cpp/arguments.h b/ddprof-lib/src/main/cpp/arguments.h index 2d400f213..435e50d2e 100644 --- a/ddprof-lib/src/main/cpp/arguments.h +++ b/ddprof-lib/src/main/cpp/arguments.h @@ -100,9 +100,10 @@ enum EventMask { EM_LOCK = 4, EM_WALL = 8, EM_NATIVEMEM = 16, - EM_METHOD_TRACE = 32 + EM_METHOD_TRACE = 32, + EM_NATIVESOCKET = 64 }; -constexpr int EVENT_MASK_SIZE = 6; +constexpr int EVENT_MASK_SIZE = 7; struct StackWalkFeatures { // Deprecated stack recovery techniques used to workaround AsyncGetCallTrace flaws @@ -192,6 +193,8 @@ class Arguments { bool _lightweight; bool _enable_method_cleanup; bool _remote_symbolication; // Enable remote symbolication for native frames + bool _nativesocket; + long _nativesocket_interval; // initial sampling period in nanoseconds; 0 = engine default Arguments(bool persistent = false) : _buf(NULL), @@ -227,7 +230,9 @@ class Arguments { _context_attributes({}), _lightweight(false), _enable_method_cleanup(true), - _remote_symbolication(false) {} + _remote_symbolication(false), + _nativesocket(false), + _nativesocket_interval(0) {} ~Arguments(); diff --git a/ddprof-lib/src/main/cpp/codeCache.cpp b/ddprof-lib/src/main/cpp/codeCache.cpp index a3c6e29d1..5203d5c5d 100644 --- a/ddprof-lib/src/main/cpp/codeCache.cpp +++ b/ddprof-lib/src/main/cpp/codeCache.cpp @@ -350,13 +350,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, "read") == 0) { + saveImport(im_read, entry); } break; case 's': - if (strcmp(name, "sigaction") == 0) { + if (strcmp(name, "send") == 0) { + saveImport(im_send, entry); + } else if (strcmp(name, "sigaction") == 0) { saveImport(im_sigaction, entry); } break; + case 'w': + if (strcmp(name, "write") == 0) { + saveImport(im_write, entry); + } + break; } } diff --git a/ddprof-lib/src/main/cpp/codeCache.h b/ddprof-lib/src/main/cpp/codeCache.h index 5c9a5b155..92b45bf47 100644 --- a/ddprof-lib/src/main/cpp/codeCache.h +++ b/ddprof-lib/src/main/cpp/codeCache.h @@ -38,6 +38,10 @@ enum ImportId { im_posix_memalign, im_aligned_alloc, im_sigaction, + im_send, + im_recv, + im_write, + im_read, NUM_IMPORTS }; diff --git a/ddprof-lib/src/main/cpp/event.h b/ddprof-lib/src/main/cpp/event.h index 752db842d..9d4457acd 100644 --- a/ddprof-lib/src/main/cpp/event.h +++ b/ddprof-lib/src/main/cpp/event.h @@ -99,6 +99,19 @@ class MallocEvent : public Event { MallocEvent() : Event(), _start_time(0), _address(0), _size(0), _weight(1.0f) {} }; +class NativeSocketEvent : public Event { +public: + u64 _start_time; // TSC ticks at call entry + u64 _end_time; // TSC ticks at call return + u8 _operation; // 0 = SEND, 1 = RECV, 2 = WRITE, 3 = READ + char _remote_addr[64]; // "ip:port" null-terminated string + u64 _bytes; // bytes transferred (return value of send/recv/write/read) + float _weight; // inverse-transform sample weight + + NativeSocketEvent() : Event(), _start_time(0), _end_time(0), _operation(0), + _bytes(0), _weight(1.0f) { _remote_addr[0] = '\0'; } +}; + class WallClockEpochEvent { public: bool _dirty; diff --git a/ddprof-lib/src/main/cpp/flightRecorder.cpp b/ddprof-lib/src/main/cpp/flightRecorder.cpp index cefbc476b..ff336e8e5 100644 --- a/ddprof-lib/src/main/cpp/flightRecorder.cpp +++ b/ddprof-lib/src/main/cpp/flightRecorder.cpp @@ -1600,6 +1600,24 @@ void Recording::recordMallocSample(Buffer *buf, int tid, u64 call_trace_id, flushIfNeeded(buf); } +void Recording::recordNativeSocketSample(Buffer *buf, int tid, u64 call_trace_id, + NativeSocketEvent *event) { + int start = buf->skip(1); + buf->putVar64(T_NATIVE_SOCKET); + buf->putVar64(event->_start_time); + buf->putVar64(tid); + buf->putVar64(call_trace_id); + buf->putVar64(event->_end_time - event->_start_time); + static const char* const kOpNames[] = {"SEND", "RECV", "WRITE", "READ"}; + buf->putUtf8(event->_operation < 4 ? kOpNames[event->_operation] : "UNKNOWN"); + buf->putUtf8(event->_remote_addr); + buf->putVar64(event->_bytes); + buf->putFloat(event->_weight); + writeCurrentContext(buf); + writeEventSizePrefix(buf, start); + flushIfNeeded(buf); +} + void Recording::recordHeapLiveObject(Buffer *buf, int tid, u64 call_trace_id, ObjectLivenessEvent *event) { int start = buf->skip(1); @@ -1845,6 +1863,9 @@ void FlightRecorder::recordEvent(int lock_index, int tid, u64 call_trace_id, case BCI_NATIVE_MALLOC: rec->recordMallocSample(buf, tid, call_trace_id, (MallocEvent *)event); break; + case BCI_NATIVE_SOCKET: + rec->recordNativeSocketSample(buf, tid, call_trace_id, (NativeSocketEvent *)event); + break; } rec->flushIfNeeded(buf); rec->addThread(lock_index, tid); diff --git a/ddprof-lib/src/main/cpp/flightRecorder.h b/ddprof-lib/src/main/cpp/flightRecorder.h index e9aa3cde1..5ea2b0dc1 100644 --- a/ddprof-lib/src/main/cpp/flightRecorder.h +++ b/ddprof-lib/src/main/cpp/flightRecorder.h @@ -283,6 +283,8 @@ class Recording { AllocEvent *event); void recordMallocSample(Buffer *buf, int tid, u64 call_trace_id, MallocEvent *event); + void recordNativeSocketSample(Buffer *buf, int tid, u64 call_trace_id, + NativeSocketEvent *event); void recordHeapLiveObject(Buffer *buf, int tid, u64 call_trace_id, ObjectLivenessEvent *event); void recordMonitorBlocked(Buffer *buf, int tid, u64 call_trace_id, diff --git a/ddprof-lib/src/main/cpp/hotspot/hotspotSupport.cpp b/ddprof-lib/src/main/cpp/hotspot/hotspotSupport.cpp index 50bf0d6d0..e4665f9bd 100644 --- a/ddprof-lib/src/main/cpp/hotspot/hotspotSupport.cpp +++ b/ddprof-lib/src/main/cpp/hotspot/hotspotSupport.cpp @@ -1106,7 +1106,31 @@ int HotspotSupport::walkJavaStack(StackWalkRequest& request) { int java_frames = 0; if (features.mixed) { java_frames = walkVM(ucontext, frames, max_depth, features, eventTypeFromBCI(request.event_type), lock_index, truncated); - } else if (request.event_type == BCI_CPU || request.event_type == BCI_WALL || request.event_type == BCI_NATIVE_MALLOC) { + } else if (request.event_type == BCI_NATIVE_MALLOC || request.event_type == BCI_NATIVE_SOCKET) { + if (cstack >= CSTACK_VM) { + java_frames = walkVM(ucontext, frames, max_depth, features, eventTypeFromBCI(request.event_type), lock_index, truncated); + } else { + AsyncSampleMutex mutex(ProfiledThread::currentSignalSafe()); + if (mutex.acquired()) { + java_frames = getJavaTraceAsync(ucontext, frames, max_depth, java_ctx, truncated); + if (java_frames > 0 && java_ctx->pc != NULL && VMStructs::hasMethodStructs()) { + VMNMethod* nmethod = CodeHeap::findNMethod(java_ctx->pc); + if (nmethod != NULL) { + fillFrameTypes(frames, java_frames, nmethod); + } + } + } + if (java_frames > 0 && VM::hotspot_version() >= 21 && java_frames < max_depth) { + VMThread* carrier = VMThread::current(); + if (carrier != nullptr && carrier->isCarryingVirtualThread()) { + frames[java_frames].bci = BCI_NATIVE_FRAME; + frames[java_frames].method_id = (jmethodID) "JVM Continuation"; + LP64_ONLY(frames[java_frames].padding = 0;) + java_frames++; + } + } + } + } else if (request.event_type == BCI_CPU || request.event_type == BCI_WALL) { if (cstack >= CSTACK_VM) { java_frames = walkVM(ucontext, frames, max_depth, features, eventTypeFromBCI(request.event_type), lock_index, truncated); } else { diff --git a/ddprof-lib/src/main/cpp/jfrMetadata.cpp b/ddprof-lib/src/main/cpp/jfrMetadata.cpp index 54e0f6a15..2830396d7 100644 --- a/ddprof-lib/src/main/cpp/jfrMetadata.cpp +++ b/ddprof-lib/src/main/cpp/jfrMetadata.cpp @@ -311,6 +311,20 @@ void JfrMetadata::initialize( << field("localRootSpanId", T_LONG, "Local Root Span ID") || contextAttributes) + << (type("datadog.NativeSocketEvent", T_NATIVE_SOCKET, "Native Socket I/O") + << category("Datadog", "Profiling") + << field("startTime", T_LONG, "Start Time", F_TIME_TICKS) + << field("eventThread", T_THREAD, "Event Thread", F_CPOOL) + << field("stackTrace", T_STACK_TRACE, "Stack Trace", F_CPOOL) + << field("duration", T_LONG, "Duration", F_DURATION_TICKS) + << field("operation", T_STRING, "Operation") + << field("remoteAddress", T_STRING, "Remote Address") + << field("bytesTransferred", T_LONG, "Bytes Transferred", F_BYTES) + << field("weight", T_FLOAT, "Sample weight") + << 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) diff --git a/ddprof-lib/src/main/cpp/jfrMetadata.h b/ddprof-lib/src/main/cpp/jfrMetadata.h index 52c2e0ae8..ac241a7a8 100644 --- a/ddprof-lib/src/main/cpp/jfrMetadata.h +++ b/ddprof-lib/src/main/cpp/jfrMetadata.h @@ -80,6 +80,7 @@ enum JfrType { T_DATADOG_COUNTER = 125, T_UNWIND_FAILURE = 126, T_MALLOC = 127, + T_NATIVE_SOCKET = 128, T_ANNOTATION = 200, T_LABEL = 201, T_CATEGORY = 202, diff --git a/ddprof-lib/src/main/cpp/jvmSupport.cpp b/ddprof-lib/src/main/cpp/jvmSupport.cpp index 6e3f5bc3a..93bbe00a0 100644 --- a/ddprof-lib/src/main/cpp/jvmSupport.cpp +++ b/ddprof-lib/src/main/cpp/jvmSupport.cpp @@ -22,7 +22,8 @@ int JVMSupport::walkJavaStack(StackWalkRequest& request) { } else if (VM::isOpenJ9() || VM::isZing()) { assert(request.event_type == BCI_CPU || request.event_type == BCI_WALL || - request.event_type == BCI_NATIVE_MALLOC); + request.event_type == BCI_NATIVE_MALLOC || + request.event_type == BCI_NATIVE_SOCKET); return asyncGetCallTrace(request.frames, request.max_depth, request.ucontext); } assert(false && "Unsupported JVM"); diff --git a/ddprof-lib/src/main/cpp/libraryPatcher.h b/ddprof-lib/src/main/cpp/libraryPatcher.h index 39cf6822c..70be3659b 100644 --- a/ddprof-lib/src/main/cpp/libraryPatcher.h +++ b/ddprof-lib/src/main/cpp/libraryPatcher.h @@ -3,6 +3,7 @@ #include "codeCache.h" #include "spinLock.h" +#include #ifdef __linux__ @@ -28,15 +29,36 @@ class LibraryPatcher { static PatchEntry _sigaction_entries[MAX_NATIVE_LIBS]; static int _sigaction_size; + // Separate tracking for socket (send/recv/write/read) patches. + // Each library can contribute up to 4 GOT slots (send/recv/write/read). + static PatchEntry _socket_entries[4 * MAX_NATIVE_LIBS]; + static int _socket_size; + static void patch_library_unlocked(CodeCache* lib); static void patch_pthread_create(); static void patch_pthread_setspecific(); static void patch_sigaction_in_library(CodeCache* lib); public: + // True while socket hooks are installed; read by Profiler::dlopen_hook + // to decide whether to re-patch after a new library is loaded. + // Set to true after the first batch of libraries is patched in patch_socket_functions(). + // Libraries loaded after profiler start are picked up on the next dlopen_hook call, + // which calls install_socket_hooks() to patch them if _socket_active is true. + // Low-probability race: stop() is called only on JVM exit; atomic is zero-cost insurance. + static std::atomic _socket_active; static void initialize(); static void patch_libraries(); static void unpatch_libraries(); static void patch_sigaction(); + static bool patch_socket_functions(); + static void unpatch_socket_functions(); + // Called from Profiler::dlopen_hook after a new library is loaded. + // No-op when socket hooks are not active. + static inline void install_socket_hooks() { + if (_socket_active.load(std::memory_order_acquire)) { + patch_socket_functions(); + } + } }; #else @@ -47,8 +69,11 @@ class LibraryPatcher { static void patch_libraries() { } static void unpatch_libraries() { } static void patch_sigaction() { } + static bool patch_socket_functions() { return false; } + static void unpatch_socket_functions() { } + static void install_socket_hooks() { } }; #endif -#endif // _LIBRARYPATCHER_H \ No newline at end of file +#endif // _LIBRARYPATCHER_H diff --git a/ddprof-lib/src/main/cpp/libraryPatcher_linux.cpp b/ddprof-lib/src/main/cpp/libraryPatcher_linux.cpp index cfb6d1802..0772a0650 100644 --- a/ddprof-lib/src/main/cpp/libraryPatcher_linux.cpp +++ b/ddprof-lib/src/main/cpp/libraryPatcher_linux.cpp @@ -7,8 +7,9 @@ #ifdef __linux__ #include "counters.h" -#include "profiler.h" #include "guards.h" +#include "nativeSocketSampler.h" +#include "profiler.h" #include #include @@ -25,6 +26,9 @@ PatchEntry LibraryPatcher::_patched_entries[MAX_NATIVE_LIBS]; int LibraryPatcher::_size = 0; PatchEntry LibraryPatcher::_sigaction_entries[MAX_NATIVE_LIBS]; int LibraryPatcher::_sigaction_size = 0; +PatchEntry LibraryPatcher::_socket_entries[4 * MAX_NATIVE_LIBS]; +int LibraryPatcher::_socket_size = 0; +std::atomic LibraryPatcher::_socket_active{false}; void LibraryPatcher::initialize() { if (_profiler_name == nullptr) { @@ -332,4 +336,169 @@ void LibraryPatcher::patch_sigaction() { } } +bool LibraryPatcher::patch_socket_functions() { + // Resolve the real libc symbols ONCE at first call and cache them. On a + // restart cycle (stop()→start()) we MUST NOT re-resolve via RTLD_NEXT: if + // any GOT slot in another DSO was missed during unpatch (e.g. its CodeCache + // disappeared), dlsym(RTLD_NEXT) could now resolve to the still-installed + // hook in that other DSO's GOT — the assignment to _orig_* would become + // self-referential and the next hook call would infinite-loop. + // + // RTLD_NEXT finds the first definition after this DSO in load order, + // bypassing unresolved lazy-binding stubs that would otherwise trigger + // _dl_runtime_resolve and silently overwrite the hook in the GOT. + // May resolve to an LD_PRELOAD interposer (e.g. libasan) — intentional. + // On musl, RTLD_NEXT returns NULL when libc is loaded before this DSO in the + // link map; fall back to RTLD_DEFAULT which finds symbols globally. + static NativeSocketSampler::send_fn cached_send = nullptr; + static NativeSocketSampler::recv_fn cached_recv = nullptr; + static NativeSocketSampler::write_fn cached_write = nullptr; + static NativeSocketSampler::read_fn cached_read = nullptr; + static bool cached = false; + if (!cached) { + cached_send = (NativeSocketSampler::send_fn) dlsym(RTLD_NEXT, "send"); + if (!cached_send) cached_send = (NativeSocketSampler::send_fn) dlsym(RTLD_DEFAULT, "send"); + cached_recv = (NativeSocketSampler::recv_fn) dlsym(RTLD_NEXT, "recv"); + if (!cached_recv) cached_recv = (NativeSocketSampler::recv_fn) dlsym(RTLD_DEFAULT, "recv"); + cached_write = (NativeSocketSampler::write_fn) dlsym(RTLD_NEXT, "write"); + if (!cached_write) cached_write = (NativeSocketSampler::write_fn) dlsym(RTLD_DEFAULT, "write"); + cached_read = (NativeSocketSampler::read_fn) dlsym(RTLD_NEXT, "read"); + if (!cached_read) cached_read = (NativeSocketSampler::read_fn) dlsym(RTLD_DEFAULT, "read"); + // Defensive: if any resolved address coincides with one of our hooks, the + // dynamic linker is already serving us the patched copy — refuse to cache + // and let the caller surface the failure. + if (cached_send == &NativeSocketSampler::send_hook || + cached_recv == &NativeSocketSampler::recv_hook || + cached_write == &NativeSocketSampler::write_hook || + cached_read == &NativeSocketSampler::read_hook) { + TEST_LOG("patch_socket_functions dlsym returned hook address; refusing to self-reference"); + cached_send = nullptr; cached_recv = nullptr; + cached_write = nullptr; cached_read = nullptr; + return false; + } + cached = (cached_send && cached_recv && cached_write && cached_read); + } + auto pre_send = cached_send; + auto pre_recv = cached_recv; + auto pre_write = cached_write; + auto pre_read = cached_read; + TEST_LOG("patch_socket_functions dlsym send=%p recv=%p write=%p read=%p", + (void*)pre_send, (void*)pre_recv, (void*)pre_write, (void*)pre_read); + if (!pre_send || !pre_recv || !pre_write || !pre_read) { + TEST_LOG("patch_socket_functions EARLY RETURN: at least one dlsym returned NULL"); + return false; + } + + const CodeCacheArray& native_libs = Libraries::instance()->native_libs(); + int num_of_libs = native_libs.count(); + + // Pre-resolve all library paths before acquiring the lock: realpath() may + // block on I/O and must not be called while holding _lock. + // We only need the is-self flag per library, so avoid a huge stack allocation. + static_assert(MAX_NATIVE_LIBS > 0, "MAX_NATIVE_LIBS must be positive"); + bool is_self[MAX_NATIVE_LIBS]; + int capped = (num_of_libs <= MAX_NATIVE_LIBS) ? num_of_libs : MAX_NATIVE_LIBS; + for (int index = 0; index < capped; index++) { + CodeCache* lib = native_libs.at(index); + is_self[index] = false; + if (lib == nullptr || lib->name() == nullptr) continue; + char path[PATH_MAX]; + char* rp = realpath(lib->name(), path); + is_self[index] = (rp != nullptr && strcmp(rp, _profiler_name) == 0); + } + + ExclusiveLockGuard locker(&_lock); + // Only assign orig pointers on the first call (no hooks installed yet). + // On re-entry via dlopen, RTLD_NEXT would resolve to the hook itself. + if (_socket_size == 0) { + NativeSocketSampler::setOriginalFunctions(pre_send, pre_recv, pre_write, pre_read); + } + // TODO: hook table (name + hook fn) should be owned by NativeSocketSampler; + // LibraryPatcher should iterate an externally-provided table rather than + // hardcoding the four socket hooks here. + auto try_patch_slot = [&](void** location, void* hook_fn, const char* fn_name, CodeCache* lib) { + if (location == nullptr) return; + for (int i = 0; i < _socket_size; i++) { + if (_socket_entries[i]._location == location) return; + } + if (_socket_size < 4 * MAX_NATIVE_LIBS) { + void* orig = (void*)__atomic_load_n(location, __ATOMIC_ACQUIRE); + _socket_entries[_socket_size]._lib = lib; + _socket_entries[_socket_size]._location = location; + _socket_entries[_socket_size]._func = orig; + __atomic_store_n(location, hook_fn, __ATOMIC_RELEASE); + _socket_size++; + } else { + Log::warn("socket patch table full (%d slots), skipping %s in %s", 4 * MAX_NATIVE_LIBS, fn_name, lib ? lib->name() : "?"); + } + }; + for (int index = 0; index < capped; index++) { + CodeCache* lib = native_libs.at(index); + if (lib == nullptr) continue; + if (lib->name() == nullptr) continue; + + if (is_self[index]) { + continue; + } + + void** send_location = (void**)lib->findImport(im_send); + void** recv_location = (void**)lib->findImport(im_recv); + void** write_location = (void**)lib->findImport(im_write); + void** read_location = (void**)lib->findImport(im_read); + + if (send_location == nullptr && recv_location == nullptr + && write_location == nullptr && read_location == nullptr) continue; + + TEST_LOG("patch_socket_functions PATCH %s send=%p recv=%p write=%p read=%p", + lib->name(), (void*)send_location, (void*)recv_location, + (void*)write_location, (void*)read_location); + + // The _lock is held during patching to protect _socket_entries and _socket_size. + // Concurrent dlopen_hook calls serialize via the same lock in install_socket_hooks(), + // ensuring slot_patched checks and updates are atomic with respect to each other. + try_patch_slot(send_location, (void*)NativeSocketSampler::send_hook, "send", lib); + try_patch_slot(recv_location, (void*)NativeSocketSampler::recv_hook, "recv", lib); + try_patch_slot(write_location, (void*)NativeSocketSampler::write_hook, "write", lib); + try_patch_slot(read_location, (void*)NativeSocketSampler::read_hook, "read", lib); + } + + TEST_LOG("patch_socket_functions DONE total_slots=%d num_libs_scanned=%d", + _socket_size, capped); + _socket_active.store(true, std::memory_order_release); + return true; +} + +void LibraryPatcher::unpatch_socket_functions() { + ExclusiveLockGuard locker(&_lock); + // Restore PLT slots BEFORE clearing _socket_active so any later hook entry + // dispatches to the real libc target before observing _socket_active=false. + // Hooks already past the acquire-load on _socket_active still proceed into + // recordEvent — that race is intrinsic to PLT teardown; the surrounding + // ExclusiveLockGuard plus the in-flight check inside recordSample bound the + // window to one in-flight call per CPU. + // + // ASSUMPTION (dlclose UAF): we write through _socket_entries[i]._location + // without checking that the owning library is still mapped. If a patched + // DSO were actually unmapped between patch and unpatch, this store would + // corrupt freed memory or SEGV. In practice this is benign because (a) the + // host JVM does not dlclose libc-importing DSOs, (b) glibc's dlclose + // refcounts and only unmaps when the final reference is dropped, and + // (c) the same risk is already accepted by unpatch_libraries() and + // unpatch_socket_functions has the same trust model. If a host that + // routinely unmaps libc-importing libraries is ever supported, gate each + // store on a /proc/self/maps lookup or hold a dlopen handle on each lib + // for the patch lifetime. + TEST_LOG("unpatch_socket_functions restoring %d slot(s)", _socket_size); + for (int index = 0; index < _socket_size; index++) { + __atomic_store_n(_socket_entries[index]._location, _socket_entries[index]._func, __ATOMIC_RELEASE); + } + _socket_active.store(false, std::memory_order_release); + _socket_size = 0; + // _orig_send/_orig_recv/_orig_write/_orig_read are intentionally NOT nulled. + // In-flight hook invocations that entered before PLT entries were restored + // above may still be executing and will dereference these pointers. + // They remain valid (pointing to the real libc functions) until the next + // patch_socket_functions() call. +} + #endif // __linux__ diff --git a/ddprof-lib/src/main/cpp/livenessTracker.cpp b/ddprof-lib/src/main/cpp/livenessTracker.cpp index afb2c1f74..cae96fabe 100644 --- a/ddprof-lib/src/main/cpp/livenessTracker.cpp +++ b/ddprof-lib/src/main/cpp/livenessTracker.cpp @@ -211,6 +211,10 @@ void LivenessTracker::stop() { Error LivenessTracker::initialize(Arguments &args) { _enabled = args._gc_generations || args._record_liveness; + // Per-call toggles that must reflect the current start's arguments even when + // the tracker's table is reused from a prior initialize (_initialized=true). + // On musl the test JVM is not forked per test, so the singleton survives. + _record_heap_usage = args._record_heap_usage; if (!_enabled) { return Error::OK; @@ -265,8 +269,6 @@ Error LivenessTracker::initialize(Arguments &args) { // enough for 1G of heap _table = (TrackingEntry *)malloc(sizeof(TrackingEntry) * _table_cap); - _record_heap_usage = args._record_heap_usage; - _gc_epoch = 0; _last_gc_epoch = 0; diff --git a/ddprof-lib/src/main/cpp/nativeSocketSampler.cpp b/ddprof-lib/src/main/cpp/nativeSocketSampler.cpp new file mode 100644 index 000000000..8ead5c5b7 --- /dev/null +++ b/ddprof-lib/src/main/cpp/nativeSocketSampler.cpp @@ -0,0 +1,389 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include "nativeSocketSampler.h" + +#if defined(__linux__) + +#include "common.h" +#include "flightRecorder.h" +#include "libraryPatcher.h" +#include "os.h" +#include "profiler.h" +#include "tsc.h" +#include "vmEntry.h" + +#include +#include +#include +#include +#include +#include +#include + +static thread_local PoissonSampler _send_sampler; +static thread_local PoissonSampler _recv_sampler; + +// Debug-only hook-fire counters, paired with TEST_LOG (common.h). Gated at +// compile time to keep release hot paths free of cross-thread atomic writes. +#ifdef DEBUG +static std::atomic _send_hook_calls{0}; +static std::atomic _recv_hook_calls{0}; +static std::atomic _write_hook_calls{0}; +static std::atomic _read_hook_calls{0}; +static std::atomic _record_accept_calls{0}; +static std::atomic _record_reject_calls{0}; +#endif + +// intentional process-lifetime singleton — matches MallocTracer pattern; no destructor needed +NativeSocketSampler* const NativeSocketSampler::_instance = new NativeSocketSampler(); +NativeSocketSampler::send_fn NativeSocketSampler::_orig_send = nullptr; +NativeSocketSampler::recv_fn NativeSocketSampler::_orig_recv = nullptr; +NativeSocketSampler::write_fn NativeSocketSampler::_orig_write = nullptr; +NativeSocketSampler::read_fn NativeSocketSampler::_orig_read = nullptr; + +std::string NativeSocketSampler::resolveAddr(int fd) { + struct sockaddr_storage ss; + socklen_t len = sizeof(ss); + if (getpeername(fd, (struct sockaddr*)&ss, &len) != 0) { + TEST_LOG("NativeSocketSampler::resolveAddr getpeername fd=%d failed errno=%d", fd, errno); + return ""; + } + char host[INET6_ADDRSTRLEN]; + int port = 0; + if (ss.ss_family == AF_INET) { + struct sockaddr_in* s = (struct sockaddr_in*)&ss; + if (inet_ntop(AF_INET, &s->sin_addr, host, sizeof(host)) == nullptr) return ""; + port = ntohs(s->sin_port); + } else if (ss.ss_family == AF_INET6) { + struct sockaddr_in6* s = (struct sockaddr_in6*)&ss; + if (inet_ntop(AF_INET6, &s->sin6_addr, host, sizeof(host)) == nullptr) return ""; + port = ntohs(s->sin6_port); + } else { + return ""; + } + // [addr]:port — INET6_ADDRSTRLEN(46) + brackets(2) + colon(1) + port(5) + NUL(1) = 55; round to 64. + static const int FORMATTED_ADDR_BUF = 64; + char buf[FORMATTED_ADDR_BUF]; + int n; + if (ss.ss_family == AF_INET6) { + n = snprintf(buf, sizeof(buf), "[%s]:%d", host, port); + } else { + n = snprintf(buf, sizeof(buf), "%s:%d", host, port); + } + // Truncation is theoretical (buf is 64 bytes, max needed is 55), but snprintf + // already NUL-terminates on truncation; suppress unused-variable warning in release. + (void)n; + return std::string(buf); +} + +bool NativeSocketSampler::isSocket(int fd) { + // Accepts any SOCK_STREAM socket (including AF_UNIX); AF_INET/AF_INET6 filtering + // is deferred to resolveAddr() which is only called for sampled events. AF_UNIX + // will produce an empty remoteAddress field in the JFR event. + if (fd < 0) return false; + if ((size_t)fd >= (size_t)FD_TYPE_CACHE_SIZE) { + int so_type; + socklen_t solen = sizeof(so_type); + return getsockopt(fd, SOL_SOCKET, SO_TYPE, &so_type, &solen) == 0 + && so_type == SOCK_STREAM; + } + // Acquire on the gen load pairs with the release on the gen-bump in start() + // and on the cache cell store below; without it, on a weakly-ordered arch + // (aarch64) a thread could observe a freshly written cell without the matching + // gen bump (or vice versa), defeating the generation-tag invalidation contract. + uint8_t gen = _fd_cache_gen.load(std::memory_order_acquire); + uint8_t cached = _fd_type_cache[fd].load(std::memory_order_acquire); + // High nibble encodes generation; entry is valid only when it matches current gen mod 16. + if ((cached >> 4) == (gen & 0xF)) { + uint8_t type = cached & 0xF; + if (type == FD_TYPE_SOCKET) return true; + if (type == FD_TYPE_NON_SOCKET) return false; + } + + int so_type; + socklen_t solen = sizeof(so_type); + int rc = getsockopt(fd, SOL_SOCKET, SO_TYPE, &so_type, &solen); + if (rc == 0) { + bool tcp = (so_type == SOCK_STREAM); + uint8_t type = tcp ? FD_TYPE_SOCKET : FD_TYPE_NON_SOCKET; + _fd_type_cache[fd].store((uint8_t)(((gen & 0xF) << 4) | type), + std::memory_order_release); + return tcp; + } + // Only cache the non-socket verdict when getsockopt definitively says + // "not a socket" (ENOTSOCK). Transient errors (EBADF on a racing close, + // EINTR, etc.) must NOT poison the cache: a sticky misclassification + // would survive fd reuse via dup2() and silently suppress sampling for + // the rest of the session. + if (errno == ENOTSOCK) { + _fd_type_cache[fd].store((uint8_t)(((gen & 0xF) << 4) | FD_TYPE_NON_SOCKET), + std::memory_order_release); + } + return false; +} + +bool NativeSocketSampler::shouldSample(u64 duration_ticks, int op, float &weight) { + PoissonSampler &sampler = (op == 0) ? _send_sampler : _recv_sampler; + return sampler.sample(duration_ticks, + (u64)_rate_limiter.interval(), + _rate_limiter.epoch(), + weight); +} + +void NativeSocketSampler::recordEvent(int fd, u64 t0, u64 t1, ssize_t bytes, u8 op) { + float weight = 0.0f; + bool sampled = shouldSample(t1 - t0, op, weight); + if (!sampled) { +#ifdef DEBUG + uint64_t n = _record_reject_calls.fetch_add(1, std::memory_order_relaxed); + if (n < 3 || (n & 0x3FF) == 0) { + TEST_LOG("NativeSocketSampler::recordEvent REJECT #%llu fd=%d op=%u bytes=%zd dur_ticks=%llu", + (unsigned long long)(n + 1), fd, (unsigned)op, bytes, + (unsigned long long)(t1 - t0)); + } +#endif + return; + } +#ifdef DEBUG + { + uint64_t n = _record_accept_calls.fetch_add(1, std::memory_order_relaxed); + if (n < 3 || (n & 0x3F) == 0) { + TEST_LOG("NativeSocketSampler::recordEvent ACCEPT #%llu fd=%d op=%u bytes=%zd dur_ticks=%llu weight=%f", + (unsigned long long)(n + 1), fd, (unsigned)op, bytes, + (unsigned long long)(t1 - t0), (double)weight); + } + } +#endif + + NativeSocketEvent event; + event._start_time = t0; + event._end_time = t1; + event._operation = op; + event._remote_addr[0] = '\0'; + { + std::lock_guard lock(_fd_cache_mutex); + auto it = _fd_cache.find(fd); + if (it != _fd_cache.end()) { + strncpy(event._remote_addr, it->second.c_str(), sizeof(event._remote_addr) - 1); + event._remote_addr[sizeof(event._remote_addr) - 1] = '\0'; + } + } + // TOCTOU: resolveAddr runs without the lock; concurrent emplace is safe (first writer wins). + // Skip resolveAddr entirely once the cache is full — otherwise the hot path + // degrades to per-event getpeername+inet_ntop+std::string for fds that will + // never be cached. + if (event._remote_addr[0] == '\0' && !_fd_cache_full.load(std::memory_order_acquire)) { + std::string resolved = resolveAddr(fd); + strncpy(event._remote_addr, resolved.c_str(), sizeof(event._remote_addr) - 1); + event._remote_addr[sizeof(event._remote_addr) - 1] = '\0'; + std::lock_guard lock(_fd_cache_mutex); + if ((int)_fd_cache.size() < MAX_FD_CACHE) { + _fd_cache.emplace(fd, resolved); + if ((int)_fd_cache.size() >= MAX_FD_CACHE) { + _fd_cache_full.store(true, std::memory_order_release); + } + } + } + // ret > 0 checked above; cast is safe. + event._bytes = (u64)bytes; + event._weight = weight; + + // Pass NULL ucontext (mirrors MallocTracer): recordSample routes + // BCI_NATIVE_SOCKET to walkVM with no signal context, which walks native + // frames via DWARF/FP and falls back to JavaFrameAnchor for Java frames. + // No isRunning() guard here: hook bodies check _socket_active before calling + // recordEvent(), so in-flight calls that slip through during unpatch are + // benign — recordSample() will simply fail to acquire a lock slot and return. + Profiler::instance()->recordSample(NULL, (u64)bytes, OS::threadId(), + BCI_NATIVE_SOCKET, 0, &event); + + _rate_limiter.recordFire(); +} + +ssize_t NativeSocketSampler::send_hook(int fd, const void* buf, size_t len, int flags) { + // Defensive guard against direct invocation outside of PLT dispatch (e.g. tests + // that obtain the static symbol address before LibraryPatcher::patch_socket_functions + // has run). Production hooks are unreachable until setOriginalFunctions() has been + // called under _lock, so this branch is not exercised on the normal path. + send_fn fn = _orig_send; + if (fn == nullptr) { errno = ENOSYS; return -1; } + if (!LibraryPatcher::_socket_active.load(std::memory_order_acquire)) return fn(fd, buf, len, flags); +#ifdef DEBUG + { + uint64_t n = _send_hook_calls.fetch_add(1, std::memory_order_relaxed); + if (n < 3 || (n & 0x3FF) == 0) { + TEST_LOG("NativeSocketSampler::send_hook #%llu fd=%d len=%zu flags=0x%x", + (unsigned long long)(n + 1), fd, len, flags); + } + } +#endif + u64 t0 = TSC::ticks(); + return record_if_positive(fd, fn(fd, buf, len, flags), t0, TSC::ticks(), 0); +} + +ssize_t NativeSocketSampler::recv_hook(int fd, void* buf, size_t len, int flags) { + recv_fn fn = _orig_recv; + if (fn == nullptr) { errno = ENOSYS; return -1; } + if (!LibraryPatcher::_socket_active.load(std::memory_order_acquire)) return fn(fd, buf, len, flags); +#ifdef DEBUG + { + uint64_t n = _recv_hook_calls.fetch_add(1, std::memory_order_relaxed); + if (n < 3 || (n & 0x3FF) == 0) { + TEST_LOG("NativeSocketSampler::recv_hook #%llu fd=%d len=%zu flags=0x%x", + (unsigned long long)(n + 1), fd, len, flags); + } + } +#endif + u64 t0 = TSC::ticks(); + return record_if_positive(fd, fn(fd, buf, len, flags), t0, TSC::ticks(), 1); +} + +ssize_t NativeSocketSampler::write_hook(int fd, const void* buf, size_t len) { + write_fn fn = _orig_write; + if (fn == nullptr) { errno = ENOSYS; return -1; } + if (!LibraryPatcher::_socket_active.load(std::memory_order_acquire)) return fn(fd, buf, len); + NativeSocketSampler* self = _instance; + if (!self->isSocket(fd)) { +#ifdef DEBUG + uint64_t n = _write_hook_calls.fetch_add(1, std::memory_order_relaxed); + if (n < 3 || (n & 0x3FF) == 0) { + TEST_LOG("NativeSocketSampler::write_hook #%llu fd=%d len=%zu is_socket=0", + (unsigned long long)(n + 1), fd, len); + } +#endif + return fn(fd, buf, len); + } +#ifdef DEBUG + { + uint64_t n = _write_hook_calls.fetch_add(1, std::memory_order_relaxed); + if (n < 3 || (n & 0x3FF) == 0) { + TEST_LOG("NativeSocketSampler::write_hook #%llu fd=%d len=%zu is_socket=1", + (unsigned long long)(n + 1), fd, len); + } + } +#endif + u64 t0 = TSC::ticks(); + return record_if_positive(fd, fn(fd, buf, len), t0, TSC::ticks(), 2); +} + +ssize_t NativeSocketSampler::read_hook(int fd, void* buf, size_t len) { + read_fn fn = _orig_read; + if (fn == nullptr) { errno = ENOSYS; return -1; } + if (!LibraryPatcher::_socket_active.load(std::memory_order_acquire)) return fn(fd, buf, len); + NativeSocketSampler* self = _instance; + if (!self->isSocket(fd)) { +#ifdef DEBUG + uint64_t n = _read_hook_calls.fetch_add(1, std::memory_order_relaxed); + if (n < 3 || (n & 0x3FF) == 0) { + TEST_LOG("NativeSocketSampler::read_hook #%llu fd=%d len=%zu is_socket=0", + (unsigned long long)(n + 1), fd, len); + } +#endif + return fn(fd, buf, len); + } +#ifdef DEBUG + { + uint64_t n = _read_hook_calls.fetch_add(1, std::memory_order_relaxed); + if (n < 3 || (n & 0x3FF) == 0) { + TEST_LOG("NativeSocketSampler::read_hook #%llu fd=%d len=%zu is_socket=1", + (unsigned long long)(n + 1), fd, len); + } + } +#endif + u64 t0 = TSC::ticks(); + return record_if_positive(fd, fn(fd, buf, len), t0, TSC::ticks(), 3); +} + +Error NativeSocketSampler::check(Arguments &args) { + if (!args._nativesocket) { + return Error("natsock profiling not requested"); + } + return Error::OK; +} + +Error NativeSocketSampler::start(Arguments &args) { + // Clear the fd cache on start so stale entries from a prior session don't + // produce misattributed events even if stop() was not called. + clearFdCache(); + // Initial sampling period: args._nativesocket_interval (ns) when > 0, + // otherwise 1 ms default. Converted to TSC ticks (time-weighted sampling). + // + // Overflow guard: the naive (interval_ns * tsc_freq) can wrap u64 when the + // configured interval is large. At a 3 GHz TSC the product overflows around + // interval_ns ≈ 6.1 s. Compute via divide-first to keep the intermediate + // bounded: ticks = interval_ns * (freq/1e9) ≈ ns_per_tick⁻¹ * interval_ns. + long init_interval; + if (args._nativesocket_interval > 0) { + u64 ns = (u64)args._nativesocket_interval; + u64 tsc_freq = TSC::frequency(); + u64 secs = ns / 1000000000ULL; + u64 sub_ns = ns % 1000000000ULL; + u64 ticks = secs * tsc_freq + (sub_ns * tsc_freq) / 1000000000ULL; + // Clamp to LONG_MAX (the field is `long`). Anything >2^63 ticks is + // effectively "never sample" — we cap at a sentinel that the PID + // controller can still drive down. + if (ticks > (u64)LONG_MAX) ticks = (u64)LONG_MAX; + init_interval = (long)ticks; + } else { + init_interval = (long)(TSC::frequency() / 1000); + } + if (init_interval < 1) { + init_interval = DEFAULT_INTERVAL_TICKS; + } + // One limiter for all four hooks (send/write and recv/read): ~83 events/s (~5000/min) total. + // A large interval driven by heavy write traffic does not suppress long blocking reads: + // time-weighted sampling gives P = 1 - exp(-duration/interval) → 1 when duration >> interval, + // so slow calls self-select regardless of interval magnitude. Only short-duration calls + // (which carry no latency signal) are suppressed when the interval is large. + _rate_limiter.start(init_interval, TARGET_EVENTS_PER_SECOND, + PID_WINDOW_SECS, PID_P_GAIN, PID_I_GAIN, PID_D_GAIN, PID_CUTOFF_S); + // Reset fd-type cache in O(1): incrementing _fd_cache_gen invalidates all + // entries without touching the 65536-entry array (generation mismatch in + // isSocket() causes a fresh getsockopt probe on next access). + // Release pairs with the acquire in isSocket() so cache cells written after + // the bump are observed alongside the new generation on weakly-ordered archs. + _fd_cache_gen.fetch_add(1, std::memory_order_release); +#ifdef DEBUG + _send_hook_calls.store(0, std::memory_order_relaxed); + _recv_hook_calls.store(0, std::memory_order_relaxed); + _write_hook_calls.store(0, std::memory_order_relaxed); + _read_hook_calls.store(0, std::memory_order_relaxed); + _record_accept_calls.store(0, std::memory_order_relaxed); + _record_reject_calls.store(0, std::memory_order_relaxed); + TEST_LOG("NativeSocketSampler::start interval_ticks=%ld tsc_freq=%llu", + init_interval, (unsigned long long)TSC::frequency()); +#endif + if (!LibraryPatcher::patch_socket_functions()) { + return Error("failed to install native socket hooks (dlsym returned NULL)"); + } + return Error::OK; +} + +void NativeSocketSampler::stop() { +#ifdef DEBUG + TEST_LOG("NativeSocketSampler::stop summary send=%llu recv=%llu write=%llu read=%llu accept=%llu reject=%llu", + (unsigned long long)_send_hook_calls.load(std::memory_order_relaxed), + (unsigned long long)_recv_hook_calls.load(std::memory_order_relaxed), + (unsigned long long)_write_hook_calls.load(std::memory_order_relaxed), + (unsigned long long)_read_hook_calls.load(std::memory_order_relaxed), + (unsigned long long)_record_accept_calls.load(std::memory_order_relaxed), + (unsigned long long)_record_reject_calls.load(std::memory_order_relaxed)); +#endif + LibraryPatcher::unpatch_socket_functions(); + clearFdCache(); +} + +void NativeSocketSampler::clearFdCache() { + std::lock_guard lock(_fd_cache_mutex); + _fd_cache.clear(); + _fd_cache_full.store(false, std::memory_order_release); +} + +#else // !__linux__ + +NativeSocketSampler* const NativeSocketSampler::_instance = new NativeSocketSampler(); + +#endif // __linux__ diff --git a/ddprof-lib/src/main/cpp/nativeSocketSampler.h b/ddprof-lib/src/main/cpp/nativeSocketSampler.h new file mode 100644 index 000000000..9ce6368be --- /dev/null +++ b/ddprof-lib/src/main/cpp/nativeSocketSampler.h @@ -0,0 +1,218 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +#ifndef _NATIVESOCKETSAMPLER_H +#define _NATIVESOCKETSAMPLER_H + +#include "arch.h" +#include "arguments.h" +#include "engine.h" +#include "event.h" + +#if defined(__linux__) + +#include "poissonSampler.h" +#include "rateLimiter.h" +#include +#include +#include +#include + +class LibraryPatcher; + +// Synchronisation strategy +// ------------------------- +// Hook functions (send_hook / recv_hook / write_hook / read_hook) run on the +// calling Java thread, NOT in a signal handler. Therefore malloc and locking +// are safe inside hooks. +// +// fd-to-addr cache : guarded by _fd_cache_mutex (std::mutex). +// TOCTOU note: the cache is checked under lock, then +// released for resolveAddr(); a concurrent thread may +// emplace the same fd before re-acquisition. emplace() +// is idempotent in that case (first writer wins). +// Address staleness on fd reuse is accepted: worst case +// is one misattributed event per reuse. +// _fd_type_cache : std::atomic array, lock-free. Entry encoding: +// bits [7:4] = generation mod 16, bits [3:0] = type +// (0=unknown, 1=TCP socket, 2=non-TCP). Valid only when +// high nibble matches _fd_cache_gen mod 16. Stale entries +// after fd reuse cause at most one extra getsockopt() call. +// _rate_limiter : RateLimiter — owns std::atomic interval, epoch, and +// event count. PID update races are resolved by CAS +// inside RateLimiter::maybeUpdateInterval(). +// Sampling state : thread_local PoissonSampler (in nativeSocketSampler.cpp). +// No cross-thread contention; each thread maintains its +// own independent Poisson process. The per-second PID +// window observes the aggregate fire count via the shared +// atomic inside RateLimiter. +// Hook install/remove : guarded by the profiler's main state lock (MutexLocker +// in Profiler::start / Profiler::stop). No deadlock +// risk because hook bodies do NOT acquire the profiler +// signal lock. + +class NativeSocketSampler : public Engine { +public: + // Typedefs for libc send/recv/write/read signatures. + typedef ssize_t (*send_fn)(int, const void*, size_t, int); + typedef ssize_t (*recv_fn)(int, void*, size_t, int); + typedef ssize_t (*write_fn)(int, const void*, size_t); + typedef ssize_t (*read_fn)(int, void*, size_t); + + static NativeSocketSampler* instance() { return _instance; } + + Error check(Arguments &args) override; + Error start(Arguments &args) override; + void stop() override; + + // Clears the fd-to-address cache and resets the fd-type cache. + // Called from both start() (to reset state on restart) and stop(). + // Intentionally NOT called on JFR chunk boundaries. + void clearFdCache(); + + // PLT hooks installed by LibraryPatcher::patch_socket_functions(). + static ssize_t send_hook(int fd, const void* buf, size_t len, int flags); + static ssize_t recv_hook(int fd, void* buf, size_t len, int flags); + static ssize_t write_hook(int fd, const void* buf, size_t len); + static ssize_t read_hook(int fd, void* buf, size_t len); + + // Called once by LibraryPatcher::patch_socket_functions() to install the + // real libc function pointers before any PLT entries are patched. + static void setOriginalFunctions(send_fn s, recv_fn r, write_fn w, read_fn rd) { + _orig_send = s; _orig_recv = r; _orig_write = w; _orig_read = rd; + } + + // For testing only: retrieve the current original function pointers. + static void getOriginalFunctions(send_fn& s, recv_fn& r, write_fn& w, read_fn& rd) { + s = _orig_send; r = _orig_recv; w = _orig_write; rd = _orig_read; + } + +private: + static NativeSocketSampler* const _instance; + + // Set once by setOriginalFunctions() (called under _lock, before PLT patching) and + // never reset to null while hooks are active. No atomic needed: the __ATOMIC_RELEASE + // on each PLT patch provides a store-store barrier that keeps these assignments + // visible before the PLT entry becomes observable; the _socket_active release/acquire + // pair establishes happens-before for any hook that sees _socket_active=true. + // + // ASSUMPTION (documented for any future port to a weaker memory model): we do not + // restart the profiler in production — start()/stop() are exercised only in tests. + // A formal data race would only be observable on a stop()→start() restart cycle when + // a stale-epoch hook is still in flight while setOriginalFunctions() rewrites these + // pointers. On x86_64 and aarch64 aligned-pointer stores are atomic by hardware so + // no value tearing occurs in practice. If restart-in-prod ever becomes a supported + // mode, declare these as std::atomic<...> with release/acquire pairing. + static send_fn _orig_send; + static recv_fn _orig_recv; + static write_fn _orig_write; + static read_fn _orig_read; + + // Target aggregate event rate: ~83 events/s (~5000/min) across all four hooks + // (send/write and recv/read) combined. + static const int TARGET_EVENTS_PER_SECOND = 83; + static const int PID_WINDOW_SECS = 1; + + // PID controller gains. Tuned for a 1-second observation window targeting + // ~83 events/s. P=31 is proportional gain; I=511 accumulates steady-state + // error over the window; D=3 damps oscillation; cutoff=15s low-passes the + // derivative to suppress high-frequency noise. + static constexpr double PID_P_GAIN = 31.0; + static constexpr double PID_I_GAIN = 511.0; + static constexpr double PID_D_GAIN = 3.0; + static constexpr double PID_CUTOFF_S = 15.0; + + // Default sampling interval in TSC ticks (equals 1 ms only on a ~1 GHz TSC; + // serves as a numeric floor for pathologically low TSC frequencies). + static const long DEFAULT_INTERVAL_TICKS = 1000000; // fallback used in start() when the TSC-derived interval rounds to < 1 + + // Rate limiter: owns the PID controller, interval, epoch, and fire counter. + // NativeSocketSampler uses it directly (not via RateLimitedSampler) because + // it has two sampling channels (send + recv) that share one rate target but + // need independent per-thread PoissonSampler state. + RateLimiter _rate_limiter; + + // fd -> "ip:port" string cache. Bounded to MAX_FD_CACHE entries; no + // eviction is performed (entries for closed/reused fds are stale until + // the next stop(), but stale addresses are a known, accepted limitation). + // Once the cache fills, _fd_cache_full latches true to skip further + // resolveAddr() calls — without this latch the hot path would degrade to + // per-event getpeername()+inet_ntop()+std::string allocation that gets + // thrown away (since emplace is gated on size). + static const int MAX_FD_CACHE = 65536; + std::unordered_map _fd_cache; + std::mutex _fd_cache_mutex; + std::atomic _fd_cache_full{false}; + + // fd-type cache for write/read hooks. Lock-free: one atomic byte per fd number. + // Encoding: bits [7:4] = generation mod 16, bits [3:0] = type (0=unknown/invalid + // — implicit zero in fresh array, never written explicitly; 1=TCP socket; + // 2=non-TCP). An entry is valid only when its high nibble equals _fd_cache_gen + // mod 16. Incrementing _fd_cache_gen invalidates all entries in O(1) without + // touching the 65536-entry array. + // + // KNOWN LIMITATION (mod-16 generation wrap): _fd_cache_gen is only consulted via + // its low 4 bits. After 16 start() cycles the generation wraps and stale entries + // from a previous incarnation become indistinguishable from current ones until each + // fd is naturally re-probed. Profiler restarts are not exercised in production + // (only in tests), so the wrap is benign in practice. If restart-in-prod ever + // becomes a supported mode, widen _fd_cache_gen to uint32_t and store the full + // generation in a wider per-fd cell. + // Fds outside [0, FD_TYPE_CACHE_SIZE) are probed on every call. + static const int FD_TYPE_CACHE_SIZE = 65536; + // FD_TYPE_UNKNOWN is the implicit value-zero sentinel for never-written entries + // and gen-mismatch entries; it is decoded by the (cached >> 4) != gen path in + // isSocket(), not by an explicit comparison against this constant. + static const uint8_t FD_TYPE_UNKNOWN = 0; + static const uint8_t FD_TYPE_SOCKET = 1; + static const uint8_t FD_TYPE_NON_SOCKET = 2; + std::atomic _fd_cache_gen{0}; // incremented on each cache reset + std::atomic _fd_type_cache[FD_TYPE_CACHE_SIZE]; + + NativeSocketSampler() = default; + + // Resolve the peer address for fd; returns empty string on failure. + std::string resolveAddr(int fd); + + // Returns true if fd is a SOCK_STREAM socket (including AF_UNIX). + // Uses the fd-type cache; calls getsockopt on first encounter per fd. + bool isSocket(int fd); + + // Decide whether to sample and compute weight. + // Returns true if the call should be recorded; sets weight out-param. + // Implements per-thread Poisson-process sampling: each thread maintains its + // own Exp-distributed countdown; when it expires the event is sampled and a + // new countdown is drawn. weight = 1 / (1 - exp(-duration/interval)). + // duration_ticks: wall time of the I/O call in TSC ticks. + // op: 0 = send, 1 = recv. + bool shouldSample(u64 duration_ticks, int op, float &weight); + + // Common recording logic shared by all four hooks. + void recordEvent(int fd, u64 t0, u64 t1, ssize_t bytes, u8 op); + + // Records the event if ret > 0; returns ret unchanged. Shared tail for all four hooks. + static inline ssize_t record_if_positive(int fd, ssize_t ret, u64 t0, u64 t1, u8 op) { + if (ret > 0) _instance->recordEvent(fd, t0, t1, ret, op); + return ret; + } +}; + +#else // !__linux__ + +class NativeSocketSampler : public Engine { +public: + static NativeSocketSampler* instance() { return _instance; } + Error check(Arguments &args) override { return Error::OK; } + Error start(Arguments &args) override { return Error::OK; } + void stop() override {} + void clearFdCache() {} +private: + static NativeSocketSampler* const _instance; + NativeSocketSampler() {} +}; + +#endif // __linux__ + +#endif // _NATIVESOCKETSAMPLER_H diff --git a/ddprof-lib/src/main/cpp/poissonSampler.h b/ddprof-lib/src/main/cpp/poissonSampler.h new file mode 100644 index 000000000..466a2fc71 --- /dev/null +++ b/ddprof-lib/src/main/cpp/poissonSampler.h @@ -0,0 +1,233 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +#ifndef _POISSONSAMPLER_H +#define _POISSONSAMPLER_H + +#include "arch.h" +#include + +/** + * @file poissonSampler.h + * + * Thread-local, dimension-agnostic Poisson-process sampler. + * + * ## Concept + * + * Many profiler engines need to sub-sample a high-frequency stream of + * measurements and produce an unbiased aggregate estimate from the surviving + * samples. The classic approach is a deterministic threshold counter (fire + * every N units), but that creates phase-locking artifacts and handles + * measurements that span multiple intervals poorly. + * + * PoissonSampler models the stream as a Poisson process: between consecutive + * sample points the gap is drawn independently from Exp(mean = @p interval). + * This guarantees memoryless inter-arrival times and correct behaviour for + * measurements of any size relative to the interval. + * + * ## Sampling decision + * + * The sampler maintains a monotonically growing accumulator @c _used and a + * Exp-distributed threshold @c _threshold. On each call: + * + * 1. @p value is added to @c _used. + * 2. If @c _used < @c _threshold: no sample, return false. + * 3. Otherwise: the threshold has been crossed. Advance @c _threshold by + * a fresh draw from Exp(@p interval) so the next gap is independent. + * Compute and return the weight (see below), return true. + * + * ## Weight formula and estimator invariant + * + * The probability that a Poisson process with rate 1/@p interval produces + * at least one event during an interval of length @p value is: + * + * P = 1 - exp(-value / interval) + * + * The inverse-transform weight is: + * + * weight = 1 / P = 1 / (1 - exp(-value / interval)) + * + * This satisfies the unbiasedness invariant for every measurement: + * + * E[weight * value | sampled] * P(sampled) = (1/P) * value * P = value + * + * Therefore sum(weight_i * value_i) over all sampled events is an unbiased + * estimator of the total accumulated dimension over the recording window, + * regardless of the distribution of individual measurement sizes: + * + * - value << interval → P ≈ value/interval, weight ≈ interval/value, + * weight * value ≈ interval (one interval per event) + * - value >> interval → P ≈ 1, weight ≈ 1, + * weight * value ≈ value (full measurement credited) + * + * ## Dimension agnosticism + * + * The class is generic over the accumulated dimension. The caller chooses + * what @p value represents; @p interval must be in the same units: + * + * - TSC ticks → latency / time-in-I/O profiling + * - Bytes → throughput / allocation profiling + * - Event count → frequency profiling (pass value = 1 per event) + * + * ## Thread safety and epoch-based reset + * + * Instances must be declared @c thread_local. All state is private to the + * owning thread; no locks or atomics are used in the hot path. + * + * To support profiler restart, the engine exposes a shared + * @c std::atomic epoch counter that it increments on @c start(). + * Each PoissonSampler caches the last seen epoch; when it differs the + * sampler reinitialises lazily on the next @c sample() call, re-seeding + * the PRNG and resetting the accumulator and threshold. + * + * ## Usage + * + * @code + * // Engine header: + * std::atomic _epoch{0}; // bumped in start() + * std::atomic _interval; // PID-controlled mean gap + * + * // Engine translation unit: + * static thread_local PoissonSampler _send_sampler; + * static thread_local PoissonSampler _recv_sampler; + * + * // In the hook / measurement path: + * float weight; + * if (_send_sampler.sample(value, (u64)_interval.load(), _epoch.load(), weight)) { + * // record event; sum(weight * value) estimates total accumulated value + * } + * @endcode + */ +class PoissonSampler { +public: + /** + * Decide whether to sample this measurement and compute its weight. + * + * @param value The measurement for this call, in the chosen unit. + * Must be in the same units as @p interval. + * A value of 0 is never sampled. + * @param interval Mean inter-sample gap in the same units as @p value. + * Controls the average number of events recorded per + * unit of accumulated @p value. Must be > 0. + * @param epoch_now Current profiler epoch from the owning engine's + * shared atomic. When it differs from the cached epoch + * the sampler resets all state before evaluating the + * current measurement. + * @param weight [out] Set on true return to 1/(1-exp(-value/interval)). + * Multiply by @p value to get this event's contribution + * to the total-accumulated-value estimate. + * @return true if this measurement should be recorded. + */ + bool sample(u64 value, u64 interval, u64 epoch_now, float &weight) { + if (value == 0 || interval == 0) { + return false; + } + if (_epoch != epoch_now) { + reset(interval, epoch_now); + } + _used += value; + if (_used < _threshold) { + return false; + } + // Threshold crossed: advance by a fresh Exp draw so the next + // inter-arrival gap is independent of all previous ones. + _threshold += nextExp(interval); + // Float precision: when value >> interval, expf(-value/interval) rounds to 0.0f, + // so weight = 1.0f / (1.0f - 0.0f) = 1.0f. This is a conservative lower bound — + // large events count as weight >= 1.0. Intentional; avoids the cost of double arithmetic. + float p = 1.0f - expf(-(float)value / (float)interval); + weight = (p > 0.0f) ? 1.0f / p : 1.0f; + return true; + } + +private: + u64 _epoch{0}; // last seen profiler epoch; 0 = not yet initialised + u64 _used{0}; // accumulated value since the last threshold crossing + u64 _threshold{0}; // next Exp-distributed threshold to cross + u64 _rng{0}; // xorshift64 PRNG state; must never be 0 + + /** + * Reinitialise all state for a new profiling session. + * + * The PRNG is seeded from the instance's own address XOR'd with a + * multiple of the new epoch. Because @c thread_local instances live at + * fixed but thread-specific addresses, and because distinct samplers + * within the same thread occupy different addresses, each (thread, + * sampler, session) triple receives an independent random stream. + */ + void reset(u64 interval, u64 epoch_now) { + // Fibonacci hashing of epoch_now spreads low-entropy epoch values + // across the full 64-bit range before XOR-ing with the address. + _rng = (u64)(uintptr_t)this ^ (epoch_now * 0x9e3779b97f4a7c15ULL); + if (_rng == 0) _rng = 1; // xorshift64 must not start at 0 + _used = 0; + _threshold = nextExp(interval); + _epoch = epoch_now; + } + + /** + * Draw one sample from Exp(mean = @p interval). + * + * Uses xorshift64 to produce a uniform pseudo-random value, then applies + * the inverse CDF of the exponential distribution: + * + * X = -interval * ln(U), U ~ Uniform(0, 1] + * + * U > 0 because the +0.5 offset in `((double)_rng + 0.5) * 2^-64` ensures the + * product is strictly positive even if _rng were 0. The xorshift64 invariant + * (_rng != 0) is independently required by the recurrence (0 is a fixed point). + * The magic constant 5.421010862427522e-20 ≈ 1/2^64 converts a u64 to [0, 1]. + * + * Use `double` (53 mantissa bits) rather than `float` (24): a 24-bit float + * scaled by 2^-64 clamps the smallest representable U to ~5e-21, capping the + * longest possible Exp draw at ~47*interval and biasing the inter-arrival + * distribution toward shorter gaps under fast traffic. With double, the smallest + * representable U is ~5e-20, which permits draws up to ~44*interval — but the + * truncation `(u64)(...)` no longer rounds away small Exp values to zero. + * + * ### Why xorshift64 instead of a C++ standard generator? + * + * The C++ facility (std::mt19937, std::minstd_rand, …) is + * unsuitable here for several reasons: + * + * 1. **Hot-path overhead.** nextExp() is called only once per fired + * event (~83 times/second at the default rate), so raw throughput + * is not the primary concern. The concern is code-size and + * instruction-cache pressure: std::mt19937 carries ~2.5 KB of + * state and its generate step touches all of it. xorshift64 fits + * in a single 8-byte field already present in the struct. + * + * 2. **Seeding.** std::random_device — the canonical seed source — + * may block, throw, or return low-entropy values on some Linux + * configurations (e.g., early boot, containers without /dev/urandom + * entropy). Our seed (instance address XOR epoch hash) is always + * available, zero-cost, and produces independent streams per thread + * and per profiling session without any OS interaction. + * + * 3. **No allocation, no exceptions.** std::random_device and the + * distribution wrappers (std::uniform_real_distribution, etc.) may + * allocate and may throw. This code runs inside PLT hooks that + * intercept arbitrary application threads; allocation and exception + * handling in that context would be unsafe. + * + * 4. **Statistical sufficiency.** xorshift64 (Marsaglia 2003) passes the + * Diehard battery; it fails some BigCrush tests for linear-algebra-based + * statistics (MatrixRank, LinearComp), but those failure modes are + * irrelevant to inverse-CDF Exp sampling for aggregate weight estimates. + * The inverse-CDF transform amplifies non-uniformity only near U ≈ 0 + * (i.e., extremely large Exp draws), which correspond to very long + * inter-sample gaps — a rare tail that has negligible effect on aggregate + * estimates. + */ + u64 nextExp(u64 interval) { + _rng ^= _rng << 13; + _rng ^= _rng >> 7; + _rng ^= _rng << 17; + double u = ((double)_rng + 0.5) * 5.421010862427522e-20; + return (u64)(-(double)interval * log(u)); + } +}; + +#endif // _POISSONSAMPLER_H diff --git a/ddprof-lib/src/main/cpp/profiler.cpp b/ddprof-lib/src/main/cpp/profiler.cpp index c70b5df17..998762485 100644 --- a/ddprof-lib/src/main/cpp/profiler.cpp +++ b/ddprof-lib/src/main/cpp/profiler.cpp @@ -8,6 +8,7 @@ #include "profiler.h" #include "asyncSampleMutex.h" #include "mallocTracer.h" +#include "nativeSocketSampler.h" #include "context.h" #include "context_api.h" #include "guards.h" @@ -705,6 +706,7 @@ void *Profiler::dlopen_hook(const char *filename, int flags) { // Patch sigaction in newly loaded libraries LibraryPatcher::patch_sigaction(); MallocTracer::installHooks(); + LibraryPatcher::install_socket_hooks(); // Extract build-ids for newly loaded libraries if remote symbolication is enabled Profiler* profiler = instance(); if (profiler != nullptr && profiler->_remote_symbolication) { @@ -1062,7 +1064,8 @@ Error Profiler::start(Arguments &args, bool reset) { (args._record_allocations || args._record_liveness || args._gc_generations ? EM_ALLOC : 0) | - (args._nativemem >= 0 ? EM_NATIVEMEM : 0); + (args._nativemem >= 0 ? EM_NATIVEMEM : 0) | + (args._nativesocket ? EM_NATIVESOCKET : 0); if (_event_mask == 0) { return Error("No profiling events specified"); @@ -1237,6 +1240,15 @@ Error Profiler::start(Arguments &args, bool reset) { activated |= EM_NATIVEMEM; } } + if (_event_mask & EM_NATIVESOCKET) { + error = NativeSocketSampler::instance()->start(args); + if (error) { + Log::warn("%s", error.message()); + error = Error::OK; // recoverable + } else { + activated |= EM_NATIVESOCKET; + } + } if (activated) { switchThreadEvents(JVMTI_ENABLE); @@ -1277,6 +1289,8 @@ Error Profiler::stop() { _alloc_engine->stop(); if (_event_mask & EM_NATIVEMEM) malloc_tracer.stop(); + if (_event_mask & EM_NATIVESOCKET) + NativeSocketSampler::instance()->stop(); if (_event_mask & EM_WALL) _wall_engine->stop(); if (_event_mask & EM_CPU) @@ -1328,6 +1342,9 @@ Error Profiler::check(Arguments &args) { if (!error && args._nativemem >= 0) { error = malloc_tracer.check(args); } + if (!error && args._nativesocket) { + error = NativeSocketSampler::instance()->check(args); + } if (!error) { if (args._cstack == CSTACK_DWARF && !DWARF_SUPPORTED) { return Error("DWARF unwinding is not supported on this platform"); diff --git a/ddprof-lib/src/main/cpp/rateLimiter.h b/ddprof-lib/src/main/cpp/rateLimiter.h new file mode 100644 index 000000000..daa4fa67e --- /dev/null +++ b/ddprof-lib/src/main/cpp/rateLimiter.h @@ -0,0 +1,119 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +#ifndef _RATELIMITER_H +#define _RATELIMITER_H + +#include "arch.h" +#include "os.h" +#include "pidController.h" +#include + +/** + * Thread-safe rate limiter based on a PID controller. + * + * Maintains a shared target event rate by adjusting a sampling interval + * (in arbitrary units — TSC ticks, bytes, counts, …) via a PID feedback + * loop driven by the aggregate observed fire count across all threads. + * + * ## Typical use + * + * One RateLimiter instance is shared across threads. Per-thread sampling + * decisions are made by a companion sampler (e.g. PoissonSampler) that + * reads interval() and epoch() from this object. After each sampled event + * the thread calls recordFire() to feed back into the rate controller. + * + * ## Epoch-based lazy reset + * + * start() bumps an epoch counter. Per-thread samplers compare their cached + * epoch against epoch() on every sample call; a mismatch triggers a lazy + * reinitialisation, so no explicit iteration over threads is needed at start. + */ +class RateLimiter { +public: + RateLimiter() = default; + + /** + * Initialise for a new profiling session. + * + * @param init_interval_units Initial sampling interval in the chosen unit + * (e.g. TSC ticks for ~1 ms). Must be >= 1. + * @param target_per_second Target aggregate fire rate (events / second). + * @param pid_window_secs PID observation window in seconds. + * @param p_gain PID proportional gain. + * @param i_gain PID integral gain. + * @param d_gain PID derivative gain. + * @param cutoff_secs PID derivative low-pass cutoff in seconds. + */ + void start(long init_interval_units, + u64 target_per_second, + int pid_window_secs, + double p_gain, double i_gain, double d_gain, + double cutoff_secs) { + _interval.store(init_interval_units, std::memory_order_release); + _event_count.store(0, std::memory_order_relaxed); + _last_update_ns.store(OS::nanotime(), std::memory_order_release); + _epoch.fetch_add(1, std::memory_order_release); + _pid = PidController(target_per_second, p_gain, i_gain, d_gain, + pid_window_secs, cutoff_secs); + } + + /** Current sampling interval in the chosen unit. */ + long interval() const { + return _interval.load(std::memory_order_relaxed); + } + + /** Current epoch; bumped on every start(). */ + u64 epoch() const { + return _epoch.load(std::memory_order_relaxed); + } + + /** + * Record one sampled event and update the PID controller at most once + * per second. Safe to call from any thread concurrently. + */ + void recordFire() { + _event_count.fetch_add(1, std::memory_order_relaxed); + maybeUpdateInterval(); + } + +private: + static const u64 ONE_SECOND_NS = 1000000000ULL; + + std::atomic _interval{1}; + std::atomic _epoch{0}; + std::atomic _event_count{0}; + std::atomic _last_update_ns{0}; + PidController _pid{1, 1.0, 1.0, 1.0, 1, 1.0}; + + void maybeUpdateInterval() { + u64 now = OS::nanotime(); + u64 prev = _last_update_ns.load(std::memory_order_relaxed); + if (now - prev < ONE_SECOND_NS) { + return; + } + if (!_last_update_ns.compare_exchange_strong(prev, now, + std::memory_order_acq_rel, + std::memory_order_relaxed)) { + return; + } + // One-event-per-window imprecision: a concurrent recordFire() after this exchange + // loses its count for this window. Accepted: the PID controller tolerates this level + // of measurement noise without instability. + long count = _event_count.exchange(0, std::memory_order_relaxed); + double signal = _pid.compute(static_cast(count), 1.0); + long new_interval = _interval.load(std::memory_order_relaxed) + - static_cast(signal); + if (new_interval < 1) { + new_interval = 1; + } + // Relaxed store: eventual consistency is acceptable. Threads reading _interval + // with relaxed loads will see the update within at most one additional window. + // Forcing release ordering here would add unnecessary cost on weak-ordering architectures. + _interval.store(new_interval, std::memory_order_relaxed); + } +}; + +#endif // _RATELIMITER_H diff --git a/ddprof-lib/src/main/cpp/vmEntry.h b/ddprof-lib/src/main/cpp/vmEntry.h index 6be4c87bb..2a346a0f3 100644 --- a/ddprof-lib/src/main/cpp/vmEntry.h +++ b/ddprof-lib/src/main/cpp/vmEntry.h @@ -34,6 +34,7 @@ enum ASGCT_CallFrameType { BCI_ERROR = -18, // method_id is an error string BCI_NATIVE_FRAME_REMOTE = -19, // method_id points to RemoteFrameInfo for remote symbolication BCI_NATIVE_MALLOC = -20, // native malloc/free sample (size stored in counter) + BCI_NATIVE_SOCKET = -21, // native socket I/O sample (bytes stored in counter) }; // See hotspot/src/share/vm/prims/forte.cpp diff --git a/ddprof-lib/src/test/cpp/nativeSocketSampler_ut.cpp b/ddprof-lib/src/test/cpp/nativeSocketSampler_ut.cpp new file mode 100644 index 000000000..31400da99 --- /dev/null +++ b/ddprof-lib/src/test/cpp/nativeSocketSampler_ut.cpp @@ -0,0 +1,376 @@ +/* + * 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. + */ + +#include + +#if defined(__linux__) + +#include "nativeSocketSampler.h" +#include "libraryPatcher.h" + +#include +#include + +// --------------------------------------------------------------------------- +// Stub tracking +// --------------------------------------------------------------------------- + +static std::atomic g_send_calls{0}; +static std::atomic g_recv_calls{0}; +static std::atomic g_send_ret{-1}; +static std::atomic g_recv_ret{-1}; +static std::atomic g_write_calls{0}; +static std::atomic g_read_calls{0}; +static std::atomic g_write_ret{-1}; +static std::atomic g_read_ret{-1}; + +static ssize_t stub_send(int /*fd*/, const void* /*buf*/, size_t /*len*/, int /*flags*/) { + g_send_calls++; + return g_send_ret.load(); +} + +static ssize_t stub_recv(int /*fd*/, void* /*buf*/, size_t /*len*/, int /*flags*/) { + g_recv_calls++; + return g_recv_ret.load(); +} + +static ssize_t stub_write(int /*fd*/, const void* /*buf*/, size_t /*len*/) { + g_write_calls++; + return g_write_ret.load(); +} + +static ssize_t stub_read(int /*fd*/, void* /*buf*/, size_t /*len*/) { + g_read_calls++; + return g_read_ret.load(); +} + +// --------------------------------------------------------------------------- +// Test fixture — installs stubs as the "original" function pointers so the +// hooks invoke them without needing GOT patching or a running JVM. +// --------------------------------------------------------------------------- + +class NativeSocketSamplerHookTest : public ::testing::Test { +protected: + NativeSocketSampler::send_fn _saved_send; + NativeSocketSampler::recv_fn _saved_recv; + NativeSocketSampler::write_fn _saved_write; + NativeSocketSampler::read_fn _saved_read; + + void SetUp() override { + NativeSocketSampler::getOriginalFunctions(_saved_send, _saved_recv, _saved_write, _saved_read); + NativeSocketSampler::setOriginalFunctions(stub_send, stub_recv, stub_write, stub_read); + g_send_calls = 0; + g_recv_calls = 0; + g_write_calls = 0; + g_read_calls = 0; + } + + void TearDown() override { + NativeSocketSampler::setOriginalFunctions(_saved_send, _saved_recv, _saved_write, _saved_read); + NativeSocketSampler::send_fn cur_send; NativeSocketSampler::recv_fn cur_recv; + NativeSocketSampler::write_fn cur_write; NativeSocketSampler::read_fn cur_read; + NativeSocketSampler::getOriginalFunctions(cur_send, cur_recv, cur_write, cur_read); + ASSERT_EQ(cur_send, _saved_send) << "Function pointers must be restored in TearDown"; + } +}; + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +/** + * Verifies that send_hook forwards the call to _orig_send and returns its + * return value when the original function indicates failure (ret <= 0 so the + * sampling path — which needs a JVM — is not entered). + */ +TEST_F(NativeSocketSamplerHookTest, SendHookCallsOrigSendAndReturnsValue) { + g_send_ret = -1; // error path avoids the JVM-dependent sampling code + char buf[16] = {}; + + ssize_t ret = NativeSocketSampler::send_hook(0, buf, sizeof(buf), 0); + + EXPECT_EQ(g_send_calls.load(), 1) << "send_hook must call _orig_send exactly once"; + EXPECT_EQ(ret, -1) << "send_hook must propagate the return value from _orig_send"; +} + +/** + * Verifies that recv_hook forwards the call to _orig_recv and returns its + * return value (same guard: ret <= 0 skips the JVM path). + */ +TEST_F(NativeSocketSamplerHookTest, RecvHookCallsOrigRecvAndReturnsValue) { + g_recv_ret = 0; + char buf[16] = {}; + + ssize_t ret = NativeSocketSampler::recv_hook(0, buf, sizeof(buf), 0); + + EXPECT_EQ(g_recv_calls.load(), 1) << "recv_hook must call _orig_recv exactly once"; + EXPECT_EQ(ret, 0) << "recv_hook must propagate the return value from _orig_recv"; +} + +/** + * Verifies that write_hook forwards the call to _orig_write and returns its + * return value when the original function indicates failure (ret <= 0 skips + * the sampling path that requires a running JVM). + * + * fd=0 (stdin) is not a socket descriptor, so getsockopt fails and isSocket() + * returns false; recordEvent() is never reached — the non-socket branch is + * exercised here. Note: AF_UNIX SOCK_STREAM would return true from isSocket(). + */ +TEST_F(NativeSocketSamplerHookTest, WriteHookCallsOrigWriteAndReturnsValue) { + g_write_ret = -1; // error path avoids the JVM-dependent sampling code + char buf[16] = {}; + + ssize_t ret = NativeSocketSampler::write_hook(0, buf, sizeof(buf)); + + EXPECT_EQ(g_write_calls.load(), 1) << "write_hook must call _orig_write exactly once"; + EXPECT_EQ(ret, -1) << "write_hook must propagate the return value from _orig_write"; +} + +/** + * Verifies that read_hook forwards the call to _orig_read and returns its + * return value (same guard: ret <= 0 skips the JVM path). + * + * fd=0 (stdin) is not a socket descriptor, so getsockopt fails and isSocket() + * returns false; the non-socket branch is exercised. Note: AF_UNIX SOCK_STREAM + * would return true from isSocket(). + */ +TEST_F(NativeSocketSamplerHookTest, ReadHookCallsOrigReadAndReturnsValue) { + g_read_ret = 0; + char buf[16] = {}; + + ssize_t ret = NativeSocketSampler::read_hook(0, buf, sizeof(buf)); + + EXPECT_EQ(g_read_calls.load(), 1) << "read_hook must call _orig_read exactly once"; + EXPECT_EQ(ret, 0) << "read_hook must propagate the return value from _orig_read"; +} + +#endif // __linux__ + +// --------------------------------------------------------------------------- +// PoissonSampler unit tests — no Linux/glibc guard needed. +// --------------------------------------------------------------------------- + +#include "poissonSampler.h" + +/** + * Verifies that sample() returns false when value is 0 or interval is 0. + */ +TEST(PoissonSamplerTest, ReturnsFalseForZeroValueOrInterval) { + PoissonSampler s; + float w = 0.0f; + + // value == 0: must not sample regardless of interval + EXPECT_FALSE(s.sample(0, 1000, /*epoch=*/1, w)) + << "sample() must return false when value is 0"; + + // interval == 0: must not sample regardless of value + EXPECT_FALSE(s.sample(1000, 0, /*epoch=*/1, w)) + << "sample() must return false when interval is 0"; +} + +/** + * Verifies that a change in epoch causes the sampler to reset so subsequent + * calls with large value cross the freshly-drawn threshold. + */ +TEST(PoissonSamplerTest, EpochChangeResetsState) { + PoissonSampler s; + float w = 0.0f; + const u64 interval = 100; + + // Prime with epoch=1 until it fires at least once. + bool fired = false; + for (int i = 0; i < 10000 && !fired; i++) { + fired = s.sample(interval, interval, /*epoch=*/1, w); + } + ASSERT_TRUE(fired) << "Sampler should have fired during priming"; + + // Bump epoch: large value should fire quickly against the fresh threshold. + bool fired_after_reset = false; + for (int i = 0; i < 10000 && !fired_after_reset; i++) { + fired_after_reset = s.sample(interval * 1000, interval, /*epoch=*/2, w); + } + EXPECT_TRUE(fired_after_reset) + << "Sampler should fire quickly after epoch reset with large value"; +} + +/** + * Verifies that when value >> interval the computed weight approaches 1.0. + * For value = 1000 * interval, exp(-1000) ≈ 0 so P ≈ 1 and weight ≈ 1. + */ +TEST(PoissonSamplerTest, HighVolumeWeightApproachesOne) { + PoissonSampler s; + float w = 0.0f; + const u64 interval = 100; + const u64 big_value = interval * 1000; + + // Use successive epoch bumps to guarantee a fresh threshold each iteration. + bool fired = false; + for (int i = 0; i < 100 && !fired; i++) { + fired = s.sample(big_value, interval, /*epoch=*/(u64)(i + 1), w); + } + ASSERT_TRUE(fired) << "Sampler must fire with value >> interval"; + EXPECT_NEAR(w, 1.0f, 1e-3f) + << "Weight must be ~1.0 when value >> interval"; +} + +// --------------------------------------------------------------------------- +// Success-path and isSocket() tests — Linux guard required for socket APIs. +// --------------------------------------------------------------------------- + +#if defined(__linux__) + +#include +#include + +// Global counter incremented by stub_send when it returns a positive value. +// Used to verify that recordEvent is reached on the success path. +static std::atomic g_send_success_calls{0}; + +static ssize_t stub_send_success(int /*fd*/, const void* /*buf*/, size_t len, int /*flags*/) { + g_send_success_calls++; + return (ssize_t)len; // report all bytes sent +} + +/** + * Verifies that send_hook calls _orig_send and propagates a successful return + * value WHEN HOOKS ARE INACTIVE (the most common in-process state during tests). + * Because _socket_active is false the hook short-circuits to _orig_send and + * never reaches recordEvent / Profiler::recordSample — exercising recordEvent + * requires a running profiler with a recorder bound, which is not feasible in + * this gtest unit harness. + * + * To exercise the active-path short-circuit (i.e., the hook's outer guard + * branch), see SendHookActivePathReachesRecorderGuard below. + */ +TEST_F(NativeSocketSamplerHookTest, SendHookSuccessPathReturnsBytes) { + g_send_success_calls = 0; + NativeSocketSampler::setOriginalFunctions(stub_send_success, stub_recv, stub_write, stub_read); + char buf[16] = {}; + + ssize_t ret = NativeSocketSampler::send_hook(0, buf, sizeof(buf), 0); + + EXPECT_EQ(g_send_success_calls.load(), 1) + << "send_hook must call _orig_send exactly once on the inactive path"; + EXPECT_EQ(ret, (ssize_t)sizeof(buf)) + << "send_hook must propagate the byte count from _orig_send"; +} + +/** + * Verifies that with _socket_active flipped to true the hook actually takes + * the active branch (calls TSC::ticks twice and routes through record_if_positive). + * No JVM/recorder is running so recordEvent's downstream Profiler::recordSample + * is benign (returns without recording); we verify the orig fn is still called + * exactly once and the return value is propagated. + */ +TEST_F(NativeSocketSamplerHookTest, SendHookActivePathReachesRecorderGuard) { + g_send_success_calls = 0; + NativeSocketSampler::setOriginalFunctions(stub_send_success, stub_recv, stub_write, stub_read); + + // Manually flip the active flag so the hook traverses the active branch. + // Restore in a guard; tearing down the fixture must observe it cleared. + bool prev = LibraryPatcher::_socket_active.exchange(true, std::memory_order_release); + + char buf[16] = {}; + ssize_t ret = NativeSocketSampler::send_hook(0, buf, sizeof(buf), 0); + + LibraryPatcher::_socket_active.store(prev, std::memory_order_release); + + EXPECT_EQ(g_send_success_calls.load(), 1) + << "send_hook must call _orig_send exactly once on the active path"; + EXPECT_EQ(ret, (ssize_t)sizeof(buf)) + << "send_hook must propagate the byte count from _orig_send on the active path"; +} + +/** + * Verifies write_hook pass-through: when hooks are inactive (_socket_active=false), + * write_hook forwards immediately to _orig_write without calling isSocket(). + * Uses a real AF_UNIX SOCK_STREAM socketpair; the pass-through path must handle + * any fd type correctly. After closing both ends the stub is still called (the + * inactive guard fires before any fd inspection). + */ +TEST(NativeSocketSamplerIsSocketTest, UnixSocketPairReturnsFalseAfterClose) { + int fds[2]; + ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0) + << "socketpair() failed: " << strerror(errno); + + NativeSocketSampler* inst = NativeSocketSampler::instance(); + ASSERT_NE(inst, nullptr); + + // _socket_active is false — write_hook forwards to _orig_write without calling isSocket(). + // (write_hook is exercised with a stub that returns the full length, then verify the + // return value is correct.) + NativeSocketSampler::send_fn saved_send; NativeSocketSampler::recv_fn saved_recv; + NativeSocketSampler::write_fn saved_write; NativeSocketSampler::read_fn saved_read; + NativeSocketSampler::getOriginalFunctions(saved_send, saved_recv, saved_write, saved_read); + NativeSocketSampler::setOriginalFunctions(saved_send, saved_recv, + [](int /*fd*/, const void* /*buf*/, size_t len) -> ssize_t { return (ssize_t)len; }, + saved_read); + + char buf[16] = {}; + ssize_t ret = NativeSocketSampler::write_hook(fds[0], buf, sizeof(buf)); + EXPECT_EQ(ret, (ssize_t)sizeof(buf)) + << "write_hook must propagate the byte count even for AF_UNIX fds"; + + NativeSocketSampler::setOriginalFunctions(saved_send, saved_recv, saved_write, saved_read); + close(fds[0]); + close(fds[1]); +} + +#endif // __linux__ (success-path / isSocket tests) + +// --------------------------------------------------------------------------- +// Arguments parsing tests — no Linux/glibc guard needed. +// --------------------------------------------------------------------------- + +#include "arguments.h" + +/** + * Verifies that a negative natsock interval is rejected by Arguments::parse. + */ +TEST(ArgumentsNatsock, NegativeIntervalRejected) { + Arguments args; + Error e = args.parse("natsock=-1us"); + ASSERT_TRUE(static_cast(e)) << "Expected error for negative natsock interval"; + ASSERT_NE(std::string(e.message()).find("must be >= 0"), std::string::npos) + << "Error message should mention 'must be >= 0'"; +} + +/** + * Verifies that natsock with no =value (empty value) is rejected. + */ +TEST(ArgumentsNatsock, EmptyValueRejected) { + Arguments args; + Error e = args.parse("natsock"); + ASSERT_TRUE(static_cast(e)) << "Expected error for natsock with no value"; +} + +/** + * Verifies that natsock=0us is accepted (zero interval is valid). + */ +TEST(ArgumentsNatsock, ZeroIntervalAccepted) { + Arguments args; + Error e = args.parse("natsock=0us"); + ASSERT_FALSE(static_cast(e)) << "natsock=0us should be accepted"; +} + +/** + * Verifies that natsock with overflow value is rejected. + */ +TEST(ArgumentsNatsock, OverflowRejected) { + Arguments args; + Error e = args.parse("natsock=99999999999999999s"); + ASSERT_TRUE(static_cast(e)) << "Expected error for natsock with overflow value"; +} diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/throughput/NativeSocketOverheadBenchmark.java b/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/throughput/NativeSocketOverheadBenchmark.java new file mode 100644 index 000000000..b3b787f34 --- /dev/null +++ b/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/throughput/NativeSocketOverheadBenchmark.java @@ -0,0 +1,173 @@ +/* + * 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. + */ +package com.datadoghq.profiler.stresstest.scenarios.throughput; + +import com.datadoghq.profiler.JavaProfiler; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.TimeUnit; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Warmup; + +/** + * Measures overhead of the nativesocket PLT write/read hooks on: + *
    + *
  • fileWrite — write() to a regular file. After the first call the + * fd-type cache classifies it as non-socket (one atomic load per subsequent + * call). This is the worst-case overhead scenario for code that does heavy + * file I/O with nativesocket enabled. + *
  • socketWrite — write() to a TCP socket (blocking, PlainSocket). + *
  • nioSocketWrite — write() via NIO SocketChannel (JDK11+ path). + *
+ * + *

Compare {@code profilerActive=false} vs {@code profilerActive=true} to + * quantify the hook overhead. Revert if fileWrite throughput degrades > 5%. + * + *

The profiler uses time-weighted (duration-based) inverse-transform + * sampling: {@code P(sample) = 1 - exp(-duration_ticks / interval_ticks)}. + * Slow I/O calls are sampled more often; fast calls are down-sampled. + * + * 

+ *   ./gradlew :ddprof-stresstest:jmh -PjmhInclude="NativeSocketOverheadBenchmark"
+ *
+ */ +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +@Fork(value = 1, warmups = 0) +@Warmup(iterations = 3, time = 2) +@Measurement(iterations = 5, time = 3) +@State(Scope.Thread) +public class NativeSocketOverheadBenchmark { + + private static final int CHUNK = 4096; + + @Param({"false", "true"}) + public boolean profilerActive; + + private Path tmpFile; + private OutputStream fileOut; + + private ServerSocket server; + private Socket client; + private Socket serverConn; + private OutputStream sockOut; + private Thread serverAcceptor; + + private ServerSocket nioServer; + private SocketChannel nioClient; + private ByteBuffer nioBuf; + + private final byte[] buf = new byte[CHUNK]; + + @Setup(Level.Trial) + public void setup() throws Exception { + if (profilerActive) { + JavaProfiler profiler = JavaProfiler.getInstance(); + Path jfr = Files.createTempFile("nativesocket-bench", ".jfr"); + profiler.execute("start,nativesocket,jfr,file=" + jfr.toAbsolutePath()); + } + + // File I/O: regular file, will be classified non-socket after first write + tmpFile = Files.createTempFile("nativesocket-bench-file", ".bin"); + fileOut = Files.newOutputStream(tmpFile); + + // Blocking TCP socket (PlainSocket / JDK 8 send path) + server = new ServerSocket(0); + serverAcceptor = new Thread(() -> { + try { + serverConn = server.accept(); + // drain so the client write buffer never fills + InputStream drain = serverConn.getInputStream(); + byte[] dbuf = new byte[CHUNK]; + while (drain.read(dbuf) != -1) { /* drain */ } + } catch (IOException ignored) {} + }); + serverAcceptor.setDaemon(true); + serverAcceptor.start(); + client = new Socket("127.0.0.1", server.getLocalPort()); + sockOut = client.getOutputStream(); + + // NIO SocketChannel (write(2) path used by JDK 11+) + nioServer = new ServerSocket(0); + Thread nioAcceptor = new Thread(() -> { + try { + Socket ac = nioServer.accept(); + // drain + InputStream drain = ac.getInputStream(); + byte[] dbuf = new byte[CHUNK]; + while (drain.read(dbuf) != -1) { /* drain */ } + } catch (IOException ignored) {} + }); + nioAcceptor.setDaemon(true); + nioAcceptor.start(); + nioClient = SocketChannel.open(new InetSocketAddress("127.0.0.1", nioServer.getLocalPort())); + nioBuf = ByteBuffer.allocate(CHUNK); + } + + @TearDown(Level.Trial) + public void teardown() throws Exception { + if (profilerActive) { + JavaProfiler.getInstance().execute("stop"); + } + fileOut.close(); + Files.deleteIfExists(tmpFile); + client.close(); + if (serverConn != null) serverConn.close(); + server.close(); + nioClient.close(); + nioServer.close(); + } + + /** write() to a regular file — measures fd-type-cache overhead on non-socket fds. */ + @Benchmark + public void fileWrite() throws IOException { + fileOut.write(buf); + } + + /** write() to a blocking TCP socket — the socket sampling path. */ + @Benchmark + public void socketWrite() throws IOException { + sockOut.write(buf); + } + + /** SocketChannel.write() — the NIO path used by JDK 11+ java.net.Socket. */ + @Benchmark + public long nioSocketWrite() throws IOException { + nioBuf.clear(); + nioBuf.put(buf); + nioBuf.flip(); + return nioClient.write(nioBuf); + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketBytesAccuracyTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketBytesAccuracyTest.java new file mode 100644 index 000000000..92347ca03 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketBytesAccuracyTest.java @@ -0,0 +1,138 @@ +package com.datadoghq.profiler.nativesocket; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.Platform; +import org.junit.jupiter.api.Assumptions; +import org.junitpioneer.jupiter.RetryingTest; +import org.openjdk.jmc.common.item.Attribute; +import org.openjdk.jmc.common.item.IAttribute; +import org.openjdk.jmc.common.item.IItem; +import org.openjdk.jmc.common.item.IItemCollection; +import org.openjdk.jmc.common.item.IItemIterable; +import org.openjdk.jmc.common.item.IMemberAccessor; +import org.openjdk.jmc.common.unit.IQuantity; +import org.openjdk.jmc.common.unit.UnitLookup; + +import java.io.IOException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Verifies that time-weighted sampling produces a statistically reasonable + * estimate of total I/O time. With time-weighted inverse-transform sampling + * the invariant is: + * 
+ *   E[ sum(weight * duration) ] = total_io_time
+ *
+ * + * We verify that {@code sum(weight * duration_ns)} is positive and within a + * generous 100x tolerance of the actual test duration, confirming that the + * weight field reflects duration-based probability rather than a degenerate + * value. + */ +public class NativeSocketBytesAccuracyTest extends AbstractProfilerTest { + + private static final IAttribute DURATION_ATTR = + Attribute.attr("duration", "duration", "Duration", UnitLookup.TIMESPAN); + private static final IAttribute WEIGHT_ATTR = + Attribute.attr("weight", "weight", "weight", UnitLookup.NUMBER); + + @Override + protected boolean isPlatformSupported() { + return Platform.isLinux() && !Platform.isMusl(); + } + + @Override + protected String getProfilerCommand() { + // 100us period keeps sampling probability high on short localhost I/O. + return "natsock=100us"; + } + + @RetryingTest(5) + public void timeWeightedEstimateIsWithinReasonableBounds() throws Exception { + Assumptions.assumeTrue(Platform.isLinux(), "nativesocket tracking is Linux-only"); + + int payloadSize = 256 * 1024; + int iterations = 100; + + long wallStart = System.nanoTime(); + doTcpSend(payloadSize, iterations); + long wallEnd = System.nanoTime(); + long wallNs = wallEnd - wallStart; + + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.NativeSocketEvent"); + assertTrue(events.hasItems(), "No NativeSocketEvent events found"); + + double scaledDurationNs = 0.0; + long sendEventCount = 0; + for (IItemIterable items : events) { + IMemberAccessor opAccessor = OPERATION.getAccessor(items.getType()); + IMemberAccessor durationAccessor = DURATION_ATTR.getAccessor(items.getType()); + IMemberAccessor weightAccessor = WEIGHT_ATTR.getAccessor(items.getType()); + if (opAccessor == null || durationAccessor == null || weightAccessor == null) continue; + for (IItem item : items) { + String op = opAccessor.getMember(item); + // Outbound direction: SEND (send syscall) or WRITE (write syscall on socket fd). + if ("SEND".equals(op) || "WRITE".equals(op)) { + IQuantity dur = durationAccessor.getMember(item); + IQuantity weight = weightAccessor.getMember(item); + if (dur != null && weight != null) { + double durationNs = dur.doubleValueIn(UnitLookup.NANOSECOND); + scaledDurationNs += durationNs * weight.doubleValue(); + sendEventCount++; + } + } + } + } + + System.out.println("Wall time of transfers: " + wallNs + " ns"); + System.out.println("Scaled I/O time (sum of duration*weight): " + scaledDurationNs + " ns"); + System.out.println("Outbound (SEND/WRITE) event count: " + sendEventCount); + + assertTrue(sendEventCount > 0, "No outbound (SEND/WRITE) events recorded"); + assertTrue(scaledDurationNs > 0.0, "sum(weight * duration) must be positive"); + + // Generous 100x tolerance: scaled estimate must not exceed 100x wall time. + // Lower bound is not enforced because very short I/O calls may all fall + // below the sampling threshold in a brief recording window. + assertTrue(scaledDurationNs <= wallNs * 100.0, + String.format("sum(weight * duration) = %.0f ns is implausibly large (wall=%d ns)", + scaledDurationNs, wallNs)); + } + + private void doTcpSend(int payloadSize, int iterations) throws Exception { + byte[] payload = new byte[payloadSize]; + try (ServerSocket server = new ServerSocket(0)) { + int port = server.getLocalPort(); + Thread serverThread = new Thread(() -> { + for (int iter = 0; iter < iterations; iter++) { + try (Socket conn = server.accept()) { + InputStream in = conn.getInputStream(); + byte[] buf = new byte[payloadSize]; + int read = 0; + while (read < payloadSize) { + int n = in.read(buf, read, payloadSize - read); + if (n < 0) break; + read += n; + } + } catch (IOException ignored) {} + } + }); + serverThread.setDaemon(true); + serverThread.start(); + + for (int iter = 0; iter < iterations; iter++) { + try (Socket client = new Socket("127.0.0.1", port)) { + client.getOutputStream().write(payload); + client.getOutputStream().flush(); + } + } + serverThread.join(5000); + } + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketDisabledTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketDisabledTest.java new file mode 100644 index 000000000..8e74fcb02 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketDisabledTest.java @@ -0,0 +1,86 @@ +package com.datadoghq.profiler.nativesocket; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.Platform; +import org.junit.jupiter.api.Assumptions; +import org.junitpioneer.jupiter.RetryingTest; +import org.openjdk.jmc.common.item.IItemCollection; + +import java.io.IOException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; + +import static org.junit.jupiter.api.Assertions.assertFalse; + +/** + * Verifies that NativeSocketEvent events are absent when the 'nativesocket' + * profiler argument is not specified. + */ +public class NativeSocketDisabledTest extends AbstractProfilerTest { + + @Override + protected boolean isPlatformSupported() { + return Platform.isLinux(); + } + + @Override + protected String getProfilerCommand() { + // cpu-only profiling, no nativesocket + return "cpu=10ms"; + } + + @RetryingTest(3) + public void noSocketEventsWithoutFeatureEnabled() throws Exception { + Assumptions.assumeTrue(Platform.isLinux(), "nativesocket tracking is Linux-only"); + + doTcpTransfer(64 * 1024, 8); + + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.NativeSocketEvent", false); + assertFalse(events.hasItems(), + "NativeSocketEvent events must not appear when nativesocket argument is absent"); + } + + private void doTcpTransfer(int payloadSize, int iterations) throws Exception { + byte[] payload = new byte[payloadSize]; + try (ServerSocket server = new ServerSocket(0)) { + int port = server.getLocalPort(); + Thread serverThread = new Thread(() -> { + for (int iter = 0; iter < iterations; iter++) { + try (Socket conn = server.accept()) { + InputStream in = conn.getInputStream(); + byte[] buf = new byte[payloadSize]; + int read = 0; + while (read < payloadSize) { + int n = in.read(buf, read, payloadSize - read); + if (n < 0) break; + read += n; + } + conn.getOutputStream().write(buf, 0, read); + conn.getOutputStream().flush(); + } catch (IOException ignored) {} + } + }); + serverThread.setDaemon(true); + serverThread.start(); + + for (int iter = 0; iter < iterations; iter++) { + try (Socket client = new Socket("127.0.0.1", port)) { + client.getOutputStream().write(payload); + client.getOutputStream().flush(); + byte[] resp = new byte[payloadSize]; + InputStream in = client.getInputStream(); + int read = 0; + while (read < payloadSize) { + int n = in.read(resp, read, payloadSize - read); + if (n < 0) break; + read += n; + } + } + } + serverThread.join(5000); + } + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketEnabledTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketEnabledTest.java new file mode 100644 index 000000000..5f0f3e7f6 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketEnabledTest.java @@ -0,0 +1,28 @@ +package com.datadoghq.profiler.nativesocket; + +import org.junit.jupiter.api.Assumptions; +import org.junitpioneer.jupiter.RetryingTest; +import org.openjdk.jmc.common.item.IItemCollection; + +import com.datadoghq.profiler.Platform; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies that the 'nativesocket' profiler argument enables socket I/O tracking + * and that NativeSocketEvent JFR events are produced. + */ +public class NativeSocketEnabledTest extends NativeSocketTestBase { + + @RetryingTest(3) + public void socketEventsProducedWhenFeatureEnabled() throws Exception { + Assumptions.assumeTrue(Platform.isLinux(), "nativesocket tracking is Linux-only"); + + doTcpTransfer(4096, 128); + + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.NativeSocketEvent"); + assertTrue(events.hasItems(), "Expected NativeSocketEvent events to be present in JFR recording"); + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketEventFieldsTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketEventFieldsTest.java new file mode 100644 index 000000000..adba7bf45 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketEventFieldsTest.java @@ -0,0 +1,111 @@ +package com.datadoghq.profiler.nativesocket; + +import com.datadoghq.profiler.Platform; +import org.junit.jupiter.api.Assumptions; +import org.junitpioneer.jupiter.RetryingTest; +import org.openjdk.jmc.common.IMCThread; +import org.openjdk.jmc.common.IMCStackTrace; +import org.openjdk.jmc.common.item.Attribute; +import org.openjdk.jmc.common.item.IAttribute; +import org.openjdk.jmc.common.item.IItem; +import org.openjdk.jmc.common.item.IItemCollection; +import org.openjdk.jmc.common.item.IItemIterable; +import org.openjdk.jmc.common.item.IMemberAccessor; +import org.openjdk.jmc.common.unit.IQuantity; +import org.openjdk.jmc.common.unit.UnitLookup; +import org.openjdk.jmc.flightrecorder.JfrAttributes; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Verifies that each NativeSocketEvent carries all required fields with valid values: + * eventThread, stackTrace, duration, operation (SEND/RECV), remoteAddress (ip:port), + * bytesTransferred (> 0), weight (> 0). + */ +public class NativeSocketEventFieldsTest extends NativeSocketTestBase { + + private static final IAttribute REMOTE_ADDRESS = + Attribute.attr("remoteAddress", "remoteAddress", "Remote address", UnitLookup.PLAIN_TEXT); + private static final IAttribute BYTES_TRANSFERRED = + Attribute.attr("bytesTransferred", "bytesTransferred", "Bytes transferred", UnitLookup.MEMORY); + private static final IAttribute DURATION = + Attribute.attr("duration", "duration", "Duration", UnitLookup.TIMESPAN); + + @RetryingTest(3) + public void allRequiredFieldsPresentAndValid() throws Exception { + Assumptions.assumeTrue(Platform.isLinux(), "nativesocket tracking is Linux-only"); + + doTcpTransfer(4096, 128); + + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.NativeSocketEvent"); + assertTrue(events.hasItems(), "No NativeSocketEvent events found"); + + boolean foundSend = false; + boolean foundRecv = false; + + for (IItemIterable items : events) { + IMemberAccessor operationAccessor = + OPERATION.getAccessor(items.getType()); + IMemberAccessor remoteAddressAccessor = + REMOTE_ADDRESS.getAccessor(items.getType()); + IMemberAccessor bytesAccessor = + BYTES_TRANSFERRED.getAccessor(items.getType()); + IMemberAccessor weightAccessor = + WEIGHT.getAccessor(items.getType()); + IMemberAccessor durationAccessor = + DURATION.getAccessor(items.getType()); + IMemberAccessor threadAccessor = + JfrAttributes.EVENT_THREAD.getAccessor(items.getType()); + IMemberAccessor stackTraceAccessor = + STACK_TRACE.getAccessor(items.getType()); + + assertNotNull(operationAccessor, "operation field accessor must be present"); + assertNotNull(remoteAddressAccessor, "remoteAddress field accessor must be present"); + assertNotNull(bytesAccessor, "bytesTransferred field accessor must be present"); + assertNotNull(weightAccessor, "weight field accessor must be present"); + assertNotNull(durationAccessor, "duration field accessor must be present"); + assertNotNull(threadAccessor, "eventThread field accessor must be present"); + assertNotNull(stackTraceAccessor, "stackTrace field accessor must be present"); + + for (IItem item : items) { + String operation = operationAccessor.getMember(item); + assertNotNull(operation, "operation must not be null"); + // op encodes the underlying syscall: SEND/RECV are emitted by send_hook/recv_hook; + // WRITE/READ are emitted by write_hook/read_hook. Java sockets typically reach + // libc via write()/read(), so foundSend covers SEND and WRITE, foundRecv covers + // RECV and READ — both directions must be observed. + assertTrue(operation.equals("SEND") || operation.equals("RECV") + || operation.equals("WRITE") || operation.equals("READ"), + "operation must be one of SEND/RECV/WRITE/READ, got: " + operation); + if ("SEND".equals(operation) || "WRITE".equals(operation)) foundSend = true; + if ("RECV".equals(operation) || "READ".equals(operation)) foundRecv = true; + + String remoteAddress = remoteAddressAccessor.getMember(item); + assertNotNull(remoteAddress, "remoteAddress must not be null"); + assertTrue(remoteAddress.contains(":"), + "remoteAddress must be in ip:port format, got: " + remoteAddress); + + IQuantity bytes = bytesAccessor.getMember(item); + assertNotNull(bytes, "bytesTransferred must not be null"); + assertTrue(bytes.longValue() > 0, + "bytesTransferred must be > 0, got: " + bytes); + + IQuantity weight = weightAccessor.getMember(item); + assertNotNull(weight, "weight must not be null"); + assertTrue(weight.doubleValue() > 0.0, + "weight must be > 0, got: " + weight); + + IQuantity duration = durationAccessor.getMember(item); + assertNotNull(duration, "duration must not be null"); + + IMCThread thread = threadAccessor.getMember(item); + assertNotNull(thread, "eventThread must not be null"); + } + } + + assertTrue(foundSend, "Expected at least one SEND event"); + assertTrue(foundRecv, "Expected at least one RECV event"); + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketEventThreadTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketEventThreadTest.java new file mode 100644 index 000000000..0aa1d45c4 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketEventThreadTest.java @@ -0,0 +1,104 @@ +package com.datadoghq.profiler.nativesocket; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.Platform; +import org.junit.jupiter.api.Assumptions; +import org.junitpioneer.jupiter.RetryingTest; +import org.openjdk.jmc.common.IMCThread; +import org.openjdk.jmc.common.item.IItem; +import org.openjdk.jmc.common.item.IItemCollection; +import org.openjdk.jmc.common.item.IItemIterable; +import org.openjdk.jmc.common.item.IMemberAccessor; +import org.openjdk.jmc.flightrecorder.JfrAttributes; + +import java.io.IOException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Verifies that the eventThread field on NativeSocketEvent is populated and + * names a non-empty thread name, indicating the I/O was attributed to the + * calling thread. + */ +public class NativeSocketEventThreadTest extends AbstractProfilerTest { + + @Override + protected boolean isPlatformSupported() { + return Platform.isLinux() && !Platform.isMusl(); + } + + @Override + protected String getProfilerCommand() { + // 100us period keeps sampling probability high on short localhost I/O. + return "natsock=100us"; + } + + @RetryingTest(3) + public void eventThreadIsPopulated() throws Exception { + Assumptions.assumeTrue(Platform.isLinux(), "nativesocket tracking is Linux-only"); + + doTcpTransfer(64 * 1024, 16); + + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.NativeSocketEvent"); + assertTrue(events.hasItems(), "No NativeSocketEvent events found"); + + for (IItemIterable items : events) { + IMemberAccessor threadAccessor = + JfrAttributes.EVENT_THREAD.getAccessor(items.getType()); + assertNotNull(threadAccessor, "eventThread accessor must be present"); + for (IItem item : items) { + IMCThread thread = threadAccessor.getMember(item); + assertNotNull(thread, "eventThread must not be null"); + String name = thread.getThreadName(); + assertNotNull(name, "thread name must not be null"); + assertFalse(name.isEmpty(), "thread name must not be empty"); + } + } + } + + private void doTcpTransfer(int payloadSize, int iterations) throws Exception { + byte[] payload = new byte[payloadSize]; + try (ServerSocket server = new ServerSocket(0)) { + int port = server.getLocalPort(); + Thread serverThread = new Thread(() -> { + for (int iter = 0; iter < iterations; iter++) { + try (Socket conn = server.accept()) { + InputStream in = conn.getInputStream(); + byte[] buf = new byte[payloadSize]; + int read = 0; + while (read < payloadSize) { + int n = in.read(buf, read, payloadSize - read); + if (n < 0) break; + read += n; + } + conn.getOutputStream().write(buf, 0, read); + conn.getOutputStream().flush(); + } catch (IOException ignored) {} + } + }); + serverThread.setDaemon(true); + serverThread.start(); + + for (int iter = 0; iter < iterations; iter++) { + try (Socket client = new Socket("127.0.0.1", port)) { + client.getOutputStream().write(payload); + client.getOutputStream().flush(); + byte[] resp = new byte[payloadSize]; + InputStream in = client.getInputStream(); + int read = 0; + while (read < payloadSize) { + int n = in.read(resp, read, payloadSize - read); + if (n < 0) break; + read += n; + } + } + } + serverThread.join(5000); + } + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketMacOsNoOpTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketMacOsNoOpTest.java new file mode 100644 index 000000000..575e6759b --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketMacOsNoOpTest.java @@ -0,0 +1,97 @@ +package com.datadoghq.profiler.nativesocket; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.Platform; +import org.junit.jupiter.api.Assumptions; +import org.junitpioneer.jupiter.RetryingTest; +import org.openjdk.jmc.common.item.IItemCollection; + +import java.io.IOException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * On macOS the nativesocket feature is a no-op stub. + * Verifies that the profiler starts without error when 'nativesocket' is specified + * and that no NativeSocketEvent events appear in the recording. + */ +public class NativeSocketMacOsNoOpTest extends AbstractProfilerTest { + + @Override + protected boolean isPlatformSupported() { + return Platform.isMac(); + } + + @Override + protected String getProfilerCommand() { + return "natsock"; + } + + @RetryingTest(3) + public void noEventsOnMacOS() throws Exception { + Assumptions.assumeTrue(Platform.isMac(), "This test targets macOS no-op behaviour"); + + // Profiler started in @BeforeEach without error; verify it is actually running + String status = profiler.getStatus(); + assertTrue(status.contains("Running : true"), + "Profiler should be running after start with nativesocket on macOS; status: " + status); + + doTcpTransfer(32 * 1024, 8); + + stopProfiler(); + + // verifyEvents with failOnEmpty=false: must not throw even if empty + IItemCollection events = verifyEvents("datadog.NativeSocketEvent", false); + assertNotNull(events); + assertFalse(events.hasItems(), + "NativeSocketEvent must not be emitted on macOS (no-op stub)"); + } + + private void doTcpTransfer(int payloadSize, int iterations) throws Exception { + byte[] payload = new byte[payloadSize]; + try (ServerSocket server = new ServerSocket(0)) { + int port = server.getLocalPort(); + Thread serverThread = new Thread(() -> { + for (int iter = 0; iter < iterations; iter++) { + try (Socket conn = server.accept()) { + try { + InputStream in = conn.getInputStream(); + byte[] buf = new byte[payloadSize]; + int read = 0; + while (read < payloadSize) { + int n = in.read(buf, read, payloadSize - read); + if (n < 0) break; + read += n; + } + conn.getOutputStream().write(buf, 0, read); + conn.getOutputStream().flush(); + } catch (IOException ignored) {} + } catch (IOException ignored) {} + } + }); + serverThread.setDaemon(true); + serverThread.start(); + + for (int iter = 0; iter < iterations; iter++) { + try (Socket client = new Socket("127.0.0.1", port)) { + client.getOutputStream().write(payload); + client.getOutputStream().flush(); + byte[] resp = new byte[payloadSize]; + InputStream in = client.getInputStream(); + int read = 0; + while (read < payloadSize) { + int n = in.read(resp, read, payloadSize - read); + if (n < 0) break; + read += n; + } + } + } + serverThread.join(5000); + } + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketRateLimitTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketRateLimitTest.java new file mode 100644 index 000000000..103f85ba9 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketRateLimitTest.java @@ -0,0 +1,168 @@ +/* + * 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. + */ + +package com.datadoghq.profiler.nativesocket; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.Platform; +import org.junit.jupiter.api.Assumptions; +import org.junitpioneer.jupiter.RetryingTest; +import org.openjdk.jmc.common.item.Attribute; +import org.openjdk.jmc.common.item.IAttribute; +import org.openjdk.jmc.common.item.IItem; +import org.openjdk.jmc.common.item.IItemCollection; +import org.openjdk.jmc.common.item.IItemIterable; +import org.openjdk.jmc.common.item.IMemberAccessor; +import org.openjdk.jmc.common.unit.IQuantity; +import org.openjdk.jmc.common.unit.UnitLookup; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.ServerSocket; +import java.net.Socket; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies that the sampler is active: when a large number of I/O operations are + * performed over a single persistent connection, only a fraction are recorded + * (events << operations), and the weight field is > 1 on at least some events, + * reflecting statistical significance. + * + * The target rate is ~5000 events/min; for a 10-second window that is ~833 events. + * We generate far more byte volume than that and assert the event count is + * substantially less than the operation count, and that at least one event + * carries weight > 1. + * + * A persistent connection is reused across all iterations to avoid the TCP + * handshake overhead that would make the test slow and unreliable on CI. + */ +public class NativeSocketRateLimitTest extends AbstractProfilerTest { + + private static final IAttribute WEIGHT_ATTR = + Attribute.attr("weight", "weight", "weight", UnitLookup.NUMBER); + + @Override + protected boolean isPlatformSupported() { + return Platform.isLinux() && !Platform.isMusl(); + } + + @Override + protected String getProfilerCommand() { + return "natsock"; + } + + @RetryingTest(3) + public void eventCountIsSubstantiallyLessThanOperationCount() throws Exception { + Assumptions.assumeTrue(Platform.isLinux(), "nativesocket tracking is Linux-only"); + + // Generate a high volume of byte-sized writes over a single connection. + // ~5000 iterations * 4 KB = 20 MB total — far above the ~5000-event/min rate limit. + int operations = doHighRateTcpTransfer(4096, 5000); + System.out.println("Total TCP operations performed: " + operations); + + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.NativeSocketEvent"); + assertTrue(events.hasItems(), "No NativeSocketEvent events found"); + + long eventCount = 0; + boolean foundWeightAboveOne = false; + + for (IItemIterable items : events) { + IMemberAccessor weightAccessor = + WEIGHT_ATTR.getAccessor(items.getType()); + for (IItem item : items) { + eventCount++; + if (weightAccessor != null) { + IQuantity w = weightAccessor.getMember(item); + if (w != null && w.doubleValue() > 1.0) { + foundWeightAboveOne = true; + } + } + } + } + + System.out.println("Recorded NativeSocketEvent count: " + eventCount); + + // Recorded events must be far fewer than raw operations (subsampling is active). + // Target rate is ~5000 events/min; for a 10-second window ~833 events is the ceiling. + // Using operations/2 (10000) provides a generous upper bound that must still be met. + assertTrue(eventCount < operations / 2, + "Too many events sampled (rate limiting not working): event count (" + eventCount + + ") should be less than operations/2 (" + operations / 2 + ")"); + + // At least some events must have weight > 1, indicating byte-weighted sampling + assertTrue(foundWeightAboveOne, + "Expected at least one event with weight > 1 (byte-weighted inverse-transform sampling)"); + } + + /** + * Sends {@code iterations} writes of {@code payloadSize} bytes over a single + * persistent TCP connection and reads the echo back. Reusing the connection + * avoids per-iteration TCP handshake overhead that would otherwise make the + * workload too slow to reliably hit the rate limit within the recording window. + * + * @return total number of individual send/recv calls performed (4 per iteration) + */ + private int doHighRateTcpTransfer(int payloadSize, int iterations) throws Exception { + byte[] payload = new byte[payloadSize]; + + try (ServerSocket server = new ServerSocket(0)) { + int port = server.getLocalPort(); + + Thread serverThread = new Thread(() -> { + try (Socket conn = server.accept()) { + InputStream in = conn.getInputStream(); + OutputStream out = conn.getOutputStream(); + byte[] buf = new byte[payloadSize]; + for (int iter = 0; iter < iterations; iter++) { + int read = 0; + while (read < payloadSize) { + int n = in.read(buf, read, payloadSize - read); + if (n < 0) return; + read += n; + } + out.write(buf, 0, payloadSize); + out.flush(); + } + } catch (IOException ignored) {} + }); + serverThread.setDaemon(true); + serverThread.start(); + + try (Socket client = new Socket("127.0.0.1", port)) { + OutputStream out = client.getOutputStream(); + InputStream in = client.getInputStream(); + byte[] resp = new byte[payloadSize]; + for (int iter = 0; iter < iterations; iter++) { + out.write(payload); + out.flush(); + int read = 0; + while (read < payloadSize) { + int n = in.read(resp, read, payloadSize - read); + if (n < 0) break; + read += n; + } + } + } + serverThread.join(10000); + } + // Each iteration: 1 send + 1 recv on client; 1 recv + 1 send on server + return iterations * 4; + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketRemoteAddressTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketRemoteAddressTest.java new file mode 100644 index 000000000..04b46e6e8 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketRemoteAddressTest.java @@ -0,0 +1,119 @@ +package com.datadoghq.profiler.nativesocket; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.Platform; +import org.junit.jupiter.api.Assumptions; +import org.junitpioneer.jupiter.RetryingTest; +import org.openjdk.jmc.common.item.Attribute; +import org.openjdk.jmc.common.item.IAttribute; +import org.openjdk.jmc.common.item.IItem; +import org.openjdk.jmc.common.item.IItemCollection; +import org.openjdk.jmc.common.item.IItemIterable; +import org.openjdk.jmc.common.item.IMemberAccessor; +import org.openjdk.jmc.common.unit.UnitLookup; + +import java.io.IOException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Verifies that the remoteAddress field in NativeSocketEvent is a non-empty + * string of the form ":" matching the known server endpoint. + */ +public class NativeSocketRemoteAddressTest extends AbstractProfilerTest { + + private static final IAttribute REMOTE_ADDRESS = + Attribute.attr("remoteAddress", "remoteAddress", "Remote address", UnitLookup.PLAIN_TEXT); + + @Override + protected boolean isPlatformSupported() { + return Platform.isLinux() && !Platform.isMusl(); + } + + @Override + protected String getProfilerCommand() { + // 100us period keeps sampling probability high enough that the 16 + // short-lived connections reliably produce at least one event whose + // remoteAddress points at the known server port. + return "natsock=100us"; + } + + @RetryingTest(3) + public void remoteAddressIsIpColonPort() throws Exception { + Assumptions.assumeTrue(Platform.isLinux(), "nativesocket tracking is Linux-only"); + + int serverPort = doTcpTransfer(64 * 1024, 16); + System.out.println("Server port: " + serverPort); + + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.NativeSocketEvent"); + assertTrue(events.hasItems(), "No NativeSocketEvent events found"); + + boolean foundMatchingAddress = false; + for (IItemIterable items : events) { + IMemberAccessor addrAccessor = + REMOTE_ADDRESS.getAccessor(items.getType()); + assertNotNull(addrAccessor, "remoteAddress accessor must exist"); + for (IItem item : items) { + String addr = addrAccessor.getMember(item); + assertNotNull(addr, "remoteAddress must not be null"); + assertFalse(addr.isEmpty(), "remoteAddress must not be empty"); + // Must match ip:port pattern + assertTrue(addr.matches("^[\\d.]+:\\d+$") || addr.matches("^\\[.*\\]:\\d+$"), + "remoteAddress '" + addr + "' does not match expected ip:port format"); + if (addr.endsWith(":" + serverPort)) { + foundMatchingAddress = true; + } + } + } + assertTrue(foundMatchingAddress, + "Expected at least one event with remoteAddress pointing to server port " + serverPort); + } + + /** Returns the server's bound port. */ + private int doTcpTransfer(int payloadSize, int iterations) throws Exception { + byte[] payload = new byte[payloadSize]; + try (ServerSocket server = new ServerSocket(0)) { + int port = server.getLocalPort(); + Thread serverThread = new Thread(() -> { + for (int iter = 0; iter < iterations; iter++) { + try (Socket conn = server.accept()) { + InputStream in = conn.getInputStream(); + byte[] buf = new byte[payloadSize]; + int read = 0; + while (read < payloadSize) { + int n = in.read(buf, read, payloadSize - read); + if (n < 0) break; + read += n; + } + conn.getOutputStream().write(buf, 0, read); + conn.getOutputStream().flush(); + } catch (IOException ignored) {} + } + }); + serverThread.setDaemon(true); + serverThread.start(); + + for (int iter = 0; iter < iterations; iter++) { + try (Socket client = new Socket("127.0.0.1", port)) { + client.getOutputStream().write(payload); + client.getOutputStream().flush(); + byte[] resp = new byte[payloadSize]; + InputStream in = client.getInputStream(); + int read = 0; + while (read < payloadSize) { + int n = in.read(resp, read, payloadSize - read); + if (n < 0) break; + read += n; + } + } + } + serverThread.join(5000); + return port; + } + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketRestartTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketRestartTest.java new file mode 100644 index 000000000..b49823db9 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketRestartTest.java @@ -0,0 +1,45 @@ +package com.datadoghq.profiler.nativesocket; + +import com.datadoghq.profiler.Platform; +import org.junit.jupiter.api.Assumptions; +import org.junitpioneer.jupiter.RetryingTest; +import org.openjdk.jmc.common.item.IItemCollection; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies that nativesocket profiling survives a stop/restart cycle. + * Events must be recorded in the second profiling session, confirming + * that lifecycle management (hook install/uninstall/reinstall) is correct. + */ +public class NativeSocketRestartTest extends NativeSocketTestBase { + + @RetryingTest(3) + public void testNativeSocketProfilerRestart() throws Exception { + Assumptions.assumeTrue(Platform.isLinux(), "nativesocket tracking is Linux-only"); + + // First session: framework already started the profiler in @BeforeEach. + doTcpTransfer(4096, 64); + stopProfiler(); + + // Second session: start manually with a fresh JFR file. + Path jfr2 = Files.createTempFile(Paths.get("/tmp/recordings"), "NativeSocketRestartTest_restart", ".jfr"); + try { + profiler.execute("start,natsock=100us,jfr,file=" + jfr2.toAbsolutePath()); + + doTcpTransfer(4096, 64); + + profiler.stop(); + + IItemCollection events = verifyEvents(jfr2, "datadog.NativeSocketEvent", true); + assertTrue(events.hasItems(), + "NativeSocketEvent events must be recorded in the second profiling session"); + } finally { + Files.deleteIfExists(jfr2); + } + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketSendRecvSeparateTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketSendRecvSeparateTest.java new file mode 100644 index 000000000..7eb7d688e --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketSendRecvSeparateTest.java @@ -0,0 +1,107 @@ +package com.datadoghq.profiler.nativesocket; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.Platform; +import org.junit.jupiter.api.Assumptions; +import org.junitpioneer.jupiter.RetryingTest; +import org.openjdk.jmc.common.item.IItem; +import org.openjdk.jmc.common.item.IItemCollection; +import org.openjdk.jmc.common.item.IItemIterable; +import org.openjdk.jmc.common.item.IMemberAccessor; + +import java.io.IOException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.concurrent.CountDownLatch; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Verifies that SEND and RECV operations are tracked as independent events. + * Generates a workload where only one side performs writes so that events + * can be attributed unambiguously to the sending or receiving thread. + */ +public class NativeSocketSendRecvSeparateTest extends AbstractProfilerTest { + + @Override + protected boolean isPlatformSupported() { + return Platform.isLinux() && !Platform.isMusl(); + } + + @Override + protected String getProfilerCommand() { + // 100us initial sampling period: a single 256KB localhost write + // completes in ~100-500us giving P ~= 0.6-1.0 per call, so both + // SEND and RECV directions are sampled reliably over 32 iterations. + return "natsock=100us"; + } + + @RetryingTest(3) + public void sendAndRecvTrackedWithSeparateCounts() throws Exception { + Assumptions.assumeTrue(Platform.isLinux(), "nativesocket tracking is Linux-only"); + + // Large volume of data to ensure sampler captures both directions + doUnidirectionalTransfer(256 * 1024, 32); + + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.NativeSocketEvent"); + assertTrue(events.hasItems(), "No NativeSocketEvent events found"); + + long sendCount = 0; + long recvCount = 0; + + for (IItemIterable items : events) { + IMemberAccessor opAccessor = OPERATION.getAccessor(items.getType()); + assertNotNull(opAccessor); + for (IItem item : items) { + String op = opAccessor.getMember(item); + // Java sockets reach libc via write()/read(); send()/recv() also possible. + // Group by direction: outbound (SEND, WRITE) vs inbound (RECV, READ). + if ("SEND".equals(op) || "WRITE".equals(op)) sendCount++; + else if ("RECV".equals(op) || "READ".equals(op)) recvCount++; + } + } + + System.out.println("Outbound (SEND/WRITE) events: " + sendCount + + ", Inbound (RECV/READ) events: " + recvCount); + assertTrue(sendCount > 0, "Expected at least one outbound (SEND/WRITE) event, got 0"); + assertTrue(recvCount > 0, "Expected at least one inbound (RECV/READ) event, got 0"); + } + + private void doUnidirectionalTransfer(int payloadSize, int iterations) throws Exception { + byte[] payload = new byte[payloadSize]; + CountDownLatch serverReady = new CountDownLatch(1); + + try (ServerSocket server = new ServerSocket(0)) { + int port = server.getLocalPort(); + Thread serverThread = new Thread(() -> { + serverReady.countDown(); + for (int iter = 0; iter < iterations; iter++) { + try (Socket conn = server.accept()) { + InputStream in = conn.getInputStream(); + byte[] buf = new byte[payloadSize]; + int read = 0; + while (read < payloadSize) { + int n = in.read(buf, read, payloadSize - read); + if (n < 0) break; + read += n; + } + } catch (IOException ignored) {} + } + }); + serverThread.setDaemon(true); + serverThread.start(); + serverReady.await(); + + for (int iter = 0; iter < iterations; iter++) { + try (Socket client = new Socket("127.0.0.1", port)) { + client.getOutputStream().write(payload); + client.getOutputStream().flush(); + } + } + serverThread.join(5000); + } + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketStackTraceTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketStackTraceTest.java new file mode 100644 index 000000000..d966b4538 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketStackTraceTest.java @@ -0,0 +1,120 @@ +package com.datadoghq.profiler.nativesocket; + +import com.datadoghq.profiler.CStackAwareAbstractProfilerTest; +import com.datadoghq.profiler.Platform; +import com.datadoghq.profiler.junit.CStack; +import com.datadoghq.profiler.junit.RetryTest; +import org.junit.jupiter.api.Assumptions; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.params.provider.ValueSource; +import org.openjdk.jmc.common.item.IItem; +import org.openjdk.jmc.common.item.IItemCollection; +import org.openjdk.jmc.common.item.IItemIterable; +import org.openjdk.jmc.common.item.IMemberAccessor; +import org.openjdk.jmc.flightrecorder.jdk.JdkAttributes; + +import java.io.IOException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Verifies that NativeSocketEvent events carry non-empty stack traces, + * and that a recognizable Java call site from the test workload appears + * in at least one event's stack trace. + * + *

Parameterized over cstack modes (vm, vmx, dwarf, fp) to exercise both + * the walkVM path (cstack >= CSTACK_VM) and the AsyncGetCallTrace path + * (dwarf/fp) through the BCI_NATIVE_SOCKET code. + */ +public class NativeSocketStackTraceTest extends CStackAwareAbstractProfilerTest { + + public NativeSocketStackTraceTest(@CStack String cstack) { + super(cstack); + } + + @Override + protected boolean isPlatformSupported() { + return Platform.isLinux() && !Platform.isMusl(); + } + + @Override + protected String getProfilerCommand() { + // 100us period keeps sampling probability high on short localhost I/O. + return "natsock=100us"; + } + + @RetryTest(3) + @TestTemplate + @ValueSource(strings = {"vm", "vmx", "dwarf", "fp"}) + public void stackTraceIsCapturedForSocketEvents() throws Exception { + Assumptions.assumeTrue(Platform.isLinux(), "nativesocket tracking is Linux-only"); + + doTcpTransfer(64 * 1024, 20); + + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.NativeSocketEvent"); + assertTrue(events.hasItems(), "No NativeSocketEvent events found"); + + boolean foundNonEmptyStackTrace = false; + for (IItemIterable items : events) { + IMemberAccessor stackTraceAccessor = + JdkAttributes.STACK_TRACE_STRING.getAccessor(items.getType()); + if (stackTraceAccessor == null) continue; + for (IItem item : items) { + String st = stackTraceAccessor.getMember(item); + if (st != null && !st.isEmpty()) { + foundNonEmptyStackTrace = true; + break; + } + } + if (foundNonEmptyStackTrace) break; + } + assertTrue(foundNonEmptyStackTrace, "Expected at least one NativeSocketEvent with a non-empty stack trace"); + } + + // Named method so it can appear as a recognizable frame + private void doTcpTransfer(int payloadSize, int iterations) throws Exception { + byte[] payload = new byte[payloadSize]; + try (ServerSocket server = new ServerSocket(0)) { + int port = server.getLocalPort(); + Thread serverThread = new Thread(() -> { + for (int iter = 0; iter < iterations; iter++) { + try (Socket conn = server.accept()) { + InputStream in = conn.getInputStream(); + byte[] buf = new byte[payloadSize]; + int read = 0; + while (read < payloadSize) { + int n = in.read(buf, read, payloadSize - read); + if (n < 0) break; + read += n; + } + conn.getOutputStream().write(buf, 0, read); + conn.getOutputStream().flush(); + } catch (IOException ignored) {} + } + }); + serverThread.setDaemon(true); + serverThread.start(); + + for (int iter = 0; iter < iterations; iter++) { + try (Socket client = new Socket("127.0.0.1", port)) { + client.getOutputStream().write(payload); + client.getOutputStream().flush(); + byte[] resp = new byte[payloadSize]; + InputStream in = client.getInputStream(); + int read = 0; + while (read < payloadSize) { + int n = in.read(resp, read, payloadSize - read); + if (n < 0) break; + read += n; + } + } + } + serverThread.join(5000); + } + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketTestBase.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketTestBase.java new file mode 100644 index 000000000..c45307959 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketTestBase.java @@ -0,0 +1,78 @@ +package com.datadoghq.profiler.nativesocket; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.Platform; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.ServerSocket; +import java.net.Socket; + +/** Base class for native-socket profiler tests that need a persistent TCP workload. */ +abstract class NativeSocketTestBase extends AbstractProfilerTest { + + @Override + protected boolean isPlatformSupported() { + return Platform.isLinux(); + } + + @Override + protected String getProfilerCommand() { + // 100us initial period keeps P high enough that fast localhost I/O + // reliably produces events across small test workloads. + return "natsock=100us"; + } + + /** + * Sends {@code iterations} writes of {@code payloadSize} bytes over a single + * persistent TCP connection and reads the echo back. Reusing the connection + * fills the TCP send buffer after ~32 iterations (at 4 KB/write, 128 KB OS buffer), + * causing subsequent writes to block for >1 ms and driving the Poisson sampler + * probability close to 1.0 for those calls. + */ + protected void doTcpTransfer(int payloadSize, int iterations) throws Exception { + byte[] payload = new byte[payloadSize]; + for (int i = 0; i < payloadSize; i++) payload[i] = (byte) (i & 0xFF); + + try (ServerSocket server = new ServerSocket(0)) { + int port = server.getLocalPort(); + Thread serverThread = new Thread(() -> { + try (Socket conn = server.accept()) { + InputStream in = conn.getInputStream(); + OutputStream out = conn.getOutputStream(); + byte[] buf = new byte[payloadSize]; + for (int iter = 0; iter < iterations; iter++) { + int read = 0; + while (read < payloadSize) { + int n = in.read(buf, read, payloadSize - read); + if (n < 0) return; + read += n; + } + out.write(buf, 0, payloadSize); + out.flush(); + } + } catch (IOException ignored) {} + }); + serverThread.setDaemon(true); + serverThread.start(); + + try (Socket client = new Socket("127.0.0.1", port)) { + OutputStream out = client.getOutputStream(); + InputStream in = client.getInputStream(); + byte[] resp = new byte[payloadSize]; + for (int iter = 0; iter < iterations; iter++) { + out.write(payload); + out.flush(); + int read = 0; + while (read < payloadSize) { + int n = in.read(resp, read, payloadSize - read); + if (n < 0) break; + read += n; + } + } + } + serverThread.join(10000); + } + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketUdpExcludedTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketUdpExcludedTest.java new file mode 100644 index 000000000..b0b3a48bc --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/nativesocket/NativeSocketUdpExcludedTest.java @@ -0,0 +1,84 @@ +package com.datadoghq.profiler.nativesocket; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.Platform; +import org.junit.jupiter.api.Assumptions; +import org.junitpioneer.jupiter.RetryingTest; +import org.openjdk.jmc.common.item.Attribute; +import org.openjdk.jmc.common.item.IAttribute; +import org.openjdk.jmc.common.item.IItem; +import org.openjdk.jmc.common.item.IItemCollection; +import org.openjdk.jmc.common.item.IItemIterable; +import org.openjdk.jmc.common.item.IMemberAccessor; +import org.openjdk.jmc.common.unit.UnitLookup; + +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetAddress; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Verifies that UDP (DatagramSocket / sendto / recvfrom) transfers do NOT + * produce NativeSocketEvent events. Only TCP blocking send/recv are in scope. + * + * This test performs only UDP transfers and expects zero NativeSocketEvent + * events in the recording. + */ +public class NativeSocketUdpExcludedTest extends AbstractProfilerTest { + + @Override + protected boolean isPlatformSupported() { + return Platform.isLinux() && !Platform.isMusl(); + } + + @Override + protected String getProfilerCommand() { + // 100us period strengthens the negative assertion: any UDP traffic + // that accidentally leaks through the TCP filter would be far more + // likely to produce a sampled event at this tighter interval. + return "natsock=100us"; + } + + @RetryingTest(3) + public void udpTransfersProduceNoSocketEvents() throws Exception { + Assumptions.assumeTrue(Platform.isLinux(), "nativesocket tracking is Linux-only"); + + doUdpTransfer(1024, 500); + + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.NativeSocketEvent", false); + assertNotNull(events); + assertFalse(events.hasItems(), + "NativeSocketEvent must not be produced for UDP (sendto/recvfrom) transfers"); + } + + private void doUdpTransfer(int payloadSize, int iterations) throws Exception { + byte[] payload = new byte[payloadSize]; + InetAddress loopback = InetAddress.getLoopbackAddress(); + + try (DatagramSocket server = new DatagramSocket(0, loopback)) { + int port = server.getLocalPort(); + Thread serverThread = new Thread(() -> { + byte[] buf = new byte[payloadSize]; + DatagramPacket pkt = new DatagramPacket(buf, buf.length); + for (int iter = 0; iter < iterations; iter++) { + try { + server.receive(pkt); + } catch (Exception ignored) {} + } + }); + serverThread.setDaemon(true); + serverThread.start(); + + try (DatagramSocket client = new DatagramSocket()) { + DatagramPacket pkt = new DatagramPacket(payload, payload.length, loopback, port); + for (int iter = 0; iter < iterations; iter++) { + client.send(pkt); + } + } + serverThread.join(5000); + } + } +}