feat: Memcached ebpf support

deepflowio · Oct 12, 2024 · 0e08c31 · 0e08c31
1 parent 575d89d
commit 0e08c31
Show file tree

Hide file tree

Showing 13 changed files with 306 additions and 137 deletions.
diff --git a/agent/src/bin/deepflow-agent-ctl.rs b/agent/src/bin/deepflow-agent-ctl.rs
@@ -173,7 +173,8 @@ struct EbpfArgs {
     ///
     /// App Protocol: All(0), Other(1),
     ///   HTTP1(20), HTTP2(21), Dubbo(40), SofaRPC(43),
-    ///   MySQL(60), PostGreSQL(61), Oracle(62), Redis(80),
+    ///   MySQL(60), PostGreSQL(61), Oracle(62), Memcached(63),
+    ///   Redis(80),
     ///   Kafka(100), MQTT(101), DNS(120), TLS(121),
     ///
     /// eg: deepflow-agent-ctl ebpf datadump --proto 20

diff --git a/agent/src/common/l7_protocol_log.rs b/agent/src/common/l7_protocol_log.rs
@@ -532,8 +532,8 @@ impl L7ProtocolChecker for L7ProtocolBitmap {
     }
 }
 
-impl From<&Vec<String>> for L7ProtocolBitmap {
-    fn from(vs: &Vec<String>) -> Self {
+impl From<&[String]> for L7ProtocolBitmap {
+    fn from(vs: &[String]) -> Self {
         let mut bitmap = L7ProtocolBitmap(0);
         for v in vs.iter() {
             if let Ok(p) = L7ProtocolParser::try_from(v.as_str()) {

diff --git a/agent/src/config/config.rs b/agent/src/config/config.rs
@@ -39,7 +39,7 @@ use tokio::runtime::Runtime;
 
 use crate::common::l7_protocol_log::L7ProtocolParser;
 use crate::dispatcher::recv_engine::DEFAULT_BLOCK_SIZE;
-use crate::flow_generator::{DnsLog, OracleLog, TlsLog};
+use crate::flow_generator::{DnsLog, MemcachedLog, OracleLog, TlsLog};
 #[cfg(any(target_os = "linux", target_os = "android"))]
 use crate::platform::{get_container_id, OsAppTag, ProcessData};
 use crate::{
@@ -1500,6 +1500,7 @@ impl Default for ApplicationProtocolInference {
                 "MySQL".to_string(),
                 "PostgreSQL".to_string(),
                 "Oracle".to_string(),
+                "Memcached".to_string(),
                 "Redis".to_string(),
                 "MongoDB".to_string(),
                 "Kafka".to_string(),
@@ -3087,6 +3088,7 @@ impl UserConfig {
     const DEFAULT_DNS_PORTS: &'static str = "53,5353";
     const DEFAULT_TLS_PORTS: &'static str = "443,6443";
     const DEFAULT_ORACLE_PORTS: &'static str = "1521";
+    const DEFAULT_MEMCACHED_PORTS: &'static str = "11211";
     const PACKET_FANOUT_MODE_MAX: u32 = 7;
 
     pub fn get_fast_path_map_size(&self, mem_size: u64) -> usize {
@@ -3155,6 +3157,20 @@ impl UserConfig {
                 Self::DEFAULT_ORACLE_PORTS.to_string(),
             );
         }
+        let memcached_str = L7ProtocolParser::Memcached(MemcachedLog::default()).as_str();
+        // memcached default only parse 11211 port. when l7_protocol_ports config without MEMCACHED, need to reserve the memcached default config.
+        if !self
+            .processors
+            .request_log
+            .filters
+            .port_number_prefilters
+            .contains_key(memcached_str)
+        {
+            new.insert(
+                memcached_str.to_string(),
+                Self::DEFAULT_MEMCACHED_PORTS.to_string(),
+            );
+        }
 
         new
     }
@@ -3816,6 +3832,7 @@ impl YamlConfig {
     const DEFAULT_DNS_PORTS: &'static str = "53,5353";
     const DEFAULT_TLS_PORTS: &'static str = "443,6443";
     const DEFAULT_ORACLE_PORTS: &'static str = "1521";
+    const DEFAULT_MEMCACHED_PORTS: &'static str = "11211";
     const PACKET_FANOUT_MODE_MAX: u32 = 7;
     const DEFAULT_L7_PROTOCOL_ENABLED: [&'static str; 7] =
         ["HTTP", "HTTP2", "MySQL", "Redis", "Kafka", "DNS", "TLS"];
@@ -4087,6 +4104,14 @@ impl YamlConfig {
                 Self::DEFAULT_ORACLE_PORTS.to_string(),
             );
         }
+        let memcached_str = L7ProtocolParser::Memcached(MemcachedLog::default()).as_str();
+        // memcached default only parse 11211 port. when l7_protocol_ports config without MEMCACHED, need to reserve the memcached default config.
+        if !self.l7_protocol_ports.contains_key(memcached_str) {
+            new.insert(
+                memcached_str.to_string(),
+                Self::DEFAULT_MEMCACHED_PORTS.to_string(),
+            );
+        }
 
         new
     }

diff --git a/agent/src/config/handler.rs b/agent/src/config/handler.rs
@@ -574,11 +574,11 @@ impl From<(&UserConfig, &DynamicConfig)> for FlowConfig {
             packet_sequence_flag: conf.processors.packet.tcp_header.header_fields_flag, // Enterprise Edition Feature: packet-sequence
             packet_sequence_block_size: conf.processors.packet.tcp_header.block_size, // Enterprise Edition Feature: packet-sequence
             l7_protocol_enabled_bitmap: L7ProtocolBitmap::from(
-                &conf
-                    .processors
+                conf.processors
                     .request_log
                     .application_protocol_inference
-                    .enabled_protocols,
+                    .enabled_protocols
+                    .as_slice(),
             ),
             l7_protocol_parse_port_bitmap: Arc::new(conf.get_protocol_port_parse_bitmap()),
             plugins: PluginConfig {
@@ -639,11 +639,11 @@ impl From<(&UserConfig, &DynamicConfig)> for FlowConfig {
                 .oracle
                 .clone(),
             obfuscate_enabled_protocols: L7ProtocolBitmap::from(
-                &conf
-                    .processors
+                conf.processors
                     .request_log
                     .tag_extraction
-                    .obfuscate_protocols,
+                    .obfuscate_protocols
+                    .as_slice(),
             ),
             server_ports: conf
                 .processors
@@ -1919,11 +1919,11 @@ impl TryFrom<(Config, UserConfig, DynamicConfig)> for ModuleConfig {
                     &conf.processors.request_log.tag_extraction.http_endpoint,
                 ),
                 obfuscate_enabled_protocols: L7ProtocolBitmap::from(
-                    &conf
-                        .processors
+                    conf.processors
                         .request_log
                         .tag_extraction
-                        .obfuscate_protocols,
+                        .obfuscate_protocols
+                        .as_slice(),
                 ),
                 l7_log_blacklist: conf.processors.request_log.filters.tag_filters.clone(),
                 l7_log_blacklist_trie: {
@@ -2057,11 +2057,11 @@ impl TryFrom<(Config, UserConfig, DynamicConfig)> for ModuleConfig {
                     MacAddr::ZERO
                 },
                 l7_protocol_enabled_bitmap: L7ProtocolBitmap::from(
-                    &conf
-                        .processors
+                    conf.processors
                         .request_log
                         .application_protocol_inference
-                        .enabled_protocols,
+                        .enabled_protocols
+                        .as_slice(),
                 ),
                 l7_protocol_parse_port_bitmap: Arc::new(conf.get_protocol_port_parse_bitmap()),
                 l7_protocol_ports: conf.get_protocol_port(),

diff --git a/agent/src/ebpf/kernel/include/common.h b/agent/src/ebpf/kernel/include/common.h
@@ -76,6 +76,7 @@ enum traffic_protocol {
 	PROTO_MYSQL = 60,
 	PROTO_POSTGRESQL = 61,
 	PROTO_ORACLE = 62,
+	PROTO_MEMCACHED = 63,
 	PROTO_REDIS = 80,
 	PROTO_MONGO = 81,
 	PROTO_KAFKA = 100,