feat: add ebpf plugins

.devcontainer/devcontainer.json

@@ -6,6 +6,11 @@
     }
   },
   "initializeCommand": ".devcontainer/gen_env.sh",
+  "privileged": true,
+  "mounts": [
+    { "source": "/sys", "target": "/sys", "type": "bind" },
+    { "source": "/", "target": "/logtail_host", "type": "bind" }
+  ],
   "runArgs": [ "--cap-add=SYS_PTRACE", "--security-opt", "seccomp=unconfined" ],
   "onCreateCommand": "sudo chown -R $(id -un):$(id -gn) /root",
   "customizations": {

.github/workflows/build-core-ut.yaml

@@ -83,7 +83,7 @@ jobs:
         run: docker build -t unittest_coverage -f ./docker/Dockerfile_coverage .
 
       - name: Unit Test
-        run: docker run -v $(pwd):$(pwd) unittest_coverage bash -c "cd $(pwd) && make unittest_core"
+        run: docker run --privileged -v /sys:/sys -v /:/logtail_host -v $(pwd):$(pwd) unittest_coverage bash -c "cd $(pwd) && make unittest_core"
 
       - name: Unit Test Coverage
         run: docker run -v $(pwd):$(pwd) unittest_coverage bash -c "cd $(pwd)/core && gcovr --gcov-ignore-errors=no_working_dir_found --root . --json coverage.json --json-summary-pretty --json-summary summary.json -e \".*\.pb\.cc\" -e \".*\.pb\.h\" -e \".*unittest.*\" -e \".*sdk.*\" -e \".*logger.*\" -e \".*config_server.*\" -e \".*go_pipeline.*\" -e \".*application.*\" -e \".*runner.*\""

.github/workflows/static-check.yaml

@@ -88,7 +88,7 @@ jobs:
         with:
           clang-format-version: '18'
           check-path: 'core'
-          exclude-regex: 'common/xxhash|labels/Relabel\.cpp|ProcessorParseContainerLogNative\.cpp|FlusherSLS\.cpp'
+          exclude-regex: 'common/xxhash|labels/Relabel\.cpp|ProcessorParseContainerLogNative\.cpp|FlusherSLS\.cpp|ebpf/driver/coolbpf'
           include-regex: '.*\.(cpp|h)$'
 
       - name: Go Plugin Lint

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "core/ebpf/driver/coolbpf"]
+	path = core/ebpf/driver/coolbpf
+	url = https://gitee.com/anolis/coolbpf.git

core/CMakeLists.txt

@@ -92,7 +92,6 @@ endif ()
 # To be compatible with low version Linux.
 if (ENABLE_COMPATIBLE_MODE)
     message(STATUS "Enable compatible mode.")
-    set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -std=c90")
     set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wl,--wrap=memcpy")
     add_definitions(-DENABLE_COMPATIBLE_MODE)
 endif ()
@@ -127,7 +126,7 @@ set(SUB_DIRECTORIES_LIST
         protobuf/sls protobuf/models
         file_server file_server/event file_server/event_handler file_server/event_listener file_server/reader file_server/polling
         prometheus prometheus/labels prometheus/schedulers prometheus/async prometheus/component
-        ebpf ebpf/observer ebpf/security ebpf/handler
+        ebpf ebpf/util ebpf/util/sampler ebpf/protocol/http ebpf/protocol ebpf/plugin/file_security ebpf/plugin/network_observer ebpf/plugin/process_security ebpf/plugin/network_security ebpf/plugin ebpf/observer ebpf/security ebpf/handler 
         parser
         )
 if (LINUX)
@@ -145,6 +144,7 @@ endif()
 
 # Module includes & set files.
 include_directories(${CMAKE_CURRENT_SOURCE_DIR})
+include_directories("${DEPS_INCLUDE_ROOT}/coolbpf")
 
 foreach (DIR_NAME ${SUB_DIRECTORIES_LIST})
     include_directories(${CMAKE_CURRENT_SOURCE_DIR}/${DIR_NAME})
@@ -221,6 +221,8 @@ endif ()
 # Generate independent libraries.
 add_subdirectory(go_pipeline)
 add_subdirectory(common)
+# Build eBPF dependencies
+add_subdirectory(ebpf/driver)
 
 # Link libraries.
 if(BUILD_LOGTAIL OR BUILD_LOGTAIL_SHARED_LIBRARY)
@@ -230,6 +232,7 @@ if(BUILD_LOGTAIL OR BUILD_LOGTAIL_SHARED_LIBRARY)
     all_link(${LOGTAIL_TARGET})
     common_link(${LOGTAIL_TARGET})
     target_link_libraries(${LOGTAIL_TARGET} provider)
+    add_dependencies(${LOGTAIL_TARGET} install_coolbpf)
 endif()
 
 # Logtail UT.

