Merge branch 'main' into fix-incorrect-tcp-rtt

CHANGELOG.md

@@ -8,6 +8,14 @@
 ### <a id="main"></a>DeepFlow release main
 
 #### Bug Fix
+* fix: the aggregation method of unsummable fields of custom data_source is wrong [#8181](https://github.com/deepflowio/deepflow/pull/8181) by [lzf575](https://github.com/lzf575)
+* fix: applicaition log decode failed causes subsequent data anomalies [#8179](https://github.com/deepflowio/deepflow/pull/8179) by [lzf575](https://github.com/lzf575)
+* fix: Bad cstr passed into pcap_compile_nopcap [#8171](https://github.com/deepflowio/deepflow/pull/8171) by [rvql](https://github.com/rvql)
+* fix: memory profile feature compile failed [#8157](https://github.com/deepflowio/deepflow/pull/8157) by [lzf575](https://github.com/lzf575)
+* fix: memory profile data cannot be compressed [#8155](https://github.com/deepflowio/deepflow/pull/8155) by [lzf575](https://github.com/lzf575)
+* fix: dispatcher pipelines is empty [#8146](https://github.com/deepflowio/deepflow/pull/8146) by [yuanchaoa](https://github.com/yuanchaoa)
+* fix: agent - eBPF fix errors caused by java_syms_update_tasks_head [#8145](https://github.com/deepflowio/deepflow/pull/8145) by [yinjiping](https://github.com/yinjiping)
+* fix: querier data visibility filter error [#8139](https://github.com/deepflowio/deepflow/pull/8139) by [xiaochaoren1](https://github.com/xiaochaoren1)
 * fix: incorrect request resource in kafka logs [#8141](https://github.com/deepflowio/deepflow/pull/8141) by [yuanchaoa](https://github.com/yuanchaoa)
 * fix: specify the go mod version when cli verify [#8135](https://github.com/deepflowio/deepflow/pull/8135) by [jiumos](https://github.com/jiumos)
 * fix: Memory profile not outputing data [#8129](https://github.com/deepflowio/deepflow/pull/8129) by [rvql](https://github.com/rvql)
@@ -191,6 +199,9 @@
 * Fix prometheus data cannot be labeled with universal tags，if slow-decoder is used. [#7100](https://github.com/deepflowio/deepflow/pull/7100)
 
 #### NEW FEATURE
+* feat: agent - Retain the profiler_regex interface [#8150](https://github.com/deepflowio/deepflow/pull/8150) by [yinjiping](https://github.com/yinjiping)
+* feat: agent - eBPF Unified Process Matching Interface [#8033](https://github.com/deepflowio/deepflow/pull/8033) by [yinjiping](https://github.com/yinjiping)
+* feat: update genesis data store [#7964](https://github.com/deepflowio/deepflow/pull/7964) by [askyrie](https://github.com/askyrie)
 * feat: support free OS memory at intervals [#8091](https://github.com/deepflowio/deepflow/pull/8091) by [lzf575](https://github.com/lzf575)
 * feat: support profile data compressed by agent [#8088](https://github.com/deepflowio/deepflow/pull/8088) by [taloric](https://github.com/taloric)
 * feat: agent - eBPF Optimize interface 'java_syms_update_main' [#8102](https://github.com/deepflowio/deepflow/pull/8102) by [yinjiping](https://github.com/yinjiping)
@@ -391,6 +402,11 @@
 * feat: server directly reports metrics of load1_by_cpu_num that can be used for alert [#7088](https://github.com/deepflowio/deepflow/pull/7088) by [lzf575](https://github.com/lzf575)
 
 #### Refactoring
+* refactor: agent is compatibles with configuration and interfaces of both old and new versions [#8084](https://github.com/deepflowio/deepflow/pull/8084) by [TomatoMr](https://github.com/TomatoMr)
+* refactor: agent adds the 'disabled-cgroups' configuration. [#8164](https://github.com/deepflowio/deepflow/pull/8164) by [TomatoMr](https://github.com/TomatoMr)
+* refactor: Remove extended proc event handler [#8161](https://github.com/deepflowio/deepflow/pull/8161) by [rvql](https://github.com/rvql)
+* refactor: Remove old profiler regex [#8158](https://github.com/deepflowio/deepflow/pull/8158) by [rvql](https://github.com/rvql)
+* refactor: agent adds a disable_cgroups option [#8140](https://github.com/deepflowio/deepflow/pull/8140) by [TomatoMr](https://github.com/TomatoMr)
 * refactor: update cli [#8134](https://github.com/deepflowio/deepflow/pull/8134) by [lzf575](https://github.com/lzf575)
 * refactor: remove unused sql [#7656](https://github.com/deepflowio/deepflow/pull/7656) by [roryye](https://github.com/roryye)
 * refactor: Adjust agent profile configurations [#7918](https://github.com/deepflowio/deepflow/pull/7918) by [rvql](https://github.com/rvql)
@@ -436,6 +452,8 @@
 
 
 #### OTHER
+* add ci to clear stale branch [#8149](https://github.com/deepflowio/deepflow/pull/8149) by [jiumos](https://github.com/jiumos)
+* failed to create tables in ByConity database under non-default organization [#8148](https://github.com/deepflowio/deepflow/pull/8148) by [lzf575](https://github.com/lzf575)
 * Update typo in comments [#8127](https://github.com/deepflowio/deepflow/pull/8127) by [Hyzhou](https://github.com/Hyzhou)
 * Cmbc [#8098](https://github.com/deepflowio/deepflow/pull/8098) by [rvql](https://github.com/rvql)
 * Update the description of `status` in l4_flow_log. [#8036](https://github.com/deepflowio/deepflow/pull/8036) by [deepflow-lifei](https://github.com/deepflow-lifei)

agent/benches/common.rs

@@ -18,11 +18,11 @@ use std::time::Instant;
 
 use criterion::*;
 
-use deepflow_agent::common::TapTyper;
+use deepflow_agent::common::CaptureNetworkTyper;
 
 fn bench_tap_typer(c: &mut Criterion) {
     c.bench_function("get_tap_type_by_vlan", |b| {
-        let tap_typer = TapTyper::new();
+        let tap_typer = CaptureNetworkTyper::new();
 
         b.iter_custom(|iters| {
             let start = Instant::now();

agent/benches/flow_generator/flow_map.rs

@@ -24,13 +24,13 @@ use deepflow_agent::{
     _reverse_meta_packet as reverse_meta_packet, common::meta_packet::ProtocolData,
 };
 
-use public::proto::common::TridentType;
+use public::proto::agent::AgentType;
 
 pub(super) fn bench(c: &mut Criterion) {
     c.bench_function("flow_map_syn_flood", |b| {
         b.iter_custom(|iters| {
             let (module_config, mut map, _) =
-                new_flow_map_and_receiver(TridentType::TtProcess, None, false);
+                new_flow_map_and_receiver(AgentType::TtProcess, None, false);
             let config = Config {
                 flow: &module_config.flow,
                 log_parser: &module_config.log_parser,
@@ -58,7 +58,7 @@ pub(super) fn bench(c: &mut Criterion) {
     c.bench_function("flow_map_with_ten_packets_flow_flood", |b| {
         b.iter_custom(|iters| {
             let (module_config, mut map, _) =
-                new_flow_map_and_receiver(TridentType::TtProcess, None, false);
+                new_flow_map_and_receiver(AgentType::TtProcess, None, false);
             let config = Config {
                 flow: &module_config.flow,
                 log_parser: &module_config.log_parser,

agent/config/deepflow-agent-windows.yaml

@@ -46,3 +46,9 @@ controller-ips:
 
 ## Team identity for server sync, defaults to ""
 #team-id:
+
+## Disabled cgroups, deepflow-agent will default to checking the CPU and memory resource usage in a loop every 10 seconds to prevent resource usage from exceeding limits
+#cgroups-disabled: false
+
+## If it is true, it means that the communication with the server uses the new RPC interface
+#new_rpc: false

agent/config/deepflow-agent.yaml

@@ -48,3 +48,9 @@ controller-ips:
 
 ## Team identity for server sync, defaults to ""
 #team-id:
+
+## Disabled cgroups, deepflow-agent will default to checking the CPU and memory resource usage in a loop every 10 seconds to prevent resource usage from exceeding limits
+#cgroups-disabled: false
+
+## If it is true, it means that the communication with the server uses the new RPC interface
+#new_rpc: false

agent/crates/public/src/enums.rs

@@ -163,59 +163,59 @@ impl Default for L4Protocol {
 }
 
 #[derive(Serialize, Debug, Clone, Copy, Hash, PartialEq, Eq, Ord)]
-pub enum TapType {
+pub enum CaptureNetworkType {
     Any,
     Idc(u8),
     Cloud,
     Max,
     Unknown,
 }
 
-impl PartialOrd for TapType {
+impl PartialOrd for CaptureNetworkType {
     fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
         u16::from(*self).partial_cmp(&u16::from(*other))
     }
 }
 
-impl TryFrom<u16> for TapType {
+impl TryFrom<u16> for CaptureNetworkType {
     type Error = &'static str;
-    fn try_from(t: u16) -> Result<TapType, Self::Error> {
+    fn try_from(t: u16) -> Result<CaptureNetworkType, Self::Error> {
         match t {
-            0 => Ok(TapType::Any),
-            3 => Ok(TapType::Cloud),
-            0xffff => Ok(TapType::Unknown),
-            v if v < 256 => Ok(TapType::Idc(v as u8)),
-            _ => Err("TapType not in [0, 256)"),
+            0 => Ok(CaptureNetworkType::Any),
+            3 => Ok(CaptureNetworkType::Cloud),
+            0xffff => Ok(CaptureNetworkType::Unknown),
+            v if v < 256 => Ok(CaptureNetworkType::Idc(v as u8)),
+            _ => Err("CaptureNetworkType not in [0, 256)"),
         }
     }
 }
 
-impl From<TapType> for u16 {
-    fn from(t: TapType) -> u16 {
+impl From<CaptureNetworkType> for u16 {
+    fn from(t: CaptureNetworkType) -> u16 {
         match t {
-            TapType::Any => 0,
-            TapType::Idc(v) => v as u16,
-            TapType::Cloud => 3,
-            TapType::Max => 256,
-            TapType::Unknown => 0xffff,
+            CaptureNetworkType::Any => 0,
+            CaptureNetworkType::Idc(v) => v as u16,
+            CaptureNetworkType::Cloud => 3,
+            CaptureNetworkType::Max => 256,
+            CaptureNetworkType::Unknown => 0xffff,
         }
     }
 }
 
-impl Default for TapType {
-    fn default() -> TapType {
-        TapType::Any
+impl Default for CaptureNetworkType {
+    fn default() -> CaptureNetworkType {
+        CaptureNetworkType::Any
     }
 }
 
-impl fmt::Display for TapType {
+impl fmt::Display for CaptureNetworkType {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            TapType::Any => write!(f, "any"),
-            TapType::Idc(n) => write!(f, "isp{}", n),
-            TapType::Cloud => write!(f, "tor"),
-            TapType::Max => write!(f, "max"),
-            TapType::Unknown => write!(f, "unknown"),
+            CaptureNetworkType::Any => write!(f, "any"),
+            CaptureNetworkType::Idc(n) => write!(f, "isp{}", n),
+            CaptureNetworkType::Cloud => write!(f, "tor"),
+            CaptureNetworkType::Max => write!(f, "max"),
+            CaptureNetworkType::Unknown => write!(f, "unknown"),
         }
     }
 }
@@ -401,6 +401,6 @@ mod tests {
     fn check_type_sizes() {
         assert_eq!(size_of::<EthernetType>(), 2);
         assert_eq!(size_of::<IpProtocol>(), 1);
-        assert_eq!(size_of::<TapType>(), 2);
+        assert_eq!(size_of::<CaptureNetworkType>(), 2);
     }
 }

agent/crates/public/src/rpc.rs

@@ -6,7 +6,8 @@ pub mod remote_exec {
     use futures::future::BoxFuture;
     use thiserror::Error;
 
-    use crate::proto::trident as pb;
+    use crate::proto::agent as pb;
+    use crate::proto::trident;
 
     pub const DEFAULT_PARAM_REGEX: &'static str = "^[A-Za-z0-9-_]+$";
 
@@ -175,6 +176,17 @@ pub mod remote_exec {
         }
     }
 
+    // FIXME: In order to be compatible with the old and new interfaces, this code should be deleted later
+    pub fn get_params_from_trident(params: &[trident::Parameter]) -> Vec<pb::Parameter> {
+        params
+            .iter()
+            .map(|p| pb::Parameter {
+                key: p.key.clone(),
+                value: p.value.clone(),
+            })
+            .collect::<Vec<_>>()
+    }
+
     #[cfg(test)]
     mod tests {
         use super::*;

agent/examples/remote_executor.rs

@@ -42,6 +42,7 @@ fn main() {
         vec!["127.0.0.1".to_owned()],
         exc.clone(),
         &stats_collector,
+        false,
     ));
 
     let runtime = Arc::new(

agent/plugins/npb_handler/src/lib.rs

@@ -21,7 +21,7 @@ use npb_pcap_policy::{NpbTunnelType, PolicyData};
 use public::{
     counter::{CounterType, CounterValue, OwnedCountable},
     leaky_bucket::LeakyBucket,
-    proto::trident::VlanMode,
+    proto::agent::VlanMode,
     queue::DebugSender,
 };
 

agent/src/bin/deepflow-agent-ctl.rs

@@ -264,7 +264,7 @@ struct RpcCmd {
 enum RpcData {
     Config,
     Platform,
-    TapTypes,
+    CaptureNetworkTypes,
     Cidr,
     Groups,
     Acls,
@@ -372,15 +372,15 @@ impl Controller {
 
                     let beacon: Beacon =
                         decode_from_std_read(&mut &buf[length..n], config::standard())?;
-                    if !vtap_map.contains(&beacon.vtap_id) {
+                    if !vtap_map.contains(&beacon.agent_id) {
                         println!(
                             "{:<14} {:<28} {:<45} {}",
-                            beacon.vtap_id,
+                            beacon.agent_id,
                             beacon.hostname,
                             a.ip(),
                             a.port()
                         );
-                        vtap_map.insert(beacon.vtap_id);
+                        vtap_map.insert(beacon.agent_id);
                     }
                 }
                 Err(e) => return Err(anyhow!("{}", e)),
@@ -398,7 +398,7 @@ impl Controller {
             RpcData::Acls => RpcMessage::Acls(None),
             RpcData::Config => RpcMessage::Config(None),
             RpcData::Platform => RpcMessage::PlatformData(None),
-            RpcData::TapTypes => RpcMessage::TapTypes(None),
+            RpcData::CaptureNetworkTypes => RpcMessage::CaptureNetworkTypes(None),
             RpcData::Cidr => RpcMessage::Cidr(None),
             RpcData::Groups => RpcMessage::Groups(None),
             RpcData::Segments => RpcMessage::Segments(None),
@@ -418,7 +418,7 @@ impl Controller {
             match resp {
                 RpcMessage::Acls(v)
                 | RpcMessage::PlatformData(v)
-                | RpcMessage::TapTypes(v)
+                | RpcMessage::CaptureNetworkTypes(v)
                 | RpcMessage::Cidr(v)
                 | RpcMessage::Groups(v)
                 | RpcMessage::Segments(v) => match v {

agent/src/collector/collector.rs

@@ -200,7 +200,7 @@ impl StashKey {
             // fast_id
             // 128          72        64    59       56        48           32          24         16        0
             // +-------------+---------+-----+--------+---------+------------+-----------+----------+---------+
-            // |             | L7Proto | MAC | CodeID | TapType | ServerPort | Direction | Protocol | L3EpcId |
+            // |             | L7Proto | MAC | CodeID | CaptureNetworkType | ServerPort | Direction | Protocol | L3EpcId |
             // +-------------+---------+-----+--------+---------+------------+-----------+----------+---------+
             Self::SINGLE_MAC_IP_PORT_APP => {
                 fast_id |= (tagger.l7_protocol as u128) << 64;
@@ -252,7 +252,7 @@ impl StashKey {
             // RESERVED
             // 20    19       16        8         0
             // ------------------------------------
-            // | MAC | CodeID | L7Proto | TapType |
+            // | MAC | CodeID | L7Proto | CaptureNetworkType |
             // ------------------------------------
             Self::EDGE_MAC_IP_PORT_APP => {
                 let tap_port_reserve = (tagger.l7_protocol as u32) << 8 | 3 << 16 | 1 << 19;
@@ -447,7 +447,7 @@ impl Stash {
                     acl_gid,
                     server_port: ip_id,
                     signal_source: flow.signal_source,
-                    vtap_id: config.vtap_id,
+                    agent_id: config.agent_id,
                     ..Default::default()
                 };
                 let meter = &acc_flow.flow_meter;
@@ -468,7 +468,7 @@ impl Stash {
                     acl_gid,
                     server_port: ip_id,
                     signal_source: flow.signal_source,
-                    vtap_id: config.vtap_id,
+                    agent_id: config.agent_id,
                     ..Default::default()
                 };
 
@@ -970,7 +970,7 @@ fn get_single_tagger(
 
     Tagger {
         global_thread_id,
-        vtap_id: config.vtap_id,
+        agent_id: config.agent_id,
         mac: if !has_mac {
             MacAddr::ZERO
         } else if ep == FLOW_METRICS_PEER_SRC {
@@ -1087,7 +1087,7 @@ fn get_edge_tagger(
 
     Tagger {
         global_thread_id,
-        vtap_id: config.vtap_id,
+        agent_id: config.agent_id,
         mac: src_mac,
         mac1: dst_mac,
         ip: src_ip,
@@ -1472,7 +1472,7 @@ mod tests {
 
     use std::collections::HashSet;
 
-    use crate::common::enums::TapType;
+    use crate::common::enums::CaptureNetworkType;
 
     use super::*;
 
@@ -1511,7 +1511,7 @@ mod tests {
         tagger.l3_epc_id = (tagger.l3_epc_id as u16 ^ 0x8000) as i16;
         let key = StashKey::new(&tagger, Ipv4Addr::UNSPECIFIED.into(), None, 0);
         assert_eq!(map.insert(key), true);
-        tagger.tap_type = TapType::Idc(255);
+        tagger.tap_type = CaptureNetworkType::Idc(255);
         let key = StashKey::new(&tagger, Ipv4Addr::UNSPECIFIED.into(), None, 0);
         assert_eq!(map.insert(key), true);
         tagger.direction = Direction::ServerToClient;
@@ -1550,7 +1550,7 @@ mod tests {
         tagger.l3_epc_id1 = (tagger.l3_epc_id as u16 ^ 0x8000) as i16;
         let key = StashKey::new(&tagger, Ipv4Addr::UNSPECIFIED.into(), None, 0);
         assert_eq!(map.insert(key), true);
-        tagger.tap_type = TapType::Idc(200);
+        tagger.tap_type = CaptureNetworkType::Idc(200);
         let key = StashKey::new(&tagger, Ipv4Addr::UNSPECIFIED.into(), None, 0);
         assert_eq!(map.insert(key), true);
         tagger.direction = Direction::ClientToServer;

agent/src/collector/flow_aggr.rs

@@ -33,7 +33,7 @@ use super::consts::*;
 use crate::collector::types::U16Set;
 use crate::common::Timestamp;
 use crate::common::{
-    enums::TapType,
+    enums::CaptureNetworkType,
     flow::CloseType,
     tagged_flow::{BoxedTaggedFlow, TaggedFlow},
 };
@@ -345,7 +345,8 @@ impl FlowAggr {
                         {
                             continue;
                         }
-                        if config.l4_log_store_tap_types[u16::from(TapType::Any) as usize]
+                        if config.l4_log_store_tap_types
+                            [u16::from(CaptureNetworkType::Any) as usize]
                             || config.l4_log_store_tap_types
                                 [u16::from(tagged_flow.flow.flow_key.tap_type) as usize]
                             || tagged_flow.flow.need_to_store