feat: Integrate coolbpf cpu profiling feature in loongcollector

core/CMakeLists.txt

@@ -158,7 +158,7 @@ if (LINUX)
     if (ENABLE_ENTERPRISE)
         set(SUB_DIRECTORIES_LIST ${SUB_DIRECTORIES_LIST} shennong shennong/sdk apm/forward)
     endif()
-    set(SUB_DIRECTORIES_LIST ${SUB_DIRECTORIES_LIST} ebpf ebpf/type ebpf/type/table ebpf/util ebpf/util/sampler ebpf/protocol/http ebpf/protocol/mysql ebpf/protocol ebpf/plugin/file_security ebpf/plugin/network_observer ebpf/plugin/process_security ebpf/plugin/network_security ebpf/plugin ebpf/observer ebpf/security
+    set(SUB_DIRECTORIES_LIST ${SUB_DIRECTORIES_LIST} ebpf ebpf/type ebpf/type/table ebpf/util ebpf/util/sampler ebpf/protocol/http ebpf/protocol/mysql ebpf/protocol ebpf/plugin/file_security ebpf/plugin/network_observer ebpf/plugin/process_security ebpf/plugin/network_security ebpf/plugin/cpu_profiling ebpf/plugin ebpf/observer ebpf/security
         prometheus prometheus/labels prometheus/schedulers prometheus/async prometheus/component
         host_monitor host_monitor/collector host_monitor/common forward forward/loongsuite
         )

core/app_config/AppConfig.h

@@ -552,6 +552,7 @@ class AppConfig {
     friend class PollingPreservedDirDepthUnittest;
     friend class InputStaticFileUnittest;
     friend class LogInputReaderUnittest;
+    friend class eBPFServerUnittest;
 #endif
 };
 

core/collection_pipeline/plugin/PluginRegistry.cpp

@@ -56,6 +56,7 @@
 #include "plugin/flusher/kafka/FlusherKafka.h"
 #endif
 #if defined(__linux__) && !defined(__ANDROID__)
+#include "plugin/input/InputCpuProfiling.h"
 #include "plugin/input/InputFileSecurity.h"
 #include "plugin/input/InputHostMeta.h"
 #include "plugin/input/InputHostMonitor.h"
@@ -78,6 +79,7 @@ DEFINE_FLAG_BOOL(enable_ebpf_network_observer, "", true);
 DEFINE_FLAG_BOOL(enable_ebpf_process_secure, "", true);
 DEFINE_FLAG_BOOL(enable_ebpf_file_secure, "", true);
 DEFINE_FLAG_BOOL(enable_ebpf_network_secure, "", false);
+DEFINE_FLAG_BOOL(enable_ebpf_cpu_profiling, "", false);
 
 using namespace std;
 
@@ -174,6 +176,9 @@ void PluginRegistry::LoadStaticPlugins() {
     if (BOOL_FLAG(enable_ebpf_network_secure)) {
         RegisterContinuousInputCreator(new StaticInputCreator<InputNetworkSecurity>(), true);
     }
+    if (BOOL_FLAG(enable_ebpf_cpu_profiling)) {
+        RegisterContinuousInputCreator(new StaticInputCreator<InputCpuProfiling>(), false);
+    }
     RegisterContinuousInputCreator(new StaticInputCreator<InputHostMeta>(), true);
     RegisterContinuousInputCreator(new StaticInputCreator<InputHostMonitor>(), true);
     RegisterContinuousInputCreator(new StaticInputCreator<InputForward>());

core/container_manager/ContainerManager.cpp

@@ -103,12 +103,21 @@ void ContainerManager::ApplyContainerDiffs() {
     auto nameConfigMap = FileServer::GetInstance()->GetAllFileDiscoveryConfigs();
     std::vector<std::shared_ptr<MatchedContainerInfo>> configResults;
     for (auto& pair : mConfigContainerDiffMap) {
+        FileDiscoveryOptions* options = nullptr;
+        const CollectionPipelineContext* ctx = nullptr;
         const auto& itr = nameConfigMap.find(pair.first);
         if (itr == nameConfigMap.end()) {
-            continue;
+            std::lock_guard<std::mutex> lock(mContainerHandlersMutex);
+            const auto& handlerItr = mContainerHandlers.find(pair.first);
+            if (handlerItr == mContainerHandlers.end()) {
+                continue;
+            }
+            options = handlerItr->second.first.first;
+            ctx = handlerItr->second.first.second;
+        } else {
+            options = itr->second.first;
+            ctx = itr->second.second;
         }
-        const auto& options = itr->second.first;
-        const auto& ctx = itr->second.second;
 
         const auto& diff = pair.second;
 
@@ -206,6 +215,20 @@ bool ContainerManager::CheckContainerDiffForAllConfig() {
             }
         }
     }
+    {
+        std::lock_guard<std::mutex> lock(mContainerHandlersMutex);
+        for (auto itr = mContainerHandlers.begin(); itr != mContainerHandlers.end(); ++itr) {
+            FileDiscoveryOptions* options = itr->second.first.first;
+            if (options->IsContainerDiscoveryEnabled()) {
+                bool isCurrentConfigUpdate = checkContainerDiffForOneConfig(options, itr->second.first.second);
+                if (isCurrentConfigUpdate) {
+                    isUpdate = true;
+                    // Invoke callback
+                    itr->second.second(mConfigContainerDiffMap[itr->first]);
+                }
+            }
+        }
+    }
     return isUpdate;
 }
 
@@ -270,6 +293,17 @@ void ContainerManager::sendMatchedContainerInfo(std::vector<std::shared_ptr<Matc
     }
 }
 