core/app_config/AppConfig.cpp

@@ -166,6 +166,12 @@ DEFINE_FLAG_STRING(metrics_report_method,
 DEFINE_FLAG_STRING(operator_service, "loong collector operator service", "");
 DEFINE_FLAG_INT32(operator_service_port, "loong collector operator service port", 8888);
 DEFINE_FLAG_INT32(k8s_meta_service_port, "loong collector operator service port", 9000);
+
+DEFINE_FLAG_STRING(loong_collector_singleton_service, "loong collector singleton service", "loongcollector-singleton");
+DEFINE_FLAG_INT32(loong_collector_singleton_port, "loong collector singleton service port", 8899);
+DEFINE_FLAG_STRING(loong_collector_operator_service, "loong collector operator service", "");
+DEFINE_FLAG_INT32(loong_collector_operator_service_port, "loong collector operator service port", 8888);
+DEFINE_FLAG_INT32(loong_collector_k8s_meta_service_port, "loong collector operator service port", 9000);
 DEFINE_FLAG_STRING(_pod_name_, "agent pod name", "");
 
 DEFINE_FLAG_STRING(app_info_file, "", "app_info.json");

core/collection_pipeline/serializer/SLSSerializer.cpp

@@ -116,13 +116,30 @@ bool SLSEventGroupSerializer::Serialize(BatchedEvents&& group, string& res, stri
         case PipelineEvent::Type::METRIC: {
             for (size_t i = 0; i < group.mEvents.size(); ++i) {
                 const auto& e = group.mEvents[i].Cast<MetricEvent>();
+                for (auto tag = e.TagsBegin(); tag != e.TagsEnd(); tag++) {
+                    LOG_DEBUG(sLogger, 
+                        ("event tags for metricname", std::string(e.GetName()))
+                        (std::string(tag->first), std::string(tag->second))
+                    );
+                }
+                for (auto tag = group.mTags.mInner.begin(); tag != group.mTags.mInner.end(); tag++) {
+                    LOG_DEBUG(sLogger, 
+                        ("group tags for metricname", std::string(e.GetName()))
+                        (std::string(tag->first), std::string(tag->second))
+                    );
+                }
+
+
                 if (e.Is<UntypedSingleValue>()) {
                     metricEventContentCache[i].first = to_string(e.GetValue<UntypedSingleValue>()->mValue);
+                    // should not happen
+                    LOG_ERROR(sLogger,
+                              ("config", mFlusher->GetContext().GetConfigName()) ("metricname", e.GetName()));
                 } else {
                     // should not happen
                     LOG_ERROR(sLogger,
                               ("unexpected error",
-                               "invalid metric event type")("config", mFlusher->GetContext().GetConfigName()));
+                               "invalid metric event type")("config", mFlusher->GetContext().GetConfigName()) ("metricname", e.GetName()));
                     continue;
                 }
                 metricEventContentCache[i].second = GetMetricLabelSize(e);
@@ -141,6 +158,18 @@ bool SLSEventGroupSerializer::Serialize(BatchedEvents&& group, string& res, stri
         case PipelineEvent::Type::SPAN:
             for (size_t i = 0; i < group.mEvents.size(); ++i) {
                 const auto& e = group.mEvents[i].Cast<SpanEvent>();
+                for (auto tag = e.TagsBegin(); tag != e.TagsEnd(); tag++) {
+                    LOG_DEBUG(sLogger, 
+                        ("event tags for spanname", std::string(e.GetName()))
+                        (std::string(tag->first), std::string(tag->second))
+                    );
+                }
+                for (auto tag = group.mTags.mInner.begin(); tag != group.mTags.mInner.end(); tag++) {
+                    LOG_DEBUG(sLogger, 
+                        ("group tags for spanname", std::string(e.GetName()))
+                        (std::string(tag->first), std::string(tag->second))
+                    );
+                }
                 size_t contentSZ = 0;
                 contentSZ += GetLogContentSize(DEFAULT_TRACE_TAG_TRACE_ID.size(), e.GetTraceId().size());
                 contentSZ += GetLogContentSize(DEFAULT_TRACE_TAG_SPAN_ID.size(), e.GetSpanId().size());

core/common/CapabilityUtil.cpp

@@ -0,0 +1,109 @@
+// Copyright 2023 iLogtail Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "CapabilityUtil.h"
+#include <unordered_map>
+#include <vector>
+#include <stdexcept>
+#include <iostream>
+#include <sstream>
+
+namespace logtail {
+
+std::unordered_map<uint64_t, std::string> capabilitiesString = {
+  {0, "CAP_CHOWN"},
+  {1, "DAC_OVERRIDE"},
+  {2, "CAP_DAC_READ_SEARCH"},
+  {3, "CAP_FOWNER"},
+  {4, "CAP_FSETID"},
+  {5, "CAP_KILL"},
+  {6, "CAP_SETGID"},
+  {7, "CAP_SETUID"},
+  {8, "CAP_SETPCAP"},
+  {9, "CAP_LINUX_IMMUTABLE"},
+  {10, "CAP_NET_BIND_SERVICE"},
+  {11, "CAP_NET_BROADCAST"},
+  {12, "CAP_NET_ADMIN"},
+  {13, "CAP_NET_RAW"},
+  {14, "CAP_IPC_LOCK"},
+  {15, "CAP_IPC_OWNER"},
+  {16, "CAP_SYS_MODULE"},
+  {17, "CAP_SYS_RAWIO"},
+  {18, "CAP_SYS_CHROOT"},
+  {19, "CAP_SYS_PTRACE"},
+  {20, "CAP_SYS_PACCT"},
+  {21, "CAP_SYS_ADMIN"},
+  {22, "CAP_SYS_BOOT"},
+  {23, "CAP_SYS_NICE"},
+  {24, "CAP_SYS_RESOURCE"},
+  {25, "CAP_SYS_TIME"},
+  {26, "CAP_SYS_TTY_CONFIG"},
+  {27, "CAP_MKNOD"},
+  {28, "CAP_LEASE"},
+  {29, "CAP_AUDIT_WRITE"},
+  {30, "CAP_AUDIT_CONTROL"},
+  {31, "CAP_SETFCAP"},
+  {32, "CAP_MAC_OVERRIDE"},
+  {33, "CAP_MAC_ADMIN"},
+  {34, "CAP_SYSLOG"},
+  {35, "CAP_WAKE_ALARM"},
+  {36, "CAP_BLOCK_SUSPEND"},
+  {37, "CAP_AUDIT_READ"},
+  {38, "CAP_PERFMON"},
+  {39, "CAP_BPF"},
+  {40, "CAP_CHECKPOINT_RESTORE"}
+};
+
+const int32_t CAP_LAST_CAP = 40;
+
+bool IsCapValid(int32_t capInt) {
+  return (capInt >= 0 && capInt <= CAP_LAST_CAP);
+}
+
+std::string GetCapability(int32_t capInt) {
+  if (!IsCapValid(capInt)) {
+    throw std::invalid_argument("invalid capability value " + std::to_string(capInt));
+  }
+
+  auto it = capabilitiesString.find(static_cast<uint64_t>(capInt));
+  if (it == capabilitiesString.end()) {
+    throw std::invalid_argument("could not map capability value " + std::to_string(capInt));
+  }
+
+  return it->second;
+}
+
+std::string GetCapabilities(uint64_t capInt) {
+  std::vector<std::string> caps;
+
+  for (uint64_t i = 0; i < 64; ++i) {
+    if ((1ULL << i) & capInt) {
+      auto it = capabilitiesString.find(i);
+      if (it != capabilitiesString.end()) {
+        caps.push_back(it->second);
+      }
+    }
+  }
+
+  std::ostringstream oss;
+  for (size_t i = 0; i < caps.size(); ++i) {
+    if (i > 0) {
+      oss << " ";
+    }
+    oss << caps[i];
+  }
+  return oss.str();
+}
+
+}

core/common/CapabilityUtil.h

@@ -0,0 +1,25 @@
+// Copyright 2023 iLogtail Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <string>
+
+namespace logtail {
+
+std::string GetCapabilities(uint64_t capInt);
+
+std::string GetCapability(int32_t capInt);
+
+}

core/common/LRUCache.h

@@ -104,13 +104,9 @@ class Cache {
      * directly anyway! :)
      */
     explicit Cache(size_t maxSize = 64, size_t elasticity = 10)
-        : maxSize_(maxSize), elasticity_(elasticity), prune_thread(&Cache::pruneThreadFunc, this) {}
+        : maxSize_(maxSize), elasticity_(elasticity) {}
 
     virtual ~Cache() {
-        stop_pruning = true;
-        if (prune_thread.joinable()) {
-            prune_thread.join();
-        }
     }
 
     size_t size() const {
@@ -252,13 +248,6 @@ class Cache {
         return count;
     }
 
-    void pruneThreadFunc() {
-        while (!stop_pruning) {
-            pruneExpired();
-            std::this_thread::sleep_for(std::chrono::seconds(60)); // 每60秒检查一次
-        }
-    }
-
 private:
     // Disallow copying.
     Cache(const Cache&) = delete;
@@ -269,8 +258,6 @@ class Cache {
     list_type keys_;
     size_t maxSize_;
     size_t elasticity_;
-    std::atomic<bool> stop_pruning{false};
-    std::thread prune_thread;
 
 #ifdef APSARA_UNIT_TEST_MAIN
     friend class LRUCacheUnittest;