feat: ebpf support dpdk

deepflowio · Aug 22, 2024 · 4fdb3b3 · 4fdb3b3
1 parent 8a13d17
commit 4fdb3b3
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 2 deletions.
diff --git a/agent/src/ebpf_dispatcher/ebpf_dispatcher.rs b/agent/src/ebpf_dispatcher/ebpf_dispatcher.rs
@@ -404,6 +404,7 @@ pub struct EbpfCollector {
 
 static mut SWITCH: bool = false;
 static mut SENDER: Option<DebugSender<Box<MetaPacket>>> = None;
+static mut DPDK_SENDER: Option<DebugSender<Box<MetaPacket>>> = None;
 static mut PROC_EVENT_SENDER: Option<DebugSender<BoxedProcEvents>> = None;
 static mut EBPF_PROFILE_SENDER: Option<DebugSender<Profile>> = None;
 static mut POLICY_GETTER: Option<PolicyGetter> = None;
@@ -423,6 +424,22 @@ impl EbpfCollector {
         CStr::from_ptr(ptr).to_string_lossy().into_owned()
     }
 
+    extern "C" fn ebpf_dpdk_callback(sd: *mut ebpf::SK_BPF_DATA) {
+        unsafe {
+            if !SWITCH || DPDK_SENDER.is_none() {
+                return;
+            }
+            let packet = MetaPacket::from_ebpf(sd);
+            if packet.is_err() {
+                warn!("meta packet parse from ebpf error: {}", packet.unwrap_err());
+                return;
+            }
+            if let Err(e) = SENDER.as_mut().unwrap().send(Box::new(packet.unwrap())) {
+                warn!("meta packet send ebpf error: {:?}", e);
+            }
+        }
+    }
+
     extern "C" fn ebpf_l7_callback(sd: *mut ebpf::SK_BPF_DATA) {
         unsafe {
             if !SWITCH || SENDER.is_none() {
@@ -533,6 +550,7 @@ impl EbpfCollector {
     fn ebpf_init(
         config: &EbpfConfig,
         sender: DebugSender<Box<MetaPacket<'static>>>,
+        dpdk_sender: Option<DebugSender<Box<MetaPacket<'static>>>>,
         proc_event_sender: DebugSender<BoxedProcEvents>,
         ebpf_profile_sender: DebugSender<Profile>,
         l7_protocol_enabled_bitmap: L7ProtocolBitmap,
@@ -814,6 +832,7 @@ impl EbpfCollector {
         unsafe {
             SWITCH = false;
             SENDER = Some(sender);
+            DPDK_SENDER = dpdk_sender;
             PROC_EVENT_SENDER = Some(proc_event_sender);
             EBPF_PROFILE_SENDER = Some(ebpf_profile_sender);
             POLICY_GETTER = Some(policy_getter);
@@ -894,6 +913,7 @@ impl EbpfCollector {
         queue_debugger: &QueueDebugger,
         stats_collector: Arc<stats::Collector>,
         exception_handler: ExceptionHandler,
+        dpdk_sender: Option<DebugSender<Box<MetaPacket<'static>>>>,
     ) -> Result<Box<Self>> {
         let ebpf_config = config.load();
         if ebpf_config.ebpf.disabled {
@@ -915,6 +935,7 @@ impl EbpfCollector {
         Self::ebpf_init(
             &ebpf_config,
             sender,
+            dpdk_sender,
             proc_event_output,
             ebpf_profile_sender,
             ebpf_config.l7_protocol_enabled_bitmap,

diff --git a/agent/src/flow_generator/protocol_logs/rpc/tars.rs b/agent/src/flow_generator/protocol_logs/rpc/tars.rs
@@ -24,7 +24,7 @@ use crate::{
     flow_generator::{
         error::Result,
         protocol_logs::{
-            pb_adapter::{L7ProtocolSendLog, L7Request, L7Response, ExtendedInfo},
+            pb_adapter::{ExtendedInfo, L7ProtocolSendLog, L7Request, L7Response},
             AppProtoHead, L7ResponseStatus, LogMessageType,
         },
     },
@@ -280,7 +280,7 @@ impl TarsInfo {
                         | TARS_SERVER_NO_SERVANT_ERR
                         | TARS_SERVER_RESET_GRID
                         | TARS_SERVER_QUEUE_TIMEOUT
-                        | TARS_ASYNC_CALL_OR_INVOKE_TIMEOUT 
+                        | TARS_ASYNC_CALL_OR_INVOKE_TIMEOUT
                         | TARS_PROXY_CONNECT_ERR
                         | TARS_SERVER_UNKNOWN_ERR => {
                             info.resp_status = L7ResponseStatus::ServerError;

diff --git a/agent/src/trident.rs b/agent/src/trident.rs
@@ -2365,6 +2365,7 @@ impl AgentComponents {
                 &queue_debugger,
                 stats_collector.clone(),
                 exception_handler.clone(),
+                None,
             ) {
                 Ok(ebpf_collector) => {
                     synchronizer