+void ContainerManager::AddContainerHandler(const std::string& name,
+                                           const FileDiscoveryConfig& config,
+                                           const std::function<void(std::shared_ptr<ContainerDiff>)>& handler) {
+    std::lock_guard<std::mutex> lock(mContainerHandlersMutex);
+    mContainerHandlers[name] = std::make_pair(config, handler);
+}
+
+void ContainerManager::RemoveContainerHandler(const std::string& name) {
+    std::lock_guard<std::mutex> lock(mContainerHandlersMutex);
+    mContainerHandlers.erase(name);
+}
 
 bool ContainerManager::checkContainerDiffForOneConfig(FileDiscoveryOptions* options,
                                                       const CollectionPipelineContext* ctx) {
@@ -1033,6 +1067,18 @@ void ContainerManager::loadContainerInfoFromContainersFormat(const Json::Value&
                 checkContainerDiffForOneConfig(options, itr->second.second);
             }
         }
+        {
+            std::lock_guard<std::mutex> lock(mContainerHandlersMutex);
+            for (auto iter = mContainerHandlers.begin(); iter != mContainerHandlers.end(); ++iter) {
+                FileDiscoveryOptions* options = iter->second.first.first;
+                if (options->IsContainerDiscoveryEnabled()) {
+                    if (checkContainerDiffForOneConfig(options, nullptr)) {
+                        // Invoke callback
+                        iter->second.second(mConfigContainerDiffMap[iter->first]);
+                    }
+                }
+            }
+        }
         LOG_INFO(sLogger, ("load container state from docker_path_config.json (v1.0.0)", configPath));
     }
 }

core/container_manager/ContainerManager.h

@@ -61,6 +61,11 @@ class ContainerManager {
     void UpdateMatchedContainerInfoPipeline(CollectionPipelineContext* ctx, size_t inputIndex);
     void RemoveMatchedContainerInfoPipeline();
 
+    void AddContainerHandler(const std::string& name,
+                             const FileDiscoveryConfig& config,
+                             const std::function<void(std::shared_ptr<ContainerDiff>)>& handler);
+    void RemoveContainerHandler(const std::string& name);
+
 private:
     void pollingLoop();
     void refreshAllContainersSnapshot();
@@ -101,6 +106,10 @@ class ContainerManager {
     mutable ReadWriteLock mMatchedContainerInfoPipelineMux;
     CollectionPipelineContext* mMatchedContainerInfoPipelineCtx = nullptr;
     size_t mMatchedContainerInfoInputIndex = 0;
+
+    std::mutex mContainerHandlersMutex;
+    using ContainerHandler = std::pair<FileDiscoveryConfig, std::function<void(std::shared_ptr<ContainerDiff>)>>;
+    std::unordered_map<std::string, ContainerHandler> mContainerHandlers;
 };
 
 } // namespace logtail

core/ebpf/Config.cpp

@@ -17,6 +17,7 @@
 #include <string>
 #include <unordered_set>
 
+#include "app_config/AppConfig.h"
 #include "common/Flags.h"
 #include "common/ParamExtractor.h"
 #include "logger/Logger.h"
@@ -555,6 +556,64 @@ bool SecurityOptions::Init(SecurityProbeType probeType,
     return true;
 }
 
