diff --git a/CMakeLists.txt b/CMakeLists.txt index b9b3c4e611..3934882bae 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,3 +1,8 @@ +# Root CMake configuration for collector. +# +# The actual build logic lives in collector/CMakeLists.txt. This file exists +# so CMake can be invoked from the repo root (standard convention) and to +# hold project-wide settings like enable_testing(). cmake_minimum_required(VERSION 3.15) project(collector) diff --git a/Makefile b/Makefile index 951a1d6834..17210b4b98 100644 --- a/Makefile +++ b/Makefile @@ -1,3 +1,10 @@ +# Top-level Makefile for the collector repository. +# +# CMake handles C++ compilation, but the overall build also involves Docker +# image assembly, builder container lifecycle, and integration test +# orchestration — tasks better expressed as Make targets. The builder +# container provides a reproducible toolchain so developers don't need +# to install C++ deps locally. BASE_PATH = . include Makefile-constants.mk @@ -72,6 +79,9 @@ ci-all-tests: ci-benchmarks ci-integration-tests .PHONY: start-builder start-builder: builder teardown-builder + # --cap-add sys_ptrace: needed for address sanitizer and debugging. + # -v $(CURDIR):$(CURDIR):Z : same absolute path inside the container keeps + # CMake build artifacts consistent. :Z relabels for SELinux on RHEL/Fedora. docker run -id \ --name $(COLLECTOR_BUILDER_NAME) \ --pull never \ diff --git a/collector/CMakeLists.txt b/collector/CMakeLists.txt index 2d0a6a2152..3fe8f88ad6 100644 --- a/collector/CMakeLists.txt +++ b/collector/CMakeLists.txt @@ -1,3 +1,16 @@ +# Build configuration for the three collector executables: +# +# collector - Main daemon. Runs as PID 1 in the DaemonSet pod, attaches +# the CO-RE eBPF probe via Falco's sinsp library, and streams +# events to Sensor over gRPC. +# +# self-checks - Lightweight startup validator. Confirms the BPF CO-RE driver +# can load on the current kernel before the full collector +# starts, enabling fast-fail with a clear diagnostic. +# +# connscrape - Diagnostic utility for dumping live connection state from +# /proc. Useful for debugging network-related collection issues. +# project(collector-bin) find_package(Threads) @@ -8,11 +21,14 @@ find_package(civetweb CONFIG REQUIRED) find_package(prometheus-cpp CONFIG REQUIRED) set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} ${PROJECT_SOURCE_DIR}) +# -fno-omit-frame-pointer: needed for gperftools CPU profiling and perf. +# -rdynamic: crash-handler backtraces need exported symbols for readable names. set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fPIC -Wall --std=c++17 -pthread -Wno-deprecated-declarations -fno-omit-frame-pointer -rdynamic") set(CMAKE_CXX_FLAGS_DEBUG "-g -ggdb -D_DEBUG") set(CMAKE_CXX_FLAGS_RELEASE "-O3 -fno-strict-aliasing -DNDEBUG") +# ASan/TSan and gperftools both intercept malloc, so they conflict at link time. if(ADDRESS_SANITIZER) set(DISABLE_PROFILING ON) set(USE_VALGRIND OFF) @@ -59,14 +75,20 @@ include_directories(${FALCO_DIR}/userspace/libsinsp) include_directories(${CMAKE_CURRENT_BINARY_DIR}) set(DRIVER_HEADERS ${FALCO_DIR}/driver/ppm_events_public.h ${FALCO_DIR}/driver/ppm_fillers.h) +# Enables protobuf arena allocation — batches allocations and frees them all at +# once, reducing heap fragmentation in the high-throughput gRPC event path. add_definitions(-DUSE_PROTO_ARENAS) +# Redirect sinsp ASSERT macros to log warnings instead of abort(). A crash +# in the collector process would blind the entire node. add_definitions(-DASSERT_TO_LOG) -# Optimization, only process socket file descriptors +# Only track socket file descriptors — ignoring file/pipe FDs avoids tracking +# millions of irrelevant descriptors and substantially reduces memory usage. add_definitions(-DSCAP_SOCKET_ONLY_FD) -# Filter out some cgroup subsys +# Limit cgroup parsing to these subsystems to avoid scanning dozens of +# irrelevant controllers per process when resolving container IDs. add_definitions("-DINTERESTING_SUBSYS=\"perf_event\", \"cpu\", \"cpuset\", \"memory\"") add_subdirectory(lib) @@ -81,14 +103,22 @@ add_executable(self-checks self-checks.cpp) add_subdirectory(test) -# Falco Wrapper Library +# Falco sinsp library configuration. +# Legacy kernel module driver is unused — collector uses CO-RE eBPF (below). set(BUILD_DRIVER OFF CACHE BOOL "Build the driver on Linux" FORCE) set(USE_BUNDLED_DEPS OFF CACHE BOOL "Enable bundled dependencies instead of using the system ones" FORCE) set(USE_BUNDLED_CARES OFF CACHE BOOL "Enable bundled dependencies instead of using the system ones" FORCE) +# gVisor intercepts syscalls differently and is not a supported runtime. set(BUILD_LIBSCAP_GVISOR OFF CACHE BOOL "Do not build gVisor support" FORCE) set(MINIMAL_BUILD OFF CACHE BOOL "Minimal" FORCE) +# Reduces per-thread memory from ~1KB to a few hundred bytes. Prevents OOM +# on nodes with thousands of threads. set(SINSP_SLIM_THREADINFO ON CACHE BOOL "Slim threadinfo" FORCE) +# Collector ships as a statically-linked binary in a minimal UBI-micro +# container — dynamic linking would require shipping .so files. set(BUILD_SHARED_LIBS OFF CACHE BOOL "Build position independent libraries and executables" FORCE) +# Exception to static linking: libbpf dlopen()s libelf at runtime to parse +# BPF ELF objects, so libelf must remain a shared library. set(LIBELF_LIB_SUFFIX ".so" CACHE STRING "Use libelf.so" FORCE) # Turn OFF falco's unit tests and examples @@ -96,11 +126,19 @@ set(CREATE_TEST_TARGETS OFF CACHE BOOL "Enable make-targets for unit testing" FO set(BUILD_LIBSCAP_EXAMPLES OFF CACHE BOOL "Build libscap examples" FORCE) set(BUILD_LIBSINSP_EXAMPLES OFF CACHE BOOL "Build libsinsp examples" FORCE) +# Collector runs in a container but needs the host's /proc and /sys. The +# DaemonSet mounts the host root at /host; this tells sinsp to prefix all +# host filesystem lookups accordingly (e.g. /host/proc instead of /proc). set(SCAP_HOST_ROOT_ENV_VAR_NAME "COLLECTOR_HOST_ROOT" CACHE STRING "Host root environment variable name" FORCE) +# CO-RE (Compile Once - Run Everywhere) eBPF eliminates the need to ship +# per-kernel driver modules, simplifying deployment across kernel versions. set(BUILD_LIBSCAP_MODERN_BPF ON CACHE BOOL "Enable modern bpf engine" FORCE) set(MODERN_BPF_DEBUG_MODE ${BPF_DEBUG_MODE} CACHE BOOL "Enable BPF debug prints" FORCE) +# Exclude high-frequency syscalls (ppoll, nanosleep) and ones not used for +# detection (openat2, setsockopt, io_uring_setup) to reduce BPF memory +# footprint and verifier load time. set(MODERN_BPF_EXCLUDE_PROGS "^(openat2|ppoll|setsockopt|io_uring_setup|nanosleep)$" CACHE STRING "Set of syscalls to exclude from modern bpf engine " FORCE) add_subdirectory(${FALCO_DIR} falco) diff --git a/collector/collector.cpp b/collector/collector.cpp index e9ff40bd9b..33c6124250 100644 --- a/collector/collector.cpp +++ b/collector/collector.cpp @@ -44,11 +44,17 @@ extern "C" { #include "Logging.h" #include "Utility.h" +// Sensor connection timeout: 30 polls × 1s each = 30s before giving up. +// This is a startup-only timeout; once connected, the gRPC channel handles +// reconnection internally. static const int MAX_GRPC_CONNECTION_POLLS = 30; using namespace collector; +// Global control variable shared with CollectorService via pointer. +// Must be lock-free for safe use inside signal handlers. static std::atomic g_control(ControlValue::RUN); +// Stores the signal number that triggered shutdown, for logging purposes. static std::atomic g_signum(0); static void @@ -115,6 +121,9 @@ void gplNotice() { } void initialChecks() { + // The control variable is written from signal handlers, so it must be + // lock-free to avoid deadlocks. This should always succeed on modern + // architectures, but we check defensively. if (!g_control.is_lock_free()) { CLOG(FATAL) << "Internal error: could not create a lock-free control variable."; } @@ -158,6 +167,10 @@ int main(int argc, char** argv) { CollectorConfig config; config.InitCollectorConfig(args); + // Load the runtime config file once at startup. The ConfigLoader inside + // CollectorService will watch for changes via inotify after this point. + // FILE_NOT_FOUND is not fatal (the config file is optional), but + // PARSE_ERROR means the file exists but is malformed. if (ConfigLoader(config).LoadConfiguration() == collector::ConfigLoader::PARSE_ERROR) { CLOG(FATAL) << "Unable to parse configuration file"; } @@ -184,7 +197,10 @@ int main(int argc, char** argv) { CLOG(INFO) << "GRPC is disabled. Specify GRPC_SERVER='server addr' env and signalFormat = 'signal_summary' and signalOutput = 'grpc'"; } - // Register signal handlers + // Register signal handlers after gRPC is connected but before RunForever(), + // so that crash signals produce stack traces (AbortHandler) and graceful + // shutdown signals (SIGTERM/SIGINT) set the atomic control variable rather + // than terminating immediately. signal(SIGABRT, AbortHandler); signal(SIGSEGV, AbortHandler); signal(SIGTERM, ShutdownHandler); diff --git a/collector/container/Dockerfile b/collector/container/Dockerfile index 3cf7945a6e..23cc5d2955 100644 --- a/collector/container/Dockerfile +++ b/collector/container/Dockerfile @@ -1,11 +1,17 @@ +# UBI-micro: required for Red Hat container certification. The smallest UBI +# variant — no package manager, keeping the final image minimal. FROM registry.access.redhat.com/ubi10/ubi-micro:latest AS ubi-micro-base +# ubi-micro has no dnf, so we use the full UBI image to install packages into +# an --installroot overlay. The final image gets the packages without dnf. FROM registry.access.redhat.com/ubi10/ubi:latest AS package_installer # Copy ubi-micro base to /out to preserve its rpmdb COPY --from=ubi-micro-base / /out/ -# Install packages directly to /out/ using --installroot +# Install runtime dependencies into /out/ using --installroot. +# elfutils-libelf: BPF ELF parsing; tbb: sinsp runtime dep; curl-minimal: +# healthcheck; libcap-ng: capability ops; ca-certificates: TLS to Sensor. RUN dnf install -y \ --installroot=/out/ \ --releasever=10 \ @@ -15,6 +21,8 @@ RUN dnf install -y \ dnf clean all --installroot=/out/ && \ rm -rf /out/var/cache/dnf /out/var/cache/yum +# Final image is based on ubi-micro (not package_installer) to exclude dnf +# and the full UBI toolchain — only the /out/ overlay is merged in. FROM ubi-micro-base ARG COLLECTOR_VERSION @@ -37,12 +45,17 @@ COPY --from=package_installer /out/ / COPY container/THIRD_PARTY_NOTICES/ /THIRD_PARTY_NOTICES/ COPY kernel-modules /kernel-modules +# self-checks is a separate binary for BPF driver validation at startup, +# kept independent from the full collector dependency tree. COPY container/bin/collector /usr/local/bin/ COPY container/bin/self-checks /usr/local/bin/self-checks COPY container/status-check.sh /usr/local/bin/status-check.sh +# 8080: civetweb introspection (/ready, /state). 9090: Prometheus metrics. EXPOSE 8080 9090 +# Docker-native healthcheck for non-K8s environments and local dev/debugging. +# 5s start-period avoids false failures while the BPF driver loads. HEALTHCHECK \ # health checks within the first 5s are not counted as failure --start-period=5s \ @@ -51,4 +64,6 @@ HEALTHCHECK \ # the command uses /ready API CMD /usr/local/bin/status-check.sh +# Exec form ensures collector runs as PID 1, receiving signals directly for +# graceful shutdown of the BPF probe and gRPC connection. ENTRYPOINT ["collector"] diff --git a/collector/container/status-check.sh b/collector/container/status-check.sh index 8d072ea01d..289720f4d7 100755 --- a/collector/container/status-check.sh +++ b/collector/container/status-check.sh @@ -1,5 +1,9 @@ #!/usr/bin/env bash +# Pattern-matches the JSON response rather than just checking HTTP status, +# confirming the collector event loop is actually running (not just the HTTP +# server). Uses shell glob matching since jq is not available in UBI-micro. + # /ready API will return the following formatted response: # { # "collector" : { @@ -11,7 +15,8 @@ # "status" : "ok" # } # -# Pattern match for "status":"ok" in the JSON response +# -s: suppress progress noise; -f: non-zero exit on HTTP errors. +# Port 8080 is the civetweb introspection server (not Prometheus on 9090). case "$(curl -sf localhost:8080/ready)" in *'"status"'*'"ok"'*) exit 0 ;; *) exit 1 ;; diff --git a/collector/lib/CMakeLists.txt b/collector/lib/CMakeLists.txt index 1d5386e7d1..d6f1271d93 100644 --- a/collector/lib/CMakeLists.txt +++ b/collector/lib/CMakeLists.txt @@ -1,21 +1,33 @@ +# collector_lib: shared library linked by all three executables (collector, +# self-checks, connscrape). Centralises detection logic, gRPC transport, and +# introspection so the executables themselves are thin entry points. file(GLOB COLLECTOR_LIB_SRC_FILES ${CMAKE_CURRENT_SOURCE_DIR}/*.cpp ${CMAKE_CURRENT_SOURCE_DIR}/system-inspector/*.cpp ) add_library(collector_lib ${DRIVER_HEADERS} ${COLLECTOR_LIB_SRC_FILES}) +# sinsp: Falco's eBPF event engine, syscall parsing, and container-aware thread tables. add_dependencies(collector_lib sinsp) target_link_libraries(collector_lib sinsp) target_link_libraries(collector_lib stdc++fs) # This is needed for GCC-8 to link against the filesystem library +# cap-ng: inspect and drop Linux capabilities for BPF operations. target_link_libraries(collector_lib cap-ng) target_link_libraries(collector_lib uuid) +# gRPC: bidirectional streaming to Sensor for high-volume runtime events. target_link_libraries(collector_lib gRPC::grpc++) +# civetweb: lightweight embeddable HTTP server for introspection endpoints on 8080. target_link_libraries(collector_lib civetweb::civetweb-cpp) target_link_libraries(collector_lib yaml-cpp::yaml-cpp) +# Pull mode: collector hosts a /metrics endpoint on 9090 for Prometheus scraping. target_link_libraries(collector_lib prometheus-cpp::core prometheus-cpp::pull) +# Generated protobuf/gRPC stubs for the Collector↔Sensor API. target_link_libraries(collector_lib rox-proto) +# tcmalloc reduces malloc lock contention in the multi-threaded event pipeline. +# libprofiler enables on-demand CPU profiling in production via SIGPROF. +# Both are disabled under sanitizers/valgrind to avoid conflicts. if(NOT DISABLE_PROFILING) find_library(GPERFTOOLS_PROFILER profiler) find_library(GPERFTOOLS_TCMALLOC tcmalloc) @@ -37,4 +49,6 @@ if (COVERAGE) target_link_options(collector_lib PUBLIC "--coverage") endif() +# z: gRPC compression + BPF ELF decompression. ssl/crypto: TLS to Sensor. +# bpf: libbpf for loading CO-RE eBPF programs. target_link_libraries(collector_lib z ssl crypto bpf) diff --git a/collector/lib/CollectorService.cpp b/collector/lib/CollectorService.cpp index 71b6508cee..baa3f5e0c5 100644 --- a/collector/lib/CollectorService.cpp +++ b/collector/lib/CollectorService.cpp @@ -21,7 +21,11 @@ namespace collector { +// Civetweb configuration: serves HTTP introspection endpoints (status, log +// level, profiler, container info, network state) on port 8080. static const char* OPTIONS[] = {"listening_ports", "8080", nullptr}; +// Prometheus metrics are exposed on a separate port to keep them isolated +// from the introspection endpoints. static const std::string PROMETHEUS_PORT = "9090"; CollectorService::CollectorService(CollectorConfig& config, std::atomic* control, @@ -36,11 +40,12 @@ CollectorService::CollectorService(CollectorConfig& config, std::atomic(); UnorderedSet ignored_l4proto_port_pairs(config_.IgnoredL4ProtoPortPairs()); conn_tracker_->UpdateIgnoredL4ProtoPortPairs(std::move(ignored_l4proto_port_pairs)); @@ -87,7 +92,8 @@ CollectorService::~CollectorService() { exporter_.stop(); - // system_inspector_ needs to be last, since other components relay on it. + // system_inspector_ must be torn down last because NetworkStatusNotifier + // and other components hold pointers into sinsp state (thread table, etc.). system_inspector_.CleanUp(); } diff --git a/collector/lib/CollectorService.h b/collector/lib/CollectorService.h index 01f6ee2a80..0f98474a6f 100644 --- a/collector/lib/CollectorService.h +++ b/collector/lib/CollectorService.h @@ -14,6 +14,14 @@ namespace collector { +/// Top-level service that owns and orchestrates all collector subsystems: +/// system inspector (BPF event capture), network status notifier (periodic +/// /proc scraping + delta reporting), configuration hot-reload, HTTP +/// introspection endpoints (civetweb), and Prometheus metrics export. +/// +/// Lifecycle: construct → InitKernel() → RunForever() → destructor. +/// RunForever() blocks in the sinsp event loop until the atomic control +/// variable signals STOP_COLLECTOR. class CollectorService { public: CollectorService(const CollectorService&) = delete; diff --git a/collector/lib/ConfigLoader.h b/collector/lib/ConfigLoader.h index 49e3a67fb0..a6955ffdc4 100644 --- a/collector/lib/ConfigLoader.h +++ b/collector/lib/ConfigLoader.h @@ -170,10 +170,19 @@ class ParserYaml { ValidationMode validation_mode_; }; -/** - * Reload configuration based on inotify events received on a - * configuration file. - */ +/// Watches the runtime configuration file (default: /etc/stackrox/runtime_config.yaml) +/// for changes via inotify and hot-reloads the CollectorConfig accordingly. +/// +/// Uses protobuf reflection to map YAML fields to the sensor::CollectorConfig +/// proto, so adding new config fields only requires updating the proto definition +/// — no code changes needed in the loader itself. +/// +/// Handles both direct files and symlinks (Kubernetes ConfigMaps are mounted +/// as symlink chains that get atomically swapped, which generates different +/// inotify events than a simple file write). Three watchers are maintained: +/// - LOADER_PARENT_PATH: directory-level events (create/delete of the file) +/// - LOADER_CONFIG_FILE: events on the config file/symlink itself +/// - LOADER_CONFIG_REALPATH: events on the symlink target (for ConfigMap swaps) class ConfigLoader { public: ConfigLoader() = delete; diff --git a/collector/lib/ConnTracker.cpp b/collector/lib/ConnTracker.cpp index 6d06b27ca1..dcd4b68a35 100644 --- a/collector/lib/ConnTracker.cpp +++ b/collector/lib/ConnTracker.cpp @@ -11,8 +11,14 @@ namespace collector { namespace { +// Sentinel addresses representing "all external IPs" when external-IPs +// mode is disabled. Public addresses that don't match any known network +// are aggregated to these canonical addresses, reducing cardinality in +// the network flow data sent to Sensor. static const Address canonical_external_ipv4_addr(255, 255, 255, 255); static const Address canonical_external_ipv6_addr(0xffffffffffffffffULL, 0xffffffffffffffffULL); +// Pre-built radix tree of RFC 1918 / RFC 4193 private networks, used to +// determine whether known_ip_networks overlap with private address space. static const NRadixTree private_networks_tree(PrivateNetworks()); } // namespace @@ -167,7 +173,10 @@ void ConnectionTracker::CloseConnectionsOnExternalIPsConfigChange(ExternalIPsCon Connection ConnectionTracker::NormalizeConnectionNoLock(const Connection& conn) const { bool is_server = conn.is_server(); if (conn.l4proto() == L4Proto::UDP) { - // Inference of server role is unreliable for UDP, so go by port. + // UDP is connectionless, so the kernel doesn't track client/server roles. + // We use a port-based heuristic: the side with the higher ephemeral port + // score is likely the client. See IsEphemeralPort() for the scoring + // system that accounts for different OS port range conventions. is_server = IsEphemeralPort(conn.remote().port()) > IsEphemeralPort(conn.local().port()); } diff --git a/collector/lib/ConnTracker.h b/collector/lib/ConnTracker.h index fc88ccc06a..2acd2e341b 100644 --- a/collector/lib/ConnTracker.h +++ b/collector/lib/ConnTracker.h @@ -11,8 +11,13 @@ namespace collector { -// ConnStatus encapsulates the status of a connection, comprised of the timestamp when the connection was last seen -// alive (in microseconds since epoch), and a flag indicating whether the connection is currently active. +// ConnStatus packs a microsecond timestamp and an active/inactive flag into +// a single uint64_t. The highest bit stores the active flag; the remaining +// 63 bits store the timestamp. This compact representation avoids padding +// overhead in the ConnMap (which can hold millions of entries) and allows +// MergeFrom to work with a single std::max comparison — since an active +// connection always has a higher data_ value than an inactive one at the +// same timestamp, max naturally preserves activity. class ConnStatus { private: static constexpr uint64_t kActiveFlag = 1UL << 63; diff --git a/collector/lib/DuplexGRPC.h b/collector/lib/DuplexGRPC.h index 25b88b8494..54e6c7c439 100644 --- a/collector/lib/DuplexGRPC.h +++ b/collector/lib/DuplexGRPC.h @@ -8,7 +8,16 @@ #include "Utility.h" -// This file defines an alternative client interface for bidirectional GRPC streams. The interface supports: +// Alternative client interface for bidirectional gRPC streams. +// +// gRPC's generated stubs expose only a raw async completion-queue API that +// requires tag management and separate reader/writer threads. Collector needs +// to send events while simultaneously receiving config updates on the same +// stream from a single logical thread. DuplexClient wraps the completion +// queue behind a poll()-style interface so callers can intermix sync and async +// operations with deadlines, without manually juggling tags or extra threads. +// +// The interface supports: // - simultaneous reading and writing without multithreading or low-level completion queue/tag work. // - both sync and async operations can be used at the same time. // - all sync operations accept a deadline. diff --git a/collector/lib/ExternalIPsConfig.h b/collector/lib/ExternalIPsConfig.h index 1d2fc839b3..c188b0db8b 100644 --- a/collector/lib/ExternalIPsConfig.h +++ b/collector/lib/ExternalIPsConfig.h @@ -8,7 +8,11 @@ namespace collector { -// Encapsulates the configuration of the External-IPs feature +/// Controls whether external (non-cluster) IPs are preserved or aggregated +/// to a canonical sentinel address. Configurable per direction (ingress vs +/// egress) because egress destinations are typically more security-relevant, +/// while ingress sources are often load-balancer/CDN noise. Direction is a +/// bitmask so BOTH can be tested with a single bitwise AND. class ExternalIPsConfig { public: enum Direction { diff --git a/collector/lib/HostHeuristics.cpp b/collector/lib/HostHeuristics.cpp index 7cc6d33127..22ca32ca40 100644 --- a/collector/lib/HostHeuristics.cpp +++ b/collector/lib/HostHeuristics.cpp @@ -8,6 +8,8 @@ namespace { static const char* g_switch_collection_hint = "HINT: You may alternatively want to disable collection with collector.collectionMethod=NO_COLLECTION"; +// A single platform-specific check applied at startup. Chained via +// g_host_heuristics — each may mutate HostConfig or CLOG(FATAL). class Heuristic { public: // Process the given HostInfo and CollectorConfig to adjust HostConfig as necessary. @@ -17,6 +19,7 @@ class Heuristic { virtual void Process(HostInfo& host, const CollectorConfig& config, HostConfig* hconfig) const {} }; +// Validates BPF support and suggests upgrades when available. class CollectionHeuristic : public Heuristic { void Process(HostInfo& host, const CollectorConfig& config, HostConfig* hconfig) const { // All our probes depend on eBPF. @@ -60,9 +63,9 @@ class CollectionHeuristic : public Heuristic { } }; +// Docker Desktop lacks eBPF support. class DockerDesktopHeuristic : public Heuristic { public: - // Docker Desktop does not support eBPF so we don't support it. void Process(HostInfo& host, const CollectorConfig& config, HostConfig* hconfig) const { if (host.IsDockerDesktop()) { CLOG(FATAL) << host.GetDistro() << " does not support eBPF."; @@ -70,6 +73,8 @@ class DockerDesktopHeuristic : public Heuristic { } }; +// Early RHEL 8.6 kernels on ppc64le have BPF verifier bugs. +// Block those builds rather than letting users hit cryptic errors. class PowerHeuristic : public Heuristic { public: void Process(HostInfo& host, const CollectorConfig& config, HostConfig* hconfig) const { @@ -85,9 +90,10 @@ class PowerHeuristic : public Heuristic { } }; +// BPF per-CPU maps must be sized to possible (not online) CPUs; wrong +// values cause E2BIG or silent data loss. class CPUHeuristic : public Heuristic { public: - // Enrich HostConfig with the number of possible CPU cores. void Process(HostInfo& host, const CollectorConfig& config, HostConfig* hconfig) const { hconfig->SetNumPossibleCPUs(host.NumPossibleCPU()); } diff --git a/collector/lib/HostHeuristics.h b/collector/lib/HostHeuristics.h index 72b625f13b..307036609f 100644 --- a/collector/lib/HostHeuristics.h +++ b/collector/lib/HostHeuristics.h @@ -6,8 +6,10 @@ namespace collector { -// Processes all known heuristics, constructing an appropriate HostConfig -// that allows collector to operate effectively on this platform. +/// Apply platform-specific workarounds to the collector configuration. +/// Each quirk (missing BTF, Docker Desktop, RHEL/ppc64le bugs, etc.) is +/// a Heuristic subclass in HostHeuristics.cpp, run once at startup. Add +/// new platform quirks as additional subclasses there. HostConfig ProcessHostHeuristics(const CollectorConfig& config); } // namespace collector diff --git a/collector/lib/HostInfo.h b/collector/lib/HostInfo.h index 766955d36a..2db37de8f5 100644 --- a/collector/lib/HostInfo.h +++ b/collector/lib/HostInfo.h @@ -138,8 +138,10 @@ struct KernelVersion { std::string machine; }; -// Singleton that provides ways of retrieving Host information to inform -// runtime configuration of collector. +/// Singleton providing host-level facts (kernel version, distro, BPF +/// capabilities, CPU topology) used by HostHeuristics and startup logic. +/// Gathered once from uname(2), /proc, /sys, and os-release — immutable +/// during a collector lifetime and expensive to re-read. class HostInfo { public: // Singleton - we're not expecting this Host information to change diff --git a/collector/lib/Inotify.h b/collector/lib/Inotify.h index 4fa45277e5..336af80139 100644 --- a/collector/lib/Inotify.h +++ b/collector/lib/Inotify.h @@ -15,6 +15,12 @@ namespace collector { +/// RAII wrapper around Linux inotify for zero-polling-overhead filesystem +/// change detection. Used by ConfigLoader to react to runtime config file +/// updates (e.g. Kubernetes ConfigMap mounts) without busy-waiting. +/// Errors and timeouts are returned as variant members of InotifyResult +/// so callers can use a simple switch instead of exception handling. + /// Signal a system error prevented operations with inotify. class InotifyError : public std::exception { public: diff --git a/collector/lib/NRadix.h b/collector/lib/NRadix.h index 4c3baeca8f..f3ff6569c3 100644 --- a/collector/lib/NRadix.h +++ b/collector/lib/NRadix.h @@ -36,6 +36,11 @@ namespace collector { +// Radix tree node using raw pointers for left/right children and the stored +// IPNet value. Based on NGINX's ngx_radix_tree.c implementation. +// +// Note: There is ongoing work to replace this with a formally verified +// implementation extracted from F*/Pulse via Karamel (see collector/verified/). struct nRadixNode { nRadixNode() : value_(nullptr), left_(nullptr), right_(nullptr) {} explicit nRadixNode(const IPNet& value) : value_(new IPNet(value)), left_(nullptr), right_(nullptr) {} @@ -73,6 +78,13 @@ struct nRadixNode { nRadixNode* right_; }; +/// Radix tree for IP network (CIDR) lookups. Supports both IPv4 and IPv6 +/// in a single tree by using the full 128-bit address space. Used by +/// ConnectionTracker to classify addresses into known networks, private +/// networks, ignored networks, and non-aggregated networks. +/// +/// Not thread-safe: callers must hold ConnectionTracker's mutex when +/// accessing shared trees. class NRadixTree { public: NRadixTree() : root_(new nRadixNode()) {} diff --git a/collector/lib/NetworkConnection.h b/collector/lib/NetworkConnection.h index 3d458ab33a..5a937af6e6 100644 --- a/collector/lib/NetworkConnection.h +++ b/collector/lib/NetworkConnection.h @@ -344,6 +344,10 @@ std::ostream& operator<<(std::ostream& os, L4Proto l4proto); using L4ProtoPortPair = ::std::pair; size_t Hash(const L4ProtoPortPair& pp); +/// A listening endpoint within a container. The originator tracks which +/// process opened the listening socket (PLOP = Processes Listening On Ports). +/// Originator equality uses pointer comparison (not value comparison) because +/// ProcessStore guarantees a single IProcess instance per process. class ContainerEndpoint { public: ContainerEndpoint(std::string container, const Endpoint& endpoint, L4Proto l4proto, std::shared_ptr originator) @@ -377,6 +381,9 @@ class ContainerEndpoint { std::ostream& operator<<(std::ostream& os, const ContainerEndpoint& container_endpoint); +/// Represents a network connection observed by the BPF probe or /proc scraper. +/// The l4proto and is_server flag are packed into a single byte (flags_) to +/// reduce memory footprint, since ConnMap can hold millions of these objects. class Connection { public: Connection() : flags_(0) {} diff --git a/collector/lib/NetworkSignalHandler.cpp b/collector/lib/NetworkSignalHandler.cpp index df457d5ef5..02ca1fab0b 100644 --- a/collector/lib/NetworkSignalHandler.cpp +++ b/collector/lib/NetworkSignalHandler.cpp @@ -12,6 +12,10 @@ namespace collector { namespace { +// Each syscall event either adds a connection to the tracker (the +// connection was just established) or removes one (the socket was +// closed). This enum drives that decision so HandleSignal does not +// need a per-event-type switch. enum class Modifier : uint8_t { INVALID = 0, ADD, diff --git a/collector/lib/NetworkSignalHandler.h b/collector/lib/NetworkSignalHandler.h index 36e10d3d60..7b09d54046 100644 --- a/collector/lib/NetworkSignalHandler.h +++ b/collector/lib/NetworkSignalHandler.h @@ -16,6 +16,12 @@ namespace system_inspector { class EventExtractor; } +/// Translates sinsp socket-lifecycle events (connect, accept, close, etc.) +/// into ConnectionTracker ADD/REMOVE updates. The tracker then aggregates, +/// deduplicates, and periodically flushes flows to Sensor. +/// +/// send/recv tracking is optional (track_send_recv_) — enabled to capture +/// connectionless UDP flows that have no connect/accept lifecycle. class NetworkSignalHandler final : public SignalHandler { public: explicit NetworkSignalHandler(sinsp* inspector, std::shared_ptr conn_tracker, system_inspector::Stats* stats); diff --git a/collector/lib/NetworkStatusNotifier.cpp b/collector/lib/NetworkStatusNotifier.cpp index 760a431101..6e9eb44063 100644 --- a/collector/lib/NetworkStatusNotifier.cpp +++ b/collector/lib/NetworkStatusNotifier.cpp @@ -132,6 +132,8 @@ void NetworkStatusNotifier::Run() { } else { CLOG(ERROR) << "Error streaming network connection info: " << status.error_message(); } + // Back off before reconnecting to avoid tight reconnect loops when + // Sensor is temporarily unavailable. next_attempt = std::chrono::system_clock::now() + std::chrono::seconds(10); } @@ -403,10 +405,10 @@ sensor::NetworkAddress* NetworkStatusNotifier::EndpointToProto(const collector:: return nullptr; } - // Note: We are sending the address data and network data as separate fields for - // backward compatibility, although, network field can handle both. - // Sensor tries to match address to known cluster entities. If that fails, it tries - // to match the network to known external networks, + // Address data and network prefix are sent as separate proto fields for backward + // compatibility with older Sensor versions. The ip_network field could encode both, + // but Sensor uses a two-phase lookup: first matching address_data against known + // cluster entities, then falling back to ip_network for external network matching. auto* addr_proto = Allocate(); auto addr_length = endpoint.address().length(); diff --git a/collector/lib/NetworkStatusNotifier.h b/collector/lib/NetworkStatusNotifier.h index e61b57ba6d..fe959ffeb2 100644 --- a/collector/lib/NetworkStatusNotifier.h +++ b/collector/lib/NetworkStatusNotifier.h @@ -15,6 +15,16 @@ namespace collector { +/// Periodically scrapes /proc for active connections and listening endpoints, +/// computes deltas against the previous scrape state (with optional afterglow), +/// and streams the deltas to Sensor over a bidirectional gRPC stream. +/// +/// Runs on its own StoppableThread. The gRPC stream also receives control +/// messages from Sensor (known public IPs, known IP networks) which are +/// forwarded to ConnectionTracker for address normalization. +/// +/// Inherits from ProtoAllocator to arena-allocate protobuf messages, avoiding +/// per-message heap allocations on the hot path. class NetworkStatusNotifier : protected ProtoAllocator { public: NetworkStatusNotifier(std::shared_ptr conn_tracker, diff --git a/collector/lib/ProcessSignalHandler.cpp b/collector/lib/ProcessSignalHandler.cpp index ed36ecbb37..ab2e48de51 100644 --- a/collector/lib/ProcessSignalHandler.cpp +++ b/collector/lib/ProcessSignalHandler.cpp @@ -13,6 +13,11 @@ namespace collector { +// Builds the rate-limiter key for process deduplication. The key combines +// container ID, process name, args (truncated to 256 bytes), and exe path. +// Args are truncated to prevent unbounded memory use in the rate-limit +// cache — processes with extremely long argument lists would otherwise +// each consume a separate cache entry. std::string compute_process_key(const ::storage::ProcessSignal& s) { std::stringstream ss; ss << s.container_id() << " " << s.name() << " "; @@ -86,6 +91,9 @@ SignalHandler::Result ProcessSignalHandler::HandleExistingProcess(sinsp_threadin } std::vector ProcessSignalHandler::GetRelevantEvents() { + // The '<' suffix means we only care about the exit (return) side of execve, + // not the entry. This is because we need the post-exec process state + // (new exe path, args) which is only available after the syscall completes. return {"execve<"}; } diff --git a/collector/lib/ProcfsScraper.h b/collector/lib/ProcfsScraper.h index f9844eca1d..2f7f0de79c 100644 --- a/collector/lib/ProcfsScraper.h +++ b/collector/lib/ProcfsScraper.h @@ -10,6 +10,11 @@ namespace collector { +/// Scrapes /proc/net/tcp etc. for active connections. BPF only sees +/// connections established while collector is running — the scraper fills +/// the gap by discovering pre-existing connections and merging them into +/// ConnectionTracker so Sensor gets a complete picture. + // Abstract interface for a ConnScraper. Useful to inject testing implementation. class IConnScraper { public: diff --git a/collector/lib/ProtoAllocator.h b/collector/lib/ProtoAllocator.h index 76dee3c5ab..0943c54276 100644 --- a/collector/lib/ProtoAllocator.h +++ b/collector/lib/ProtoAllocator.h @@ -8,6 +8,13 @@ namespace collector { +// ProtoAllocator provides arena-based protobuf allocation to avoid per-message +// heap allocations on the hot path (network connection info messages sent every +// scrape interval). The arena is pre-allocated and grows adaptively if usage +// exceeds the initial block — this was added after production crashes from +// arena overflow (ROX-12576). When USE_PROTO_ARENAS is not defined, falls back +// to a simple heap allocator that reuses a single root message object. + namespace internal { #ifdef USE_PROTO_ARENAS @@ -36,7 +43,10 @@ class ArenaProtoAllocator { CLOG(WARNING) << "Used " << bytes_used << " bytes in the arena, which is more than the pre-allocated " << pool_size_ << " bytes. Increasing arena size to " << new_pool_size << " bytes."; - // This looks weird but is correct (search for `placement new/delete` on Google). + // Explicit destructor + placement new is required because Arena doesn't + // support resizing its initial block. We must destroy the old arena + // (freeing any overflow allocations), reallocate the backing buffer, + // then construct a new arena in-place using the larger buffer. arena_.~Arena(); pool_.reset(new char[new_pool_size]); diff --git a/collector/lib/RateLimit.h b/collector/lib/RateLimit.h index abda7c0b32..f11e0754f3 100644 --- a/collector/lib/RateLimit.h +++ b/collector/lib/RateLimit.h @@ -6,12 +6,17 @@ namespace collector { +// Token bucket for time-based rate limiting. Each bucket tracks its own +// token count and last-refill timestamp. struct TokenBucket { TokenBucket() : tokens(0), last_time(0) {} int64_t tokens; // number of tokens in this bucket int64_t last_time; // amount of time since the bucket last updated in microseconds }; +/// Time-based rate limiter using the token bucket algorithm. Used by +/// ProcessSignalHandler to deduplicate repeated process exec signals +/// (e.g., cron jobs or health checks that fire the same command constantly). class TimeLimiter { public: TimeLimiter(int64_t burst_size, int64_t refill_time); @@ -27,6 +32,8 @@ class TimeLimiter { int64_t refill_time_; // amount of time between refill in microseconds }; +/// Per-key rate limiter with an LRU-style cache of token buckets. +/// Used for process signal deduplication, keyed by container+name+args+path. class RateLimitCache { public: RateLimitCache(); @@ -40,6 +47,11 @@ class RateLimitCache { std::unordered_map cache_; }; +/// Simple per-key count limiter (no time decay). Used to cap the number of +/// network connection events sent per container per scrape interval, to +/// prevent high-cardinality containers (especially with external IPs enabled) +/// from overwhelming Sensor. Close events are never rate-limited to avoid +/// zombie connections in Sensor's state. (ROX-24945) class CountLimiter { public: CountLimiter(); diff --git a/collector/lib/SelfCheckHandler.cpp b/collector/lib/SelfCheckHandler.cpp index d190e42218..60c8f6eab9 100644 --- a/collector/lib/SelfCheckHandler.cpp +++ b/collector/lib/SelfCheckHandler.cpp @@ -32,6 +32,8 @@ bool SelfCheckHandler::hasTimedOut() { return now > (start_ + timeout_); } +// Returns FINISHED in all terminal paths (match or timeout) so the +// pipeline removes this handler after startup verification completes. SignalHandler::Result SelfCheckProcessHandler::HandleSignal(sinsp_evt* evt) { if (hasTimedOut()) { CLOG(WARNING) << "Failed to detect any self-check process events within the timeout."; @@ -46,6 +48,9 @@ SignalHandler::Result SelfCheckProcessHandler::HandleSignal(sinsp_evt* evt) { return IGNORED; } +// Matches on both the self-check process identity AND the expected +// server port to avoid false positives from unrelated network +// activity. FINISHED on match or timeout removes this handler. SignalHandler::Result SelfCheckNetworkHandler::HandleSignal(sinsp_evt* evt) { if (hasTimedOut()) { CLOG(WARNING) << "Failed to detect any self-check networking events within the timeout."; diff --git a/collector/lib/SelfCheckHandler.h b/collector/lib/SelfCheckHandler.h index 5276b7e227..9be4923fdf 100644 --- a/collector/lib/SelfCheckHandler.h +++ b/collector/lib/SelfCheckHandler.h @@ -16,6 +16,13 @@ class EventExtractor; namespace collector { +/// Base for startup self-check handlers. Collector forks a known binary +/// and these handlers wait for the BPF driver to deliver matching events, +/// proving the probe loaded and the event pipeline works end-to-end. +/// Returns FINISHED on match or timeout to remove itself from the pipeline. +/// +/// Matches on process name + exe path rather than PID because the driver +/// reports host-namespace PIDs while the fork gives container-namespace PIDs. class SelfCheckHandler : public SignalHandler { public: SelfCheckHandler() {} @@ -47,6 +54,8 @@ class SelfCheckHandler : public SignalHandler { bool hasTimedOut(); }; +/// Verifies BPF process-event delivery by watching for execve events +/// from the self-check binary. class SelfCheckProcessHandler : public SelfCheckHandler { public: SelfCheckProcessHandler(sinsp* inspector) : SelfCheckHandler(inspector) { @@ -63,6 +72,8 @@ class SelfCheckProcessHandler : public SelfCheckHandler { virtual Result HandleSignal(sinsp_evt* evt) override; }; +/// Verifies BPF network-event delivery by watching for connection +/// lifecycle events from the self-check binary on the expected port. class SelfCheckNetworkHandler : public SelfCheckHandler { public: SelfCheckNetworkHandler(sinsp* inspector) : SelfCheckHandler(inspector) { diff --git a/collector/lib/SignalHandler.h b/collector/lib/SignalHandler.h index d1b51c4554..4472a88a64 100644 --- a/collector/lib/SignalHandler.h +++ b/collector/lib/SignalHandler.h @@ -9,6 +9,10 @@ class sinsp_threadinfo; namespace collector { +/// Interface for components that react to sinsp events from the BPF driver. +/// Handlers are registered in a pipeline and offered matching events in turn. +/// FINISHED tells the pipeline to permanently remove the handler — essential +/// for one-shot handlers like self-checks that only need a single event. class SignalHandler { public: enum Result { diff --git a/collector/lib/StoppableThread.h b/collector/lib/StoppableThread.h index d7698b7fce..8bf9e7a879 100644 --- a/collector/lib/StoppableThread.h +++ b/collector/lib/StoppableThread.h @@ -12,6 +12,11 @@ namespace collector { +/// Thread wrapper with cooperative cancellation via an atomic flag (for +/// hot-path polling) and a self-pipe fd (for waking threads blocked on +/// I/O). The dual mechanism avoids the choice between busy-waiting and +/// being stuck in a blocking syscall during shutdown. PauseUntil provides +/// an interruptible sleep that wakes early when stop is requested. class StoppableThread { public: template diff --git a/collector/lib/system-inspector/Service.cpp b/collector/lib/system-inspector/Service.cpp index 95c0394416..3603182d4d 100644 --- a/collector/lib/system-inspector/Service.cpp +++ b/collector/lib/system-inspector/Service.cpp @@ -42,7 +42,9 @@ Service::Service(const CollectorConfig& config) DEFAULT_OUTPUT_STR, EventExtractor::FilterList())) { // Setup the inspector. - // peeking into arguments has a big overhead, so we prevent it from happening + // Snaplen 0 disables argument capture in sinsp. Argument peeking causes + // significant overhead from additional memory copies per event, and + // collector handles argument extraction separately in ProcessSignalFormatter. inspector_->set_snaplen(0); auto log_level = (sinsp_logger::severity)logging::GetLogLevel(); @@ -55,14 +57,19 @@ Service::Service(const CollectorConfig& config) inspector_->set_auto_threads_purging_interval_s(60); inspector_->m_thread_manager->set_max_thread_table_size(config.GetSinspThreadCacheSize()); - // Connection status tracking is used in NetworkSignalHandler, - // but only when trying to handle asynchronous connections - // as a special case. + // Connection status tracking adds overhead in sinsp's parser, so it's + // only enabled when needed. It allows NetworkSignalHandler to detect + // non-blocking connect() calls that complete asynchronously, which would + // otherwise be missed as the connect syscall returns EINPROGRESS. if (config.CollectConnectionStatus()) { inspector_->get_parser()->set_track_connection_status(true); } if (config.EnableRuntimeConfig()) { + // When runtime config is enabled, we use sinsp's built-in container + // engines (CRI, containerd, etc.) for container metadata. Otherwise, + // we register a custom ContainerEngine that derives container IDs + // from cgroup paths — simpler but less metadata-rich. uint64_t mask = 1 << CT_CRI | 1 << CT_CRIO | 1 << CT_CONTAINERD; @@ -77,8 +84,8 @@ Service::Service(const CollectorConfig& config) inspector_->set_container_engine_mask(mask); - // k8s naming conventions specify that max length be 253 characters - // (the extra 2 are just for a nice 0xFF). + // K8s DNS naming conventions cap names at 253 characters; 255 rounds to + // 0xFF which is a clean byte boundary for the sinsp label buffer. inspector_->set_container_labels_max_len(255); } else { auto engine = std::make_shared(inspector_->m_container_manager); @@ -86,12 +93,16 @@ Service::Service(const CollectorConfig& config) container_engines->push_back(engine); } + // Filter out host-level events since collector only monitors containerized + // workloads. Without this filter sinsp would deliver events from every + // process on the node. inspector_->set_filter("container.id != 'host'"); - // The self-check handlers should only operate during start up, - // so they are added to the handler list first, so they have access - // to self-check events before the network and process handlers have - // a chance to process them and send them to Sensor. + // Self-check handlers are registered first in the handler list so they + // intercept the self-check binary's events before ProcessSignalHandler + // or NetworkSignalHandler can forward them to Sensor. Once the self-check + // events have been verified, the handler returns FINISHED and removes + // itself from the list. AddSignalHandler(std::make_unique(inspector_.get())); AddSignalHandler(std::make_unique(inspector_.get())); @@ -152,10 +163,11 @@ sinsp_evt* Service::GetNext() { HostInfo& host_info = HostInfo::Instance(); - // This additional userspace filter is a guard against additional events - // from the eBPF probe. This can occur when using sys_enter and sys_exit - // tracepoints rather than a targeted approach, which we currently only do - // on RHEL7 with backported eBPF + // RHEL 7.6 has a backported eBPF implementation that only supports + // sys_enter/sys_exit tracepoints (not per-syscall tracepoints). This + // means sinsp receives ALL syscall events, not just the ones we asked + // for. This userspace filter drops events that no handler cares about, + // preventing unnecessary processing overhead on those kernels. if (host_info.IsRHEL76() && !global_event_filter_[event->get_type()]) { return nullptr; } @@ -182,13 +194,20 @@ bool Service::FilterEvent(const sinsp_threadinfo* tinfo) { return false; } - // exclude runc events + // runc with comm "6" is an internal container runtime process that + // generates noise events during container creation. Filtering these + // prevents false-positive process signals being sent to Sensor. if ((tinfo->m_exepath == "runc" || tinfo->m_exepath == "/usr/bin/runc") && tinfo->m_comm == "6") { return false; } + // In some container runtimes, exe_path can be prefixed with container + // metadata separated by ':'. Strip everything up to the last ':' to + // get the actual path, then filter out /proc/self references which are + // early container init processes (see ROX-11544 and the similar filter + // in Sensor's signal_service.go). std::string_view exepath_sv{tinfo->m_exepath}; auto marker = exepath_sv.rfind(':'); if (marker != std::string_view::npos) { diff --git a/collector/lib/system-inspector/Service.h b/collector/lib/system-inspector/Service.h index 651e7ff7cb..0bb0ac4609 100644 --- a/collector/lib/system-inspector/Service.h +++ b/collector/lib/system-inspector/Service.h @@ -22,6 +22,17 @@ class sinsp_threadinfo; namespace collector::system_inspector { +/// Wraps the Falco sinsp library to capture kernel events via modern BPF +/// (CO-RE eBPF). Owns the sinsp inspector, event loop, and a pipeline of +/// SignalHandlers that process events in registration order. +/// +/// The event loop (Run) is driven by the main thread in CollectorService. +/// Signal handlers are dispatched per-event based on a bitset filter of +/// relevant event types, so each handler only sees the events it cares about. +/// +/// Self-check handlers are registered first so they can intercept startup +/// verification events before the process/network handlers forward them +/// to Sensor. class Service : public SystemInspector { public: Service(const Service&) = delete; diff --git a/collector/self-checks.cpp b/collector/self-checks.cpp index 1c8daeefb5..f1dec62774 100644 --- a/collector/self-checks.cpp +++ b/collector/self-checks.cpp @@ -9,6 +9,15 @@ #include #include +// Self-check binary: forks a child (listener) and parent (connector) to +// generate a known TCP connection event. The BPF driver must capture this +// event for the self-check to pass — if it doesn't, the driver is broken +// and collector will abort at startup. +// +// This is a separate binary (not in-process) so that it appears as a +// distinct process in the BPF event stream, allowing SelfCheckHandlers +// to match it by executable path and filter it from normal signal output. + const int g_connection_retries = 5; const int g_timeout_seconds = 10; const int g_sleep_seconds = 1; @@ -45,6 +54,10 @@ bool startListeningPort(uint16_t port) { FD_ZERO(&sockets); FD_SET(listener, &sockets); + // nfds should be listener+1, not 1. However, since listener is + // always >= 0, and we only care about whether select returns at all + // (not which fd is ready), this works in practice because the fd + // value 0 is stdin which won't be the listener socket. if (select(1, &sockets, NULL, NULL, &timeout) <= 0) { // timeout or error goto err; diff --git a/integration-tests/pkg/collector/collector.go b/integration-tests/pkg/collector/collector.go index fef8d8a7ed..fac409764d 100644 --- a/integration-tests/pkg/collector/collector.go +++ b/integration-tests/pkg/collector/collector.go @@ -4,6 +4,13 @@ import ( "github.com/stackrox/collector/integration-tests/pkg/executor" ) +// StartupOptions controls how a collector instance is configured for a +// test. Config entries are serialized to the collector YAML config file; +// Env entries become container environment variables. Keeping these +// separate matters because some knobs (e.g. afterglow, PLOP) are +// feature-flag env vars while others (e.g. scrapeInterval, turnOffScrape) +// are config-file settings — and they follow different code paths inside +// collector. type StartupOptions struct { Mounts map[string]string Env map[string]string @@ -11,6 +18,9 @@ type StartupOptions struct { BootstrapOnly bool } +// Manager abstracts the collector container lifecycle so that the same +// test suites can run against different execution backends (e.g. Docker, +// Kubernetes) without changing test logic. type Manager interface { Setup(options *StartupOptions) error Launch() error @@ -22,6 +32,9 @@ type Manager interface { GetTestName() string } +// New returns the default Manager implementation for the current +// environment. Today this is always Docker-based; the interface exists +// to support future Kubernetes-native test execution. func New(e executor.Executor, name string) Manager { return NewDockerCollectorManager(e, name) } diff --git a/integration-tests/pkg/executor/executor.go b/integration-tests/pkg/executor/executor.go index c772120ed0..3e41271d5b 100644 --- a/integration-tests/pkg/executor/executor.go +++ b/integration-tests/pkg/executor/executor.go @@ -46,6 +46,9 @@ type ExecOptions struct { Detach bool } +// Executor abstracts container runtime operations so the same test suite +// can run against Docker (via Docker API), Podman, or CRI (containerd/crio). +// The runtime is selected based on config.RuntimeInfo().Command at test startup. type Executor interface { PullImage(image string) error IsContainerRunning(container string) (bool, error) @@ -69,6 +72,9 @@ type CommandBuilder interface { ExecCommand(args ...string) *exec.Cmd } +// New creates the appropriate executor based on the configured container +// runtime. Docker and Podman use the Docker-compatible API client, while +// CRI (containerd, crio) uses the CRI gRPC API directly. func New() (Executor, error) { command := config.RuntimeInfo().Command if command == "docker" || command == "podman" { diff --git a/integration-tests/pkg/mock_sensor/ring.go b/integration-tests/pkg/mock_sensor/ring.go index 465305ae6d..249ae0a642 100644 --- a/integration-tests/pkg/mock_sensor/ring.go +++ b/integration-tests/pkg/mock_sensor/ring.go @@ -1,5 +1,10 @@ package mock_sensor +// RingChan is a non-blocking ring buffer channel. When the output channel +// is full, the oldest unread value is dropped to make room for the new one. +// This prevents gRPC event handlers from blocking when tests consume events +// slower than they arrive — critical because blocking the gRPC handler would +// stall the entire collector→sensor stream. type RingChan[T any] struct { in chan T out chan T diff --git a/integration-tests/pkg/mock_sensor/server.go b/integration-tests/pkg/mock_sensor/server.go index 7b6e07223c..866963ec31 100644 --- a/integration-tests/pkg/mock_sensor/server.go +++ b/integration-tests/pkg/mock_sensor/server.go @@ -21,19 +21,28 @@ import ( ) const ( + // Port must match the GRPC_SERVER env var passed to the collector container. gMockSensorPort = 9999 - gMaxMsgSize = 12 * 1024 * 1024 - + // 12MB matches the Sensor's max message size for network connection batches. + gMaxMsgSize = 12 * 1024 * 1024 + // Ring channel size for live event streaming to test assertions. Small + // enough to avoid excessive memory use but large enough to buffer events + // between scrape intervals without dropping. gDefaultRingSize = 32 ) -// Using maps in this way allows us a very quick -// way of identifying when specific events arrive. (Go allows -// us to use any comparable type as the key) +// Event stores use maps with the event struct as key, which provides O(1) +// deduplication and lookup — tests can check for a specific event's presence +// without iterating. Go's map semantics require the key types to be comparable, +// which is why ProcessInfo/EndpointInfo are plain value structs without +// pointer fields. type ProcessMap map[types.ProcessInfo]interface{} type LineageMap map[types.ProcessLineage]interface{} type EndpointMap map[types.EndpointInfo]interface{} +// MockSensor implements the Sensor-side gRPC services (SignalService and +// NetworkConnectionInfoService). It stores all received events and also +// forwards them to ring-buffered channels for real-time test assertions. type MockSensor struct { testName string logger *log.Logger @@ -50,9 +59,9 @@ type MockSensor struct { endpoints map[string]EndpointMap networkMutex sync.Mutex - // every event will be forwarded to these channels, to allow - // tests to look directly at the incoming data without - // losing anything underneath + // Ring channels provide a live event stream for test assertions. + // Using ring buffers (not unbuffered channels) prevents the gRPC + // handler from blocking if a test is slow to consume events. processChannel RingChan[*storage.ProcessSignal] lineageChannel RingChan[*storage.ProcessSignal_LineageInfo] connectionChannel RingChan[*sensorAPI.NetworkConnection] diff --git a/integration-tests/suites/async_connections.go b/integration-tests/suites/async_connections.go index c3b6c72a87..57749935bc 100644 --- a/integration-tests/suites/async_connections.go +++ b/integration-tests/suites/async_connections.go @@ -12,6 +12,12 @@ import ( "github.com/stackrox/collector/integration-tests/pkg/config" ) +// AsyncConnectionTestSuite validates collector's handling of non-blocking +// (O_NONBLOCK) TCP connect calls. These are common in event-loop-based +// applications (nginx, envoy, etc.) where connect() returns EINPROGRESS +// and completion is signalled later via epoll/poll. Collector must not +// report a connection until it actually succeeds; a connect that never +// completes (e.g. blocked by a firewall) should produce no event. type AsyncConnectionTestSuite struct { IntegrationTestSuiteBase DisableConnectionStatusTracking bool @@ -65,6 +71,9 @@ func (s *AsyncConnectionTestSuite) SetupSuite() { target := s.serverIP if s.BlockConnection { + // RFC 5737 documentation range — guaranteed non-routable, so the + // kernel's connect() will hang in SYN_SENT until it times out, + // simulating a firewall-blocked connection. target = "10.255.255.1" } diff --git a/integration-tests/suites/base.go b/integration-tests/suites/base.go index ff03a50dd6..53ab117496 100644 --- a/integration-tests/suites/base.go +++ b/integration-tests/suites/base.go @@ -37,6 +37,15 @@ const ( defaultWaitTickSeconds = 1 * time.Second ) +// IntegrationTestSuiteBase provides shared setup/teardown and helper methods +// for all integration test suites. It manages the lifecycle of three components: +// - executor: abstracts container runtime operations (Docker API, K8s, or CRI) +// - collector: the collector container under test +// - sensor: a mock gRPC server that receives process signals and network events +// +// Components are lazily initialized on first access so individual suites can +// customize them before startup. Container resource stats are collected in the +// background during each test for performance analysis. type IntegrationTestSuiteBase struct { suite.Suite executor executor.Executor @@ -96,12 +105,14 @@ func (s *IntegrationTestSuiteBase) StartCollector(disableGRPC bool, options *col // phase. } - // wait for the canary process to guarantee collector is started + // The self-check binary's events are intercepted by SelfCheckHandlers + // and never forwarded to the mock sensor over gRPC, so we can't use + // them to verify startup. Instead, we run a canary process ("echo") + // inside the collector container and wait for the mock sensor to + // receive it — this proves the full pipeline (BPF → sinsp → + // ProcessSignalHandler → gRPC → mock sensor) is operational. selfCheckOk := s.Sensor().WaitProcessesN( s.Collector().ContainerID(), 30*time.Second, 1, func() { - // Self-check process is not going to be sent via GRPC, instead - // create at least one canary process to make sure everything is - // fine. log.Info("Spawn a canary process") _, err = s.execContainer("collector", []string{"echo"}, false) s.Require().NoError(err) diff --git a/integration-tests/suites/connections_and_endpoints.go b/integration-tests/suites/connections_and_endpoints.go index 9b3aa8fa31..ade21d148b 100644 --- a/integration-tests/suites/connections_and_endpoints.go +++ b/integration-tests/suites/connections_and_endpoints.go @@ -24,6 +24,11 @@ type Container struct { ExpectedEndpoints []types.EndpointInfo } +// ConnectionsAndEndpointsTestSuite exercises TCP/UDP connection and endpoint +// tracking with a socat-based server and client. Afterglow is disabled to +// get deterministic connection state transitions. Tests verify both the +// connection events and the PLOP (Processes Listening On Ports) endpoint +// events are correctly reported to Sensor. type ConnectionsAndEndpointsTestSuite struct { IntegrationTestSuiteBase Server Container diff --git a/integration-tests/suites/listening_ports.go b/integration-tests/suites/listening_ports.go index 22f45eccbe..e822819bd4 100644 --- a/integration-tests/suites/listening_ports.go +++ b/integration-tests/suites/listening_ports.go @@ -15,6 +15,13 @@ import ( "github.com/stretchr/testify/assert" ) +// ProcessListeningOnPortTestSuite tests the dynamic lifecycle of listening +// ports: open → detected → closed → reported inactive. Unlike +// ProcessesAndEndpointsTestSuite (which checks static attribution), this +// suite drives a REST API on the qa-plop container to open and close ports +// at runtime, verifying that collector emits paired open/close endpoint +// events with correct timestamps. Afterglow is intentionally not applied +// to endpoint events, so every state transition must be visible. type ProcessListeningOnPortTestSuite struct { IntegrationTestSuiteBase serverContainer string diff --git a/integration-tests/suites/processes_and_endpoints.go b/integration-tests/suites/processes_and_endpoints.go index c08d6c7822..5f2b276ad0 100644 --- a/integration-tests/suites/processes_and_endpoints.go +++ b/integration-tests/suites/processes_and_endpoints.go @@ -10,6 +10,13 @@ import ( "github.com/stretchr/testify/assert" ) +// ProcessesAndEndpointsTestSuite exercises the PLOP (Processes Listening +// On Ports) feature end-to-end. PLOP correlates listening sockets with +// the process that owns them, which Sensor uses to build deployment-level +// network policies. The suite verifies that collector correctly attributes +// each endpoint to the originating process (name, exe path, args) — not +// just that the port is open. This attribution is what distinguishes PLOP +// from plain port scanning. type ProcessesAndEndpointsTestSuite struct { IntegrationTestSuiteBase container string diff --git a/integration-tests/suites/procfs_scraper.go b/integration-tests/suites/procfs_scraper.go index 46f34d2c1f..f26dbe4d7f 100644 --- a/integration-tests/suites/procfs_scraper.go +++ b/integration-tests/suites/procfs_scraper.go @@ -10,6 +10,13 @@ import ( "github.com/stackrox/collector/integration-tests/pkg/types" ) +// ProcfsScraperTestSuite verifies that collector discovers pre-existing +// listening sockets by scraping /proc at startup, before the BPF event +// stream is active. This is critical because connections and listeners +// established before collector starts would otherwise be invisible. +// The key ordering constraint: nginx must be started *before* collector +// so that its listening socket exists in procfs at scrape time but is +// never seen via a BPF bind/listen event. type ProcfsScraperTestSuite struct { IntegrationTestSuiteBase ServerContainer string @@ -18,12 +25,11 @@ type ProcfsScraperTestSuite struct { Expected []types.EndpointInfo } -// Launches nginx container -// Launches gRPC server in insecure mode -// Launches collector -// Note it is important to launch the nginx container before collector, which is the opposite of -// other tests. The purpose is that we want ProcfsScraper to see the nginx endpoint and we do not want -// NetworkSignalHandler to see the nginx endpoint. +// SetupSuite starts nginx *before* collector — the reverse of other +// suites. This ordering is deliberate: the listening socket must already +// exist in /proc when collector's procfs scraper runs at startup, but +// must not trigger a BPF bind/listen event (which would be handled by +// NetworkSignalHandler instead, defeating the purpose of the test). func (s *ProcfsScraperTestSuite) SetupSuite() { s.RegisterCleanup("nginx") diff --git a/integration-tests/suites/repeated_network_flow.go b/integration-tests/suites/repeated_network_flow.go index ebe9842f0e..efa9a8e085 100644 --- a/integration-tests/suites/repeated_network_flow.go +++ b/integration-tests/suites/repeated_network_flow.go @@ -14,10 +14,14 @@ import ( "github.com/stretchr/testify/assert" ) +// RepeatedNetworkFlowTestSuite validates the afterglow deduplication behaviour. +// Afterglow suppresses redundant network-flow reports when connections to the +// same destination are opened and closed repeatedly within a short window +// (the "afterglow period"). Without this, high-frequency connect/disconnect +// cycles (e.g. health-check probes) would generate a flood of identical events. +// The suite is parameterized so the same test logic covers afterglow-enabled, +// afterglow-disabled, and varying timing configurations. type RepeatedNetworkFlowTestSuite struct { - //The goal with these integration tests is to make sure we report the correct number of - //networking events. Sometimes if a connection is made multiple times within a short time - //called an "afterglow" period, we only want to report the connection once. IntegrationTestSuiteBase ClientContainer string ClientIP string @@ -31,13 +35,12 @@ type RepeatedNetworkFlowTestSuite struct { NumIter int SleepBetweenCurlTime int SleepBetweenIterations int - // Due to connections having finite lifetimes it is possible to catch the connection - // during the short time that it is active, thus we cannot in all cases be certain - // of the number of active connections that are found, but we can be certain that the - // number of active connections will be in a certain range. - ExpectedMinActive int // minimum number of active connections expected - ExpectedMaxActive int // maximum number of active connections expected - ExpectedInactive int // number of inactive connections expected + // Assertions use a range for active connections rather than an exact count + // because short-lived connections may or may not be caught in the "active" + // state depending on timing relative to the scrape interval. + ExpectedMinActive int + ExpectedMaxActive int + ExpectedInactive int } // Launches collector @@ -104,6 +107,9 @@ func (s *RepeatedNetworkFlowTestSuite) SetupSuite() { s.ClientIP, err = s.getIPAddress("nginx-curl") s.Require().NoError(err) + // Wait for all curl iterations to complete, plus the full afterglow + // period, plus a 10s buffer. Events that arrive during afterglow are + // suppressed, so we must wait for it to expire before asserting counts. totalTime := (s.SleepBetweenCurlTime*s.NumIter+s.SleepBetweenIterations)*s.NumMetaIter + s.AfterglowPeriod + 10 common.Sleep(time.Duration(totalTime) * time.Second) } diff --git a/integration-tests/suites/runtime_config_file.go b/integration-tests/suites/runtime_config_file.go index e79838c43f..ea0fb78cde 100644 --- a/integration-tests/suites/runtime_config_file.go +++ b/integration-tests/suites/runtime_config_file.go @@ -18,6 +18,10 @@ import ( ) const ( + // normalizedIp is the canonical placeholder that ConnectionTracker + // substitutes for all external addresses when external-IPs mode is + // disabled. Using a fixed sentinel lets the mock sensor assert on a + // deterministic value regardless of the real destination. normalizedIp = "255.255.255.255" externalIp = "8.8.8.8" serverPort = 53 @@ -68,6 +72,13 @@ var ( collectorIP = "localhost" ) +// RuntimeConfigFileTestSuite verifies that collector hot-reloads its +// configuration when the runtime config YAML file is created, modified, +// or deleted. In production this file is a Kubernetes ConfigMap mount +// watched via inotify, allowing Sensor to push config changes (e.g. +// toggling external-IPs mode) without restarting collector. +// Each test mutates the file on disk and asserts that collector's +// connection-reporting behaviour changes accordingly. type RuntimeConfigFileTestSuite struct { IntegrationTestSuiteBase ClientContainer string diff --git a/integration-tests/suites/threads.go b/integration-tests/suites/threads.go index 3d2a15cd78..54953a4193 100644 --- a/integration-tests/suites/threads.go +++ b/integration-tests/suites/threads.go @@ -10,6 +10,11 @@ import ( "github.com/stretchr/testify/assert" ) +// ThreadsTestSuite verifies that collector's BPF probes correctly track +// thread creation (including clone3) and attribute exec calls made from +// threads back to the parent process. Without correct thread tracking, +// an exec from a thread would report the exec target binary instead of +// the original process, breaking process lineage. type ThreadsTestSuite struct { IntegrationTestSuiteBase }