feature: add net collector

Author: WRPStephanie

core/host_monitor/Constants.cpp

@@ -23,6 +23,9 @@ namespace logtail {
 std::filesystem::path PROCESS_DIR = "/proc";
 const std::filesystem::path PROCESS_STAT = "stat";
 const std::filesystem::path PROCESS_LOADAVG = "loadavg";
+const std::filesystem::path PROCESS_NET_SOCKSTAT = "net/sockstat";
+const std::filesystem::path PROCESS_NET_SOCKSTAT6 = "net/sockstat6";
+const std::filesystem::path PROCESS_NET_DEV = "net/dev";
 const int64_t SYSTEM_HERTZ = sysconf(_SC_CLK_TCK);
 
 } // namespace logtail

core/host_monitor/Constants.h

@@ -25,6 +25,10 @@ namespace logtail {
 extern std::filesystem::path PROCESS_DIR;
 const extern std::filesystem::path PROCESS_STAT;
 const extern std::filesystem::path PROCESS_LOADAVG;
+
+const extern std::filesystem::path PROCESS_NET_SOCKSTAT;
+const extern std::filesystem::path PROCESS_NET_SOCKSTAT6;
+const extern std::filesystem::path PROCESS_NET_DEV;
 const extern int64_t SYSTEM_HERTZ;
 
 #ifdef __ENTERPRISE__

core/host_monitor/HostMonitorInputRunner.cpp

@@ -35,6 +35,7 @@
 #include "host_monitor/Constants.h"
 #include "host_monitor/HostMonitorTimerEvent.h"
 #include "host_monitor/collector/CPUCollector.h"
+#include "host_monitor/collector/NetCollector.h"
 #include "host_monitor/collector/ProcessEntityCollector.h"
 #include "host_monitor/collector/SystemCollector.h"
 #include "logger/Logger.h"
@@ -55,6 +56,7 @@ HostMonitorInputRunner::HostMonitorInputRunner() {
     RegisterCollector<ProcessEntityCollector>();
     RegisterCollector<CPUCollector>();
     RegisterCollector<SystemCollector>();
+    RegisterCollector<NetCollector>();
 
     size_t threadPoolSize = 1;
     // threadPoolSize should be greater than 0

core/host_monitor/LinuxSystemInterface.cpp

@@ -17,6 +17,7 @@
 #include "host_monitor/LinuxSystemInterface.h"
 
 #include <chrono>
+#include <string>
 
 using namespace std;
 using namespace std::chrono;
@@ -71,6 +72,195 @@ bool GetHostLoadavg(vector<string>& lines, string& errorMessage) {
     }
     return true;
 }
+bool ReadSocketStat(const std::filesystem::path& path, uint64_t& tcp) {
+    tcp = 0;
+    if (!path.empty()) {
+        std::vector<std::string> sockstatLines;
+        std::string errorMessage;
+        if (!CheckExistance(path)) {
+            errorMessage = "file does not exist: " + (path).string();
+            return false;
+        }
+
+        if (GetFileLines(path, sockstatLines, true, &errorMessage) != 0 || sockstatLines.empty()) {
+            return false;
+        }
+
+        for (auto const& line : sockstatLines) {
+            if (line.size() >= 5 && (line.substr(0, 4) == "TCP:" || line.substr(0, 5) == "TCP6:")) {
+                std::vector<std::string> metrics;
+                boost::split(metrics, line, boost::is_any_of(" "), boost::token_compress_on);
+                if (metrics.size() >= 9) {
+                    tcp += static_cast<uint64_t>(std::stoull(metrics[6])); // tw
+                    tcp += static_cast<uint64_t>(std::stoull(metrics[8])); // alloc
+                }
+            }
+        }
+    }
+    return true;
+}
+
+bool ReadNetLink(std::vector<uint64_t>& tcpStateCount) {
+    static std::atomic_int sequence_number = 1;
+    int fd;
+    // struct inet_diag_msg *r;
+    // 使用netlink socket与内核通信
+    fd = socket(AF_NETLINK, SOCK_RAW, NETLINK_INET_DIAG);
+    if (fd < 0) {
+        LOG_WARNING(sLogger,
+                    ("ReadNetLink, socket(AF_NETLINK, SOCK_RAW, NETLINK_INET_DIAG) failed, error msg: ",
+                     std::string(strerror(errno))));
+        return false;
+    }
+
+
+    // 存在多个netlink socket时，必须单独bind,并通过nl_pid来区分
+    struct sockaddr_nl nladdr_bind {};
+    memset(&nladdr_bind, 0, sizeof(nladdr_bind));
+    nladdr_bind.nl_family = AF_NETLINK;
+    nladdr_bind.nl_pad = 0;
+    nladdr_bind.nl_pid = getpid();
+    nladdr_bind.nl_groups = 0;
+    if (bind(fd, (struct sockaddr*)&nladdr_bind, sizeof(nladdr_bind))) {
+        LOG_WARNING(sLogger, ("ReadNetLink, bind netlink socket failed, error msg: ", std::string(strerror(errno))));
+        close(fd);
+        return false;
+    }
+    struct sockaddr_nl nladdr {};
+    memset(&nladdr, 0, sizeof(nladdr));
+    nladdr.nl_family = AF_NETLINK;
+    struct NetLinkRequest req {};
+    memset(&req, 0, sizeof(req));
+    req.nlh.nlmsg_len = sizeof(req);
+    req.nlh.nlmsg_type = TCPDIAG_GETSOCK;
+    req.nlh.nlmsg_flags = NLM_F_ROOT | NLM_F_MATCH | NLM_F_REQUEST;
+    // sendto kernel
+    req.nlh.nlmsg_pid = getpid();
+    req.nlh.nlmsg_seq = ++sequence_number;
+    req.r.idiag_family = AF_INET;
+    req.r.idiag_states = 0xfff;
+    req.r.idiag_ext = 0;
+    struct iovec iov {};
+    memset(&iov, 0, sizeof(iov));
+    iov.iov_base = &req;
+    iov.iov_len = sizeof(req);
+    struct msghdr msg {};
+    memset(&msg, 0, sizeof(msg));
+    msg.msg_name = (void*)&nladdr;
+    msg.msg_namelen = sizeof(nladdr);
+    msg.msg_iov = &iov;
+    msg.msg_iovlen = 1;
+
+    if (sendmsg(fd, &msg, 0) < 0) {
+        LOG_WARNING(sLogger, ("ReadNetLink, sendmsg(2) failed, error msg: ", std::string(strerror(errno))));
+        close(fd);
+        return false;
+    }
+    char buf[8192];
+    iov.iov_base = buf;
+    iov.iov_len = sizeof(buf);
+
+    uint64_t received_count = 0;
+    uint64_t MAX_RECV_COUNT = std::numeric_limits<uint64_t>::max();
+    while (received_count < MAX_RECV_COUNT) {
+        received_count++;
+        // struct nlmsghdr *h;
+        memset(&msg, 0, sizeof(msg));
+        msg.msg_name = (void*)&nladdr;
+        msg.msg_namelen = sizeof(nladdr);
+        msg.msg_iov = &iov;
+        msg.msg_iovlen = 1;
+        ssize_t status = recvmsg(fd, (struct msghdr*)&msg, 0);
+        if (status < 0) {
+            if (errno == EINTR || errno == EAGAIN) {
+                continue;
+            }
+            LOG_WARNING(sLogger, ("ReadNetLink, recvmsg(2) failed, error msg: ", std::string(strerror(errno))));
+            close(fd);
+            return false;
+        } else if (status == 0) {
+            LOG_WARNING(sLogger,
+                        ("ReadNetLink, Unexpected zero-sized  reply from netlink socket. error msg: ",
+                         std::string(strerror(errno))));
+            close(fd);
+            return true;
+        }
+
+        // h = (struct nlmsghdr *) buf;
+        for (auto h = (struct nlmsghdr*)buf; NLMSG_OK(h, status); h = NLMSG_NEXT(h, status)) {
+            if (static_cast<uint64_t>(h->nlmsg_seq) != static_cast<uint64_t>(sequence_number)) {
+                // sequence_number is not equal
+                // h = NLMSG_NEXT(h, status);
+                continue;
+            }
+
+            if (h->nlmsg_type == NLMSG_DONE) {
+                close(fd);
+                return true;
+            } else if (h->nlmsg_type == NLMSG_ERROR) {
+                if (h->nlmsg_len < NLMSG_LENGTH(sizeof(struct nlmsgerr))) {
+                    LOG_WARNING(sLogger, ("ReadNetLink ", "message truncated"));
+                } else {
+                    auto msg_error = (struct nlmsgerr*)NLMSG_DATA(h);
+                    LOG_WARNING(sLogger, ("ReadNetLink, Received error, error msg: ", msg_error));
+                }
+                close(fd);
+                return false;
+            }
+            auto r = (struct inet_diag_msg*)NLMSG_DATA(h);
+            /*This code does not(need to) distinguish between IPv4 and IPv6.*/
+            if (r->idiag_state > TCP_CLOSING || r->idiag_state < TCP_ESTABLISHED) {
+                // Ignoring connection with unknown state
+                continue;
+            }
+            tcpStateCount[r->idiag_state]++;
+            // h = NLMSG_NEXT(h, status);
+        }
+    }
+    close(fd);
+    return true;
+}
+
+bool GetNetStateByNetLink(NetState& netState) {
+    std::vector<uint64_t> tcpStateCount(TCP_CLOSING + 1, 0);
+    if (ReadNetLink(tcpStateCount) == false) {
+        return false;
+    }
+    uint64_t tcp = 0, tcpSocketStat = 0;
+
+    if (ReadSocketStat(PROCESS_DIR / PROCESS_NET_SOCKSTAT, tcp)) {
+        tcpSocketStat += tcp;
+    }
+    if (ReadSocketStat(PROCESS_DIR / PROCESS_NET_SOCKSTAT6, tcp)) {
+        tcpSocketStat += tcp;
+    }
+
+    int total = 0;
+    for (int i = TCP_ESTABLISHED; i <= TCP_CLOSING; i++) {
+        if (i == TCP_SYN_SENT || i == TCP_SYN_RECV) {
+            total += tcpStateCount[i];
+        }
+        netState.tcpStates[i] = tcpStateCount[i];
+    }
+    // 设置为-1表示没有采集
+    netState.tcpStates[TCP_TOTAL] = total + tcpSocketStat;
+    netState.tcpStates[TCP_NON_ESTABLISHED] = netState.tcpStates[TCP_TOTAL] - netState.tcpStates[TCP_ESTABLISHED];
+    return true;
+}
+
+bool GetHostNetDev(vector<string>& lines, string& errorMessage) {
+    errorMessage.clear();
+    if (!CheckExistance(PROCESS_DIR / PROCESS_NET_DEV)) {
+        errorMessage = "file does not exist: " + (PROCESS_DIR / PROCESS_NET_DEV).string();
+        return false;
+    }
+
+    int ret = GetFileLines(PROCESS_DIR / PROCESS_NET_DEV, lines, true, &errorMessage);
+    if (ret != 0 || lines.empty()) {
+        return false;
+    }
+    return true;
+}
 
 bool LinuxSystemInterface::GetSystemInformationOnce(SystemInformation& systemInfo) {
     std::vector<std::string> lines;
@@ -221,4 +411,66 @@ bool LinuxSystemInterface::GetCPUCoreNumInformationOnce(CpuCoreNumInformation& c
     return true;
 }
 
+bool LinuxSystemInterface::GetTCPStatInformationOnce(TCPStatInformation& tcpStatInfo) {
+    NetState netState;
+
+    bool ret = false;
+
+    ret = GetNetStateByNetLink(netState);
+
+    if (ret) {
+        tcpStatInfo.stat.tcpEstablished = (netState.tcpStates[TCP_ESTABLISHED]);
+        tcpStatInfo.stat.tcpListen = (netState.tcpStates[TCP_LISTEN]);
+        tcpStatInfo.stat.tcpTotal = (netState.tcpStates[TCP_TOTAL]);
+        tcpStatInfo.stat.tcpNonEstablished = (netState.tcpStates[TCP_NON_ESTABLISHED]);
+    }
+
+    return ret;
+}
+
+bool LinuxSystemInterface::GetNetRateInformationOnce(NetRateInformation& netRateInfo) {
+    //  /proc/net/dev
+    std::vector<std::string> netDevLines = {};
+    std::string errorMessage;
+    bool ret = GetHostNetDev(netDevLines, errorMessage);
+    if (!ret || netDevLines.empty()) {
+        return false;
+    }
+
+    for (size_t i = 2; i < netDevLines.size(); ++i) {
+        auto pos = netDevLines[i].find_first_of(':');
+        std::string devCounterStr = netDevLines[i].substr(pos + 1);
+        std::string devName = netDevLines[i].substr(0, pos);
+        std::vector<std::string> netDevMetric;
+        boost::algorithm::trim(devCounterStr);
+        boost::split(netDevMetric, devCounterStr, boost::is_any_of(" "), boost::token_compress_on);
+
+        if (netDevMetric.size() >= 16) {
+            NetInterfaceMetric information;
+            int index = 0;
+            boost::algorithm::trim(devName);
+            information.name = devName;
+            information.rxBytes = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+            information.rxPackets = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+            information.rxErrors = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+            information.rxDropped = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+            information.rxOverruns = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+            information.rxFrame = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+            // skip compressed multicast
+            index += 2;
+            information.txBytes = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+            information.txPackets = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+            information.txErrors = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+            information.txDropped = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+            information.txOverruns = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+            information.txCollisions = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+            information.txCarrier = boost::lexical_cast<uint64_t>(netDevMetric[index++]);
+
+            information.speed = -1;
+            netRateInfo.metrics.push_back(information);
+        }
+    }
+    return true;
+}
+
 } // namespace logtail

core/host_monitor/LinuxSystemInterface.h

@@ -39,6 +39,8 @@ class LinuxSystemInterface : public SystemInterface {
     bool GetCPUInformationOnce(CPUInformation& cpuInfo) override;
     bool GetProcessListInformationOnce(ProcessListInformation& processListInfo) override;
     bool GetProcessInformationOnce(pid_t pid, ProcessInformation& processInfo) override;
+    bool GetTCPStatInformationOnce(TCPStatInformation& tcpStatInfo) override;
+    bool GetNetRateInformationOnce(NetRateInformation& netRateInfo) override;
 
     bool GetSystemLoadInformationOnce(SystemLoadInformation& systemLoadInfo) override;
     bool GetCPUCoreNumInformationOnce(CpuCoreNumInformation& cpuCoreNumInfo) override;

core/host_monitor/SystemInterface.cpp

@@ -114,6 +114,28 @@ bool SystemInterface::GetCPUCoreNumInformation(CpuCoreNumInformation& cpuCoreNum
         errorType);
 }
 
+bool SystemInterface::GetTCPStatInformation(TCPStatInformation& tcpStatInfo) {
+    const std::string errorType = "TCP stat";
+    return MemoizedCall(
+        mTCPStatInformationCache,
+        [this](BaseInformation& info) {
+            return this->GetTCPStatInformationOnce(static_cast<TCPStatInformation&>(info));
+        },
+        tcpStatInfo,
+        errorType);
+}
+
+bool SystemInterface::GetNetRateInformation(NetRateInformation& netRateInfo) {
+    const std::string errorType = "Net rate";
+    return MemoizedCall(
+        mNetRateInformationCache,
+        [this](BaseInformation& info) {
+            return this->GetNetRateInformationOnce(static_cast<NetRateInformation&>(info));
+        },
+        netRateInfo,
+        errorType);
+}
+
 template <typename F, typename InfoT, typename... Args>
 bool SystemInterface::MemoizedCall(
     SystemInformationCache<InfoT, Args...>& cache, F&& func, InfoT& info, const std::string& errorType, Args... args) {