+bool CpuProfilingOption::Init(const Json::Value& config,
+                              const CollectionPipelineContext* mContext,
+                              const std::string& sName) {
+    std::string errorMsg;
+
+    if (!GetOptionalUIntParam(config, "CollectIntervalMs", mCollectIntervalMs, errorMsg)) {
+        PARAM_WARNING_IGNORE(mContext->GetLogger(),
+                             mContext->GetAlarm(),
+                             errorMsg,
+                             sName,
+                             mContext->GetConfigName(),
+                             mContext->GetProjectName(),
+                             mContext->GetLogstoreName(),
+                             mContext->GetRegion());
+    }
+
+    if (!GetOptionalStringParam(config, "AppName", mAppName, errorMsg)) {
+        PARAM_WARNING_IGNORE(mContext->GetLogger(),
+                             mContext->GetAlarm(),
+                             errorMsg,
+                             sName,
+                             mContext->GetConfigName(),
+                             mContext->GetProjectName(),
+                             mContext->GetLogstoreName(),
+                             mContext->GetRegion());
+    }
+
+    if (!GetOptionalStringParam(config, "Language", mLanguage, errorMsg)) {
+        PARAM_WARNING_IGNORE(mContext->GetLogger(),
+                             mContext->GetAlarm(),
+                             errorMsg,
+                             sName,
+                             mContext->GetConfigName(),
+                             mContext->GetProjectName(),
+                             mContext->GetLogstoreName(),
+                             mContext->GetRegion());
+    }
+
+    if (!GetOptionalListFilterParam<std::string>(config, "CommandLines", mCmdlines, errorMsg)) {
+        PARAM_WARNING_IGNORE(mContext->GetLogger(),
+                             mContext->GetAlarm(),
+                             errorMsg,
+                             sName,
+                             mContext->GetConfigName(),
+                             mContext->GetProjectName(),
+                             mContext->GetLogstoreName(),
+                             mContext->GetRegion());
+    }
+
+    if (AppConfig::GetInstance()->IsPurageContainerMode()) {
+        if (!mContainerDiscovery.Init(config, *mContext, sName)) {
+            return false;
+        }
+    }
+
+    return true;
+}
+
 //////
 void eBPFAdminConfig::LoadEbpfConfig(const Json::Value& confJson) {
     // receive_event_chan_cap (Optional)

core/ebpf/Config.h

@@ -21,6 +21,7 @@
 #include "json/json.h"
 
 #include "collection_pipeline/CollectionPipelineContext.h"
+#include "container_manager/ContainerDiscoveryOptions.h"
 #include "ebpf/include/export.h"
 
 namespace logtail::ebpf {
@@ -48,6 +49,23 @@ class SecurityOptions {
     SecurityProbeType mProbeType;
 };
 
+/////////////////////  /////////////////////
+
+class CpuProfilingOption {
+public:
+    bool Init(const Json::Value& config, const CollectionPipelineContext* mContext, const std::string& sName);
+
+    uint32_t mCollectIntervalMs = 0; // 0 means use default
+    std::vector<std::string> mCmdlines;
+    std::string mAppName;
+    // Since ebpf may collect mulit-language profiling data,
+    // the language field should be set by user.
+    std::string mLanguage = "java";
+    ContainerDiscoveryOptions mContainerDiscovery;
+};
+
+using PluginOptions = std::variant<SecurityOptions*, ObserverNetworkOption*, CpuProfilingOption*>;
+
 ///////////////////// Process Level Config /////////////////////
 
 struct AdminConfig {

core/ebpf/EBPFServer.cpp

@@ -32,6 +32,7 @@
 #include "ebpf/plugin/AbstractManager.h"
 #include "logger/Logger.h"
 #include "monitor/metric_models/ReentrantMetricsRecord.h"
+#include "plugin/cpu_profiling/CpuProfilingManager.h"
 #include "plugin/file_security/FileSecurityManager.h"
 #include "plugin/network_observer/NetworkObserverManager.h"
 #include "plugin/network_security/NetworkSecurityManager.h"
@@ -59,7 +60,8 @@ bool EnvManager::IsSupportedEnv(PluginType type) {
             break;
         case PluginType::FILE_SECURITY:
         case PluginType::NETWORK_SECURITY:
-        case PluginType::PROCESS_SECURITY: {
+        case PluginType::PROCESS_SECURITY:
+        case PluginType::CPU_PROFILING: {
             status = mArchSupport && mBTFSupport;
             break;
         }
@@ -282,10 +284,10 @@ bool EBPFServer::startPluginInternal(const std::string& pipelineName,
                                      uint32_t pluginIndex,
                                      PluginType type,
                                      const logtail::CollectionPipelineContext* ctx,
-                                     const std::variant<SecurityOptions*, ObserverNetworkOption*>& options,
+                                     const PluginOptions& options,
                                      const PluginMetricManagerPtr& metricManager) {
     bool isNeedProcessCache = false;
-    if (type != PluginType::NETWORK_OBSERVE) {
+    if (type != PluginType::NETWORK_OBSERVE && type != PluginType::CPU_PROFILING) {
         isNeedProcessCache = true;
     }
     auto& pluginMgr = getPluginState(type).mManager;
@@ -346,6 +348,17 @@ bool EBPFServer::startPluginInternal(const std::string& pipelineName,
                 }
                 break;
             }
+
+            case PluginType::CPU_PROFILING: {
+                if (!pluginMgr) {
+                    auto mgr = CpuProfilingManager::Create(
+                        mProcessCacheManager, mEBPFAdapter, mCommonEventQueue, &mEventPool);
+                    mgr->SetMetrics(mRecvKernelEventsTotal);
+                    pluginMgr = mgr;
+                }
+                break;
+            }
+
             default:
                 LOG_ERROR(sLogger, ("Unknown plugin type", int(type)));
                 return false;
@@ -371,7 +384,8 @@ bool EBPFServer::startPluginInternal(const std::string& pipelineName,
     }
 
     updatePluginState(type, pipelineName, ctx->GetProjectName(), PluginStateOperation::kAddPipeline, pluginMgr);
-    if (type != PluginType::PROCESS_SECURITY && type != PluginType::NETWORK_OBSERVE) {
+    if (type != PluginType::PROCESS_SECURITY && type != PluginType::NETWORK_OBSERVE
+        && type != PluginType::CPU_PROFILING) {
         RegisterPluginPerfBuffers(type);
     }
 
@@ -391,7 +405,7 @@ bool EBPFServer::EnablePlugin(const std::string& pipelineName,
                               uint32_t pluginIndex,
                               PluginType type,
                               const CollectionPipelineContext* ctx,
-                              const std::variant<SecurityOptions*, ObserverNetworkOption*>& options,
+                              const PluginOptions& options,
                               const PluginMetricManagerPtr& mgr) {
     if (!IsSupportedEnv(type)) {
         return false;
@@ -550,14 +564,17 @@ void EBPFServer::pollPerfBuffers() {
         mProcessCacheManager->ClearProcessExpiredCache();
 
         // TODO (@qianlu.kk) adapt to ConsumePerfBufferData
-        auto& pluginState = getPluginState(PluginType::NETWORK_OBSERVE);
-        if (!pluginState.mValid.load(std::memory_order_acquire)) {
-            continue;
-        }
-        std::shared_lock<std::shared_mutex> lock(pluginState.mMtx);
-        if (pluginState.mManager) {
-            auto* mgr = static_cast<NetworkObserverManager*>(pluginState.mManager.get());
-            mgr->PollPerfBuffer(0); // 0 means non-blocking r(ef: https://libbpf.readthedocs.io/en/latest/api.html)
+        std::vector<PluginState*> pluginStatePtrs
+            = {&getPluginState(PluginType::NETWORK_OBSERVE), &getPluginState(PluginType::CPU_PROFILING)};
+        for (auto& pluginStatePtr : pluginStatePtrs) {
+            auto& pluginState = *pluginStatePtr;
+            if (!pluginState.mValid.load(std::memory_order_acquire)) {
+                continue;
+            }
+            std::shared_lock<std::shared_mutex> lock(pluginState.mMtx);
+            if (pluginState.mManager) {
+                pluginState.mManager->PollPerfBuffer(0); // 0 means non-blocking
+            }
         }
     }
 }

core/ebpf/EBPFServer.h

@@ -92,7 +92,7 @@ class EBPFServer : public InputRunner {
                       uint32_t pluginIndex,
                       PluginType type,
                       const logtail::CollectionPipelineContext* ctx,
-                      const std::variant<SecurityOptions*, ObserverNetworkOption*>& options,
+                      const PluginOptions& options,
                       const PluginMetricManagerPtr& mgr);
 
     bool DisablePlugin(const std::string& pipelineName, PluginType type);
@@ -118,7 +118,7 @@ class EBPFServer : public InputRunner {
                              uint32_t pluginIndex,
                              PluginType type,
                              const logtail::CollectionPipelineContext* ctx,
-                             const std::variant<SecurityOptions*, ObserverNetworkOption*>& options,
+                             const PluginOptions& options,
                              const PluginMetricManagerPtr& metricManager);
     EBPFServer();