Fixed #TASK_QT-13496 保证了程序的正常运行，删除queue，删除expire_sock_record函数

ebpftracer/ebpf/netflow/kflowd.bpf.c

@@ -309,102 +309,6 @@ static __always_inline int submit_sock_record(void* ctx, struct SOCK_INFO *sinfo
     return 0;
 }
 
-/* check for expired records */
-static __always_inline void expire_sock_records(void* ctx) {
-    struct SOCK_INFO *sq_sinfo;
-    struct SOCK_QUEUE sq = {0};
-    struct STATS     *s;
-    __u64             qlen = 0;
-    __u64             ts_now;
-    __u32             zero = 0;
-    int               cnt;
-    const volatile __u32 agg_idle_timeout = 15;
-    const volatile __u32 agg_active_timeout = 1800;
-
-    s = bpf_map_lookup_elem(&stats, &zero);
-    if (s) {
-        qlen = s->q_push_added + s->q_push_updated - s->q_pop_expired - s->q_pop_ignored - s->q_pop_missed;
-        if (!qlen)
-            return;
-        if (debug_proc(NULL, NULL))
-            bpf_printk("EXPIRE_SOCK_RECORDS: %lu records in queue", qlen);
-    }
-    ts_now = bpf_ktime_get_ns();
-#pragma unroll
-    for (cnt = 0; cnt < SOCK_EXP_MAX; cnt++) {
-        if (s && cnt >= qlen)
-            break;
-        // if (!bpf_map_pop_elem(&queue_socks, &sq)) {
-        if(1) {
-            sq_sinfo = bpf_map_lookup_elem(&hash_socks, &sq.key);
-            if (sq_sinfo) {
-                __u64 ts_last = MAX(sq_sinfo->rx_ts, sq_sinfo->tx_ts);
-                if (debug_proc(NULL, NULL))
-                    bpf_printk("Popped socket key %lx with lport %u and rport %u from queue", sq.key, sq_sinfo->lport,
-                               sq_sinfo->rport);
-                if (sq.ts < ts_last) {
-                    if (debug_proc(NULL, NULL))
-                        bpf_printk("Ignored socket key %lx with %lu outdated timestamp %lu", sq.key, ts_last - sq.ts,
-                                   sq.ts);
-                    if (s)
-                        s->q_pop_ignored++;
-                } else if (sq.ts > ts_last) {
-                    // bpf_printk("WARNING: Timestamp for %s socket key %lx is %lu greater than last timestamp",
-                    //            GET_ROLE_STR(sq_sinfo->role), sq.key, sq.ts - ts_last);
-                    bpf_printk("time out\n");
-                    if (s)
-                        s->q_pop_missed++;
-                } else if (ts_now - sq.ts > agg_idle_timeout * (u64)1e9 ||
-                           ts_now - sq_sinfo->ts_first > agg_active_timeout * (u64)1e9) {
-                    /* set udp state to close on idle timeout */
-                    if (sq_sinfo->proto == IPPROTO_UDP && ts_now - sq.ts > agg_idle_timeout * (u64)1e9)
-                        sq_sinfo->state = UDP_CLOSE;
-                    submit_sock_record(ctx, sq_sinfo);
-                    if (s)
-                        s->q_pop_expired++;
-                    if (debug_proc(NULL, NULL)) {
-                        // bpf_printk("Expired socket %s key %lx for pid %u", GET_ROLE_STR(sq_sinfo->role), sq.key,
-                        //            sq_sinfo->pid);
-                        bpf_printk("Expired socket\n");
-                        if (ts_now - sq.ts > agg_idle_timeout * (u64)1e9)
-                            bpf_printk("Expired socket key with %lu idle timestamp %lu", ts_now - sq.ts, sq.ts);
-                        else
-                            bpf_printk("Expired socket key with %lu active timestamp %lu", ts_now - sq_sinfo->ts_first,
-                                       sq_sinfo->ts_first);
-                    }
-                    /* delete only UDP socket since TCP socket is deleted on TCP_CLOSE in state machine */
-                    if (sq_sinfo->proto == IPPROTO_UDP) {
-                        if (bpf_map_delete_elem(&hash_socks, &sq.key))
-                            // bpf_printk("WARNING: Failed to delete %s socket for pid %u", GET_ROLE_STR(sq_sinfo->role),
-                            //            sq_sinfo->pid);
-                            bpf_printk("bpf_map_delete_elem failed\n");
-                    }
-                } else {
-                    // if (!bpf_map_push_elem(&queue_socks, &sq, BPF_EXIST)) {
-                    if(1){
-                        if (s)
-                            s->q_push_readded++;
-                        if (debug_proc(NULL, NULL))
-                            bpf_printk("Repushed socket key %lx with lport %u and rport %u to queue", sq.key,
-                                       sq_sinfo->lport, sq_sinfo->rport);
-                    }
-                }
-            } else if (debug_proc(NULL, NULL)) {
-                if (s)
-                    s->q_pop_missed++;
-                bpf_printk("Popped socket key %lx not found", sq.key);
-            }
-        }
-    }
-
-    if (debug_proc(NULL, NULL) && s) {
-        qlen = s->q_push_added + s->q_push_updated - s->q_pop_expired - s->q_pop_ignored - s->q_pop_missed;
-        bpf_printk("  QUEUE: %lu records", qlen);
-        bpf_printk("  PUSH: %u added   %u updated   %u readded", s->q_push_added, s->q_push_updated, s->q_push_readded);
-        bpf_printk("  POP: %u expired   %u ignored   %u missed\n", s->q_pop_expired, s->q_pop_ignored, s->q_pop_missed);
-    }
-}
-
 /* handle tcp socket tx and rx events */
 static __always_inline int handle_tcp_event(void *ctx, const struct SOCK_EVENT_INFO *event) {
     pid_t               pid = bpf_get_current_pid_tgid() >> 32;
@@ -793,9 +697,6 @@ static __always_inline int handle_tcp_packet(void* ctx, struct cw_net_sock *sock
     __u32               cntf;
     __u32               zero = 0;
 
-    /* clean expired records */
-    // expire_sock_records(ctx);//TODO 考虑优化此函数
-
     /* try to get sock from buffer if zero */
     if (!sock) {
         // sock = BPF_CORE_READ(skb, sk);
@@ -1085,25 +986,6 @@ static __always_inline int handle_tcp_packet(void* ctx, struct cw_net_sock *sock
             //     bpf_printk("Updated tcp %s flags of socket %lx for pid %u", isrx ? "rx" : "tx", key, sinfo->pid);
             sq.key = key;
             sq.ts = ts_now;
-            // if (!bpf_map_push_elem(&queue_socks, &sq, BPF_EXIST)) {
-            if(1){
-                s = bpf_map_lookup_elem(&stats, &zero);
-                if (s) {
-                    if (sinfo->rx_events + sinfo->tx_events == 1)
-                        s->q_push_added++;
-                    else
-                        s->q_push_updated++;
-                }
-                // if (debug_proc(sinfo->comm, NULL))
-                //     bpf_printk("Pushed tcp key %lx with lport %u and rport %u to queue", key, sinfo->lport,
-                //                sinfo->rport);
-                if (s) {
-                    __u32 qlen =
-                        s->q_push_added + s->q_push_updated - s->q_pop_expired - s->q_pop_ignored - s->q_pop_missed;
-                    // if (debug_proc(sinfo->comm, NULL))
-                    //     bpf_printk("%lu records in queue", qlen);
-                }
-            }
         } 
         else
             bpf_printk("WARNING: Failed to update tcp flags of socket %lx for pid %u", key,
@@ -1331,8 +1213,6 @@ static __always_inline int handle_udp_event(void *ctx, const struct SOCK_EVENT_I
     //         break;
     //     }
     // }
-    /* clean expired records */
-    // expire_sock_records(ctx);
 
     /* lookup and update socket */
     // key = KEY_SOCK(BPF_CORE_READ(sock, __sk_common.skc_hash));
@@ -1424,13 +1304,6 @@ static __always_inline int handle_udp_event(void *ctx, const struct SOCK_EVENT_I
             // bpf_printk("bpf_map_update_elem success \n");
             sq.key = key;
             sq.ts = ts_now;
-            // if (!bpf_map_push_elem(&queue_socks, &sq, BPF_EXIST)) {
-            if(1){
-                if (s)
-                    s->q_push_updated++;
-                // if (debug_proc(sinfo->comm, NULL))
-                //     bpf_printk("Pushed udp socket %lx with lport %u and rport %u to queue\n", key, lport, rport);
-            }
         } else
             // bpf_printk("WARNING: Failed to update %s socket %lx for pid %u\n", GET_ROLE_STR(sinfo->role), key, pid);
             bpf_printk("failed to update\n");
@@ -1585,13 +1458,6 @@ static __always_inline int handle_udp_event(void *ctx, const struct SOCK_EVENT_I
             bpf_printk("bpf_map_update_elem success\n");
             sq.key = key;
             sq.ts = ts_now;
-            // if (!bpf_map_push_elem(&queue_socks, &sq, BPF_EXIST)) {
-            if(1){
-                if (s)
-                    s->q_push_added++;
-                // if (debug_proc(sinfo->comm, NULL))
-                    // bpf_printk("Pushed first udp socket %lx with lport %u and rport %u to queue", key, lport, rport);
-            }
         } else
             // bpf_printk("WARNING: Failed to add new %s socket for pid %u\n", GET_ROLE_STR(sinfo->role), pid);
             bpf_printk("bpf_map_update_elem failed\n");