fix: include custom container plugin and filtering

collector/CMakeLists.txt

@@ -49,6 +49,8 @@ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/lib/CollectorVersion.h.in ${CMAKE_CUR
 
 set(FALCO_DIR ${PROJECT_SOURCE_DIR}/../falcosecurity-libs)
 
+add_subdirectory(container-plugin)
+
 add_subdirectory(${PROJECT_SOURCE_DIR}/proto)
 
 include_directories(${PROJECT_SOURCE_DIR}/lib)

collector/Makefile

@@ -9,9 +9,9 @@ COLLECTOR_BIN_DIR = $(CMAKE_DIR)/collector
 LIBSINSP_BIN_DIR = $(CMAKE_DIR)/collector/EXCLUDE_FROM_DEFAULT_BUILD/libsinsp
 SRC_MOUNT_DIR = /tmp/collector
 
-HDRS := $(wildcard lib/*.h) $(shell find $(BASE_PATH)/falcosecurity-libs/userspace -name '*.h')
+HDRS := $(wildcard lib/*.h) $(wildcard container-plugin/*.h) $(shell find $(BASE_PATH)/falcosecurity-libs/userspace -name '*.h')
 
-SRCS := $(wildcard lib/*.cpp) collector.cpp
+SRCS := $(wildcard lib/*.cpp) $(wildcard container-plugin/*.cpp) collector.cpp
 
 COLLECTOR_BUILD_DEPS := $(HDRS) $(SRCS) $(shell find $(BASE_PATH)/falcosecurity-libs -name '*.h' -o -name '*.cpp' -o -name '*.c')
 
@@ -37,8 +37,10 @@ cmake-build/collector: cmake-configure/collector $(COLLECTOR_BUILD_DEPS)
 
 container/bin/collector: cmake-build/collector
 	mkdir -p container/bin
+	mkdir -p container/libs
 	cp "$(COLLECTOR_BIN_DIR)/collector" container/bin/collector
 	cp "$(COLLECTOR_BIN_DIR)/self-checks" container/bin/self-checks
+	cp "$(COLLECTOR_BIN_DIR)/collector-container-plugin.so" container/libs/collector-container-plugin.so
 
 .PHONY: collector
 collector: container/bin/collector txt-files

collector/container-plugin/CMakeLists.txt

@@ -0,0 +1,12 @@
+add_library(collector-container-plugin SHARED ContainerPlugin.cpp)
+
+set_target_properties(collector-container-plugin PROPERTIES
+    PREFIX ""
+    OUTPUT_NAME "collector-container-plugin"
+    LIBRARY_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/collector"
+)
+
+target_include_directories(collector-container-plugin PRIVATE
+    ${FALCO_DIR}/userspace
+    ${FALCO_DIR}/driver
+)

collector/container-plugin/ContainerID.h

@@ -0,0 +1,39 @@
+#pragma once
+
+#include <algorithm>
+#include <cctype>
+#include <optional>
+#include <string_view>
+
+namespace collector::container_plugin {
+
+constexpr size_t kContainerIDLength = 64;
+constexpr size_t kShortContainerIDLength = 12;
+
+inline std::optional<std::string_view> ExtractContainerIDFromCgroup(std::string_view cgroup) {
+  const auto scope = cgroup.rfind(".scope");
+  if (scope != std::string_view::npos) {
+    cgroup.remove_suffix(cgroup.size() - scope);
+  }
+  if (cgroup.size() < kContainerIDLength + 1) {
+    return {};
+  }
+  const auto id_start = cgroup.size() - kContainerIDLength;
+  const char separator = cgroup[id_start - 1];
+  if (separator != '/' && separator != '-' && separator != ':') {
+    return {};
+  }
+  const std::string_view parent = cgroup.substr(0, id_start - 1);
+  constexpr std::string_view kConmonSuffix = "-conmon";
+  if (parent.size() >= kConmonSuffix.size() &&
+      parent.substr(parent.size() - kConmonSuffix.size()) == kConmonSuffix) {
+    return {};
+  }
+  const std::string_view id = cgroup.substr(id_start);
+  if (!std::all_of(id.begin(), id.end(), [](char c) { return std::isxdigit(static_cast<unsigned char>(c)); })) {
+    return {};
+  }
+  return id.substr(0, kShortContainerIDLength);
+}
+
+}  // namespace collector::container_plugin

collector/container-plugin/ContainerPlugin.cpp

@@ -0,0 +1,260 @@
+#include <cctype>
+#include <cstdint>
+#include <ppm_events_public.h>
+#include <string>
+#include <string_view>
+
+#include <plugin/plugin_api.h>
+
+#include "ContainerID.h"
+
+namespace {
+
+constexpr std::string_view kPluginName = "collector-container";
+constexpr std::string_view kPluginVersion = "0.1.0";
+constexpr std::string_view kHostContainerID = "host";
+
+struct plugin_state {
+  std::string last_error;
+  ss_plugin_table_t* threads = nullptr;
+  ss_plugin_table_field_t* cgroups = nullptr;
+  ss_plugin_table_field_t* cgroup_path = nullptr;
+  ss_plugin_table_field_t* container_id = nullptr;
+  std::string extracted_container_id;
+  const char* extracted_container_id_ptr = nullptr;
+};
+
+std::string ContainerIDFromCgroup(std::string_view cgroup) {
+  return std::string(collector::container_plugin::ExtractContainerIDFromCgroup(cgroup).value_or(std::string_view{}));
+}
+
+struct cgroup_iteration_state {
+  plugin_state* plugin;
+  ss_plugin_table_reader_vtable_ext* reader;
+  ss_plugin_table_t* cgroup_table;
+  std::string container_id;
+};
+
+ss_plugin_bool FindContainerID(ss_plugin_table_iterator_state_t* data, ss_plugin_table_entry_t* entry) {
+  auto* state = reinterpret_cast<cgroup_iteration_state*>(data);
+  ss_plugin_state_data value{};
+  if (state->reader->read_entry_field(state->cgroup_table, entry, state->plugin->cgroup_path, &value) != SS_PLUGIN_SUCCESS) {
+    return 0;
+  }
+  if (value.str != nullptr) {
+    state->container_id = ContainerIDFromCgroup(value.str);
+  }
+  return 1;
+}
+
+ss_plugin_rc CacheContainerID(plugin_state* state,
+                              ss_plugin_table_entry_t* thread,
+                              ss_plugin_table_reader_vtable_ext* reader,
+                              ss_plugin_table_writer_vtable_ext* writer) {
+  ss_plugin_state_data cgroups{};
+  if (reader->read_entry_field(state->threads, thread, state->cgroups, &cgroups) != SS_PLUGIN_SUCCESS ||
+      cgroups.table == nullptr) {
+    state->last_error = "failed to read thread cgroups";
+    return SS_PLUGIN_FAILURE;
+  }
+
+  cgroup_iteration_state iteration{state, reader, cgroups.table, {}};
+  if (!reader->iterate_entries(cgroups.table, FindContainerID,
+                               reinterpret_cast<ss_plugin_table_iterator_state_t*>(&iteration))) {
+    state->last_error = "failed to inspect thread cgroups";
+    return SS_PLUGIN_FAILURE;
+  }
+
+  ss_plugin_state_data value{};
+  const std::string id = iteration.container_id.empty() ? std::string(kHostContainerID) : iteration.container_id;
+  value.str = id.c_str();
+  if (writer->write_entry_field(state->threads, thread, state->container_id, &value) != SS_PLUGIN_SUCCESS) {
+    state->last_error = "failed to cache thread container ID";
+    return SS_PLUGIN_FAILURE;
+  }
+  return SS_PLUGIN_SUCCESS;
+}
+
+struct thread_iteration_state {
+  plugin_state* plugin;
+  ss_plugin_table_reader_vtable_ext* reader;
+  ss_plugin_table_writer_vtable_ext* writer;
+};
+
+ss_plugin_bool CacheInitialThread(ss_plugin_table_iterator_state_t* data, ss_plugin_table_entry_t* entry) {
+  auto* state = reinterpret_cast<thread_iteration_state*>(data);
+  return CacheContainerID(state->plugin, entry, state->reader, state->writer) == SS_PLUGIN_SUCCESS;
+}
+
+}  // namespace
+
+extern "C" const char* plugin_get_required_api_version() {
+  return PLUGIN_API_VERSION_STR;
+}
+
+extern "C" const char* plugin_get_version() {
+  return kPluginVersion.data();
+}
+
+extern "C" const char* plugin_get_name() {
+  return kPluginName.data();
+}
+
+extern "C" const char* plugin_get_description() {
+  return "Caches Collector container IDs in Falco thread state";
+}
+
+extern "C" const char* plugin_get_contact() {
+  return "https://github.com/stackrox/collector";
+}
+
+extern "C" const char* plugin_get_required_event_schema_version(ss_plugin_t*) {
+  return "4.1.0";
+}
+
+extern "C" ss_plugin_t* plugin_init(const ss_plugin_init_input* input, ss_plugin_rc* rc) {
+  auto* state = new plugin_state;
+  *rc = SS_PLUGIN_FAILURE;
+  if (input == nullptr || input->tables == nullptr || input->tables->fields_ext == nullptr ||
+      input->tables->reader_ext == nullptr || input->tables->writer_ext == nullptr) {
+    state->last_error = "Falco table API is unavailable";
+    return reinterpret_cast<ss_plugin_t*>(state);
+  }
+
+  state->threads = input->tables->get_table(input->owner, "threads", SS_PLUGIN_ST_INT64);
+  if (state->threads == nullptr) {
+    state->last_error = "failed to access Falco threads table";
+    return reinterpret_cast<ss_plugin_t*>(state);
+  }
+  state->cgroups = input->tables->fields_ext->get_table_field(state->threads, "cgroups", SS_PLUGIN_ST_TABLE);
+  state->container_id = input->tables->fields_ext->add_table_field(state->threads, "container_id", SS_PLUGIN_ST_STRING);
+  if (state->cgroups == nullptr || state->container_id == nullptr) {
+    state->last_error = "failed to access Falco thread cgroup or container ID fields";
+    return reinterpret_cast<ss_plugin_t*>(state);
+  }
+
+  ss_plugin_table_entry_t* entry = input->tables->writer_ext->create_table_entry(state->threads);
+  ss_plugin_state_data cgroups{};
+  if (entry == nullptr || input->tables->reader_ext->read_entry_field(state->threads, entry, state->cgroups, &cgroups) != SS_PLUGIN_SUCCESS ||
+      cgroups.table == nullptr) {
+    if (entry != nullptr) {
+      input->tables->writer_ext->destroy_table_entry(state->threads, entry);
+    }
+    state->last_error = "failed to access Falco cgroup table";
+    return reinterpret_cast<ss_plugin_t*>(state);
+  }
+  state->cgroup_path = input->tables->fields_ext->get_table_field(cgroups.table, "second", SS_PLUGIN_ST_STRING);
+  input->tables->writer_ext->destroy_table_entry(state->threads, entry);
+  if (state->cgroup_path == nullptr) {
+    state->last_error = "failed to access Falco cgroup path field";
+    return reinterpret_cast<ss_plugin_t*>(state);
+  }
+
+  *rc = SS_PLUGIN_SUCCESS;
+  return reinterpret_cast<ss_plugin_t*>(state);
+}
+
+extern "C" void plugin_destroy(ss_plugin_t* plugin) {
+  delete reinterpret_cast<plugin_state*>(plugin);
+}
+
+extern "C" const char* plugin_get_last_error(ss_plugin_t* plugin) {
+  return reinterpret_cast<plugin_state*>(plugin)->last_error.c_str();
+}
+
+extern "C" const char* plugin_get_parse_event_sources() {
+  return "[\"syscall\"]";
+}
+
+extern "C" uint16_t* plugin_get_parse_event_types(uint32_t* count, ss_plugin_t*) {
+  static uint16_t event_types[] = {
+      PPME_SYSCALL_CLONE_20_X,
+      PPME_SYSCALL_FORK_20_X,
+      PPME_SYSCALL_VFORK_20_X,
+      PPME_SYSCALL_CLONE3_X,
+      PPME_SYSCALL_EXECVE_16_X,
+      PPME_SYSCALL_EXECVE_17_X,
+      PPME_SYSCALL_EXECVE_18_X,
+      PPME_SYSCALL_EXECVE_19_X,
+      PPME_SYSCALL_EXECVEAT_X,
+      PPME_SYSCALL_CHROOT_X,
+  };
+  *count = sizeof(event_types) / sizeof(event_types[0]);
+  return event_types;
+}
+
+extern "C" const char* plugin_get_fields() {
+  return R"([{"type":"string","name":"container.id","desc":"Cached container ID for the event thread"}])";
+}
+
+extern "C" const char* plugin_get_extract_event_sources() {
+  return "[\"syscall\"]";
+}
+
+extern "C" ss_plugin_rc plugin_extract_fields(ss_plugin_t* plugin,
+                                              const ss_plugin_event_input* event,
+                                              const ss_plugin_field_extract_input* input) {
+  auto* state = reinterpret_cast<plugin_state*>(plugin);
+  ss_plugin_state_data key{};
+  key.s64 = static_cast<int64_t>(event->evt->tid);
+  ss_plugin_table_entry_t* thread = input->table_reader_ext->get_table_entry(state->threads, &key);
+  if (thread == nullptr) {
+    for (uint32_t i = 0; i < input->num_fields; ++i) {
+      input->fields[i].res_len = 0;
+    }
+    return SS_PLUGIN_SUCCESS;
+  }
+
+  ss_plugin_state_data value{};
+  const ss_plugin_rc read_rc = input->table_reader_ext->read_entry_field(state->threads, thread, state->container_id, &value);
+  input->table_reader_ext->release_table_entry(state->threads, thread);
+  if (read_rc != SS_PLUGIN_SUCCESS || value.str == nullptr) {
+    state->last_error = "failed to read cached thread container ID";
+    return SS_PLUGIN_FAILURE;
+  }
+
+  state->extracted_container_id = value.str;
+  state->extracted_container_id_ptr = state->extracted_container_id.c_str();
+  for (uint32_t i = 0; i < input->num_fields; ++i) {
+    if (input->fields[i].field_id != 0) {
+      input->fields[i].res_len = 0;
+      continue;
+    }
+    input->fields[i].res.str = &state->extracted_container_id_ptr;
+    input->fields[i].res_len = 1;
+  }
+  return SS_PLUGIN_SUCCESS;
+}
+
+extern "C" ss_plugin_rc plugin_parse_event(ss_plugin_t* plugin,
+                                           const ss_plugin_event_input* event,
+                                           const ss_plugin_event_parse_input* input) {
+  auto* state = reinterpret_cast<plugin_state*>(plugin);
+  ss_plugin_state_data key{};
+  key.s64 = static_cast<int64_t>(event->evt->tid);
+  ss_plugin_table_entry_t* thread = input->table_reader_ext->get_table_entry(state->threads, &key);
+  if (thread == nullptr) {
+    return SS_PLUGIN_SUCCESS;
+  }
+  const ss_plugin_rc rc = CacheContainerID(state, thread, input->table_reader_ext, input->table_writer_ext);
+  input->table_reader_ext->release_table_entry(state->threads, thread);
+  return rc;
+}
+
+extern "C" ss_plugin_rc plugin_capture_open(ss_plugin_t* plugin, const ss_plugin_capture_listen_input* input) {
+  auto* state = reinterpret_cast<plugin_state*>(plugin);
+  thread_iteration_state iteration{state, input->table_reader_ext, input->table_writer_ext};
+  if (!input->table_reader_ext->iterate_entries(
+          state->threads, CacheInitialThread,
+          reinterpret_cast<ss_plugin_table_iterator_state_t*>(&iteration))) {
+    if (state->last_error.empty()) {
+      state->last_error = "failed to cache initial thread container IDs";
+    }
+    return SS_PLUGIN_FAILURE;
+  }
+  return SS_PLUGIN_SUCCESS;
+}
+
+extern "C" ss_plugin_rc plugin_capture_close(ss_plugin_t*, const ss_plugin_capture_listen_input*) {
+  return SS_PLUGIN_SUCCESS;
+}

collector/container/Dockerfile

@@ -39,6 +39,7 @@ COPY container/THIRD_PARTY_NOTICES/ /THIRD_PARTY_NOTICES/
 COPY kernel-modules /kernel-modules
 COPY container/bin/collector /usr/local/bin/
 COPY container/bin/self-checks /usr/local/bin/self-checks
+COPY container/libs/collector-container-plugin.so /usr/local/lib/collector/collector-container-plugin.so
 COPY container/status-check.sh /usr/local/bin/status-check.sh
 
 EXPOSE 8080 9090

collector/container/dev.Dockerfile

@@ -23,6 +23,7 @@ COPY container/THIRD_PARTY_NOTICES/ /THIRD_PARTY_NOTICES/
 COPY kernel-modules /kernel-modules
 COPY container/bin/collector /usr/local/bin/
 COPY container/bin/self-checks /usr/local/bin/self-checks
+COPY container/libs/collector-container-plugin.so /usr/local/lib/collector/collector-container-plugin.so
 COPY container/status-check.sh /usr/local/bin/status-check.sh
 
 EXPOSE 8080 9090

collector/container/konflux.Dockerfile

@@ -135,6 +135,7 @@ COPY --from=package_installer /out/ /
 
 COPY --from=builder ${CMAKE_BUILD_DIR}/collector/collector /usr/local/bin/
 COPY --from=builder ${CMAKE_BUILD_DIR}/collector/self-checks /usr/local/bin/
+COPY --from=builder ${CMAKE_BUILD_DIR}/collector/collector-container-plugin.so /usr/local/lib/collector/collector-container-plugin.so
 
 COPY LICENSE /licenses/LICENSE
 

collector/lib/CollectorConfig.cpp

@@ -82,6 +82,7 @@ PathEnvVar tls_client_cert_path("ROX_COLLECTOR_TLS_CLIENT_CERT");
 PathEnvVar tls_client_key_path("ROX_COLLECTOR_TLS_CLIENT_KEY");
 
 BoolEnvVar disable_process_arguments("ROX_COLLECTOR_NO_PROCESS_ARGUMENTS", false);
+StringEnvVar container_plugin_path("ROX_COLLECTOR_CONTAINER_PLUGIN_PATH", "/usr/local/lib/collector/collector-container-plugin.so");
 }  // namespace
 
 constexpr bool CollectorConfig::kTurnOffScrape;
@@ -97,6 +98,7 @@ CollectorConfig::CollectorConfig() {
   scrape_interval_ = kScrapeInterval;
   turn_off_scrape_ = kTurnOffScrape;
   collection_method_ = kCollectionMethod;
+  container_plugin_path_ = container_plugin_path.value();
 }
 
 void CollectorConfig::InitCollectorConfig(CollectorArgs* args) {